open/llm-ingestion-okf
1
0
Fork 0
Code Issues Releases 1 Activity

feat(cli): okf build, one installed command for folder in, bundle out

Browse source 
Until now "run the door over a folder" was a shell loop over two scripts
under `tools/`, with nine flags between them and a `--path-prefix` rule
that lived in a code block in a measurement report. Neither script was
packaged (`pyproject.toml` ships `src/llm_ingestion_okf` only), so the
path the published K1/K2 numbers were measured on was reachable from a
clone and nowhere else.

`okf build <folder> --bundle <dir>` is that path, packaged, declared as a
console script and installed with the wheel. It is orchestration only:
the proposer and the corpus harness MOVED into the package
(`llm_ingestion_okf.propose`, `llm_ingestion_okf.corpus`) and the two
`tools/` scripts became thin entry points to them, so the published
reproduction blocks still run and there is exactly one implementation of
each rule. Neither move adds a dependency or a model call.

Two decisions belong to this layer and are stated where they are made.
A document's proposed paths are scoped by its RELATIVE PATH minus the
extension, not its basename: the door walks recursively now, and two
documents named alike in different folders would otherwise collide on a
path Door B is supposed to make impossible rather than merely detect.
And omitted timestamps do not come from the clock -- `--ingested-at` and
`--proposed-at` default to one shared epoch constant, because a
wall-clock default would put a changing byte in the artifact and take
rebuild-equals-incremental away from every caller who did not pass them.

Arm C and Arm D stay off and are not exposed here.

Measured on the 43-file K2 corpus, one invocation against the two-script
bundle of 2026-09-03: N = 43 computed, merged 39/43, coded rejections
4/43 (`extractor_unknown` 3, `extractor_empty_pdf` 1), K1b 39 + 4 = 43,
exit 0, 779.43 s. 1107 of 1108 files byte-identical. The one that
differs is the root `index.md`, by exactly the `log.md` link a commit
fifteen hours younger than the stored artifact adds -- appending that
line to the stored file reproduces the new one byte for byte. Against
the two scripts at THIS commit the trees agree in full, which is what
the byte-identity test holds.

Suite 1127 passed after `git add` (1113 before), mypy --strict clean,
ruff clean.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Kjell Tore Guttormsen 2026-09-07 05:06:33 +02:00
commit 4ce14ae5dd
15 changed files with 2042 additions and 1244 deletions
Download patch file Download diff file Expand all files Collapse all files

13
CLAUDE.md
View file

@ -234,6 +234,19 @@ and fixtures, never code.
- Test: `pytest`
- Lint: `ruff check .` + `ruff format --check .`
- Type check: `mypy --strict src/`
- Build a bundle: `okf build <folder> --bundle <dir> --bundle-id <id>
--okf-version <v>` — the installed console script (`[project.scripts]`),
the packaged form of what used to be a shell loop over two `tools/`
scripts. It is orchestration only: the proposer and the corpus harness
live in `llm_ingestion_okf.propose` and `llm_ingestion_okf.corpus`, and
the `tools/` scripts are thin entry points to the same functions so the
published reproduction blocks still run. Path scope for a document's
proposals is its RELATIVE path minus the extension (the door walks
recursively, and two same-named documents in different folders must not
collide); `--ingested-at` and `--proposed-at` default to one shared epoch
constant rather than the clock, because a wall-clock default takes
rebuild-equals-incremental away from anyone who omits them. Arm C and
Arm D are off and not exposed here.
## Workflow

50
README.md
View file

@ -52,6 +52,56 @@ against `v0.4.0` is the one combination that fails.
OKF v0.2 pilot set only; pin it only if you are one of them (see
[Upstream OKF versions](#upstream-okf-versions)).
## Build
Installing the package installs one command. A folder of documents in, an OKF
bundle out:
```
okf build ./documents --bundle ./bundle --bundle-id my-bundle --okf-version 0.2
```
It walks the folder recursively, proposes a segmentation for each document with
the mechanical rules, replays those proposals through the bundle inbox, writes
the bundle and its `log.md`, and prints the run's numbers. Every proposal is
marked `PROPOSED` and `adjudicated: false` — the command segments nothing a
human has approved, and says so in the artifact.
The last line that matters is the conservation identity: `merged + coded
rejections == N`, where `N` is the folder's file count read at run time. **The
run exits non-zero when it does not hold**, and names the unaccounted files, so
a pipeline cannot mistake a partial bundle for a complete one.
Flags worth knowing: `--segments off` ingests each document as one concept and
asks for no root values; `--plans-dir` keeps the proposals instead of
discarding them; `--report` writes the full report to a file as well as stdout.
`--ingested-at` and `--proposed-at` default to `1970-01-01T00:00:00Z` rather
than the clock, so two builds of the same folder are byte-identical — a
wall-clock default would break rebuild-equals-incremental for every caller who
did not pass them.
Measured 2026-09-07 on a 43-file corpus (33 `pdf`, 5 `docx`, 2 `xlsx`, and
three files no reader accepts), one
`okf build` invocation replacing the shell loop over `tools/` that produced the
same corpus's bundle on 2026-09-03:
| figure | value |
|---|---|
| `N` (folder file count, computed) | 43 |
| merged | 39/43 |
| coded rejections | 4/43 (`extractor_unknown` 3, `extractor_empty_pdf` 1) |
| K1b | `39 + 4 = 43 = N`, exit `0` |
| files written | 1108 |
| identical to the 2026-09-03 bundle | 1107/1108 |
| wall time | 779.43 s total, 18.126 s per file |
The one file that differs is the root `index.md`, by exactly one line: the link
to the bundle's own `log.md`, added by a commit that postdates the stored
artifact by fifteen hours. Appending that line to the stored `index.md`
reproduces the new one byte for byte. The command's own byte-identity test
compares it against the two scripts at the current commit, where the two agree
over the whole tree.
## Implemented scope (v1)
The library provides three entry points for getting content into an OKF

6
pyproject.toml
View file

@ -34,6 +34,12 @@ classifiers = [
# keep and we do not need to demand.
dependencies = ["llm-ingestion-guard>=1.2,<2.0"]
# The installed command. `okf build <folder> --bundle <dir>` is the packaged
# form of a path that was two unpackaged scripts under `tools/` and nine flags
# -- reachable only from a clone, which is not where a consumer stands.
[project.scripts]
okf = "llm_ingestion_okf.cli:main"
[project.optional-dependencies]
# Binary file-type extraction parsers. OPT-IN ONLY: this extra pulls binary
# wheels (pillow, pypdfium2) and a transitive tree that core must never have —

287
src/llm_ingestion_okf/cli.py Normal file
View file

@ -0,0 +1,287 @@
"""`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`) and Arm D (`--outline-run`) are OFF here and are
not exposed: they are measurement arms, both 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())

488
src/llm_ingestion_okf/corpus.py Normal file
View file

@ -0,0 +1,488 @@
"""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 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,
) -> 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,
)
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)
def link_log_in_root_index(bundle: Path, profile: BundleProfile) -> None:
"""Point the root index at the log, so the walk section 8 supports reaches it.
Measured on the K2 artifact: the bundle carried a conformant root `log.md`
that no index named, so a consumer entering at `index.md` never reached the
one file carrying `N`.
A LOCAL choice, not a conformance requirement, and the difference is worth
stating rather than implying. Section 9 puts `log.md` at any level and
section 8 has an index enumerate its directory's contents, but upstream's
own bundles do not link it: measured at `9a15b13`, 0 of the 24 shipped
`index.md` files name the single `log.md` in the set. Upstream therefore
shows the link is not REQUIRED -- not that it is disallowed.
It belongs to the harness and not the library. The log's content IS the
run's outcome, so it cannot exist when the indexes are projected; an index
that enumerated it off the directory would gain the link only from the
second run onward and break rebuild-equals-incremental, the property the
segmented bundle is built on. Writing it after the log instead keeps both
runs identical.
THE MEMBERSHIP TEST IS LOAD-BEARING, and measured rather than assumed: the
two reprojections do not treat this line the same way. The per-directory
one drops every managed line before re-emitting its block, so the link is
gone by the time this runs. The flat one keeps a managed line whose target
is not an owned concept -- deliberately, because claiming somebody else's
link on the strength of a regex would delete curated content -- so `log.md`
survives there. Appending unconditionally therefore doubled the entry on
the second unsegmented run. Re-writing the line only when it is absent is
idempotent under both, without either side having to know about the other.
"""
index_path = bundle / profile.index.name
if not index_path.is_file():
return
body = index_path.read_text(encoding="utf-8")
link = profile.index.render_link(LOG_TITLE, LOG_NAME) + "\n"
if link in body.splitlines(keepends=True):
return
index_path.write_text(body + link, encoding="utf-8", newline="")
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="")
link_log_in_root_index(bundle, profile)
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())

