feat(row6): a proposal whose approach declared no requirement is unsupported
Stress round 6 validated three falsification arms, and every validated approach rested only on run-level declarations nobody can attribute to one approach. declare_requirement now takes a required approach_id (a mandate id or own-proposal; an unknown id is refused naming the valid ones), and a ValidatedProposal whose approach has neither a mandate requirement nor a declaration under its own id becomes validator.Unsupported - a Rejection subclass carrying the validator's own ruling, reported as `unsupported` in coverage, the outcome artefact, the settlement and the judge, and never counted or summed. The rule is active whenever the debate held the declaration tool, the micro base included; the road and pre-pass paths are untouched. Declaration quality is not judged, so the rule can be satisfied by declaring any document the run read. The v1 gate's row 6 probes pass; its artefact half reads IKKE MÅLT because stress round 6 predates approach-addressed declarations, and IKKE MÅLT is never green - it fails the exit code. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
parent
9847e014e7
commit
938a1ca30e
23 changed files with 718 additions and 115 deletions
292
tests/test_row6_declaration_rule_loadbearing.py
Normal file
292
tests/test_row6_declaration_rule_loadbearing.py
Normal file
|
|
@ -0,0 +1,292 @@
|
|||
"""Row 6 — a proposal whose approach declared no requirement cannot carry ``validated``.
|
||||
|
||||
Before this rule every numeric stage could pass and the run would stamp ``validated`` whether or
|
||||
not anything in the knowledge base had been named as binding the direction. Measured on stress
|
||||
round 6: three falsification arms validated, and every one of the ten validated approaches had only
|
||||
run-level declarations — which a judge cannot attribute to any one approach.
|
||||
|
||||
What each arm pins:
|
||||
|
||||
(a) on the shipped micro base, where the debate holds ``declare_requirement``, a silent approach
|
||||
that used to validate is now ``unsupported`` — and so is the run's own proposal;
|
||||
(b) a declaration filed under the approach's id is what lets it validate, and ONLY that approach;
|
||||
(c) a requirement written into the mandate counts as the approach's own declaration;
|
||||
(d) the declaration's quality is not judged: any requirement the run read is accepted;
|
||||
(e) an id no approach carries is refused with the valid ids named, and nothing is recorded;
|
||||
(f) ``unsupported`` is never counted, summed or selected as a success, yet the validator's own
|
||||
ruling is kept on the record (``provenance.validator_decision``) and in the artefact;
|
||||
(g) with the rule inactive (no declaration rung offered) the old ruling stands;
|
||||
(h) the judge reads the addressed declaration as the approach's own, a legacy one as ``run``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from portfolio_optimiser import okf, outbox, stress
|
||||
from portfolio_optimiser.explore import DeclaredRequirement, navigator_tools, requirement_payload
|
||||
from portfolio_optimiser.ir import AffectedItem, SavingsProposal
|
||||
from portfolio_optimiser.mandate import Approach, BindingRequirement, Mandate, settle
|
||||
from portfolio_optimiser.run import _evaluate_mandate, run_project
|
||||
from portfolio_optimiser.simulation import scripted_factory
|
||||
from portfolio_optimiser.validator import (
|
||||
UNSUPPORTED_REASON,
|
||||
Rejection,
|
||||
Unsupported,
|
||||
ValidatedProposal,
|
||||
rejection_stage,
|
||||
)
|
||||
from portfolio_optimiser.verdicts import VerdictStore
|
||||
|
||||
_BUNDLE = Path(__file__).resolve().parents[1] / "shared" / "examples" / "bygg-energi-mikro"
|
||||
_BASE_ID = "bygg-energi-mikro"
|
||||
_PID = "BYGG-KONTOR-NORD"
|
||||
_REPLY = (
|
||||
'{"measure":"LED-retrofit av kontorbelysning","affected_items":'
|
||||
'[{"code":"ENERGI-TOTAL-EL","quantity":300000,"unit_cost":1.0}],"claimed_saving_nok":30000}'
|
||||
)
|
||||
_CHECKER = "Reasoning holds.\nVERDICT: APPROVE"
|
||||
_CONCEPTS = [f.name for f in okf.navigate_bundle(str(_BUNDLE)).context_files]
|
||||
|
||||
|
||||
def _mandate(**approach: Any) -> Mandate:
|
||||
return Mandate(
|
||||
objective="Kutt energikostnad",
|
||||
approaches=(Approach(id="a1", label="LED-retrofit", description="reason", **approach),),
|
||||
)
|
||||
|
||||
|
||||
def _script(*declare_for: str, ref: str = "Krav 1.1-1") -> dict[str, Any]:
|
||||
steps: list[Any] = []
|
||||
if declare_for:
|
||||
steps += [
|
||||
{"call": "read_file", "args": {"bundle_id": _BASE_ID, "path": n}} for n in _CONCEPTS[:3]
|
||||
]
|
||||
steps += [
|
||||
{
|
||||
"call": "declare_requirement",
|
||||
"args": {
|
||||
"bundle_id": _BASE_ID,
|
||||
"path": _CONCEPTS[0],
|
||||
"ref": ref,
|
||||
"approach_id": aid,
|
||||
},
|
||||
}
|
||||
for aid in declare_for
|
||||
]
|
||||
return {"proposer": [*steps, _REPLY, _REPLY, _REPLY, _REPLY, _REPLY], "checker": _CHECKER}
|
||||
|
||||
|
||||
async def _run(tmp_path: Path, script: dict[str, Any], mandate: Mandate) -> Any:
|
||||
return await run_project(
|
||||
_PID,
|
||||
"local",
|
||||
docs_dir=str(_BUNDLE),
|
||||
bundle_dir=str(_BUNDLE),
|
||||
store=VerdictStore(verdicts=[]),
|
||||
client_factory=scripted_factory(script, []),
|
||||
mandate=mandate,
|
||||
outbox_dir=str(tmp_path),
|
||||
run_id="r6",
|
||||
)
|
||||
|
||||
|
||||
def _statuses(result: Any) -> dict[str, str]:
|
||||
return {row.id: row.status for row in result.coverage}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------
|
||||
# (a)-(d) the rule on the real run
|
||||
# ---------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_silent_approach_on_the_micro_base_is_unsupported(tmp_path: Path) -> None:
|
||||
result = await _run(tmp_path, _script(), _mandate())
|
||||
assert _statuses(result) == {"a1": "unsupported", "own-proposal": "unsupported"}
|
||||
assert {row.detail for row in result.coverage} == {UNSUPPORTED_REASON}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_only_the_approach_that_declared_is_validated(tmp_path: Path) -> None:
|
||||
result = await _run(tmp_path, _script("a1"), _mandate())
|
||||
assert _statuses(result) == {"a1": "validated", "own-proposal": "unsupported"}
|
||||
debate = json.loads((tmp_path / "r6-debate.json").read_text(encoding="utf-8"))
|
||||
assert [r["approach_id"] for r in debate["requirements"]] == ["a1"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_declaration_for_the_own_proposal_does_not_stand_in(tmp_path: Path) -> None:
|
||||
result = await _run(tmp_path, _script("own-proposal"), _mandate())
|
||||
assert _statuses(result) == {"a1": "unsupported", "own-proposal": "validated"}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_requirement_written_into_the_mandate_counts(tmp_path: Path) -> None:
|
||||
requirement = BindingRequirement(path=_CONCEPTS[0], ref="Krav 1.1-1")
|
||||
result = await _run(tmp_path, _script(), _mandate(requirement=requirement))
|
||||
assert _statuses(result)["a1"] == "validated"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_the_declarations_quality_is_not_judged(tmp_path: Path) -> None:
|
||||
result = await _run(tmp_path, _script("a1", ref="anything at all"), _mandate())
|
||||
assert _statuses(result)["a1"] == "validated"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------
|
||||
# (e) the address
|
||||
# ---------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _tool(approach_ids: list[str] | None) -> tuple[Any, list[DeclaredRequirement]]:
|
||||
opened: list[Any] = []
|
||||
declared: list[DeclaredRequirement] = []
|
||||
tools = navigator_tools(
|
||||
[str(_BUNDLE)], opened=opened, requirements=declared, approach_ids=approach_ids
|
||||
)
|
||||
from portfolio_optimiser.explore import ToolCall
|
||||
|
||||
opened += [ToolCall(name="read_file", bundle_id=_BASE_ID, path=n) for n in _CONCEPTS[:3]]
|
||||
return {t.name: t for t in tools}["declare_requirement"], declared
|
||||
|
||||
|
||||
def test_an_unknown_approach_id_is_refused_naming_the_valid_ones() -> None:
|
||||
tool, declared = _tool(["a1", "own-proposal"])
|
||||
reply = tool.func(bundle_id=_BASE_ID, path=_CONCEPTS[0], ref="K", approach_id="a9")
|
||||
assert reply["refusal"] == "UnknownApproach"
|
||||
assert "'a1'" in reply["refused"] and "'own-proposal'" in reply["refused"]
|
||||
assert declared == []
|
||||
ok = tool.func(bundle_id=_BASE_ID, path=_CONCEPTS[0], ref="K", approach_id="a1")
|
||||
assert ok["declared"] is True and ok["approach_id"] == "a1"
|
||||
assert declared == [
|
||||
DeclaredRequirement(bundle_id=_BASE_ID, path=_CONCEPTS[0], ref="K", approach_id="a1")
|
||||
]
|
||||
|
||||
|
||||
def test_without_a_commission_any_label_is_recorded_but_never_an_empty_one() -> None:
|
||||
tool, declared = _tool(None)
|
||||
assert tool.func(bundle_id=_BASE_ID, path=_CONCEPTS[0], ref="K", approach_id=" ")["refusal"]
|
||||
assert declared == []
|
||||
tool.func(bundle_id=_BASE_ID, path=_CONCEPTS[0], ref="K", approach_id="LED-retrofit")
|
||||
assert requirement_payload(declared) == [
|
||||
{"bundle_id": _BASE_ID, "path": _CONCEPTS[0], "ref": "K", "approach_id": "LED-retrofit"}
|
||||
]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------
|
||||
# (f) never counted as a success; the validator's ruling kept
|
||||
# ---------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_unsupported_is_never_a_success_but_keeps_the_validators_ruling(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
result = await _run(tmp_path, _script(), _mandate())
|
||||
assert isinstance(result.outcome, Unsupported)
|
||||
assert not isinstance(result.outcome, ValidatedProposal)
|
||||
assert result.provenance.validator_decision == "validated"
|
||||
text = settle(result.coverage)
|
||||
assert "UNSUPPORTED" in text and "Validated: 0 of 2" in text
|
||||
artefact = json.loads((tmp_path / "r6-a1-outcome.json").read_text(encoding="utf-8"))
|
||||
assert artefact["outcome_type"] == "unsupported"
|
||||
assert artefact["reason"] == UNSUPPORTED_REASON
|
||||
assert artefact["p50"] > 0
|
||||
proposal = json.loads((tmp_path / "r6-a1-proposal.json").read_text(encoding="utf-8"))
|
||||
assert proposal["provenance"]["validator_decision"] == "validated"
|
||||
|
||||
|
||||
def test_the_stage_label_names_the_new_falsifier() -> None:
|
||||
assert rejection_stage(UNSUPPORTED_REASON) == "unsupported"
|
||||
|
||||
|
||||
def _validated() -> ValidatedProposal:
|
||||
proposal = SavingsProposal(
|
||||
project_id=_PID,
|
||||
measure="m",
|
||||
affected_items=[AffectedItem(code="X", quantity=10.0, unit_cost=10.0)],
|
||||
claimed_saving_nok=5.0,
|
||||
)
|
||||
return ValidatedProposal(proposal=proposal, p10=1.0, p50=2.0, p90=3.0, nominal_feasible=4.0)
|
||||
|
||||
|
||||
def test_the_outcome_payload_carries_both_halves() -> None:
|
||||
v = _validated()
|
||||
payload = outbox.outcome_payload(
|
||||
Unsupported(proposal=v.proposal, reason=UNSUPPORTED_REASON, validated=v),
|
||||
checker_verdict="approve",
|
||||
verdict_id="k",
|
||||
)
|
||||
assert (payload["outcome_type"], payload["p90"]) == ("unsupported", 3.0)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------
|
||||
# (g) inactive rule
|
||||
# ---------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
("declared", "status"), [(None, "validated"), ([], "unsupported")], ids=["inactive", "active"]
|
||||
)
|
||||
async def test_the_rule_acts_only_when_a_declaration_rung_was_offered(
|
||||
declared: list[DeclaredRequirement] | None, status: str
|
||||
) -> None:
|
||||
v = _validated()
|
||||
|
||||
async def evaluate(_approach: Approach | None) -> ValidatedProposal | Rejection:
|
||||
return v
|
||||
|
||||
_, rows, _ = await _evaluate_mandate(
|
||||
Mandate(objective="o", approaches=(Approach(id="a1", label="l"),)),
|
||||
evaluate,
|
||||
declared=declared,
|
||||
)
|
||||
assert {r.id: r.status for r in rows} == {"a1": status, "own-proposal": status}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------
|
||||
# (h) the judge
|
||||
# ---------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_the_judge_attributes_addressed_declarations_and_labels_legacy_ones() -> None:
|
||||
a1 = Approach(id="a1", label="l")
|
||||
new = [{"path": "p1", "approach_id": "a1"}, {"path": "p2", "approach_id": "own-proposal"}]
|
||||
assert stress._attributable(a1, new) == (("p1",), "approach")
|
||||
assert stress._attributable(Approach(id="a2", label="l"), new) == ((), "absent")
|
||||
assert stress._attributable(a1, [{"path": "p0"}]) == (("p0",), "run")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_run_offered_no_declaration_rung_keeps_the_validators_ruling(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
"""The road path holds no knowledge base, so no ``declare_requirement`` exists there: nothing
|
||||
could have been declared, and the rule stays out of it. Drives the REAL ``run_project`` —
|
||||
the arm above only proves ``_evaluate_mandate`` honours ``declared=None``, not that the run
|
||||
passes it."""
|
||||
from portfolio_optimiser.reference_domain import load_reference_projects
|
||||
from portfolio_optimiser.validator import proposal_for
|
||||
|
||||
project = load_reference_projects()[0]
|
||||
reply = proposal_for(project, ["05.2", "03.1"], claimed_saving_nok=200_000).model_dump_json()
|
||||
docs = tmp_path / "docs"
|
||||
docs.mkdir()
|
||||
(docs / "kilde.md").write_text("Cost saving measure candidates for the project.\n", "utf-8")
|
||||
result = await run_project(
|
||||
project.id,
|
||||
"local",
|
||||
docs_dir=str(docs),
|
||||
store=VerdictStore(verdicts=[]),
|
||||
client_factory=scripted_factory({"proposer": reply, "checker": _CHECKER}, []),
|
||||
mandate=Mandate(
|
||||
objective="o", approaches=(Approach(id="a1", label="l"),), allow_own_proposals=False
|
||||
),
|
||||
)
|
||||
assert _statuses(result) == {"a1": "validated"}
|
||||
Loading…
Add table
Add a link
Reference in a new issue