245 lines
9.3 KiB
Python
245 lines
9.3 KiB
Python
"""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.inbox import GateDecision, process_inbox
|
|
from llm_ingestion_okf.profiles import DEFAULT, SEGMENTED_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
|
|
|
|
|
|
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,
|
|
entries: tuple[tuple[str, str, str | None], ...],
|
|
**overrides: Any,
|
|
) -> SegmentationPlan:
|
|
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": 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"]
|