portfolio-optimiser/tests/test_requirement_comparison_loadbearing.py

"""P22 DEL B — ``declare_requirement`` answers with a COMPARISON, not a confirmation.

MEASURED (P21 funn 2, re-measured at the head of okt 126 against the six round-5 debate traces):
``requirement_hit`` is **0 of 20** approach rows and **0 of 12** declarations — the third round in
a row at zero. P21/C1 made the runs LOOK first, and it worked on its own terms: distinct documents
opened before a declaration went from 1,1,1,2,5,13 to 3,3,5,7,11,12. The hit did not move. The
runs were made to read more, not righter.

P20/A1 had already made the reply carry the document's OWN title and number instead of echoing the
caller's arguments. What nobody said was whether that document has anything to do with the
direction the run is committed to. This turns the reply into that comparison.

**A REPORT, never a gate.** The declaration is recorded either way, and arm (c) is what holds that
open. A gate on word overlap would refuse legitimate declarations — a requirement can bind a
measure without sharing a word with the name someone gave it — which is precisely how the
alternative rule P21/C1 measured and rejected failed, one rung over.

**Measured before it was built**, offline against the six traces: the rule speaks on **10 of 12**
declarations and stays quiet on 2 (both fv412, on ``materialer``). A rule that spoke on 12 of 12,
or on 0 of 12, could not tell the two classes apart.
"""

from __future__ import annotations

from pathlib import Path
from typing import Any

import pytest
from portfolio_optimiser.simulation import scripted_factory

from portfolio_optimiser import okf
from portfolio_optimiser.explore import ToolCall, navigator_tools
from portfolio_optimiser.mandate import Approach, Mandate
from portfolio_optimiser.run import run_project

_EXAMPLES = Path(__file__).resolve().parents[1] / "shared" / "examples"
_TUNNEL = _EXAMPLES / "tunnel-hauglia"
_BASE_ID = "tunnel-hauglia"

#: The fixture document the arms declare. Its title is ASCII-clean, which is what lets a label
#: share a word with it without a marker carrying a multibyte character into a scripted run.
_DOC = "tiltak-portalskjerming.md"

#: A direction whose words are IN that title ("Portalskjerming: senke L20 ...").
_MATCHING = "Billigere portalskjerming"
#: A direction that shares nothing with it. Checked against the document's own tokens, both ways.
_FOREIGN = "Asfaltdekke gjenbruk"


def _wired(labels: tuple[str, ...]) -> tuple[dict[str, Any], list[ToolCall], list[Any]]:
    opened: list[ToolCall] = []
    declared: list[Any] = []
    tools = navigator_tools((str(_TUNNEL),), opened=opened, requirements=declared, labels=labels)
    return {t.name: t for t in tools}, opened, declared


def _declare(
    labels: tuple[str, ...], *, ref: str = "Krav 1.1-1"
) -> tuple[dict[str, Any], list[Any]]:
    tools, opened, declared = _wired(labels)
    for name in [f.name for f in okf.navigate_bundle(str(_TUNNEL)).context_files][:3]:
        opened.append(ToolCall(name="read_file", bundle_id=_BASE_ID, path=name))
    opened.append(ToolCall(name="read_file", bundle_id=_BASE_ID, path=_DOC))
    answer = tools["declare_requirement"].func(
        approach_id="a1", bundle_id=_BASE_ID, path=_DOC, ref=ref
    )
    return answer, declared


# --- (a)/(b) the discriminator: SAME document, SAME ref, different directions ---------------------


def test_a_direction_that_shares_a_word_is_told_which_one() -> None:
    """KNOWN-POSITIVE. The document's title carries the direction's own word, and the reply says
    so — the half that keeps the report from being one that only ever complains."""
    answer, _ = _declare((_MATCHING,))
    assert answer["overlap"] == ["portalskjerming"], answer["overlap"]
    assert "portalskjerming" in answer["compare"]


def test_a_direction_that_shares_nothing_is_told_that_too() -> None:
    """KNOWN-NEGATIVE, and the measured case: the SAME document and the SAME ``ref`` as the arm
    above — the only difference is the direction. That is what makes this a comparison rather than
    a confirmation, and it is exactly the 10-of-12 class the round-5 traces fall into."""
    answer, _ = _declare((_FOREIGN,))
    assert answer["overlap"] == []
    assert "No word of any of them appears" in answer["compare"], answer["compare"]
    assert repr(_FOREIGN) in answer["compare"], answer["compare"]


# --- (c) a REPORT, never a gate ------------------------------------------------------------------


def test_a_declaration_with_no_overlap_is_still_recorded() -> None:
    """The whole difference between this rung and P21/C1's. A requirement can bind a measure
    without sharing a word with the name someone gave it, so refusing here would refuse
    legitimate declarations — the failure of the alternative rule the C1 measurement rejected."""
    answer, declared = _declare((_FOREIGN,))
    assert answer["declared"] is True
    assert len(declared) == 1, "a reported mismatch must not swallow the declaration"


# --- (d) it reads the DOCUMENT, never the caller's own argument ----------------------------------


