258 lines
9 KiB
Python
258 lines
9 KiB
Python
"""Per-directory indexes: a nested bundle a consumer can still walk.
|
|
|
|
A segmented bundle has directories, and a bundle whose nested concepts are
|
|
reachable only by guessing a path is the filing cabinet the index exists to
|
|
replace. So every directory carries an index, and every parent index carries a
|
|
navigation entry pointing at each child's index -- the root is an entry point
|
|
to the whole tree, not just to its own level.
|
|
|
|
Two traps this closes, both measured:
|
|
|
|
- `link_in_index`'s idempotence keys on the SUBSTRING `f"]({target})"`. Once
|
|
targets are relative subdirectory paths that matcher is ambiguous:
|
|
`](krav/3-1/a.md)` also contains `](3-1/a.md)`. The writer here recomputes
|
|
each index whole and recognises managed lines with the ANCHORED
|
|
`link_pattern`, never that substring.
|
|
- ordering is routed through ONE named helper, `_index_sort_key`. A future
|
|
consumer-controlled ordering is then a parameter, not a refactor of every
|
|
place that happened to sort.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from llm_ingestion_okf.inbox import GateDecision, process_inbox
|
|
from llm_ingestion_okf.profiles import DEFAULT, SEGMENTED_V1, STRUCTURED_V1
|
|
from llm_ingestion_okf.segmentation import SegmentationPlan, parse_segmentation_plan
|
|
|
|
INGESTED_AT = "2026-07-25T12:00:00Z"
|
|
PLAN_AT = "2026-08-30T09:00:00Z"
|
|
|
|
DOCUMENT = "Krav i konseptet.\n" * 20
|
|
|
|
PATHS = (
|
|
"krav/forord.md",
|
|
"krav/3-1/brannkonsept.md",
|
|
"krav/3-1/roemning.md",
|
|
"krav/3-2/baereevne.md",
|
|
"krav/3-2/slokkeanlegg.md",
|
|
)
|
|
|
|
|
|
def gate(text: str) -> GateDecision:
|
|
return GateDecision(sanitized_text=text, disposition="warn")
|
|
|
|
|
|
def drop(inbox: Path, name: str, text: str = DOCUMENT) -> Path:
|
|
inbox.mkdir(parents=True, exist_ok=True)
|
|
path = inbox / name
|
|
path.write_text(text, encoding="utf-8", newline="")
|
|
return path
|
|
|
|
|
|
def build_plan(source_bytes: bytes, paths: tuple[str, ...] = PATHS, **overrides: Any):
|
|
payload: dict[str, Any] = {
|
|
"version": "1",
|
|
"source_sha256": hashlib.sha256(source_bytes).hexdigest(),
|
|
"extractor_id": "md",
|
|
"extractor_version": "1.0.0",
|
|
"adjudicated_at": "2026-08-30T08:00:00Z",
|
|
"entries": [
|
|
{
|
|
"segment_id": f"s{index}",
|
|
"path": path,
|
|
"title": f"Del {index}",
|
|
"okf_type": "requirement",
|
|
"span": [index * 10, index * 10 + 10],
|
|
"ingested_at": PLAN_AT,
|
|
}
|
|
for index, path in enumerate(paths)
|
|
],
|
|
}
|
|
payload.update(overrides)
|
|
return parse_segmentation_plan(payload)
|
|
|
|
|
|
def run(
|
|
tmp: Path,
|
|
*,
|
|
plan: SegmentationPlan | None = None,
|
|
profile=SEGMENTED_V1,
|
|
round_name: str = "round",
|
|
bundle_name: str = "bundle",
|
|
):
|
|
return process_inbox(
|
|
tmp / round_name,
|
|
tmp / bundle_name,
|
|
INGESTED_AT,
|
|
okf_type="requirement",
|
|
gate=gate,
|
|
profile=profile,
|
|
root_frontmatter_values={"bundle_id": "b-1"} if profile is SEGMENTED_V1 else None,
|
|
segmentation=plan,
|
|
)
|
|
|
|
|
|
def build(tmp: Path, bundle_name: str = "bundle") -> Path:
|
|
source = drop(tmp / "round", "n500.md")
|
|
run(tmp, plan=build_plan(source.read_bytes()), bundle_name=bundle_name)
|
|
return tmp / bundle_name
|
|
|
|
|
|
def indexes(bundle: Path, profile=SEGMENTED_V1) -> dict[str, list]:
|
|
"""Every index in the bundle, parsed into its managed entries."""
|
|
found: dict[str, list] = {}
|
|
for path in sorted(bundle.rglob(profile.index.name)):
|
|
entries = [
|
|
profile.index.parse_entry(line)
|
|
for line in path.read_text(encoding="utf-8").splitlines()
|
|
]
|
|
# `Path(".")` is what `relative_to` gives for the bundle root; the rest
|
|
# of this module keys the root as the empty string.
|
|
relative = path.parent.relative_to(bundle).as_posix()
|
|
found["" if relative == "." else relative] = [
|
|
entry for entry in entries if entry is not None
|
|
]
|
|
return found
|
|
|
|
|
|
def tree(bundle: Path) -> dict[str, bytes]:
|
|
return {
|
|
str(path.relative_to(bundle)): path.read_bytes()
|
|
for path in sorted(bundle.rglob("*"))
|
|
if path.is_file()
|
|
}
|
|
|
|
|
|
# --- S8: every directory has an index, and every concept is in one --------
|
|
|
|
|
|
def test_every_directory_carries_an_index_with_at_least_one_entry(tmp_path: Path) -> None:
|
|
bundle = build(tmp_path)
|
|
directories = {
|
|
path.parent.relative_to(bundle).as_posix()
|
|
for path in bundle.rglob("*.md")
|
|
if path.is_file() and path.name != SEGMENTED_V1.index.name
|
|
}
|
|
assert len(directories) >= 2
|
|
|
|
found = indexes(bundle)
|
|
assert directories <= set(found)
|
|
for directory, entries in found.items():
|
|
assert entries, f"{directory or '<root>'} has an index with no entries"
|
|
|
|
|
|
def test_the_concept_entries_across_all_indexes_are_exactly_the_concepts(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
bundle = build(tmp_path)
|
|
targets: set[str] = set()
|
|
for directory, entries in indexes(bundle).items():
|
|
prefix = f"{directory}/" if directory else ""
|
|
for entry in entries:
|
|
if not entry.target.endswith(SEGMENTED_V1.index.name):
|
|
targets.add(prefix + entry.target)
|
|
assert targets == set(PATHS)
|
|
|
|
|
|
def test_no_entry_has_an_empty_label_or_a_description_echoing_it(tmp_path: Path) -> None:
|
|
for entries in indexes(build(tmp_path)).values():
|
|
for entry in entries:
|
|
assert entry.label.strip()
|
|
description = getattr(entry, "description", None)
|
|
if description is not None:
|
|
assert description.strip()
|
|
assert description != entry.label
|
|
|
|
|
|
# --- reachability: the root is an entry point to the whole tree -----------
|
|
|
|
|
|
def test_every_concept_is_reachable_from_the_root_by_index_links_only(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
bundle = build(tmp_path)
|
|
found = indexes(bundle)
|
|
reached: set[str] = set()
|
|
frontier = [""]
|
|
seen_directories: set[str] = set()
|
|
while frontier:
|
|
directory = frontier.pop()
|
|
if directory in seen_directories:
|
|
continue
|
|
seen_directories.add(directory)
|
|
prefix = f"{directory}/" if directory else ""
|
|
for entry in found.get(directory, []):
|
|
target = prefix + entry.target
|
|
if entry.target.endswith(SEGMENTED_V1.index.name):
|
|
frontier.append(str(Path(target).parent))
|
|
else:
|
|
reached.add(target)
|
|
assert reached == set(PATHS)
|
|
|
|
|
|
def test_a_parent_index_names_each_child_directory(tmp_path: Path) -> None:
|
|
bundle = build(tmp_path)
|
|
krav = [entry.target for entry in indexes(bundle)["krav"]]
|
|
assert f"3-1/{SEGMENTED_V1.index.name}" in krav
|
|
assert f"3-2/{SEGMENTED_V1.index.name}" in krav
|
|
|
|
|
|
def test_the_root_index_keeps_its_frontmatter_block(tmp_path: Path) -> None:
|
|
bundle = build(tmp_path)
|
|
assert (
|
|
(bundle / "index.md").read_text(encoding="utf-8").startswith("---\nbundle_id: b-1\n---\n\n")
|
|
)
|
|
# Nested indexes carry none: the asymmetry is the shape both consumers
|
|
# confirmed independently, not one repo's preference.
|
|
assert not (bundle / "krav" / "index.md").read_text(encoding="utf-8").startswith("---")
|
|
|
|
|
|
# --- S8b: determinism -----------------------------------------------------
|
|
|
|
|
|
def test_two_builds_from_identical_inputs_are_byte_identical(tmp_path: Path) -> None:
|
|
first = build(tmp_path, bundle_name="one")
|
|
(tmp_path / "round").rename(tmp_path / "spent")
|
|
(tmp_path / "spent").rename(tmp_path / "round")
|
|
second = build(tmp_path, bundle_name="two")
|
|
assert tree(first) == tree(second)
|
|
|
|
|
|
def test_a_second_round_over_the_same_inputs_changes_nothing(tmp_path: Path) -> None:
|
|
bundle = build(tmp_path)
|
|
before = tree(bundle)
|
|
source = drop(tmp_path / "again", "n500.md")
|
|
run(tmp_path, plan=build_plan(source.read_bytes()), round_name="again")
|
|
assert tree(bundle) == before
|
|
|
|
|
|
def test_nested_targets_do_not_confuse_the_entry_matcher(tmp_path: Path) -> None:
|
|
# `link_in_index` keys idempotence on the substring `](target)`, and
|
|
# `](krav/3-1/a.md)` contains `](3-1/a.md)`. An index recomputed whole with
|
|
# an anchored matcher cannot be fooled that way; a substring matcher would
|
|
# drop or double an entry here.
|
|
source = drop(tmp_path / "round", "n500.md")
|
|
plan = build_plan(source.read_bytes(), ("krav/3-1/a.md", "3-1/a.md"))
|
|
run(tmp_path, plan=plan)
|
|
bundle = tmp_path / "bundle"
|
|
assert (bundle / "krav/3-1/a.md").is_file()
|
|
assert (bundle / "3-1/a.md").is_file()
|
|
targets = [entry.target for entry in indexes(bundle)["3-1"]]
|
|
assert targets.count("a.md") == 1
|
|
|
|
|
|
# --- the shipped profiles keep exactly one root index --------------------
|
|
|
|
|
|
def test_default_and_structured_write_one_root_index_only(tmp_path: Path) -> None:
|
|
for profile, name in ((DEFAULT, "flat"), (STRUCTURED_V1, "structured")):
|
|
drop(tmp_path / name, "n500.md")
|
|
run(tmp_path, profile=profile, round_name=name, bundle_name=name + "-bundle")
|
|
bundle = tmp_path / (name + "-bundle")
|
|
assert [path.relative_to(bundle).as_posix() for path in bundle.rglob("index.md")] == [
|
|
"index.md"
|
|
]
|