portfolio-optimiser/tests/test_scripted_explore_door_loadbearing.py

"""MAJOR-2 (docs/2026-08-25-syretest-vei-ab.md) — ``--explore --scripted-replies`` must not
crash with a raw ``KeyError: 'navigator'``.

``_SCRIPTED_ROLES = ("proposer", "checker")`` (``run.py:1531``) is the debate's two roles.
``explore()`` asks the SAME ``client_factory`` for three more: ``manager``, ``navigator``,
``hypothesiser`` (``explore.py:576``). ``_load_scripted_replies`` was fail-fast for the two roles
it knew about — measured (docs/2026-08-25-syretest-vei-ab.md § MAJOR-2) to let the three it did not
know about surface exactly the ``KeyError`` deep inside ``scripted_factory``'s lookup its own
docstring warns against, mid-run, after the banner had already printed.

The fix widens the required-role set to include the exploration's three roles WHEN ``--explore``
is in play, so a missing role is refused BY NAME before any model/agent work starts — the same
door, never a second one.

MAJOR-1 (``docs/2026-09-02-misjonsreview-v2.md`` § 7) — and then the door that no longer crashed
turned out to be VACUOUS BY CONSTRUCTION, with an artefact that could not see it.

One constant string per role means no scripted role can ever emit a ``function_call``: measured,
0 tool calls / 0 approaches / 1 round on 4/4 bases, while every other field of
``{run_id}-exploration.json`` looked like a run that had worked. The third test above says as much
in its own docstring and asserts only the absence of a traceback — which was honest, and is
exactly the ceiling this section raises. An operator following this repo's measurement ladder
("prove as much as possible for free before the paid step") could not prove the navigator opens
anything, and after a PAID run nobody could read whether it had.

Two halves, each asserted so the OTHER cannot carry it:

(a) ``ExplorationToolRecorder`` — a ``FunctionMiddleware`` on the exploration agents recording the
    tool NAME and the ``bundle_id`` it was asked for (never the result), in invocation order, onto
    the caller-owned ``ExplorationTrace``; ``trace_payload`` writes it beside ``quick_validations``.
(b) ``_load_scripted_replies`` accepts, for the exploration roles, a LIST of steps where a step may
    be ``{"call": "<tool>", "args": {...}}``, and ``scripted_factory`` then builds a client that
    emits a ``function_call``.

Without (a) the record is empty though the tools ran; without (b) the tools never run though the
recorder is wired. The end-to-end test is RED against either detached, and the constant-string
control proves the recorder does not invent entries.
"""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any

from portfolio_optimiser import explore as ex
from portfolio_optimiser import run

_BUNDLE_DIR = Path(__file__).resolve().parents[1] / "shared" / "examples" / "bygg-energi-mikro"
_PID = "BYGG-KONTOR-NORD"

_PROPOSER_REPLY = json.dumps(
    {
        "measure": "LED-retrofit",
        "affected_items": [{"code": "ENERGI-TOTAL-EL", "quantity": 300000, "unit_cost": 1.0}],
        "claimed_saving_nok": 30000,
    }
)


def _config_file(tmp_path: Path, **overrides: Any) -> str:
    path = tmp_path / "exploration.json"
    path.write_text(
        json.dumps(
            {
                "max_rounds": 4,
                "max_tokens": 100_000,
                "max_stall_count": 2,
                "max_reset_count": 1,
                "max_plan_revisions": 0,
                "enable_plan_review": False,
                **overrides,
            }
        ),
        encoding="utf-8",
    )
    return str(path)


def _replies_file(tmp_path: Path, roles: dict[str, str]) -> str:
    path = tmp_path / "replies.json"
    path.write_text(json.dumps(roles), encoding="utf-8")
    return str(path)


def _base_argv(tmp_path: Path, replies_path: str) -> list[str]:
    return [
        _PID,
        "--docs-dir",
        str(_BUNDLE_DIR),
        "--bundle-dir",
        str(_BUNDLE_DIR),
        "--explore",
        "Find the cheapest saving.",
        "--explore-config",
        _config_file(tmp_path),
        "--scripted-replies",
        replies_path,
    ]


