open/portfolio-optimiser
1
0
Fork 0
Code Issues Releases 3 Activity
portfolio-optimiser/tests/test_proposal_review_loop_loadbearing.py
Kjell Tore Guttormsen c84e8bf6f1 feat(p19): a direction must NAME the requirement that binds it, and have READ it 
Two paid rounds scored 0 of 26 fasit concepts opened -- the same number twice.
P18 closed the navigation side (a listing is a window, an invented path is
refused by name) and it did not move, which makes it a ROLE question: nothing
in the loop ever asked the model to say what requirement binds the direction it
committed to, so opening one was never on the critical path to an answer.

A PREMISE OF THE ORDER WAS FELLED BEFORE ANYTHING WAS BUILT ON IT. A1 places
the demand in _INSTRUCTIONS[HYPOTHESISER_ROLE] alone. Measured: the stress
command sends --mandate and NOT --explore, the two are refused together by
name, and none of the nine round-1/2 outboxes holds a {run_id}-exploration.json
-- the hypothesiser never runs in a stress round, so A3 would have been
unreachable in exactly the paid runs this order commissions.

A2's own sentence resolves it: the refusal goes to the model "som en tur den
kan rette (samme mekanisme som quick_validate's nekt), ikke som en raise" --
and quick_validate IS a tool. declare_requirement therefore lives in
navigator_tools, held by BOTH roles that navigate (the exploration, and since
S2c the debate). It EXISTS only when the caller offers both sinks, which keeps
every pre-P19 call site byte-identical; one sink without the other is refused
at construction. 'opened' is the SAME list ExplorationToolRecorder fills, so
the refusal reads the run's own read trace.

The marked hypothesis carries 'requirement' as a REQUIRED key: omitted is a
hard error, explicit null is legal and needs 'why_none', a half-named one is
refused. A minted approach carries it; a seed never acquires one. The proposer
prompt names it only when the field exists, and the judge counts a hit against
THIS approach's fasit concepts, never against the base.

Load-bearing measured (12 arms), four mutations all red against the whole
suite, green control 1711/5 and demo-transcript.stdout byte-unchanged.
A-iii's predicted signature was FALSIFIED: the golden stays green because the
demo runs without a mandate, so _build_messages' approach branch is never
taken there. A-iv was GREEN first -- the repo's vacuous-gate class, 24th time:
the arm drove _attributable while the hit is computed at the call site.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-15 01:21:47 +02:00

1804 lines
81 KiB
Python
Raw Blame History

"""MAJOR-2 — a human must be able to answer the proposal that is ON THE TABLE, and the answer
must be USED.
Measured before this door existed (``docs/2026-09-02-misjonsreview-v2.md`` § 3, re-measured at
HEAD ``da5f10f``): ``grep -n "prior_" generate.py`` finds exactly ONE seam — ``prior_rejection``,
the MACHINE's own falsification fed back inside the same run — and the only human seam in the
loop is ``plan_reviewer``, which fires BEFORE any hypothesis exists. A human's verdict lands in
the NEXT run (Step 7's inbox). So "improved proposals given feedback" was, for a person, a
two-run loop with days between it.
**The gate is the SECOND half of that phrase.** A door that renders the candidate, reads a line
and discards it passes "the expert was asked" and fails the målbilde — this repo's vacuous-gate
class, fifteen times over. Every arm here is therefore built so that an **always-approve
reviewer is RED**: the discriminator is that a ``revise`` reaches the NEXT generation attempt's
prompt VERBATIM, that the proposer answers differently because of it, and that the run carries
the validator's ruling on THAT attempt (D6).
Three things the door must NOT do, gated as hard as the things it must: ``approve`` mints no
``Verdict`` (F2 stands); a ``revise`` never writes ``provenance.validator_decision`` and never
touches the checker gate (the two falsifiers stay unblended — the human is a third voice that
gates nothing except asking for one more attempt); and a run with NO reviewer is byte-identical
to today, golden transcript included.
"""
from __future__ import annotations
import inspect
import io
import json
import subprocess
import sys
from collections.abc import Sequence
from pathlib import Path
from typing import Any
import pytest
from spikes._harness import FakeChatClient, message_texts
from portfolio_optimiser import hitl, hosting
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 import run
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]
_EXAMPLES = _REPO / "shared" / "examples"
_BUNDLE_DIR = _EXAMPLES / "bygg-energi-mikro"
_PID = "BYGG-KONTOR-NORD"
_RUN_ID = "proposal-review-door"
#: A string that can reach a generation prompt ONLY by travelling through the expert's feedback:
#: asserted absent from every file of the fixture bundle and from the base prompt (Step 2), so a
#: prompt that carries it can have got it from exactly one place.
_FEEDBACK_SENTINEL = "PROPOSAL-REVIEW-SENTINEL-9c41ae"
#: The shape of the defect BLOCKER-1 measured, asserted NEGATIVELY: any object rendered by its
#: default repr onto a surface a human reads. Stronger than a positive sentinel alone, which a
#: renderer that prints nothing at all would also satisfy.
_REPR_LEAK = " object at 0x"
def _proposal(*, amount: float = 30000.0, measure: str = "LED-retrofit") -> SavingsProposal:
return SavingsProposal(
project_id=_PID,
measure=measure,
affected_items=[AffectedItem(code="ENERGI-TOTAL-EL", quantity=300000.0, unit_cost=1.0)],
claimed_saving_nok=amount,
)
def _validated(*, amount: float = 30000.0, p50: float = 42.0) -> ValidatedProposal:
return ValidatedProposal(
proposal=_proposal(amount=amount),
p10=p50 - 1.0,
p50=p50,
p90=p50 + 1.0,
nominal_feasible=p50 + 2.0,
)
def _review(
*,
approach_id: str | None = None,
attempt: int = 0,
decision: str = "approve",
feedback: str = "",
honoured: bool = True,
p50: float = 42.0,
) -> pr.ProposalReview:
return pr.ProposalReview(
approach_id=approach_id,
attempt=attempt,
decision=decision, # type: ignore[arg-type]
feedback=feedback,
honoured=honoured,
proposal=_validated(p50=p50),
)
# ---------------------------------------------------------------------------------------------
# Group A0 (Step 1) — the types, the error's CLASS, and the two renderers.
# ---------------------------------------------------------------------------------------------
def test_a_revision_must_say_what_to_revise() -> None:
"""Detach point: ``ProposalReviewDecision.revise`` accepting an empty string. A bare
``revise`` is not a decision — it is a line the door must ask again for. The type refuses it
at construction so no surface can invent the distinction for itself."""
with pytest.raises(ValueError):
pr.ProposalReviewDecision.revise("")
with pytest.raises(ValueError):
pr.ProposalReviewDecision.revise(" ")
assert pr.ProposalReviewDecision.revise("do X").feedback == "do X"
assert pr.ProposalReviewDecision.approve().feedback is None
def test_the_input_error_is_a_runtime_error_and_never_a_value_error() -> None:
"""Detach point: making ``ProposalReviewInputError`` a ``ValueError``. **The class IS the
channel**, decided by three measurements (brief § Constraints): a ``ValueError`` would land
on ``run.py``'s refusal tuple and print ``run refused:`` for a run whose argv was fine and
which had already spent tokens; and it would sit ONE frame from ``_fetch_parsed``'s
``except (ValidationError, ValueError, TypeError)`` catch-all, where a human-input error
would be appended to ``parse_failures`` and the model re-called until the ledger fired.
Both halves are asserted, because ``issubclass(X, RuntimeError)`` alone stays green on a
class that inherits from both."""
assert issubclass(pr.ProposalReviewInputError, RuntimeError)
assert not issubclass(pr.ProposalReviewInputError, ValueError)
def test_the_notice_is_none_exactly_when_no_reviewer_was_offered() -> None:
"""Detach point: a renderer that always returns its line. The ``announce`` rule: a run that
offered no reviewer has nothing to say, and an empty row would read as a dropped line."""
assert pr.proposal_review_notice((), offered=False) is None
assert pr.proposal_review_notice((_review(),), offered=False) is None
assert pr.proposal_review_notice((), offered=True) is not None
def test_zero_reviews_with_a_reviewer_offered_says_so_in_words() -> None:
"""Detach point: returning ``None`` on zero reviews with a reviewer present (M33).
A DELIBERATE departure from the announce rule's zero-is-silence half, and the reason is the
operator: someone who passed ``--proposal-review`` and sees nothing cannot tell "no candidate
was ever validated, so nobody was asked" from "the door hung". The departure is stated in the
invariant row rather than left as an inconsistency."""
line = pr.proposal_review_notice((), offered=True)
assert line == " proposal review offered, never consulted (no candidate validated)"
def test_the_notice_counts_answers_candidates_and_the_unhonoured_ones() -> None:
"""Detach point: dropping the un-honoured tail, or counting entries instead of candidates.
An un-honoured revise (no attempt left to buy) is a FACT the operator must be able to read:
the expert asked for a change that the run could not make."""
reviews = (
_review(approach_id="a1", attempt=0, decision="revise", feedback="F", honoured=True),
_review(approach_id="a1", attempt=1, decision="approve"),
_review(approach_id="a2", attempt=0, decision="revise", feedback="G", honoured=False),
)
assert pr.proposal_review_notice(reviews, offered=True) == (
" proposal review: 3 answer(s) across 2 candidate(s) — 1 approve, 2 revise "
"(1 not honoured)"
)
honoured_only = (
_review(approach_id="a1", attempt=0, decision="revise", feedback="F", honoured=True),
_review(approach_id="a1", attempt=1, decision="approve"),
)
assert pr.proposal_review_notice(honoured_only, offered=True) == (
" proposal review: 2 answer(s) across 1 candidate(s) — 1 approve, 1 revise"
)
def test_the_payload_carries_every_field_and_takes_its_key_from_the_injected_function() -> None:
"""Detach point: dropping ``verdict_key`` (M28), or deriving it inside the module.
``verdicts.verdict_key`` takes ``ProposalFeatures`` and the only derivation from a
``SavingsProposal`` is ``run._features_of`` — but ``verdicts.py`` imports MAF and ``run`` is
a cycle from here, so the key function is INJECTED and ``_features_of`` stays the single
home (kø-(p)). The arm asserts the injected callable is handed the REVIEWED proposal, not a
re-derived one."""
seen: list[SavingsProposal] = []
def key_of(proposal: SavingsProposal) -> str:
seen.append(proposal)
return f"key-{len(seen)}"
reviews = (
_review(approach_id="a1", attempt=0, decision="revise", feedback="F", honoured=False),
_review(approach_id=None, attempt=1, decision="approve", p50=99.5),
)
payload = pr.proposal_reviews_payload(reviews, key_of=key_of)
assert list(payload) == ["reviews"]
assert payload["reviews"] == [
{
"approach_id": "a1",
"attempt": 0,
"decision": "revise",
"feedback": "F",
"honoured": False,
"verdict_key": "key-1",
"p50": 42.0,
},
{
"approach_id": None,
"attempt": 1,
"decision": "approve",
"feedback": "",
"honoured": True,
"verdict_key": "key-2",
"p50": 99.5,
},
]
assert seen == [reviews[0].proposal.proposal, reviews[1].proposal.proposal]
# The payload is a PLAIN mapping: ``outbox`` stays MAF-free and byte-deterministic.
assert json.loads(json.dumps(payload)) == payload
# ---------------------------------------------------------------------------------------------
# Group A1 (Step 2) — ``prior_feedback``: the third composable block of the hypothesis prompt.
# ---------------------------------------------------------------------------------------------
@pytest.fixture(scope="module")
def project():
return load_reference_projects()[0] # FV42-GSV-E1
def test_prior_feedback_none_keeps_the_base_prompt_byte_identical(project) -> None:
"""CONTROL: the new block is INERT when nobody reviewed. A test that can only go green
proves nothing, and this is the half that keeps every existing run, golden and nav-fixture
untouched — the contract ``prior_rejection`` and ``approach`` already state."""
base = message_texts(_build_messages(project, "ctx"))[0]
assert base == message_texts(_build_messages(project, "ctx", prior_feedback=None))[0]
assert _FEEDBACK_SENTINEL not in base
def test_the_experts_words_reach_the_next_prompt_verbatim(project) -> None:
"""Detach point: dropping the ``prior_feedback`` block (M3 — print-and-discard).
VERBATIM and only the text: never the previous proposal JSON. That is the ``prior_rejection``
rule, and it is the same reason — the model must address what the human said, not parrot the
candidate they were unhappy with."""
text = message_texts(
_build_messages(
project, "ctx", prior_feedback=f"Use 150000, not 200000. {_FEEDBACK_SENTINEL}"
)
)[0]
assert _FEEDBACK_SENTINEL in text
assert "asked for a revision" in text
assert "Use 150000, not 200000." in text
def test_a_rejection_and_a_standing_feedback_compose_in_order(project) -> None:
"""The two blocks compose, and the ORDER is fixed: base -> approach head -> rejection ->
feedback.
This is the shape T6 measures end to end: a revise whose bought attempt the validator then
rejects leaves the attempt after it carrying BOTH — the human's instruction stands until the
human next answers, the machine's reason is per-attempt (only the most recent, as today)."""
rejection = Rejection(proposal=_proposal(), reason="claimed saving 270000 exceeds P90 121057")
text = message_texts(
_build_messages(
project, "ctx", prior_rejection=rejection, prior_feedback=_FEEDBACK_SENTINEL
)
)[0]
assert text.count(rejection.reason) == 1
assert text.count(_FEEDBACK_SENTINEL) == 1
assert text.index("REJECTED by the deterministic validator") < text.index(_FEEDBACK_SENTINEL)
def test_the_feedback_sentinel_cannot_arrive_from_the_knowledge_base() -> None:
"""The known-positive control on every sentinel arm in this file: the marker exists NOWHERE
in the fixture bundle, so a prompt that carries it can only have got it through the expert's
feedback block. Paired with an assertion that the scan actually reads the files — a scan that
silently finds nothing makes a gate that can only go green."""
texts = [
path.read_text(encoding="utf-8")
for path in sorted(_BUNDLE_DIR.rglob("*"))
if path.is_file()
]
assert texts, "the fixture bundle must exist for this control to mean anything"
assert any("ENERGI-TOTAL-EL" in text for text in texts) # the scan CAN find a known string
assert not any(_FEEDBACK_SENTINEL in text for text in texts)
# ---------------------------------------------------------------------------------------------
# Group A (Step 3) — the reviewer INSIDE the attempt loop. Driven at ``generate_via_llm``.
# ---------------------------------------------------------------------------------------------
# FV42-GSV-E1 cost codes 05.2 + 03.1 -> affected total 1_482_500, degenerate Monte Carlo
# P90 = 0.30 x 1_482_500 = 444_750 (the ``test_step5_refine_loadbearing`` fixture, reused).
_CODES = ["05.2", "03.1"]
_BAD_CLAIM = 800_000 # parseable, but above P90 -> the DETERMINISTIC validator rejects it
_FIRST_CLAIM = 200_000 # validates
_REVISED_CLAIM = 150_000 # validates too, and differs -> "the answer was USED" is observable
def _meter(*, rounds: int = 20) -> TokenMeter:
return TokenMeter(Budget(max_tokens=10**9, max_rounds=rounds))
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`` 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:
super().__init__()
self._flip_key = flip_key
self._first = first_reply
self._revised = revised_reply
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
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:
"""One reviewer double: scripted decisions, every request recorded. The script's LAST entry
repeats, so an always-approve or always-revise policy is a one-element script."""
def __init__(self, script: Sequence[pr.ProposalReviewDecision]) -> None:
self._script = list(script)
self.calls: list[pr.ProposalReviewRequest] = []
def __call__(self, request: pr.ProposalReviewRequest) -> pr.ProposalReviewDecision:
self.calls.append(request)
return self._script[min(len(self.calls) - 1, len(self._script) - 1)]
def _replies(project) -> tuple[str, str, str]:
"""(first, revised, bad) as the model would emit them — built through ``proposal_for`` so the
SUT parses exactly what the test built."""
return (
proposal_for(project, _CODES, claimed_saving_nok=_FIRST_CLAIM).model_dump_json(),
proposal_for(project, _CODES, claimed_saving_nok=_REVISED_CLAIM).model_dump_json(),
proposal_for(project, _CODES, claimed_saving_nok=_BAD_CLAIM).model_dump_json(),
)
async def test_t1_a_revise_reaches_the_next_prompt_verbatim_and_changes_the_outcome(
project,
) -> None:
"""T1 — THE discriminator. Detach points: the reviewer call (M1), ignoring the decision (M2),
dropping the ``prior_feedback`` injection (M3), truncating the feedback into the sink (M26).
A door that shows the candidate, reads the answer and discards it passes "the expert was
asked". The gate is the second half: attempt 2's prompt must carry the expert's words
VERBATIM, and the outcome must DIFFER from the always-approve control's."""
first, revised, _ = _replies(project)
feedback = f"Bruk {_REVISED_CLAIM}, ikke {_FIRST_CLAIM}. {_FEEDBACK_SENTINEL}"
client = _FeedbackAwareChatClient(_FEEDBACK_SENTINEL, first, revised)
reviewer = _RecordingReviewer(
[pr.ProposalReviewDecision.revise(feedback), pr.ProposalReviewDecision.approve()]
)
reviews: list[pr.ProposalReview] = []
result = await generate_via_llm(
client, project, "ctx", _meter(), reviewer=reviewer, reviews=reviews
)
assert client.call_count == 2
assert _FEEDBACK_SENTINEL not in " ".join(client.received_texts[0])
assert feedback in " ".join(client.received_texts[1]) # VERBATIM, not a paraphrase
assert isinstance(result.outcome, ValidatedProposal)
assert result.outcome.proposal.claimed_saving_nok == _REVISED_CLAIM
assert [(r.decision, r.feedback, r.honoured) for r in reviews] == [
("revise", feedback, True),
("approve", "", True),
]
# CONTROL: the same fixture with an always-approve reviewer stops at one call and keeps the
# original amount — so T1's flip cannot be an artefact of the client.
control_client = _FeedbackAwareChatClient(_FEEDBACK_SENTINEL, first, revised)
control_reviews: list[pr.ProposalReview] = []
control = await generate_via_llm(
control_client,
project,
"ctx",
_meter(),
reviewer=_RecordingReviewer([pr.ProposalReviewDecision.approve()]),
reviews=control_reviews,
)
assert control_client.call_count == 1
assert isinstance(control.outcome, ValidatedProposal)
assert control.outcome.proposal.claimed_saving_nok == _FIRST_CLAIM
assert [r.decision for r in control_reviews] == ["approve"]
@pytest.mark.parametrize("max_attempts", [2, 3])
async def test_t2_always_revise_stops_at_the_attempt_budget(project, max_attempts: int) -> None:
"""T2 — the loop stays BOUNDED: a revise consumes one of the attempts the loop already has,
it does not open a new one (M8). Parametrised because a fixed expected count cannot tell a
bound from a coincidence — the counter has to MOVE with the budget.
The final record is ``honoured=False``: the last attempt has nothing left to buy."""
first, revised, _ = _replies(project)
client = _FeedbackAwareChatClient("never-appears", first, revised)
reviewer = _RecordingReviewer([pr.ProposalReviewDecision.revise("again")])
reviews: list[pr.ProposalReview] = []
result = await generate_via_llm(
client,
project,
"ctx",
_meter(),
max_attempts=max_attempts,
reviewer=reviewer,
reviews=reviews,
)
assert client.call_count == max_attempts
assert len(reviews) == max_attempts
assert [r.honoured for r in reviews] == [True] * (max_attempts - 1) + [False]
assert isinstance(result.outcome, ValidatedProposal)
async def test_t2b_validated_then_revise_on_every_attempt_never_hits_an_assert(project) -> None:
"""The loop's EXIT CONTRACT. Detach point: reverting to ``assert last is not None`` (M29).
Before MAJOR-2 the exit rested on ``last``, which only a validator REJECTION sets — so
"validated -> revise" on every attempt would reach the end of the loop with ``last is None``
and die on the assert with a traceback (and, under ``-O``, on ``None.proposal``). The carrier
is explicit now, and the last ruling is what the run gets."""
first, revised, _ = _replies(project)
client = _FeedbackAwareChatClient("never-appears", first, revised)
reviews: list[pr.ProposalReview] = []
result = await generate_via_llm(
client,
project,
"ctx",
_meter(),
max_attempts=3,
reviewer=_RecordingReviewer([pr.ProposalReviewDecision.revise("again")]),
reviews=reviews,
)
assert isinstance(result.outcome, ValidatedProposal)
assert result.outcome.proposal.claimed_saving_nok == _FIRST_CLAIM
assert result.refinements == () # no validator rejection was ever fed back
async def test_t3a_the_round_ledger_and_not_max_attempts_stops_the_revisions(project) -> None:
"""T3a — a revise costs a ROUND, and the shared ledger is often what binds. Detach point:
``attempts_remaining`` computed from ``max_attempts`` alone (M38).
With ``max_rounds=2`` and ``max_attempts=10`` the loop must stop after two fetches and return
the last validated ruling. Under an attempt-only ``remaining`` the second attempt would buy a
third and the meter would raise ``BudgetExceeded(rounds, 2, 3)`` — so "no exception, exactly
two calls" is the discriminator, not a mere count."""
first, revised, _ = _replies(project)
client = _FeedbackAwareChatClient("never-appears", first, revised)
reviews: list[pr.ProposalReview] = []
result = await generate_via_llm(
client,
project,
"ctx",
_meter(rounds=2),
max_attempts=10,
reviewer=_RecordingReviewer([pr.ProposalReviewDecision.revise("again")]),
reviews=reviews,
)
assert client.call_count == 2
assert [r.honoured for r in reviews] == [True, False]
assert isinstance(result.outcome, ValidatedProposal)
async def test_t3b_a_revise_whose_follow_up_never_fetched_is_not_honoured(project) -> None:
"""T3b — ``honoured`` means **the attempt this revise bought actually FETCHED a reply**, not
"was bought". Detach point: setting ``honoured=True`` at revise time (M40).
The bought attempt's reply does not parse, so the inner parse-retry ticks the ledger again and
``BudgetExceeded`` fires before ``_fetch_parsed`` ever returns. The expert asked for a change
the run never made, and the artefact must be able to say so — beside ``rounds limit=2
observed=3`` (``observed != limit``, the kø-(y) rule)."""
first, revised, _ = _replies(project)
client = FakeChatClient(scripted=[first, "not json at all"], default_reply="still not json")
reviews: list[pr.ProposalReview] = []
failures: list[ParseFailure] = []
with pytest.raises(BudgetExceeded) as excinfo:
await generate_via_llm(
client,
project,
"ctx",
_meter(rounds=2),
max_attempts=10,
parse_failures=failures,
reviewer=_RecordingReviewer([pr.ProposalReviewDecision.revise(_FEEDBACK_SENTINEL)]),
reviews=reviews,
)
assert (excinfo.value.kind, excinfo.value.limit, excinfo.value.observed) == ("rounds", 2, 3)
assert [(r.decision, r.feedback, r.honoured) for r in reviews] == [
("revise", _FEEDBACK_SENTINEL, False)
]
async def test_t4_a_validator_rejected_attempt_is_never_shown_to_the_expert(project) -> None:
"""T4 — only a ``ValidatedProposal`` reaches the reviewer. A rejection is already fed back
INFORMED (Step 5); asking a human to comment on numbers the machine just refuted spends the
human on the machine's job. Detach point: calling the reviewer before ``validate_proposal``
(M6)."""
_, _, bad = _replies(project)
client = FakeChatClient(scripted=[bad], default_reply=bad)
reviewer = _RecordingReviewer([pr.ProposalReviewDecision.revise("x")])
reviews: list[pr.ProposalReview] = []
result = await generate_via_llm(
client, project, "ctx", _meter(), max_attempts=1, reviewer=reviewer, reviews=reviews
)
assert reviewer.calls == []
assert reviews == []
assert isinstance(result.outcome, Rejection)
assert result.refinements == ()
async def test_t5_the_request_carries_this_attempts_candidate_and_its_context(project) -> None:
"""T5 — what the expert is shown. Detach points: a constant ``attempts_remaining`` (M38), a
dropped approach key (M10), a dropped checker verdict.
``attempts_remaining`` is the LEDGER-aware number: with ``max_rounds=1`` the ledger has no
headroom after attempt 1, so a revise can buy nothing even though ``max_attempts=3`` — an
attempt-only computation would show ``2`` and read false at the terminal."""
first, revised, _ = _replies(project)
client = _FeedbackAwareChatClient("never-appears", first, revised)
reviewer = _RecordingReviewer([pr.ProposalReviewDecision.approve()])
await generate_via_llm(
client,
project,
"ctx",
_meter(rounds=1),
max_attempts=3,
reviewer=reviewer,
reviews=[],
review_key=("a2", "Night setback"),
checker_verdict="approve",
)
(request,) = reviewer.calls
assert request.project_id == project.id
assert (request.approach_id, request.approach_label) == ("a2", "Night setback")
assert request.attempt == 0
assert request.attempts_remaining == 0 # the LEDGER binds, not max_attempts
assert request.checker_verdict == "approve"
assert isinstance(request.proposal, ValidatedProposal)
assert request.proposal.proposal.claimed_saving_nok == _FIRST_CLAIM
async def test_t6_feedback_is_sticky_while_the_machines_reason_is_per_attempt(project) -> None:
"""T6 — composition. Detach points: clearing the feedback after one attempt (M4), or
accumulating rejections (M5).
A1 validates -> the expert asks for a revision. A2 is REJECTED by the validator (and the
expert is NOT asked about it, T4). A3's prompt must carry BOTH the expert's standing
instruction and A2's fresh reason — the human's words stand until the human next answers,
the machine's reason is only the most recent."""
first, revised, bad = _replies(project)
client = FakeChatClient(scripted=[first, bad, revised], default_reply=revised)
reviewer = _RecordingReviewer(
[pr.ProposalReviewDecision.revise(_FEEDBACK_SENTINEL), pr.ProposalReviewDecision.approve()]
)
reviews: list[pr.ProposalReview] = []
result = await generate_via_llm(
client, project, "ctx", _meter(), max_attempts=3, reviewer=reviewer, reviews=reviews
)
reason = f"{_BAD_CLAIM}"
third = " ".join(client.received_texts[2])
assert _FEEDBACK_SENTINEL in " ".join(client.received_texts[1])
assert third.count(_FEEDBACK_SENTINEL) == 1
assert "REJECTED by the deterministic validator" in third
assert third.count(reason) == 1
assert len(reviewer.calls) == 2 # A2 was never shown
assert isinstance(result.outcome, ValidatedProposal)
async def test_t7_a_revise_with_nothing_left_to_buy_leaves_the_validated_ruling_standing(
project,
) -> None:
"""T7 / criterion 12 — an un-honoured revise buys nothing, so the last ruling (the validated
one) stands. Detach point: ``honoured`` constant ``True`` (M9).
The record must be able to STATE the fact rather than leave it to be inferred: the expert
asked for a change the run could not make."""
first, revised, _ = _replies(project)
client = _FeedbackAwareChatClient(_FEEDBACK_SENTINEL, first, revised)
reviews: list[pr.ProposalReview] = []
result = await generate_via_llm(
client,
project,
"ctx",
_meter(),
max_attempts=1,
reviewer=_RecordingReviewer([pr.ProposalReviewDecision.revise(_FEEDBACK_SENTINEL)]),
reviews=reviews,
)
assert client.call_count == 1 # nothing further was generated
assert isinstance(result.outcome, ValidatedProposal)
assert result.outcome.proposal.claimed_saving_nok == _FIRST_CLAIM
assert [(r.decision, r.honoured) for r in reviews] == [("revise", False)]
async def test_t13_an_honoured_revise_rejected_next_leaves_the_validators_last_ruling(
project,
) -> None:
"""Criterion 13 / D6 — the validator's LAST ruling wins, and the reviewed candidate is not
lost. Detach points: falling back to the earlier validated proposal (M27), dropping the
reviewed proposal from the record (M28).
Falling back would hand the run the very candidate the expert asked to change — a silent
override of a human decision — and would stamp ``validator_decision`` for a candidate that
was not the last one ruled on."""
first, _, bad = _replies(project)
client = FakeChatClient(scripted=[first, bad], default_reply=bad)
reviews: list[pr.ProposalReview] = []
result = await generate_via_llm(
client,
project,
"ctx",
_meter(),
max_attempts=2,
reviewer=_RecordingReviewer([pr.ProposalReviewDecision.revise(_FEEDBACK_SENTINEL)]),
reviews=reviews,
)
assert isinstance(result.outcome, Rejection)
assert result.outcome.proposal.claimed_saving_nok == _BAD_CLAIM
(record,) = reviews
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"
def _debate_entries(sink: list[str]) -> list[str]:
"""Everything in the sink that is NOT a generation prompt — i.e. the debate's turns.
The complement of ``_GENERATION_MARK`` rather than a positive debate marker, because the claim
being guarded is exactly "nothing that is not generation moved": a positive marker would let a
turn the classifier does not know about drift unseen. Every "N generation prompts" assertion in
this file is paired with an equality of THIS list between the treated and the control run
(SC1/SC4) — without the pairing, an arm that counted a debate turn as a generation prompt, or a
door that perturbed the debate, would read as a correct generation count."""
return [blob for blob in sink if _GENERATION_MARK not in blob]
# 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.
The per-candidate attempt indices below are a GENERATION count (two attempts each), so SC1
requires them paired with the debate's count: a control run whose reviewer always approves must
leave the non-generation entries identical."""
mandate = Mandate(
objective="Cut energy cost without rebuilding.",
approaches=(_A1, _A2),
allow_own_proposals=True,
)
reviewer = _RunReviewer()
treated_sink: list[str] = []
result = await _run_with(
outbox_dir=tmp_path / "outbox",
reviewer=reviewer,
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(result, RunResult) and isinstance(control, RunResult)
assert _debate_entries(treated_sink), "no debate turns reached the sink: the pairing is vacuous"
assert _debate_entries(treated_sink) == _debate_entries(control_sink)
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 finished in (treated, control):
assert [row.status for row in finished.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))
# SC1's pairing: the counts above mean "one more GENERATION prompt" only if the debate did not
# move. The non-empty check is the vacuity guard — two empty lists are equal for free.
assert _debate_entries(treated_sink), "no debate turns reached the sink: the pairing is vacuous"
assert _debate_entries(treated_sink) == _debate_entries(control_sink)
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)
]
class _WriteRefused(OSError):
"""A disk that will not take the artefact. Its own subclass so the control arm can assert that
THIS error escaped, rather than that some ``OSError`` did."""
def _refuse_writer(monkeypatch: pytest.MonkeyPatch, name: str) -> None:
def _raise(*_a: Any, **_k: Any) -> None:
raise _WriteRefused(f"{name} refused")
monkeypatch.setattr(run.outbox, name, _raise)
def _budget_stop_select(blob: str, _role: str) -> str:
"""T9's shape: the bought attempt's reply never parses, the retry exhausts the round ledger,
and ``BudgetExceeded`` leaves ``run_project`` as an exception with BOTH ``finally`` writers
armed (parse failures exist, and a reviewer was given)."""
if _GENERATION_MARK not in blob:
return "ok"
return "not json at all" if _FEEDBACK_SENTINEL in blob else _run_reply("Belysning", 30_000)
async def test_a_write_error_never_displaces_the_stop_that_was_already_in_flight(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
"""The review writer runs from a ``finally`` on EVERY reviewed run, so an ``OSError`` from it
lands while ``BudgetExceeded`` (or ``ProposalReviewInputError``) is propagating — and a bare
call there REPLACES the in-flight exception. Neither ``except ProposalReviewInputError`` nor
the CLI's refusal tuple catches ``OSError``, so the operator got a traceback instead of the
``run stopped:`` line, and the reason the run ended was gone.
Detach point: dropping the ``in_flight`` guard around the writers. The stop must survive, and
the write failure must still be SAID — swallowed silently, a missing artefact would look like
a run that had nothing to record (T10/T11's whole distinction)."""
_refuse_writer(monkeypatch, "write_proposal_reviews")
with pytest.raises(BudgetExceeded):
await _run_with(
outbox_dir=tmp_path / "outbox",
reviewer=_RunReviewer(),
select=_budget_stop_select,
max_rounds=2,
)
err = capsys.readouterr().err
assert f"{_RUN_ID}-proposal-reviews.json" in err
assert "write_proposal_reviews refused" in err
async def test_the_parse_failure_writer_carries_the_same_guard(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
"""The SIBLING writer in the same ``finally``. It has the identical shape and runs only when a
parse failed — rarer, never absent — so it gets the same guard and its own witness: one
helper, two call sites, and a change to only one of them is what a shared seam exists to
prevent (kø-(p))."""
_refuse_writer(monkeypatch, "write_parse_failures")
with pytest.raises(BudgetExceeded):
await _run_with(
outbox_dir=tmp_path / "outbox",
reviewer=_RunReviewer(),
select=_budget_stop_select,
max_rounds=2,
)
assert f"{_RUN_ID}-parse-failures.json" in capsys.readouterr().err
async def test_a_write_error_with_nothing_in_flight_is_still_an_error(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""The CONTROL, and the reason the guard is conditional rather than a blanket ``except``. On a
run that otherwise SUCCEEDED there is no stop reason to protect, so a disk that refused the
artefact must reach the caller: 'the outbox is unwritable' silently downgraded to a clean
return is how a run reports success it cannot evidence.
Without this arm the guard could be mutated to swallow unconditionally and stay green."""
_refuse_writer(monkeypatch, "write_proposal_reviews")
with pytest.raises(_WriteRefused):
await _run_with(outbox_dir=tmp_path / "outbox", reviewer=_RunReviewer())
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
# ---------------------------------------------------------------------------------------------
# Group C (Step 5) — the terminal door, driven directly with in-memory streams.
# ---------------------------------------------------------------------------------------------
def _request(
*, attempts_remaining: int = 2, label: str = "Night setback"
) -> pr.ProposalReviewRequest:
return pr.ProposalReviewRequest(
project_id=_PID,
approach_id="a2",
approach_label=label,
attempt=0,
attempts_remaining=attempts_remaining,
proposal=_validated(p50=17_500.0),
checker_verdict="approve",
)
def _door(text: str) -> tuple[pr.ProposalReviewer, io.StringIO]:
out = io.StringIO()
return pr.terminal_proposal_reviewer(stream_in=io.StringIO(text), stream_out=out), out
def test_t14_the_candidate_is_shown_as_text_and_never_as_a_repr() -> None:
"""T14 — BLOCKER-1's lesson, applied at construction. Detach point: rendering the request or
the proposal with ``str()`` (M25).
An expert who answers ``approve`` on ``<...ValidatedProposal object at 0x10d...>`` has signed
blindly. Both halves are asserted: a POSITIVE sentinel only the text path can emit (the
measure, the claimed saving, the percentile), and the NEGATIVE shape of the defect itself —
the positive alone would be satisfied by a renderer that prints nothing."""
door, out = _door("approve\n")
door(_request())
rendered = out.getvalue()
assert "LED-retrofit" in rendered # the measure, from the typed IR
assert "30000" in rendered # the claimed saving
assert "17500" in rendered # the validator's p50
assert "Night setback" in rendered # WHICH candidate is being asked about
assert "attempts remaining: 2" in rendered
assert "checker: approve" in rendered
assert "approve" in rendered and "revise" in rendered
assert _REPR_LEAK not in rendered
def test_t15_anything_outside_the_closed_vocabulary_is_asked_again() -> None:
"""T15 — fail-closed on the expert's OWN input. Detach point: treating any non-revise line as
an approval (M17).
Three things are re-asked and none of them is a decision: a typo, a blank line, and a bare
``revise`` (which says nothing to revise). Validation, never repair."""
door, out = _door("yes please\n\nrevise\nrevise F\napprove\n")
first = door(_request())
second = door(_request())
assert first == pr.ProposalReviewDecision.revise("F")
assert second == pr.ProposalReviewDecision.approve()
rendered = out.getvalue()
assert rendered.count("PROPOSAL REVIEW") == 2
assert rendered.count("Not an answer:") == 3
def test_the_door_refuses_a_revise_it_cannot_buy_and_says_why() -> None:
"""D1(a) — the THIRD door state: a re-ask, not a third word. Detach point: letting the door
return a revise with nothing left to buy.
The line states a FACT ("no attempts remain for this candidate; answer approve") rather than
offering a remedy this CLI cannot perform — there is no ``--max-attempts`` flag, and D1 adds
none. The expert still SEES the candidate, so silence is never read as consent."""
door, out = _door("revise F\napprove\n")
decision = door(_request(attempts_remaining=0))
assert decision == pr.ProposalReviewDecision.approve()
assert "no attempts remain for this candidate; answer approve" in out.getvalue()
def test_t16_end_of_input_is_never_a_sign_off() -> None:
"""T16 (unit) — EOF raises. Detach point: reading end of input as an approval (M15).
Reading silence as approval would let a run carry a candidate nobody signed for, and do it
invisibly. The message names the flag and the condition so the CLI's ``run stopped:`` line can
print it verbatim."""
door, _ = _door("")
with pytest.raises(pr.ProposalReviewInputError) as excinfo:
door(_request())
assert "end of input" in str(excinfo.value)
assert "--proposal-review" in str(excinfo.value)
def test_t20_the_streams_are_resolved_at_call_time(monkeypatch: pytest.MonkeyPatch) -> None:
"""T20 — the ``shared_root()`` idiom, F4's measurement. Detach point: capturing
``sys.stdin``/``sys.stdout`` at construction (M21).
A closure that bound the streams when the factory ran could not be driven by a caller — or a
test — that replaces them afterwards, and the only way left to exercise the door would be a
subprocess. The reviewer is built FIRST, the streams swapped AFTER, and only then called."""
door = pr.terminal_proposal_reviewer()
out = io.StringIO()
monkeypatch.setattr(sys, "stdin", io.StringIO("revise later\n"))
monkeypatch.setattr(sys, "stdout", out)
assert door(_request()) == pr.ProposalReviewDecision.revise("later")
assert "PROPOSAL REVIEW" in out.getvalue()
# ---------------------------------------------------------------------------------------------
# Group C (Step 6) — the CLI door: the flag, the four refusals, the ``run stopped:`` channel and
# the notice. The two door arms run in a CHILD (P4: stdin/stdout/rc are answered in a subprocess,
# never with ``capsys``); the refusal arms run in-process with the model factory RAISING, because
# at the exit code a refusal after the spend looks exactly like a refusal before it.
# ---------------------------------------------------------------------------------------------
_CLI_FEEDBACK = f"Bruk 40000, ikke 30000. {_FEEDBACK_SENTINEL}"
_CLI_MEASURE = "Behovsstyrt belysning i fellesarealer"
def _cli_replies_file(
tmp_path: Path, *, claims: tuple[int, ...] = (30_000, 40_000), name: str = "replies.json"
) -> str:
"""A TWO-ENTRY proposer step list. A single constant string would make attempt 2 byte-identical
to attempt 1 — the door would go green while proving nothing about the answer being used."""
path = tmp_path / name
path.write_text(
json.dumps(
{
"proposer": [_run_reply(_CLI_MEASURE, claim) for claim in claims],
"checker": "VERDICT: APPROVE",
}
),
encoding="utf-8",
)
return str(path)
def _cli_argv(tmp_path: Path, *, outbox: str, replies: str | None = None) -> list[str]:
return [
_RUN_PID,
"--docs-dir",
str(_BUNDLE_DIR),
"--bundle-dir",
str(_BUNDLE_DIR),
"--scripted-replies",
replies or _cli_replies_file(tmp_path),
"--proposal-review",
"--outbox-dir",
str(tmp_path / outbox),
"--run-id",
_RUN_ID,
]
def _child(argv: list[str], *, stdin: str) -> subprocess.CompletedProcess[str]:
return subprocess.run(
[sys.executable, "-m", "portfolio_optimiser.run", *argv],
input=stdin,
capture_output=True,
text=True,
cwd=str(_REPO),
)
def _refuse_model(monkeypatch: pytest.MonkeyPatch) -> None:
"""The CLI's ONE injection point (``test_run_cli_loadbearing``'s seam). A refusal that fires
after the spend is indistinguishable from one that fires before it at the exit code — økt 57's
rule — so every refusal arm asserts on ZERO model clients, not only on rc 1."""
def _raise(_profile: Any) -> Any:
raise AssertionError("a model client was built on a path that must make no model calls")
monkeypatch.setattr(run, "_default_factory", _raise)
def test_t13_the_flag_answers_the_review_from_a_real_argv_and_the_answer_is_used(
tmp_path: Path,
) -> None:
"""T13 — the CLI door, end to end in a CHILD. Detach points: the argparse flag (M20), the
reviewer wiring at the dispatch, the sink, the artefact.
The child settles three things no in-process arm can: that the flag exists on the parsed
command line, that a real pipe reaches the reviewer, and that stdout renders the candidate as
TEXT. The proposer's second step carries a DIFFERENT amount, so "the answer was used" is the
outcome, not the presence of a record.
(The plan's separate T19 — "a bare, unscripted flag parse" — is folded in here: without a
script that argv would call a live model, so the argparse witness is this child's own
``unrecognized arguments`` assertion.)
**The attempt indices are SC4's count PROXY at this door, and nothing more — stated rather
than left to be read as a prompt-level count.** The child runs in its own interpreter, so the
``ScriptedChatClient`` sink lives in that process and there is no CLI flag that dumps it: the
prompt-level half of SC4 (the feedback VERBATIM in the second generation prompt, and the
debate's turn count unchanged against a control) is measured in-process, where the sink is
reachable — T1 for the verbatim injection, T5-run and T8 for the debate pairing. What this arm
settles that none of those can is that a real argv, a real pipe and a real ``run_project``
bought a second attempt whose OUTCOME differs (40 000, entry 2 of the step list)."""
proc = _child(_cli_argv(tmp_path, outbox="outbox"), stdin=f"revise {_CLI_FEEDBACK}\napprove\n")
assert "unrecognized arguments" not in proc.stderr, proc.stderr
assert "Traceback" not in proc.stderr, proc.stderr
assert proc.returncode == 0, proc.stderr
payload = _reviews_on_disk(tmp_path / "outbox")
assert [(r["decision"], r["honoured"]) for r in payload["reviews"]] == [
("revise", True),
("approve", True),
]
assert payload["reviews"][0]["feedback"] == _CLI_FEEDBACK # VERBATIM
# SC4's count proxy at this door (see the docstring): two attempt indices, not two counted
# prompts — the child's sink is in the child.
assert [r["attempt"] for r in payload["reviews"]] == [0, 1]
outcome = json.loads(
(tmp_path / "outbox" / f"{_RUN_ID}-proposal.json").read_text(encoding="utf-8")
)
assert outcome["proposal"]["claimed_saving_nok"] == 40_000 # entry 2 — the answer was USED
assert _CLI_MEASURE in proc.stdout # the candidate was rendered as TEXT
assert _REPR_LEAK not in proc.stdout
assert "proposal review: 2 answer(s)" in proc.stdout
def test_t16_input_that_ends_without_an_answer_stops_the_run_on_its_own_channel(
tmp_path: Path,
) -> None:
"""T16 — EOF at the CLI. Detach points: reading EOF as approve (M15), not catching
``ProposalReviewInputError`` by name (M16).
``run stopped:`` is a DISTINCT channel from ``run refused:``: the argv was fine and the run
had already spent tokens, so "refused" would mislabel it — and a ``ValueError``-shaped error
would sit one frame from ``_fetch_parsed``'s catch-all and be captured as a *parse failure*
instead. Both are asserted: no traceback, and no ``{run_id}-parse-failures.json``.
T13's run (same fixture, same flags, a SEPARATE outbox) is the positive control that
``-outcome.json`` CAN appear when the door is answered."""
control = _child(
_cli_argv(tmp_path, outbox="answered"), stdin=f"revise {_CLI_FEEDBACK}\napprove\n"
)
assert control.returncode == 0, control.stderr
assert (tmp_path / "answered" / f"{_RUN_ID}-outcome.json").exists()
proc = _child(_cli_argv(tmp_path, outbox="silent"), stdin="")
assert proc.returncode == 1
assert "Traceback" not in proc.stderr, proc.stderr
assert "run refused:" not in proc.stderr
assert "run stopped:" in proc.stderr
assert "--proposal-review" in proc.stderr
assert "end of input" in proc.stderr
silent = tmp_path / "silent"
assert not (silent / f"{_RUN_ID}-outcome.json").exists()
assert not (silent / f"{_RUN_ID}-parse-failures.json").exists()
# The ``finally`` still wrote the record, and it is empty of decisions.
assert _reviews_on_disk(silent) == {"reviews": [], "run_id": _RUN_ID}
def test_t17_the_door_belongs_to_single_project_mode(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
"""T17 — the documented partition. Detach point: dropping the ``single_only`` row (M18).
Concurrent portfolio waves would share ONE terminal and interleave prompts from several
projects. The assert names ``--portfolio``, never the shared ``--proposal-review``: a dropped
row falls through to a neighbouring refusal that names the flag too, and asserting on the
shared token would stay green against no partition entry at all (økt 57's own mutation)."""
_refuse_model(monkeypatch)
assert run.main(["--portfolio", "--proposal-review"]) == 1
assert "--portfolio" in capsys.readouterr().err
def test_t18_the_door_is_refused_in_report_mode(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
"""T18 — report mode returns ABOVE every dispatch, so a flag missing from
``report_forbidden`` is a silent DROP, not a refusal (the F4 gap). Detach point: dropping the
row (M19).
Run against an argv report mode would otherwise ACCEPT, with the rc-0 control proving it —
otherwise rc 1 could come from the missing ``--ledger`` rather than from this row."""
_refuse_model(monkeypatch)
ledger = tmp_path / "ledger.json"
# A JSON ARRAY: ``SavingsLedger.load`` refuses a dict, and an rc-1 from THAT would make the
# arm green for the wrong reason — which is precisely what the rc-0 control below catches.
ledger.write_text(json.dumps([]), encoding="utf-8")
assert run.main(["--report", "--ledger", str(ledger)]) == 0
capsys.readouterr()
assert run.main(["--report", "--ledger", str(ledger), "--proposal-review"]) == 1
assert "mode-exclusive" in capsys.readouterr().err
def test_the_door_and_a_dry_run_contradict(
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
"""Detach point: dropping the ``--live-dry-run`` refusal (M30).
The dry-run cut returns ABOVE generation, so no candidate ever reaches a reviewer — the flag
would be accepted and then silently inert. Asserts on the PARTNER token."""
_refuse_model(monkeypatch)
rc = run.main(
[
_RUN_PID,
"--docs-dir",
str(_BUNDLE_DIR),
"--bundle-dir",
str(_BUNDLE_DIR),
"--live-dry-run",
"--proposal-review",
]
)
assert rc == 1
assert "--live-dry-run" in capsys.readouterr().err
def test_the_door_has_nothing_to_answer_under_proposals_from_mandate(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
"""Detach point: dropping the ``--proposals-from-mandate`` refusal (M31).
That mode is SYNC by construction — it settles the commission against the derived baseline and
returns at the terminal dispatch, above any generation — so the flag would exit 0 having asked
nobody anything. The rc-0 control (same argv without the flag) proves the refusal is this row
and not some other precondition."""
_refuse_model(monkeypatch)
mandate_path = tmp_path / "mandate.json"
mandate_path.write_text(
Mandate(
objective="Finn kostnadsbesparelser i K2",
approaches=(
Approach(
id="a1",
label="Redusert sprengningsvolum i sone A",
affected_codes=("21.1",),
claimed_saving_nok=200_000.0,
),
),
allow_own_proposals=False,
).model_dump_json(),
encoding="utf-8",
)
priced = str(_REPO / "tests" / "fixtures" / "k2-prisskjema-SYNTETISK")
argv = [
"K2",
"--docs-dir",
priced,
"--bundle-dir",
priced,
"--derive-cost-baseline",
"--proposals-from-mandate",
"--mandate",
str(mandate_path),
]
assert run.main(argv) == 0
capsys.readouterr()
assert run.main([*argv, "--proposal-review"]) == 1
assert "--proposals-from-mandate" in capsys.readouterr().err
def test_the_door_and_a_parked_exploration_contradict(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
"""The FOURTH return above generation. Detach point: dropping the ``--checkpoint-dir``
refusal (M39).
A parked exploration returns on the park leg before any candidate exists, so the flag would be
accepted and never used. The refusal names ``--resume``, which is where the door DOES compose
— a refusal that only forbids leaves the operator without the door that works."""
_refuse_model(monkeypatch)
rc = run.main(
[
_RUN_PID,
"--docs-dir",
str(_BUNDLE_DIR),
"--bundle-dir",
str(_BUNDLE_DIR),
"--explore",
"Find the cheapest saving.",
"--checkpoint-dir",
str(tmp_path / "checkpoints"),
"--outbox-dir",
str(tmp_path / "outbox"),
"--run-id",
_RUN_ID,
"--proposal-review",
]
)
assert rc == 1
err = capsys.readouterr().err
assert "--checkpoint-dir" in err
assert "--resume" in err
#: Day 1's parked exploration, driven exactly as an operator drives it: a real ``--explore`` leg
#: with ``enable_plan_review`` armed and NO reviewer at the terminal, so the loop parks. The
#: proposal-review door is deliberately absent here — that argv is the refusal this arm's sibling
#: (``…and_a_parked_exploration_contradict``) keeps green.
_RESUME_RUN_ID = "proposal-review-resume"
_RESUME_MANAGER_REPLY = json.dumps(
{
"is_request_satisfied": {"reason": "r", "answer": True},
"is_in_loop": {"reason": "r", "answer": False},
"is_progress_being_made": {"reason": "r", "answer": True},
"next_speaker": {"reason": "r", "answer": "hypothesiser"},
"instruction_or_question": {"reason": "r", "answer": "go"},
}
)
def _resume_replies_file(tmp_path: Path) -> str:
"""Both time-scales' roles in one file: the exploration's three and the debate's two. A
missing ``navigator`` key is the ``KeyError`` M40 of økt 64 — the resume leg builds an
exploration too, so it needs the same roster the park leg did."""
path = tmp_path / "resume-replies.json"
path.write_text(
json.dumps(
{
"proposer": _run_reply(_CLI_MEASURE, 30_000),
"checker": "VERDICT: APPROVE",
"manager": _RESUME_MANAGER_REPLY,
"navigator": "NAVIGATOR: read the index.",
"hypothesiser": "HYPOTHESIS: "
+ json.dumps(
{
"label": "Night setback",
"rationale": "y",
"requirement": None,
"why_none": "scripted",
}
),
}
),
encoding="utf-8",
)
return str(path)
def _resume_config_file(tmp_path: Path) -> str:
path = tmp_path / "resume-exploration.json"
path.write_text(
json.dumps(
{
"max_rounds": 4,
"max_tokens": 200_000,
"max_stall_count": 2,
"max_reset_count": 1,
"max_plan_revisions": 2,
"enable_plan_review": True,
}
),
encoding="utf-8",
)
return str(path)
def _park_then_answer(tmp_path: Path) -> None:
"""Day 1 and the days between: park a real plan review, then drop the expert's answer in the
inbox. Both halves go through the shipped surfaces — a hand-written question file would test
the fixture, not the door."""
outbox = tmp_path / "outbox"
assert (
run.main(
[
_RUN_PID,
"--docs-dir",
str(_BUNDLE_DIR),
"--bundle-dir",
str(_BUNDLE_DIR),
"--explore",
"Find the cheapest saving.",
"--explore-config",
_resume_config_file(tmp_path),
"--scripted-replies",
_resume_replies_file(tmp_path),
"--outbox-dir",
str(outbox),
"--checkpoint-dir",
str(tmp_path / "checkpoints"),
"--run-id",
_RESUME_RUN_ID,
]
)
== 0
)
question = json.loads(
(outbox / f"{_RESUME_RUN_ID}-plan-review.json").read_text(encoding="utf-8")
)
inbox = tmp_path / "review-inbox"
inbox.mkdir(parents=True, exist_ok=True)
(inbox / f"{_RESUME_RUN_ID}-plan-review-answer.json").write_text(
json.dumps(
{
"run_id": _RESUME_RUN_ID,
"request_id": question["request_id"],
"decision": "approve",
}
),
encoding="utf-8",
)
def test_the_door_composes_with_resume(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""A3, and the WITNESS the review found missing: the ``--checkpoint-dir`` refusal points the
operator at ``--resume``, and ``--resume`` itself REQUIRES ``--checkpoint-dir`` — so before
this arm the recommended composition was refused by the very message that recommended it, and
nothing in the suite could see it. Detach point: reverting the guard to a bare
``if args.checkpoint_dir is not None``.
The argv is a REAL resume: day 1 parks a real plan review through ``run.main``, the expert's
answer is dropped in a real inbox, and day N passes ``--resume … --checkpoint-dir …
--review-inbox … --proposal-review``. The arm it replaced sent none of those three flags and
was a plain single-project run T13 already covered — green against the defect it was named
for. ``run_project`` is a recorder, so what is measured is the WIRING at the fall-through, and
``_refuse_model`` proves the recorded call is the ONLY place a model would have been reached
from."""
_park_then_answer(tmp_path)
calls: list[dict[str, Any]] = []
async def _recorder(*args: Any, **kwargs: Any) -> Any:
calls.append(kwargs)
raise ValueError("recorded")
monkeypatch.setattr(run, "run_project", _recorder)
_refuse_model(monkeypatch)
run.main(
[
_RUN_PID,
"--docs-dir",
str(_BUNDLE_DIR),
"--bundle-dir",
str(_BUNDLE_DIR),
"--scripted-replies",
_resume_replies_file(tmp_path),
"--outbox-dir",
str(tmp_path / "outbox"),
"--checkpoint-dir",
str(tmp_path / "checkpoints"),
"--review-inbox",
str(tmp_path / "review-inbox"),
"--resume",
_RESUME_RUN_ID,
"--proposal-review",
]
)
assert len(calls) == 1, "the resume must fall through to the full-run dispatch, not be refused"
assert calls[0]["proposal_reviewer"] is not None
def test_a_reviewer_nobody_could_consult_says_so_on_stdout(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
"""The notice's zero case, at the CLI. Detach points: the renderer returning ``None`` on zero
with a reviewer present (M33), and the print itself (M37).
An operator who passed ``--proposal-review`` and sees nothing cannot tell "no candidate was
ever validated, so nobody was asked" from "the door hung"."""
monkeypatch.setattr(sys, "stdin", io.StringIO(""))
rc = run.main(
_cli_argv(
tmp_path,
outbox="outbox",
# BOTH attempts claim above the P90 cap (90 000), so the validator rejects each one
# and the reviewer is never reached — offered, never consulted.
# As many entries as ``max_attempts`` (3), so every attempt PARSES and is rejected
# by the deterministic gate. A shorter list would fall through to the selector's
# default reply, which never parses, and the round ledger would fire instead — the
# honesty limit ``_load_scripted_replies`` states, reproduced here by accident once.
replies=_cli_replies_file(
tmp_path, claims=(200_000, 200_000, 200_000), name="rejected.json"
),
)
)
assert rc == 0
assert "proposal review offered, never consulted" in capsys.readouterr().out
def test_a_run_with_no_reviewer_prints_no_review_line(
tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""CONTROL for the notice (M32): the renderer must return ``None`` when no reviewer was
offered, so a run that answered nobody says nothing."""
argv = [a for a in _cli_argv(tmp_path, outbox="outbox") if a != "--proposal-review"]
assert run.main(argv) == 0
assert "proposal review" not in capsys.readouterr().out
# ---------------------------------------------------------------------------------------------
# Group D (Steps 7 and 8) — the hosted refusal, and the library dispatcher.
# ---------------------------------------------------------------------------------------------
class _RunProjectRecorder:
def __init__(self) -> None:
self.calls: list[dict[str, Any]] = []
async def __call__(self, *args: Any, **kwargs: Any) -> Any:
self.calls.append(kwargs)
raise AssertionError("run_project must not be reached on a refused invocation")
async def test_t21_the_hosted_surface_refuses_the_door_by_name(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""T21 — the hosted refusal. Detach points: deleting the named refusal (M22), forwarding the
field instead (M23).
**The discriminator is the TEXT, not the status.** The whitelist already answers any unknown
field with ``400 unknown field(s): …``, so a detached named refusal would STILL return 400
naming the field — the arm would pass while the message said nothing about where the door
actually is. So the named message points at the CLI and at ``/readiness``, and the control
(an ordinary unknown field) asserts the generic message shares none of it.
Zero ``run_project`` calls, not merely a 400: at the status code a refusal after the spend
looks identical to one before it (økt 57's rule)."""
recorder = _RunProjectRecorder()
monkeypatch.setattr(hosting, "run_project", recorder)
payload = {"project_id": _RUN_PID, "docs_dir": str(_BUNDLE_DIR)}
with pytest.raises(hosting.InvocationRefused) as excinfo:
await hosting.invoke({**payload, "proposal_review": True})
message = str(excinfo.value)
assert "--proposal-review" in message
assert "/readiness" in message
assert recorder.calls == []
with pytest.raises(hosting.InvocationRefused) as generic:
await hosting.invoke({**payload, "outbox_dir": "/x"})
assert "--proposal-review" not in str(generic.value)
assert "outbox_dir" in str(generic.value)
def test_the_refused_name_enters_neither_half_of_the_whitelist() -> None:
"""The partition arm. Fase 4e's two asserts (every forwarded field IS a ``run_project``
parameter, every consumed field is NOT) must stay untouched, which is why this is a
PRE-whitelist check on the raw payload rather than a fourth list entry.
The F4 precedent is measured as NOT analogous: ``enable_plan_review`` is a NESTED key inside
the whitelisted ``explore_contract``, which is why it can be named at all."""
assert set(hosting._REFUSED_BY_NAME).isdisjoint(hosting._ALLOWED_FIELDS)
assert set(hosting._REFUSED_BY_NAME).isdisjoint(inspect.signature(run_project).parameters)
# ...and the parameter it corresponds to is a real one, so the refusal names a door that exists.
assert "proposal_reviewer" in inspect.signature(run_project).parameters
async def test_t22_the_dispatcher_threads_one_reviewer_into_every_base(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""T22 — the multi-base dispatcher gets its OWN witness (the S7a-3 rule: every door that opens
a base gets its own mutation, because an unwitnessed copy can regress alone). Detach point:
the dispatcher stops threading it (M36).
ONE object is threaded, asserted with ``is`` and not ``==``: the dispatch is SEQUENTIAL, so a
single terminal reviewer composes — and a fresh reviewer per base would be a different object
with identical behaviour, which ``==`` on a plain callable cannot tell apart (the identity
lesson ``test_multibase_loadbearing``'s store arm was corrected into).
The signature arm is paired with it for the Fase-4e reason: a ``**kwargs`` recorder cannot see
a name ``run_project`` does not actually take."""
calls: list[dict[str, Any]] = []
async def _recorder(project_id: str, profile: Any = "local", **kwargs: Any) -> Any:
calls.append({"project_id": project_id, **kwargs})
raise RuntimeError("recorded")
reviewer = _RunReviewer()
monkeypatch.setattr(run, "run_project", _recorder)
with pytest.raises(RuntimeError):
await run.run_mandate_across_bundles(
Mandate(
objective="o",
approaches=(
Approach(id="a", label="A", bundle_id="tunnel-hauglia"),
Approach(id="b", label="B", bundle_id="veglys-fv-soer"),
),
),
(str(_EXAMPLES / "tunnel-hauglia"), str(_EXAMPLES / "veglys-fv-soer")),
proposal_reviewer=reviewer,
)
assert calls, "the dispatcher must have reached run_project at least once"
assert all(c["proposal_reviewer"] is reviewer for c in calls)
assert "proposal_reviewer" in inspect.signature(run_project).parameters
def test_run_portfolio_takes_no_reviewer() -> None:
"""The ABSENCE arm (brief § Non-Goals). Concurrent portfolio waves share one terminal, which
is the ``--portfolio`` partition's own reason — so a later "symmetry" edit must be a RED test,
not a silent widening."""
assert "proposal_reviewer" not in inspect.signature(run.run_portfolio).parameters
def test_the_readme_block_names_every_flag_the_cli_refuses_the_door_with() -> None:
"""The README-claim arm, in the Fase-3 form (``test_public_surface_claims_loadbearing``):
a claim the PUBLISHED surface makes about itself, read as RAW TEXT, because prose is the one
thing no behavioural test can see.
The door has five partner refusals. A README block that documents the door but omits one of
them sends a stranger into a refusal the page said nothing about — the same drift the wheel
filename gate exists for, and the reason the handover gate matches names rather than prose.
The block is EXTRACTED (the ``--proposal-review`` prose block, from its bold heading to the
next bold heading) rather than substring-matched over the whole file, because ``--portfolio``,
``--report`` and ``--checkpoint-dir`` all appear elsewhere in the README for their own reasons
— a whole-file check would be green on exactly the prose it is supposed to protect.
This arm carries no M-number: the M-list closed at M40. It was driven RED TWICE, because
"the block is missing" and "the block is incomplete" are different failures and only the
second is what this gate is for. (i) Written and measured against the README as it stood
BEFORE the block was added — extraction found nothing, and ``assert block`` failed. (ii) With
the block in place, the ``--checkpoint-dir`` sentence was rewritten to describe the refusal
without NAMING the flag; the partner loop went red on its own. The control that the extractor
is not silently finding nothing is the ``--plan-review`` block at the end, which has existed
since F4.
"""
readme = (_REPO / "README.md").read_text(encoding="utf-8")
block = _readme_block(readme, "Answering the proposal review (`--proposal-review`)")
assert block, "the README must carry a --proposal-review prose block"
for partner in (
"--portfolio",
"--report",
"--live-dry-run",
"--proposals-from-mandate",
"--checkpoint-dir",
):
assert partner in block, f"the README block never names the refusal with {partner}"
assert "--resume" in block, "the --checkpoint-dir refusal must name the door that DOES work"
assert "OKF bundle" not in block, "customer-facing prose says knowledge base, never OKF bundle"
# CONTROL: the extractor finds a block that has existed since F4, so an empty result above is
# a missing block and not a broken extractor.
plan_block = _readme_block(readme, "Answering the plan review (`--plan-review`)")
assert plan_block and "enable_plan_review" in plan_block
def _readme_block(readme: str, heading: str) -> str:
marker = f"**{heading}.**"
start = readme.find(marker)
if start == -1:
return ""
nxt = readme.find("\n **", start + len(marker))
return readme[start : nxt if nxt != -1 else len(readme)]
Reference in a new issue View git blame Copy permalink