feat(inbox): Door B flow against an injected guard gate (Phase 2 step 3)

`process_inbox(inbox_dir, bundle_dir, ingested_at, *, okf_type, gate)`:
per dropped file, bytes -> extract_text -> guard gate -> render -> collision
gate -> write -> index link. Returns an InboxResult whose four buckets
(persisted / quarantined / rejected / failed) are disjoint and complete, so a
file that vanished shows up as a missing entry rather than as nothing.

The guard is INJECTED rather than imported. The library calls no guard
function and makes no security decision: the Gate adapter returns a
GateDecision carrying the guard's own disposition value and the sanitized
text, and the flow only branches on it. That keeps the core dependency-free
while B2 (the guard's CI channel) is still open, and lets the persist and
refuse branches run deterministically against a test double.

Decisions taken with the operator this session:

- What is persisted is the gate's SANITIZED text, not the extracted text.
  Screening one string and writing another would make the verdict a statement
  about bytes nobody kept. `sanitize` is exported by the guard precisely for
  callers composing the checklist themselves, so this is sanctioned API, not
  a reimplementation. Door B has no model call, so the fenced text the
  bookends produce for a transform is never persisted.
- Quarantine is reported apart from rejection. QUARANTINE_REVIEW means "hold
  for human review" — an operator queue — where FAIL_SECURE is a decision.
  No quarantine directory in v1; that stays an extension point.

Fails closed by construction: only the guard's non-blocking floor (`warn`)
persists, so a renamed member, a future disposition or an adapter typo lands
in `rejected` rather than being guessed safe.

One bad file never aborts the run. Only three conditions fail the whole run,
and each is wrong for every file at once: an invalid `ingested_at`, a reserved
`okf_type`, and a missing inbox directory.

New code `inbox_slug_collision`: two dropped files reducing to one generated
name are BOTH refused. Persisting one would let iteration order decide the
winner, and overwriting would lose the other's content.

Phase 1 primitives are reused, never duplicated, so four helpers become
package-internal names (write_bytes, link_in_index, parse_frontmatter,
INDEX_NAME) and link_in_index learns to append to an empty index — Door A
always seeds its index with bundle_summary, but Door B has no summary to
invent and must not open with a blank line.

src/llm_ingestion_okf/__init__.py

@@ -15,6 +15,11 @@ what it is given. The guard-calling persist gates arrive with Doors B and C.
 
 Door A (spec-based ingestion) public surface: materialize_bundle plus the
 typed error hierarchy rooted in IngestError.
+
+Door B (bundle inbox) public surface: process_inbox plus its result types.
+Its persist gate is INJECTED -- the caller supplies a Gate adapter over
+llm-ingestion-guard and process_inbox obeys the verdict; the library imports
+no guard function itself and makes no security decision of its own.
 """
 
 from .errors import (
@@ -27,6 +32,15 @@ from .errors import (
     SourceError,
 )
 from .extract import extract_text
+from .inbox import (
+    BlockedFile,
+    FailedFile,
+    Gate,
+    GateDecision,
+    InboxResult,
+    PersistedFile,
+    process_inbox,
+)
 from .manifest import (
     Extraction,
     FileSource,
@@ -40,20 +54,27 @@ from .materialize import IngestResult, materialize_bundle
 __version__ = "0.3.2"
 
 __all__ = [
+    "BlockedFile",
     "Extraction",
     "ExtractionError",
+    "FailedFile",
     "FileSource",
+    "Gate",
+    "GateDecision",
     "HttpSource",
+    "InboxResult",
     "IngestError",
     "IngestResult",
     "Manifest",
     "ManifestError",
     "MaterializationError",
     "NetworkGateError",
+    "PersistedFile",
     "RenderError",
     "SourceError",
     "SqlSource",
     "extract_text",
     "load_manifest",
     "materialize_bundle",
+    "process_inbox",
 ]

src/llm_ingestion_okf/errors.py

@@ -101,6 +101,9 @@ class MaterializationError(IngestError):
     - `inbox_slug_too_long` — the generated inbox filename would exceed the
       255-byte filesystem limit (Door B; never a truncated name, which would
       be lossy and could collide with another long name sharing its prefix)
+    - `inbox_slug_collision` — two files dropped in the same run reduce to one
+      generated filename (Door B); both are refused rather than letting
+      iteration order decide which one survives
     - `inbox_title_invalid` — an inbox title is multi-line or contains `[`/`]`,
       either of which would break frontmatter or an index link
     - `inbox_source_file_invalid` — an inbox `source_file` is multi-line and

src/llm_ingestion_okf/inbox.py

@@ -1,4 +1,4 @@
-"""Door B inbox provenance rendering and filename slugging (Phase 2 step 2).
+"""Door B: the bundle inbox — provenance rendering, slugging, and the flow.
 
 Pure functions on top of the Phase 1 primitives: a dropped file's name is
 reduced to the Phase 1 id grammar and namespaced `inbox-{slug}.md` — disjoint
@@ -11,8 +11,10 @@ extracted text, so provenance stays re-verifiable against the operator's file.
 `ingested_at` is explicit and validated by the same rule as Door A: no
 wall-clock anywhere. Output is LF-only with exactly one trailing newline.
 
-No guard call and no model call in this module — the persist gate arrives with
+`process_inbox` is the flow: per file, extracted text goes through the guard
-the inbox flow (step 3).
+gate before anything is written, and only the gate's non-blocking floor
+persists. No model call anywhere, and no security decision here — the gate
+supplies the verdict and this module only obeys it.
 """
 
 from __future__ import annotations
@@ -20,10 +22,19 @@ from __future__ import annotations
 import hashlib
 import re
 import unicodedata
+from collections.abc import Callable
+from dataclasses import dataclass
 from pathlib import Path
 
-from .errors import MaterializationError
+from .errors import IngestError, MaterializationError, SourceError
-from .materialize import validate_ingested_at
+from .extract import extract_text
+from .materialize import (
+    INDEX_NAME,
+    link_in_index,
+    parse_frontmatter,
+    validate_ingested_at,
+    write_bytes,
+)
 
 _RESERVED_OKF_TYPE = "verdict"
 
@@ -156,3 +167,253 @@ def render_inbox_concept(
     }
     rendered = "\n".join(f"{key}: {value}" for key, value in frontmatter.items())
     return f"---\n{rendered}\n---\n\n{_normalize_body(text)}"
+
+
+# --- the guard seam -------------------------------------------------------
+
+# The guard's non-blocking floor. `Disposition` is a `str, Enum` in
+# llm-ingestion-guard, so its VALUE is the stable thing to compare against
+# across the pinned `>=0.2,<0.3` range. Pinned as a constant here rather than
+# imported, because the dependency is injected (see `Gate`): the step-4
+# adapter's signature smoke test is what catches a rename in the guard.
+_DISPOSITION_PERSIST = "warn"
+_DISPOSITION_QUARANTINE = "quarantine_review"
+
+
+@dataclass(frozen=True)
+class GateDecision:
+    """One guard verdict, carried verbatim from `llm-ingestion-guard`.
+
+    `sanitized_text` is what the gate actually screened, and therefore the only
+    text that may be persisted — screening one string and writing another would
+    make the verdict a statement about bytes nobody kept. `disposition` is the
+    guard's `Disposition` VALUE and `reasons` its audit trail; neither is
+    interpreted here beyond the single persist/do-not-persist branch.
+    """
+
+    sanitized_text: str
+    disposition: str
+    reasons: tuple[str, ...] = ()
+
+
+# The persist gate, injected. The library never imports the guard directly:
+# the caller supplies the adapter, which keeps this module dependency-free and
+# lets the flow's branches be exercised deterministically by a test double.
+Gate = Callable[[str], GateDecision]
+
+
+# --- per-file outcomes ----------------------------------------------------
+
+
+@dataclass(frozen=True)
+class PersistedFile:
+    """A dropped file that cleared the gate and was written."""
+
+    source_file: str
+    path: Path
+    reasons: tuple[str, ...] = ()
+
+
+@dataclass(frozen=True)
+class BlockedFile:
+    """A dropped file the guard refused. Quarantine and rejection are reported
+    separately: quarantine means "hold for human review" and is the operator's
+    queue, while a fail-secure verdict is a decision, not a queue."""
+
+    source_file: str
+    disposition: str
+    reasons: tuple[str, ...]
+
+
+@dataclass(frozen=True)
+class FailedFile:
+    """A dropped file this library could not process — extraction, filename or
+    collision. Always a typed error, never a leaked stdlib exception."""
+
+    source_file: str
+    error: IngestError
+
+
+@dataclass(frozen=True)
+class InboxResult:
+    """Every dropped file's outcome, in sorted filename order.
+
+    Four disjoint buckets, and every file lands in exactly one: a run's report
+    is complete by construction, so a file that silently vanished would show up
+    as a missing entry rather than as nothing at all.
+    """
+
+    persisted: tuple[PersistedFile, ...]
+    quarantined: tuple[BlockedFile, ...]
+    rejected: tuple[BlockedFile, ...]
+    failed: tuple[FailedFile, ...]
+
+
+def _is_inbox_owned(path: Path) -> bool:
+    # Door B's ownership marker: `generated: true` AND a `source_file`
+    # reference. Deliberately disjoint from Door A's test, which keys on
+    # `ingest_manifest` — a Door A concept is never this door's to replace,
+    # and curated content carries neither key.
+    frontmatter = parse_frontmatter(path)
+    return frontmatter.get("generated") == "true" and "source_file" in frontmatter
+
+
+def process_inbox(
+    inbox_dir: Path,
+    bundle_dir: Path,
+    ingested_at: str,
+    *,
+    okf_type: str,
+    gate: Gate,
+) -> InboxResult:
+    """Convert every file dropped in `inbox_dir` into an OKF concept.
+
+    An explicit operator command, never a watcher or a scheduler (spec §9's
+    human-in-the-loop rule). `ingested_at` is required and stamped verbatim, as
+    at Door A. `gate` is the guard adapter: extracted text is screened before
+    anything is written, and only the guard's non-blocking floor persists —
+    anything else, INCLUDING a disposition this library does not recognise,
+    fails closed.
+
+    One bad file never aborts the run. Extraction failures, unusable filenames
+    and collisions are reported per file in :class:`InboxResult` while the
+    remaining files still process. Only three conditions fail the whole run,
+    and all three are wrong for every file at once: an invalid `ingested_at`,
+    a reserved `okf_type`, and a missing inbox directory.
+    """
+    validate_ingested_at(ingested_at)
+    if okf_type.lower() == _RESERVED_OKF_TYPE:
+        raise MaterializationError(
+            f"okf_type must not be {_RESERVED_OKF_TYPE!r} (reserved layer)",
+            code="okf_type_reserved",
+        )
+    inbox = Path(inbox_dir)
+    if not inbox.is_dir():
+        raise SourceError(f"inbox directory does not exist: {inbox}", code="source_root_missing")
+
+    # Top-level only, sorted: the operator's nested structure is theirs, and a
+    # deterministic order is what makes a re-run comparable.
+    dropped = sorted((path for path in inbox.iterdir() if path.is_file()), key=lambda p: p.name)
+
+    persisted: 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.
+    named: list[tuple[Path, str]] = []
+    slug_owners: dict[str, list[Path]] = {}
+    for path in dropped:
+        try:
+            name = inbox_filename(inbox_slug(path.name))
+        except IngestError as exc:
+            failed.append(FailedFile(source_file=path.name, error=exc))
+            continue
+        named.append((path, name))
+        slug_owners.setdefault(name, []).append(path)
+
+    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(
+                        f"{path.name!r} and "
+                        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",
+                    ),
+                )
+            )
+
+    # 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
+    # pre-existing curated content by a later file's check.
+    bundle = Path(bundle_dir)
+    pre_existing = (
+        {path.name for path in bundle.glob("*.md") if path.name != INDEX_NAME}
+        if bundle.is_dir()
+        else set()
+    )
+    owned = {name for name in pre_existing if _is_inbox_owned(bundle / name)}
+
+    for path, name in named:
+        if name in colliding:
+            continue
+        if name in pre_existing and name not in owned:
+            failed.append(
+                FailedFile(
+                    source_file=path.name,
+                    error=MaterializationError(
+                        f"generated filename {name!r} collides with an existing file that does "
+                        "not carry the inbox marker — refusing to overwrite curated content (§3)",
+                        code="collision_unstamped",
+                    ),
+                )
+            )
+            continue
+        try:
+            source_bytes = path.read_bytes()
+            text = extract_text(path.name, source_bytes)
+            decision = gate(text)
+            if decision.disposition != _DISPOSITION_PERSIST:
+                blocked = BlockedFile(
+                    source_file=path.name,
+                    disposition=decision.disposition,
+                    reasons=decision.reasons,
+                )
+                # Quarantine is a queue for the operator; everything else —
+                # fail-secure, or a disposition from outside the pinned range —
+                # is a refusal. Unknown values land here by construction.
+                if decision.disposition == _DISPOSITION_QUARANTINE:
+                    quarantined.append(blocked)
+                else:
+                    rejected.append(blocked)
+                continue
+            title = unicodedata.normalize("NFC", path.stem)
+            content = render_inbox_concept(
+                decision.sanitized_text,
+                okf_type=okf_type,
+                title=title,
+                source_file=path.name,
+                source_bytes=source_bytes,
+                ingested_at=ingested_at,
+            )
+        except OSError as exc:
+            failed.append(
+                FailedFile(
+                    source_file=path.name,
+                    error=SourceError(
+                        f"cannot read dropped file {path.name}: {exc}", code="source_file_missing"
+                    ),
+                )
+            )
+            continue
+        except IngestError as exc:
+            failed.append(FailedFile(source_file=path.name, error=exc))
+            continue
+
+        bundle.mkdir(parents=True, exist_ok=True)
+        written = write_bytes(bundle, name, content)
+        persisted.append(
+            PersistedFile(source_file=path.name, path=written, reasons=decision.reasons)
+        )
+
+    # §6 index — the last disk mutation, and only when something was written.
+    if persisted:
+        index_path = bundle / INDEX_NAME
+        if not index_path.is_file():
+            write_bytes(bundle, INDEX_NAME, "")
+        for entry in persisted:
+            link_in_index(
+                bundle, entry.path.name, unicodedata.normalize("NFC", Path(entry.source_file).stem)
+            )
+
+    return InboxResult(
+        persisted=tuple(persisted),
+        quarantined=tuple(quarantined),
+        rejected=tuple(rejected),
+        failed=tuple(sorted(failed, key=lambda entry: entry.source_file)),
+    )

