"""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" # --------------------------------------------------------------------------------------------- # 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"}) 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_refused_for_a_debate_role(tmp_path, capsys) -> None: """The list form is the EXPLORATION's, and is refused elsewhere by name rather than accepted and quietly ignored. The debate's proposer is driven by ``generate``'s own call, not by an agent loop that would invoke a tool between turns, so a script of calls there would describe a rehearsal that cannot happen. """ replies = _replies_file( tmp_path, { "proposer": [{"call": "retrieve_cost_docs", "args": {"query": "x"}}], "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 == 1 err = capsys.readouterr().err assert "run refused" in err and "proposer" in err, err