2
src/llm_ingestion_okf/inbox.py
View file

@ -338,7 +338,7 @@ def walk_inbox(
) -> tuple[tuple[Path, ...], tuple[SkippedPath, ...]]:
"""Every dropped file at any depth, sorted by relative path, plus the skips.
ONE implementation, shared with `tools/okf_corpus_run.py`: the corpus
ONE implementation, shared with the corpus harness: the corpus
measurement counts the denominator N, and a walk that disagreed with the
door's would report a count over a different set of files than the one that
was ingested.

764
src/llm_ingestion_okf/propose.py Executable file
View file

@ -0,0 +1,764 @@
#!/usr/bin/env python3
"""Propose a segmentation plan for one document. A human adjudicates it.
Pipeline step 3. It lives in the package because `okf build` has to reach it
from an INSTALLED copy, where `tools/` does not exist -- it was outside `src/`
until then, and the reason it could move is that the reason it sat outside was
never about dependencies: every rule below is mechanical, so nothing here adds
a model call to a package that promises none.
What DID have to survive the move is the separation the old location expressed
physically: the split of a document into units of knowledge is a judgement, and
the run path replays a decision somebody already made. That separation is
carried by `adjudicated: false` and the `PROPOSED` marker, which is where it
belonged all along -- a directory boundary cannot enforce it, and a caller who
ingests a proposal unadjudicated could always do so.
## What the research says this tool may and may not claim
Topic 2 measured the OKF reference agent's granularity criteria against
`_okf-canonical`: it splits on **what a thing is**, not on layout, and makes
"multiple `write_concept_doc` calls ... rather than dumping everything into one
doc". Four of its gates are semantic and need a model. A handful of MECHANICAL
rules port today, and those are the ones below.
Topic 1b measured heading derivation on the K2 corpus: 11 of 11 prose headings
recovered -- from ONE document. 23 of 33 PDFs carry no outline at all and 95 %
of the outline entries that do exist are AutoCAD export metadata. The
denominator is 1. A rule validated on n=1 is not validated, and this tool says
so by marking every entry it emits `PROPOSED` rather than adjudicated.
Topic 1a measured that the best deterministic heading rule from poppler is a
CONJUNCTION -- `size AND bold`, via `-fontfullname` -- at recall 1.000 and
precision 0.846, and that adding weight as a DISJUNCT makes precision worse
(0.786 -> 0.524). That path is implemented here and nowhere else: poppler is a
SYSTEM binary the `[extract]` extra cannot express, so it may never be on the
run path or in a golden fixture.
## The one rule that is not a heuristic
**Nothing here is ever adjudicated.** `adjudicated: false` sits at the top of
every artifact and `PROPOSED` in every entry's `derived` list. A plan is
replayed deterministically and forever by the run path, so a proposal that
could pass for an adjudication would put a machine's guess where a human's
judgement is supposed to be, permanently and silently.
Stdlib only. No network: the model-backed path this tool deliberately does not
have would need the per-run network opt-in, and the socket-free test suite
proves the absence rather than assuming it.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import re
import sys
import unicodedata
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from .errors import IngestError
from .extract import extract_text
from .materialize import reduce_to_id_grammar
from .segmentation import observed_extractor_version
#: Stamped into every entry's `derived` list. The marker is what keeps a
#: proposal from being mistaken for the judgement the run path replays.
PROPOSED_MARKER = "PROPOSED"
#: This tool's identity, written into the artifact so an operator reading a
#: plan six months later can tell what produced it.
PROPOSER_ID = "okf-propose-segments"
PROPOSER_VERSION = "1"
#: The rules that survived Topic 2's port test. Each entry names exactly one,
#: so a proposal an operator disagrees with is traceable to the rule that made
#: it rather than to the tool as a whole.
RULE_HEADING = "rule:heading"
RULE_TABLE_BLOCK = "rule:table-block"
RULE_POPPLER_SIZE_AND_BOLD = "rule:poppler-size-and-bold"
#: Arm C only. NOT one of Topic 2's ported rules and not a heading rule at
#: all: it names the fact that a span was cut because it was too long, which
#: is a judgement about SIZE and says nothing about where a unit of knowledge
#: begins. It is emitted ALONGSIDE the rule that proposed the origin span, so
#: an operator reading a part can still see what opened it.
RULE_SIZE_SPLIT = "rule:size-split"
#: Arm D only. Like Arm C it is NOT one of Topic 2's ported rules and NOT
#: defined upstream: `docs/2026-09-02-k3-k4-k5-metode.md` contains no
#: occurrence of the word "arm" at all, so this definition was written for the
#: brief of order 20260906T213322Z and is reported as the author's. Unlike Arm
#: C it says nothing about size -- it names the fact that the DOCUMENT ITSELF
#: declared a chapter there, by numbering it in an ascending run its own
#: outline sustains.
RULE_OUTLINE = "rule:outline"
RULE_NAMES = (
RULE_HEADING,
RULE_TABLE_BLOCK,
RULE_POPPLER_SIZE_AND_BOLD,
RULE_SIZE_SPLIT,
RULE_OUTLINE,
)
#: How many characters of context each side of a quote anchor carries. Enough
#: to separate two occurrences of a repeated heading, short enough that an
#: edit NEAR a segment does not invalidate the anchor FOR it -- the anchor
#: exists to survive shifts, so making it fragile would defeat it.
ANCHOR_CONTEXT = 48
#: Norwegian and English function words. A heading made only of these names no
#: unit of knowledge -- it is a connective that happened to sit on its own line.
#: Topic 2's stop-word gate, and the only place this tool judges wording.
STOP_WORDS = frozenset(
{
"and",
"as",
"at",
"av",
"be",
"by",
"da",
"de",
"den",
"der",
"det",
"en",
"er",
"et",
"for",
"fra",
"i",
"in",
"is",
"it",
"med",
"of",
"og",
"om",
"on",
"or",
"over",
"paa",
"som",
"til",
"the",
"to",
"under",
"ved",
"with",
}
)
# An ATX heading, or a numbered section opening a line (`3.1 Brannkonsept`).
# A BARE integer is not a section number, for the same reason `structure.py`
# refuses one: `12 ting` is an ordinary line and admitting it would cut a
# document at every list item.
#
# That claim still holds, and Arm D does not weaken it. `_OUTLINE` below admits
# a bare integer ONLY inside an ascending run the document sustains for at
# least a declared length -- which is a property of the whole text, not of the
# line -- and the rule is off unless a caller asks for it. An UNGATED widening
# was measured and rejected: 1681 raw hits against 618 candidates, admitting
# list items, quantities and page furniture. The gate is what makes the signal
# a signal.
_ATX = re.compile(r"^(?P<hashes>#{1,6})\s+(?P<title>\S.*?)\s*$")
_NUMBERED = re.compile(r"^(?P<number>\d+(?:\.\d+)+)\s+(?P<title>\S.*?)\s*$")
_TABLE_ROW = re.compile(r"^\s*\|.*\|\s*$")
# Arm D's grammar. Integer-only BY CONSTRUCTION: `\s+` after the optional
# separator is what keeps `1.1 Brannkonsept` out, because `_NUMBERED` requires
# a dot and this requires whitespace, so no line can match both. No exclusion
# clause is written for that: a filter with a measured effect of zero is dead
# code that reads like a guard.
_OUTLINE = re.compile(r"^\s{0,4}(?P<number>\d{1,2})[.)]?\s+(?P<title>\S.*?)\s*$")
# A contents line carries the page it points at (`Innledning 6`). Measured on
# the K2 corpus: stripping it changes 0 of the 144 outline counts and 9 emitted
# titles. It is load-bearing anyway, because titles become concept paths
# through `_segment_path` -- an unstripped page number would become part of a
# filename.
_TRAILING_PAGE_NUMBER = re.compile(r"[\s.]+\d{1,4}\s*$")
class ProposerError(Exception):
"""The run failed. NOT 'nothing to propose' -- the two must stay distinct."""
@dataclass(frozen=True)
class Candidate:
"""One proposed boundary, before it becomes an entry."""
title: str
level: int
number: str | None
rule: str
start: int
end: int
#: True when this candidate is one PART of a longer span that Arm C cut.
#: Kept on the candidate rather than recomputed at write time so the entry
#: and the reason it exists cannot drift apart.
split: bool = False
def _is_stop_word_only(title: str) -> bool:
words = [word for word in re.split(r"[^\w]+", title.lower()) if word]
return bool(words) and all(word in STOP_WORDS for word in words)
def _strip_page_number(title: str) -> str:
"""Remove a trailing page number from a contents-listing title.
Deliberately NOT applied to a title that is only digits: `477` has no
separator before the number, so the pattern cannot match it and the title
survives for the stop-word and junk paths to see. Emptying it would fall
back to the `seksjon` stem and dress junk as a named section.
"""
return _TRAILING_PAGE_NUMBER.sub("", title)
def outline_lines(text: str) -> list[tuple[int, int, str]]:
"""Every line the outline grammar admits, as `(line index, integer, title)`.
Module level and importable on purpose: the reach instrument measures this
rule, and an instrument that re-implements the grammar it measures is
measuring a second definition that can silently drift from the shipped one.
"""
found: list[tuple[int, int, str]] = []
for index, line in enumerate(text.splitlines()):
match = _OUTLINE.match(line)
if match is None:
continue
title = _strip_page_number(match.group("title")).strip()
if not title or _is_stop_word_only(title):
continue
found.append((index, int(match.group("number")), title))
return found
def outline_runs(
entries: list[tuple[int, int, str]], minimum: int
) -> list[list[tuple[int, int, str]]]:
"""The maximal ascending runs among `entries`, each at least `minimum` long.
A run is anchored at `1` and every later member is its predecessor plus
one; a number that is neither is skipped without closing the run, so a
stray page number between two chapters does not truncate the outline. A new
`1` closes the current run and opens the next, which is what makes a
contents listing and the body it lists two runs rather than one.
Returned in document order. The CALLER chooses among them -- last-run
selection was measured against the alternatives and is stated where it is
applied, not hidden in here.
"""
runs: list[list[tuple[int, int, str]]] = []
current: list[tuple[int, int, str]] = []
for entry in entries:
number = entry[1]
if number == 1:
if current:
runs.append(current)
current = [entry]
elif current and number == current[-1][1] + 1:
current.append(entry)
if current:
runs.append(current)
return [run for run in runs if len(run) >= minimum]
def find_candidates(text: str, *, outline_run: int = 0) -> list[Candidate]:
"""Every boundary the mechanical rules propose, in document order.
Two gates from Topic 2 are applied here and both REMOVE candidates:
- the **stop-word gate**: a heading made only of function words is not a
unit of knowledge;
- the **orphan check**: a heading with no body under it proposes nothing,
because an empty concept is the silent skip this library refuses
everywhere else.
`outline_run` is Arm D's gate and it is OFF at 0: the function then behaves
exactly as it did before the rule existed. At `N >= 1` the document's own
numbered outline contributes boundaries where the integers sustain an
ascending run of at least `N`.
"""
lines = text.splitlines(keepends=True)
offsets: list[int] = []
position = 0
for line in lines:
offsets.append(position)
position += len(line)
end_of_text = position
# Computed BEFORE the loop, and that is a correctness requirement rather
# than a style choice: run selection is a whole-text decision (the LAST
# maximal run wins, because a contents listing precedes the body it lists),
# and a forward scan cannot know which run is last. Deciding it up front is
# also what keeps `marked` sorted by construction -- appending outline
# candidates in a second pass would leave `end < start` on some spans, and
# `text[start:end]` is then `""`, so the orphan check DELETES them
# silently. Silent loss, not a raise: nothing would announce it.
admitted: dict[int, str] = {}
if outline_run > 0:
runs = outline_runs(outline_lines(text), outline_run)
if runs:
# LAST run, not longest and not first. Measured against both:
# first-run opens segments inside the table of contents on 14/39
# documents; longest-run differs on 5/39 with no measured reason to
# prefer it. "Later occurrence wins" states the document's own
# ordering rather than a property of this corpus.
admitted = {index: title for index, _, title in runs[-1]}
marked: list[tuple[int, Candidate]] = []
in_table = False
for index, line in enumerate(lines):
if _TABLE_ROW.match(line):
if not in_table:
in_table = True
marked.append(
(
index,
Candidate(
title=f"Tabell linje {index + 1}",
level=9,
number=None,
rule=RULE_TABLE_BLOCK,
start=offsets[index],
end=end_of_text,
),
)
)
continue
in_table = False
outline_title = admitted.get(index)
if outline_title is not None:
outline_match = _OUTLINE.match(line)
assert outline_match is not None, "an admitted index still matches the grammar"
marked.append(
(
index,
Candidate(
title=outline_title,
level=1,
number=outline_match.group("number"),
rule=RULE_OUTLINE,
start=offsets[index],
end=end_of_text,
),
)
)
continue
atx = _ATX.match(line)
numbered = _NUMBERED.match(line)
if atx is None and numbered is None:
continue
if atx is not None:
title = atx.group("title")
level = len(atx.group("hashes"))
inner = _NUMBERED.match(title)
number = inner.group("number") if inner else None
else:
assert numbered is not None
title = numbered.group("title")
number = numbered.group("number")
level = number.count(".") + 1
# The stop-word gate. Applied to the TITLE, after any section number
# has been split off, so `3.1 Og` is judged on `Og`.
if _is_stop_word_only(title):
continue
marked.append(
(
index,
Candidate(
title=title,
level=level,
number=number,
rule=RULE_HEADING,
start=offsets[index],
end=end_of_text,
),
)
)
candidates: list[Candidate] = []
for position_in_list, (_, candidate) in enumerate(marked):
following = marked[position_in_list + 1 :]
end = offsets[following[0][0]] if following else end_of_text
body = text[candidate.start : end]
# The orphan check: everything after the heading line itself.
if not body.splitlines()[1:] or not "".join(body.splitlines()[1:]).strip():
continue
candidates.append(
Candidate(
title=candidate.title,
level=candidate.level,
number=candidate.number,
rule=candidate.rule,
start=candidate.start,
end=end,
)
)
return candidates
def _cut_points(text: str, start: int, end: int, cap: int) -> list[int]:
"""Where to cut `text[start:end]` so no part exceeds `cap` characters.
The cut prefers a PARAGRAPH boundary (a blank line) inside the window, then
a line boundary, and only then cuts mid-line. The order is the whole
content of the rule: a cut that lands mid-sentence splits one unit of
knowledge for no reason other than arithmetic, and the K3 categories count
that as `too fine`. The last resort exists anyway, because a document whose
body is one unbroken line is exactly where a cap that quietly stopped
binding would be least defensible.
"""
cuts: list[int] = []
position = start
while end - position > cap:
window_end = position + cap
paragraph = text.rfind("\n\n", position, window_end)
if paragraph != -1:
cut = paragraph + 2
else:
line = text.rfind("\n", position, window_end)
cut = line + 1 if line != -1 else window_end
# rfind can only return an index at or after `position`, so every
# branch advances. The assertion states that rather than trusting it:
# a cut that did not advance would loop forever on a corpus run.
assert cut > position, f"cut {cut} did not advance past {position}"
cuts.append(cut)
position = cut
return cuts
def subdivide(text: str, candidates: list[Candidate], cap: int) -> list[Candidate]:
"""Arm C. Arm B's candidates, with every over-long span cut down to `cap`.
ARM C IS NOT DEFINED IN `docs/2026-09-02-k3-k4-k5-metode.md`; that file
contains no occurrence of the word. This definition was written for order
20260904T145630Z and is reported as the author's, not as a ratified one.
Two callers' cases, one rule. When Arm B found boundaries but a span still
runs long (a PDF whose headings are its table of contents, so the trailing
segment absorbs the body), the span is cut. When Arm B found NO boundary at
all, the whole document is that span -- which is the `no declared
structure` case § 10 names, and 23 of 33 PDFs in the K2 corpus are in it.
A document with no boundaries that is already under the cap proposes
NOTHING, exactly as Arm B does. Arm C fires on size; where size is not the
problem it has nothing to say, and a one-entry plan would only dress a
single concept in a plan file.
"""
if cap <= 0:
return candidates
if not candidates:
if len(text) <= cap:
return []
# The synthetic span. Its rule is the size rule alone, because no
# heading rule proposed it -- there was no heading.
candidates = [
Candidate(
title="Del",
level=1,
number=None,
rule=RULE_SIZE_SPLIT,
start=0,
end=len(text),
split=False,
)
]
unnumbered_parts = True
else:
unnumbered_parts = False
out: list[Candidate] = []
for candidate in candidates:
cuts = _cut_points(text, candidate.start, candidate.end, cap)
if not cuts:
out.append(candidate)
continue
edges = [candidate.start, *cuts, candidate.end]
for part, (start, end) in enumerate(zip(edges, edges[1:]), start=1):
if unnumbered_parts:
title = f"Del {part}"
else:
title = candidate.title if part == 1 else f"{candidate.title} (del {part})"
out.append(
Candidate(
title=title,
level=candidate.level,
number=candidate.number,
rule=candidate.rule,
start=start,
end=end,
split=True,
)
)
return out
def _segment_path(candidate: Candidate, taken: set[str], prefix: str = "") -> str:
title = unicodedata.normalize("NFC", candidate.title)
# The section number becomes the DIRECTORY, so leaving it in the stem too
# yields `3-1/3-1-brannkonsept.md` -- correct and unreadable.
if candidate.number and title.startswith(candidate.number):
title = title[len(candidate.number) :]
stem = reduce_to_id_grammar(title)
if not stem:
stem = "seksjon"
directory = reduce_to_id_grammar(candidate.number or "") if candidate.number else ""
# The caller's scope comes FIRST and is never deduplicated against: it is
# the same for every entry in this document by construction, and that is
# the whole point -- one document's sections must not be able to claim
# another's path.
head = f"{prefix}/" if prefix else ""
path = f"{head}{directory}/{stem}.md" if directory else f"{head}{stem}.md"
suffix = 2
while path in taken:
path = f"{head}{directory}/{stem}-{suffix}.md" if directory else f"{head}{stem}-{suffix}.md"
suffix += 1
taken.add(path)
return path
def build_plan(
source: Path,
text: str,
source_bytes: bytes,
*,
okf_type: str,
proposed_at: str,
path_prefix: str = "",
max_segment_chars: int = 0,
outline_run: int = 0,
) -> dict[str, Any]:
"""The artifact. Every entry PROPOSED, the plan itself never adjudicated."""
taken: set[str] = set()
extractor_id = source.suffix.lower().lstrip(".") or "none"
entries: list[dict[str, Any]] = []
candidates = find_candidates(text, outline_run=outline_run)
for candidate in subdivide(text, candidates, max_segment_chars):
entries.append(
{
"segment_id": f"p{len(entries) + 1}",
"path": _segment_path(candidate, taken, path_prefix),
"title": candidate.title,
"okf_type": okf_type,
"span": [candidate.start, candidate.end],
"ingested_at": proposed_at,
# The offsets are a hint the anchor may correct. Written at
# proposal time because that is the only moment the text the
# adjudicator will judge and the offsets naming it are known
# to agree -- reconstructing it later would anchor to whatever
# the extraction had already become.
"anchor": {
"quote": text[candidate.start : candidate.end],
"prefix": text[max(0, candidate.start - ANCHOR_CONTEXT) : candidate.start],
"suffix": text[candidate.end : candidate.end + ANCHOR_CONTEXT],
},
# PROPOSED first, then the rule that proposed it. `derived` is
# this library's existing "which of these did we infer" marker,
# so a consumer that already distrusts derived fields
# distrusts these by construction.
# PROPOSED first, then the rule that proposed the span, then
# -- for an Arm C part only -- the size rule that cut it. Two
# names rather than one on those entries: the heading rule is
# still what opened the span, and dropping it would make a part
# untraceable to anything but arithmetic.
"derived": (
[PROPOSED_MARKER, candidate.rule, RULE_SIZE_SPLIT]
if candidate.split and candidate.rule != RULE_SIZE_SPLIT
else [PROPOSED_MARKER, candidate.rule]
),
}
)
return {
"version": "1",
"source_sha256": hashlib.sha256(source_bytes).hexdigest(),
# The hash the offsets actually depend on. Source bytes alone cannot
# see a converter reshaping its output, so the staleness signal this
# plan is supposed to carry did not exist until this line did.
"text_sha256": hashlib.sha256(text.encode("utf-8")).hexdigest(),
"extractor_id": extractor_id,
# The EXTRACTOR's version, not this tool's. `PROPOSER_VERSION` sat here
# and named the wrong thing: a converter bump left the field frozen at
# the proposer's own number, so the component could not move.
"extractor_version": observed_extractor_version(extractor_id),
"adjudicated_at": proposed_at,
# NOT a timestamp question. `adjudicated_at` records when this artifact
# was produced; this records whether a human has looked at it, and it is
# false until one replaces the file.
"adjudicated": False,
"proposed_by": f"{PROPOSER_ID}/{PROPOSER_VERSION}",
"entries": entries,
}
def run(
source: Path,
out: Path,
*,
okf_type: str,
proposed_at: str,
path_prefix: str = "",
max_segment_chars: int = 0,
outline_run: int = 0,
) -> int:
if max_segment_chars < 0:
raise ProposerError(
f"--max-segment-chars {max_segment_chars} is negative; the cap is a "
"character count, and 0 means off (Arm B)"
)
if outline_run < 0:
raise ProposerError(
f"--outline-run {outline_run} is negative; the gate is a run LENGTH, "
"and 0 means off (Arm B)"
)
# Reduced HERE, before anything is read: a prefix that survives to the
# entries as an empty component would produce exactly the unscoped paths
# the caller asked to avoid, and would do it silently.
#
# PER COMPONENT, because the prefix carries a DIRECTORY now that Door B
# walks the inbox recursively and records a relative `source_file`.
# Reducing the whole string would fold `/` into a `-` and flatten
# `sub/sub2` into the single component `sub-sub2` -- a bundle shaped unlike
# the inbox it came from, and unlike what the caller wrote.
components = (
[reduce_to_id_grammar(part) for part in path_prefix.split("/")] if path_prefix else []
)
if path_prefix and not all(components):
raise ProposerError(
f"--path-prefix {path_prefix!r} has a component that reduces to nothing under "
"the id grammar ([a-z0-9][a-z0-9-]*); refusing to write unscoped paths under a "
"scope that was asked for"
)
scope = "/".join(components)
if not source.is_file():
raise ProposerError(f"source is not a file: {source}")
try:
source_bytes = source.read_bytes()
except OSError as exc:
raise ProposerError(f"cannot read {source}: {exc}") from exc
try:
text = extract_text(source.name, source_bytes)
except IngestError as exc:
raise ProposerError(f"cannot extract text from {source.name}: {exc}") from exc
payload = build_plan(
source,
text,
source_bytes,
okf_type=okf_type,
proposed_at=proposed_at,
path_prefix=scope,
max_segment_chars=max_segment_chars,
outline_run=outline_run,
)
# Nothing to propose is an OUTCOME, and it is not an artifact. An empty
# plan cannot be replayed -- `process_inbox` refuses one, because a plan
# naming no entry would persist nothing for a document that was dropped --
# so the only thing a zero-entry file can do is fail a run later. Its own
# exit status, distinct from 2, so a driver can tell "this document lands
# as one flat concept" from "stop".
if not payload["entries"]:
print(
f"{PROPOSER_ID}: nothing to propose for {source.name} — the mechanical "
"rules found no boundary. No artifact written; this document lands as "
"one concept unless someone segments it by hand.",
file=sys.stderr,
)
return 1
out.parent.mkdir(parents=True, exist_ok=True)
out.write_bytes((json.dumps(payload, indent=2, ensure_ascii=False) + "\n").encode("utf-8"))
print(
f"{PROPOSER_ID}: proposed {len(payload['entries'])} segment(s) -> {out}\n"
f"{PROPOSER_ID}: every entry is PROPOSED. Adjudicate before ingesting.",
file=sys.stderr,
)
return 0
def parse_args(argv: list[str] | None) -> argparse.Namespace:
parser = argparse.ArgumentParser(
prog=PROPOSER_ID,
description="Propose a segmentation plan. A human adjudicates it before use.",
)
parser.add_argument("source", type=Path, help="the document to segment")
parser.add_argument("--out", type=Path, required=True, help="where to write the artifact")
parser.add_argument("--okf-type", default="reference", help="okf_type for every entry")
parser.add_argument(
"--path-prefix",
default="",
help=(
"scope every entry's path under this directory, `/`-separated for a "
"nested one (each component is reduced on its own). Required for a corpus: "
"section numbering is document-local, so two documents propose the same "
"path and Door B refuses both. An argument rather than something this "
"tool derives -- it sees one document and cannot know what else is in "
"the bundle"
),
)
parser.add_argument(
"--max-segment-chars",
type=int,
default=0,
metavar="N",
help=(
"Arm C: cut any proposed span longer than N characters at the nearest "
"paragraph boundary, the whole document counting as one span when the "
"mechanical rules find no boundary at all. 0 (the default) is OFF and "
"leaves the artifact byte-identical to Arm B. Arm C is the author's "
"definition, written for order 20260904T145630Z; it is not defined in "
"the K3 method file"
),
)
parser.add_argument(
"--outline-run",
type=int,
default=0,
metavar="N",
help=(
"Arm D: also propose a boundary at each line of the document's own "
"numbered outline (the bare integers the heading grammar cannot "
"match, since it requires a dot), but only where those integers "
"sustain an ascending run of at least N entries, and only for the "
"LAST such run when the outline repeats, because a contents listing "
"precedes the body it lists. 0 (the default) is OFF and leaves the "
"artifact byte-identical to Arm B. Arm D is the author's definition, "
"written for order 20260906T213322Z; it is not defined upstream, and "
"the K3 method file does not name it either"
),
)
parser.add_argument(
"--proposed-at",
default="1970-01-01T00:00:00Z",
help="the timestamp written into the artifact; explicit so a run is reproducible",
)
return parser.parse_args(argv)
def main(argv: list[str] | None = None) -> int:
args = parse_args(argv)
try:
return run(
args.source,
args.out,
okf_type=args.okf_type,
proposed_at=args.proposed_at,
path_prefix=args.path_prefix,
max_segment_chars=args.max_segment_chars,
outline_run=args.outline_run,
)
except ProposerError as exc:
print(f"{PROPOSER_ID}: FAILED - {exc}", file=sys.stderr)
print(
f"{PROPOSER_ID}: this is NOT 'nothing to propose'. Nothing was written.",
file=sys.stderr,
)
return 2
if __name__ == "__main__":
raise SystemExit(main())