src/llm_ingestion_okf/materialize.py

@@ -30,7 +30,7 @@ from .render import render_fenced_block, render_table
 
 _LOGGER = logging.getLogger(__name__)
 
-_INDEX_NAME = "index.md"
+INDEX_NAME = "index.md"
 _INGESTED_AT_RE = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$")
 
 # One managed index line: `- [<label>](<target>)`. Anchored full-line —
@@ -82,7 +82,7 @@ def _render_frontmatter(frontmatter: dict[str, str]) -> str:
     )
 
 
-def _parse_frontmatter(path: Path) -> dict[str, str]:
+def parse_frontmatter(path: Path) -> dict[str, str]:
     lines = path.read_text(encoding="utf-8").splitlines()
     if not lines or lines[0].strip() != "---":
         return {}
@@ -108,7 +108,7 @@ def _is_ingest_owned(path: Path, manifest_stem: str) -> bool:
     # the same manifest wrote, while a DIFFERENT manifest sharing the bundle
     # keeps its own. rsplit strips the trailing `@{sha}`, so a stem that itself
     # contains `@` still compares correctly.
-    frontmatter = _parse_frontmatter(path)
+    frontmatter = parse_frontmatter(path)
     if frontmatter.get("generated") != "true":
         return False
     reference = frontmatter.get("ingest_manifest")
