feat(inbox): accept one segmentation plan per document

src/llm_ingestion_okf/inbox.py

@@ -21,7 +21,7 @@ from __future__ import annotations
 
 import hashlib
 import unicodedata
-from collections.abc import Callable, Mapping
+from collections.abc import Callable, Mapping, Sequence
 from dataclasses import dataclass, replace
 from pathlib import Path, PurePosixPath
 
@@ -387,19 +387,61 @@ def _bundle_id_key(profile: BundleProfile) -> str:
     return profile.segmentation.bundle_id_key
 
 
-def _plan_covering(plan: SegmentationPlan | None, source_bytes: bytes) -> SegmentationPlan | None:
+def _resolve_plans(
+    segmentation: SegmentationPlan | None,
+    segmentations: Mapping[str, SegmentationPlan] | None,
+) -> tuple[SegmentationPlan, ...]:
+    """The plans this run replays, from either call form, never from both.
+
+    The mapping is keyed by SOURCE FILENAME because that is the name an
+    operator reads and maintains, but the key never selects anything -- see
+    :func:`_plan_covering`. Keeping selection on content identity is what lets
+    a renamed file still find its plan, and stops a plan filed under the wrong
+    name from segmenting the wrong document.
+
+    Both forms at once is REFUSED rather than merged. They are two ways to say
+    the same thing, and merging them would let a caller hold a plan in each and
+    never learn the two disagreed.
+    """
+    if segmentation is not None and segmentations:
+        raise SegmentationError(
+            "both `segmentation` and `segmentations` were given — pass one form or the "
+            "other; merging them would hide a disagreement between two plans for the "
+            "same document",
+            code="segmentation_plan_invalid",
+        )
+    plans = tuple(segmentations.values()) if segmentations else ()
+    if segmentation is not None:
+        plans = (segmentation,)
+
+    seen: dict[str, SegmentationPlan] = {}
+    for plan in plans:
+        if plan.source_sha256 in seen and plan is not seen[plan.source_sha256]:
+            raise SegmentationError(
+                f"two segmentation plans claim source_sha256 {plan.source_sha256!r} — "
+                "selection is by content identity, so which one segmented the document "
+                "would depend on mapping order; refusing rather than picking one",
+                code="segmentation_plan_invalid",
+            )
+        seen[plan.source_sha256] = plan
+    return plans
+
+
+def _plan_covering(
+    plans: Sequence[SegmentationPlan], source_bytes: bytes
+) -> SegmentationPlan | None:
     """The plan for THESE bytes, or None when this file is not plan-covered.
 
     Selection is by content hash, so a run may drop several documents while
-    only one of them is segmented; every other file keeps today's one-concept
+    only some of them are segmented; every other file keeps today's one-concept
     rule verbatim. The hash SELECTS; :func:`assert_plan_applies` VALIDATES,
     and the two are deliberately different questions -- see S5b.
     """
-    if plan is None:
-        return None
-    if plan.source_sha256 != hashlib.sha256(source_bytes).hexdigest():
-        return None
-    return plan
+    digest = hashlib.sha256(source_bytes).hexdigest()
+    for plan in plans:
+        if plan.source_sha256 == digest:
+            return plan
+    return None
 
 
 def _render_segments(
@@ -532,6 +574,7 @@ def process_inbox(
     profile: BundleProfile = DEFAULT,
     root_frontmatter_values: Mapping[str, str] | None = None,
     segmentation: SegmentationPlan | None = None,
+    segmentations: Mapping[str, SegmentationPlan] | None = None,
 ) -> InboxResult:
     """Convert every file dropped in `inbox_dir` into an OKF concept.
 
@@ -555,7 +598,8 @@ def process_inbox(
     # `materialize.py` states the same rule for Door A, and a door that half-built
     # a bundle before refusing would be worse than one that never started.
     root_head = _render_root_frontmatter(root_frontmatter_values or {}, profile=profile)
-    if segmentation is not None:
+    plans = _resolve_plans(segmentation, segmentations)
+    if plans:
         if profile.segmentation is None:
             raise SegmentationError(
                 "a segmentation plan was passed to a profile that does not declare the "
@@ -602,7 +646,10 @@ def process_inbox(
     # hash matched a drop but whose entry paths were then refused is a covered
     # document with a bad plan, and it must keep reporting its own per-file
     # code rather than being re-reported as a plan that matched nothing.
-    plan_matched = False
+    # WHICH plans matched, not merely that one did: a corpus run where four of
+    # five plans matched would otherwise report success over four segmented
+    # documents and one silently flat one.
+    matched_hashes: set[str] = set()
     for path in dropped:
         try:
             # Read HERE rather than in the write loop: a plan is selected by
@@ -621,8 +668,9 @@ def process_inbox(
             )
             continue
         try:
-            covering = _plan_covering(segmentation, source_bytes)
-            plan_matched = plan_matched or covering is not None
+            covering = _plan_covering(plans, source_bytes)
+            if covering is not None:
+                matched_hashes.add(covering.source_sha256)
             targets: tuple[str, ...]
             if covering is None:
                 targets = (inbox_filename(inbox_slug(path.name), profile=profile),)
@@ -643,11 +691,13 @@ def process_inbox(
     # rather than "was every file examined?", so a file that could not be read
     # cannot mask the refusal. Still before any disk mutation: Phase 1 only
     # named things.
-    if segmentation is not None and not plan_matched:
+    unmatched = sorted(set(plan.source_sha256 for plan in plans) - matched_hashes)
+    if unmatched:
         raise SegmentationError(
-            f"the segmentation plan's source_sha256 {segmentation.source_sha256!r} matches "
-            f"none of the {len(dropped)} dropped file(s) — nothing would be segmented and "
-            "the run would report success over a flat bundle; check the hash against the "
+            f"{len(unmatched)} of {len(plans)} segmentation plan(s) match none of the "
+            f"{len(dropped)} dropped file(s) — first unmatched source_sha256 "
+            f"{unmatched[0]!r}; nothing would be segmented for those documents and the "
+            "run would report success over a flat bundle; check each hash against the "
             "bytes it was adjudicated over",
             code="segmentation_plan_unmatched",
         )
@@ -728,7 +778,7 @@ def process_inbox(
             text = extract_text(
                 path.name, source_bytes, renderer=_resolve_renderer(profile, path.name)
             )
-            covering = _plan_covering(segmentation, source_bytes)
+            covering = _plan_covering(plans, source_bytes)
             if covering is not None:
                 blocked = _render_segments(
                     covering,