"""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)}" )