portfolio-optimiser/tests/test_ingest_loadbearing.py

"""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 llm_ingestion_okf import ManifestError

from conftest import expected_generated_stamp
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]], *, filename: str | None = None
) -> Path:
    """Write a file-source manifest. ``filename`` overrides the default name-by-count.

    The default encodes the extraction count, which keeps unrelated cases in one tmp_path
    from overwriting each other. It is the WRONG default for a RE-ingest, though: the
    library keys ownership on the manifest's STEM (§10.2), so writing an edited manifest
    under a new name models "a second, rival manifest sharing the bundle" rather than "the
    same manifest, edited" — and the collision gate then correctly refuses to touch the
    files the first name wrote. Pass ``filename`` to keep the stem stable across an edit.
    Measured 2026-08-05: this distinction is invisible at the pinned v0.3.2 and load-bearing
    from v0.4.0 on, so pinning the stem makes the re-ingest case mean the same thing on both.
    """
    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 = filename if filename is not None else 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(ManifestError) as exc:
        materialize(manifest_path, bundle_dir, ingested_at=_INGESTED_AT)
    assert exc.value.code == "okf_type_reserved"

    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"] == expected_generated_stamp(_INGESTED_AT)
        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()

    # ONE manifest filename, rewritten in place between the two runs: this case is a
    # RE-ingest of an EDITED manifest, and ownership is keyed on the stem (§10.2). Writing
    # the reduced form under a second name would make it a rival manifest instead, which the
    # collision gate rightly refuses — the scenario this test exists to cover would never run.
    _REINGEST_MANIFEST = "reingest-manifest.json"
    manifest_two = _write_project(
        tmp_path, copy.deepcopy(_EXTRACTIONS), filename=_REINGEST_MANIFEST
    )
    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(  # drop `meta` — same file, edited
        tmp_path, copy.deepcopy(_EXTRACTIONS)[:1], filename=_REINGEST_MANIFEST
    )
    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)}"
                )