def test_explore_with_debate_only_scripted_replies_is_refused_by_name(tmp_path, capsys) -> None:
    """A ``--scripted-replies`` file that only answers the debate (proposer/checker) — exactly the
    file MAJOR-2 was measured against — is refused BY NAME, never left to crash mid-run.

    Detach point: revert ``_SCRIPTED_ROLES`` to the fixed debate-only tuple used for ``--explore``
    too → this raises an unhandled ``KeyError`` instead of returning 1 (RED, reproduces MAJOR-2).
    """
    replies = _replies_file(tmp_path, {"proposer": _PROPOSER_REPLY, "checker": "VERDICT: APPROVE"})

    rc = run.main(_base_argv(tmp_path, replies))

    assert rc == 1, "a role the exploration needs is missing — this must be a clean refusal"
    err = capsys.readouterr().err
    assert "run refused" in err
    for role in ("manager", "navigator", "hypothesiser"):
        assert role in err, f"the refusal must name the missing role {role!r}"


def test_explore_with_debate_only_scripted_replies_never_reaches_a_model_call(
    tmp_path, capsys, monkeypatch
) -> None:
    """The refusal fires at the DOOR — before ``explore()`` is even entered. A spy on ``explore``
    proves zero exploration work happened, the same way econ 57's outbox/run-id hoist was proved.

    Detach point: let the flag through and catch the ``KeyError`` further in → this spy would still
    record a call (RED).
    """
    calls: list[object] = []
    monkeypatch.setattr(
        "portfolio_optimiser.run.explore",
        lambda *a, **kw: (
            calls.append((a, kw)) or (_ for _ in ()).throw(AssertionError("unreached"))
        ),
    )
    replies = _replies_file(tmp_path, {"proposer": _PROPOSER_REPLY, "checker": "VERDICT: APPROVE"})

    rc = run.main(_base_argv(tmp_path, replies))

    assert rc == 1
    assert calls == [], "explore() must never be entered when a required role is missing"


def test_explore_with_the_full_five_role_scripted_replies_does_not_crash(tmp_path, capsys) -> None:
    """With every role ``explore()`` can ask for supplied as a constant string, the CLI door must
    run the loop to completion (or a typed budget/refusal outcome) — never a raw traceback.

    A constant per-role reply cannot answer every stage-specific shape the magentic manager can be
    prompted with (facts / plan / progress-ledger JSON / final answer all differ) — so this does
    not assert the exploration finds anything, only that the documented crash is gone.
    """
    replies = _replies_file(
        tmp_path,
        {
            "proposer": _PROPOSER_REPLY,
            "checker": "VERDICT: APPROVE",
            "manager": '{"is_request_satisfied": {"reason": "r", "answer": true}, '
            '"is_in_loop": {"reason": "r", "answer": false}, '
            '"is_progress_being_made": {"reason": "r", "answer": true}, '
            '"next_speaker": {"reason": "r", "answer": "hypothesiser"}, '
            '"instruction_or_question": {"reason": "r", "answer": "go"}}',
            "navigator": "NAVIGATOR: read the index.",
            "hypothesiser": "HYPOTHESIS: "
            + json.dumps(
                {"label": "x", "rationale": "y", "requirement": None, "why_none": "scripted"}
            ),
        },
    )

    rc = run.main(_base_argv(tmp_path, replies))

    err = capsys.readouterr().err
    assert "KeyError" not in err
    assert "Traceback" not in err
    assert rc in (0, 1), f"expected a clean exit, got rc={rc} stderr={err!r}"


# ---------------------------------------------------------------------------------------------
# MAJOR-1 (a) — THE RECORDER: name + the base asked for, in order, never the result
# ---------------------------------------------------------------------------------------------


class _FakeTool:
    def __init__(self, name: str) -> None:
        self.name = name


class _FakeContext:
    """The two attributes the middleware reads, and nothing else.

    Invoked DIRECTLY rather than through a driven run, for the reason
    ``test_explore_loadbearing``'s tool-freedom test calls the tool bodies directly: until half
    (b) exists no scripted client emits a tool call, so a middleware test routed through one would
    exercise nothing — the vacuity this section is about, rebuilt inside its own gate.
    """

    def __init__(self, name: str, arguments: Any) -> None:
        self.function = _FakeTool(name)
        self.arguments = arguments


async def _invoke(recorder: "ex.ExplorationToolRecorder", context: Any) -> bool:
    called = False

    async def _next() -> None:
        nonlocal called
        called = True

    await recorder.process(context, _next)
    return called


