llm-ingestion-okf/tests/test_segmented_rounds.py

"""Rounds: a bundle built up over several drops keeps its concept IDs.

The operator's requirement in their own words -- documents arrive "en eller
flere ganger, altsa additivt" -- with one extra edge under 1-to-N that flat
Door B never had: the SPLIT itself can change between rounds. A document
adjudicated as one concept in round 1 may be re-adjudicated into four in round
2, and a later round may name fewer segments than the one before it.

Two failures follow if ownership stays keyed on the concept FILENAME, which is
what it was:

- the pre-existing scan and the reprojection both use FLAT globs, so a nested
  concept is invisible to them -- our own file looks like curated content and
  the collision gate fires on it;
- a round yielding fewer segments ORPHANS the ones it no longer names. They
  survive in the incremental bundle and are absent from a scratch rebuild, so
  the two diverge silently. That is the S7 acceptance test failing for a reason
  no per-step test would catch.

The fix is to ask a different question: not "is this filename owned?" but
"which concepts are owned by source_file X?".

**One deviation from the plan, stated rather than absorbed.** The plan's S4
row also asks that a deprecated parent carry `status: deprecated`. Nothing in
this library populates `status`: it is NAMED as a facet key in `profiles.py`
but no code path ever writes it, and a segment body is a slice of extracted
text, so it cannot declare one either. Deriving it would be new machinery in a
step whose file list is `inbox.py` alone. What IS load-bearing about S4 -- that
a re-split makes no concept ID disappear, adds at least two, and keeps the
parent with an edge to its children -- is asserted below.
"""

from __future__ import annotations

import hashlib
from pathlib import Path
from typing import Any

from llm_ingestion_okf.extract import extract_text
from llm_ingestion_okf.inbox import GateDecision, process_inbox
from llm_ingestion_okf.profiles import DEFAULT, SEGMENTED_V1
from llm_ingestion_okf.segmentation import (
    SegmentationPlan,
    observed_extractor_version,
    parse_segmentation_plan,
)

INGESTED_AT = "2026-07-25T12:00:00Z"
PLAN_AT = "2026-08-30T09:00:00Z"

DOCUMENT = "Krav i konseptet.\n" * 20


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 _extracted_text_sha256(source_bytes: bytes, filename: str = "n500.md") -> str:
    return hashlib.sha256(extract_text(filename, source_bytes).encode("utf-8")).hexdigest()


def build_plan(
    source_bytes: bytes,
    entries: tuple[tuple[str, str, str | None], ...],
    **overrides: Any,
) -> SegmentationPlan:
    payload: dict[str, Any] = {
        "version": "1",
        "source_sha256": hashlib.sha256(source_bytes).hexdigest(),
        # The hash of the CANONICAL EXTRACTED text, which is what the spans
        # index. Equal to the source hash on a `.md` passthrough and computed
        # rather than copied, so the fixture keeps saying which one it means.
        "text_sha256": _extracted_text_sha256(source_bytes),
        "extractor_id": "md",
        "extractor_version": observed_extractor_version("md"),
        "adjudicated_at": "2026-08-30T08:00:00Z",
        "entries": [
            {
                "segment_id": segment_id,
                "path": path,
                "title": f"Del {segment_id}",
                "okf_type": "requirement",
                "span": [index * 10, index * 10 + 10],
                "ingested_at": PLAN_AT,
                **({"parent_id": parent} if parent else {}),
            }
            for index, (segment_id, path, parent) in enumerate(entries)
        ],
    }
    payload.update(overrides)
    return parse_segmentation_plan(payload)


