portfolio-optimiser/tests/test_okf.py

"""OKF bundle navigation (okf.py) — the framework-neutral, D7-portable context seam.

These tests pin the minimal navigation the Step-1 ExpeL wiring depends on: read ``index.md``,
follow intra-bundle cross-links, parse each file's frontmatter, classify by ``type``, and locate
the candidate IR projection. Robustness (OKF SPEC §4): broken links are tolerated, never raised.

No ``agent_framework``/``mcp`` import is allowed in ``okf`` — guarded by ``test_okf_is_maf_free``.
"""

from __future__ import annotations

import ast
import os
from pathlib import Path

import pytest

from portfolio_optimiser import okf

# Framework-neutral, D7-portable modules that must never import MAF/mcp (C2:
# the guard previously scanned only okf.py; dimension.py is now covered too).
_MAF_FREE_MODULES = [
    "okf.py",
    "dimension.py",
    "outbox.py",
    "costsim.py",
    "hitl.py",
    "notify.py",
    "semretrieval.py",
]

BUNDLE_DIR = Path(__file__).resolve().parents[1] / "shared" / "examples" / "bygg-energi-mikro"


def test_navigate_classifies_every_typed_file() -> None:
    """The energi bundle resolves to its six typed OKF files (index + project + hypothesis +
    methodology + reference + verdict), each carrying its declared ``type``."""
    bundle = okf.navigate_bundle(str(BUNDLE_DIR))
    types = {f.name: f.type for f in bundle.files}
    assert types == {
        "index.md": "index",
        "bygg-kontor-nord.md": "project",
        "tiltak-led-retrofit.md": "hypothesis",
        "metode-ipmvp-a.md": "methodology",
        "kilder-realiseringsgap.md": "reference",
        "verdict-led-fro.md": "verdict",
    }


def test_navigate_exposes_verdicts_and_hypothesis() -> None:
    """The navigation surfaces exactly one ``type: verdict`` file (the ExpeL seed) and the
    candidate hypothesis — the two the Step-1 wiring keys on."""
    bundle = okf.navigate_bundle(str(BUNDLE_DIR))
    assert [f.name for f in bundle.verdicts] == ["verdict-led-fro.md"]
    assert bundle.verdicts[0].frontmatter["realization_rate"] == "0.82"
    assert bundle.hypothesis is not None
    assert bundle.hypothesis.name == "tiltak-led-retrofit.md"


def test_navigate_index_summary_is_progressive_disclosure() -> None:
    """``index_summary`` is the index body (the progressive-disclosure entry point), not stuffing
    the whole bundle."""
    bundle = okf.navigate_bundle(str(BUNDLE_DIR))
    assert "progressiv disclosure" in bundle.index_summary.lower()


def test_bundle_context_excludes_verdict_layer() -> None:
    """Fase 2b LOAD-BEARING (okf level): ``bundle_context`` renders the concept files for the agent
    prompt but EXCLUDES the ``type: verdict`` file, so the realization signal reaches a prompt only
    via the gated ExpeL fold — never by stuffing it into the read-context (målbilde §2/§4)."""
    bundle = okf.navigate_bundle(str(BUNDLE_DIR))
    assert {f.type for f in bundle.context_files} == {
        "project",
        "hypothesis",
        "methodology",
        "reference",
    }
    context = okf.bundle_context(bundle)
    assert "0.82" not in context  # the verdict's realization signal is NOT stuffed in
    assert (
        bundle.verdicts[0].frontmatter["realization_rate"] == "0.82"
    )  # though it IS in the bundle
    assert "## hypothesis:" in context  # concept files ARE rendered (progressive disclosure)
    assert "progressiv disclosure" in context.lower()  # the index summary is the entry point