def test_the_recorder_keeps_every_call_in_order_with_the_base_it_named() -> None:
    """Order and repetition ARE the signal. ``mcp_tools.ToolCallRecorder``'s sorted, de-duplicated
    set answers "what was contacted" because its record is an egress claim; this one must answer
    "what was opened, and in what sequence" — a set cannot tell a rehearsal that read a base from
    one that only listed them, which is the whole of MAJOR-1.

    Detach point: sort or de-duplicate the sink → RED on both the order and the repeat.
    """
    import asyncio

    sink: list[ex.ToolCall] = []
    recorder = ex.ExplorationToolRecorder(sink)

    asyncio.run(_invoke(recorder, _FakeContext("read_bundle", {"bundle_id": "b2"})))
    asyncio.run(_invoke(recorder, _FakeContext("list_bundles", {})))
    asyncio.run(_invoke(recorder, _FakeContext("read_bundle", {"bundle_id": "b2"})))

    assert [(c.name, c.bundle_id) for c in sink] == [
        ("read_bundle", "b2"),
        ("list_bundles", ""),
        ("read_bundle", "b2"),
    ], (
        "invocation order and repeats are the record; a sorted, de-duplicated one answers something else"
    )


def test_a_call_is_recorded_but_never_altered_or_blocked() -> None:
    """It observes only — ``call_next`` is always awaited. A trace that changed the run it traces
    would not be a trace (the ``ToolCallRecorder`` rule, restated one layer up)."""
    import asyncio

    sink: list[ex.ToolCall] = []
    assert (
        asyncio.run(_invoke(ex.ExplorationToolRecorder(sink), _FakeContext("read_file", {})))
        is True
    )
    assert len(sink) == 1


def test_the_base_argument_is_read_from_both_shapes_the_context_allows() -> None:
    """``FunctionInvocationContext.arguments`` is ``BaseModel | Mapping[str, Any]`` (measured
    against the installed signature), so BOTH are read rather than one assumed. A tool that names
    no base yields ``""`` — a label invented for a base nobody named is the false attribution
    ``ToolCallRecorder`` refuses for unconfigured tools.
    """
    import asyncio
    from types import SimpleNamespace

    sink: list[ex.ToolCall] = []
    recorder = ex.ExplorationToolRecorder(sink)
    asyncio.run(
        _invoke(recorder, _FakeContext("read_bundle", SimpleNamespace(bundle_id="modelled")))
    )
    asyncio.run(_invoke(recorder, _FakeContext("list_bundles", None)))
    asyncio.run(_invoke(recorder, _FakeContext("read_bundle", {"bundle_id": 7})))

    assert [c.bundle_id for c in sink] == ["modelled", "", ""]


def test_the_record_leaves_the_run_in_the_artefact_beside_the_advisory_verdicts() -> None:
    """A field no artefact carries is a field nobody can read after the run — MAJOR-1's second
    half. ``trace_payload`` is the ONE rendering of a trace, so it is asserted there.

    Detach point: drop the ``tool_calls`` key from ``trace_payload`` → RED here and end-to-end.
    """
    trace = ex.ExplorationTrace()
    trace.tool_calls.append(ex.ToolCall(name="read_bundle", bundle_id="bygg-energi-mikro", path=""))

    payload = ex.trace_payload(trace, stop=None, completed=True, mandate=None)

    # ``path`` joined the record in S7a-3 pkt. 3 and is ``""`` here because ``read_bundle`` takes
    # none; its own gate is ``tests/test_tool_call_path_loadbearing.py``.
    # P19 DEL C added HOW the level was asked for (``filter``/``offset``/``limit``); the
    # identity of the call is still these three, so the arm compares on them.
    identity = [{k: c[k] for k in ("name", "bundle_id", "path")} for c in payload["tool_calls"]]
    assert identity == [
        {"name": "read_bundle", "bundle_id": "bygg-energi-mikro", "path": ""}
    ], "the artefact must say which bases were opened, or a paid run leaves no record that any were"
    assert json.dumps(payload), "the payload must stay plain data — the RAW layer is MAF-free"


