portfolio-optimiser/tests/test_proposal_review_loop_loadbearing.py

"""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 io
import json
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
from portfolio_optimiser import proposal_review as pr
from portfolio_optimiser.budget import Budget, BudgetExceeded, TokenMeter
from portfolio_optimiser.generate import ParseFailure, _build_messages, generate_via_llm
from portfolio_optimiser.ir import AffectedItem, SavingsProposal
from portfolio_optimiser.mandate import OWN_PROPOSAL_ID, Approach, Mandate
from portfolio_optimiser.reference_domain import load_reference_projects
from portfolio_optimiser.run import RunResult, run_project
from portfolio_optimiser.simulation import ScriptedChatClient
from portfolio_optimiser.validator import Rejection, ValidatedProposal, proposal_for
from portfolio_optimiser.verdicts import VerdictStore

_REPO = Path(__file__).resolve().parents[1]
_BUNDLE_DIR = _REPO / "shared" / "examples" / "bygg-energi-mikro"
_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 run in (treated, control):
        assert [row.status for row in run.coverage].count("not_evaluated") == 0

    def _gen_prompts(sink: list[str], label: str) -> list[str]:
        return [b for b in sink if _GENERATION_MARK in b and label in b]

    assert len(_gen_prompts(treated_sink, _A2.label)) == 2  # the revise bought one more
    assert len(_gen_prompts(control_sink, _A2.label)) == 1
    assert _gen_prompts(treated_sink, _A3.label) == _gen_prompts(control_sink, _A3.label)
    assert all(_FEEDBACK_SENTINEL not in b for b in _gen_prompts(treated_sink, _A3.label))


async def test_t9_a_budget_stop_inside_generation_still_leaves_the_record(tmp_path: Path) -> None:
    """T9 — the artefact is written from a ``finally``. Detach points: writing after the return
    (M11), building the record from a RETURNED list instead of the caller-owned sink (M35).

    The bought attempt's reply does not parse, the parse-retry exhausts the round ledger, and
    ``BudgetExceeded`` leaves ``run_project`` as an exception. This is precisely the run whose
    record matters most, and precisely the one a return value cannot reach."""
    outbox = tmp_path / "outbox"

    def select(blob: str, _role: str) -> str:
        if _GENERATION_MARK not in blob:
            return "ok"
        return "not json at all" if _FEEDBACK_SENTINEL in blob else _run_reply("Belysning", 30_000)

    with pytest.raises(BudgetExceeded):
        await _run_with(outbox_dir=outbox, reviewer=_RunReviewer(), select=select, max_rounds=2)

    payload = _reviews_on_disk(outbox)
    assert payload["run_id"] == _RUN_ID
    assert [(r["decision"], r["feedback"], r["honoured"]) for r in payload["reviews"]] == [
        ("revise", f"{_FEEDBACK_SENTINEL}-None", False)
    ]


async def test_t10_a_reviewer_nobody_could_consult_still_writes_an_empty_record(
    tmp_path: Path,
) -> None:
    """T10 — written IFF a reviewer was given, INCLUDING when empty (D4). A reviewer that was
    offered and never consulted is a fact the artefact must be able to state; inferring it from an
    absent file would confuse it with T11's reviewer-less run."""
    outbox = tmp_path / "outbox"
    reviewer = _RunReviewer()
    result = await _run_with(
        outbox_dir=outbox,
        reviewer=reviewer,
        select=_keyed_selector(first_claim=200_000, revised_claim=200_000),  # always rejected
    )
    assert isinstance(result, RunResult)
    assert isinstance(result.outcome, Rejection)
    assert reviewer.calls == []
    assert _reviews_on_disk(outbox) == {"reviews": [], "run_id": _RUN_ID}


