llm-ingestion-okf/tests/test_segmented_index.py

"""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, `IndexPolicy.sort_entries`,
  which BOTH doors call. It lives on the policy rather than in this door
  because a consumer-controlled ordering wired into one door alone is a
  profile field the other ignores in silence -- see `test_index_sort.py`.
"""

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"
        ]