"""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": "", "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"}), }, ) 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")) payload = ex.trace_payload(trace, stop=None, completed=True) assert payload["tool_calls"] == [{"name": "read_bundle", "bundle_id": "bygg-energi-mikro"}], ( "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"