# ---------------------------------------------------------------------------------------------
# MAJOR-1 (b) — THE SCRIPT CAN NOW CALL A TOOL, AND THE END-TO-END GATE OVER BOTH HALVES
# ---------------------------------------------------------------------------------------------

_BUNDLE_ID = _BUNDLE_DIR.name


def _ledger(*, satisfied: bool, speaker: str = "navigator") -> str:
    return json.dumps(
        {
            "is_request_satisfied": {"reason": "r", "answer": satisfied},
            "is_in_loop": {"reason": "r", "answer": False},
            "is_progress_being_made": {"reason": "r", "answer": True},
            "next_speaker": {"reason": "r", "answer": speaker},
            "instruction_or_question": {"reason": "r", "answer": "open a base"},
        }
    )


#: The manager answered as a STEP LIST — one entry per stage, in the order the orchestrator asks
#: (facts, plan, ledger, ledger, final answer). This is the same measured reason
#: ``simulation._exploration_manager_reply`` keys on the PROMPT rather than the project id: the
#: manager is asked five DIFFERENT questions and a single constant answers all of them with the
#: first round's ledger — which is why the pre-existing constant-reply scenario concludes after one
#: round having given no participant a turn (measured here). Text steps, no calls: the manager
#: carries no tools.
_MANAGER_STAGES = [
    "FACTS: the base is anchored.",
    "PLAN: - let the navigator open a base",
    _ledger(satisfied=False),
    _ledger(satisfied=True),
    "FINAL: the exploration is done.",
]
_HYPOTHESIS = "HYPOTHESIS: " + json.dumps(
    {"label": "x", "rationale": "y", "requirement": None, "why_none": "scripted"}
)


def _artefact(tmp_path: Path, run_id: str) -> dict[str, Any]:
    path = tmp_path / "outbox" / f"{run_id}-exploration.json"
    assert path.exists(), "the exploration artefact must be written even when the run failed"
    return json.loads(path.read_text(encoding="utf-8"))


def _explore_argv(tmp_path: Path, replies_path: str, run_id: str) -> list[str]:
    return [
        *_base_argv(tmp_path, replies_path),
        "--outbox-dir",
        str(tmp_path / "outbox"),
        "--run-id",
        run_id,
    ]


def test_a_scripted_navigator_can_open_a_base_and_the_artefact_says_which(tmp_path) -> None:
    """THE GOAL, end to end and over BOTH halves: a script of
    ``list_bundles -> read_bundle -> text`` makes the navigator actually call the tools, and the
    artefact an operator reads afterwards names which base was opened.

    RED without (b): a step LIST is refused at the door, so the run never starts. RED without (a):
    the tools run and ``tool_calls`` is empty — indistinguishable from the vacuous rehearsal this
    whole section exists to make visible.
    """
    replies = _replies_file(
        tmp_path,
        {
            "proposer": _PROPOSER_REPLY,
            "checker": "VERDICT: APPROVE",
            "manager": list(_MANAGER_STAGES),
            "navigator": [
                {"call": "list_bundles"},
                {"call": "read_bundle", "args": {"bundle_id": _BUNDLE_ID}},
                "NAVIGATOR: read the index.",
            ],
            "hypothesiser": _HYPOTHESIS,
        },
    )

    rc = run.main(_explore_argv(tmp_path, replies, "scripted-tools"))
    assert rc in (0, 1), rc

    calls = _artefact(tmp_path, "scripted-tools")["tool_calls"]
    assert [c["name"] for c in calls][:2] == ["list_bundles", "read_bundle"], (
        "an offline rehearsal must be able to prove the navigator OPENS something — that is the "
        f"whole of the free half of the measurement ladder; got {calls!r}"
    )
    assert calls[1]["bundle_id"] == _BUNDLE_ID, (
        "the name alone answers 'a tool ran'; WHICH base it was asked for is what tells an "
        "operator the navigator went where it was pointed"
    )


