fix(inbox): refuse a segmentation plan that matches no dropped file

A plan is selected by content hash, so a mistyped source_sha256 matched
nothing, every dropped file fell through to the one-concept rule, and
process_inbox returned an ordinary success over a flat bundle. The
operator asked for segmentation, got none, and had no error to read --
the silent skip this library refuses everywhere else. vegnormal-okf is
about to run an N500 corpus through this path, where a silent zero
would read as "the corpus has no concepts".

The refusal asks whether a covering plan was FOUND, not whether every
file was examined, so an unreadable drop cannot mask it; and coverage
is recorded at selection, not after path validation, so a matched plan
with a refused entry path still reports its own per-file code. The
first cut got that second question wrong and an existing collision test
caught it; the case is now pinned by its own test, verified red against
the earlier form.

New code segmentation_plan_unmatched, registered in the SegmentationError
docstring register in the same commit. Fail-fast before any disk
mutation. The four existing goldens are byte-identical to baseline.

Co-Authored-By: Claude <claude-opus-5>

src/llm_ingestion_okf/errors.py

@@ -177,6 +177,12 @@ class SegmentationError(IngestError):
       offset had silently moved
     - `segmentation_unsupported_profile` — a plan was passed to a profile that
       does not declare the segmentation capability
+    - `segmentation_plan_unmatched` — the plan is well-formed but its
+      `source_sha256` matches no dropped file, so nothing would be segmented
+      and the run would report an ordinary success over a flat bundle. A
+      mistyped hash is the likely cause and it is unreadable from the result;
+      refusing is the only way the operator learns that the judgement they
+      adjudicated was never replayed
     """
 
 

src/llm_ingestion_okf/inbox.py

@@ -531,6 +531,11 @@ def process_inbox(
     # would silently claim the first's concepts.
     named: list[tuple[Path, tuple[str, ...], bytes, bool]] = []
     slug_owners: dict[str, list[Path]] = {}
+    # Recorded at the moment of SELECTION, not after validation: a plan whose
+    # 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
     for path in dropped:
         try:
             # Read HERE rather than in the write loop: a plan is selected by
@@ -550,6 +555,7 @@ def process_inbox(
             continue
         try:
             covering = _plan_covering(segmentation, source_bytes)
+            plan_matched = plan_matched or covering is not None
             targets: tuple[str, ...]
             if covering is None:
                 targets = (inbox_filename(inbox_slug(path.name), profile=profile),)
@@ -562,6 +568,23 @@ def process_inbox(
         for target in targets:
             slug_owners.setdefault(target, []).append(path)
 
+    # A plan that covered nothing is a misuse, not an outcome. `_plan_covering`
+    # selects on content hash, so a mistyped `source_sha256` matches no drop,
+    # every file falls through to the one-concept rule, and the run returns an
+    # ordinary success over a flat bundle -- the silent skip this library
+    # refuses everywhere else. Asked as "was a covering plan actually found?"
+    # 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:
+        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 "
+            "bytes it was adjudicated over",
+            code="segmentation_plan_unmatched",
+        )
+
     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

tests/test_segmented_inbox.py

@@ -347,3 +347,97 @@ def test_a_plan_covering_one_of_two_documents_leaves_the_other_flat(tmp_path: Pa
     assert set(PATHS) <= set(found)
     assert "inbox-v720.md" in found
     assert len(found) == 6
+
+
+# --- a plan that matches nothing is refused, never silent ------------------
+
+
+def test_a_plan_matching_no_dropped_file_is_refused(tmp_path: Path) -> None:
+    """The silent skip this library refuses everywhere else.
+
+    `_plan_covering` selects on content hash, so a mistyped `source_sha256`
+    matches nothing, every dropped file falls through to the one-concept rule,
+    and the run reports a perfectly ordinary success. The operator asked for
+    segmentation and got a flat bundle with no error to read. `vegnormal-okf`
+    is about to put an N500 corpus through this path; a silent zero there would
+    read as "the corpus has no concepts".
+    """
+    source = drop(tmp_path / "round", "n500.md", DOCUMENT)
+    plan = build_plan(source.read_bytes(), DOCUMENT, source_sha256="0" * 64)
+    with pytest.raises(SegmentationError) as excinfo:
+        run(tmp_path, plan=plan)
+    assert excinfo.value.code == "segmentation_plan_unmatched"
+    assert tree(tmp_path / "bundle") == {}
+
+
+def test_a_plan_matching_one_of_several_dropped_files_is_not_refused(tmp_path: Path) -> None:
+    """The negative control for the check above.
+
+    A run may legitimately drop many documents while only one is plan-covered
+    -- that is the whole point of hash selection. A check that fired here would
+    have replaced a silent skip with a refusal of the normal case, so this test
+    is what keeps the new gate honest rather than merely loud.
+    """
+    source = drop(tmp_path / "round", "n500.md", DOCUMENT)
+    drop(tmp_path / "round", "v720.md", "V720 Tunnel: egne krav.\n")
+    run(tmp_path, plan=build_plan(source.read_bytes(), DOCUMENT))
+    assert len(concepts(tmp_path / "bundle")) == 6
+
+
+def test_an_unreadable_dropped_file_does_not_mask_an_unmatched_plan(tmp_path: Path) -> None:
+    """A file the run could not read is not evidence that the plan matched.
+
+    Phase 1 skips a file it cannot read and records a `FailedFile`. If the
+    unmatched-plan check asked "was any file left unexamined?" instead of "was
+    a covering plan actually found?", an unreadable drop would suppress the
+    refusal and restore exactly the silence this closes. The plan here is
+    hashed over the unreadable file's OWN bytes, so it is the only drop that
+    could ever have matched.
+    """
+    drop(tmp_path / "round", "n500.md", DOCUMENT)
+    other = "V720 Tunnel: egne krav.\n"
+    unreadable = drop(tmp_path / "round", "locked.md", other)
+    unreadable.chmod(0o000)
+    try:
+        # The premise, asserted rather than assumed: a test that ran as root
+        # would read the file fine and pass for the wrong reason.
+        with pytest.raises(OSError):
+            unreadable.read_bytes()
+        plan = build_plan(other.encode("utf-8"), other, paths=("krav/tunnel.md",))
+        with pytest.raises(SegmentationError) as excinfo:
+            run(tmp_path, plan=plan)
+        assert excinfo.value.code == "segmentation_plan_unmatched"
+        assert tree(tmp_path / "bundle") == {}
+    finally:
+        unreadable.chmod(0o644)
+
+
+def test_a_matched_plan_with_a_refused_path_keeps_its_own_per_file_code(tmp_path: Path) -> None:
+    """A matched plan with bad entries is not an unmatched plan.
+
+    Written because the first cut of the check got this backwards: it asked
+    whether any document reached the naming stage as covered, so a plan whose
+    hash matched but whose entry paths were then refused looked identical to a
+    plan that matched nothing. The operator would have been told to check a
+    hash that was already correct. Coverage is recorded at SELECTION.
+    """
+    source = drop(tmp_path / "round", "n500.md", DOCUMENT)
+    plan = build_plan(source.read_bytes(), DOCUMENT, paths=("krav/" + "a" * 300 + ".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_without_the_capability_an_unmatched_plan_is_still_the_earlier_refusal(
+    tmp_path: Path,
+) -> None:
+    """Order matters: the profile check runs first and keeps its own code.
+
+    Both conditions hold in this call -- no capability AND no matching file --
+    and the operator's first problem is the profile, not the hash.
+    """
+    source = drop(tmp_path / "round", "n500.md", DOCUMENT)
+    plan = build_plan(source.read_bytes(), DOCUMENT, source_sha256="0" * 64)
+    with pytest.raises(SegmentationError) as excinfo:
+        run(tmp_path, plan=plan, profile=DEFAULT, values={})
+    assert excinfo.value.code == "segmentation_unsupported_profile"