portfolio-optimiser/tests/test_v1_probes.py

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