def test_the_comparison_never_reads_the_callers_own_ref() -> None:
    """``ref`` is the caller's argument echoed back, and a comparison against the caller's own
    input can only ever agree — P20/A1's rule (read off the base, never off the arguments)
    applied to the half P20 did not reach. A ``ref`` stuffed with the direction's words must not
    manufacture an overlap."""
    answer, _ = _declare((_FOREIGN,), ref="Asfaltdekke gjenbruk krav")
    assert answer["overlap"] == [], answer["overlap"]


# --- (e) no directions -> the P20 reply, unchanged ------------------------------------------------


def test_without_directions_the_reply_is_the_one_p20_shipped() -> None:
    """The exploration mints its own directions, so at declaration time it HAS none — and every
    call site that passes no labels must carry no comparison keys (row 6 added ``approach_id`` to
    every reply, the address the declaration was filed under). Absent keys, not empty ones:
    "there was nothing to compare against" and "we compared and found nothing" are different
    facts, and only one of them is true here."""
    answer, _ = _declare(())
    assert set(answer) == {
        "declared",
        "bundle_id",
        "path",
        "ref",
        "approach_id",
        "title",
        "req_number",
        "binds",
    }, sorted(answer)


# --- (f)/(g) the matching rule's two measured properties -----------------------------------------


def test_matching_is_generous_in_both_directions() -> None:
    """The failure direction chosen on purpose. A direction whose word is a PREFIX of the
    document's own word counts, and so does the reverse — Norwegian inflects ("rundkjoring" vs
    "Rundkjoringer") and a strict rule would report "no overlap" on a declaration that was right,
    which is the only one of the two errors that can push a model away from a correct answer."""
    # label word LONGER than the document's own ("portalskjerming" is a prefix of it)
    assert _declare(("Portalskjermingen paa nordsiden",))[0]["overlap"] == ["portalskjermingen"]
    # and SHORTER: the document says "senke", the direction "senkekostnader"
    assert _declare(("Senkekostnader",))[0]["overlap"] == ["senkekostnader"]


def test_short_words_cannot_manufacture_an_overlap() -> None:
    """Below the minimum every label shares "for", "med", "til" with half a corpus, and the report
    would speak of an overlap nobody meant. The control is the arm above: the same rule that
    rejects these still finds the real word."""
    answer, _ = _declare(("Ny og med til for",))
    assert answer["overlap"] == [], answer["overlap"]


# --- (h) the commission's labels actually REACH the tool in a run --------------------------------

_PID = "TUNNEL-HAUGLIA"
_VERDICT_INPUT = {"decision": "approved", "rationale": "expert reviewed (test)"}
_VALID_REPLY = (
    '{"measure":"LED-retrofit","affected_items":'
    '[{"code":"ENERGI-TOTAL-EL","quantity":300000,"unit_cost":1.0}],'
    '"claimed_saving_nok":30000}'
)
_CHECKER_REPLY = "Reasoning holds.\nVERDICT: APPROVE"


@pytest.mark.asyncio
async def test_a_commissioned_run_reaches_the_tool_with_its_own_directions() -> None:
    """The WIRING, measured behaviourally rather than by reading ``run.py``. A source assertion is
    a lint; this drives the real debate with a step manuscript that declares, and reads the
    comparison back out of the tool's own answer. Without ``labels=`` in ``run.py`` the reply
    carries no ``compare`` at all and this arm falls."""
    concepts = [f.name for f in okf.navigate_bundle(str(_TUNNEL)).context_files]
    script = {
        "proposer": [
            *(
                {"call": "read_file", "args": {"bundle_id": _BASE_ID, "path": name}}
                for name in concepts[:3]
            ),
            {"call": "read_file", "args": {"bundle_id": _BASE_ID, "path": _DOC}},
            {
                "call": "declare_requirement",
                "args": {
                    "bundle_id": _BASE_ID,
                    "path": _DOC,
                    "ref": "Krav 1.1-1",
                    "approach_id": "a1",
                },
            },
            _VALID_REPLY,
            _VALID_REPLY,
            _VALID_REPLY,
            _VALID_REPLY,
        ],
        "checker": _CHECKER_REPLY,
    }
    seen: list[str] = []

    def factory(role: str) -> Any:
        client = scripted_factory(script, [])(role)
        original = client._inner_get_response

        def recording(*, messages, options, stream=False, **kwargs):  # type: ignore[no-untyped-def]
            for message in messages:
                for content in getattr(message, "contents", ()) or ():
                    value = getattr(content, "result", None)
                    if value:
                        seen.append(str(value))
            return original(messages=messages, options=options, stream=stream, **kwargs)

        client._inner_get_response = recording
        return client

    await run_project(
        _PID,
        "local",
        docs_dir=str(_TUNNEL),
        bundle_dir=str(_TUNNEL),
        verdict_input=_VERDICT_INPUT,
        client_factory=factory,
        mandate=Mandate(
            objective="Find savings",
            approaches=(Approach(id="a1", label=_FOREIGN, description="expert's reason"),),
        ),
    )

    blob = "\n".join(seen)
    assert "No word of any of them appears" in blob, (
        "the commission's directions never reached the declaration rung: " + blob[-600:]
    )
    assert _FOREIGN in blob