def run(
    tmp: Path,
    *,
    plan: SegmentationPlan | None = None,
    profile=SEGMENTED_V1,
    round_name: str = "round",
):
    return process_inbox(
        tmp / round_name,
        tmp / "bundle",
        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 concept_ids(bundle: Path, profile=SEGMENTED_V1) -> list[str]:
    """Sorted bundle-relative concept IDs -- the path minus the suffix.

    IDs, never basenames: OKF v0.2 §2 defines a concept ID as the path of the
    concept's file within the bundle, and two files called `a.md` in different
    directories are two different concepts.
    """
    if not bundle.is_dir():
        return []
    suffix = profile.paths.concept_suffix
    return sorted(
        str(path.relative_to(bundle))[: -len(suffix)]
        for path in bundle.rglob(f"*{suffix}")
        if path.is_file() and path.name != profile.index.name
    )


def frontmatter_of(path: Path) -> dict[str, str]:
    head = path.read_text(encoding="utf-8").split("---\n")[1]
    return dict(
        line.split(": ", 1) for line in head.splitlines() if ": " in line and line[:1] != " "
    )


# --- S4 / S6: a re-split moves no ID --------------------------------------


ROUND_1 = (("s0", "krav/brannkonsept.md", None),)
ROUND_2 = (
    ("s0", "krav/brannkonsept.md", None),
    ("s1", "krav/brannkonsept/seksjonering.md", "s0"),
    ("s2", "krav/brannkonsept/roemning.md", "s0"),
)


def test_a_resplit_loses_no_concept_id_and_adds_at_least_two(tmp_path: Path) -> None:
    source = drop(tmp_path / "one", "n500.md")
    run(tmp_path, plan=build_plan(source.read_bytes(), ROUND_1), round_name="one")
    first = concept_ids(tmp_path / "bundle")

    drop(tmp_path / "two", "n500.md")
    run(tmp_path, plan=build_plan(source.read_bytes(), ROUND_2), round_name="two")
    second = concept_ids(tmp_path / "bundle")

    # comm -23: present in round 1, absent from round 2. Nothing may appear
    # here -- a consumer has already linked to every one of those IDs.
    assert [item for item in first if item not in second] == []
    # comm -13: new in round 2. A re-split that added nothing is not a re-split.
    assert len([item for item in second if item not in first]) >= 2


def test_the_parent_survives_the_resplit_and_its_children_point_at_it(
    tmp_path: Path,
) -> None:
    source = drop(tmp_path / "one", "n500.md")
    run(tmp_path, plan=build_plan(source.read_bytes(), ROUND_1), round_name="one")
    drop(tmp_path / "two", "n500.md")
    run(tmp_path, plan=build_plan(source.read_bytes(), ROUND_2), round_name="two")

    bundle = tmp_path / "bundle"
    assert (bundle / "krav/brannkonsept.md").is_file()
    for child in ("seksjonering.md", "roemning.md"):
        keys = frontmatter_of(bundle / "krav/brannkonsept" / child)
        assert keys["parent"] == "s0"


def test_ids_are_stable_across_a_round_that_changes_nothing(tmp_path: Path) -> None:
    source = drop(tmp_path / "one", "n500.md")
    plan = build_plan(source.read_bytes(), ROUND_2)
    run(tmp_path, plan=plan, round_name="one")
    first = concept_ids(tmp_path / "bundle")
    drop(tmp_path / "two", "n500.md")
    run(tmp_path, plan=plan, round_name="two")
    assert concept_ids(tmp_path / "bundle") == first


# --- stale-segment retirement ---------------------------------------------


def test_a_round_naming_fewer_segments_removes_exactly_the_unnamed_ones(
    tmp_path: Path,
) -> None:
    source = drop(tmp_path / "one", "n500.md")
    run(tmp_path, plan=build_plan(source.read_bytes(), ROUND_2), round_name="one")
    assert len(concept_ids(tmp_path / "bundle")) == 3

    drop(tmp_path / "two", "n500.md")
    run(tmp_path, plan=build_plan(source.read_bytes(), ROUND_1), round_name="two")

    # Exactly the un-named ones. An orphan surviving here is invisible to a
    # scratch rebuild, and the two bundles diverge with nothing failing.
    assert concept_ids(tmp_path / "bundle") == ["krav/brannkonsept"]
    assert not (tmp_path / "bundle" / "krav/brannkonsept/seksjonering.md").exists()


def test_retirement_leaves_every_other_documents_concepts_alone(tmp_path: Path) -> None:
    source = drop(tmp_path / "one", "n500.md")
    other = drop(tmp_path / "one", "v720.md", "Tunnelkrav.\n" * 20)
    run(tmp_path, plan=build_plan(source.read_bytes(), ROUND_2), round_name="one")
    assert "inbox-v720" in concept_ids(tmp_path / "bundle")

    drop(tmp_path / "two", "n500.md")
    run(tmp_path, plan=build_plan(source.read_bytes(), ROUND_1), round_name="two")

    # v720 was not in round 2's inbox at all. Ownership is per source_file, so
    # nothing about n500's re-split may touch it.
    assert (tmp_path / "bundle" / "inbox-v720.md").is_file()
    assert other.name == "v720.md"


# --- the four shipped profiles keep FLAT scans ----------------------------


def test_default_ignores_a_nested_file_when_scanning_for_collisions(
    tmp_path: Path,
) -> None:
    # Switching the shared glob to rglob would change all four profiles'
    # ownership scans and break the byte-stability pin. The branch is on the
    # capability, and this plants a nested file to prove DEFAULT never looks.
    bundle = tmp_path / "bundle"
    (bundle / "krav").mkdir(parents=True)
    (bundle / "krav" / "inbox-n500.md").write_text(
        "---\ntype: note\ngenerated: true\nsource_file: n500.md\n---\n\nnested\n",
        encoding="utf-8",
    )
    drop(tmp_path / "round", "n500.md", "Flat.\n")
    result = run(tmp_path, profile=DEFAULT)

    assert result.failed == ()
    assert (bundle / "inbox-n500.md").is_file()
    # Untouched: DEFAULT does not own it, does not see it, does not retire it.
    assert (bundle / "krav" / "inbox-n500.md").read_text(encoding="utf-8").endswith("nested\n")


def test_default_still_writes_one_flat_concept_per_file(tmp_path: Path) -> None:
    drop(tmp_path / "round", "n500.md", "Flat.\n")
    drop(tmp_path / "round", "v720.md", "Annet.\n")
    run(tmp_path, profile=DEFAULT)
    assert concept_ids(tmp_path / "bundle", DEFAULT) == ["inbox-n500", "inbox-v720"]