3
tests/test_adjudicate.py
View file

@ -40,7 +40,8 @@ from llm_ingestion_okf.segmentation import parse_segmentation_plan
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools"))
import okf_adjudicate # noqa: E402
import okf_propose_segments # noqa: E402
from llm_ingestion_okf import propose as okf_propose_segments # noqa: E402
DOCUMENT = """# N500 Vegbygging

401
tests/test_cli_build.py Normal file
View file

@ -0,0 +1,401 @@
"""`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:
"""Both arms stay off by default -- measured on the artifact, not the flag.
`rule:size-split` and `rule:outline` are the markers the two arms write
into a plan's `derived` list, so their absence is the arms being off.
"""
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
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")
# --- 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

10
tests/test_corpus_run.py
View file

@ -22,14 +22,11 @@ passes silently, every green run above it becomes meaningless.
from __future__ import annotations
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools"))
import okf_corpus_run # noqa: E402
from llm_ingestion_okf import corpus as okf_corpus_run
INGESTED_AT = "2026-07-25T12:00:00Z"
@ -248,10 +245,9 @@ SEGMENTABLE = (
def _propose(source: Path, out: Path) -> None:
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools"))
import okf_propose_segments
from llm_ingestion_okf import propose
assert okf_propose_segments.run(source, out, okf_type="reference", proposed_at=INGESTED_AT) == 0
assert propose.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(

3
tests/test_outline_measure.py
View file

@ -28,7 +28,8 @@ from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools"))
import okf_outline_measure # noqa: E402
import okf_propose_segments # noqa: E402
from llm_ingestion_okf import propose as okf_propose_segments # noqa: E402
OUTLINE_DOC = """1 Innledning

5
tests/test_propose_segments.py
View file

@ -21,16 +21,13 @@ from __future__ import annotations
import json
import socket
import sys
from pathlib import Path
import pytest
from llm_ingestion_okf.segmentation import parse_segmentation_plan
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools"))
import okf_propose_segments # noqa: E402
from llm_ingestion_okf import propose as okf_propose_segments
DOCUMENT = """# N500 Vegbygging

494
tools/okf_corpus_run.py
View file

@ -1,497 +1,25 @@
"""Run a corpus through the whole path and report numbers, never a claim.
#!/usr/bin/env python3
"""Thin entry point. The implementation is `llm_ingestion_okf.corpus`.
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.
It moved into the package when `okf build` was packaged: the command runs the
same harness, and an installed copy cannot import this directory. This file
stays because the published reproduction blocks in
`docs/2026-09-03-k2-bundle-rebuild.md` and `docs/2026-09-04-k3-arm-c.md` name
it, and a measurement whose command no longer runs is a measurement nobody can
repeat.
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.
No logic here, deliberately: a second copy of the conservation check is a
second thing that can be right while the shipped one is wrong.
"""
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
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 ( # noqa: E402
GateDecision,
InboxResult,
process_inbox,
relative_source,
walk_inbox,
)
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",
"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 llm_ingestion_okf._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,
) -> 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,
)
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 "
"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)
def link_log_in_root_index(bundle: Path, profile: BundleProfile) -> None:
"""Point the root index at the log, so the walk section 8 supports reaches it.
Measured on the K2 artifact: the bundle carried a conformant root `log.md`
that no index named, so a consumer entering at `index.md` never reached the
one file carrying `N`.
A LOCAL choice, not a conformance requirement, and the difference is worth
stating rather than implying. Section 9 puts `log.md` at any level and
section 8 has an index enumerate its directory's contents, but upstream's
own bundles do not link it: measured at `9a15b13`, 0 of the 24 shipped
`index.md` files name the single `log.md` in the set. Upstream therefore
shows the link is not REQUIRED -- not that it is disallowed.
It belongs to the harness and not the library. The log's content IS the
run's outcome, so it cannot exist when the indexes are projected; an index
that enumerated it off the directory would gain the link only from the
second run onward and break rebuild-equals-incremental, the property the
segmented bundle is built on. Writing it after the log instead keeps both
runs identical.
THE MEMBERSHIP TEST IS LOAD-BEARING, and measured rather than assumed: the
two reprojections do not treat this line the same way. The per-directory
one drops every managed line before re-emitting its block, so the link is
gone by the time this runs. The flat one keeps a managed line whose target
is not an owned concept -- deliberately, because claiming somebody else's
link on the strength of a regex would delete curated content -- so `log.md`
survives there. Appending unconditionally therefore doubled the entry on
the second unsegmented run. Re-writing the line only when it is absent is
idempotent under both, without either side having to know about the other.
"""
index_path = bundle / profile.index.name
if not index_path.is_file():
return
body = index_path.read_text(encoding="utf-8")
link = profile.index.render_link(LOG_TITLE, LOG_NAME) + "\n"
if link in body.splitlines(keepends=True):
return
index_path.write_text(body + link, encoding="utf-8", newline="")
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="")
link_log_in_root_index(bundle, profile)
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
from llm_ingestion_okf.corpus import main # noqa: E402
if __name__ == "__main__":
raise SystemExit(main())

3
tools/okf_outline_measure.py
View file

@ -45,12 +45,11 @@ from dataclasses import dataclass, field
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
sys.path.insert(0, str(Path(__file__).resolve().parent))
from llm_ingestion_okf.errors import ExtractionError # noqa: E402
from llm_ingestion_okf.extract import extract_text # noqa: E402
from llm_ingestion_okf.materialize import reduce_to_id_grammar # noqa: E402
from okf_propose_segments import ( # noqa: E402
from llm_ingestion_okf.propose import ( # noqa: E402
RULE_OUTLINE,
Candidate,
_segment_path,

753
tools/okf_propose_segments.py Executable file → Normal file
View file

@ -1,758 +1,25 @@
#!/usr/bin/env python3
"""Propose a segmentation plan for one document. A human adjudicates it.
"""Thin entry point. The implementation is `llm_ingestion_okf.propose`.
Pipeline step 3, and deliberately OUTSIDE the package. `src/` promises zero
model calls on the run path, and the split of a document into units of
knowledge is a judgement. Keeping the judgement lane out here is what lets the
run path stay a deterministic replay of a decision somebody already made.
It moved into the package when `okf build` was packaged: the command has to
reach the proposer from an installed copy, where this directory does not
exist. This file stays because the published reproduction blocks in
`docs/2026-09-03-k2-bundle-rebuild.md` and `docs/2026-09-04-k3-arm-c.md` name
it, and a measurement whose command no longer runs is a measurement nobody can
repeat.
## What the research says this tool may and may not claim
Topic 2 measured the OKF reference agent's granularity criteria against
`_okf-canonical`: it splits on **what a thing is**, not on layout, and makes
"multiple `write_concept_doc` calls ... rather than dumping everything into one
doc". Four of its gates are semantic and need a model. A handful of MECHANICAL
rules port today, and those are the ones below.
Topic 1b measured heading derivation on the K2 corpus: 11 of 11 prose headings
recovered -- from ONE document. 23 of 33 PDFs carry no outline at all and 95 %
of the outline entries that do exist are AutoCAD export metadata. The
denominator is 1. A rule validated on n=1 is not validated, and this tool says
so by marking every entry it emits `PROPOSED` rather than adjudicated.
Topic 1a measured that the best deterministic heading rule from poppler is a
CONJUNCTION -- `size AND bold`, via `-fontfullname` -- at recall 1.000 and
precision 0.846, and that adding weight as a DISJUNCT makes precision worse
(0.786 -> 0.524). That path is implemented here and nowhere else: poppler is a
SYSTEM binary the `[extract]` extra cannot express, so it may never be on the
run path or in a golden fixture.
## The one rule that is not a heuristic
**Nothing here is ever adjudicated.** `adjudicated: false` sits at the top of
every artifact and `PROPOSED` in every entry's `derived` list. A plan is
replayed deterministically and forever by the run path, so a proposal that
could pass for an adjudication would put a machine's guess where a human's
judgement is supposed to be, permanently and silently.
Stdlib only. No network: the model-backed path this tool deliberately does not
have would need the per-run network opt-in, and the socket-free test suite
proves the absence rather than assuming it.
No logic here, deliberately: two copies of a proposal rule would let the
published path and the packaged one drift apart while both stayed green.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import re
import sys
import unicodedata
from dataclasses import dataclass
from pathlib import Path
from typing import Any
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.materialize import reduce_to_id_grammar # noqa: E402
from llm_ingestion_okf.segmentation import observed_extractor_version # noqa: E402
#: Stamped into every entry's `derived` list. The marker is what keeps a
#: proposal from being mistaken for the judgement the run path replays.
PROPOSED_MARKER = "PROPOSED"
#: This tool's identity, written into the artifact so an operator reading a
#: plan six months later can tell what produced it.
PROPOSER_ID = "okf-propose-segments"
PROPOSER_VERSION = "1"
#: The rules that survived Topic 2's port test. Each entry names exactly one,
#: so a proposal an operator disagrees with is traceable to the rule that made
#: it rather than to the tool as a whole.
RULE_HEADING = "rule:heading"
RULE_TABLE_BLOCK = "rule:table-block"
RULE_POPPLER_SIZE_AND_BOLD = "rule:poppler-size-and-bold"
#: Arm C only. NOT one of Topic 2's ported rules and not a heading rule at
#: all: it names the fact that a span was cut because it was too long, which
#: is a judgement about SIZE and says nothing about where a unit of knowledge
#: begins. It is emitted ALONGSIDE the rule that proposed the origin span, so
#: an operator reading a part can still see what opened it.
RULE_SIZE_SPLIT = "rule:size-split"
#: Arm D only. Like Arm C it is NOT one of Topic 2's ported rules and NOT
#: defined upstream: `docs/2026-09-02-k3-k4-k5-metode.md` contains no
#: occurrence of the word "arm" at all, so this definition was written for the
#: brief of order 20260906T213322Z and is reported as the author's. Unlike Arm
#: C it says nothing about size -- it names the fact that the DOCUMENT ITSELF
#: declared a chapter there, by numbering it in an ascending run its own
#: outline sustains.
RULE_OUTLINE = "rule:outline"
RULE_NAMES = (
RULE_HEADING,
RULE_TABLE_BLOCK,
RULE_POPPLER_SIZE_AND_BOLD,
RULE_SIZE_SPLIT,
RULE_OUTLINE,
)
#: How many characters of context each side of a quote anchor carries. Enough
#: to separate two occurrences of a repeated heading, short enough that an
#: edit NEAR a segment does not invalidate the anchor FOR it -- the anchor
#: exists to survive shifts, so making it fragile would defeat it.
ANCHOR_CONTEXT = 48
#: Norwegian and English function words. A heading made only of these names no
#: unit of knowledge -- it is a connective that happened to sit on its own line.
#: Topic 2's stop-word gate, and the only place this tool judges wording.
STOP_WORDS = frozenset(
{
"and",
"as",
"at",
"av",
"be",
"by",
"da",
"de",
"den",
"der",
"det",
"en",
"er",
"et",
"for",
"fra",
"i",
"in",
"is",
"it",
"med",
"of",
"og",
"om",
"on",
"or",
"over",
"paa",
"som",
"til",
"the",
"to",
"under",
"ved",
"with",
}
)
# An ATX heading, or a numbered section opening a line (`3.1 Brannkonsept`).
# A BARE integer is not a section number, for the same reason `structure.py`
# refuses one: `12 ting` is an ordinary line and admitting it would cut a
# document at every list item.
#
# That claim still holds, and Arm D does not weaken it. `_OUTLINE` below admits
# a bare integer ONLY inside an ascending run the document sustains for at
# least a declared length -- which is a property of the whole text, not of the
# line -- and the rule is off unless a caller asks for it. An UNGATED widening
# was measured and rejected: 1681 raw hits against 618 candidates, admitting
# list items, quantities and page furniture. The gate is what makes the signal
# a signal.
_ATX = re.compile(r"^(?P<hashes>#{1,6})\s+(?P<title>\S.*?)\s*$")
_NUMBERED = re.compile(r"^(?P<number>\d+(?:\.\d+)+)\s+(?P<title>\S.*?)\s*$")
_TABLE_ROW = re.compile(r"^\s*\|.*\|\s*$")
# Arm D's grammar. Integer-only BY CONSTRUCTION: `\s+` after the optional
# separator is what keeps `1.1 Brannkonsept` out, because `_NUMBERED` requires
# a dot and this requires whitespace, so no line can match both. No exclusion
# clause is written for that: a filter with a measured effect of zero is dead
# code that reads like a guard.
_OUTLINE = re.compile(r"^\s{0,4}(?P<number>\d{1,2})[.)]?\s+(?P<title>\S.*?)\s*$")
# A contents line carries the page it points at (`Innledning 6`). Measured on
# the K2 corpus: stripping it changes 0 of the 144 outline counts and 9 emitted
# titles. It is load-bearing anyway, because titles become concept paths
# through `_segment_path` -- an unstripped page number would become part of a
# filename.
_TRAILING_PAGE_NUMBER = re.compile(r"[\s.]+\d{1,4}\s*$")
class ProposerError(Exception):
"""The run failed. NOT 'nothing to propose' -- the two must stay distinct."""
@dataclass(frozen=True)
class Candidate:
"""One proposed boundary, before it becomes an entry."""
title: str
level: int
number: str | None
rule: str
start: int
end: int
#: True when this candidate is one PART of a longer span that Arm C cut.
#: Kept on the candidate rather than recomputed at write time so the entry
#: and the reason it exists cannot drift apart.
split: bool = False
def _is_stop_word_only(title: str) -> bool:
words = [word for word in re.split(r"[^\w]+", title.lower()) if word]
return bool(words) and all(word in STOP_WORDS for word in words)
def _strip_page_number(title: str) -> str:
"""Remove a trailing page number from a contents-listing title.
Deliberately NOT applied to a title that is only digits: `477` has no
separator before the number, so the pattern cannot match it and the title
survives for the stop-word and junk paths to see. Emptying it would fall
back to the `seksjon` stem and dress junk as a named section.
"""
return _TRAILING_PAGE_NUMBER.sub("", title)
def outline_lines(text: str) -> list[tuple[int, int, str]]:
"""Every line the outline grammar admits, as `(line index, integer, title)`.
Module level and importable on purpose: the reach instrument measures this
rule, and an instrument that re-implements the grammar it measures is
measuring a second definition that can silently drift from the shipped one.
"""
found: list[tuple[int, int, str]] = []
for index, line in enumerate(text.splitlines()):
match = _OUTLINE.match(line)
if match is None:
continue
title = _strip_page_number(match.group("title")).strip()
if not title or _is_stop_word_only(title):
continue
found.append((index, int(match.group("number")), title))
return found
def outline_runs(
entries: list[tuple[int, int, str]], minimum: int
) -> list[list[tuple[int, int, str]]]:
"""The maximal ascending runs among `entries`, each at least `minimum` long.
A run is anchored at `1` and every later member is its predecessor plus
one; a number that is neither is skipped without closing the run, so a
stray page number between two chapters does not truncate the outline. A new
`1` closes the current run and opens the next, which is what makes a
contents listing and the body it lists two runs rather than one.
Returned in document order. The CALLER chooses among them -- last-run
selection was measured against the alternatives and is stated where it is
applied, not hidden in here.
"""
runs: list[list[tuple[int, int, str]]] = []
current: list[tuple[int, int, str]] = []
for entry in entries:
number = entry[1]
if number == 1:
if current:
runs.append(current)
current = [entry]
elif current and number == current[-1][1] + 1:
current.append(entry)
if current:
runs.append(current)
return [run for run in runs if len(run) >= minimum]
def find_candidates(text: str, *, outline_run: int = 0) -> list[Candidate]:
"""Every boundary the mechanical rules propose, in document order.
Two gates from Topic 2 are applied here and both REMOVE candidates:
- the **stop-word gate**: a heading made only of function words is not a
unit of knowledge;
- the **orphan check**: a heading with no body under it proposes nothing,
because an empty concept is the silent skip this library refuses
everywhere else.
`outline_run` is Arm D's gate and it is OFF at 0: the function then behaves
exactly as it did before the rule existed. At `N >= 1` the document's own
numbered outline contributes boundaries where the integers sustain an
ascending run of at least `N`.
"""
lines = text.splitlines(keepends=True)
offsets: list[int] = []
position = 0
for line in lines:
offsets.append(position)
position += len(line)
end_of_text = position
# Computed BEFORE the loop, and that is a correctness requirement rather
# than a style choice: run selection is a whole-text decision (the LAST
# maximal run wins, because a contents listing precedes the body it lists),
# and a forward scan cannot know which run is last. Deciding it up front is
# also what keeps `marked` sorted by construction -- appending outline
# candidates in a second pass would leave `end < start` on some spans, and
# `text[start:end]` is then `""`, so the orphan check DELETES them
# silently. Silent loss, not a raise: nothing would announce it.
admitted: dict[int, str] = {}
if outline_run > 0:
runs = outline_runs(outline_lines(text), outline_run)
if runs:
# LAST run, not longest and not first. Measured against both:
# first-run opens segments inside the table of contents on 14/39
# documents; longest-run differs on 5/39 with no measured reason to
# prefer it. "Later occurrence wins" states the document's own
# ordering rather than a property of this corpus.
admitted = {index: title for index, _, title in runs[-1]}
marked: list[tuple[int, Candidate]] = []
in_table = False
for index, line in enumerate(lines):
if _TABLE_ROW.match(line):
if not in_table:
in_table = True
marked.append(
(
index,
Candidate(
title=f"Tabell linje {index + 1}",
level=9,
number=None,
rule=RULE_TABLE_BLOCK,
start=offsets[index],
end=end_of_text,
),
)
)
continue
in_table = False
outline_title = admitted.get(index)
if outline_title is not None:
outline_match = _OUTLINE.match(line)
assert outline_match is not None, "an admitted index still matches the grammar"
marked.append(
(
index,
Candidate(
title=outline_title,
level=1,
number=outline_match.group("number"),
rule=RULE_OUTLINE,
start=offsets[index],
end=end_of_text,
),
)
)
continue
atx = _ATX.match(line)
numbered = _NUMBERED.match(line)
if atx is None and numbered is None:
continue
if atx is not None:
title = atx.group("title")
level = len(atx.group("hashes"))
inner = _NUMBERED.match(title)
number = inner.group("number") if inner else None
else:
assert numbered is not None
title = numbered.group("title")
number = numbered.group("number")
level = number.count(".") + 1
# The stop-word gate. Applied to the TITLE, after any section number
# has been split off, so `3.1 Og` is judged on `Og`.
if _is_stop_word_only(title):
continue
marked.append(
(
index,
Candidate(
title=title,
level=level,
number=number,
rule=RULE_HEADING,
start=offsets[index],
end=end_of_text,
),
)
)
candidates: list[Candidate] = []
for position_in_list, (_, candidate) in enumerate(marked):
following = marked[position_in_list + 1 :]
end = offsets[following[0][0]] if following else end_of_text
body = text[candidate.start : end]
# The orphan check: everything after the heading line itself.
if not body.splitlines()[1:] or not "".join(body.splitlines()[1:]).strip():
continue
candidates.append(
Candidate(
title=candidate.title,
level=candidate.level,
number=candidate.number,
rule=candidate.rule,
start=candidate.start,
end=end,
)
)
return candidates
def _cut_points(text: str, start: int, end: int, cap: int) -> list[int]:
"""Where to cut `text[start:end]` so no part exceeds `cap` characters.
The cut prefers a PARAGRAPH boundary (a blank line) inside the window, then
a line boundary, and only then cuts mid-line. The order is the whole
content of the rule: a cut that lands mid-sentence splits one unit of
knowledge for no reason other than arithmetic, and the K3 categories count
that as `too fine`. The last resort exists anyway, because a document whose
body is one unbroken line is exactly where a cap that quietly stopped
binding would be least defensible.
"""
cuts: list[int] = []
position = start
while end - position > cap:
window_end = position + cap
paragraph = text.rfind("\n\n", position, window_end)
if paragraph != -1:
cut = paragraph + 2
else:
line = text.rfind("\n", position, window_end)
cut = line + 1 if line != -1 else window_end
# rfind can only return an index at or after `position`, so every
# branch advances. The assertion states that rather than trusting it:
# a cut that did not advance would loop forever on a corpus run.
assert cut > position, f"cut {cut} did not advance past {position}"
cuts.append(cut)
position = cut
return cuts
def subdivide(text: str, candidates: list[Candidate], cap: int) -> list[Candidate]:
"""Arm C. Arm B's candidates, with every over-long span cut down to `cap`.
ARM C IS NOT DEFINED IN `docs/2026-09-02-k3-k4-k5-metode.md`; that file
contains no occurrence of the word. This definition was written for order
20260904T145630Z and is reported as the author's, not as a ratified one.
Two callers' cases, one rule. When Arm B found boundaries but a span still
runs long (a PDF whose headings are its table of contents, so the trailing
segment absorbs the body), the span is cut. When Arm B found NO boundary at
all, the whole document is that span -- which is the `no declared
structure` case § 10 names, and 23 of 33 PDFs in the K2 corpus are in it.
A document with no boundaries that is already under the cap proposes
NOTHING, exactly as Arm B does. Arm C fires on size; where size is not the
problem it has nothing to say, and a one-entry plan would only dress a
single concept in a plan file.
"""
if cap <= 0:
return candidates
if not candidates:
if len(text) <= cap:
return []
# The synthetic span. Its rule is the size rule alone, because no
# heading rule proposed it -- there was no heading.
candidates = [
Candidate(
title="Del",
level=1,
number=None,
rule=RULE_SIZE_SPLIT,
start=0,
end=len(text),
split=False,
)
]
unnumbered_parts = True
else:
unnumbered_parts = False
out: list[Candidate] = []
for candidate in candidates:
cuts = _cut_points(text, candidate.start, candidate.end, cap)
if not cuts:
out.append(candidate)
continue
edges = [candidate.start, *cuts, candidate.end]
for part, (start, end) in enumerate(zip(edges, edges[1:]), start=1):
if unnumbered_parts:
title = f"Del {part}"
else:
title = candidate.title if part == 1 else f"{candidate.title} (del {part})"
out.append(
Candidate(
title=title,
level=candidate.level,
number=candidate.number,
rule=candidate.rule,
start=start,
end=end,
split=True,
)
)
return out
def _segment_path(candidate: Candidate, taken: set[str], prefix: str = "") -> str:
title = unicodedata.normalize("NFC", candidate.title)
# The section number becomes the DIRECTORY, so leaving it in the stem too
# yields `3-1/3-1-brannkonsept.md` -- correct and unreadable.
if candidate.number and title.startswith(candidate.number):
title = title[len(candidate.number) :]
stem = reduce_to_id_grammar(title)
if not stem:
stem = "seksjon"
directory = reduce_to_id_grammar(candidate.number or "") if candidate.number else ""
# The caller's scope comes FIRST and is never deduplicated against: it is
# the same for every entry in this document by construction, and that is
# the whole point -- one document's sections must not be able to claim
# another's path.
head = f"{prefix}/" if prefix else ""
path = f"{head}{directory}/{stem}.md" if directory else f"{head}{stem}.md"
suffix = 2
while path in taken:
path = f"{head}{directory}/{stem}-{suffix}.md" if directory else f"{head}{stem}-{suffix}.md"
suffix += 1
taken.add(path)
return path
def build_plan(
source: Path,
text: str,
source_bytes: bytes,
*,
okf_type: str,
proposed_at: str,
path_prefix: str = "",
max_segment_chars: int = 0,
outline_run: int = 0,
) -> dict[str, Any]:
"""The artifact. Every entry PROPOSED, the plan itself never adjudicated."""
taken: set[str] = set()
extractor_id = source.suffix.lower().lstrip(".") or "none"
entries: list[dict[str, Any]] = []
candidates = find_candidates(text, outline_run=outline_run)
for candidate in subdivide(text, candidates, max_segment_chars):
entries.append(
{
"segment_id": f"p{len(entries) + 1}",
"path": _segment_path(candidate, taken, path_prefix),
"title": candidate.title,
"okf_type": okf_type,
"span": [candidate.start, candidate.end],
"ingested_at": proposed_at,
# The offsets are a hint the anchor may correct. Written at
# proposal time because that is the only moment the text the
# adjudicator will judge and the offsets naming it are known
# to agree -- reconstructing it later would anchor to whatever
# the extraction had already become.
"anchor": {
"quote": text[candidate.start : candidate.end],
"prefix": text[max(0, candidate.start - ANCHOR_CONTEXT) : candidate.start],
"suffix": text[candidate.end : candidate.end + ANCHOR_CONTEXT],
},
# PROPOSED first, then the rule that proposed it. `derived` is
# this library's existing "which of these did we infer" marker,
# so a consumer that already distrusts derived fields
# distrusts these by construction.
# PROPOSED first, then the rule that proposed the span, then
# -- for an Arm C part only -- the size rule that cut it. Two
# names rather than one on those entries: the heading rule is
# still what opened the span, and dropping it would make a part
# untraceable to anything but arithmetic.
"derived": (
[PROPOSED_MARKER, candidate.rule, RULE_SIZE_SPLIT]
if candidate.split and candidate.rule != RULE_SIZE_SPLIT
else [PROPOSED_MARKER, candidate.rule]
),
}
)
return {
"version": "1",
"source_sha256": hashlib.sha256(source_bytes).hexdigest(),
# The hash the offsets actually depend on. Source bytes alone cannot
# see a converter reshaping its output, so the staleness signal this
# plan is supposed to carry did not exist until this line did.
"text_sha256": hashlib.sha256(text.encode("utf-8")).hexdigest(),
"extractor_id": extractor_id,
# The EXTRACTOR's version, not this tool's. `PROPOSER_VERSION` sat here
# and named the wrong thing: a converter bump left the field frozen at
# the proposer's own number, so the component could not move.
"extractor_version": observed_extractor_version(extractor_id),
"adjudicated_at": proposed_at,
# NOT a timestamp question. `adjudicated_at` records when this artifact
# was produced; this records whether a human has looked at it, and it is
# false until one replaces the file.
"adjudicated": False,
"proposed_by": f"{PROPOSER_ID}/{PROPOSER_VERSION}",
"entries": entries,
}
def run(
source: Path,
out: Path,
*,
okf_type: str,
proposed_at: str,
path_prefix: str = "",
max_segment_chars: int = 0,
outline_run: int = 0,
) -> int:
if max_segment_chars < 0:
raise ProposerError(
f"--max-segment-chars {max_segment_chars} is negative; the cap is a "
"character count, and 0 means off (Arm B)"
)
if outline_run < 0:
raise ProposerError(
f"--outline-run {outline_run} is negative; the gate is a run LENGTH, "
"and 0 means off (Arm B)"
)
# Reduced HERE, before anything is read: a prefix that survives to the
# entries as an empty component would produce exactly the unscoped paths
# the caller asked to avoid, and would do it silently.
#
# PER COMPONENT, because the prefix carries a DIRECTORY now that Door B
# walks the inbox recursively and records a relative `source_file`.
# Reducing the whole string would fold `/` into a `-` and flatten
# `sub/sub2` into the single component `sub-sub2` -- a bundle shaped unlike
# the inbox it came from, and unlike what the caller wrote.
components = (
[reduce_to_id_grammar(part) for part in path_prefix.split("/")] if path_prefix else []
)
if path_prefix and not all(components):
raise ProposerError(
f"--path-prefix {path_prefix!r} has a component that reduces to nothing under "
"the id grammar ([a-z0-9][a-z0-9-]*); refusing to write unscoped paths under a "
"scope that was asked for"
)
scope = "/".join(components)
if not source.is_file():
raise ProposerError(f"source is not a file: {source}")
try:
source_bytes = source.read_bytes()
except OSError as exc:
raise ProposerError(f"cannot read {source}: {exc}") from exc
try:
text = extract_text(source.name, source_bytes)
except IngestError as exc:
raise ProposerError(f"cannot extract text from {source.name}: {exc}") from exc
payload = build_plan(
source,
text,
source_bytes,
okf_type=okf_type,
proposed_at=proposed_at,
path_prefix=scope,
max_segment_chars=max_segment_chars,
outline_run=outline_run,
)
# Nothing to propose is an OUTCOME, and it is not an artifact. An empty
# plan cannot be replayed -- `process_inbox` refuses one, because a plan
# naming no entry would persist nothing for a document that was dropped --
# so the only thing a zero-entry file can do is fail a run later. Its own
# exit status, distinct from 2, so a driver can tell "this document lands
# as one flat concept" from "stop".
if not payload["entries"]:
print(
f"{PROPOSER_ID}: nothing to propose for {source.name} — the mechanical "
"rules found no boundary. No artifact written; this document lands as "
"one concept unless someone segments it by hand.",
file=sys.stderr,
)
return 1
out.parent.mkdir(parents=True, exist_ok=True)
out.write_bytes((json.dumps(payload, indent=2, ensure_ascii=False) + "\n").encode("utf-8"))
print(
f"{PROPOSER_ID}: proposed {len(payload['entries'])} segment(s) -> {out}\n"
f"{PROPOSER_ID}: every entry is PROPOSED. Adjudicate before ingesting.",
file=sys.stderr,
)
return 0
def parse_args(argv: list[str] | None) -> argparse.Namespace:
parser = argparse.ArgumentParser(
prog=PROPOSER_ID,
description="Propose a segmentation plan. A human adjudicates it before use.",
)
parser.add_argument("source", type=Path, help="the document to segment")
parser.add_argument("--out", type=Path, required=True, help="where to write the artifact")
parser.add_argument("--okf-type", default="reference", help="okf_type for every entry")
parser.add_argument(
"--path-prefix",
default="",
help=(
"scope every entry's path under this directory, `/`-separated for a "
"nested one (each component is reduced on its own). Required for a corpus: "
"section numbering is document-local, so two documents propose the same "
"path and Door B refuses both. An argument rather than something this "
"tool derives -- it sees one document and cannot know what else is in "
"the bundle"
),
)
parser.add_argument(
"--max-segment-chars",
type=int,
default=0,
metavar="N",
help=(
"Arm C: cut any proposed span longer than N characters at the nearest "
"paragraph boundary, the whole document counting as one span when the "
"mechanical rules find no boundary at all. 0 (the default) is OFF and "
"leaves the artifact byte-identical to Arm B. Arm C is the author's "
"definition, written for order 20260904T145630Z; it is not defined in "
"the K3 method file"
),
)
parser.add_argument(
"--outline-run",
type=int,
default=0,
metavar="N",
help=(
"Arm D: also propose a boundary at each line of the document's own "
"numbered outline (the bare integers the heading grammar cannot "
"match, since it requires a dot), but only where those integers "
"sustain an ascending run of at least N entries, and only for the "
"LAST such run when the outline repeats, because a contents listing "
"precedes the body it lists. 0 (the default) is OFF and leaves the "
"artifact byte-identical to Arm B. Arm D is the author's definition, "
"written for order 20260906T213322Z; it is not defined upstream, and "
"the K3 method file does not name it either"
),
)
parser.add_argument(
"--proposed-at",
default="1970-01-01T00:00:00Z",
help="the timestamp written into the artifact; explicit so a run is reproducible",
)
return parser.parse_args(argv)
def main(argv: list[str] | None = None) -> int:
args = parse_args(argv)
try:
return run(
args.source,
args.out,
okf_type=args.okf_type,
proposed_at=args.proposed_at,
path_prefix=args.path_prefix,
max_segment_chars=args.max_segment_chars,
outline_run=args.outline_run,
)
except ProposerError as exc:
print(f"{PROPOSER_ID}: FAILED - {exc}", file=sys.stderr)
print(
f"{PROPOSER_ID}: this is NOT 'nothing to propose'. Nothing was written.",
file=sys.stderr,
)
return 2
from llm_ingestion_okf.propose import main # noqa: E402
if __name__ == "__main__":
raise SystemExit(main())