feat(inbox): collision gate over segment paths and additive concept reporting

src/llm_ingestion_okf/inbox.py

@@ -264,6 +264,12 @@ class InboxResult:
     quarantined: tuple[BlockedFile, ...]
     rejected: tuple[BlockedFile, ...]
     failed: tuple[FailedFile, ...]
+    # One entry per CONCEPT, where `persisted` is one per SOURCE FILE. A new
+    # field rather than a changed meaning: under 1-to-N the two counts diverge,
+    # and redefining `persisted` would silently change what every existing
+    # consumer's number means. Without segmentation the two are equal, which is
+    # what makes this additive rather than a second thing to keep in step.
+    concepts: tuple[PersistedFile, ...] = ()
 
 
 def _is_inbox_owned(path: Path) -> bool:
@@ -275,6 +281,21 @@ def _is_inbox_owned(path: Path) -> bool:
     return frontmatter.get("generated") == "true" and "source_file" in frontmatter
 
 
+def _check_segment_path(path: str) -> str:
+    """Refuse a segment path the filesystem cannot hold, measured PER COMPONENT.
+
+    `check_filename_length` measures one name against NAME_MAX, which is a
+    per-directory-entry limit. Measuring the JOINED path against it gets the
+    question backwards in both directions: a perfectly legal deep hierarchy
+    would be refused, and an illegal component inside a short path would be
+    accepted and then fail at the write with an errno that differs per platform
+    -- the untyped, unportable failure the length gate exists to replace.
+    """
+    for component in path.split("/"):
+        check_filename_length(component, code="inbox_slug_too_long")
+    return path
+
+
 def _bundle_id_key(profile: BundleProfile) -> str:
     assert profile.segmentation is not None
     return profile.segmentation.bundle_id_key
