"""v1 gate probes — the named tests the v1 gate (``python -m portfolio_optimiser.evals.v1_gate``) runs to decide two of its rows. Every test here that is RED today carries ``xfail(strict=True)``, so the ordinary suite stays green while the gap is real, and the gate runs the file with ``--runxfail`` so the gap shows as red there. ``strict`` is the other half: the day a capability makes one of these pass, the suite goes RED on the XPASS until the marker is removed — a closed gap cannot stay labelled open. **Row 3 (feedback types with a way in AND an action).** Types 1, 3 and 7 are proven by EXISTING tests elsewhere in the suite (registered by node id in ``evals/v1_gate.json``). The five types with no complete surface (2, 4, 5, 6, 8) get a probe here that is red BECAUSE the surface is missing, never a missing test. Each probe measures the absence (the CLI's own ``--help``); if a matching option appears it STILL fails, naming the option — "partial is no", and a door with no observed action is exactly partial. Such a probe goes green only when it is rewritten to drive the new door and observe what it does. **Row 6 (a validated proposal whose approach declared no requirement).** Two probes against the real ``run_project`` on the shipped micro base, where the debate holds ``declare_requirement``: no declaration anywhere, and a declaration filed for ANOTHER approach (the run's own proposal). Neither may leave ``a1`` validated — a declaration counts only under its own approach's id. """ from __future__ import annotations import contextlib import io import json import re from pathlib import Path from typing import Any import pytest from portfolio_optimiser import okf, run from portfolio_optimiser.mandate import Approach, BindingRequirement, Mandate from portfolio_optimiser.run import run_project from portfolio_optimiser.simulation import scripted_factory from portfolio_optimiser.verdicts import VerdictStore _BUNDLE = Path(__file__).resolve().parents[1] / "shared" / "examples" / "bygg-energi-mikro" _BASE_ID = "bygg-energi-mikro" _PID = "BYGG-KONTOR-NORD" _VALID_REPLY = ( '{"measure":"LED-retrofit av kontorbelysning","affected_items":' '[{"code":"ENERGI-TOTAL-EL","quantity":300000,"unit_cost":1.0}],"claimed_saving_nok":30000}' ) _CHECKER_REPLY = "Reasoning holds.\nVERDICT: APPROVE" _NO_SURFACE = "v1 probe: no surface" _PARTIAL = "v1 probe: surface without an observed action" def _cli_options() -> set[str]: """Every option string the CLI's own ``--help`` prints — the surface, measured.""" buffer = io.StringIO() with contextlib.redirect_stdout(buffer), pytest.raises(SystemExit): run.main(["--help"]) return set(re.findall(r"--[a-z][a-z-]*", buffer.getvalue())) def _surface_or_fail(type_no: int, what: str, keywords: tuple[str, ...]) -> None: hits = sorted(o for o in _cli_options() if any(k in o for k in keywords)) if not hits: pytest.fail(f"{_NO_SURFACE}: type {type_no} ({what}) — no CLI option matches {keywords}") pytest.fail( f"{_PARTIAL}: type {type_no} ({what}) — {hits} appeared; rewrite this probe to drive it " "and observe the action" ) # --------------------------------------------------------------------------------------------- # Row 3 — types 3 and 7: the way in is the REAL flag, and the action shows in the RESULT # --------------------------------------------------------------------------------------------- _REQUIREMENT = {"path": "tiltak-led-retrofit.md", "ref": "Krav 1"} def _cli_run(tmp_path: Path, replies: dict[str, Any], *extra: str) -> tuple[int, Path, str]: """One in-process CLI run on the micro base with scripted replies, writing an outbox.""" replies_file = tmp_path / "replies.json" replies_file.write_text(json.dumps(replies), encoding="utf-8") out = tmp_path / "out" buffer = io.StringIO() with contextlib.redirect_stdout(buffer): rc = run.main( [ _PID, "--bundle-dir", str(_BUNDLE), "--scripted-replies", str(replies_file), "--outbox-dir", str(out), "--run-id", "probe", *extra, ] ) return rc, out, buffer.getvalue() def test_type_3_a_commissioned_angle_is_evaluated_through_the_cli(tmp_path: Path) -> None: """``--mandate`` is the way in; the action is a VALIDATED outcome for that angle in the outbox. A settlement line alone is not enough — a ``NOT EVALUATED`` row prints the id too.""" mandate = tmp_path / "mandate.json" mandate.write_text( json.dumps( { "objective": "Kutt energikostnad", "approaches": [ {"id": "ny-vinkling", "label": "LED-retrofit", "requirement": _REQUIREMENT} ], "allow_own_proposals": False, } ), encoding="utf-8", ) rc, out, stdout = _cli_run( tmp_path, {"proposer": _VALID_REPLY, "checker": _CHECKER_REPLY}, "--mandate", str(mandate) ) assert rc == 0, stdout coverage = json.loads((out / "probe-coverage.json").read_text(encoding="utf-8")) assert [(r["id"], r["status"]) for r in coverage["rows"]] == [("ny-vinkling", "validated")] outcome = json.loads((out / "probe-ny-vinkling-outcome.json").read_text(encoding="utf-8")) assert outcome["outcome_type"] == "validated" @pytest.mark.asyncio async def test_type_3_a_new_angle_changes_the_outcome() -> None: """Adding an angle in a later round changes what the run carries: the new angle's own reply (reachable only if its label reached the prompt) becomes the selected, validated outcome.""" labels = {"Behovsstyrt belysning": 30_000, "Nattsenking av temperatur": 60_000} def select(prompt: str, _role: str) -> str: claimed = next((v for k, v in labels.items() if k in prompt), 10_000) return _VALID_REPLY.replace("30000}", f"{claimed}}}") requirement = BindingRequirement(**_REQUIREMENT) first = Approach(id="a1", label="Behovsstyrt belysning", requirement=requirement) second = Approach(id="a2", label="Nattsenking av temperatur", requirement=requirement) async def outcome(*approaches: Approach) -> Any: return await run_project( _PID, "local", docs_dir=str(_BUNDLE), bundle_dir=str(_BUNDLE), store=VerdictStore(verdicts=[]), client_factory=scripted_factory({"proposer": select, "checker": _CHECKER_REPLY}, []), mandate=Mandate(objective="o", approaches=approaches, allow_own_proposals=False), ) before = await outcome(first) after = await outcome(first, second) assert before.outcome.proposal.claimed_saving_nok == 30_000 assert {r.id: r.status for r in after.coverage} == {"a1": "validated", "a2": "validated"} assert after.outcome.proposal.claimed_saving_nok == 60_000 def test_type_7_the_mcp_flag_puts_a_service_the_run_calls_into_the_result( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """``--mcp-config`` is the way in; the action is the RESULT recording that the debate called the configured service. Without the flag the same script leaves no external call.""" from tests.test_b4_mcp_call_trace_loadbearing import _as_context_manager, _lookup_unit_price monkeypatch.setattr( run, "build_mcp_tools", lambda _c: [_as_context_manager(_lookup_unit_price)] ) config = tmp_path / "mcp.json" config.write_text( json.dumps( { "servers": [ { "name": "prisregister", "transport": "http", "url": "https://intern.example/mcp", "allowed_tools": ["lookup_unit_price"], "timeout_seconds": 15, } ] } ), encoding="utf-8", ) replies = { "proposer": [ {"call": "lookup_unit_price", "args": {"code": "ENERGI-TOTAL-EL"}}, *([_VALID_REPLY] * 4), ], "checker": _CHECKER_REPLY, } def calls(sub: str, *extra: str) -> list[Any]: (tmp_path / sub).mkdir() rc, out, stdout = _cli_run(tmp_path / sub, replies, *extra) assert rc == 0, stdout proposal = json.loads((out / "probe-proposal.json").read_text(encoding="utf-8")) return list(proposal["provenance"]["external_calls"]) assert calls("with", "--mcp-config", str(config)) == [ {"server": "prisregister", "tool": "lookup_unit_price"} ] assert calls("without") == [] # --------------------------------------------------------------------------------------------- # Row 3 — the five types without a complete surface # --------------------------------------------------------------------------------------------- @pytest.mark.xfail(strict=True, reason="type 2: no typed removal; only revise free text") def test_type_2_remove_a_direction_has_a_typed_door() -> None: _surface_or_fail(2, "take a direction away", ("drop", "remove", "exclude", "withdraw")) @pytest.mark.xfail(strict=True, reason="type 4: no surface relaxes a requirement") def test_type_4_relax_a_requirement_has_a_door() -> None: _surface_or_fail(4, "relax a requirement", ("relax", "waive", "loosen")) @pytest.mark.xfail(strict=True, reason="type 5: concept graph edits have no CLI door") def test_type_5_edit_the_concept_graph_has_a_door() -> None: _surface_or_fail(5, "edit the concept graph", ("promote", "concept", "graph")) @pytest.mark.xfail(strict=True, reason="type 6: no skills flag") def test_type_6_skills_per_analysis_has_a_door() -> None: _surface_or_fail(6, "skills per analysis", ("skill",)) @pytest.mark.xfail(strict=True, reason="type 8: no door for inline context such as meeting notes") def test_type_8_inline_context_has_a_door() -> None: _surface_or_fail(8, "inline context", ("note", "minutes", "inline", "attach")) # --------------------------------------------------------------------------------------------- # Row 6 — a validated proposal must rest on a declaration its approach made # --------------------------------------------------------------------------------------------- def _mandate() -> Mandate: return Mandate( objective="Kutt energikostnad", approaches=(Approach(id="a1", label="LED-retrofit", description="expert's reason"),), ) async def _statuses(script: dict[str, Any], tmp_path: Path, base: Path) -> dict[str, str]: result = await run_project( _PID, "local", docs_dir=str(base), bundle_dir=str(base), store=VerdictStore(verdicts=[]), client_factory=scripted_factory(script, []), mandate=_mandate(), outbox_dir=str(tmp_path / "out"), run_id="v1-row6", ) return {row.id: row.status for row in result.coverage} def _declaring_script(base: Path, approach_id: str) -> dict[str, Any]: concepts = [f.name for f in okf.navigate_bundle(str(base)).context_files][:3] return { "proposer": [ *({"call": "read_file", "args": {"bundle_id": _BASE_ID, "path": n}} for n in concepts), { "call": "declare_requirement", "args": { "bundle_id": _BASE_ID, "path": concepts[0], "ref": "Krav 1.1-1", "approach_id": approach_id, }, }, _VALID_REPLY, _VALID_REPLY, _VALID_REPLY, _VALID_REPLY, ], "checker": _CHECKER_REPLY, } def _declared(tmp_path: Path) -> list[dict[str, Any]]: debate = json.loads((tmp_path / "out" / "v1-row6-debate.json").read_text(encoding="utf-8")) return list(debate["requirements"]) @pytest.mark.asyncio async def test_row6_an_approach_that_declared_nothing_cannot_be_validated(tmp_path: Path) -> None: base = _BUNDLE statuses = await _statuses( {"proposer": _VALID_REPLY, "checker": _CHECKER_REPLY}, tmp_path, base ) assert _declared(tmp_path) == [] # precondition: nothing was declared anywhere assert statuses["a1"] != "validated", "validated without any declared requirement" @pytest.mark.asyncio async def test_row6_a_run_level_declaration_does_not_stand_in_for_the_approach( tmp_path: Path, ) -> None: """The debate declares a requirement — but for the run's OWN proposal, not for ``a1``.""" base = _BUNDLE statuses = await _statuses(_declaring_script(base, "own-proposal"), tmp_path, base) declared = _declared(tmp_path) assert [d.get("approach_id") for d in declared] == ["own-proposal"] # precondition assert statuses["a1"] != "validated", "validated on a declaration the approach never made"