"""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, assert_plan_applies, parse_segmentation_plan, plan_cache_key, ) 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" # --- S5b: the cache key is the extractor, not the hash alone --------------- # # Source bytes cannot see an extractor swap or a version bump. Both invalidate # every stored offset while `source_sha256` stays identical, so the mismatch # has to be LOUD -- a silent re-derivation would replay a human's adjudication # against text that human never saw. def parsed_plan(**overrides: Any) -> SegmentationPlan: return parse_segmentation_plan(plan(**overrides)) def applies_fails(subject: SegmentationPlan, **overrides: str) -> SegmentationError: arguments = { "source_sha256": subject.source_sha256, "extractor_id": subject.extractor_id, "extractor_version": subject.extractor_version, } arguments.update(overrides) with pytest.raises(SegmentationError) as excinfo: assert_plan_applies(subject, **arguments) return excinfo.value def test_the_cache_key_is_the_three_tuple() -> None: subject = parsed_plan() assert plan_cache_key(subject) == ( subject.source_sha256, subject.extractor_id, subject.extractor_version, ) def test_two_plans_differing_only_in_extractor_id_have_different_cache_keys() -> None: # The whole point of S5b: the hash alone would call these one cached # adjudication, and replay the first plan's offsets against the second # extraction. one = parsed_plan(extractor_id="text") other = parsed_plan(extractor_id="pdfplumber") assert one.source_sha256 == other.source_sha256 assert plan_cache_key(one) != plan_cache_key(other) assert plan_cache_key(one)[0] == plan_cache_key(other)[0] def test_an_identical_triple_applies_without_raising() -> None: subject = parsed_plan() assert ( assert_plan_applies( subject, source_sha256=subject.source_sha256, extractor_id=subject.extractor_id, extractor_version=subject.extractor_version, ) is None ) def test_a_changed_extractor_version_is_refused_and_named() -> None: error = applies_fails(parsed_plan(), extractor_version="1.0.1") assert error.code == "segmentation_extractor_mismatch" assert "extractor_version" in str(error) assert "1.0.1" in str(error) def test_a_changed_extractor_id_is_refused_and_named() -> None: error = applies_fails(parsed_plan(), extractor_id="pdfplumber") assert error.code == "segmentation_extractor_mismatch" assert "extractor_id" in str(error) def test_a_changed_source_hash_is_refused_and_named() -> None: error = applies_fails(parsed_plan(), source_sha256="b" * 64) assert error.code == "segmentation_extractor_mismatch" assert "source_sha256" in str(error) def test_the_message_names_every_differing_component_not_just_the_first() -> None: error = applies_fails(parsed_plan(), extractor_id="pdfplumber", extractor_version="1.0.1") assert "extractor_id" in str(error) assert "extractor_version" in str(error) assert "source_sha256" not in str(error)