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

feat(explore): U4+U13 del 3 - multi-base er en PARTISJON, ikke en videre signatur (ORDRE 20260825T080753Z) [skip-docs]

Browse source 
Ordrens premiss («endrer run_project sin signatur») er FALSIFISERT foer bygging,
og ordren ba selv om nettopp den sjekken. run_project kan ikke ta mer enn en
bundle_dir: paa bundle-stien avleder den FIRE enkeltverdier fra DEN basen -
prosjektet (_project_from_bundle fail-faster naar basens egen project_id ikke er
den forespurte), validatorens stage-0-baseline, agentenes lesekontekst og
ExpeL-noekkelen - og returnerer ETT stemplet RunResult. En andre katalog paa den
signaturen ville tvunget et stille velg-en for alle fire. Planens egen setning
sier det samme lest naert: «pipelinen kjoeres per bundle som i dag
(run_portfolio-formen)» = N kall, ikke ett kall med N.

Levert form, tre soemmer:
- mandate.Approach.bundle_id (default "" - hvert mandat skrevet foer i dag er
  fortsatt gyldig og dispatchbart uendret)
- mandate.route_by_bundle - ren partisjon, fail-fast paa et mandat som ikke kan
  utfoeres som skrevet (load_mandate-regelen). En base som ingen approach navngir
  kjoeres ikke; med NOEYAKTIG en base absorberer den alt uten navn, som ikke er en
  gjetning men det eneste mulige svaret.
- run.run_mandate_across_bundles - dispatchen. INGEN project_id-parameter:
  hver base sitt prosjekt leses fra DEN basens egen IR-projeksjon, altsaa den
  verdien _project_from_bundle allerede fail-faster mot. En delt VerdictStore
  traades paa tvers (kryss-base-laering, run_portfolio-formen), og med
  portfolio_meter gjelder de to S3.4-tennene som HAR mening her: oppstartsnekt
  (BudgetRefused) og aldri-startet + budget_stop + not_evaluated-rader.

INGEN eksisterende kaller endrer signatur - CLI, hosting og simulation sender
fortsatt en base hver, og kan fortsatt gjoere det.

explore(): hver myntet approach baerer bundle_id; markoeren kan navngi basen;
en umerket markoer med flere baser NEKTES (HypothesisParseError), med en base
resolveres den. Froe-approaches VALIDERES men skrives ALDRI om (§ C.6 doer 1 er
en bevaringsregel) - og valideringen skjer FOER foerste modellkall, samme
oekt-57-hoist-grunn: ved unntaket alene ser en nekt etter forbruket identisk ut.
explore()-docstringens gamle aerlighets-grense («venter paa at run_project tar
mer enn en bundle_dir») er RETTET - den ville vaert usann fra i dag.

Load-bearing MAALT, 12 mutasjoner alle roede mot HELE suiten, groenn kontroll
1020/5 (fra 999/5 + 21 nye tester); golden demo-transcript.stdout BYTE-UENDRET
(ea8c534773acdbe41ae68f2c55724d69aaf8be4f): detach myntet bundle_id (2) ·
stille gjennomfall ved >1 base (1) · ukjent id resolvert etter rekkefoelge (1) ·
froe-sjekk etter forbruket (2) · ruteren gjetter foerste base (1) · uroutbar
approach droppet (1) · dispatchen kollapser til en base (5) · project_id fra
foerste base (2) · detach aldri-startet-tannen (1) · unreached urapportert (1) ·
fersk store per base (1) · detach oppstartsnekten (2).

EN MUTASJON FALSIFISERTE TESTEN FOERST (repoets vakuoes-gate-klasse, aattende
gang): store-testen sammenlignet med ==, og VerdictStore er en pydantic-modell
med VERDI-likhet - tre ulike tomme stores er alle like, saa «fersk store per
base» lot hele suiten staa groenn. Delt INSTANS er paastanden, saa testen
asserterer naa paa `is`.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01L3YHobQC3WzYVoxSgZus4d
This commit is contained in:
Kjell Tore Guttormsen 2026-08-25 13:35:39 +02:00
commit 18af86e422
6 changed files with 1051 additions and 17 deletions
Download patch file Download diff file Expand all files Collapse all files

2
tests/test_explore_callsites_loadbearing.py
View file