@@ -435,37 +456,72 @@ def process_inbox(
     dropped = sorted((path for path in inbox.iterdir() if path.is_file()), key=lambda p: p.name)
 
     persisted: list[PersistedFile] = []
+    concepts: list[PersistedFile] = []
     quarantined: list[BlockedFile] = []
     rejected: list[BlockedFile] = []
     failed: list[FailedFile] = []
 
     # Phase 1: name every file BEFORE any gate call or write, so an intra-run
-    # slug collision is caught while both files can still be refused together.
+    # collision is caught while both files can still be refused together. Under
-    named: list[tuple[Path, str]] = []
+    # 1-to-N a document does not claim ONE name — it claims the whole set of
+    # paths its plan expands to, and the gate is keyed on that set. Keyed on one
+    # name per file, the same defect returns one level down: the second document
+    # would silently claim the first's concepts.
+    named: list[tuple[Path, tuple[str, ...], bytes]] = []
     slug_owners: dict[str, list[Path]] = {}
     for path in dropped:
         try:
-            name = inbox_filename(inbox_slug(path.name), profile=profile)
+            # Read HERE rather than in the write loop: a plan is selected by
-        except IngestError as exc:
+            # content hash, so the set of names a document claims is not knowable
-            failed.append(FailedFile(source_file=path.name, error=exc))
+            # without its bytes, and the whole point of this phase is to know
-            continue
+            # every name before anything happens.
-        named.append((path, name))
+            source_bytes = path.read_bytes()
-        slug_owners.setdefault(name, []).append(path)
+        except OSError as exc:
-
-    colliding = {name for name, owners in slug_owners.items() if len(owners) > 1}
-    for name in sorted(colliding):
-        for path in slug_owners[name]:
             failed.append(
                 FailedFile(
                     source_file=path.name,
-                    error=MaterializationError(
+                    error=SourceError(
-                        f"{path.name!r} and "
+                        f"cannot read dropped file {path.name}: {exc}", code="source_file_missing"
-                        f"{', '.join(repr(other.name) for other in slug_owners[name] if other != path)}"
-                        f" both reduce to {name!r} — rename one; refusing to pick a winner",
-                        code="inbox_slug_collision",
                     ),
                 )
             )
+            continue
+        try:
+            covering = _plan_covering(segmentation, source_bytes)
+            targets: tuple[str, ...]
+            if covering is None:
+                targets = (inbox_filename(inbox_slug(path.name), profile=profile),)
+            else:
+                targets = tuple(_check_segment_path(item.path) for item in covering.entries)
+        except IngestError as exc:
+            failed.append(FailedFile(source_file=path.name, error=exc))
+            continue
+        named.append((path, targets, source_bytes))
+        for target in targets:
+            slug_owners.setdefault(target, []).append(path)
+
+    contested = {name for name, owners in slug_owners.items() if len(owners) > 1}
+    # One refusal per DOCUMENT, not per contested path: a document expanding to
+    # five colliding paths is one thing the operator has to fix, and five
+    # identical entries would report the same rename five times.
+    for path in sorted(
+        {owner for name in contested for owner in slug_owners[name]}, key=lambda item: item.name
+    ):
+        claimed = sorted(name for name in contested if path in slug_owners[name])
+        others = sorted(
+            {other.name for name in claimed for other in slug_owners[name] if other != path}
+        )
+        failed.append(
+            FailedFile(
+                source_file=path.name,
+                error=MaterializationError(
+                    f"{path.name!r} and {', '.join(repr(other) for other in others)}"
+                    f" both reduce to {', '.join(repr(name) for name in claimed)}"
+                    " — rename one; refusing to pick a winner",
+                    code="inbox_slug_collision",
+                ),
+            )
+        )
 
     # Phase 2: the §3 ownership scan, evaluated against the bundle as it was
     # BEFORE this run — a file written below must never be mistaken for
@@ -482,16 +538,18 @@ def process_inbox(
     )
     owned = {name for name in pre_existing if _is_inbox_owned(bundle / name)}
 
-    for path, name in named:
+    for path, targets, source_bytes in named:
-        if name in colliding:
+        if any(name in contested for name in targets):
             continue
-        if name in pre_existing and name not in owned:
+        unstamped = [name for name in targets if name in pre_existing and name not in owned]
+        if unstamped:
             failed.append(
                 FailedFile(
                     source_file=path.name,
                     error=MaterializationError(
-                        f"generated filename {name!r} collides with an existing file that does "
+                        f"generated filename {unstamped[0]!r} collides with an existing file "
-                        "not carry the inbox marker — refusing to overwrite curated content (§3)",
+                        "that does not carry the inbox marker — refusing to overwrite curated "
+                        "content (§3)",
                         code="collision_unstamped",
                     ),
                 )
@@ -499,7 +557,6 @@ def process_inbox(
             continue
         outputs: list[tuple[str, str, tuple[str, ...]]] = []
         try:
-            source_bytes = path.read_bytes()
             text = extract_text(path.name, source_bytes)
             covering = _plan_covering(segmentation, source_bytes)
             if covering is not None:
@@ -548,7 +605,7 @@ def process_inbox(
                     _validate_facets(structure, profile)
                 outputs.append(
                     (
-                        name,
+                        targets[0],
                         render_inbox_concept(
                             decision.sanitized_text,
                             okf_type=okf_type,
@@ -582,7 +639,12 @@ def process_inbox(
             # creates one. Without this the very first hierarchical write fails.
             (bundle / target_name).parent.mkdir(parents=True, exist_ok=True)
             written = write_bytes(bundle, target_name, content)
-            persisted.append(PersistedFile(source_file=path.name, path=written, reasons=reasons))
+            concepts.append(PersistedFile(source_file=path.name, path=written, reasons=reasons))
+        # One entry per SOURCE FILE, whatever the document expanded into. That
+        # is what `persisted` has always meant, so an existing consumer's count
+        # does not change under a profile that segments.
+        if outputs:
+            persisted.append(concepts[-len(outputs)])
 
     # §6 index — the last disk mutation, and only when something was written.
     if persisted:
@@ -592,7 +654,7 @@ def process_inbox(
         else:
             _refresh_root_frontmatter(index_path, root_head)
         if profile.index.facets is None:
-            for entry in persisted:
+            for entry in concepts:
                 link_in_index(
                     bundle,
                     entry.path.name,
@@ -607,6 +669,7 @@ def process_inbox(
         quarantined=tuple(quarantined),
         rejected=tuple(rejected),
         failed=tuple(sorted(failed, key=lambda entry: entry.source_file)),
+        concepts=tuple(concepts),
     )
 
 

tests/test_segmented_collisions.py

@@ -0,0 +1,202 @@
+"""The collision gate under 1-to-N, and reporting that stays additive.
+
+Today's gate names every dropped file BEFORE any gate call or write, so two
+files reducing to one slug are refused TOGETHER rather than letting iteration
+order decide which one survives. Under segmentation a document no longer claims
+one name -- it claims the whole SET of paths its plan expands to -- so the gate
+has to be keyed on that set or the same defect returns one level down: the
+second document silently claims the first's concepts.
+
+Two length rules that look alike and are not. `check_filename_length` measures
+ONE name against NAME_MAX, which is a per-directory-entry limit. Measuring a
+joined hierarchical path against it gets the question backwards in both
+directions: it would refuse a perfectly legal deep path, and it would accept an
+illegal component sitting in a short one.
+
+Reporting is extended ADDITIVELY. `persisted` keeps its per-source-file
+meaning, so an existing consumer reading it sees exactly what it saw before,
+and the per-concept expansion arrives as a new `concepts` field beside it.
+"""
+
+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.materialize import NAME_MAX_BYTES
+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"
+
+# Long enough that every span this module declares (`index * 10` onward)
+# lands inside it -- a span past the end is a different refusal, and would
+# mask the collision these tests are about.
+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, paths: tuple[str, ...], **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": 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",
+):
+    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 tree(bundle: Path) -> dict[str, bytes]:
+    if not bundle.is_dir():
+        return {}
+    return {
+        str(path.relative_to(bundle)): path.read_bytes()
+        for path in sorted(bundle.rglob("*"))
+        if path.is_file()
+    }
+
+
+# --- the gate, keyed on the whole set of segment paths --------------------
+
+
+def test_two_documents_claiming_one_segment_path_are_both_refused(tmp_path: Path) -> None:
+    # Identical bytes, so ONE plan covers both documents and both expand onto
+    # the same paths. Refused together, before any gate call or write.
+    drop(tmp_path / "round", "n500.md")
+    drop(tmp_path / "round", "v720.md")
+    plan = build_plan(
+        DOCUMENT.encode("utf-8"), ("krav/3-1/brannkonsept.md", "krav/3-2/roemning.md")
+    )
+
+    result = run(tmp_path, plan=plan)
+
+    assert {entry.error.code for entry in result.failed} == {"inbox_slug_collision"}
+    assert {entry.source_file for entry in result.failed} == {"n500.md", "v720.md"}
+    assert tree(tmp_path / "bundle") == {}
+    assert result.persisted == ()
+
+
+def test_the_collision_message_names_the_contested_path(tmp_path: Path) -> None:
+    drop(tmp_path / "round", "n500.md")
+    drop(tmp_path / "round", "v720.md")
+    plan = build_plan(DOCUMENT.encode("utf-8"), ("krav/3-1/brannkonsept.md",))
+    result = run(tmp_path, plan=plan)
+    assert all("krav/3-1/brannkonsept.md" in str(entry.error) for entry in result.failed)
+
+
+def test_a_flat_run_still_refuses_two_names_reducing_to_one_slug(tmp_path: Path) -> None:
+    drop(tmp_path / "round", "note.md", "a\n")
+    drop(tmp_path / "round", "note.txt", "b\n")
+    result = run(tmp_path, profile=DEFAULT)
+    assert {entry.error.code for entry in result.failed} == {"inbox_slug_collision"}
+    assert tree(tmp_path / "bundle") == {}
+
+
+# --- the length rule is PER COMPONENT -------------------------------------
+
+
+def test_a_single_over_long_component_is_refused(tmp_path: Path) -> None:
+    source = drop(tmp_path / "round", "n500.md")
+    too_long = "a" * (NAME_MAX_BYTES + 1)
+    plan = build_plan(source.read_bytes(), (f"krav/{too_long}.md",))
+    result = run(tmp_path, plan=plan)
+    assert {entry.error.code for entry in result.failed} == {"inbox_slug_too_long"}
+    assert tree(tmp_path / "bundle") == {}
+
+
+def test_a_joined_path_over_name_max_with_legal_components_is_accepted(tmp_path: Path) -> None:
+    # The check the joined measurement gets backwards. Every component here is
+    # well under NAME_MAX; the joined path is well over it, and the filesystem
+    # does not care -- NAME_MAX is a per-entry limit.
+    source = drop(tmp_path / "round", "n500.md")
+    deep = "/".join(f"niva-{index}-{'x' * 40}" for index in range(6))
+    target = f"{deep}/krav.md"
+    assert len(target.encode("utf-8")) > NAME_MAX_BYTES
+    assert all(len(part.encode("utf-8")) <= NAME_MAX_BYTES for part in target.split("/"))
+
+    result = run(tmp_path, plan=build_plan(source.read_bytes(), (target,)))
+    assert result.failed == ()
+    assert target in tree(tmp_path / "bundle")
+
+
+# --- additive reporting ---------------------------------------------------
+
+
+def test_concepts_holds_one_entry_per_concept_and_persisted_one_per_source(
+    tmp_path: Path,
+) -> None:
+    source = drop(tmp_path / "round", "n500.md")
+    plan = build_plan(source.read_bytes(), ("krav/a.md", "krav/b.md", "krav/c.md"))
+    result = run(tmp_path, plan=plan)
+
+    assert len(result.concepts) == 3
+    assert {entry.path.name for entry in result.concepts} == {"a.md", "b.md", "c.md"}
+    # `persisted` keeps its existing meaning: one entry per SOURCE FILE. A
+    # consumer reading it sees exactly what it saw before segmentation existed.
+    assert len(result.persisted) == 1
+    assert result.persisted[0].source_file == "n500.md"
+
+
+def test_under_default_the_two_fields_agree(tmp_path: Path) -> None:
+    drop(tmp_path / "round", "n500.md")
+    drop(tmp_path / "round", "v720.md", "annet\n")
+    result = run(tmp_path, profile=DEFAULT)
+    assert len(result.persisted) == 2
+    assert result.concepts == result.persisted
+
+
+def test_concepts_names_every_segment_path_it_wrote(tmp_path: Path) -> None:
+    source = drop(tmp_path / "round", "n500.md")
+    plan = build_plan(source.read_bytes(), ("krav/3-1/a.md", "krav/3-2/b.md"))
+    result = run(tmp_path, plan=plan)
+    bundle = tmp_path / "bundle"
+    assert {str(entry.path.relative_to(bundle)) for entry in result.concepts} == {
+        "krav/3-1/a.md",
+        "krav/3-2/b.md",
+    }
+    assert all(entry.source_file == "n500.md" for entry in result.concepts)