test(ingest): load-bearing seam quartet + MAF-free guard, detach-proven (I2)

tests/test_ingest_loadbearing.py

@@ -0,0 +1,258 @@
+"""I2 load-bearing seams (ingest spec §11's table; the method-spec §11 detach-RED regime).
+
+Each test names its seam and the detach that turns it RED — scenarios are built so the seam
+under test is the ONLY thing preventing the observable artifact (the Fase-2 green-but-dead
+trap). Pattern: tests/test_step8_promotion_loadbearing.py.
+
+- VERDICT RESERVATION (§3): a manifest mapping to ``type: verdict`` — otherwise fully valid
+  and materializable, so detaching the validator WOULD produce a verdict file + link — is
+  rejected at load, before any source call, and writes NOTHING. RED when the
+  case-insensitive ``okf_type`` validator is removed. (The reserved-FILENAME half of §3 —
+  ``index.md``/``promoted-verdict-*`` — is structurally unreachable: the §4 id grammar plus
+  the mandatory ``ingest-`` prefix make no valid manifest able to produce those names. It
+  is asserted below as a defensive invariant; its load-bearing bite is the okf_type check.)
+- PROVENANCE STAMPING (§7): every generated file carries the 7-key layer, readable by the
+  UNCHANGED ``okf.parse_frontmatter``. RED when the stamp keys are dropped from rendering.
+- NAVIGABILITY (§2/§11): the generated bundle is consumable by the UNCHANGED bundle
+  navigation — every ``ingest-{id}.md`` reachable via index cross-links, classified by its
+  ``okf_type``, provenance riding through unknown-field preservation, content rendered by
+  ``bundle_context``. RED when index linking is detached (file unreachable).
+- RE-INGEST LAYER SAFETY (§3/§6/§10) — with the removal filter ACTIVE: the second
+  materialization uses a REDUCED manifest so the §6 managed-line removal actually fires
+  while a REAL promoted verdict (``verdicts.promote_verdict``) and the fixture's curated
+  links are present. An identical-manifest second run never removes anything and would
+  stay green under a loose filter — exactly the trap this test closes. RED when the
+  removal/ownership filter over-matches (a bare-substring filter deletes the promoted
+  managed-form line) or replacement stops honouring the ingest stamp.
+- MAF-FREE MODULE (context-seam purity): ``ingest.py`` imports no ``agent_framework``/
+  ``mcp`` and no MAF-coupled project modules (only ``okf``/``retrieval``, both
+  stdlib-pure) — mirrors ``tests/test_okf.py::test_okf_is_maf_free``, which covers only
+  ``okf.py``. RED the moment e.g. ``contracts`` (→ ``backends`` → MAF) is imported.
+"""
+
+from __future__ import annotations
+
+import ast
+import copy
+import json
+import shutil
+from pathlib import Path
+from typing import Any
+
+import pytest
+from pydantic import ValidationError
+
+from portfolio_optimiser import okf
+from portfolio_optimiser.ingest import Extraction, materialize
+from portfolio_optimiser.verdicts import Verdict, bundle_candidate_features, promote_verdict
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+CURATED_BUNDLE = REPO_ROOT / "shared" / "examples" / "bygg-energi-mikro"
+INGEST_SOURCE = REPO_ROOT / "src" / "portfolio_optimiser" / "ingest.py"
+
+_INGESTED_AT = "2026-07-03T12:00:00Z"
+
+_EXTRACTIONS: list[dict[str, Any]] = [
+    {"id": "costs", "title": "Project costs", "query": "costs.csv", "okf_type": "dataset", "max_rows": 100},
+    {"id": "meta", "title": "Catalogue metadata", "query": "meta.csv", "okf_type": "reference", "max_rows": 10},
+]
+_FILES = {
+    "costs.csv": b"item,cost\nled,120\n",
+    "meta.csv": b"key,value\nowner,anlegg\n",
+}
+
+
+def _write_project(tmp_path: Path, extractions: list[dict[str, Any]]) -> Path:
+    catalogue = tmp_path / "catalogue"
+    catalogue.mkdir(exist_ok=True)
+    for name, content in _FILES.items():
+        (catalogue / name).write_bytes(content)
+    manifest = {
+        "manifest_version": 1,
+        "source": {"type": "file", "id": "prosjekt-arkiv", "root": "catalogue"},
+        "bundle_summary": "Cost extracts from the project archive.",
+        "extractions": extractions,
+    }
+    name = f"manifest-{len(extractions)}.json"
+    manifest_path = tmp_path / name
+    manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
+    return manifest_path
+
+
+# --- Seam 1: verdict reservation (§3) ------------------------------------------------------------
+
+
+def test_verdict_typed_manifest_is_rejected_and_writes_nothing(tmp_path: Path) -> None:
+    """LOAD-BEARING RESERVATION: the manifest is otherwise fully materializable (real CSVs,
+    writable target), so detaching the okf_type validator WOULD produce a machine-generated
+    ``type: verdict`` file that the next run's seeding turns into an "approved" store entry
+    around the gate. RED when the validator is removed."""
+    extractions = copy.deepcopy(_EXTRACTIONS)
+    extractions[0]["okf_type"] = "Verdict"  # case-insensitive rejection (§3/§4)
+    manifest_path = _write_project(tmp_path, extractions)
+    bundle_dir = tmp_path / "bundle"
+
+    with pytest.raises(ValidationError):
+        materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT)
+
+    assert not bundle_dir.exists(), "a verdict-typed manifest must write NOTHING (no index, no files)"
+
+
+def test_reserved_filename_namespace_is_structurally_unreachable() -> None:
+    """Defensive invariant (§3, honestly non-detachable): for EVERY valid extraction id the
+    generated name ``ingest-{id}.md`` is disjoint from ``index.md`` and ``promoted-verdict-*``
+    by construction — no manifest input can exercise a filename collision with the reserved
+    namespace, so the load-bearing bite lives in the okf_type test above."""
+    extraction = Extraction(
+        id="promoted-verdict-x", title="t", query="q.csv", okf_type="dataset", max_rows=1
+    )
+    name = f"ingest-{extraction.id}.md"
+    assert name.startswith("ingest-") and name != "index.md"
+    assert not name.startswith("promoted-verdict-")
+
+
+# --- Seam 2: provenance stamping (§7) -------------------------------------------------------------
+
+
+def test_every_generated_file_carries_the_provenance_layer(tmp_path: Path) -> None:
+    """LOAD-BEARING PROVENANCE: the §7 layer (all seven keys, ``generated: true``, the
+    ``{stem}@{hash16}`` manifest reference) is present on EVERY generated file, readable by
+    the unchanged ``okf.parse_frontmatter``. RED when the stamp keys are dropped."""
+    manifest_path = _write_project(tmp_path, copy.deepcopy(_EXTRACTIONS))
+    written = materialize(manifest_path, tmp_path / "bundle", ingested_at=_INGESTED_AT)
+
+    assert len(written) == 2
+    for path in written:
+        fm = okf.parse_frontmatter(path)
+        for key in (
+            "type",
+            "title",
+            "source_system",
+            "source_query",
+            "ingested_at",
+            "ingest_manifest",
+            "generated",
+        ):
+            assert key in fm, f"provenance key {key!r} missing from {path.name}"
+        assert fm["generated"] == "true"
+        assert fm["source_system"] == "prosjekt-arkiv"
+        assert fm["ingested_at"] == _INGESTED_AT
+        stem, _, digest = fm["ingest_manifest"].partition("@")
+        assert stem == manifest_path.stem and len(digest) == 16
+
+
+# --- Seam 3: navigability via UNCHANGED okf.py (§2/§11) -------------------------------------------
+
+
+def test_generated_bundle_is_navigable_via_unchanged_okf(tmp_path: Path) -> None:
+    """LOAD-BEARING NAVIGABILITY (durable version of the planning session's ad-hoc proof):
+    ``okf.navigate_bundle`` — untouched by I2 — reaches every generated file through index
+    cross-links, classifies it by okf_type, preserves the provenance fields (unknown-field
+    preservation, OKF §4), and ``bundle_context`` renders the extracted content. RED when
+    index linking is detached: the file exists but is unreachable (navigation follows ONLY
+    index cross-links)."""
+    manifest_path = _write_project(tmp_path, copy.deepcopy(_EXTRACTIONS))
+    bundle_dir = tmp_path / "bundle"
+    materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT)
+
+    bundle = okf.navigate_bundle(str(bundle_dir))
+    by_name = {f.name: f for f in bundle.files}
+    assert "ingest-costs.md" in by_name and "ingest-meta.md" in by_name
+    assert by_name["ingest-costs.md"].type == "dataset"
+    assert by_name["ingest-meta.md"].type == "reference"
+    assert by_name["ingest-costs.md"].frontmatter["ingest_manifest"]  # provenance rides through
+    context = okf.bundle_context(bundle)
+    assert "| led | 120 |" in context and "Project costs" in context
+
+
+# --- Seam 4: re-ingest layer safety, removal filter ACTIVE (§3/§6/§10) ----------------------------
+
+
+def test_reingest_with_active_removal_preserves_promoted_and_curated(tmp_path: Path) -> None:
+    """LOAD-BEARING LAYER SAFETY: materialize into a copy of the curated fixture bundle,
+    promote a REAL verdict through ``verdicts.promote_verdict``, then re-materialize with a
+    REDUCED manifest so the §6 removal filter actually fires. The dropped extraction's file
+    and link vanish; the promoted verdict's file and exact managed-form index line — and
+    every curated byte — survive. RED when the filter over-matches (bare substring) or the
+    stamp ownership rule is detached."""
+    bundle_dir = tmp_path / "bundle"
+    shutil.copytree(CURATED_BUNDLE, bundle_dir)  # promotion + ingest WRITE: throwaway copy
+    curated_index_lines = (bundle_dir / "index.md").read_text(encoding="utf-8").splitlines()
+
+    manifest_two = _write_project(tmp_path, copy.deepcopy(_EXTRACTIONS))
+    materialize(manifest_two, bundle_dir, ingested_at=_INGESTED_AT)
+
+    promoted_path = promote_verdict(
+        str(bundle_dir),
+        Verdict(
+            id="I2-REINGEST",
+            proposal_features=bundle_candidate_features(str(bundle_dir)),
+            decision="approved",
+            rationale="godkjent — layer-safety-fixture",
+        ),
+        approver="persona",
+        experiment="i2-reingest",
+        timestamp="2026-07-03",
+    )
+    promoted_line = f"- [Promotert ekspert-vurdering (gated)]({promoted_path.name})"
+    curated_bytes_before = {
+        p.name: p.read_bytes()
+        for p in sorted(bundle_dir.glob("*.md"))
+        if not p.name.startswith("ingest-") and p.name != "index.md"
+    }
+
+    manifest_one = _write_project(tmp_path, copy.deepcopy(_EXTRACTIONS)[:1])  # drop `meta`
+    materialize(manifest_one, bundle_dir, ingested_at=_INGESTED_AT)
+
+    index_after = (bundle_dir / "index.md").read_text(encoding="utf-8")
+    assert not (bundle_dir / "ingest-meta.md").exists()
+    assert "](ingest-meta.md)" not in index_after, "dropped extraction's link must be removed"
+    assert (bundle_dir / "ingest-costs.md").exists()
+    assert promoted_path.exists(), "re-ingest deleted the promoted verdict file (§3 violated)"
+    assert promoted_line in index_after.splitlines(), (
+        "the promoted verdict's index link did not survive an ACTIVE removal pass — the "
+        "filter over-matched the managed-line shape promote_verdict writes"
+    )
+    for line in curated_index_lines:
+        assert line in index_after.splitlines(), f"curated index line lost: {line!r}"
+    curated_bytes_after = {
+        p.name: p.read_bytes()
+        for p in sorted(bundle_dir.glob("*.md"))
+        if not p.name.startswith("ingest-") and p.name != "index.md"
+    }
+    assert curated_bytes_after == curated_bytes_before, "only stamped files may be replaced"
+
+
+# --- Seam 5: MAF-free module guard ----------------------------------------------------------------
+
+
+def test_ingest_module_is_maf_free_and_context_layer_pure() -> None:
+    """The ingest module is a context-seam citizen (D7-portable): NO ``agent_framework``,
+    NO ``mcp``, and no MAF-coupled project modules — only ``okf``/``retrieval`` (both
+    stdlib-pure; ``contracts`` would smuggle MAF in via ``backends``). Mirrors
+    ``test_okf.py::test_okf_is_maf_free``, which covers only ``okf.py``."""
+    tree = ast.parse(INGEST_SOURCE.read_text(encoding="utf-8"))
+    forbidden_roots = {"agent_framework", "mcp"}
+    allowed_internal = {"portfolio_optimiser.okf", "portfolio_optimiser.retrieval"}
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Import):
+            for alias in node.names:
+                root = alias.name.split(".")[0]
+                assert root not in forbidden_roots, f"MAF import in ingest.py: {alias.name}"
+                assert root != "portfolio_optimiser", (
+                    "import whole project modules explicitly via `from` so the guard can "
+                    f"check them: {alias.name}"
+                )
+        elif isinstance(node, ast.ImportFrom):
+            module = node.module or ""
+            root = module.split(".")[0]
+            assert root not in forbidden_roots, f"MAF import in ingest.py: {module}"
+            if root == "portfolio_optimiser":
+                if module == "portfolio_optimiser":
+                    imported = {f"portfolio_optimiser.{alias.name}" for alias in node.names}
+                else:
+                    imported = {module}
+                assert imported <= allowed_internal, (
+                    f"ingest.py imports MAF-coupled project module(s): "
+                    f"{sorted(imported - allowed_internal)}"
+                )