"""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" # 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 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)) 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 # --------------------------------------------------------------------------------------------- # 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.)""" 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 assert [r["attempt"] for r in payload["reviews"]] == [0, 1] # exactly two generation attempts 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 def test_the_door_composes_with_resume(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: """A3, VERIFIED rather than deferred: the resume block yields the parked exploration's mandate and falls through to the SAME full-run dispatch, so the reviewer built at the CLI is reached. Detach point: refusing ``--resume`` together with the door, or forgetting to pass the reviewer on that path. ``run_project`` is replaced by a recorder, so the arm measures the WIRING rather than a whole resumed exploration.""" 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", _cli_replies_file(tmp_path), "--outbox-dir", str(tmp_path / "outbox"), "--run-id", _RUN_ID, "--proposal-review", ] ) assert len(calls) == 1 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)]