def _dimension_bundle(tmp_path) -> str:
    (tmp_path / "index.md").write_text(
        "---\ntype: index\n---\n\n# Bundle\n\n"
        "- [energi](energi-method.md)\n"
        "- [asfalt](asfalt-method.md)\n"
        "- [shared](shared-note.md)\n"
        "- [verdict](verdict-x.md)\n",
        encoding="utf-8",
    )
    (tmp_path / "energi-method.md").write_text(
        "---\ntype: methodology\ndimension: energi\n---\n\nENERGI-SENTINEL body\n", encoding="utf-8"
    )
    (tmp_path / "asfalt-method.md").write_text(
        "---\ntype: methodology\ndimension: asfalt\n---\n\nASFALT-SENTINEL body\n", encoding="utf-8"
    )
    (tmp_path / "shared-note.md").write_text(
        "---\ntype: reference\n---\n\nSHARED-SENTINEL body\n", encoding="utf-8"
    )
    (tmp_path / "verdict-x.md").write_text(
        "---\ntype: verdict\ndimension: energi\n---\n\nVERDICT-SENTINEL body\n", encoding="utf-8"
    )
    return str(tmp_path)


def test_bundle_context_dimension_filter(tmp_path) -> None:
    """SC7 forutsetning: with ``dimension="energi"`` only energi-marked + unmarked concept files
    render; an asfalt-marked file is omitted. ``dimension=None`` is byte-identical to the no-arg
    call (backward compat — protects the verdict-exclusion + step7/8 load-bearing tests).
    ``type: verdict`` stays excluded in every case."""
    bundle = okf.navigate_bundle(_dimension_bundle(tmp_path))

    scoped = okf.bundle_context(bundle, dimension="energi")
    assert "ENERGI-SENTINEL" in scoped  # energi-marked concept file rendered
    assert "SHARED-SENTINEL" in scoped  # unmarked knowledge is never dropped
    assert "ASFALT-SENTINEL" not in scoped  # other-dimension file filtered out
    assert "VERDICT-SENTINEL" not in scoped  # verdict layer still excluded

    default = okf.bundle_context(bundle)
    assert okf.bundle_context(bundle, dimension=None) == default  # None == today, byte-identical
    assert "ASFALT-SENTINEL" in default  # no filter -> asfalt present
    assert "VERDICT-SENTINEL" not in default  # verdict still excluded


def test_navigate_tolerates_broken_links(tmp_path) -> None:
    """OKF SPEC §4: a consumer MUST tolerate broken links. An index linking a missing file
    navigates without raising, simply omitting the absent target."""
    (tmp_path / "index.md").write_text(
        "---\ntype: index\n---\n\nSee [gone](missing.md) and [here](real.md).\n",
        encoding="utf-8",
    )
    (tmp_path / "real.md").write_text("---\ntype: project\n---\n\nbody\n", encoding="utf-8")
    bundle = okf.navigate_bundle(str(tmp_path))
    names = {f.name for f in bundle.files}
    assert names == {"index.md", "real.md"}  # missing.md silently skipped, no raise


def test_navigate_skips_null_byte_link(tmp_path) -> None:
    """OKF SPEC §4 (okf.py:8-9): a broken link is NEVER raised. A link target carrying an embedded
    null byte has no path separator, so it slips past the ``/`` pre-filter and reaches path
    resolution, where ``os.path.realpath`` raises ``ValueError`` — which MUST be absorbed
    fail-closed (skipped), not propagated out of ``navigate_bundle``."""
    (tmp_path / "index.md").write_text(
        "---\ntype: index\n---\n\nSee [bad](a\x00b.md) and [ok](real.md).\n",
        encoding="utf-8",
    )
    (tmp_path / "real.md").write_text("---\ntype: project\n---\n\nbody\n", encoding="utf-8")
    bundle = okf.navigate_bundle(str(tmp_path))  # must not raise
    names = {f.name for f in bundle.files}
    assert names == {"index.md", "real.md"}  # null-byte target silently skipped


def test_navigate_skips_bundle_escaping_symlink_link(tmp_path) -> None:
    """OKF SPEC §4 + fail-closed path-safety: a same-dir link whose target is a symlink escaping the
    bundle is skipped (``safe_resolve`` raises ``PathSecurityError`` -> ``_load_file`` -> None),
    never read. Closes the untested path-safety-at-link-resolution branch of ``navigate_bundle``."""
    outside = tmp_path / "outside.md"
    outside.write_text("---\ntype: secret\n---\n\nEXFIL\n", encoding="utf-8")
    bundle_dir = tmp_path / "bundle"
    bundle_dir.mkdir()
    (bundle_dir / "index.md").write_text(
        "---\ntype: index\n---\n\nSee [escape](evil.md).\n", encoding="utf-8"
    )
    os.symlink(str(outside), str(bundle_dir / "evil.md"))
    bundle = okf.navigate_bundle(str(bundle_dir))
    names = {f.name for f in bundle.files}
    assert names == {"index.md"}  # escaping symlink target refused, never navigated
    assert all("EXFIL" not in f.body for f in bundle.files)


