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

feat(segmentation): segmentation plan data model and typed errors

Browse source
This commit is contained in:
Kjell Tore Guttormsen 2026-08-31 23:56:57 +02:00
commit 0a11c860ce
3 changed files with 610 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

33
src/llm_ingestion_okf/errors.py
View file

@ -147,6 +147,39 @@ class MaterializationError(IngestError):
"""
class SegmentationError(IngestError):
"""A segmentation plan is unusable, or does not apply (Door B, 1-to-N).
A concept is "a single unit of knowledge within a bundle" (OKF v0.2 §2),
not a file someone dropped, so splitting one document into several is a
judgement. A judgement cannot be made on a run path that promises zero
model calls, so it is made once, written down as data, adjudicated by a
human, and replayed deterministically. Every failure here is that replay
refusing to guess: a plan that no longer matches its extraction is never
silently re-derived, because the offsets it carries would then point into
text no one adjudicated.
Codes:
- `segmentation_plan_invalid` — the plan's shape is wrong: a missing or
wrongly-typed field, an empty entry list, or a `parent_id` naming no
entry in the same plan
- `segmentation_path_invalid` — an entry's path is not a bundle-relative
`/`-separated path (absolute, empty, or containing `.`/`..`), a
component reduces to nothing under the id grammar, or two entries claim
one path after normalisation
- `segmentation_span_invalid` — a span is not a half-open pair of
non-negative offsets with `start < end`, or it reaches past the end of
the canonical extracted text
- `segmentation_duplicate_id` — two entries share a `segment_id`
- `segmentation_extractor_mismatch` — the plan was adjudicated against a
different extraction. Source bytes cannot see an extractor swap or a
version bump, so the hash alone would still match while every stored
offset had silently moved
- `segmentation_unsupported_profile` — a plan was passed to a profile that
does not declare the segmentation capability
"""
class NetworkGateError(IngestError):
"""A network source was used without the per-run opt-in flag (spec §8).

317
src/llm_ingestion_okf/segmentation.py Normal file
View file

