feat(inbox): own concepts by source_file and retire stale segments

src/llm_ingestion_okf/inbox.py

@@ -281,6 +281,52 @@ def _is_inbox_owned(path: Path) -> bool:
     return frontmatter.get("generated") == "true" and "source_file" in frontmatter
 
 
+def _owned_concepts_by_source(bundle: Path, profile: BundleProfile) -> dict[str, set[str]]:
+    """Every inbox-owned concept in the bundle, grouped by the `source_file` it came from.
+
+    The question ownership has to answer under 1-to-N. Keyed on the concept
+    FILENAME -- which is what it was -- a round that re-adjudicates a document
+    into fewer segments ORPHANS the ones it no longer names: they survive in the
+    incremental bundle, are absent from a scratch rebuild, and the two diverge
+    with nothing failing. Grouping by `source_file` is what makes "these are the
+    concepts this document currently owns" expressible at all.
+
+    Recursive, and reached only when the profile declares the capability: the
+    four shipped profiles keep flat scans, because switching the shared glob
+    would change their ownership behaviour and break their byte-stability pin.
+    """
+    owned: dict[str, set[str]] = {}
+    if not bundle.is_dir():
+        return owned
+    for path in sorted(bundle.rglob(f"*{profile.paths.concept_suffix}")):
+        if not path.is_file() or path.name == profile.index.name:
+            continue
+        frontmatter = parse_frontmatter(path)
+        if frontmatter.get("generated") != "true":
+            continue
+        source_file = frontmatter.get("source_file")
+        if source_file is not None:
+            owned.setdefault(source_file, set()).add(path.relative_to(bundle).as_posix())
+    return owned
+
+
+def _retire_stale_segments(bundle: Path, stale: set[str]) -> None:
+    """Delete concepts this round's plan no longer names, and prune what empties.
+
+    An emptied directory is deleted too: a scratch rebuild never creates it, so
+    leaving it behind is exactly the kind of one-sided difference `diff -r`
+    reports and S7 refuses.
+    """
+    for relative in sorted(stale):
+        target = bundle / relative
+        target.unlink(missing_ok=True)
+    for relative in sorted(stale, key=lambda item: item.count("/"), reverse=True):
+        parent = (bundle / relative).parent
+        while parent != bundle and parent.is_dir() and not any(parent.iterdir()):
+            parent.rmdir()
+            parent = parent.parent
+
+
 def _check_segment_path(path: str) -> str:
     """Refuse a segment path the filesystem cannot hold, measured PER COMPONENT.
 
@@ -467,7 +513,7 @@ def process_inbox(
     # 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]] = []
+    named: list[tuple[Path, tuple[str, ...], bytes, bool]] = []
     slug_owners: dict[str, list[Path]] = {}
     for path in dropped:
         try:
@@ -496,7 +542,7 @@ def process_inbox(
         except IngestError as exc:
             failed.append(FailedFile(source_file=path.name, error=exc))
             continue
-        named.append((path, targets, source_bytes))
+        named.append((path, targets, source_bytes, covering is not None))
         for target in targets:
             slug_owners.setdefault(target, []).append(path)
 
@@ -527,18 +573,34 @@ def process_inbox(
     # BEFORE this run — a file written below must never be mistaken for
     # pre-existing curated content by a later file's check.
     bundle = Path(bundle_dir)
-    pre_existing = (
-        {
-            path.name
-            for path in bundle.glob(f"*{profile.paths.concept_suffix}")
-            if path.name != profile.index.name
-        }
-        if bundle.is_dir()
-        else set()
-    )
+    owned_by_source: dict[str, set[str]] = {}
+    if profile.segmentation is not None:
+        # RECURSIVE, and only here. A nested concept is invisible to a flat
+        # glob, so our own file would look like curated content and the §3
+        # collision gate would fire on it.
+        owned_by_source = _owned_concepts_by_source(bundle, profile)
+        pre_existing = (
+            {
+                path.relative_to(bundle).as_posix()
+                for path in bundle.rglob(f"*{profile.paths.concept_suffix}")
+                if path.is_file() and path.name != profile.index.name
+            }
+            if bundle.is_dir()
+            else set()
+        )
+    else:
+        pre_existing = (
+            {
+                path.name
+                for path in bundle.glob(f"*{profile.paths.concept_suffix}")
+                if path.name != profile.index.name
+            }
+            if bundle.is_dir()
+            else set()
+        )
     owned = {name for name in pre_existing if _is_inbox_owned(bundle / name)}
 
-    for path, targets, source_bytes in named:
+    for path, targets, source_bytes, segmented_file in named:
         if any(name in contested for name in targets):
             continue
         unstamped = [name for name in targets if name in pre_existing and name not in owned]
@@ -645,6 +707,11 @@ def process_inbox(
         # does not change under a profile that segments.
         if outputs:
             persisted.append(concepts[-len(outputs)])
+        if segmented_file:
+            # Round N owns exactly what round N's plan names. Everything this
+            # document owned before and does not now is retired HERE, after the
+            # writes, so a failure above leaves the previous round intact.
+            _retire_stale_segments(bundle, owned_by_source.get(path.name, set()) - set(targets))
 
     # §6 index — the last disk mutation, and only when something was written.
     if persisted:
@@ -743,10 +810,24 @@ def _reproject_index(bundle: Path, profile: BundleProfile) -> None:
     """
     assert profile.index.facets is not None
     documents: dict[str, DocumentStructure] = {}
-    for path in sorted(bundle.glob(f"*{profile.paths.concept_suffix}")):
-        if path.name == profile.index.name or not _is_inbox_owned(path):
+    # M4's second flat glob. Under the capability a concept lives at a nested
+    # path, and a flat scan would reproject an index that silently omits every
+    # one of them -- the reprojection is the whole-bundle recompute that makes
+    # rebuild equal an incremental update, so a scan that cannot see a file
+    # makes that equality false rather than merely incomplete.
+    segmented = profile.segmentation is not None
+    found = (
+        bundle.rglob(f"*{profile.paths.concept_suffix}")
+        if segmented
+        else bundle.glob(f"*{profile.paths.concept_suffix}")
+    )
+    for path in sorted(found):
+        if not path.is_file() or path.name == profile.index.name or not _is_inbox_owned(path):
             continue
-        documents[path.name] = structure_from_frontmatter(parse_frontmatter(path))
+        # The concept ID is the bundle-relative path (OKF v0.2 §2), and two
+        # files called `a.md` in different directories are two concepts.
+        key = path.relative_to(bundle).as_posix() if segmented else path.name
+        documents[key] = structure_from_frontmatter(parse_frontmatter(path))
 
     resolved = resolve_structure(documents)
     block = [