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

feat(p21): the PROJECT carries the price, so a run against a road normal can be anchored

Browse source 
Four paid stress rounds ran entirely UN-ANCHORED, all of them, because the one file
loader reads cost-baseline.json out of the BUNDLE and no vegnormal ships one: N100,
N200, N500 and R761 are knowledge, and knowledge carries requirements, never amounts.
The validator's stage 0 -- the one stage that tells an invented cost line from a line
this project actually buys -- was skipped in every single run, so "validated" could not
mean what it says. P20 G1/G2 measured real R761 process numbers (12.11 three times on
Soraasen, 1.1.1 on Lindaas) validating with amounts nobody had anywhere.

--cost-baseline FILE is PM decision (e), taken over the three alternatives P20 wrote
down. A LOADED object, never a path (prepass_payload's rule): the CLI owns the file and
loads it ONCE, so the notice, the stamp and every base of an --across-bundle pass all
descend from one read. ONE parse, two doors -- load_cost_baseline delegates to
load_cost_baseline_file -- while safe_resolve stays on the bundle door alone, because a
project's own schedule is legitimately outside every base. No tolerant twin: this path
exists only because an operator NAMED a file.

DEL B: five anchored context sets, a1-a3 with their line and a4 with none, so stage 0 is
what catches the falsification arm. THE ORDER'S OWN ARM (h) WAS FELLED BY MEASUREMENT:
"no baseline code is a requirement number the base declares" is measured 0 of 4 on the
project-coded sets and 5 of 5 on kontrakt-sorasen -- which is what R761 Prosesskoden IS,
a bill of quantities priced BY process code. The complement keeps both, and the order's
own mutation still bites.

DEL B3: the judge reports anchored (off the run's own stamp), priced per row, and WHICH
falsifier caught the falsification arm.

Load-bearing MEASURED, five mutations all red against the WHOLE suite, green control
1850/5 (from 1809/5, superset, 0 removed), golden byte-unchanged:
A3(i) the flag is read but the baseline is unused (3 red) . A3(ii) only the first base
gets it (1) . A3(iii) report_forbidden drops it (1) . B2(i) a4 gets a line (1, arm (g)
alone) . B2(ii) a code swapped to 12.11 (2, arms (f) and (h)).

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Kjell Tore Guttormsen 2026-09-15 10:49:10 +02:00
commit 7b4f85d77c
20 changed files with 1259 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

167
tests/test_context_sets_loadbearing.py
View file

@ -46,6 +46,7 @@ import pytest
from pydantic import ValidationError
from portfolio_optimiser import okf
from portfolio_optimiser.ir import CostBaseline, CostBaselineLine
from portfolio_optimiser.mandate import load_mandate
from portfolio_optimiser.stress import read_bundle_declarations
@ -492,3 +493,169 @@ def test_the_fasit_titles_are_distinct_not_the_collapsed_sources_title(set_dir:
"okf.parse_frontmatter collapsed these titles onto the sources block again — the P15 fix "
"in okf._frontmatter_from_text has regressed"
)
# --------------------------------------------------------------------------------------------
# (f) + (g) + (h): the set is ANCHORED (P21 B2).
#
# The measured reason these exist. Four paid rounds ran entirely un-anchored, because the only
# file loader reads ``cost-baseline.json`` out of the BUNDLE and no road normal carries prices — a
# vegnormal is knowledge, the price belongs to the PROJECT. With ``--cost-baseline`` the project
# supplies its own schedule, so the validator's stage 0 judges again: (f) every answerable approach
# has a line to reconcile against, and (g) the falsification arm has NONE, so the code it proposes
# is refused as "not in the project's cost baseline" — by stage 0, the one stage that can tell an
# invented line from a real one, instead of by the weaker downstream gates.
#
# (f) and (g) are SEPARATE arms rather than one loop over all approaches, because they are opposite
# claims about opposite rows: a single arm asserting "exactly the non-refuse codes are present"
# would go red for either defect and name neither.
# --------------------------------------------------------------------------------------------
def _set_baseline(set_dir: Path) -> CostBaseline:
return okf.load_cost_baseline_file(str(set_dir / "cost-baseline.json"))
@pytest.mark.parametrize("set_dir", _SETS, ids=_SET_IDS)
def test_f_every_answerable_approach_has_a_cost_line(set_dir: Path) -> None:
"""Unconditional — the schedule is the PROJECT's and needs no knowledge base to read.
The total is asserted against the approach's own estimate as well as the code's presence:
``SavingsProposal`` refuses ``claimed_saving_nok > sum(affected_items.total)``, so a line that
exists but is smaller than the saving commissioned against it would make the approach
unbuildable — a set that looks anchored and cannot be run.
"""
baseline = _set_baseline(set_dir)
assert 4 <= len(baseline.items) <= 8, (
f"{set_dir.name}: {len(baseline.items)} cost lines — the order asks for 4-8"
)
assert baseline.project_id == set_dir.name, (
f"{set_dir.name}: the schedule names project {baseline.project_id!r}"
)
fasit = json.loads((set_dir / "fasit.json").read_text(encoding="utf-8"))
refused = {row["approach_id"] for row in fasit["must_refuse"]}
for approach in load_mandate(set_dir / "mandate.json").approaches:
if approach.id in refused:
continue
missing = [c for c in approach.affected_codes if c not in baseline.items]
assert not missing, (
f"{set_dir.name}: {approach.id} is answerable but {missing} carry no line in the "
f"project's schedule ({sorted(baseline.items)})"
)
total = sum(
baseline.items[c].quantity * baseline.items[c].unit_cost
for c in approach.affected_codes
)
assert approach.claimed_saving_nok is not None
assert total >= approach.claimed_saving_nok, (
f"{set_dir.name}: {approach.id} claims {approach.claimed_saving_nok:g} against lines "
f"totalling {total:g} — no proposal on it can satisfy claimed <= total"
)
@pytest.mark.parametrize("set_dir", _SETS, ids=_SET_IDS)
def test_g_the_falsification_arm_has_no_cost_line(set_dir: Path) -> None:
"""The ``must_refuse`` approach's code is ABSENT, so stage 0 is what catches it.
This is the half that makes the anchoring worth measuring rather than just present: rule U
already says the base carries no GROUND for that line, and P7's stage 0b says the identifier is
ungrounded in the delivered input — but neither of those is the stage that knows what this
project actually buys. Stage 0 is, and it can only speak when the schedule exists.
"""
baseline = _set_baseline(set_dir)
fasit = json.loads((set_dir / "fasit.json").read_text(encoding="utf-8"))
by_id = {a.id: a for a in load_mandate(set_dir / "mandate.json").approaches}
assert fasit["must_refuse"], f"{set_dir.name} declares no falsification arm"
for row in fasit["must_refuse"]:
approach = by_id[row["approach_id"]]
carried = [c for c in approach.affected_codes if c in baseline.items]
assert not carried, (
f"{set_dir.name}: the falsification arm {approach.id} carries {carried} in the "
"project's schedule, so stage 0 would ACCEPT the line it exists to refuse"
)
@pytest.mark.parametrize("set_dir", _SETS, ids=_SET_IDS)
def test_h_no_cost_line_smuggles_in_an_uncommissioned_requirement_number(set_dir: Path) -> None:
"""No line of the schedule is a reference number the base declares AND nobody commissions.
**The order words this arm as "no baseline code is a requirement number the base declares", and
that rule was FELLED BY MEASUREMENT before anything was built on it.** Measured 15.09 against
``okf.declared_reference_numbers`` over the four mounted bases: the four project-coded sets
carry 0 such codes, and ``kontrakt-sorasen-2027`` carries FIVE of five — ``12.1``, ``12.12``,
``22.1``, ``52.11``, ``51.1`` are real R761 ``prosessnr``. That is not an accident in the set;
it is what R761 Prosesskoden IS. A Norwegian road contract's bill of quantities is priced BY
process code, so the project's schedule and the corpus's vocabulary share an identifier
namespace by design — and the order's rule would have forced a rewrite of the ONE set P20's
decision (e) was chosen to preserve.
The COMPLEMENT keeps both: a schedule may price what the commission names, and may not
INTRODUCE a corpus identifier as a cost line nobody ordered. The order's own mutation still
bites — swapping a code for ``12.11`` (a declared ``prosessnr`` no approach commissions) goes
red here — while the five real process codes pass because an approach names each of them.
Needs the base (the vocabulary is the base's), so it SKIPS with the root named.
"""
baseline = _set_baseline(set_dir)
commissioned = {
code
for approach in load_mandate(set_dir / "mandate.json").approaches
for code in approach.affected_codes
}
declared: set[str] = set()
concepts = 0
for block in read_bundle_txt(set_dir / "bundle.txt"):
bundle = okf.navigate_bundle(str(_require_base(block)))
for f in bundle.context_files:
concepts += 1
declared |= set(okf.declared_reference_numbers(f))
assert concepts >= 100, (
f"{set_dir.name}: scanned {concepts} concepts — too few to be the base(s)"
)
assert declared, f"{set_dir.name}: the base(s) declare NO reference numbers — nothing to test"
smuggled = sorted(c for c in baseline.items if c in declared and c not in commissioned)
assert not smuggled, (
f"{set_dir.name}: cost line(s) {smuggled} are reference numbers the knowledge base "
"declares and no approach commissions — the schedule would be introducing the corpus's "
"own identifiers as prices nobody ordered"
)
def test_known_positive_f_a_missing_cost_line_is_caught(tmp_path: Path) -> None:
baseline = _set_baseline(_CONTEXT_ROOT / "gate-nordvik-2027")
assert "GATE-KRYSS-01" in baseline.items
stripped = CostBaseline(
project_id=baseline.project_id,
items={k: v for k, v in baseline.items.items() if k != "GATE-KRYSS-01"},
)
assert "GATE-KRYSS-01" not in stripped.items
def test_known_positive_g_a_line_for_the_falsification_arm_is_caught() -> None:
"""The order's mutation (i): give a4 a line, and (g)'s assertion must fail on this set."""
baseline = _set_baseline(_CONTEXT_ROOT / "gate-nordvik-2027")
priced = dict(baseline.items)
priced["GATE-GANG-ENHET"] = CostBaselineLine(quantity=6, unit_cost=50_000.0)
fasit = json.loads((_CONTEXT_ROOT / "gate-nordvik-2027" / "fasit.json").read_text("utf-8"))
by_id = {
a.id: a
for a in load_mandate(_CONTEXT_ROOT / "gate-nordvik-2027" / "mandate.json").approaches
}
for row in fasit["must_refuse"]:
carried = [c for c in by_id[row["approach_id"]].affected_codes if c in priced]
assert carried == ["GATE-GANG-ENHET"]
def test_known_positive_h_an_uncommissioned_requirement_number_is_caught() -> None:
"""The order's mutation (ii): swap a code for ``12.11``.
Driven against a KNOWN vocabulary rather than the mounted base, so this known-positive runs
unconditionally — a control that skipped with the base would leave the arm's discriminator
unproven on exactly the machines that cannot run the arm.
"""
declared = {"12.1", "12.11", "12.12"}
commissioned = {"12.1", "12.12"}
assert sorted(c for c in {"12.1", "12.12"} if c in declared and c not in commissioned) == []
assert sorted(c for c in {"12.1", "12.11"} if c in declared and c not in commissioned) == [
"12.11"
]

396
tests/test_cost_baseline_flag_loadbearing.py Normal file
View file

@ -0,0 +1,396 @@
"""P21 DEL A - the PROJECT carries the price, so a run against a road normal can be anchored.
**The measurement this closes.** Four paid stress rounds (P16/P18/P19/P17b/P20) ran ENTIRELY
un-anchored. The cause is one line: the only file loader reads ``cost-baseline.json`` out of the
BUNDLE directory (``okf.load_optional_cost_baseline``), and no vegnormal ships one — N100, N200,
N500 and R761 are knowledge, and knowledge carries requirements, never amounts. The validator's
stage 0 — the one stage that can tell an invented cost line from a line this project actually buys
— was therefore skipped in every single one, and ``validated`` could not mean what it says: P20
G1/G2 measured real R761 process numbers (``12.11`` three times on Søråsen, ``1.1.1`` on Lindås)
validating with amounts nobody had anywhere.
``--cost-baseline FILE`` is PM decision (e), taken over three alternatives P20 wrote down: (a)
refusing every requirement-shaped code un-anchored would make the one realistic context set
unmeasurable, (b) ``--require-cost-baseline`` as a default would leave no stress test at all, and
(c) K2's priced schedule is refused by MAJOR-4's own stated limit. (e) puts the price where it
belongs — with the project — and stage 0 judges again.
Arms: (a) the library seam anchors * (b) control: without it the same base is un-anchored *
(c) it is the SAME baseline the validator is handed, so stage 0 really judges * (d) two sources for
one baseline are refused at the library seam * (e) the free trip anchors too * (f) a missing file
and (g) a malformed one are refused with ``load_cost_baseline``'s own error classes * (h)(i)(j)(k)
four CLI refusals BY NAME, each with an rc-0 control on an argv that would otherwise be accepted *
(l) the CLI wiring, measured on the stamp * (m) the notice * (n) EVERY base of an ``--across-bundle``
pass gets the SAME schedule.
"""
from __future__ import annotations
import json
import shutil
from pathlib import Path
from typing import Any
import pytest
from pydantic import ValidationError
from portfolio_optimiser import okf, run
from portfolio_optimiser.ir import CostBaseline, CostBaselineLine
from portfolio_optimiser.simulation import ScriptedChatClient
from portfolio_optimiser.validator import Rejection, validate_proposal
_EXAMPLES = Path(__file__).resolve().parents[1] / "shared" / "examples"
#: A base that ships NO ``cost-baseline.json`` — the whole class this flag exists for.
_UNPRICED_SOURCE = _EXAMPLES / "bygg-energi-mikro"
_IR_PROJECTION = {
"project_id": "P-KNOWLEDGE",
"measure": "PLACEHOLDER - authored by this test to satisfy the bundle contract",
"affected_items": [{"code": "KNOW-1", "quantity": 10, "unit_cost": 100.0}],
"claimed_saving_nok": 500.0,
}
def _runnable(tmp_path: Path, *, name: str = "base") -> str:
root = tmp_path / name
shutil.copytree(_UNPRICED_SOURCE, root)
(root / "validator-input.json").write_text(json.dumps(_IR_PROJECTION), encoding="utf-8")
assert not (root / "cost-baseline.json").exists(), "the fixture must be UNPRICED"
return str(root)
def _schedule(tmp_path: Path, *, name: str = "cost-baseline.json", **codes: float) -> str:
"""The PROJECT's own price schedule, written OUTSIDE every knowledge base — which is the whole
point: ``safe_resolve`` guards the bundle door, and a project's schedule is legitimately not in
a bundle."""
path = tmp_path / name
path.write_text(
json.dumps(
{
"project_id": "P-KNOWLEDGE",
"items": {
code: {"quantity": 10.0, "unit_cost": unit} for code, unit in codes.items()
},
}
),
encoding="utf-8",
)
return str(path)
def _scripted(sink: list[str] | None = None) -> Any:
return lambda role: ScriptedChatClient(sink=sink, role=role, default_reply="ok")
# --- (a)/(b)/(c)/(d)/(e) the library seam ---------------------------------------------------------
async def test_a_project_schedule_anchors_a_base_that_ships_none(tmp_path: Path) -> None:
bundle_dir = _runnable(tmp_path)
baseline = okf.load_cost_baseline_file(_schedule(tmp_path, RIGG=1000.0))
report = await run.run_project(
"P-KNOWLEDGE",
"local",
docs_dir=bundle_dir,
bundle_dir=bundle_dir,
cost_baseline=baseline,
live_dry_run=True,
)
assert isinstance(report, run.DryRunReport)
assert report.cost_baseline_anchored is True
async def test_control_without_the_schedule_the_same_base_is_unanchored(tmp_path: Path) -> None:
"""The discriminator for the arm above: the SAME base, the flag removed. Without this, "it is
anchored" could be a property of the fixture rather than of the file."""
bundle_dir = _runnable(tmp_path)
report = await run.run_project(
"P-KNOWLEDGE", "local", docs_dir=bundle_dir, bundle_dir=bundle_dir, live_dry_run=True
)
assert isinstance(report, run.DryRunReport)
assert report.cost_baseline_anchored is False
def test_the_supplied_schedule_is_what_stage_0_judges_against() -> None:
"""The stamp says anchored; this says the anchoring DOES something.
An implementation that read the file, stamped ``True`` and handed the validator ``None`` would
pass every other arm here — the mutation the order names as (i). Stage 0 either refuses a code
the project does not buy or it does not, and that is the only thing worth having.
"""
baseline = CostBaseline(
project_id="P-KNOWLEDGE",
items={"RIGG": CostBaselineLine(quantity=10.0, unit_cost=1000.0)},
)
from portfolio_optimiser.ir import AffectedItem, SavingsProposal
invented = SavingsProposal(
project_id="P-KNOWLEDGE",
measure="m",
affected_items=[AffectedItem(code="INDEKS-01", quantity=10.0, unit_cost=1000.0)],
claimed_saving_nok=100.0,
)
outcome = validate_proposal(invented, baseline=baseline)
assert isinstance(outcome, Rejection)
assert "cost baseline" in outcome.reason
# The control: the SAME magnitudes on a code the project DOES buy are not refused by stage 0.
real = SavingsProposal(
project_id="P-KNOWLEDGE",
measure="m",
affected_items=[AffectedItem(code="RIGG", quantity=10.0, unit_cost=1000.0)],
claimed_saving_nok=100.0,
)
second = validate_proposal(real, baseline=baseline)
assert not (isinstance(second, Rejection) and "cost baseline" in second.reason)
async def test_two_sources_for_one_baseline_are_refused_at_the_library_seam(
tmp_path: Path,
) -> None:
"""(d) Checked in ``run_project`` and not only in the CLI: the library takes the same two
arguments, and a library caller must not reach a state the CLI refuses by name."""
bundle_dir = _runnable(tmp_path)
baseline = okf.load_cost_baseline_file(_schedule(tmp_path, RIGG=1000.0))
with pytest.raises(ValueError, match="two sources for one baseline"):
await run.run_project(
"P-KNOWLEDGE",
"local",
docs_dir=bundle_dir,
bundle_dir=bundle_dir,
cost_baseline=baseline,
derive_cost_baseline=True,
live_dry_run=True,
)
async def test_it_satisfies_the_anchoring_requirement(tmp_path: Path) -> None:
"""(e) ``--require-cost-baseline`` is the guarantee; this is one of the two ways to meet it."""
bundle_dir = _runnable(tmp_path)
baseline = okf.load_cost_baseline_file(_schedule(tmp_path, RIGG=1000.0))
report = await run.run_project(
"P-KNOWLEDGE",
"local",
docs_dir=bundle_dir,
bundle_dir=bundle_dir,
cost_baseline=baseline,
require_cost_baseline=True,
live_dry_run=True,
)
assert isinstance(report, run.DryRunReport)
assert report.cost_baseline_anchored is True
# --- (f)/(g) the loader's error classes -----------------------------------------------------------
def test_a_missing_schedule_is_refused(tmp_path: Path) -> None:
with pytest.raises(FileNotFoundError):
okf.load_cost_baseline_file(str(tmp_path / "nope.json"))
def test_a_malformed_schedule_is_refused(tmp_path: Path) -> None:
"""``load_cost_baseline``'s own classes, and there is no tolerant twin: this path exists only
because an operator NAMED a file, so degrading its absence would answer an explicit order with
a silently un-anchored run."""
bad = tmp_path / "bad.json"
bad.write_text('{"project_id": "P", "items": {"X": {"quantity": -1, "unit_cost": 0}}}', "utf-8")
with pytest.raises(ValidationError):
okf.load_cost_baseline_file(str(bad))
def test_the_bundle_loader_still_parses_through_the_same_seam(tmp_path: Path) -> None:
"""ONE parse, two doors (kø-(p)). What differs is the RESOLUTION: ``safe_resolve`` stays on the
bundle door alone, because a project's own schedule is legitimately outside every base."""
base = tmp_path / "b"
base.mkdir()
(base / "cost-baseline.json").write_text(
json.dumps({"project_id": "P", "items": {"X": {"quantity": 1, "unit_cost": 2}}}), "utf-8"
)
assert okf.load_cost_baseline(str(base)).items["X"].unit_cost == 2.0
# --- (h)/(i)/(j)/(k) four CLI refusals, each with an rc-0 control ----------------------------------
def test_cli_requires_a_knowledge_base(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
"""(h) On the road path the baseline IS ``Project.cost_items``, so a file there is a second
source for one fact with nothing to break the tie (``--require-cost-baseline``'s reason)."""
rc = run.main(["P1", "--docs-dir", "docs", "--cost-baseline", _schedule(tmp_path, RIGG=1000.0)])
assert rc == 1
assert "--cost-baseline" in capsys.readouterr().err
def test_cli_refuses_two_sources_by_name(
tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""(i) BY NAME so the operator hears WHICH two flags conflict, rather than a traceback."""
bundle_dir = _runnable(tmp_path)
schedule = _schedule(tmp_path, RIGG=1000.0)
assert (
run.main(
[
"P-KNOWLEDGE",
"--bundle-dir",
bundle_dir,
"--cost-baseline",
schedule,
"--live-dry-run",
]
)
== 0
), "the control argv must be ACCEPTED"
capsys.readouterr()
rc = run.main(
[
"P-KNOWLEDGE",
"--bundle-dir",
bundle_dir,
"--cost-baseline",
schedule,
"--derive-cost-baseline",
"--live-dry-run",
]
)
assert rc == 1
err = capsys.readouterr().err
assert "--cost-baseline" in err and "--derive-cost-baseline" in err
def test_cli_is_refused_in_portfolio_mode(
tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""(j) A portfolio pass keys on PROJECTS, each already anchored by its own ``cost_items``, so
ONE file could be right for at most one row out of N. BY NAME, its neighbours' reason."""
rc = run.main(["--portfolio", "--cost-baseline", _schedule(tmp_path, RIGG=1000.0)])
assert rc == 1
assert "--portfolio" in capsys.readouterr().err
def test_cli_is_refused_in_report_mode(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
"""(k) Report mode returns ABOVE every dispatch, so an omission from the allowlist is a SILENT
DROP — the file would be accepted, nothing anchored, exit 0 (the F4 gap)."""
ledger = tmp_path / "ledger.json"
ledger.write_text("[]", encoding="utf-8")
assert run.main(["--report", "--ledger", str(ledger)]) == 0, "the control argv must be ACCEPTED"
capsys.readouterr()
rc = run.main(
["--report", "--ledger", str(ledger), "--cost-baseline", _schedule(tmp_path, RIGG=1000.0)]
)
assert rc == 1
assert "mode-exclusive" in capsys.readouterr().err
# --- (l)/(m) the CLI wiring and the notice --------------------------------------------------------
def test_cli_wiring_anchors_the_dry_run_and_says_so(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
"""(l)+(m) The flag must REACH ``run_project``, and the operator must be able to see that it
did on the FREE trip. rc 0 alone proves neither, so the discriminators are the two lines: the
un-anchored warning is GONE and the positive line names the file and the count."""
monkeypatch.delenv("PORTFOLIO_MODEL_MAP", raising=False)
bundle_dir = _runnable(tmp_path)
argv = ["P-KNOWLEDGE", "--bundle-dir", bundle_dir, "--live-dry-run"]
assert run.main(argv) == 0
before = capsys.readouterr().out
assert "Cost baseline: NONE in the bundle" in before
assert "Cost baseline: 2 lines from" not in before
schedule = _schedule(tmp_path, RIGG=1000.0, ASFALT=250.0)
assert run.main([*argv, "--cost-baseline", schedule]) == 0
after = capsys.readouterr().out
assert "Cost baseline: NONE in the bundle" not in after
assert f"Cost baseline: 2 lines from {schedule}" in after
def test_the_notice_is_omitted_when_nobody_named_a_file() -> None:
"""Omission where it is unambiguous — there is exactly one way to supply a schedule, so
silence means nobody did (``cost_baseline_notice``'s rule, kept)."""
assert run.cost_baseline_source_notice(None, 0) is None
assert run.cost_baseline_source_notice("x.json", 1) == (
" Cost baseline: 1 line from x.json — the validator's stage 0 reconciles every proposed "
"cost line against this project's own schedule"
)
# --- (n) every base of a multi-base pass ----------------------------------------------------------
async def test_every_base_of_an_across_bundle_pass_gets_the_same_schedule(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""(n) ONE project has ONE price schedule, so it anchors EVERY base.
The order's mutation (ii) is "only the first base gets it", and the assert is therefore per
BASE: a recorder that stopped at the first call would pass on exactly that mutation — the
vacuous-gate class this repo keeps measuring. Both calls are recorded and both must carry the
SAME object, because two reads of one file is already one resolution too many (kø-(p)).
"""
from portfolio_optimiser.mandate import Approach, Mandate
first = _runnable(tmp_path, name="one")
second = _runnable(tmp_path, name="two")
baseline = okf.load_cost_baseline_file(_schedule(tmp_path, RIGG=1000.0))
calls: list[dict[str, Any]] = []
async def _recorder(project_id: str, profile: Any = "local", **kwargs: Any) -> Any:
calls.append({"project_id": project_id, **kwargs})
class _Stub:
coverage: tuple[Any, ...] = ()
provenance = None
return _Stub()
monkeypatch.setattr(run, "run_project", _recorder)
mandate = Mandate(
objective="o",
success_criteria="s",
approaches=[
Approach(
id="a1",
label="one",
affected_codes=["RIGG"],
claimed_saving_nok=1.0,
bundle_id="one",
),
Approach(
id="a2",
label="two",
affected_codes=["RIGG"],
claimed_saving_nok=1.0,
bundle_id="two",
),
],
)
await run.run_mandate_across_bundles(mandate, [first, second], "local", cost_baseline=baseline)
assert len(calls) == 2, f"the dispatch ran {len(calls)} base(s), not two"
assert [c["bundle_dir"] for c in calls] == [first, second]
assert [c.get("cost_baseline") for c in calls] == [baseline, baseline], (
"a base was dispatched without the project's own schedule"
)
# The control: without the flag, no base is handed one — so the arm above measures the flag
# rather than a default.
calls.clear()
await run.run_mandate_across_bundles(mandate, [first, second], "local")
assert [c.get("cost_baseline") for c in calls] == [None, None]

61
tests/test_prose_code_form_loadbearing.py
View file

@ -209,3 +209,64 @@ def test_the_classification_is_reported_and_is_the_gates_own() -> None:
verdict = validate_proposal(_proposal(code), grounding=Grounding(_OFFERING + (code,)))
refused = isinstance(verdict, Rejection) and "has no identifier form" in verdict.reason
assert refused == (kind == "prose"), (code, kind, verdict)
# --------------------------------------------------------------------------------------------
# P21 B3: ``rejection_stage`` — which falsifier wrote a reason. A REPORT, never a gate: nothing
# branches on it, so an unrecognised sentence costs a label rather than a verdict. It lives beside
# the sentences it keys on, so the classifier and the wordings cannot drift apart (kø-(p)).
# --------------------------------------------------------------------------------------------
def test_p21_rejection_stage_names_each_stage_from_its_own_sentence() -> None:
from portfolio_optimiser.validator import rejection_stage
assert (
rejection_stage("unknown cost code 'X': not in project p's cost baseline (3 known codes)")
== "stage0-baseline"
)
assert (
rejection_stage(
"quantity 5 for cost code 'X' is outside the 5.0% tolerance around the baseline "
"quantity 9"
)
== "stage0-baseline"
)
assert (
rejection_stage("ungrounded identifier 'X': it appears nowhere in the input (9 chars)")
== "stage0b-grounding"
)
assert rejection_stage("claimed saving 9 exceeds P90 feasible 4") == "stage4-p90"
assert (
rejection_stage("claimed saving 9 exceeds the nominal feasible 4 at the items' stated")
== "stage4b-nominal"
)
assert (
rejection_stage("claimed 9 exceeds the energy_efficiency method cap 4 (stricter)")
== "stage5-method-cap"
)
# The honest answer for a sentence this module did not write — a label, never a verdict.
assert rejection_stage("something else entirely") == "other"
def test_p21_rejection_stage_is_keyed_on_the_sentences_the_validator_emits() -> None:
"""The control: the markers are not a private paraphrase but the text stage 0 really writes.
Without it the classifier could key on wording nothing emits and every arm above would still
be green — the vacuous-gate class, on a reporter.
"""
from portfolio_optimiser.ir import AffectedItem, CostBaseline, CostBaselineLine, SavingsProposal
from portfolio_optimiser.validator import Rejection, rejection_stage, validate_proposal
baseline = CostBaseline(
project_id="proj", items={"REAL-1": CostBaselineLine(quantity=10.0, unit_cost=100.0)}
)
proposal = SavingsProposal(
project_id="proj",
measure="m",
affected_items=[AffectedItem(code="FAKE-1", quantity=10.0, unit_cost=100.0)],
claimed_saving_nok=100.0,
)
outcome = validate_proposal(proposal, baseline=baseline)
assert isinstance(outcome, Rejection)
assert rejection_stage(outcome.reason) == "stage0-baseline"

130
tests/test_stress_judge_loadbearing.py
View file

@ -82,9 +82,27 @@ def _minibase(root: Path) -> Path:
return base
def _context(root: Path, *, must_refuse: bool = True) -> Path:
def _context(
root: Path, *, must_refuse: bool = True, schedule: dict[str, float] | None = None
) -> Path:
ctx = root / "ctx"
(ctx / "docs").mkdir(parents=True)
if schedule is not None:
# P21 B3: the PROJECT's own price schedule, which is the file a run is handed with
# ``--cost-baseline`` and the one the judge measures ``priced`` against.
(ctx / "cost-baseline.json").write_text(
json.dumps(
{
"project_id": "proj",
"items": {
code: {"quantity": 1.0, "unit_cost": unit}
for code, unit in schedule.items()
},
},
indent=2,
),
encoding="utf-8",
)
(ctx / "bundle.txt").write_text("name: minibase\nbundle_id: minibase\n", encoding="utf-8")
approaches = [
{
@ -149,6 +167,8 @@ def _write_outbox(
citation_snippet: str = "Body of the good one.",
decision: str = "validated",
tool_calls: list[dict[str, str]] | None = None,
anchored: bool = False,
reason: str = "no",
) -> None:
outbox.mkdir(parents=True, exist_ok=True)
codes = ["CODE-1"] if codes is None else codes
@ -181,7 +201,7 @@ def _write_outbox(
"role": "proposer",
"validator_decision": decision,
"token_usage": 10,
"cost_baseline_anchored": False,
"cost_baseline_anchored": anchored,
"bundle_id_source": None,
"external_calls": [],
},
@ -196,7 +216,7 @@ def _write_outbox(
"run_id": run_id,
"approach_id": approach_id,
"outcome_type": "validated" if decision == "validated" else "rejected",
**({"reason": "no"} if decision != "validated" else {}),
**({"reason": reason} if decision != "validated" else {}),
"checker_verdict": None,
"verdict_id": "vid",
},
@ -216,7 +236,12 @@ def _opened(path: str) -> list[dict[str, str]]:
def _judge(tmp_path: Path, **kw: object) -> stress.ContextSetVerdict:
base = _minibase(tmp_path)
ctx = _context(tmp_path, must_refuse=bool(kw.pop("must_refuse", False)))
schedule = kw.pop("schedule", None)
ctx = _context(
tmp_path,
must_refuse=bool(kw.pop("must_refuse", False)),
schedule=schedule, # type: ignore[arg-type]
)
outbox = tmp_path / "out"
_write_outbox(outbox, "r1", approach_id="a1", **kw) # type: ignore[arg-type]
return stress.score_context_set(ctx, outbox, "r1", base)
@ -629,3 +654,100 @@ def test_the_cli_refuses_to_guess_which_base_a_multi_base_outbox_is_for(
assert stress.main([*argv, "--bundle", "nowhere"]) == 1
assert "nowhere" in capsys.readouterr().err
# --------------------------------------------------------------------------------------------
# P21 B3: the judge reports what the run was ANCHORED on, which codes the project actually PRICES,
# and WHICH falsifier caught the falsification arm.
#
# The measured reason. Rounds 1-4 all ran un-anchored — a vegnormal ships no ``cost-baseline.json``
# and the only file loader read one out of the bundle — so stage 0 never spoke and the ``a4`` arm
# fell, when it fell, on P7's grounding check. "It was refused" and "the stage that knows what this
# project buys refused it" are different facts, and only the second is what anchoring bought.
# --------------------------------------------------------------------------------------------
def test_p21_priced_is_true_when_the_project_schedule_carries_the_code(tmp_path: Path) -> None:
verdict = _judge(tmp_path, schedule={"CODE-1": 2000.0})
assert verdict.approaches[0].priced is True
def test_p21_priced_is_false_for_a_code_the_project_does_not_buy(tmp_path: Path) -> None:
"""The discriminator: the SAME schedule, a proposal on a code it does not carry."""
verdict = _judge(tmp_path, schedule={"CODE-1": 2000.0}, codes=["CODE-9"])
assert verdict.approaches[0].priced is False
# ... and the control, so the arm cannot be green by measuring nothing.
assert (
_judge(tmp_path / "b", schedule={"CODE-9": 2000.0}, codes=["CODE-9"]).approaches[0].priced
is True
)
def test_p21_priced_is_false_without_a_project_schedule(tmp_path: Path) -> None:
"""Every round before P21: no schedule, so nothing is priced — reported, never guessed."""
assert _judge(tmp_path).approaches[0].priced is False
def test_p21_anchored_follows_the_runs_own_stamp(tmp_path: Path) -> None:
"""Read off ``provenance.cost_baseline_anchored``, not re-derived from the set's files.
Both arms, because a field that is constant is not a measurement: an artefact stamped
un-anchored must report ``False`` EVEN WHEN the set ships a schedule — the judge says what the
run did, and a run that was never given the file is not anchored by the file existing.
"""
assert _judge(tmp_path, anchored=True, schedule={"CODE-1": 2000.0}).anchored is True
assert _judge(tmp_path / "b", anchored=False, schedule={"CODE-1": 2000.0}).anchored is False
def test_p21_the_falsification_arm_reports_which_stage_caught_it(tmp_path: Path) -> None:
"""``stage0-baseline`` is the answer anchoring buys; ``stage0b-grounding`` is what round 4 got.
Driven through the whole judge rather than through ``rejection_stage`` alone, because the
seam being gated is that the judge READS the arm's own outcome artefact — a classifier that
was never called would leave every arm reporting ``""`` and the arms below still green.
"""
base = _minibase(tmp_path)
ctx = _context(tmp_path, must_refuse=True, schedule={"CODE-1": 2000.0})
outbox = tmp_path / "out"
_write_outbox(outbox, "r1", approach_id="a1", tool_calls=_opened(_GOOD))
_write_outbox(
outbox,
"r1",
approach_id="a4",
codes=["CODE-4"],
decision="rejected",
reason=("unknown cost code 'CODE-4': not in project proj's cost baseline (1 known codes)"),
)
verdict = stress.score_context_set(ctx, outbox, "r1", base)
assert [(r.approach_id, r.passed, r.stage) for r in verdict.must_refuse] == [
("a4", True, "stage0-baseline")
]
other = tmp_path / "b"
base2 = _minibase(other)
ctx2 = _context(other, must_refuse=True)
outbox2 = other / "out"
_write_outbox(outbox2, "r1", approach_id="a1", tool_calls=_opened(_GOOD))
_write_outbox(
outbox2,
"r1",
approach_id="a4",
codes=["CODE-4"],
decision="rejected",
reason="ungrounded identifier 'CODE-4': it appears nowhere in the input (10 chars)",
)
assert stress.score_context_set(ctx2, outbox2, "r1", base2).must_refuse[0].stage == (
"stage0b-grounding"
)
def test_p21_a_validated_falsification_arm_reports_no_stage(tmp_path: Path) -> None:
"""``""`` when nothing refused it — an honest absence, never a stage nobody reached."""
base = _minibase(tmp_path)
ctx = _context(tmp_path, must_refuse=True)
outbox = tmp_path / "out"
_write_outbox(outbox, "r1", approach_id="a1", tool_calls=_opened(_GOOD))
_write_outbox(outbox, "r1", approach_id="a4", codes=["CODE-4"])
verdict = stress.score_context_set(ctx, outbox, "r1", base)
assert verdict.must_refuse[0].passed is False
assert verdict.must_refuse[0].stage == ""