Kjell Tore Guttormsen
aa87eb8818
Door B listed `inbox.iterdir()` and kept only top-level files. A file in a subdirectory was neither ingested nor refused: it appeared in none of the result's buckets, so a nested drop produced a bundle that was silently short of what was dropped and no count said so. That broke the K1b identity for any inbox with folders in it. Operator decision 2026-09-06. - `walk_inbox` is the ONE walk rule, shared with `tools/okf_corpus_run.py`: the denominator N is now counted over exactly the set of files the door ingests, rather than over a second listing that happened to agree. - Sorted on the whole relative path, not the basename, so the order is a function of the tree; that is what keeps rebuild-from-scratch byte-equal to an incremental update. - A concept's `source_file` is the path relative to the inbox root, `/`-separated. The concept NAME still comes from the basename, so two folders holding one basename hit the existing §3 collision refusal instead of one silently claiming the other's concept. - Dot-directories and a bundle directory inside the inbox are skipped with a CODE, in a new `InboxResult.skipped`. Recursion makes the door's own output reachable as its own input; a silent skip would be the same absence-without-a-denominator defect one level down. - `--path-prefix` reduces per component and rejoins with `/`, so the caller driving a nested corpus can carry the relative directory. Reducing the whole string folded the separator into a `-` and flattened `sub/sub2`. `tests/test_inbox_flow.py::test_subdirectories_are_not_walked` asserted the opposite and is superseded in place, with the reason written down. Measured on the K2 corpus (flat, N=43): 39/43 merged, 4 coded, K1b holds. The bundle digest is `1472e98aec8643c5beee540f4c42b5e437bd26e7c61d69a91bcff799f06a6d13` over 1108 files -- byte-identical to a run of the same corpus at190086fWITHOUT this change (`diff -r` exit 0), so recursion costs a flat inbox nothing. It differs from the stored 2026-09-03 artifact by one line in `index.md` (`- [Corpus run history](log.md)`), which95eb271added 15 hours after that bundle was built. Suite 1113 passed, `ruff` clean, `mypy --strict src/ tools/` clean. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
441 lines
17 KiB
Python
441 lines
17 KiB
Python
"""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", "cleanbody\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_walked(tmp_path: Path) -> None:
|
||
"""Superseded, deliberately, by the operator's decision of 2026-09-06.
|
||
|
||
This test used to assert the opposite -- top-level only, "the operator's
|
||
nested structure is theirs". The rule it pinned was not a boundary, it was
|
||
a silent loss: a file under a folder appeared in NONE of the result's
|
||
buckets, so the count said nothing was there rather than that nothing was
|
||
looked at. The nested structure is still the operator's, and it survives
|
||
in the recorded `source_file`; it is the disappearance that is gone. The
|
||
rule's own suite is `tests/test_inbox_recursion.py`.
|
||
"""
|
||
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] == ["nested/deep.md", "top.md"]
|
||
assert (bundle / "inbox-deep.md").is_file()
|