@ -0,0 +1,317 @@
"""The segmentation plan: one document's split into many concepts, as data.
OKF v0.2 §2 defines a concept as "a single unit of knowledge within a bundle"
and a concept ID as the path of its file within the bundle. Neither ties a
concept to a source file, and Appendix A presents v0.1 -> v0.2 as a
de-monolithization. Door B nevertheless emitted exactly one flat concept per
dropped file, which is the form the SPEC names as the one being migrated away
from. No conformance test caught that and none could: §11 checks that every
non-reserved `.md` has parsable frontmatter with a non-empty `type`, so a
bundle of one giant concept is fully conformant. Conformance is the floor, not
the proof.
Splitting a document into units of knowledge is a JUDGEMENT, and this
library's run path promises zero model calls. The resolution is to make the
judgement once, write it down here as data, have a human adjudicate it, and
replay it deterministically forever after. A plan is therefore authored input,
never something this module infers: nothing below proposes a split, and the
proposer that does (`tools/okf_propose_segments.py`) lives outside the package
and marks every entry it emits as PROPOSED rather than adjudicated.
Two properties this module exists to protect:
1. **Offsets index the CANONICAL EXTRACTED TEXT, never the source bytes.** You
cannot slice a PDF's bytes and recover prose, and even a `.csv` is
re-rendered into a table before it becomes a concept body. A span is a
window on whatever `extract.extract_text` returned.
2. **Paths are normalised through the id grammar at entry.** macOS/APFS hands
filenames back DECOMPOSED, so the same visual path reduces two ways
depending on which normal form it arrived in. Normalising once, here,
is what keeps a concept ID from silently moving between rounds -- the one
failure that cannot be repaired after the fact, because consumers have
already linked to the old ID.
Pure: no filesystem, no bundle, no door, no model call, no network.
"""
from __future__ import annotations
from collections.abc import Mapping, Sequence
from dataclasses import dataclass, field
from pathlib import PurePosixPath
from typing import Any
from .errors import SegmentationError
from .materialize import reduce_to_id_grammar
#: The top-level keys a plan payload must carry. Every one is required: a plan
#: missing its extractor identity would still parse, and would then be replayed
#: against an extraction nobody checked it against.
PLAN_FIELDS = (
"version",
"source_sha256",
"extractor_id",
"extractor_version",
"adjudicated_at",
"entries",
)
#: The keys every entry must carry. `parent_id` and `derived` are optional --
#: a flat plan has no parents, and an entry adjudicated from scratch derived
#: nothing.
ENTRY_FIELDS = ("segment_id", "path", "title", "okf_type", "span", "ingested_at")
#: Path components refused outright, before the id grammar is consulted. An
#: empty component is a leading, trailing or doubled `/`; `.` and `..` are
#: traversal. Refused here rather than resolved, because a plan is authored and
#: an authored `..` is a mistake worth naming, not a path worth normalising.
FORBIDDEN_COMPONENTS = ("", ".", "..")
@dataclass(frozen=True)
class SegmentEntry:
"""One concept a document expands into.
`span` is half-open over the canonical extracted text. `path` is
bundle-relative, `/`-separated and already normalised (see
:func:`normalize_segment_path`) -- the concept ID is this path minus the
suffix, so it is fixed the moment the plan is adjudicated.
"""
segment_id: str
path: str
title: str
okf_type: str
span: tuple[int, int]
ingested_at: str
parent_id: str | None = None
derived: frozenset[str] = field(default_factory=frozenset)
@dataclass(frozen=True)
class SegmentationPlan:
"""An adjudicated split, keyed to the extraction it was adjudicated against.
The three extractor fields are not decoration. Source bytes cannot see an
extractor swap or a version bump, so `source_sha256` alone would still
match while every offset in `entries` had silently moved -- see
:func:`assert_plan_applies`.
"""
version: str
source_sha256: str
extractor_id: str
extractor_version: str
adjudicated_at: str
entries: tuple[SegmentEntry, ...]
def _require_str(payload: Mapping[str, Any], key: str, *, where: str) -> str:
if key not in payload:
raise SegmentationError(
f"{where} is missing the required field {key!r} — a plan is replayed "
"verbatim, so an absent field cannot be inferred",
code="segmentation_plan_invalid",
)
value = payload[key]
if not isinstance(value, str) or not value:
raise SegmentationError(
f"{where} field {key!r} must be a non-empty string, got {value!r}",
code="segmentation_plan_invalid",
)
return value
def normalize_segment_path(path: str, *, where: str) -> str:
"""The bundle-relative path an entry claims, reduced to the id grammar.
Every component is reduced separately, because reducing the joined string
would collapse the `/` separators into `-` and flatten the hierarchy the
plan exists to express. The last component's suffix is preserved rather
than reduced (`brannkonsept.md` must not become `brannkonsept-md`), which
is the same split Door B already makes on a dropped filename.
"""
if not isinstance(path, str) or not path:
raise SegmentationError(
f"{where} must carry a non-empty bundle-relative path, got {path!r}",
code="segmentation_path_invalid",
)
if "\\" in path:
raise SegmentationError(
f"{where} path {path!r} contains a backslash — paths are `/`-separated "
"and bundle-relative on every platform",
code="segmentation_path_invalid",
)
components = path.split("/")
forbidden = [item for item in components if item in FORBIDDEN_COMPONENTS]
if forbidden:
raise SegmentationError(
f"{where} path {path!r} is not bundle-relative — it is absolute, or it "
f"contains {', '.join(repr(item) for item in forbidden)}; refusing to "
"resolve traversal in authored data",
code="segmentation_path_invalid",
)
normalized: list[str] = []
last = len(components) - 1
for index, component in enumerate(components):
suffix = PurePosixPath(component).suffix if index == last else ""
stem = component[: len(component) - len(suffix)] if suffix else component
reduced = reduce_to_id_grammar(stem)
if not reduced:
raise SegmentationError(
f"{where} path {path!r} has a component {component!r} that reduces to "
"nothing under the id grammar ([a-z0-9][a-z0-9-]*) — refusing to "
"invent a directory name",
code="segmentation_path_invalid",
)
normalized.append(reduced + suffix.lower())
return "/".join(normalized)
def _parse_span(value: Any, *, where: str) -> tuple[int, int]:
if (
not isinstance(value, Sequence)
or isinstance(value, (str, bytes))
or len(value) != 2
or not all(isinstance(offset, int) for offset in value)
):
raise SegmentationError(
f"{where} span must be a two-item [start, end] of integer offsets into "
f"the canonical extracted text, got {value!r}",
code="segmentation_span_invalid",
)
start, end = int(value[0]), int(value[1])
if start < 0 or end <= start:
raise SegmentationError(
f"{where} span [{start}, {end}] is not a half-open range of non-negative "
"offsets with start < end — an empty or reversed span names no text",
code="segmentation_span_invalid",
)
return (start, end)
def _parse_derived(value: Any, *, where: str) -> frozenset[str]:
if not isinstance(value, Sequence) or isinstance(value, (str, bytes)):
raise SegmentationError(
f"{where} field 'derived' must be a list of field names, got {value!r}",
code="segmentation_plan_invalid",
)
if not all(isinstance(name, str) and name for name in value):
raise SegmentationError(
f"{where} field 'derived' must hold non-empty field names, got {value!r}",
code="segmentation_plan_invalid",
)
return frozenset(value)
def _parse_entry(payload: Any, *, position: int) -> SegmentEntry:
where = f"segmentation entry {position}"
if not isinstance(payload, Mapping):
raise SegmentationError(
f"{where} must be a mapping, got {payload!r}",
code="segmentation_plan_invalid",
)
for key in ENTRY_FIELDS:
if key not in payload:
raise SegmentationError(
f"{where} is missing the required field {key!r} — a plan is replayed "
"verbatim, so an absent field cannot be inferred",
code="segmentation_plan_invalid",
)
segment_id = _require_str(payload, "segment_id", where=where)
where = f"segmentation entry {segment_id!r}"
parent_id = payload.get("parent_id")
if parent_id is not None and (not isinstance(parent_id, str) or not parent_id):
raise SegmentationError(
f"{where} field 'parent_id' must be a non-empty string or absent, got {parent_id!r}",
code="segmentation_plan_invalid",
)
return SegmentEntry(
segment_id=segment_id,
path=normalize_segment_path(payload["path"], where=where),
title=_require_str(payload, "title", where=where),
okf_type=_require_str(payload, "okf_type", where=where),
span=_parse_span(payload["span"], where=where),
ingested_at=_require_str(payload, "ingested_at", where=where),
parent_id=parent_id,
derived=_parse_derived(payload.get("derived", ()), where=where),
)
def parse_segmentation_plan(payload: Mapping[str, Any]) -> SegmentationPlan:
"""Validate an authored plan fail-fast, or refuse it with a typed code.
Fail-fast rather than best-effort: a plan is the record of a human
judgement, and a partially-honoured one would materialize a bundle nobody
adjudicated. Entries keep their authored order — the plan states the
document's own sequence, which no sort here could recover.
"""
if not isinstance(payload, Mapping):
raise SegmentationError(
f"a segmentation plan must be a mapping, got {payload!r}",
code="segmentation_plan_invalid",
)
for key in PLAN_FIELDS:
if key not in payload:
raise SegmentationError(
f"the segmentation plan is missing the required field {key!r} — a plan "
"is replayed verbatim, so an absent field cannot be inferred",
code="segmentation_plan_invalid",
)
raw_entries = payload["entries"]
if (
not isinstance(raw_entries, Sequence)
or isinstance(raw_entries, (str, bytes))
or not raw_entries
):
raise SegmentationError(
"a segmentation plan must name at least one entry — an empty plan would "
"silently persist nothing for a document that was dropped",
code="segmentation_plan_invalid",
)
entries = tuple(
_parse_entry(item, position=position) for position, item in enumerate(raw_entries)
)
seen_ids: set[str] = set()
for item in entries:
if item.segment_id in seen_ids:
raise SegmentationError(
f"two segmentation entries share the segment_id {item.segment_id!r} — "
"refusing to let plan order decide which one a parent points at",
code="segmentation_duplicate_id",
)
seen_ids.add(item.segment_id)
seen_paths: set[str] = set()
for item in entries:
if item.path in seen_paths:
raise SegmentationError(
f"two segmentation entries claim the path {item.path!r} after "
"normalisation — refusing to let one segment silently overwrite "
"the other",
code="segmentation_path_invalid",
)
seen_paths.add(item.path)
for item in entries:
if item.parent_id is not None and item.parent_id not in seen_ids:
raise SegmentationError(
f"segmentation entry {item.segment_id!r} names parent_id "
f"{item.parent_id!r}, which no entry in this plan carries — a "
"hierarchy is resolved inside one plan or not at all",
code="segmentation_plan_invalid",
)
return SegmentationPlan(
version=_require_str(payload, "version", where="the segmentation plan"),
source_sha256=_require_str(payload, "source_sha256", where="the segmentation plan"),
extractor_id=_require_str(payload, "extractor_id", where="the segmentation plan"),
extractor_version=_require_str(payload, "extractor_version", where="the segmentation plan"),
adjudicated_at=_require_str(payload, "adjudicated_at", where="the segmentation plan"),
entries=entries,
)