@@ -133,7 +133,7 @@ def _render_concept_file(
     return f"---\n{_render_frontmatter(frontmatter)}\n---\n\n{body}"
 
 
-def _write_bytes(bundle_dir: Path, name: str, content: str) -> Path:
+def write_bytes(bundle_dir: Path, name: str, content: str) -> Path:
     # LF-only + exactly one trailing newline are byte-level guarantees (§5),
     # so the write is raw bytes — never write_text, whose platform newline
     # translation would break golden byte-determinism.
@@ -172,14 +172,17 @@ def _update_index_lines(
         index_path.write_bytes("".join(updated).encode("utf-8"))
 
 
-def _link_in_index(bundle_dir: Path, target_name: str, label: str) -> None:
+def link_in_index(bundle_dir: Path, target_name: str, label: str) -> None:
     # §6: idempotent by target — a link whose target is already present in
     # the index is never added twice.
-    index_path = safe_resolve(bundle_dir, _INDEX_NAME)
+    index_path = safe_resolve(bundle_dir, INDEX_NAME)
     body = index_path.read_bytes().decode("utf-8")
     if f"]({target_name})" in body:
         return
-    prefix = body if body.endswith("\n") else body + "\n"
+    # An empty index needs no separator: Door A always seeds its index with
+    # bundle_summary first, but Door B has no summary to invent, so its index
+    # starts empty and must not open with a blank line.
+    prefix = body if (body == "" or body.endswith("\n")) else body + "\n"
     index_path.write_bytes(f"{prefix}- [{label}]({target_name})\n".encode())
 
 
@@ -278,7 +281,7 @@ def materialize_bundle(
     owned = {
         path.name
         for path in sorted(bundle.glob("*.md"))
-        if path.name != _INDEX_NAME and _is_ingest_owned(path, manifest_file.stem)
+        if path.name != INDEX_NAME and _is_ingest_owned(path, manifest_file.stem)
     }
     # §3 collision gate — BEFORE any mutation: a staged filename occupied by
     # a file WITHOUT the stamp is curated content; never overwrite it.
@@ -293,20 +296,20 @@ def materialize_bundle(
     # §5 replacement: remove every stamped file, then write the new set.
     for name in sorted(owned):
         (bundle / name).unlink()
-    written = tuple(_write_bytes(bundle, name, content) for name, content in staged)
+    written = tuple(write_bytes(bundle, name, content) for name, content in staged)
 
     # §6 index generation — the last disk mutation. A fresh index gets
     # bundle_summary as its body; links are appended in extraction order.
-    index_path = bundle / _INDEX_NAME
+    index_path = bundle / INDEX_NAME
     labels_by_target = {
         generated_filename(extraction.id): extraction.title for extraction in manifest.extractions
     }
     if not index_path.is_file():
-        _write_bytes(bundle, _INDEX_NAME, manifest.bundle_summary + "\n")
+        write_bytes(bundle, INDEX_NAME, manifest.bundle_summary + "\n")
     else:
         # Links whose target is an ingest-owned file removed this run MUST be
         # removed; all other links — curated and promoted — are preserved.
         _update_index_lines(index_path, owned - staged_names, labels_by_target)
     for extraction in manifest.extractions:
-        _link_in_index(bundle, generated_filename(extraction.id), extraction.title)
+        link_in_index(bundle, generated_filename(extraction.id), extraction.title)
     return IngestResult(written=written, stamp=stamp)

tests/test_error_codes.py

@@ -28,7 +28,13 @@ from llm_ingestion_okf.errors import (
     SourceError,
 )
 from llm_ingestion_okf.extract import extract_text
-from llm_ingestion_okf.inbox import inbox_filename, inbox_slug, render_inbox_concept
+from llm_ingestion_okf.inbox import (
+    GateDecision,
+    inbox_filename,
+    inbox_slug,
+    process_inbox,
+    render_inbox_concept,
+)
 from llm_ingestion_okf.manifest import load_manifest, load_manifest_bytes
 from llm_ingestion_okf.materialize import materialize_bundle
 from llm_ingestion_okf.render import sql_value_to_text
@@ -369,6 +375,23 @@ def test_inbox_slug_too_long() -> None:
     assert code_of(excinfo) == "inbox_slug_too_long"
 
 
+def test_inbox_slug_collision(tmp_path: Path) -> None:
+    # Two dropped names reducing to one generated filename: reported per file,
+    # so this code surfaces in InboxResult.failed rather than as a raise.
+    inbox = tmp_path / "inbox"
+    inbox.mkdir()
+    (inbox / "note.md").write_text("a\n", encoding="utf-8")
+    (inbox / "note.txt").write_text("b\n", encoding="utf-8")
+    result = process_inbox(
+        inbox,
+        tmp_path / "bundle",
+        INGESTED_AT,
+        okf_type="note",
+        gate=lambda text: GateDecision(sanitized_text=text, disposition="warn"),
+    )
+    assert {entry.error.code for entry in result.failed} == {"inbox_slug_collision"}
+
+
 def test_inbox_title_invalid() -> None:
     with pytest.raises(MaterializationError) as excinfo:
         inbox_concept(title="broken [link]")

tests/test_inbox_flow.py

@@ -0,0 +1,434 @@
+"""Door B inbox flow against a stub guard (Phase 2 step 3).
+
+The stub stands in for the pinned `llm-ingestion-guard` surface so the persist,
+quarantine and reject branches run deterministically without the dependency
+installed. It is a TEST DOUBLE and lives here on purpose — never in `src/`.
+
+What the flow owes the caller, and what this suite pins:
+
+- the guard decides; the library only branches on the verdict and never
+  re-derives one (an unrecognised disposition must fail CLOSED);
+- what is persisted is the guard's SANITIZED text, so the bytes screened at
+  the gate are exactly the bytes written;
+- one bad file never aborts the run — every per-file failure is reported and
+  the remaining files still process;
+- the collision gate runs before any mutation and never overwrites curated
+  content, exactly as at Door A.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+
+import pytest
+
+from llm_ingestion_okf.inbox import GateDecision, InboxResult, process_inbox
+
+INGESTED_AT = "2026-07-25T12:00:00Z"
+
+# The guard's Disposition is a `str, Enum`; these are its VALUES, restated here
+# independently of the library constant so a drift in either is visible.
+WARN = "warn"
+QUARANTINE_REVIEW = "quarantine_review"
+FAIL_SECURE = "fail_secure"
+
+
+# --- the stub guard -------------------------------------------------------
+
+
+@dataclass
+class StubGuard:
+    """A gate whose verdict is scripted per source text.
+
+    `sanitize` models the guard stripping carriers: the default implementation
+    removes zero-width characters, so a test can assert that what landed on
+    disk is the sanitized text and not the extracted text.
+    """
+
+    disposition: str = WARN
+    reasons: tuple[str, ...] = ()
+    calls: list[str] | None = None
+
+    def __call__(self, text: str) -> GateDecision:
+        if self.calls is not None:
+            self.calls.append(text)
+        return GateDecision(
+            sanitized_text=text.replace("​", ""),
+            disposition=self.disposition,
+            reasons=self.reasons,
+        )
+
+
+@dataclass
+class PerFileGuard:
+    """A gate that returns a different disposition per source text fragment."""
+
+    by_marker: dict[str, str]
+
+    def __call__(self, text: str) -> GateDecision:
+        for marker, disposition in self.by_marker.items():
+            if marker in text:
+                return GateDecision(
+                    sanitized_text=text, disposition=disposition, reasons=(f"matched {marker}",)
+                )
+        return GateDecision(sanitized_text=text, disposition=WARN, reasons=())
+
+
+def drop(inbox: Path, name: str, content: str) -> None:
+    inbox.mkdir(parents=True, exist_ok=True)
+    (inbox / name).write_text(content, encoding="utf-8", newline="")
+
+
+def run(
+    tmp_path: Path, gate: object, *, okf_type: str = "note", ingested_at: str = INGESTED_AT
+) -> tuple[InboxResult, Path]:
+    bundle = tmp_path / "bundle"
+    result = process_inbox(
+        tmp_path / "inbox",
+        bundle,
+        ingested_at,
+        okf_type=okf_type,
+        gate=gate,  # type: ignore[arg-type]
+    )
+    return result, bundle
+
+
+# --- the persist branch ---------------------------------------------------
+
+
+def test_a_benign_file_is_persisted_as_an_inbox_concept(tmp_path: Path) -> None:
+    drop(tmp_path / "inbox", "My Notes.md", "Body text\n")
+
+    result, bundle = run(tmp_path, StubGuard())
+
+    assert [entry.source_file for entry in result.persisted] == ["My Notes.md"]
+    written = bundle / "inbox-my-notes.md"
+    assert written.is_file()
+    body = written.read_text(encoding="utf-8")
+    assert body.startswith("---\ntype: note\ntitle: My Notes\nsource_file: My Notes.md\n")
+    assert "generated: true\n" in body
+    assert body.endswith("Body text\n")
+
+
+def test_the_persisted_body_is_the_guards_sanitized_text(tmp_path: Path) -> None:
+    """The gate screens the sanitized text, so the sanitized text is what must
+    land on disk — screening one string and persisting another would make the
+    gate's verdict a statement about bytes nobody kept.
+    """
+    drop(tmp_path / "inbox", "note.md", "clean​body\n")
+
+    _, bundle = run(tmp_path, StubGuard())
+
+    written = (bundle / "inbox-note.md").read_text(encoding="utf-8")
+    assert "cleanbody" in written
+    assert "​" not in written
+
+
+def test_the_index_gets_a_link_labelled_with_the_readable_name(tmp_path: Path) -> None:
+    drop(tmp_path / "inbox", "My Notes.md", "Body\n")
+
+    _, bundle = run(tmp_path, StubGuard())
+
+    index = (bundle / "index.md").read_text(encoding="utf-8")
+    assert "- [My Notes](inbox-my-notes.md)\n" in index
+    assert not index.startswith("\n")
+
+
+def test_a_re_run_is_idempotent(tmp_path: Path) -> None:
+    """Same bytes + same ingested_at + same verdict => same bundle. The inbox
+    file is this door's to replace, and the index link is not duplicated.
+    """
+    drop(tmp_path / "inbox", "note.md", "Body\n")
+
+    _, bundle = run(tmp_path, StubGuard())
+    first = sorted((path.name, path.read_bytes()) for path in bundle.glob("*.md"))
+    _, bundle = run(tmp_path, StubGuard())
+    second = sorted((path.name, path.read_bytes()) for path in bundle.glob("*.md"))
+
+    assert first == second
+
+
+def test_files_are_processed_in_sorted_order(tmp_path: Path) -> None:
+    for name in ("c.md", "a.md", "b.md"):
+        drop(tmp_path / "inbox", name, f"body of {name}\n")
+    calls: list[str] = []
+
+    result, _ = run(tmp_path, StubGuard(calls=calls))
+
+    assert [entry.source_file for entry in result.persisted] == ["a.md", "b.md", "c.md"]
+    assert calls == ["body of a.md\n", "body of b.md\n", "body of c.md\n"]
+
+
+# --- the blocking branches ------------------------------------------------
+
+
+def test_quarantine_review_is_not_persisted_and_is_reported_separately(tmp_path: Path) -> None:
+    drop(tmp_path / "inbox", "suspect.md", "Body\n")
+
+    result, bundle = run(tmp_path, StubGuard(disposition=QUARANTINE_REVIEW, reasons=("held",)))
+
+    assert result.persisted == ()
+    assert [entry.source_file for entry in result.quarantined] == ["suspect.md"]
+    assert result.quarantined[0].reasons == ("held",)
+    assert result.rejected == ()
+    assert not (bundle / "inbox-suspect.md").exists()
+
+
+def test_fail_secure_is_not_persisted_and_is_reported_as_rejected(tmp_path: Path) -> None:
+    drop(tmp_path / "inbox", "poisoned.md", "Body\n")
+
+    result, bundle = run(tmp_path, StubGuard(disposition=FAIL_SECURE, reasons=("critical",)))
+
+    assert result.persisted == ()
+    assert [entry.source_file for entry in result.rejected] == ["poisoned.md"]
+    assert result.quarantined == ()
+    assert not (bundle / "inbox-poisoned.md").exists()
+
+
+def test_a_blocked_file_leaves_the_bundle_byte_identical(tmp_path: Path) -> None:
+    """The persist-gate proof: a blocked file produces ZERO new files and does
+    not touch the ones already there — not even the index.
+    """
+    drop(tmp_path / "inbox", "ok.md", "Body\n")
+    result, bundle = run(tmp_path, StubGuard())
+    assert result.persisted != ()
+    before = sorted((path.name, path.read_bytes()) for path in bundle.iterdir())
+
+    (tmp_path / "inbox" / "ok.md").unlink()
+    drop(tmp_path / "inbox", "poisoned.md", "Body\n")
+    run(tmp_path, StubGuard(disposition=FAIL_SECURE))
+
+    after = sorted((path.name, path.read_bytes()) for path in bundle.iterdir())
+    assert after == before
+
+
+@pytest.mark.parametrize("disposition", ["", "pass", "ok", "PERSIST", "warn "])
+def test_an_unrecognised_disposition_fails_closed(tmp_path: Path, disposition: str) -> None:
+    """Only the guard's non-blocking floor persists. Anything the library does
+    not recognise — a renamed member, a future disposition, a typo in an
+    adapter — must refuse to persist rather than guess it is safe.
+    """
+    drop(tmp_path / "inbox", "note.md", "Body\n")
+
+    result, bundle = run(tmp_path, StubGuard(disposition=disposition))
+
+    assert result.persisted == ()
+    assert [entry.source_file for entry in result.rejected] == ["note.md"]
+    assert list(bundle.glob("*.md")) == []
+
+
+# --- one bad file never aborts the run ------------------------------------
+
+
+def test_an_extraction_failure_does_not_abort_the_run(tmp_path: Path) -> None:
+    drop(tmp_path / "inbox", "good.md", "Body\n")
+    drop(tmp_path / "inbox", "archive.zip", "not text\n")
+
+    result, bundle = run(tmp_path, StubGuard())
+
+    assert [entry.source_file for entry in result.persisted] == ["good.md"]
+    assert [entry.source_file for entry in result.failed] == ["archive.zip"]
+    assert result.failed[0].error.code == "extractor_unknown"
+    assert (bundle / "inbox-good.md").is_file()
+
+
+def test_a_blocking_verdict_on_one_file_does_not_abort_the_run(tmp_path: Path) -> None:
+    drop(tmp_path / "inbox", "good.md", "harmless\n")
+    drop(tmp_path / "inbox", "bad.md", "poison\n")
+
+    result, bundle = run(tmp_path, PerFileGuard({"poison": FAIL_SECURE}))
+
+    assert [entry.source_file for entry in result.persisted] == ["good.md"]
+    assert [entry.source_file for entry in result.rejected] == ["bad.md"]
+    assert (bundle / "inbox-good.md").is_file()
+    assert not (bundle / "inbox-bad.md").exists()
+
+
+def test_an_unusable_filename_is_a_per_file_failure(tmp_path: Path) -> None:
+    drop(tmp_path / "inbox", "!!!.md", "Body\n")
+    drop(tmp_path / "inbox", "good.md", "Body\n")
+
+    result, _ = run(tmp_path, StubGuard())
+
+    assert [entry.source_file for entry in result.persisted] == ["good.md"]
+    assert [entry.source_file for entry in result.failed] == ["!!!.md"]
+    assert result.failed[0].error.code == "inbox_slug_empty"
+
+
+def test_an_over_long_filename_is_a_per_file_failure(tmp_path: Path) -> None:
+    """The reachable window is narrow but real: the dropped name itself cannot
+    exceed 255 bytes (the filesystem refuses to create it), so the longest stem
+    that can arrive is 252 characters — and `inbox-` + 252 + `.md` is 261, over
+    the limit. Any stem from 247 to 252 triggers the gate.
+    """
+    drop(tmp_path / "inbox", "x" * 252 + ".md", "Body\n")
+    drop(tmp_path / "inbox", "good.md", "Body\n")
+
+    result, _ = run(tmp_path, StubGuard())
+
+    assert [entry.source_file for entry in result.persisted] == ["good.md"]
+    assert [entry.error.code for entry in result.failed] == ["inbox_slug_too_long"]
+
+
+def test_a_title_that_would_break_an_index_link_is_a_per_file_failure(tmp_path: Path) -> None:
+    # The title is the readable filename stem, so a bracket in the dropped name
+    # reaches the same gate Door A applies to a manifest title.
+    drop(tmp_path / "inbox", "report [v2].md", "Body\n")
+
+    result, _ = run(tmp_path, StubGuard())
+
+    assert result.persisted == ()
+    assert [entry.error.code for entry in result.failed] == ["inbox_title_invalid"]
+
+
+# --- the collision gate ---------------------------------------------------
+
+
+def test_an_unstamped_collision_is_refused_without_overwriting(tmp_path: Path) -> None:
+    """Curated content occupying a generated name is never overwritten — the
+    same section 3 rule Door A enforces, applied per file so the rest of the
+    run still completes.
+    """
+    bundle = tmp_path / "bundle"
+    bundle.mkdir()
+    curated = bundle / "inbox-note.md"
+    curated.write_text("curated, no marker\n", encoding="utf-8")
+    drop(tmp_path / "inbox", "note.md", "Body\n")
+    drop(tmp_path / "inbox", "other.md", "Body\n")
+
+    result, _ = run(tmp_path, StubGuard())
+
+    assert curated.read_text(encoding="utf-8") == "curated, no marker\n"
+    assert [entry.error.code for entry in result.failed] == ["collision_unstamped"]
+    assert [entry.source_file for entry in result.persisted] == ["other.md"]
+
+
+def test_the_collision_gate_is_evaluated_before_any_write(tmp_path: Path) -> None:
+    """Pre-mutation: ownership is judged against the bundle as it was BEFORE
+    this run. A file written earlier in the same run must never be mistaken for
+    pre-existing curated content by a later file's check.
+    """
+    bundle = tmp_path / "bundle"
+    bundle.mkdir()
+    (bundle / "inbox-zzz.md").write_text("curated\n", encoding="utf-8")
+    drop(tmp_path / "inbox", "aaa.md", "Body\n")
+    drop(tmp_path / "inbox", "zzz.md", "Body\n")
+
+    result, _ = run(tmp_path, StubGuard())
+
+    # The colliding file is refused; the non-colliding one still lands.
+    assert [entry.error.code for entry in result.failed] == ["collision_unstamped"]
+    assert [entry.source_file for entry in result.persisted] == ["aaa.md"]
+    assert (bundle / "inbox-zzz.md").read_text(encoding="utf-8") == "curated\n"
+
+
+def test_two_dropped_files_that_slug_alike_are_both_refused(tmp_path: Path) -> None:
+    """`note.md` and `note.txt` both reduce to `inbox-note.md`. Persisting one
+    and silently dropping the other would make the outcome depend on iteration
+    order, and overwriting would lose the first file's content. Neither is
+    acceptable, and the library will not pick a winner: both are refused and
+    the operator renames one.
+    """
+    drop(tmp_path / "inbox", "note.md", "from markdown\n")
+    drop(tmp_path / "inbox", "note.txt", "from text\n")
+    drop(tmp_path / "inbox", "other.md", "Body\n")
+
+    result, bundle = run(tmp_path, StubGuard())
+
+    assert [entry.source_file for entry in result.failed] == ["note.md", "note.txt"]
+    assert {entry.error.code for entry in result.failed} == {"inbox_slug_collision"}
+    assert not (bundle / "inbox-note.md").exists()
+    assert [entry.source_file for entry in result.persisted] == ["other.md"]
+
+
+def test_a_previous_inbox_file_is_this_doors_to_replace(tmp_path: Path) -> None:
+    drop(tmp_path / "inbox", "note.md", "first\n")
+    run(tmp_path, StubGuard())
+    drop(tmp_path / "inbox", "note.md", "second\n")
+
+    result, bundle = run(tmp_path, StubGuard())
+
+    assert [entry.source_file for entry in result.persisted] == ["note.md"]
+    assert (bundle / "inbox-note.md").read_text(encoding="utf-8").endswith("second\n")
+
+
+def test_a_door_a_concept_is_not_this_doors_to_replace(tmp_path: Path) -> None:
+    """Namespace disjointness is the first defence, but the ownership rule is
+    the second: a file carrying Door A's stamp is never claimed here.
+    """
+    bundle = tmp_path / "bundle"
+    bundle.mkdir()
+    stamped = bundle / "inbox-note.md"
+    stamped.write_text(
+        "---\ntype: dataset\ningest_manifest: m@abc\ngenerated: true\n---\n\nbody\n",
+        encoding="utf-8",
+    )
+    drop(tmp_path / "inbox", "note.md", "Body\n")
+
+    result, _ = run(tmp_path, StubGuard())
+
+    assert [entry.error.code for entry in result.failed] == ["collision_unstamped"]
+    assert "ingest_manifest: m@abc" in stamped.read_text(encoding="utf-8")
+
+
+# --- run-level fail-fast --------------------------------------------------
+
+
+def test_an_invalid_ingested_at_fails_the_whole_run(tmp_path: Path) -> None:
+    """Not a per-file problem: a bad timestamp would stamp every file wrongly,
+    so it is refused before any file is read.
+    """
+    from llm_ingestion_okf import MaterializationError
+
+    drop(tmp_path / "inbox", "note.md", "Body\n")
+
+    with pytest.raises(MaterializationError) as excinfo:
+        run(tmp_path, StubGuard(), ingested_at="2026-07-25")
+    assert excinfo.value.code == "ingested_at_invalid"
+
+
+def test_a_reserved_okf_type_fails_the_whole_run(tmp_path: Path) -> None:
+    from llm_ingestion_okf import MaterializationError
+
+    drop(tmp_path / "inbox", "note.md", "Body\n")
+
+    with pytest.raises(MaterializationError) as excinfo:
+        run(tmp_path, StubGuard(), okf_type="verdict")
+    assert excinfo.value.code == "okf_type_reserved"
+
+
+def test_a_missing_inbox_directory_is_refused(tmp_path: Path) -> None:
+    from llm_ingestion_okf import SourceError
+
+    with pytest.raises(SourceError) as excinfo:
+        process_inbox(
+            tmp_path / "nope",
+            tmp_path / "bundle",
+            INGESTED_AT,
+            okf_type="note",
+            gate=StubGuard(),  # type: ignore[arg-type]
+        )
+    assert excinfo.value.code == "source_root_missing"
+
+
+def test_an_empty_inbox_writes_nothing(tmp_path: Path) -> None:
+    (tmp_path / "inbox").mkdir()
+
+    result, bundle = run(tmp_path, StubGuard())
+
+    assert result == InboxResult(persisted=(), quarantined=(), rejected=(), failed=())
+    assert not bundle.exists() or list(bundle.iterdir()) == []
+
+
+def test_subdirectories_are_not_walked(tmp_path: Path) -> None:
+    """Top-level only, like every other inbox in this ecosystem: a nested tree
+    is the operator's structure, not ours to flatten into one namespace.
+    """
+    drop(tmp_path / "inbox" / "nested", "deep.md", "Body\n")
+    drop(tmp_path / "inbox", "top.md", "Body\n")
+
+    result, bundle = run(tmp_path, StubGuard())
+
+    assert [entry.source_file for entry in result.persisted] == ["top.md"]
+    assert not (bundle / "inbox-deep.md").exists()