415 lines
15 KiB
Python
415 lines
15 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, `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_OKF_V0_2,
|
|
SEGMENTED_V1,
|
|
STRUCTURED_V1,
|
|
)
|
|
from llm_ingestion_okf.extract import extract_text
|
|
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
|
|
|
|
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 _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,
|
|
paths: tuple[str, ...] = PATHS,
|
|
*,
|
|
entries_override: dict[str, dict[str, Any]] | None = None,
|
|
text: str | None = None,
|
|
**overrides: Any,
|
|
):
|
|
verdicts = entries_override or {}
|
|
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": (
|
|
hashlib.sha256(text.encode("utf-8")).hexdigest()
|
|
if text is not None
|
|
else _extracted_text_sha256(source_bytes)
|
|
),
|
|
"extractor_id": "md",
|
|
"extractor_version": observed_extractor_version("md"),
|
|
"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,
|
|
**({"adjudication": verdicts[path]} if path in verdicts else {}),
|
|
}
|
|
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 {"bundle_id": "b-1", "okf_version": "0.2"}
|
|
if profile is SEGMENTED_OKF_V0_2
|
|
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"
|
|
]
|
|
|
|
|
|
# --- B2: the adjudication state, under the profile that asks for it --------
|
|
#
|
|
# The wire form is a CONTRACT with `portfolio-optimiser` (`docs/plan/
|
|
# office-intake.md` § 5), not a naming choice this repo may revise: key
|
|
# `adjudication`, CLOSED value set `proposed` | `adjudicated`, and when the
|
|
# value is `adjudicated` three further keys -- `adjudicated_by`,
|
|
# `adjudicated_at` (ISO 8601) and `adjudication_dwell_s` (an INTEGER number of
|
|
# seconds). The consumer's half is that ABSENCE means the state `unknown` (an
|
|
# older bundle), never a collapse to `absent`, so a producer emitting the key
|
|
# inconsistently would make that distinction unmeasurable on their side.
|
|
#
|
|
# The discriminator is `SegmentationPolicy.adjudication_key`, never
|
|
# `profile.segmentation is not None`: BOTH segmented profiles satisfy the
|
|
# latter, so keying on it would write the state into `SEGMENTED_V1` too and
|
|
# move a byte-pinned golden.
|
|
|
|
VERDICT = {
|
|
"adjudicated_by": "ktg",
|
|
"adjudicated_at": "2026-09-02T10:00:00Z",
|
|
"adjudication_dwell_s": 41,
|
|
}
|
|
|
|
|
|
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] != " "
|
|
)
|
|
|
|
|
|
def build_v0_2(tmp: Path, *, adjudicated: tuple[str, ...] = (), bundle_name: str = "bundle"):
|
|
source = drop(tmp / "round", "n500.md")
|
|
plan = build_plan(
|
|
source.read_bytes(),
|
|
entries_override={path: dict(VERDICT) for path in adjudicated},
|
|
)
|
|
return run(tmp, plan=plan, profile=SEGMENTED_OKF_V0_2, bundle_name=bundle_name)
|
|
|
|
|
|
def test_an_unratified_segment_carries_the_proposed_state(tmp_path: Path) -> None:
|
|
build_v0_2(tmp_path)
|
|
values = frontmatter_of(tmp_path / "bundle" / "krav" / "forord.md")
|
|
assert values["adjudication"] == "proposed"
|
|
assert "adjudicated_by" not in values
|
|
assert "adjudication_dwell_s" not in values
|
|
|
|
|
|
def test_a_ratified_segment_carries_its_adjudicator_time_and_dwell(tmp_path: Path) -> None:
|
|
build_v0_2(tmp_path, adjudicated=("krav/forord.md",))
|
|
values = frontmatter_of(tmp_path / "bundle" / "krav" / "forord.md")
|
|
assert values["adjudication"] == "adjudicated"
|
|
assert values["adjudicated_by"] == "ktg"
|
|
assert values["adjudicated_at"] == "2026-09-02T10:00:00Z"
|
|
# An INTEGER number of seconds, per B2 -- not a float and not a duration
|
|
# string. A ratified flag with no per-item time is unfalsifiable, and this
|
|
# is the same field that instruments adjudication throughput.
|
|
assert values["adjudication_dwell_s"] == "41"
|
|
assert int(values["adjudication_dwell_s"]) == 41
|
|
|
|
|
|
def test_the_state_is_projected_as_an_index_facet(tmp_path: Path) -> None:
|
|
build_v0_2(tmp_path, adjudicated=("krav/forord.md",))
|
|
entries = indexes(tmp_path / "bundle", profile=SEGMENTED_OKF_V0_2)
|
|
facets = [entry.facets for listing in entries.values() for entry in listing if entry.facets]
|
|
states = {facet.get("adjudication") for facet in facets}
|
|
assert "adjudicated" in states
|
|
assert "proposed" in states
|
|
|
|
|
|
def test_the_facet_survives_a_reprojection_of_the_whole_bundle(tmp_path: Path) -> None:
|
|
"""The index is a PROJECTION recomputed from stored frontmatter each round.
|
|
|
|
A state that lived only in the index would be lost the moment the index
|
|
was rebuilt, which is every round.
|
|
"""
|
|
build_v0_2(tmp_path, adjudicated=("krav/forord.md",))
|
|
first = tree(tmp_path / "bundle")
|
|
build_v0_2(tmp_path, adjudicated=("krav/forord.md",))
|
|
assert tree(tmp_path / "bundle") == first
|
|
|
|
|
|
def test_a_state_outside_the_closed_set_is_refused(tmp_path: Path) -> None:
|
|
"""Closed means closed: a value outside it is an error, not an extension.
|
|
|
|
Reaches the door the way an outside value actually can -- declared in the
|
|
dropped document's own frontmatter, before this library writes anything.
|
|
"""
|
|
body = "---\nadjudication: nesten-ferdig\n---\n\n" + DOCUMENT
|
|
source = drop(tmp_path / "round", "n500.md", body)
|
|
# The span covers the block, which is the only way a declared value reaches
|
|
# derivation at all: a segment body is a SLICE, so a frontmatter block the
|
|
# span misses is simply not part of that concept.
|
|
plan = build_plan(
|
|
source.read_bytes(),
|
|
paths=("krav/forord.md",),
|
|
text=body,
|
|
entries=[
|
|
{
|
|
"segment_id": "s0",
|
|
"path": "krav/forord.md",
|
|
"title": "Forord",
|
|
"okf_type": "requirement",
|
|
"span": [0, len(body)],
|
|
"ingested_at": PLAN_AT,
|
|
}
|
|
],
|
|
)
|
|
result = run(tmp_path, plan=plan, profile=SEGMENTED_OKF_V0_2)
|
|
assert {entry.error.code for entry in result.failed} == {"index_facet_invalid"}
|
|
|
|
|
|
def test_segmented_v1_writes_no_adjudication_state_at_all(tmp_path: Path) -> None:
|
|
"""The discriminator, measured: the older profile's bytes must not move."""
|
|
build(tmp_path, bundle_name="bundle")
|
|
values = frontmatter_of(tmp_path / "bundle" / "krav" / "forord.md")
|
|
assert "adjudication" not in values
|
|
assert "adjudicated_by" not in values
|