portfolio-optimiser-claude/tests/test_checker_gate_loadbearing.py

"""Load-bearing: the checker gate (method-spec §3 Step 4, §11).

Either the checker actually gates, or the debate must not be called
maker-checker. RED-conditions proved here: the checker's surfaced output is
detached from the gate, OR its explicit REJECT no longer overrides an
otherwise-validated outcome. The gate is opt-in-reject (fail-open): APPROVE or
a missing marker never blocks; a validator rejection stands regardless of the
checker; the two falsifiers are recorded SEPARATELY (§9) — the provenance
field mirrors only the validator, the checker's decision is its own field.
"""

from __future__ import annotations

import json

from _scripted import ScriptedClient, reply

from portfolio_optimiser_claude.budget import BudgetMeter
from portfolio_optimiser_claude.contracts import TerminationContract
from portfolio_optimiser_claude.loop import (
    CheckerVerdict,
    apply_checker_gate,
    run_project,
)
from portfolio_optimiser_claude.validator import Rejection, ValidatedProposal

# 100 × 1000 = 100_000 NOK total; nominal feasible 30_000 == p90 (no band).
VALID_PROPOSAL = {
    "project_id": "p1",
    "measure": "led-retrofit",
    "affected_items": [{"code": "E01", "quantity": 100, "unit_cost": 1000}],
    "claimed_saving_nok": 25000,
}
# IR-valid (≤ total) but above p90 → the validator rejects it.
OVERCLAIMED_PROPOSAL = dict(VALID_PROPOSAL, claimed_saving_nok=90000)

VALIDATED = ValidatedProposal(
    validates=True,
    claimed_saving_nok=25000,
    nominal_feasible=30000,
    p10=30000,
    p50=30000,
    p90=30000,
)


def _meter() -> BudgetMeter:
    return BudgetMeter(TerminationContract(max_rounds=50, max_tokens=10_000))


class TestApplyCheckerGate:
    def test_explicit_reject_overrides_a_validated_outcome(self) -> None:
        # THE load-bearing seam: red if the override is detached.
        verdict = CheckerVerdict(decision="reject", reason="double-counted savings")
        outcome = apply_checker_gate(VALIDATED, verdict)
        assert isinstance(outcome, Rejection)
        assert outcome.reason.startswith("double-counted savings")

    def test_approve_never_blocks(self) -> None:
        verdict = CheckerVerdict(decision="approve", reason=None)
        assert apply_checker_gate(VALIDATED, verdict) is VALIDATED

    def test_absent_marker_never_blocks(self) -> None:
        # Fail-open: the validator remains the sole gate on such runs.
        verdict = CheckerVerdict(decision="absent", reason=None)
        assert apply_checker_gate(VALIDATED, verdict) is VALIDATED

    def test_a_validator_rejection_stands_regardless_of_the_checker(self) -> None:
        validator_rejection = Rejection(reason="claimed saving exceeds the feasible bound")
        verdict = CheckerVerdict(decision="approve", reason=None)
        assert apply_checker_gate(validator_rejection, verdict) is validator_rejection


class TestRunProjectGating:
    def test_checker_reject_flips_an_otherwise_validated_run(self) -> None:
        # Speiltest (sesjonsplan S8): numbers pass the validator, the checker's
        # surfaced REJECT flips the run outcome — red without the override.
        client = ScriptedClient(
            replies=[
                reply("reasoning"),
                reply("VERDICT: REJECT - assumptions unsupported"),
                reply(json.dumps(VALID_PROPOSAL)),
            ]
        )
        result = run_project(client, "context", meter=_meter(), max_debate_rounds=1)
        assert isinstance(result.outcome, Rejection)
        assert result.outcome.reason.startswith("assumptions unsupported")
        # §9: the two falsifiers are never conflated — the provenance mirror
        # says the NUMBERS passed; the checker's decision is its own field.
        assert result.validator_decision == "validated"
        assert result.checker_decision == "reject"

    def test_the_gate_consumes_the_checkers_last_surfaced_output(self) -> None:
        # Two debate rounds, two distinct REJECT reasons: the gate must carry
        # the LAST one — red if the surfaced-output seam is detached.
        client = ScriptedClient(
            replies=[
                reply("reasoning v1"),
                reply("VERDICT: REJECT - EARLY-REASON"),
                reply("reasoning v2"),
                reply("VERDICT: REJECT - FINAL-REASON"),
                reply(json.dumps(VALID_PROPOSAL)),
            ]
        )
        result = run_project(client, "context", meter=_meter(), max_debate_rounds=2)
        assert isinstance(result.outcome, Rejection)
        assert result.outcome.reason.startswith("FINAL-REASON")
        assert "EARLY-REASON" not in result.outcome.reason

    def test_an_approved_run_passes_through_ungated(self) -> None:
        client = ScriptedClient(
            replies=[
                reply("reasoning"),
                reply("sound. VERDICT: APPROVE"),
                reply(json.dumps(VALID_PROPOSAL)),
            ]
        )
        result = run_project(client, "context", meter=_meter(), max_debate_rounds=1)
        assert isinstance(result.outcome, ValidatedProposal)
        assert result.validator_decision == "validated"
        assert result.checker_decision == "approve"

    def test_a_missing_marker_is_fail_open(self) -> None:
        client = ScriptedClient(
            replies=[
                reply("reasoning"),
                reply("no explicit verdict line here"),
                reply(json.dumps(VALID_PROPOSAL)),
            ]
        )
        result = run_project(client, "context", meter=_meter(), max_debate_rounds=1)
        assert isinstance(result.outcome, ValidatedProposal)
        assert result.checker_decision == "absent"

    def test_a_validator_rejection_stands_even_when_the_checker_approves(self) -> None:
        client = ScriptedClient(
            replies=[
                reply("reasoning"),
                reply("VERDICT: APPROVE"),
                reply(json.dumps(OVERCLAIMED_PROPOSAL)),
                reply(json.dumps(OVERCLAIMED_PROPOSAL)),
                reply(json.dumps(OVERCLAIMED_PROPOSAL)),
            ]
        )
        result = run_project(client, "context", meter=_meter(), max_debate_rounds=1, max_attempts=3)
        assert isinstance(result.outcome, Rejection)
        assert result.validator_decision == "rejected"
        assert result.checker_decision == "approve"

    def test_the_debates_converged_output_feeds_generation(self) -> None:
        # §3 Step 3: the debate's converged proposer output feeds Step 2's
        # generation context.
        client = ScriptedClient(
            replies=[
                reply("CONVERGED-REASONING-MARKER"),
                reply("VERDICT: APPROVE"),
                reply(json.dumps(VALID_PROPOSAL)),
            ]
        )
        run_project(client, "context", meter=_meter(), max_debate_rounds=1)
        generation_prompt = client.prompts("proposer")[1]
        assert "CONVERGED-REASONING-MARKER" in generation_prompt