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

feat(major2): run_project owns the review sink, keys it per approach and writes it from a finally [skip-docs]

Browse source 
Ordre 20260904T173146Z-8102814273-from-portfolio-optimiser, steg 4 av 10.

run_project faar proposal_reviewer (keyword-only, None => byte-identisk kjoering), eier
sinken expert_reviews ved siden av parse_failures, og skriver
{run_id}-proposal-reviews.json fra den EKSISTERENDE genererings-finally-en.

Skriveregelen er IFF en reviewer ble gitt, OGSAA naar lista er tom (D4). Begge halvdeler
er baerende og trekker hver sin vei: write_debate_tools skriver ubetinget fordi DER er det
tomme tilfellet regresjonen; her maa en reviewer-LOES kjoering la utboksen staa byte-identisk
(to eksisterende tester pinner et EKSAKT fire-navns-listing), mens en reviewer som ble tilbudt
og aldri konsultert er et faktum artefaktet maa kunne SI.

Noeklingen: med mandat er hver post noeklet - kjoeringens eget forslag paa OWN_PROPOSAL_ID -
og None betyr kun EN ting: det fantes intet mandat. RunResult.expert_revisions bygges FRA
sinken, aldri ved siden av (kø-(p)).

RODT foer impl: 8 armer.

REGRESJON FANGET AV FULL SUITE OG RETTET HER: steg 3s _FeedbackAwareChatClient kopierte
_inner_get_response-kroppen og gjorde test_scripted_client_consolidation
::test_inner_get_response_collapsed_to_two_sites roed. Doblen overstyrer naa _next_reply i
stedet - basen har alt lagt DENNE kallets prompt i received_texts naar den ber om et svar,
saa sommen holder uten en tredje kopi av kroppen, og registeret i vakten trenger ingen ny
oppfoering. Aa registrere fila som foreign lineage var ikke mulig og heller ikke riktig:
Group B bruker ScriptedChatClient, som den vakten nekter for nettopp den lista.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Kjell Tore Guttormsen 2026-09-05 06:58:18 +02:00
commit 59f3fce20a
3 changed files with 410 additions and 27 deletions
Download patch file Download diff file Expand all files Collapse all files

348
tests/test_proposal_review_loop_loadbearing.py
View file