async def test_t11_a_run_with_no_reviewer_leaves_the_outbox_byte_identical(
    tmp_path: Path,
) -> None:
    """T11 — the CONTROL that keeps every existing run untouched. Detach point: writing the
    artefact with no reviewer (M13).

    Two reviewer-less runs into separate outbox directories: the same EXACT four names, and the
    same bytes. ``-proposal-reviews.json`` must be ABSENT — that absence is what makes T10's
    ``{"reviews": []}`` mean 'offered, never consulted' rather than 'not written'."""
    first, second = tmp_path / "a", tmp_path / "b"
    for outbox in (first, second):
        result = await _run_with(outbox_dir=outbox, reviewer=None)
        assert isinstance(result, RunResult)

    expected = [
        f"{_RUN_ID}-debate.json",
        f"{_RUN_ID}-outcome.json",
        f"{_RUN_ID}-proposal.json",
        f"{_RUN_ID}-runconfig.json",
    ]
    assert sorted(p.name for p in first.glob("*.json")) == expected
    assert not _reviews_artefact(first).exists()
    for name in expected:
        assert (first / name).read_bytes() == (second / name).read_bytes()


async def test_t12_an_approve_is_not_an_expert_verdict(tmp_path: Path) -> None:
    """T12 — ``approve`` mints NO ``Verdict``. Detach point: minting one (M14).

    F2 stands: a verdict arises only from ``--decision/--rationale`` or the Step-7 inbox.
    ``approve`` means "stop asking; I take it as is", which is a different act from judging the
    measure's worth — and the control proves a verdict IS still minted the F2 way, so the arm
    cannot pass merely because verdicts stopped working."""
    reviewed = await _run_with(outbox_dir=tmp_path / "reviewed", reviewer=_RunReviewer())
    assert isinstance(reviewed, RunResult)
    assert reviewed.expert_revisions  # the expert DID answer
    assert reviewed.verdict is None
    assert reviewed.verdict_key not in {v.id for v in reviewed.store.verdicts}

    judged = await _run_with(
        outbox_dir=tmp_path / "judged",
        run_id="run-judged",
        reviewer=_RunReviewer(),
        verdict_input=_VERDICT_INPUT,
    )
    assert isinstance(judged, RunResult)
    assert judged.verdict is not None
    assert judged.verdict.id == judged.verdict_key


async def test_the_record_carries_the_reviewed_candidates_key_and_p50(tmp_path: Path) -> None:
    """The payload arm: every entry is keyed the way an expert verdict on that candidate would
    be. Detach point: dropping ``verdict_key`` from the record (M28).

    The approve entry sits on the candidate the run ended up carrying, so its key must equal the
    run's own ``verdict_key`` — the join back into the Step-7 inbox channel."""
    outbox = tmp_path / "outbox"
    result = await _run_with(outbox_dir=outbox, reviewer=_RunReviewer())
    assert isinstance(result, RunResult)
    entries = _reviews_on_disk(outbox)["reviews"]
    assert [e["decision"] for e in entries] == ["revise", "approve"]
    assert entries[-1]["verdict_key"] == result.verdict_key
    assert entries[-1]["p50"] == pytest.approx(result.outcome.p50)
    assert entries[0]["verdict_key"] != entries[-1]["verdict_key"]  # a DIFFERENT candidate


async def test_the_review_artefact_is_invisible_to_the_hitl_registry(tmp_path: Path) -> None:
    """A RATCHET, green at construction (the B4 M4 precedent): ``hitl._read_outbox_proposals``
    globs ``*-proposal.json``, which cannot match ``*-proposal-reviews.json`` today. The arm
    guards a future rename that would make ``hitl pending`` read a review as a proposal."""
    outbox, inbox = tmp_path / "outbox", tmp_path / "inbox"
    inbox.mkdir()
    await _run_with(outbox_dir=outbox, reviewer=_RunReviewer())
    with_artefact = hitl.pending(str(outbox), str(inbox))
    _reviews_artefact(outbox).unlink()
    assert len(hitl.pending(str(outbox), str(inbox))) == len(with_artefact) == 1


# ---------------------------------------------------------------------------------------------
# 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()