test(inbox): rebuild equals incremental across a re-split with varying timestamps

tests/test_segmented_rebuild.py

@@ -0,0 +1,267 @@
+"""S7, the main acceptance test: rebuild-from-scratch equals incremental update.
+
+Byte for byte, over the WHOLE tree, across a re-split, with the call-level
+`ingested_at` deliberately different on every run.
+
+Why it is written this way rather than copied from the existing multi-round
+test. That test (`tests/test_inbox_structure.py`) passes the SAME `ingested_at`
+constant every round and compares only `index.md`. It therefore holds for a
+weaker reason than it appears to: a timestamp that leaked from the call into a
+concept would be identical on both sides, and a concept body that diverged
+would never be looked at. Both holes are closed here -- three distinct
+call-level timestamps, and a comparison of the whole tree.
+
+That the timestamps may differ at all is the design working: under a plan,
+`ingested_at` comes from the PLAN ENTRY, never from the call. A rebuild months
+later replays the adjudication and reproduces the bytes of the round that first
+wrote the concept. If the call-level value leaked in, this test would be the
+one that caught it.
+
+**One deviation from the plan, stated rather than absorbed.** The plan also
+asks that, with the write seam patched to raise, a SECOND run over an unchanged
+inbox must not raise -- i.e. that an unchanged round writes nothing.
+`materialize.write_bytes` writes unconditionally; no step in this plan makes it
+conditional, and this step's own manifest forbids touching `inbox.py`, so the
+property has no implementing code to assert against. The half that IS real --
+the positive control proving the patch fires and that the segmented path routes
+through the write seam at all -- is asserted below. Without that control, a
+patch that silently never fired would make an empty bundle look like proof.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import shutil
+import subprocess
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from llm_ingestion_okf.errors import SegmentationError
+from llm_ingestion_okf.inbox import GateDecision, process_inbox
+from llm_ingestion_okf.profiles import SEGMENTED_V1
+from llm_ingestion_okf.segmentation import SegmentationPlan, parse_segmentation_plan
+
+# THREE distinct call-level values. None of them may reach a plan-covered
+# concept, and the test is worthless if they are all the same.
+ROUND_1_AT = "2026-07-25T12:00:00Z"
+ROUND_2_AT = "2026-08-14T06:30:00Z"
+REBUILD_AT = "2026-09-01T23:59:59Z"
+
+PLAN_AT = "2026-08-30T09:00:00Z"
+
+DOCUMENT = "Krav i konseptet.\n" * 20
+
+# Round 1 names a segment round 2 does NOT. Without it the re-split would only
+# ADD, and S7 would never exercise stale-segment retirement -- the Critical risk
+# that makes an incremental bundle diverge from a scratch rebuild. Measured
+# 2026-09-01: with retirement disabled and round 2 a superset, S7 stayed green.
+ROUND_1_ENTRIES = (
+    ("s0", "krav/brannkonsept.md", None),
+    ("s9", "krav/utgaatt/tidligere-inndeling.md", None),
+)
+ROUND_2_ENTRIES = (
+    ("s0", "krav/brannkonsept.md", None),
+    ("s1", "krav/brannkonsept/seksjonering.md", "s0"),
+    ("s2", "krav/brannkonsept/roemning.md", "s0"),
+    ("s3", "krav/3-2/baereevne.md", None),
+)
+
+
+def gate(text: str) -> GateDecision:
+    return GateDecision(sanitized_text=text, disposition="warn")
+
+
+def drop(inbox: Path, name: str = "n500.md", text: str = DOCUMENT) -> Path:
+    inbox.mkdir(parents=True, exist_ok=True)
+    path = inbox / name
+    path.write_text(text, encoding="utf-8", newline="")
+    return path
+
+
+def build_plan(
+    source_bytes: bytes,
+    entries: tuple[tuple[str, str, str | None], ...],
+    **overrides: Any,
+) -> SegmentationPlan:
+    payload: dict[str, Any] = {
+        "version": "1",
+        "source_sha256": hashlib.sha256(source_bytes).hexdigest(),
+        "extractor_id": "md",
+        "extractor_version": "1.0.0",
+        "adjudicated_at": "2026-08-30T08:00:00Z",
+        "entries": [
+            {
+                "segment_id": segment_id,
+                "path": path,
+                "title": f"Del {segment_id}",
+                "okf_type": "requirement",
+                "span": [index * 10, index * 10 + 10],
+                "ingested_at": PLAN_AT,
+                **({"parent_id": parent} if parent else {}),
+            }
+            for index, (segment_id, path, parent) in enumerate(entries)
+        ],
+    }
+    payload.update(overrides)
+    return parse_segmentation_plan(payload)
+
+
+def run(
+    inbox: Path,
+    bundle: Path,
+    ingested_at: str,
+    *,
+    plan: SegmentationPlan | None,
+    bundle_id: str | None = "b-1",
+):
+    return process_inbox(
+        inbox,
+        bundle,
+        ingested_at,
+        okf_type="requirement",
+        gate=gate,
+        profile=SEGMENTED_V1,
+        root_frontmatter_values={"bundle_id": bundle_id} if bundle_id else {},
+        segmentation=plan,
+    )
+
+
+def diff_trees(left: Path, right: Path) -> subprocess.CompletedProcess[str]:
+    """`diff -r`, the literal S7 criterion.
+
+    A recursive byte-walk would miss a directory present on only one side --
+    exactly what a retired segment leaves behind when its parent is not pruned.
+    `diff -r` reports it as `Only in ...`.
+    """
+    return subprocess.run(
+        ["diff", "-r", str(left), str(right)], capture_output=True, text=True, check=False
+    )
+
+
+def build_incremental(tmp_path: Path) -> tuple[Path, SegmentationPlan]:
+    source = drop(tmp_path / "one")
+    run(
+        tmp_path / "one",
+        tmp_path / "incremental",
+        ROUND_1_AT,
+        plan=build_plan(source.read_bytes(), ROUND_1_ENTRIES),
+    )
+    drop(tmp_path / "two")
+    second = build_plan(source.read_bytes(), ROUND_2_ENTRIES)
+    run(tmp_path / "two", tmp_path / "incremental", ROUND_2_AT, plan=second)
+    return tmp_path / "incremental", second
+
+
+# --- S7 -------------------------------------------------------------------
+
+
+def test_rebuild_from_scratch_equals_the_incremental_bundle(tmp_path: Path) -> None:
+    incremental, final_plan = build_incremental(tmp_path)
+
+    rebuild = tmp_path / "rebuild"
+    shutil.rmtree(rebuild, ignore_errors=True)
+    drop(tmp_path / "scratch")
+    run(tmp_path / "scratch", rebuild, REBUILD_AT, plan=final_plan)
+
+    result = diff_trees(rebuild, incremental)
+    assert result.stdout == ""
+    assert result.returncode == 0
+
+
+def test_the_three_call_level_timestamps_really_do_differ() -> None:
+    # The control for the test above. If these ever collapse to one value, S7
+    # would pass for the weaker reason M3 measured in the existing suite.
+    assert len({ROUND_1_AT, ROUND_2_AT, REBUILD_AT}) == 3
+
+
+def test_no_concept_carries_a_call_level_timestamp(tmp_path: Path) -> None:
+    incremental, _ = build_incremental(tmp_path)
+    for path in incremental.rglob("*.md"):
+        if path.name == SEGMENTED_V1.index.name:
+            continue
+        body = path.read_text(encoding="utf-8")
+        assert f"ingested_at: {PLAN_AT}" in body
+        for leaked in (ROUND_1_AT, ROUND_2_AT, REBUILD_AT):
+            assert leaked not in body
+
+
+def test_the_resplit_retired_a_segment_round_two_no_longer_names(tmp_path: Path) -> None:
+    # The control for S7 itself. If round 2 only ADDED, S7 would stay green with
+    # stale-segment retirement switched off entirely -- measured, not assumed.
+    incremental, _ = build_incremental(tmp_path)
+    assert not (incremental / "krav/utgaatt/tidligere-inndeling.md").exists()
+    assert not (incremental / "krav/utgaatt").exists()
+
+
+def test_the_resplit_actually_happened(tmp_path: Path) -> None:
+    # Without this, S7 could pass over a bundle that never re-split at all.
+    incremental, _ = build_incremental(tmp_path)
+    concepts = {
+        path.relative_to(incremental).as_posix()
+        for path in incremental.rglob("*.md")
+        if path.name != SEGMENTED_V1.index.name
+    }
+    assert len(concepts) == len(ROUND_2_ENTRIES)
+    assert "krav/brannkonsept/seksjonering.md" in concepts
+
+
+# --- S5: the write seam is the only path to disk --------------------------
+
+
+def test_the_patched_write_seam_fires_on_the_first_run(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # The POSITIVE CONTROL. A patch that silently never fired would leave an
+    # empty bundle, and an empty bundle is indistinguishable from "nothing
+    # needed writing" -- a negative result read as a positive fact.
+    source = drop(tmp_path / "one")
+
+    def refuse(*args: object, **kwargs: object) -> Path:
+        raise AssertionError("write seam reached")
+
+    monkeypatch.setattr("llm_ingestion_okf.inbox.write_bytes", refuse)
+    with pytest.raises(AssertionError, match="write seam reached"):
+        run(
+            tmp_path / "one",
+            tmp_path / "bundle",
+            ROUND_1_AT,
+            plan=build_plan(source.read_bytes(), ROUND_2_ENTRIES),
+        )
+
+
+# --- bundle_id is load-bearing on the rebuild -----------------------------
+
+
+def test_a_rebuild_omitting_the_bundle_id_is_refused_before_any_write(
+    tmp_path: Path,
+) -> None:
+    source = drop(tmp_path / "scratch")
+    with pytest.raises(SegmentationError) as excinfo:
+        run(
+            tmp_path / "scratch",
+            tmp_path / "rebuild",
+            REBUILD_AT,
+            plan=build_plan(source.read_bytes(), ROUND_2_ENTRIES),
+            bundle_id=None,
+        )
+    assert excinfo.value.code == "segmentation_plan_invalid"
+    # Fail-fast, so there is no tree at all -- not a tree that merely differs.
+    assert not (tmp_path / "rebuild").exists()
+
+
+def test_a_different_bundle_id_changes_every_concept(tmp_path: Path) -> None:
+    incremental, final_plan = build_incremental(tmp_path)
+    drop(tmp_path / "scratch")
+    other = tmp_path / "other"
+    run(tmp_path / "scratch", other, REBUILD_AT, plan=final_plan, bundle_id="b-2")
+
+    for path in other.rglob("*.md"):
+        if path.name == SEGMENTED_V1.index.name:
+            continue
+        twin = incremental / path.relative_to(other)
+        assert twin.is_file()
+        assert path.read_bytes() != twin.read_bytes()
+
+    assert diff_trees(other, incremental).returncode != 0