def test_load_ir_projection_returns_candidate_ir() -> None:
    """The bundle's IR projection (``validator-input.json``) is the candidate measure's cost-IR —
    the pre-hypothesis ExpeL query key source."""
    ir = okf.load_ir_projection(str(BUNDLE_DIR))
    assert ir["project_id"] == "BYGG-KONTOR-NORD"
    assert [a["code"] for a in ir["affected_items"]] == ["ENERGI-TOTAL-EL"]
    assert ir["claimed_saving_nok"] == 30000


def test_parse_frontmatter_reads_scalar_fields() -> None:
    """The minimal frontmatter reader returns the leading ``---`` block as key:value strings."""
    fm = okf.parse_frontmatter(BUNDLE_DIR / "verdict-led-fro.md")
    assert fm["type"] == "verdict"
    assert fm["decision"] == "approved_with_adjustment"


def test_render_frontmatter_roundtrips_consumed_fields(tmp_path) -> None:
    """Step-8 writer: ``render_frontmatter`` + ``write_concept_file`` emit a block that
    ``parse_frontmatter`` re-reads with the fields ``seed_store_from_bundle`` consumes
    (``type``, ``decision``, ``description``) intact. NOT a bijection — only these scalar fields
    are guaranteed to survive write -> read."""
    fm = {
        "type": "verdict",
        "decision": "approved",
        "description": "LED-retrofit godkjent (realiseringsgrad=0.57)",
        "verdict_id": "abc123",
    }
    okf.write_concept_file(str(tmp_path), "promoted-verdict-abc123.md", fm, "body prose\n")
    parsed = okf.parse_frontmatter(tmp_path / "promoted-verdict-abc123.md")
    assert parsed["type"] == "verdict"
    assert parsed["decision"] == "approved"
    assert parsed["description"] == "LED-retrofit godkjent (realiseringsgrad=0.57)"
    assert parsed["verdict_id"] == "abc123"


def test_render_frontmatter_single_lines_scalars(tmp_path) -> None:
    """A multi-line rationale must NOT corrupt the line-oriented frontmatter block (parse stops at
    ``---``). Newlines in a scalar value are flattened to spaces, so every following key survives."""
    fm = {
        "type": "verdict",
        "description": "line one\nline two\n---\nnot a delimiter",
        "decision": "approved",
    }
    okf.write_concept_file(str(tmp_path), "f.md", fm, "body\n")
    parsed = okf.parse_frontmatter(tmp_path / "f.md")
    assert "\n" not in parsed["description"]
    assert parsed["decision"] == "approved"  # the trailing key was NOT lost to a spurious ---


def _minimal_bundle(tmp_path) -> str:
    (tmp_path / "index.md").write_text(
        "---\ntype: index\n---\n\n# Bundle\n\n- [proj](bygg.md)\n", encoding="utf-8"
    )
    (tmp_path / "bygg.md").write_text("---\ntype: project\n---\n\nbody\n", encoding="utf-8")
    return str(tmp_path)


def test_link_in_index_makes_concept_file_navigable(tmp_path) -> None:
    """Step-8 writer: a file written into a bundle is reachable by ``navigate_bundle`` only after
    ``link_in_index`` adds an intra-bundle cross-link the navigator follows (``_LINK_RE``)."""
    bundle_dir = _minimal_bundle(tmp_path)
    fm = {"type": "verdict", "decision": "approved", "description": "d"}
    okf.write_concept_file(bundle_dir, "promoted-verdict-x.md", fm, "body\n")
    assert "promoted-verdict-x.md" not in {f.name for f in okf.navigate_bundle(bundle_dir).files}
    added = okf.link_in_index(bundle_dir, "promoted-verdict-x.md", "Promotert")
    assert added is True
    bundle = okf.navigate_bundle(bundle_dir)
    assert "promoted-verdict-x.md" in {f.name for f in bundle.files}
    assert [f.name for f in bundle.verdicts] == ["promoted-verdict-x.md"]