def test_a_constant_string_script_records_no_tool_calls(tmp_path) -> None:
    """The CONTROL, and the measurement MAJOR-1 reported: one constant string per role cannot
    emit a ``function_call``, so the rehearsal calls nothing. The recorder must SAY so rather than
    invent entries — a record that filled itself in would be worth less than none.
    """
    replies = _replies_file(
        tmp_path,
        {
            "proposer": _PROPOSER_REPLY,
            "checker": "VERDICT: APPROVE",
            "manager": list(_MANAGER_STAGES),
            "navigator": "NAVIGATOR: read the index.",
            "hypothesiser": _HYPOTHESIS,
        },
    )

    rc = run.main(_explore_argv(tmp_path, replies, "scripted-constant"))
    assert rc in (0, 1), rc

    assert _artefact(tmp_path, "scripted-constant")["tool_calls"] == [], (
        "a constant-reply rehearsal opens nothing, and the artefact must be able to say that"
    )


def _refuse(tmp_path, capsys, step: Any, run_id: str) -> str:
    replies = _replies_file(
        tmp_path,
        {
            "proposer": _PROPOSER_REPLY,
            "checker": "VERDICT: APPROVE",
            "manager": list(_MANAGER_STAGES),
            "navigator": [step],
            "hypothesiser": _HYPOTHESIS,
        },
    )
    rc = run.main(_explore_argv(tmp_path, replies, run_id))
    err = capsys.readouterr().err
    assert rc == 1, err
    assert "run refused" in err and "navigator" in err, err
    return err


def test_a_step_that_names_no_tool_is_refused_by_name(tmp_path, capsys) -> None:
    """Validation, NEVER repair (the ``write_concept_file`` rule). A step that forgot to say WHAT
    to call is refused at the door — letting it through would reach ``step["call"]`` inside the
    client and crash mid-run with a raw ``KeyError``, which is precisely the MAJOR-2 failure the
    first half of this file exists to have closed.

    **This arm was FALSIFIED first** (this repo's vacuous-gate class, twelfth time). It originally
    used ``{"invoke": "list_bundles"}``, and stayed GREEN with the shape check detached — the
    UNKNOWN-KEY branch below caught it instead, so the refusal under test had no witness at all.
    The two branches are now exercised by two arms with two mutations.
    """
    _refuse(tmp_path, capsys, {"args": {"bundle_id": _BUNDLE_ID}}, "scripted-no-call")


def test_a_step_naming_an_unknown_key_is_refused_by_name(tmp_path, capsys) -> None:
    """The whitelist half (the hosted surface's ``_ALLOWED_FIELDS`` precedent, one layer down): a
    key a step does not carry is refused BY NAME, never dropped in silence. An operator who wrote
    ``result`` expecting the rehearsal to assert something would otherwise be told nothing.
    """
    err = _refuse(
        tmp_path, capsys, {"call": "list_bundles", "result": "[]"}, "scripted-unknown-key"
    )
    assert "result" in err, err


def test_a_step_list_is_accepted_for_a_debate_role(tmp_path) -> None:
    """REWRITTEN at S2c, not deleted: this arm used to pin the refusal of a step list for the
    debate's roles, on the ground that the proposer answers ``generate``'s own call rather than an
    agent loop that could invoke a tool between turns. Since S2c the proposer and the checker ARE
    agents holding the four navigator tools, so that ground is measurably false and the refusal
    would have kept the free half of the measurement ladder away from the very seam S2c builds —
    the identical vacuity MAJOR-1 closed for the navigator.

    What replaces it is the positive: a scripted proposer opens the base, and the debate's own
    artefact names WHICH document. RED if the list form is refused for a debate role again, and RED
    if the debate's ``ExplorationToolRecorder`` is detached (the run succeeds, the trace is empty).
    """
    replies = _replies_file(
        tmp_path,
        {
            "proposer": [
                {"call": "read_bundle", "args": {"bundle_id": "bygg-energi-mikro"}},
                _PROPOSER_REPLY,
            ],
            "checker": "VERDICT: APPROVE",
            "manager": list(_MANAGER_STAGES),
            "navigator": "NAVIGATOR: read the index.",
            "hypothesiser": _HYPOTHESIS,
        },
    )

    rc = run.main(_explore_argv(tmp_path, replies, "scripted-debate-list"))

    assert rc == 0
    payload = json.loads(
        (tmp_path / "outbox" / "scripted-debate-list-debate.json").read_text(encoding="utf-8")
    )
    first = payload["tool_calls"][0]
    assert {k: first[k] for k in ("name", "bundle_id", "path")} == {
        "name": "read_bundle",
        "bundle_id": "bygg-energi-mikro",
        "path": "",
    }, payload["tool_calls"]