@ -540,7 +540,7 @@ def test_a_seeded_mandate_still_leads_the_shaped_one() -> None:
A refusal that pointed at a door which did not open would be worse than no message at all.
"""
seed = Approach(id="expert-1", label="expert's own", description="the domain expert asked")
minted = explore._mint_approaches((seed,), [(_LABEL, "shaped in the loop")])
minted = explore._mint_approaches((seed,), [(_LABEL, "shaped in the loop", "")])
assert [a.id for a in minted] == ["expert-1", "hypothesis-1"]
assert isinstance(Mandate(objective="o", approaches=minted, allow_own_proposals=True), Mandate)

8
tests/test_explore_loadbearing.py
View file

@ -947,9 +947,9 @@ def test_a_marked_line_that_will_not_parse_is_a_hard_error() -> None:
than following the tolerant RAW-inbox rule, which belongs to folders anyone may drop files in.
"""
with pytest.raises(explore.HypothesisParseError):
explore._parse_hypotheses([f"{explore.HYPOTHESIS_MARKER} not json at all"])
explore._parse_hypotheses([f"{explore.HYPOTHESIS_MARKER} not json at all"], ())
with pytest.raises(explore.HypothesisParseError):
explore._parse_hypotheses([f'{explore.HYPOTHESIS_MARKER} {{"label": "no rationale"}}'])
explore._parse_hypotheses([f'{explore.HYPOTHESIS_MARKER} {{"label": "no rationale"}}'], ())
def test_unmarked_prose_is_not_a_failure() -> None:
@ -958,7 +958,9 @@ def test_unmarked_prose_is_not_a_failure() -> None:
Without it, T24 would pass on an implementation that refused every hypothesiser turn that was
not a hypothesis, which would make the loop unusable and the strictness meaningless.
"""
assert explore._parse_hypotheses(["I looked at the index and nothing stands out yet."]) == []
assert (
explore._parse_hypotheses(["I looked at the index and nothing stands out yet."], ()) == []
)
def test_a_review_nobody_can_answer_is_refused_before_the_first_model_call() -> None:

669
tests/test_multibase_loadbearing.py Normal file
View file

@ -0,0 +1,669 @@
"""U4 + U13, del 3 — MULTI-BASE (``Approach.bundle_id``, plan § C.7).
**The premise this file corrects, measured before anything was built.** § C.7 and økt 56's own
honesty limit read as though the deliverable were "``run_project`` accepts more than one
``bundle_dir``". It cannot, and the refusal is structural rather than stylistic: on the bundle
path ``run_project`` derives FOUR single-valued things from THE bundle — the project
(``_project_from_bundle``, which fail-fasts when ``validator-input.json``'s ``project_id`` is not
the requested one), the validator's stage-0 cost baseline (S4.0's whole point being that the gate
is anchored to THAT project's real cost lines), the agents' read context, and the ExpeL query key
— and it returns ONE ``RunResult`` with ONE ``ProvenanceStamp``. A second ``bundle_dir`` would
force a silent pick-one for all four, which is the guessed-shape class this repo refuses.
§ C.7's own sentence says the same thing once read closely: *"pipelinen kjøres per bundle som i dag
(``run_portfolio``-formen)"* — N calls, one per base, not one call taking N. So the delivered shape
is: each approach RECORDS its base, a pure router PARTITIONS the mandate by base, and a thin
dispatch runs the existing ``run_project`` once per base. **No existing caller's signature changes**
— CLI, hosting and simulation each still pass one base, and each still may.
Three seams, each with its own detach signature:
* ``mandate.Approach.bundle_id`` — the field, defaulting to ``""`` so every mandate written before
this session stays valid;
* ``mandate.route_by_bundle`` — the partition, fail-fast on a commission that cannot be executed as
written (``load_mandate``'s rule: a run must never proceed on a silently degraded commission);
* ``run.run_mandate_across_bundles`` — the dispatch, whose ``project_id`` per base comes from THAT
base's own IR projection and never from a caller-supplied constant.
"""
from __future__ import annotations
import json
import shutil
from collections.abc import Callable
from pathlib import Path
from typing import Any
import pytest
from agent_framework import BaseChatClient
from portfolio_optimiser import explore, run as run_module
from portfolio_optimiser.budget import BudgetRefused, PortfolioBudget, PortfolioMeter
from portfolio_optimiser.explore import ExplorationContract, ExplorationError, HypothesisParseError
from portfolio_optimiser.mandate import (
Approach,
Mandate,
MandateRoutingError,
route_by_bundle,
)
from portfolio_optimiser.run import RunResult, run_mandate_across_bundles
from portfolio_optimiser.simulation import ScriptedChatClient
from portfolio_optimiser.verdicts import VerdictStore
_EXAMPLES = Path(__file__).resolve().parents[1] / "shared" / "examples"
#: Three bases with three DISTINCT project ids — which is what makes "the project comes from the
#: base, not from the caller" a claim a test can actually falsify.
_BYGG = _EXAMPLES / "bygg-energi-mikro" # BYGG-KONTOR-NORD
_TUNNEL = _EXAMPLES / "tunnel-hauglia" # TUNNEL-HAUGLIA
_VEGLYS = _EXAMPLES / "veglys-fv-soer" # VEGLYS-FV-SOER
_VERDICT_INPUT = {"decision": "approved", "rationale": "expert reviewed (sim)"}
# ---------------------------------------------------------------------------------------------
# 1. The field. mandate.py stays pydantic + stdlib (D7-portable, test_okf_is_maf_free).
# ---------------------------------------------------------------------------------------------
def test_an_approach_records_which_knowledge_base_it_belongs_to() -> None:
"""T1: ``Approach`` carries ``bundle_id``, and it DEFAULTS to the empty string.
The default is what keeps every mandate JSON written before this session valid, and every
existing ``Approach(...)`` constructor call unaffected — the same reason ``RunResult.coverage``
defaults. Empty means "no base named", which is a legitimate statement when the run has only
one base to name.
"""
assert Approach(id="a", label="A").bundle_id == ""
assert Approach(id="a", label="A", bundle_id="tunnel-hauglia").bundle_id == "tunnel-hauglia"
# ---------------------------------------------------------------------------------------------
# 2. The router. Pure, framework-neutral, and fail-fast on a commission it cannot execute.
# ---------------------------------------------------------------------------------------------
def _mandate(*approaches: Approach, own: bool = True) -> Mandate:
return Mandate(objective="find savings", approaches=approaches, allow_own_proposals=own)
def test_the_router_partitions_the_mandate_one_sub_mandate_per_named_base() -> None:
"""T2: each base gets a sub-mandate carrying ONLY its own approaches, in ``bundle_ids`` order.
Order is taken from the configured bases rather than from first appearance among the
approaches, so the dispatch's spend order is a property of how the run was configured and not
of how a model happened to sequence its hypotheses.
"""
a = Approach(id="a", label="A", bundle_id="tunnel-hauglia")
b = Approach(id="b", label="B", bundle_id="veglys-fv-soer")
c = Approach(id="c", label="C", bundle_id="tunnel-hauglia")
routed = route_by_bundle(_mandate(a, b, c), ("veglys-fv-soer", "tunnel-hauglia"))
assert [bundle_id for bundle_id, _ in routed] == ["veglys-fv-soer", "tunnel-hauglia"]
assert [ap.id for ap in routed[0][1].approaches] == ["b"]
assert [ap.id for ap in routed[1][1].approaches] == ["a", "c"]
# The commission's own fields travel with every partition: each sub-run is still working on the
# same objective and under the same "and/or your own" permission.
assert routed[0][1].objective == "find savings"
assert routed[0][1].allow_own_proposals is True
def test_a_single_base_absorbs_every_unassigned_approach() -> None:
"""T3: with exactly ONE base configured, an approach naming none routes to it.
Not a guess — with one configured base there is no other value the field could take, and
``_bundle_index`` already guarantees the id is unique. This is what keeps a single-base
mandate (every mandate that exists today) dispatchable unchanged.
"""
routed = route_by_bundle(_mandate(Approach(id="a", label="A")), ("tunnel-hauglia",))
assert [bundle_id for bundle_id, _ in routed] == ["tunnel-hauglia"]
assert [ap.id for ap in routed[0][1].approaches] == ["a"]
def test_an_unassigned_approach_among_several_bases_is_refused_not_guessed() -> None:
"""T4: the discriminator for T3 — with TWO bases, an unnamed approach REFUSES the dispatch.
Silently sending it to the first base would evaluate a commissioned direction against a
project nobody asked about and report it as done. Refusing is ``load_mandate``'s rule applied
one layer on: a run must never proceed on a silently degraded commission, because the coverage
report would then describe work nobody ordered.
"""
with pytest.raises(MandateRoutingError) as exc:
route_by_bundle(_mandate(Approach(id="a", label="A")), ("tunnel-hauglia", "veglys-fv-soer"))
# The REPR, never the bare letter: "a" is a substring of almost any English sentence, so an
# assertion on it would hold against a refusal raised for an entirely different reason.
assert "'a'" in str(exc.value)
def test_an_approach_naming_an_unconfigured_base_is_refused_by_name() -> None:
"""T5: an approach whose ``bundle_id`` matches no configured base refuses, naming both.
Resolving it by position instead would be the S3.2 key-collision class: an approach evaluated
against a base it does not belong to, with nothing in the report saying so.
"""
approach = Approach(id="a", label="A", bundle_id="does-not-exist")
with pytest.raises(MandateRoutingError) as exc:
route_by_bundle(_mandate(approach), ("tunnel-hauglia", "veglys-fv-soer"))
message = str(exc.value)
assert "does-not-exist" in message
assert "tunnel-hauglia" in message
def test_the_routing_refusal_is_a_value_error() -> None:
"""T6: ``MandateRoutingError`` subclasses ``ValueError`` — a TYPE claim, not a taxonomy note.
økt 57 measured the cost of getting this wrong the other way: ``ExplorationError`` is a
``RuntimeError`` and therefore fell outside ``run.main``'s ``(ValueError, FileNotFoundError,
ValidationError)`` refusal tuple and outside hosting's 400 arm, so a caller's configuration
mistake would have left as a traceback and a 500. A routing refusal is exactly that class of
caller mistake, so it is born inside both nets rather than retrofitted into them.
"""
assert issubclass(MandateRoutingError, ValueError)
def test_routing_against_no_base_at_all_is_refused() -> None:
"""T7: zero configured bases refuses rather than returning an empty plan.
An empty plan reads as "there was nothing to do", which is indistinguishable from a mandate
that was fully evaluated against nothing — the omitted-row silence ``ApproachOutcome``'s
``not_evaluated`` status exists to remove.
"""
with pytest.raises(MandateRoutingError):
route_by_bundle(_mandate(Approach(id="a", label="A")), ())
# ---------------------------------------------------------------------------------------------
# 3. Assignment: what the exploration puts in the field, and what it refuses to put there.
# ---------------------------------------------------------------------------------------------
_CONTRACT = ExplorationContract(
max_rounds=6,
max_tokens=100_000,
max_stall_count=2,
max_reset_count=1,
max_plan_revisions=0,
enable_plan_review=False,
)
_PROMPT = "Find the cheapest saving available."
def _ledger_json(*, satisfied: bool, speaker: str) -> 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": "Shape one hypothesis."},
}
)
def _manager_script(ledgers: list[str]) -> Callable[[str, str], str]:
"""Route a manager prompt blob to its scripted reply. The ORDER of these branches is
load-bearing and was measured in økt 56 (§ F, A6): the selector receives the CONCATENATION of
every message in the call, so a later-stage prompt still carries the earlier stage's text."""
def _select(blob: str, _role: str) -> str:
if "provide the final answer" in blob:
return "FINAL: exploration done."
if "pure JSON format" in blob:
return ledgers.pop(0) if ledgers else _ledger_json(satisfied=True, speaker="navigator")
if "went wrong on this last run" in blob:
return "PLAN-UPDATE: revised plan."
if "rewrite the following fact sheet" in blob:
return "FACTS-UPDATE: revised facts."
if "bullet-point plan" in blob:
return "PLAN: - ask the hypothesiser"
if "pre-survey" in blob:
return "FACTS: the bundle is anchored."
return "{}"
return _select
def _factory(
*, ledgers: list[str], hypothesiser: list[str], sink: list[str] | None = None
) -> Callable[[str], BaseChatClient]:
"""One fresh ``ScriptedChatClient`` per role. ``sink`` records every prompt that reached a
client, which is how "refused BEFORE the first model call" becomes measurable rather than
merely asserted at the exit code."""
def factory(role: str) -> BaseChatClient:
if role == explore.MANAGER_ROLE:
return ScriptedChatClient(reply_selector=_manager_script(ledgers), role=role, sink=sink)
if role == explore.HYPOTHESISER_ROLE:
replies = list(hypothesiser)
def _hyp(_blob: str, _role: str) -> str:
return replies.pop(0) if replies else "nothing further."
return ScriptedChatClient(reply_selector=_hyp, role=role, sink=sink)
return ScriptedChatClient("NAVIGATOR: index read.", role=role, sink=sink)
return factory
def _hypothesis_line(label: str, rationale: str, bundle_id: str | None = None) -> str:
payload: dict[str, Any] = {"label": label, "rationale": rationale}
if bundle_id is not None:
payload["bundle_id"] = bundle_id
return f"{explore.HYPOTHESIS_MARKER} " + json.dumps(payload)
def _two_round_ledgers() -> list[str]:
return [
_ledger_json(satisfied=False, speaker=explore.HYPOTHESISER_ROLE),
_ledger_json(satisfied=True, speaker=explore.HYPOTHESISER_ROLE),
]
@pytest.mark.asyncio
async def test_a_marked_hypothesis_may_name_its_base_and_the_mandate_carries_it() -> None:
"""T8: the marker's ``bundle_id`` reaches ``Approach.bundle_id``.
Asserted DIRECTLY on the returned mandate rather than through anything downstream: with a
single base configured a router that dropped the field would route identically, so a
behavioural assertion here could not tell an implementation that assigns from one that does
not. Two bases are configured for the same reason.
"""
result = await explore.explore(
_PROMPT,
contract=_CONTRACT,
bundle_dirs=(str(_TUNNEL), str(_VEGLYS)),
client_factory=_factory(
ledgers=_two_round_ledgers(),
hypothesiser=[_hypothesis_line("LED retrofit", "old fixtures", "veglys-fv-soer")],
),
)
assert [a.bundle_id for a in result.mandate.approaches] == ["veglys-fv-soer"]
@pytest.mark.asyncio
async def test_with_one_base_a_marker_that_names_none_still_yields_an_assigned_approach() -> None:
"""T9: one base configured, marker silent — the minted approach carries that base's id.
Again asserted on the FIELD. This is the arm that would go quietly green under an
implementation that never assigns anything, which is precisely why T8 configures two bases and
why neither test reads the field through the router.
"""
result = await explore.explore(
_PROMPT,
contract=_CONTRACT,
bundle_dirs=(str(_TUNNEL),),
client_factory=_factory(
ledgers=_two_round_ledgers(),
hypothesiser=[_hypothesis_line("LED retrofit", "old fixtures")],
),
)
assert [a.bundle_id for a in result.mandate.approaches] == ["tunnel-hauglia"]
@pytest.mark.asyncio
async def test_with_several_bases_a_marker_that_names_none_is_refused() -> None:
"""T10: the discriminator for T9 — two bases and a silent marker is a hard error.
A marked line is a claim the loop committed to; one that cannot be routed is a claim it could
not finish making. That is the ``write_concept_file`` rule (validation, never repair) and NOT
the tolerant RAW-inbox rule, because this is the product of the run.
"""
with pytest.raises(HypothesisParseError):
await explore.explore(
_PROMPT,
contract=_CONTRACT,
bundle_dirs=(str(_TUNNEL), str(_VEGLYS)),
client_factory=_factory(
ledgers=_two_round_ledgers(),
hypothesiser=[_hypothesis_line("LED retrofit", "old fixtures")],
),
)
@pytest.mark.asyncio
async def test_a_marker_naming_an_unconfigured_base_is_refused() -> None:
"""T11: a ``bundle_id`` matching no configured base refuses — ``_resolve_bundle``'s own rule.
The navigator's tools already refuse an unknown base id rather than resolving it by order; a
hypothesis that names one must not be treated more leniently than a read of one.
"""
with pytest.raises(ExplorationError):
await explore.explore(
_PROMPT,
contract=_CONTRACT,
bundle_dirs=(str(_TUNNEL), str(_VEGLYS)),
client_factory=_factory(
ledgers=_two_round_ledgers(),
hypothesiser=[_hypothesis_line("LED", "old fixtures", "no-such-base")],
),
)
@pytest.mark.asyncio
async def test_a_seed_naming_an_unknown_base_is_refused_before_the_first_model_call() -> None:
"""T12: an unroutable SEED refuses with ZERO model calls made.
The assertion is on the sink, not on the exception, and that is the økt-57 outbox-hoist
precedent: at the exception alone, a refusal AFTER the exploration has spent its whole budget
looks identical to one before. The expert's mandate cannot be dispatched either way — what is
at stake is whether they pay to find out.
"""
sink: list[str] = []
with pytest.raises(ExplorationError):
await explore.explore(
_PROMPT,
contract=_CONTRACT,
bundle_dirs=(str(_TUNNEL), str(_VEGLYS)),
seed_approaches=(Approach(id="s1", label="Seed", bundle_id="no-such-base"),),
client_factory=_factory(ledgers=_two_round_ledgers(), hypothesiser=[], sink=sink),
)
assert sink == []
@pytest.mark.asyncio
async def test_an_unassigned_seed_with_several_bases_is_refused_before_the_first_call() -> None:
"""T13: a seed naming no base, with more than one configured, refuses — also before spending.
Same defect as T10 one door earlier (§ C.6 door 1 rather than the loop's own findings). The
expert configured several bases; which one their hypothesis belongs to is a thing only they
know, and inventing it would put their name on a direction they did not commission.
"""
sink: list[str] = []
with pytest.raises(ExplorationError):
await explore.explore(
_PROMPT,
contract=_CONTRACT,
bundle_dirs=(str(_TUNNEL), str(_VEGLYS)),
seed_approaches=(Approach(id="s1", label="Seed"),),
client_factory=_factory(ledgers=_two_round_ledgers(), hypothesiser=[], sink=sink),
)
assert sink == []
@pytest.mark.asyncio
async def test_a_seed_is_never_rewritten_only_validated() -> None:
"""T14: with one base, an unassigned seed comes back EXACTLY as it was written.
§ C.6 door 1 is a preservation rule: the seed is the expert's own words, and the mandate hands
``description`` to the proposer verbatim. Filling in ``bundle_id`` on their behalf would be
repair of an input, which this repo refuses even when the repair is obviously right — the
single-base default belongs to the ROUTER, at consumption, where it is unambiguous by
construction.
"""
seed = Approach(id="s1", label="Night setback", description="the expert's own words")
result = await explore.explore(
_PROMPT,
contract=_CONTRACT,
bundle_dirs=(str(_TUNNEL),),
seed_approaches=(seed,),
client_factory=_factory(ledgers=_two_round_ledgers(), hypothesiser=[]),
)
assert result.mandate.approaches[0] == seed
assert result.mandate.approaches[0].bundle_id == ""
# ---------------------------------------------------------------------------------------------
# 4. The dispatch. One run_project per base — the existing function, composed, never widened.
# ---------------------------------------------------------------------------------------------
def _recorder(
calls: list[dict[str, Any]], *, meter: PortfolioMeter | None = None, charge: int = 0
) -> Callable[..., Any]:
"""A stand-in for ``run_project`` that records exactly the kwargs it was called with.
Deliberately paired with ``test_the_dispatch_composes_with_the_real_run_project`` below: a
recorder that swallows ``**kwargs`` proves the dispatch's ROUTING but would happily accept an
argument ``run_project`` does not take — the Fase 4e defect, one layer up.
``charge`` credits the global ledger the way a real run does (its ``TokenMeter`` is bound to the
same ``PortfolioMeter``). Without it the remainder never moves, and an admission check made once
per base could not be told apart from one made once per pass.
"""
async def _fake(project_id: str, profile: Any = "local", **kwargs: Any) -> Any:
calls.append({"project_id": project_id, "profile": profile, **kwargs})
if meter is not None and charge:
meter.record(charge)
return _stub_result()
return _fake
def _stub_result() -> Any:
class _Stub:
coverage: tuple[Any, ...] = ()
provenance = None
return _Stub()
@pytest.mark.asyncio
async def test_the_dispatch_runs_one_pipeline_per_base_with_that_bases_approaches(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""T15: two bases -> two ``run_project`` calls, each bound to ITS base and ITS approaches.
This is § C.7's "pipelinen kjøres per bundle" made falsifiable. A dispatch that collapsed the
partition would still return a result — one run, one outcome — and nothing else in the suite
would notice.
"""
calls: list[dict[str, Any]] = []
monkeypatch.setattr(run_module, "run_project", _recorder(calls))
mandate = _mandate(
Approach(id="a", label="A", bundle_id="tunnel-hauglia"),
Approach(id="b", label="B", bundle_id="veglys-fv-soer"),
)
await run_mandate_across_bundles(
mandate,
(str(_TUNNEL), str(_VEGLYS)),
verdict_input=_VERDICT_INPUT,
store=VerdictStore(verdicts=[]),
)
assert len(calls) == 2
assert [c["bundle_dir"] for c in calls] == [str(_TUNNEL), str(_VEGLYS)]
assert [[ap.id for ap in c["mandate"].approaches] for c in calls] == [["a"], ["b"]]
@pytest.mark.asyncio
async def test_each_bases_project_id_comes_from_that_base_not_from_the_caller(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""T16: the ``project_id`` per call is read from THAT base's own IR projection.
The dispatch takes no ``project_id`` argument at all, and that is the design point rather than
an omission: ``_project_from_bundle`` already fail-fasts when a bundle's ``project_id`` is not
the requested one, so a caller-supplied constant could only ever be right for one of N bases.
Turning that existing fail-fast into the routing key removes the guess entirely.
"""
calls: list[dict[str, Any]] = []
monkeypatch.setattr(run_module, "run_project", _recorder(calls))
mandate = _mandate(
Approach(id="a", label="A", bundle_id="tunnel-hauglia"),
Approach(id="b", label="B", bundle_id="veglys-fv-soer"),
)
await run_mandate_across_bundles(
mandate,
(str(_TUNNEL), str(_VEGLYS)),
verdict_input=_VERDICT_INPUT,
store=VerdictStore(verdicts=[]),
)
assert [c["project_id"] for c in calls] == ["TUNNEL-HAUGLIA", "VEGLYS-FV-SOER"]
@pytest.mark.asyncio
async def test_one_base_is_dispatched_exactly_as_a_single_run(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""T17: the control — a single base makes exactly ONE call, carrying the whole mandate.
Without this arm T15 could pass on an implementation that fanned out over configured bases
regardless of what the mandate said, spending a run on every base a caller merely made
available.
"""
calls: list[dict[str, Any]] = []
monkeypatch.setattr(run_module, "run_project", _recorder(calls))
await run_mandate_across_bundles(
_mandate(Approach(id="a", label="A"), Approach(id="b", label="B")),
(str(_TUNNEL),),
verdict_input=_VERDICT_INPUT,
store=VerdictStore(verdicts=[]),
)
assert len(calls) == 1
assert [ap.id for ap in calls[0]["mandate"].approaches] == ["a", "b"]
@pytest.mark.asyncio
async def test_a_base_that_cannot_be_funded_is_never_started_and_its_approaches_are_reported(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""T18: the S3.4 admission tooth — an unfundable base is NEVER STARTED, and it SAYS SO.
Two halves, and both are load-bearing. Never-started is the point: a base that is merely
aborted mid-run has already cost real model calls. And reporting is the ``not_evaluated``
rule — an approach the pass never reached must appear as unreached, because an omitted row is
indistinguishable from an approach nobody commissioned.
"""
calls: list[dict[str, Any]] = []
meter = PortfolioMeter(PortfolioBudget(max_total_tokens=1_000, max_tokens_per_run=500))
monkeypatch.setattr(run_module, "run_project", _recorder(calls, meter=meter, charge=600))
result = await run_mandate_across_bundles(
_mandate(
Approach(id="a", label="A", bundle_id="tunnel-hauglia"),
Approach(id="b", label="B", bundle_id="veglys-fv-soer"),
),
(str(_TUNNEL), str(_VEGLYS)),
verdict_input=_VERDICT_INPUT,
store=VerdictStore(verdicts=[]),
portfolio_meter=meter,
)
# Base 1 is funded (1000 left, 500 required) and spends 600; base 2 then has 400 left against a
# 500 reserve. The SECOND base is the one that must never start.
assert [c["bundle_dir"] for c in calls] == [str(_TUNNEL)]
assert result.stopped_early is True
assert result.budget_stop is not None
assert {row.id for row in result.unreached} == {"b"}
assert all(row.status == "not_evaluated" for row in result.unreached)
@pytest.mark.asyncio
async def test_a_pass_that_can_fund_nothing_at_all_is_refused_at_startup() -> None:
"""T19: a global remainder below one run's reserve raises ``BudgetRefused`` before anything
loads — ``run_portfolio``'s startup refusal, same primitive, same reason: a pass that has room
for zero runs is a caller error, not a result.
"""
meter = PortfolioMeter(
PortfolioBudget(max_total_tokens=1_000, max_tokens_per_run=500), spent=1_000
)
with pytest.raises(BudgetRefused):
await run_mandate_across_bundles(
_mandate(Approach(id="a", label="A")),
(str(_TUNNEL),),
verdict_input=_VERDICT_INPUT,
store=VerdictStore(verdicts=[]),
portfolio_meter=meter,
)
def _reply_for(code: str, quantity: float, unit_cost: float, claimed: int) -> str:
return json.dumps(
{
"measure": "Redusert omfang",
"affected_items": [{"code": code, "quantity": quantity, "unit_cost": unit_cost}],
"claimed_saving_nok": claimed,
}
)
@pytest.mark.asyncio
async def test_the_dispatch_composes_with_the_real_run_project(tmp_path: Path) -> None:
"""T20: every kwarg the dispatch passes is a REAL ``run_project`` parameter.
The Fase 4e proof, one layer up: every routing test above uses a stand-in that swallows
``**kwargs``, so the dispatch could name an argument ``run_project`` does not take — or pass
one twice — and not a single one of them would notice, while a live call raised ``TypeError``.
Driven against two real bases end to end, offline, through the scripted client seam.
"""
bases = []
for src in (_BYGG, _TUNNEL):
dst = tmp_path / src.name
shutil.copytree(src, dst)
bases.append(str(dst))
def factory(role: str) -> BaseChatClient:
return ScriptedChatClient(
reply_selector=lambda _blob, _role: _reply_for("ENERGI-TOTAL-EL", 300000, 1.0, 30_000),
role=role,
)
result = await run_mandate_across_bundles(
_mandate(
Approach(id="a", label="A", bundle_id="bygg-energi-mikro"),
Approach(id="b", label="B", bundle_id="tunnel-hauglia"),
),
tuple(bases),
"local",
verdict_input=_VERDICT_INPUT,
store=VerdictStore(verdicts=[]),
client_factory=factory,
max_rounds=1,
)
assert [r.bundle_id for r in result.runs] == ["bygg-energi-mikro", "tunnel-hauglia"]
assert [r.project_id for r in result.runs] == ["BYGG-KONTOR-NORD", "TUNNEL-HAUGLIA"]
assert all(isinstance(r.result, RunResult) for r in result.runs)
# Each run answered for ITS OWN commissioned approach, and for nobody else's.
assert [{row.id for row in r.result.coverage} for r in result.runs] == [
{"a", "own-proposal"},
{"b", "own-proposal"},
]
@pytest.mark.asyncio
async def test_one_store_is_threaded_across_every_base(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""T21: the SAME ``VerdictStore`` instance reaches every base's run.
The cross-base learning claim, and the same one ``run_portfolio`` makes across projects: a
verdict captured while evaluating base k must be able to reach base k+1's hypothesis. A fresh
store per base would leave the loop looking wired while carrying nothing between the runs.
**Asserted on IDENTITY, and that is a correction the mutation forced** (this repo's vacuous-gate
class, eighth occurrence). The first version compared with ``==``, and ``VerdictStore`` is a
pydantic model with VALUE equality — so a mutation handing every base its own
``VerdictStore(verdicts=[])`` left the whole suite green: three distinct empty stores are all
equal to one another. Sharing an instance is the actual claim, so ``is`` is the actual test.
"""
calls: list[dict[str, Any]] = []
monkeypatch.setattr(run_module, "run_project", _recorder(calls))
store = VerdictStore(verdicts=[])
await run_mandate_across_bundles(
_mandate(
Approach(id="a", label="A", bundle_id="tunnel-hauglia"),
Approach(id="b", label="B", bundle_id="veglys-fv-soer"),
),
(str(_TUNNEL), str(_VEGLYS)),
verdict_input=_VERDICT_INPUT,
store=store,
)
assert len(calls) == 2
assert all(c["store"] is store for c in calls)