"""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 null byte is an INVALID PATH COMPONENT, so path resolution raises ``ValueError`` — which MUST be absorbed fail-closed (skipped), not propagated out of ``navigate_bundle``. This is the *malformed* sub-class of method spec §3 Step 1's "a target that fails to resolve for ANY reason (missing file, invalid path component, escape) is skipped, not raised" (§11 row "Navigation boundary"). The rationale rests on the RESOLVE guard alone, never on the retired "a path separator means out-of-bundle" heuristic: this test stays green — and its reason stays true — once that heuristic goes and nested targets become legal.""" (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 ``.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" )