260
tests/test_segmentation.py Normal file
View file

@ -0,0 +1,260 @@
"""The segmentation plan: authored data that turns one document into many concepts.
A concept is "a single unit of knowledge within a bundle" (OKF v0.2 §2), not a
file that was dropped. Splitting one document into several is therefore a
judgement, and a judgement made by a model cannot live on a run path that
promises zero model calls. The resolution is to make the decision ONCE, write it
down as data, have a human adjudicate it, and replay it deterministically
thereafter — so this module is pure: no filesystem, no bundle, no door, no model.
Every path is normalised through the id grammar at entry, because macOS hands
filenames back DECOMPOSED: the same visual name would otherwise reduce two ways
depending on which form it arrived in, and silent ID motion between rounds is
the one failure this library cannot repair after the fact.
"""
from __future__ import annotations
import unicodedata
from typing import Any
import pytest
from llm_ingestion_okf.errors import SegmentationError
from llm_ingestion_okf.segmentation import (
SegmentationPlan,
SegmentEntry,
parse_segmentation_plan,
)
def entry(**overrides: Any) -> dict[str, Any]:
payload: dict[str, Any] = {
"segment_id": "s1",
"path": "krav/3-1/brannkonsept.md",
"title": "Brannkonsept",
"okf_type": "requirement",
"span": [0, 40],
"ingested_at": "2026-08-31T12:00:00Z",
}
payload.update(overrides)
return payload
def plan(**overrides: Any) -> dict[str, Any]:
payload: dict[str, Any] = {
"version": "1",
"source_sha256": "a" * 64,
"extractor_id": "text",
"extractor_version": "1.0.0",
"adjudicated_at": "2026-08-31T11:00:00Z",
"entries": [entry()],
}
payload.update(overrides)
return payload
def parse_fails(payload: dict[str, Any]) -> SegmentationError:
with pytest.raises(SegmentationError) as excinfo:
parse_segmentation_plan(payload)
return excinfo.value
# --- the happy path -------------------------------------------------------
def test_a_valid_payload_parses_to_the_expected_plan() -> None:
parsed = parse_segmentation_plan(
plan(
entries=[
entry(),
entry(
segment_id="s2",
path="krav/3-1/roemning.md",
title="Roemning",
span=[40, 90],
parent_id="s1",
derived=["title"],
),
]
)
)
assert isinstance(parsed, SegmentationPlan)
assert parsed.version == "1"
assert parsed.source_sha256 == "a" * 64
assert parsed.extractor_id == "text"
assert parsed.extractor_version == "1.0.0"
assert parsed.adjudicated_at == "2026-08-31T11:00:00Z"
assert len(parsed.entries) == 2
first, second = parsed.entries
assert isinstance(first, SegmentEntry)
assert first.segment_id == "s1"
assert first.path == "krav/3-1/brannkonsept.md"
assert first.title == "Brannkonsept"
assert first.okf_type == "requirement"
assert first.span == (0, 40)
assert first.ingested_at == "2026-08-31T12:00:00Z"
assert first.parent_id is None
assert first.derived == frozenset()
assert second.parent_id == "s1"
assert second.derived == frozenset({"title"})
def test_entries_keep_plan_order_rather_than_being_sorted() -> None:
parsed = parse_segmentation_plan(
plan(
entries=[
entry(segment_id="s2", path="b.md", span=[10, 20]),
entry(segment_id="s1", path="a.md", span=[0, 10]),
]
)
)
assert [item.segment_id for item in parsed.entries] == ["s2", "s1"]
def test_the_plan_and_its_entries_are_frozen() -> None:
parsed = parse_segmentation_plan(plan())
with pytest.raises(Exception):
parsed.entries[0].path = "other.md" # type: ignore[misc]
with pytest.raises(Exception):
parsed.version = "2" # type: ignore[misc]
# --- the four codes this step's parser can raise --------------------------
#
# The registry names six. `segmentation_extractor_mismatch` is raised by
# `assert_plan_applies` and `segmentation_unsupported_profile` by Door B, so
# neither is reachable from the parser; each is covered where it becomes
# reachable rather than asserted here against code that does not exist yet.
def test_an_empty_entry_list_is_refused_as_plan_invalid() -> None:
error = parse_fails(plan(entries=[]))
assert error.code == "segmentation_plan_invalid"
def test_a_missing_top_level_field_is_refused_as_plan_invalid() -> None:
payload = plan()
del payload["extractor_version"]
error = parse_fails(payload)
assert error.code == "segmentation_plan_invalid"
assert "extractor_version" in str(error)
def test_a_missing_entry_field_is_refused_as_plan_invalid() -> None:
broken = entry()
del broken["okf_type"]
error = parse_fails(plan(entries=[broken]))
assert error.code == "segmentation_plan_invalid"
assert "okf_type" in str(error)
def test_a_parent_id_naming_no_entry_is_refused_as_plan_invalid() -> None:
error = parse_fails(plan(entries=[entry(parent_id="nobody")]))
assert error.code == "segmentation_plan_invalid"
assert "nobody" in str(error)
def test_a_leading_slash_is_refused_as_path_invalid() -> None:
error = parse_fails(plan(entries=[entry(path="/krav/a.md")]))
assert error.code == "segmentation_path_invalid"
def test_a_parent_traversal_component_is_refused_as_path_invalid() -> None:
error = parse_fails(plan(entries=[entry(path="krav/../../etc/passwd.md")]))
assert error.code == "segmentation_path_invalid"
def test_a_component_reducing_to_nothing_is_refused_as_path_invalid() -> None:
error = parse_fails(plan(entries=[entry(path="krav/---/a.md")]))
assert error.code == "segmentation_path_invalid"
def test_two_entries_on_one_path_are_refused_as_path_invalid() -> None:
error = parse_fails(
plan(
entries=[
entry(segment_id="s1", path="krav/a.md", span=[0, 10]),
entry(segment_id="s2", path="krav/a.md", span=[10, 20]),
]
)
)
assert error.code == "segmentation_path_invalid"
assert "krav/a.md" in str(error)
def test_two_entries_sharing_an_id_are_refused_as_duplicate_id() -> None:
error = parse_fails(
plan(
entries=[
entry(segment_id="s1", path="krav/a.md", span=[0, 10]),
entry(segment_id="s1", path="krav/b.md", span=[10, 20]),
]
)
)
assert error.code == "segmentation_duplicate_id"
assert "s1" in str(error)
def test_a_negative_span_offset_is_refused_as_span_invalid() -> None:
error = parse_fails(plan(entries=[entry(span=[-1, 10])]))
assert error.code == "segmentation_span_invalid"
def test_a_span_that_does_not_advance_is_refused_as_span_invalid() -> None:
error = parse_fails(plan(entries=[entry(span=[10, 10])]))
assert error.code == "segmentation_span_invalid"
def test_a_reversed_span_is_refused_as_span_invalid() -> None:
error = parse_fails(plan(entries=[entry(span=[40, 10])]))
assert error.code == "segmentation_span_invalid"
def test_a_span_of_the_wrong_shape_is_refused_as_span_invalid() -> None:
error = parse_fails(plan(entries=[entry(span=[0, 10, 20])]))
assert error.code == "segmentation_span_invalid"
# --- normalisation --------------------------------------------------------
def test_a_decomposed_path_normalises_to_the_composed_one() -> None:
# macOS/APFS hands names back decomposed. The same visual path must reduce
# to the same stored path from either form, or a re-run silently moves an
# ID -- the failure this library cannot repair afterwards.
composed = unicodedata.normalize("NFC", "krav/bygningsdelér.md")
decomposed = unicodedata.normalize("NFD", composed)
assert composed != decomposed
from_composed = parse_segmentation_plan(plan(entries=[entry(path=composed)]))
from_decomposed = parse_segmentation_plan(plan(entries=[entry(path=decomposed)]))
assert from_composed.entries[0].path == from_decomposed.entries[0].path
stored = from_composed.entries[0].path
assert stored == unicodedata.normalize("NFC", stored)
def test_a_path_is_lowercased_and_reduced_component_by_component() -> None:
parsed = parse_segmentation_plan(plan(entries=[entry(path="Krav/3.1 Brann/Konsept A.md")]))
assert parsed.entries[0].path == "krav/3-1-brann/konsept-a.md"
def test_the_final_suffix_survives_reduction() -> None:
parsed = parse_segmentation_plan(plan(entries=[entry(path="krav/brannkonsept.md")]))
assert parsed.entries[0].path.endswith(".md")
def test_two_paths_differing_only_in_normal_form_collide_as_duplicates() -> None:
composed = unicodedata.normalize("NFC", "krav/bygningsdelér.md")
decomposed = unicodedata.normalize("NFD", composed)
error = parse_fails(
plan(
entries=[
entry(segment_id="s1", path=composed, span=[0, 10]),
entry(segment_id="s2", path=decomposed, span=[10, 20]),
]
)
)
assert error.code == "segmentation_path_invalid"