@ -30,15 +30,19 @@ from pathlib import Path
from typing import Any
import pytest
from agent_framework import ChatResponse, ChatResponseUpdate, Message
from spikes._harness import FakeChatClient, message_texts
from portfolio_optimiser import hitl
from portfolio_optimiser import proposal_review as pr
from portfolio_optimiser.budget import Budget, BudgetExceeded, TokenMeter
from portfolio_optimiser.generate import ParseFailure, _build_messages, generate_via_llm
from portfolio_optimiser.ir import AffectedItem, SavingsProposal
from portfolio_optimiser.mandate import OWN_PROPOSAL_ID, Approach, Mandate
from portfolio_optimiser.reference_domain import load_reference_projects
from portfolio_optimiser.run import RunResult, run_project
from portfolio_optimiser.simulation import ScriptedChatClient
from portfolio_optimiser.validator import Rejection, ValidatedProposal, proposal_for
from portfolio_optimiser.verdicts import VerdictStore
_REPO = Path(__file__).resolve().parents[1]
_BUNDLE_DIR = _REPO / "shared" / "examples" / "bygg-energi-mikro"
@ -300,10 +304,15 @@ class _FeedbackAwareChatClient(FakeChatClient):
"""A proposer whose reply depends on the PROMPT: once the expert's sentinel arrives through
``prior_feedback`` it answers with the revised amount, otherwise with the first one.
The ``_ReasonAwareChatClient`` form from Step 5's gate, keyed on the human's words instead of
the machine's. It is what makes "the answer was used" observable: a door that renders the
candidate, reads the line and discards it leaves the flip key out of attempt 2's prompt, so
the outcome equals the always-approve control's and the arm goes RED.
The ``_ReasonAwareChatClient`` idea from Step 5's gate, keyed on the human's words instead of
the machine's — but implemented by overriding ``_next_reply`` rather than
``_inner_get_response``, so no scripted body is copied and the S2.5 consolidation guard
(``test_scripted_client_consolidation``) stays green. The base has already appended THIS
call's prompt to ``received_texts`` by the time it asks for a reply, so the seam is enough.
It is what makes "the answer was used" observable: a door that renders the candidate, reads
the line and discards it leaves the flip key out of attempt 2's prompt, so the outcome equals
the always-approve control's and the arm goes RED.
"""
def __init__(self, flip_key: str, first_reply: str, revised_reply: str) -> None:
@ -312,29 +321,12 @@ class _FeedbackAwareChatClient(FakeChatClient):
self._first = first_reply
self._revised = revised_reply
def _inner_get_response(
self, *, messages: Sequence[Message], stream: bool, options: Any, **kwargs: Any
) -> Any:
received = message_texts(messages)
self.received_texts.append(received)
def _next_reply(self) -> str:
# ``total_tokens`` is a spike-measurement field nothing here reads (the fake response
# carries no ``usage_details``, so the meter never charges from it) and is left alone.
self.call_count += 1
reply = self._revised if self._flip_key in " ".join(received) else self._first
if stream:
async def _agen() -> Any:
yield ChatResponseUpdate(
role="assistant", contents=[{"type": "text", "text": reply}]
)
return self._build_response_stream(_agen())
async def _coro() -> ChatResponse:
return ChatResponse(
messages=[Message(role="assistant", contents=[reply])], response_id="fake"
)
return _coro()
received = " ".join(self.received_texts[-1]) if self.received_texts else ""
return self._revised if self._flip_key in received else self._first
class _RecordingReviewer:
@ -645,3 +637,305 @@ async def test_t13_an_honoured_revise_rejected_next_leaves_the_validators_last_r
assert (record.decision, record.honoured) == ("revise", True)
# The candidate the expert LOOKED at survives in the record, keyed and priced.
assert record.proposal.proposal.claimed_saving_nok == _FIRST_CLAIM
# ---------------------------------------------------------------------------------------------
# Group B (Step 4) — the REAL ``run_project``: the sink, the keying, the artefact, and the
# controls that keep a reviewer-less run byte-identical.
# ---------------------------------------------------------------------------------------------
_RUN_PID = "BYGG-KONTOR-NORD"
_VERDICT_INPUT = {"decision": "approved", "rationale": "expert reviewed (sim)"}
#: The instruction line ``generate._build_messages`` puts in EVERY generation prompt and nowhere
#: else — the one identifier that separates a generation call from a debate turn.
_GENERATION_MARK = "Respond with ONLY a JSON object"
# BYGG-KONTOR-NORD: affected total 300000 x 1.0 -> degenerate Monte Carlo P90 = 90000. A claim
# <= 90000 validates; above it the deterministic validator rejects.
_A1 = Approach(id="led-retrofit", label="Behovsstyrt belysning i fellesarealer")
_A2 = Approach(id="hvac-swap", label="Utskifting av ventilasjonsaggregat")
_A3 = Approach(id="tetting", label="Tetting av klimaskjerm mot kaldloft")
def _run_reply(measure: str, claimed: int) -> str:
return (
f'{{"measure":"{measure}","affected_items":'
f'[{{"code":"ENERGI-TOTAL-EL","quantity":300000,"unit_cost":1.0}}],'
f'"claimed_saving_nok":{claimed}}}'
)
def _keyed_selector(*, revised_claim: int = 40_000, first_claim: int = 30_000):
"""Answer by WHICH approach the prompt carries, and with a DIFFERENT amount once the expert's
sentinel has arrived through ``prior_feedback`` — the canonical ``reply_selector`` seam."""
def select(blob: str, _role: str) -> str:
if _GENERATION_MARK not in blob:
return "ok"
label = next(
(a.label for a in (_A1, _A2, _A3) if a.label in blob), "Systemets eget forslag"
)
claim = revised_claim if _FEEDBACK_SENTINEL in blob else first_claim
return _run_reply(label, claim)
return select
def _run_factory(select, sink: list[str] | None = None):
def factory(role: str):
return ScriptedChatClient(sink=sink, role=role, reply_selector=select, default_reply="ok")
return factory
class _RunReviewer:
"""Revise ONCE per candidate (when there is an attempt to buy), then approve. The feedback is
keyed to the approach, which is what makes "feedback never crosses approaches" observable."""
def __init__(self, revise_for: set[str | None] | None = None) -> None:
self.calls: list[pr.ProposalReviewRequest] = []
self._revised: set[str | None] = set()
self._revise_for = revise_for
def __call__(self, request: pr.ProposalReviewRequest) -> pr.ProposalReviewDecision:
self.calls.append(request)
wanted = self._revise_for is None or request.approach_id in self._revise_for
if wanted and request.approach_id not in self._revised and request.attempts_remaining > 0:
self._revised.add(request.approach_id)
return pr.ProposalReviewDecision.revise(f"{_FEEDBACK_SENTINEL}-{request.approach_id}")
return pr.ProposalReviewDecision.approve()
async def _run_with(
*,
outbox_dir: Path,
run_id: str = _RUN_ID,
reviewer=None,
select=None,
mandate=None,
sink: list[str] | None = None,
verdict_input: dict[str, str] | None = None,
max_rounds: int = 40,
):
return await run_project(
_RUN_PID,
"local",
docs_dir=str(_BUNDLE_DIR),
bundle_dir=str(_BUNDLE_DIR),
verdict_input=verdict_input,
store=VerdictStore(verdicts=[]),
client_factory=_run_factory(select or _keyed_selector(), sink),
outbox_dir=str(outbox_dir),
run_id=run_id,
mandate=mandate,
proposal_reviewer=reviewer,
meter=TokenMeter(Budget(max_tokens=10**9, max_rounds=max_rounds)),
)
def _reviews_artefact(outbox_dir: Path, run_id: str = _RUN_ID) -> Path:
return outbox_dir / f"{run_id}-proposal-reviews.json"
def _reviews_on_disk(outbox_dir: Path, run_id: str = _RUN_ID) -> dict[str, Any]:
return json.loads(_reviews_artefact(outbox_dir, run_id).read_text(encoding="utf-8"))
async def test_t8_every_candidate_gets_its_own_keyed_answer(tmp_path: Path) -> None:
"""T8 — the record is KEYED per candidate. Detach points: ``approach_id`` always ``None``
(M10), feedback hoisted out of the per-approach call (M34).
With a commissioned mandate one expert is asked N times at one terminal, so an artefact that
cannot say WHICH candidate a sentence was about is an artefact nobody can act on. The run's
own proposal is keyed ``OWN_PROPOSAL_ID`` — recording ``None`` there would make it
indistinguishable from a non-mandate run's entry, which is the property keying exists for."""
reviewer = _RunReviewer()
result = await _run_with(
outbox_dir=tmp_path / "outbox",
reviewer=reviewer,
mandate=Mandate(
objective="Cut energy cost without rebuilding.",
approaches=(_A1, _A2),
allow_own_proposals=True,
),
)
assert isinstance(result, RunResult)
revisions = {
(r.approach_id, r.feedback) for r in result.expert_revisions if r.decision == "revise"
}
assert revisions == {
("led-retrofit", f"{_FEEDBACK_SENTINEL}-led-retrofit"),
("hvac-swap", f"{_FEEDBACK_SENTINEL}-hvac-swap"),
(OWN_PROPOSAL_ID, f"{_FEEDBACK_SENTINEL}-{OWN_PROPOSAL_ID}"),
}
# The attempt index restarts per candidate: each approach is its own ``generate_via_llm`` call.
by_key: dict[str | None, list[int]] = {}
for record in result.expert_revisions:
by_key.setdefault(record.approach_id, []).append(record.attempt)
assert by_key == {"led-retrofit": [0, 1], "hvac-swap": [0, 1], OWN_PROPOSAL_ID: [0, 1]}
async def test_t5run_a_revise_on_one_approach_leaves_the_next_untouched(tmp_path: Path) -> None:
"""T5-run — feedback never crosses approaches. Detach point: a run-level feedback carried
across the per-approach calls (M34).
Approach 3's generation prompts must be BYTE-IDENTICAL to the control's. Approach 1 is
deliberately NOT asserted: ``_evaluate_mandate`` is sequential, so it finishes before approach
2's reviewer is ever called and its identity holds by construction — asserting it would be the
vacuous half. Both arms run under a meter provably large enough that ``coverage`` holds no
``not_evaluated`` row, so 'unchanged' cannot mean 'never reached'."""
mandate = Mandate(
objective="Cut energy cost without rebuilding.",
approaches=(_A1, _A2, _A3),
allow_own_proposals=False,
)
treated_sink: list[str] = []
treated = await _run_with(
outbox_dir=tmp_path / "treated",
run_id="run-treated",
reviewer=_RunReviewer(revise_for={"hvac-swap"}),
mandate=mandate,
sink=treated_sink,
)
control_sink: list[str] = []
control = await _run_with(
outbox_dir=tmp_path / "control",
run_id="run-control",
reviewer=_RunReviewer(revise_for=set()),
mandate=mandate,
sink=control_sink,
)
assert isinstance(treated, RunResult) and isinstance(control, RunResult)
for run in (treated, control):
assert [row.status for row in run.coverage].count("not_evaluated") == 0
def _gen_prompts(sink: list[str], label: str) -> list[str]:
return [b for b in sink if _GENERATION_MARK in b and label in b]
assert len(_gen_prompts(treated_sink, _A2.label)) == 2 # the revise bought one more
assert len(_gen_prompts(control_sink, _A2.label)) == 1
assert _gen_prompts(treated_sink, _A3.label) == _gen_prompts(control_sink, _A3.label)
assert all(_FEEDBACK_SENTINEL not in b for b in _gen_prompts(treated_sink, _A3.label))
async def test_t9_a_budget_stop_inside_generation_still_leaves_the_record(tmp_path: Path) -> None:
"""T9 — the artefact is written from a ``finally``. Detach points: writing after the return
(M11), building the record from a RETURNED list instead of the caller-owned sink (M35).
The bought attempt's reply does not parse, the parse-retry exhausts the round ledger, and
``BudgetExceeded`` leaves ``run_project`` as an exception. This is precisely the run whose
record matters most, and precisely the one a return value cannot reach."""
outbox = tmp_path / "outbox"
def select(blob: str, _role: str) -> str:
if _GENERATION_MARK not in blob:
return "ok"
return "not json at all" if _FEEDBACK_SENTINEL in blob else _run_reply("Belysning", 30_000)
with pytest.raises(BudgetExceeded):
await _run_with(outbox_dir=outbox, reviewer=_RunReviewer(), select=select, max_rounds=2)
payload = _reviews_on_disk(outbox)
assert payload["run_id"] == _RUN_ID
assert [(r["decision"], r["feedback"], r["honoured"]) for r in payload["reviews"]] == [
("revise", f"{_FEEDBACK_SENTINEL}-None", False)
]
async def test_t10_a_reviewer_nobody_could_consult_still_writes_an_empty_record(
tmp_path: Path,
) -> None:
"""T10 — written IFF a reviewer was given, INCLUDING when empty (D4). A reviewer that was
offered and never consulted is a fact the artefact must be able to state; inferring it from an
absent file would confuse it with T11's reviewer-less run."""
outbox = tmp_path / "outbox"
reviewer = _RunReviewer()
result = await _run_with(
outbox_dir=outbox,
reviewer=reviewer,
select=_keyed_selector(first_claim=200_000, revised_claim=200_000), # always rejected
)
assert isinstance(result, RunResult)
assert isinstance(result.outcome, Rejection)
assert reviewer.calls == []
assert _reviews_on_disk(outbox) == {"reviews": [], "run_id": _RUN_ID}
async def test_t11_a_run_with_no_reviewer_leaves_the_outbox_byte_identical(
tmp_path: Path,
) -> None:
"""T11 — the CONTROL that keeps every existing run untouched. Detach point: writing the
artefact with no reviewer (M13).
Two reviewer-less runs into separate outbox directories: the same EXACT four names, and the
same bytes. ``-proposal-reviews.json`` must be ABSENT — that absence is what makes T10's
``{"reviews": []}`` mean 'offered, never consulted' rather than 'not written'."""
first, second = tmp_path / "a", tmp_path / "b"
for outbox in (first, second):
result = await _run_with(outbox_dir=outbox, reviewer=None)
assert isinstance(result, RunResult)
expected = [
f"{_RUN_ID}-debate.json",
f"{_RUN_ID}-outcome.json",
f"{_RUN_ID}-proposal.json",
f"{_RUN_ID}-runconfig.json",
]
assert sorted(p.name for p in first.glob("*.json")) == expected
assert not _reviews_artefact(first).exists()
for name in expected:
assert (first / name).read_bytes() == (second / name).read_bytes()
async def test_t12_an_approve_is_not_an_expert_verdict(tmp_path: Path) -> None:
"""T12 — ``approve`` mints NO ``Verdict``. Detach point: minting one (M14).
F2 stands: a verdict arises only from ``--decision/--rationale`` or the Step-7 inbox.
``approve`` means "stop asking; I take it as is", which is a different act from judging the
measure's worth — and the control proves a verdict IS still minted the F2 way, so the arm
cannot pass merely because verdicts stopped working."""
reviewed = await _run_with(outbox_dir=tmp_path / "reviewed", reviewer=_RunReviewer())
assert isinstance(reviewed, RunResult)
assert reviewed.expert_revisions # the expert DID answer
assert reviewed.verdict is None
assert reviewed.verdict_key not in {v.id for v in reviewed.store.verdicts}
judged = await _run_with(
outbox_dir=tmp_path / "judged",
run_id="run-judged",
reviewer=_RunReviewer(),
verdict_input=_VERDICT_INPUT,
)
assert isinstance(judged, RunResult)
assert judged.verdict is not None
assert judged.verdict.id == judged.verdict_key
async def test_the_record_carries_the_reviewed_candidates_key_and_p50(tmp_path: Path) -> None:
"""The payload arm: every entry is keyed the way an expert verdict on that candidate would
be. Detach point: dropping ``verdict_key`` from the record (M28).
The approve entry sits on the candidate the run ended up carrying, so its key must equal the
run's own ``verdict_key`` — the join back into the Step-7 inbox channel."""
outbox = tmp_path / "outbox"
result = await _run_with(outbox_dir=outbox, reviewer=_RunReviewer())
assert isinstance(result, RunResult)
entries = _reviews_on_disk(outbox)["reviews"]
assert [e["decision"] for e in entries] == ["revise", "approve"]
assert entries[-1]["verdict_key"] == result.verdict_key
assert entries[-1]["p50"] == pytest.approx(result.outcome.p50)
assert entries[0]["verdict_key"] != entries[-1]["verdict_key"] # a DIFFERENT candidate
async def test_the_review_artefact_is_invisible_to_the_hitl_registry(tmp_path: Path) -> None:
"""A RATCHET, green at construction (the B4 M4 precedent): ``hitl._read_outbox_proposals``
globs ``*-proposal.json``, which cannot match ``*-proposal-reviews.json`` today. The arm
guards a future rename that would make ``hitl pending`` read a review as a proposal."""
outbox, inbox = tmp_path / "outbox", tmp_path / "inbox"
inbox.mkdir()
await _run_with(outbox_dir=outbox, reviewer=_RunReviewer())
with_artefact = hitl.pending(str(outbox), str(inbox))
_reviews_artefact(outbox).unlink()
assert len(hitl.pending(str(outbox), str(inbox))) == len(with_artefact) == 1