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 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