"""P17b DEL 1 — ONE run, SEVERAL knowledge bases, from the CLI (order ``20260915T014020Z``). ``run_mandate_across_bundles`` has existed since økt 58 and was reachable from FIVE test files and from NO command line (measured 15.09: ``grep -n across-bundle run.py`` = 0 hits). The operator directive of 14.09 is that ``po`` must demonstrably work against THE BASES — plural — a given run says it will use, and a library function no operator can invoke is not that demonstration. **Three things this file pins, and each is a different claim.** *The minting rule.* N runs need N ``run_id``s — the engine's own docstring says so, and refuses to default a key this repo requires a caller to supply. The rule is ``-`` and it is OPERATOR-CHOSEN (14.09), not a preference argued here. The engine therefore takes an ``outbox_for(bundle_id) -> (outbox_dir, run_id)`` CALLBACK rather than an ``outbox_dir``: the caller supplies the ids, which is exactly the contract the engine documents, and the naming convention stays in ``main()`` where the operator's decision lives. *Why a callback rather than a caller-side loop.* The order offered both. A caller that ran ``run_project`` itself over ``route_by_bundle``'s sub-mandates would have to re-implement the id reconciliation, the shared ``VerdictStore``, the per-base ``project_id`` resolution, the collision accounting and both budget teeth — five rules that already have exactly one home. Two copies of a dispatch loop is kø-(p) with a much larger surface than the one-parameter alternative. *The summary file.* ``/-multibase.json`` answers the question no per-base artefact can: in which ORDER the bases were spent, which ``run_id`` each one got, what was never reached, which candidates two bases both described, and why each base stopped. Its ``stop_reason`` is READ BACK from each base's own ``{run_id}-coverage.json`` rather than recomputed, because that file is where P19 D2 put the fact and a second derivation of it would be free to disagree. """ from __future__ import annotations import json import shutil from pathlib import Path from typing import Any import pytest from portfolio_optimiser import explore, okf, run as run_module _EXAMPLES = Path(__file__).resolve().parents[1] / "shared" / "examples" _BYGG = _EXAMPLES / "bygg-energi-mikro" # BYGG-KONTOR-NORD _TUNNEL = _EXAMPLES / "tunnel-hauglia" # TUNNEL-HAUGLIA _PROPOSAL = json.dumps( { "measure": "Redusert omfang", "affected_items": [{"code": "ENERGI-TOTAL-EL", "quantity": 300000.0, "unit_cost": 1.0}], "claimed_saving_nok": 30000, } ) #: The CHECKER carries the tool-calling step list, never the proposer: a proposer script is #: consumed a SECOND time by the fresh client ``generate_via_llm`` builds, so a leading #: ``function_call`` there would answer the generation call too and burn an attempt on a parse #: failure (``_load_scripted_replies``' own stated honesty limit). The checker runs in the debate #: and nowhere else, so one entry there buys exactly one tool call per base. _REPLIES: dict[str, Any] = { "proposer": _PROPOSAL, "checker": [{"call": "list_bundles", "args": {}}, "VERDICT: APPROVE"], } def _mount(tmp_path: Path, *sources: Path) -> list[str]: out = [] for src in sources: dst = tmp_path / src.name shutil.copytree(src, dst) out.append(str(dst)) return out def _mandate_file( tmp_path: Path, *, rows: list[tuple[str, str]], name: str = "mandate.json" ) -> str: path = tmp_path / name path.write_text( json.dumps( { "objective": "kutt kostnad i begge basene", "success_criteria": "minst én tilnærming validerer", "allow_own_proposals": False, "approaches": [ { "id": approach_id, "label": f"Tilnærming {approach_id}", "affected_codes": ["ENERGI-TOTAL-EL"], "claimed_saving_nok": 30000.0, "bundle_id": bundle_id, } for approach_id, bundle_id in rows ], } ), encoding="utf-8", ) return str(path) def _replies_file(tmp_path: Path, payload: dict[str, Any] | None = None) -> str: path = tmp_path / "replies.json" path.write_text(json.dumps(payload if payload is not None else _REPLIES), encoding="utf-8") return str(path) def _argv( bases: list[str], *, mandate: str, run_id: str, outbox: Path, replies: str, extra: list[str] ) -> list[str]: argv = [] for base in bases: argv += ["--across-bundle", base] return argv + [ "--mandate", mandate, "--run-id", run_id, "--outbox-dir", str(outbox), "--scripted-replies", replies, "--max-rounds", "2", *extra, ] # --------------------------------------------------------------------------------------------- # (i) Two outboxes and one summary, each keyed by the operator-chosen minting rule. # --------------------------------------------------------------------------------------------- def test_each_base_writes_its_own_artefact_set_under_its_own_minted_run_id(tmp_path: Path) -> None: """(i): ``-`` per base, the full artefact set, and ONE summary beside them. The discriminator against a single shared ``run_id`` is not that files exist but that they are DISTINCT: with one key the second base would overwrite the first and the outbox would hold one base's answers under a name claiming to cover both. """ bases = _mount(tmp_path, _BYGG, _TUNNEL) outbox = tmp_path / "out" rc = run_module.main( _argv( bases, mandate=_mandate_file( tmp_path, rows=[("a", "bygg-energi-mikro"), ("b", "tunnel-hauglia")] ), run_id="X", outbox=outbox, replies=_replies_file(tmp_path), extra=[], ) ) assert rc == 0 for bundle_id, approach in (("bygg-energi-mikro", "a"), ("tunnel-hauglia", "b")): stem = f"X-{bundle_id}" assert (outbox / f"{stem}-{approach}-proposal.json").is_file() assert (outbox / f"{stem}-{approach}-outcome.json").is_file() assert (outbox / f"{stem}-debate.json").is_file() assert (outbox / f"{stem}-runconfig.json").is_file() assert (outbox / f"{stem}-coverage.json").is_file() summary = json.loads((outbox / "X-multibase.json").read_text(encoding="utf-8")) assert summary["run_id"] == "X" assert [row["bundle_id"] for row in summary["runs"]] == [ "bygg-energi-mikro", "tunnel-hauglia", ] assert [row["run_id"] for row in summary["runs"]] == [ "X-bygg-energi-mikro", "X-tunnel-hauglia", ] assert summary["unreached"] == [] assert summary["collisions"] == [] assert summary["stopped_early"] is False assert summary["budget_stop"] is None assert summary["completed"] is True # Read BACK from each base's own coverage file, never recomputed here. assert [row["stop_reason"] for row in summary["runs"]] == ["", ""] def test_the_summary_reports_the_stop_reason_each_base_recorded(tmp_path: Path) -> None: """(i), the half a clean pass cannot show: a base the round cap cut short says so. An unparseable proposer burns the round cap, the FIRST approach to hit it re-raises by design (``_evaluate_mandate`` only swallows mid-list), and the pass therefore ends rc 1 — which is exactly what round 3 measured on ``kontrakt-sorasen-2027-04``. The summary must still be there, must say ``rounds`` rather than the empty string that means "the run finished", and must say ``completed: false`` rather than leaving an empty ``unreached`` to be read as "nothing was left unreached". """ bases = _mount(tmp_path, _BYGG) outbox = tmp_path / "out" rc = run_module.main( [ "--across-bundle", bases[0], "--mandate", _mandate_file(tmp_path, rows=[("a", "bygg-energi-mikro")]), "--run-id", "X", "--outbox-dir", str(outbox), "--scripted-replies", _replies_file(tmp_path, {"proposer": "not json at all", "checker": "VERDICT: APPROVE"}), "--max-rounds", "1", ] ) assert rc == 1 summary = json.loads((outbox / "X-multibase.json").read_text(encoding="utf-8")) assert summary["runs"][0]["stop_reason"] == "rounds" assert summary["completed"] is False assert summary["runs"][0]["run_id"] == "X-bygg-energi-mikro" # --------------------------------------------------------------------------------------------- # (ii) + (iii) The two refusals the dispatch owns, both as rc 1 with a readable message. # --------------------------------------------------------------------------------------------- def test_an_approach_naming_a_base_this_run_does_not_configure_is_refused( tmp_path: Path, capsys: pytest.CaptureFixture[str] ) -> None: """(ii): ``MandateRoutingError`` reaches the operator as rc 1 and names the base.""" bases = _mount(tmp_path, _BYGG) rc = run_module.main( _argv( bases, mandate=_mandate_file(tmp_path, rows=[("a", "en-base-som-ikke-er-med")]), run_id="X", outbox=tmp_path / "out", replies=_replies_file(tmp_path), extra=[], ) ) assert rc == 1 err = capsys.readouterr().err assert "en-base-som-ikke-er-med" in err assert "bygg-energi-mikro" in err def test_two_bases_declaring_one_id_are_refused( tmp_path: Path, capsys: pytest.CaptureFixture[str] ) -> None: """(iii): the id is how a mandate NAMES a base, so two bases answering to it is a refusal. Both copies DECLARE the same id on their root ``index.md`` — mount-derived ids could never collide under two directory names, so a test built on directory names alone would prove nothing about the branch that fires here. """ bases = _mount(tmp_path, _BYGG) second = tmp_path / "andre-base" shutil.copytree(_BYGG, second) bases.append(str(second)) for base in bases: index = Path(base) / "index.md" text = index.read_text(encoding="utf-8") index.write_text(text.replace("---\n", "---\nbundle_id: delt-id\n", 1), encoding="utf-8") rc = run_module.main( _argv( bases, mandate=_mandate_file(tmp_path, rows=[("a", "delt-id")]), run_id="X", outbox=tmp_path / "out", replies=_replies_file(tmp_path), extra=[], ) ) assert rc == 1 assert "delt-id" in capsys.readouterr().err # --------------------------------------------------------------------------------------------- # (iv) ONE store across the pass — the event, not a substring both branches share. # --------------------------------------------------------------------------------------------- def test_the_second_base_sees_the_verdict_the_first_base_minted( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """(iv): identity of the store AND the verdict ids present when each base started. ``VerdictStore`` is a pydantic model with VALUE equality, so three distinct EMPTY stores are all ``==`` (økt 58's measured vacuity). Identity is the claim, and it is paired with the EVENT: the set of verdict ids already in the store when base 2 began must contain the id base 1 minted, which a fresh-store-per-base implementation cannot produce. """ bases = _mount(tmp_path, _BYGG, _TUNNEL) seen: list[tuple[int, frozenset[str]]] = [] real = run_module.run_project async def recording(*args: Any, **kwargs: Any) -> Any: store = kwargs["store"] seen.append((id(store), frozenset(v.id for v in store.verdicts))) return await real(*args, **kwargs) monkeypatch.setattr(run_module, "run_project", recording) rc = run_module.main( _argv( bases, mandate=_mandate_file( tmp_path, rows=[("a", "bygg-energi-mikro"), ("b", "tunnel-hauglia")] ), run_id="X", outbox=tmp_path / "out", replies=_replies_file(tmp_path), extra=["--decision", "approved", "--rationale", "expert reviewed (scripted)"], ) ) assert rc == 0 assert len(seen) == 2 assert seen[0][0] == seen[1][0], "each base was handed a FRESH store, so nothing carries over" assert seen[0][1] == frozenset(), "base 1 started with a store that already held something" assert seen[1][1], "base 2 started with an EMPTY store — base 1's verdict never reached it" # --------------------------------------------------------------------------------------------- # (v) Two bases, two ``opened`` sinks — the requirement gate reads the base it is asked about. # --------------------------------------------------------------------------------------------- def test_the_requirement_gate_reads_the_second_bases_own_opened_list(tmp_path: Path) -> None: """(v), driven through the tool bodies DIRECTLY. A ``ScriptedChatClient`` returns TEXT and never emits a ``function_call`` for a constant reply, so no scripted run reaches a tool body — økt 56's measured vacuity, and the reason the repo's own answer is to call each tool by name. Base 1 opens a document; declaring THAT path against base 2 must refuse, because base 2's own sink never saw it. """ first, second = _mount(tmp_path, _BYGG, _TUNNEL) opened_1: list[explore.ToolCall] = [] opened_2: list[explore.ToolCall] = [] reqs_1: list[explore.DeclaredRequirement] = [] reqs_2: list[explore.DeclaredRequirement] = [] tools_1 = { t.name: t for t in explore.navigator_tools([first], opened=opened_1, requirements=reqs_1) } tools_2 = { t.name: t for t in explore.navigator_tools([second], opened=opened_2, requirements=reqs_2) } bundle_1 = okf.reconcile_bundle_id(first).id bundle_2 = okf.reconcile_bundle_id(second).id doc_1 = okf.navigate_bundle(first).context_files[0].name doc_2 = okf.navigate_bundle(second).context_files[0].name # ``opened`` is filled by the RECORDER middleware, never by the tool bodies (S2c/MAJOR-1: # what a run read is a property of the invocation, not of the tool's return value), so the # arm records the call the way a run does and then asks the gate about it. tools_1["read_file"].func(bundle_1, doc_1) opened_1.append(explore.ToolCall(name="read_file", bundle_id=bundle_1, path=doc_1)) assert not opened_2, "the two bases shared one opened sink" refusal = tools_2["declare_requirement"].func(bundle_2, doc_1, "Krav 1") assert refusal["refusal"] == "RequirementNotRead", refusal assert not reqs_2, "base 2 recorded a requirement it never read" tools_2["read_file"].func(bundle_2, doc_2) opened_2.append(explore.ToolCall(name="read_file", bundle_id=bundle_2, path=doc_2)) accepted = tools_2["declare_requirement"].func(bundle_2, doc_2, "Krav 1") assert accepted.get("declared") is True, accepted assert [r.path for r in reqs_2] == [doc_2] assert not reqs_1, "the two bases shared one requirements sink" def test_each_bases_debate_artefact_lists_only_its_own_tool_calls(tmp_path: Path) -> None: """(v), the half the direct call cannot see: ``run_project`` builds the sinks PER BASE. The arm above proves the gate reads the sink it is given; this proves the dispatch gives each base its own. With one shared sink base 2's artefact would carry base 1's call as well, so the discriminator is the COUNT and not the presence. """ bases = _mount(tmp_path, _BYGG, _TUNNEL) outbox = tmp_path / "out" rc = run_module.main( _argv( bases, mandate=_mandate_file( tmp_path, rows=[("a", "bygg-energi-mikro"), ("b", "tunnel-hauglia")] ), run_id="X", outbox=outbox, replies=_replies_file(tmp_path), extra=[], ) ) assert rc == 0 counts = [] for bundle_id in ("bygg-energi-mikro", "tunnel-hauglia"): payload = json.loads((outbox / f"X-{bundle_id}-debate.json").read_text(encoding="utf-8")) counts.append([call["name"] for call in payload["tool_calls"]]) assert counts == [["list_bundles"], ["list_bundles"]], counts # --------------------------------------------------------------------------------------------- # The three partition rows. Each refusal is paired with an rc-0 control on an argv that is # otherwise ACCEPTED — without it, rc 1 could come from anywhere in the argv. # --------------------------------------------------------------------------------------------- def test_the_flag_is_refused_in_portfolio_mode_by_name( tmp_path: Path, capsys: pytest.CaptureFixture[str] ) -> None: assert ( run_module.main( ["--portfolio", "--across-bundle", str(tmp_path), "--goals", str(tmp_path / "g.json")] ) == 1 ) assert "--across-bundle" in capsys.readouterr().err def test_the_flag_is_refused_in_report_mode_by_name( tmp_path: Path, capsys: pytest.CaptureFixture[str] ) -> None: ledger = tmp_path / "ledger.json" ledger.write_text("[]", encoding="utf-8") assert run_module.main(["--report", "--ledger", str(ledger)]) == 0, "control: report mode works" capsys.readouterr() assert ( run_module.main(["--report", "--ledger", str(ledger), "--across-bundle", str(tmp_path)]) == 1 ) assert "mode-exclusive" in capsys.readouterr().err def test_a_dry_run_drills_every_base_and_stops_before_the_first_call( tmp_path: Path, capsys: pytest.CaptureFixture[str] ) -> None: """The third row is a WIRING, not a refusal: the drill must cover every configured base. Silently dropping the flag here is the F4 class — the dry run would fall through to the single-project branch, which has no ``PROJECT_ID`` in this argv at all. """ bases = _mount(tmp_path, _BYGG, _TUNNEL) rc = run_module.main( [ *sum([["--across-bundle", b] for b in bases], []), "--mandate", _mandate_file(tmp_path, rows=[("a", "bygg-energi-mikro"), ("b", "tunnel-hauglia")]), "--run-id", "X", "--outbox-dir", str(tmp_path / "out"), "--live-dry-run", ] ) out = capsys.readouterr().out assert rc == 0 assert out.count("LIVE-DRY-RUN OK") == 2 assert "bygg-energi-mikro" in out and "tunnel-hauglia" in out # The per-base notices are the discriminator against ONE drill that merely names two bases: # ``bygg-energi-mikro`` ships no ``cost-baseline.json`` and ``tunnel-hauglia`` does, so exactly # one unanchored notice and exactly one grounding offer must appear — and a drill of only the # first, or only the second, produces a different count either way. assert out.count("Grounding offer") == 1 assert out.count("Cost baseline: NONE") == 1 @pytest.mark.parametrize( "missing, token", [("--mandate", "--mandate"), ("--run-id", "--run-id"), ("--outbox-dir", "--outbox-dir")], ) def test_the_flag_requires_the_three_things_a_multi_base_pass_cannot_invent( tmp_path: Path, capsys: pytest.CaptureFixture[str], missing: str, token: str ) -> None: base = _mount(tmp_path, _BYGG)[0] full = { "--mandate": _mandate_file(tmp_path, rows=[("a", "bygg-energi-mikro")]), "--run-id": "X", "--outbox-dir": str(tmp_path / "out"), } argv = ["--across-bundle", base] for name, value in full.items(): if name != missing: argv += [name, value] assert run_module.main(argv) == 1 assert token in capsys.readouterr().err @pytest.mark.parametrize( "extra", [ ["--bundle-dir", "somewhere"], ["--explore", "en prompt"], ["--prepass-payload", "payload.json"], ["--proposals-from-mandate"], ["--docs-dir", "somewhere"], ["--mcp-config", "servers.json"], ["--semantic-retrieval"], ["--checkpoint-dir", "ckpt"], ["--review-inbox", "inbox"], ], ) def test_the_flag_refuses_every_single_base_mode_by_name( tmp_path: Path, capsys: pytest.CaptureFixture[str], extra: list[str] ) -> None: base = _mount(tmp_path, _BYGG)[0] argv = [ "--across-bundle", base, "--mandate", _mandate_file(tmp_path, rows=[("a", "bygg-energi-mikro")]), "--run-id", "X", "--outbox-dir", str(tmp_path / "out"), *extra, ] assert run_module.main(argv) == 1 err = capsys.readouterr().err assert "--across-bundle" in err and extra[0] in err def test_a_positional_project_id_is_refused_rather_than_dropped( tmp_path: Path, capsys: pytest.CaptureFixture[str] ) -> None: """Each base's project comes from THAT base's own IR projection, never from argv. A caller-supplied constant could be right for one base out of N, and accepting it silently would look honoured while the dispatch used something else — the silent-drop class. """ base = _mount(tmp_path, _BYGG)[0] argv = [ "EN-PROSJEKT-ID", "--across-bundle", base, "--mandate", _mandate_file(tmp_path, rows=[("a", "bygg-energi-mikro")]), "--run-id", "X", "--outbox-dir", str(tmp_path / "out"), ] assert run_module.main(argv) == 1 err = capsys.readouterr().err assert "PROJECT_ID" in err and "--across-bundle" in err def test_the_anchoring_requirement_is_carried_to_every_base_not_dropped( tmp_path: Path, capsys: pytest.CaptureFixture[str] ) -> None: """``--require-cost-baseline`` is HONOURED here, and refuses before the first model call. P19 F1's named remedy, and P17b's own paid run is what makes it matter: a requirement number was accepted as a cost code on a base with no schedule. Accepting the flag and dropping it would be the F4 class on the one guarantee an operator asked for by name — so the arm asserts the REFUSAL, and its control asserts that the same argv without the flag runs to rc 0. ``bygg-energi-mikro`` ships no ``cost-baseline.json`` and ``tunnel-hauglia`` does, so the refusal must come from the first base rather than from "neither is anchored". """ bases = _mount(tmp_path, _BYGG, _TUNNEL) mandate = _mandate_file(tmp_path, rows=[("a", "bygg-energi-mikro"), ("b", "tunnel-hauglia")]) replies = _replies_file(tmp_path) outbox = tmp_path / "out" assert ( run_module.main( _argv( bases, mandate=mandate, run_id="X", outbox=outbox, replies=replies, extra=["--require-cost-baseline"], ) ) == 1 ) err = capsys.readouterr().err assert "bygg-energi-mikro" in err or "cost baseline" in err.lower() assert not (outbox / "X-bygg-energi-mikro-a-proposal.json").is_file(), ( "the refusal fired AFTER the run had already been paid for" ) assert ( run_module.main( _argv( bases, mandate=mandate, run_id="Y", outbox=outbox, replies=replies, extra=[], ) ) == 0 ), "control: the same argv without the flag runs"