open/portfolio-optimiser
1
0
Fork 0
Code Issues Releases 3 Activity

feat(p17b): ONE commission, SEVERAL bases -- reachable from the command line

Browse source 
``run_mandate_across_bundles`` has existed since session 58, reachable from FIVE
test files and from NO command line (measured: ``grep -n across-bundle run.py``
= 0 hits). ``--across-bundle <dir>``, repeated once per base, is that door.

The engine takes a CALLBACK rather than an outbox directory. Its own docstring
has always said N runs need N ``run_id``s and that minting them there would
default a key this repo requires a caller to supply -- so ``outbox_for`` is that
contract KEPT, not relaxed, and the operator-chosen ``<run-id>-<bundle_id>``
rule lives in ``main()`` where the decision was made. The order's alternative (a
caller running ``run_project`` itself over ``route_by_bundle``'s sub-mandates)
would be a second copy of the loop's id reconciliation, shared store, per-base
project resolution, collision accounting and both budget teeth.

``resolve_bundle_routing`` is ONE resolution shared by the engine and the
dry-run arm: a free trip answering with a different project id, or tolerating a
duplicate id the paid dispatch refuses, would rehearse a different run.

``{run-id}-multibase.json`` is written from a ``finally`` and every row is built
from the resolution plus disk, so the pass a cap cut short still leaves the
record. ``completed`` is a required field for ``ExplorationTrace.completed``'s
reason. ``stop_reason`` is read BACK from each base's own coverage artefact.

Load-bearing MEASURED (17 arms), four mutations all red against the WHOLE suite,
green control 1761/5 (from 1744/5, superset, 0 removed), golden byte-unchanged.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Kjell Tore Guttormsen 2026-09-15 04:24:48 +02:00
commit 5e4c497a84
5 changed files with 994 additions and 37 deletions
Download patch file Download diff file Expand all files Collapse all files

522
tests/test_across_bundles_cli_loadbearing.py Normal file
View file

@ -0,0 +1,522 @@
"""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 ``<run-id>-<bundle_id>`` 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.* ``<outbox>/<run-id>-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): ``<run-id>-<bundle_id>`` 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"],
],
)
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