def test_link_in_index_is_idempotent(tmp_path) -> None:
    """Linking the same target twice adds exactly one bullet (returns ``False`` the second time) —
    so re-promoting an existing verdict does not double-link the index."""
    bundle_dir = _minimal_bundle(tmp_path)
    assert okf.link_in_index(bundle_dir, "promoted-verdict-x.md", "A") is True
    assert okf.link_in_index(bundle_dir, "promoted-verdict-x.md", "B") is False
    body = (tmp_path / "index.md").read_text(encoding="utf-8")
    assert body.count("(promoted-verdict-x.md)") == 1


def test_link_in_index_success_preserves_existing_bytes_and_order(tmp_path) -> None:
    """LOAD-BEARING (ingest-spec §6 byte-preservation, Step-8 promotion invariant): on the SUCCESS
    path (a link IS added), ``link_in_index`` preserves the pre-existing index byte-for-byte AND in
    order, appending ONLY the new bullet. Existing coverage asserts byte-equality solely on the
    REFUSAL path (``test_step8_promotion_loadbearing.py`` Test A) and mere line MEMBERSHIP elsewhere,
    so a writer that kept every link but reordered/rewrote the existing body would pass the whole
    suite. The existing links are deliberately NON-sorted so a reordering mutation (e.g. ``sorted``)
    flips this RED."""
    original = (
        "---\ntype: index\n---\n\n# Bundle\n\n"
        "- [zeta](zeta.md)\n- [alpha](alpha.md)\n- [mid](mid.md)\n"
    )
    (tmp_path / "index.md").write_text(original, encoding="utf-8")
    added = okf.link_in_index(str(tmp_path), "promoted-verdict-x.md", "Promotert")
    assert added is True
    result = (tmp_path / "index.md").read_text(encoding="utf-8")
    assert result == original + "- [Promotert](promoted-verdict-x.md)\n"


def _dynamic_import_targets(node: ast.Call) -> list[str]:
    """Name any dynamic-import call: ``__import__(...)`` or ``<anything>.import_module(...)``.

    Shared by the MAF-free guard here and the no-network guard in
    ``tests/test_semretrieval_loadbearing.py`` — both walk imports statically, and both are blind
    to a dynamic import by construction."""
    func = node.func
    if isinstance(func, ast.Name) and func.id == "__import__":
        return ["__import__"]
    if isinstance(func, ast.Attribute) and func.attr == "import_module":
        return ["import_module"]
    return []


@pytest.mark.parametrize("module_name", _MAF_FREE_MODULES)
def test_okf_is_maf_free(module_name: str) -> None:
    """D7 portability: each framework-neutral module IMPORTS no ``agent_framework`` / ``mcp`` (a
    docstring may name them to document the constraint, exactly as ``retrieval.py`` does) — checked
    via the AST, not a raw substring, so the prose claim doesn't trip the guard. Parametrized over
    ``_MAF_FREE_MODULES`` so ``dimension.py`` is guarded alongside ``okf.py`` (C2)."""
    src = (
        Path(__file__).resolve().parents[1] / "src" / "portfolio_optimiser" / module_name
    ).read_text(encoding="utf-8")
    imported: list[str] = []
    dynamic: list[str] = []
    for node in ast.walk(ast.parse(src)):
        if isinstance(node, ast.Import):
            imported += [a.name for a in node.names]
        elif isinstance(node, ast.ImportFrom):
            imported.append(node.module or "")
        elif isinstance(node, ast.Call):
            dynamic += _dynamic_import_targets(node)
    forbidden = [m for m in imported if m.split(".")[0] in {"agent_framework", "mcp"}]
    assert forbidden == [], f"{module_name} must not import MAF/mcp, found: {forbidden}"
    # A RATCHET, green today: none of these modules imports ``importlib`` or calls ``__import__``.
    # The sweep above walks only ``ast.Import``/``ast.ImportFrom``, so a single
    # ``importlib.import_module("portfolio_optimiser.verdicts")`` would sail straight past it and
    # pull MAF into a module this guard certifies as MAF-free. Any dynamic import is refused
    # outright rather than argument-inspected: a computed target cannot be judged statically.
    assert dynamic == [], (
        f"{module_name} performs dynamic import(s) {dynamic} — the MAF-free guard is a STATIC "
        "check and cannot see through them; use a normal import so it stays enforceable"
    )