open/portfolio-optimiser-claude
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity Actions

feat(loop): S8 — D7 agentic loop: budget meter, maker-checker gate, informed refinement

Browse source 
Spec §3 steps 2–5 + §8, TDD-ed offline (scripted, honesty-marked stand-in):
- budget.py: BudgetMeter over TerminationContract — provider-reported usage
  only (missing usage fails closed), structured BudgetExceeded stop event.
- loop.py: ModelClient protocol; blind parse-retry generation (never silent
  repair); round-capped debate with turn safety net and mandated VERDICT
  line; opt-in-reject checker gate (explicit REJECT overrides a validated
  outcome, validator rejection stands); most-recent-reason-verbatim informed
  refinement under max_attempts; validator_decision stamped BEFORE override,
  checker_decision as its own result field (§9, never conflated).
- 45 new tests (121 total, no API key); four detach proofs run RED and
  reverted green: checker override, informed block, surfaced checker output,
  stamp-before-override.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QdSfQdND84oeq2mbjueLTS
This commit is contained in:
Kjell Tore Guttormsen 2026-07-03 07:21:02 +02:00
commit 9a4caeb419
7 changed files with 1003 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

57
src/portfolio_optimiser_claude/budget.py Normal file
View file

@ -0,0 +1,57 @@
"""The budget meter (method-spec §8) — never an unbounded loop, anywhere.
Token accounting comes from the PROVIDER-REPORTED usage after each model call —
never a word-count or character proxy; on counting paths a response missing
usage fails CLOSED (``UsageAccountingError``), not silently uncounted. Crossing
a cap raises ``BudgetExceeded``, a STRUCTURED stop event carrying the breached
kind, the limit, and the observed value — never a silent hang. The caps come
from the fail-fast startup ``TerminationContract`` (§10), which already refuses
non-positive values.
"""
from __future__ import annotations
from typing import Literal
from portfolio_optimiser_claude.contracts import TerminationContract
BudgetKind = Literal["tokens", "rounds"]
class UsageAccountingError(Exception):
"""A response missing provider-reported usage on a counting path (§8)."""
class BudgetExceeded(Exception):
"""The structured stop event: breached kind + limit + observed value (§8)."""
def __init__(self, kind: BudgetKind, limit: int, observed: int) -> None:
super().__init__(f"budget exceeded: {kind} observed {observed} > limit {limit}")
self.kind: BudgetKind = kind
self.limit = limit
self.observed = observed
class BudgetMeter:
"""Run-scoped usage meter over the startup termination contract (§8)."""
def __init__(self, termination: TerminationContract) -> None:
self._termination = termination
self.tokens_used = 0
self.rounds_used = 0
def charge_tokens(self, usage_tokens: int | None) -> None:
"""Charge provider-reported usage; a missing usage fails closed (§8)."""
if usage_tokens is None:
raise UsageAccountingError(
"response carries no usage — token accounting must fail closed (§8)"
)
self.tokens_used += usage_tokens
if self.tokens_used > self._termination.max_tokens:
raise BudgetExceeded("tokens", self._termination.max_tokens, self.tokens_used)
def charge_round(self) -> None:
"""Charge one round tick (debate rounds and between-attempt ticks, §8)."""
self.rounds_used += 1
if self.rounds_used > self._termination.max_rounds:
raise BudgetExceeded("rounds", self._termination.max_rounds, self.rounds_used)

327
src/portfolio_optimiser_claude/loop.py Normal file
View file

@ -0,0 +1,327 @@
"""The agentic loop, steps 2–5 (method-spec §3): generate, debate, gate, refine.
The model is reached ONLY through the ``ModelClient`` protocol — an injected
client (the Claude Agent SDK on the run path, S10; a scripted stand-in in the
offline suite, honesty rule §1). Step 2: a reply that fails to parse into the
typed IR is retried BLIND, never silently accepted or repaired downstream,
bounded by the budget meter (§8). Step 3: the round-capped maker-checker
debate, fresh state per run, the checker instructed to end with exactly one
verdict line. Step 4 (second falsifier): the checker gate — opt-in-reject
(fail-open), parsed from the checker's LAST surfaced output; an explicit
REJECT overrides an otherwise-validated outcome; a validator rejection stands
regardless. Step 5: informed refinement — the most recent rejection REASON
verbatim in the next prompt (never history, never the prior proposal JSON),
under the existing ``max_attempts`` and meter caps. The two falsifiers are
recorded separately (§9): ``validator_decision`` mirrors the deterministic
validator only, stamped BEFORE any checker override.
"""
from __future__ import annotations
import json
import re
from dataclasses import dataclass
from typing import Literal, Protocol
from pydantic import ValidationError
from portfolio_optimiser_claude.budget import BudgetMeter
from portfolio_optimiser_claude.ir import SavingsProposal
from portfolio_optimiser_claude.provenance import stamp_validator_decision
from portfolio_optimiser_claude.validator import Rejection, ValidatedProposal, validate_proposal
_PROPOSER_ROLE = "proposer"
_CHECKER_ROLE = "checker"
# §3 Step 3: the checker's mandated closing line, both marker forms spelled out.
_CHECKER_INSTRUCTION = (
"End your reply with exactly one verdict line: `VERDICT: APPROVE` if the "
"reasoning holds, or `VERDICT: REJECT - <short reason>` if not."
)
# §3 Step 4: parsed case-insensitively; the reject marker takes precedence and
# its trailing text is the reason.
_REJECT_PATTERN = re.compile(r"VERDICT:\s*REJECT(?:\s*[-–—:]\s*(.*))?", re.IGNORECASE)
_APPROVE_PATTERN = re.compile(r"VERDICT:\s*APPROVE", re.IGNORECASE)
CheckerDecision = Literal["approve", "reject", "absent"]
@dataclass(frozen=True)
class ModelReply:
"""One model reply: text + provider-reported usage (+ real model id, §9)."""
text: str
usage_tokens: int | None = None
model: str | None = None
class ModelClient(Protocol):
"""The injected model seam — SDK client on the run path, stand-in in tests."""
def complete(self, prompt: str, *, role: str) -> ModelReply: ...
# --- Step 2: hypothesise (structured candidate generation) ---------------------------------
def _parse_candidate(text: str, default_project_id: str | None) -> SavingsProposal:
raw = json.loads(text)
if not isinstance(raw, dict):
raise ValueError("candidate reply is not a JSON object")
if default_project_id is not None:
# §3 Step 2: project_id MAY be defaulted from the project when omitted.
raw.setdefault("project_id", default_project_id)
return SavingsProposal.model_validate(raw)
def generate_candidate(
client: ModelClient,
prompt: str,
*,
meter: BudgetMeter,
default_project_id: str | None = None,
) -> SavingsProposal:
"""Ask for exactly one candidate as JSON for the IR — blind parse-retry (§3 Step 2).
A reply that fails to parse into the typed IR is retried with the SAME
prompt, never silently accepted or repaired downstream. The retry loop is
bounded by the budget meter: a round tick is charged between attempts (§8).
"""
while True:
model_reply = client.complete(prompt, role=_PROPOSER_ROLE)
meter.charge_tokens(model_reply.usage_tokens)
try:
# JSONDecodeError and pydantic's ValidationError are ValueErrors.
return _parse_candidate(model_reply.text, default_project_id)
except (ValueError, ValidationError):
meter.charge_round()
# --- Step 3: debate (maker-checker) ---------------------------------------------------------
@dataclass(frozen=True)
class DebateResult:
"""The debate's converged output + the checker's LAST surfaced reply (§3)."""
proposer_output: str
checker_last: str
rounds: int
def check_turn_safety_net(turns: int, max_rounds: int) -> None:
"""The turn-count termination safety net ABOVE the round cap (§3 Step 3, §8)."""
if turns > 2 * max_rounds + 2:
raise RuntimeError(
f"debate turn-count safety net tripped: {turns} turns with max_rounds={max_rounds}"
)
def _proposer_debate_prompt(context: str, critique: str | None) -> str:
prompt = (
f"{context}\n\n"
"Propose the reasoning for exactly one cost-saving candidate measure "
"for this project."
)
if critique is not None:
prompt += f"\n\nThe checker challenged your reasoning:\n{critique}\n\nAddress it."
return prompt
def _checker_prompt(proposer_output: str) -> str:
return (
f"Check the proposer's reasoning for flaws:\n\n{proposer_output}\n\n{_CHECKER_INSTRUCTION}"
)
def run_debate(
client: ModelClient, context: str, *, max_rounds: int, meter: BudgetMeter
) -> DebateResult:
"""Alternate proposer/checker turns — round-capped, fresh state per run (§3 Step 3).
Converges when the checker approves; otherwise the checker's critique
feeds the next proposer turn until the round cap. All state is local to
this call — nothing survives from one project run into the next.
"""
if max_rounds <= 0:
raise ValueError(f"max_rounds must be positive, got {max_rounds}")
proposer_output = ""
checker_last = ""
critique: str | None = None
rounds = 0
turns = 0
for _ in range(max_rounds):
turns += 1
check_turn_safety_net(turns, max_rounds)
proposer_reply = client.complete(
_proposer_debate_prompt(context, critique), role=_PROPOSER_ROLE
)
meter.charge_tokens(proposer_reply.usage_tokens)
proposer_output = proposer_reply.text
turns += 1
check_turn_safety_net(turns, max_rounds)
checker_reply = client.complete(_checker_prompt(proposer_output), role=_CHECKER_ROLE)
meter.charge_tokens(checker_reply.usage_tokens)
checker_last = checker_reply.text
rounds += 1
meter.charge_round()
if parse_checker_verdict(checker_last).decision == "approve":
break
critique = checker_last
return DebateResult(proposer_output=proposer_output, checker_last=checker_last, rounds=rounds)
# --- Step 4 (second falsifier): the checker gate --------------------------------------------
@dataclass(frozen=True)
class CheckerVerdict:
"""The parsed checker decision: approve / reject(+reason) / absent (§3 Step 4)."""
decision: CheckerDecision
reason: str | None
def parse_checker_verdict(text: str) -> CheckerVerdict:
"""Parse the verdict marker case-insensitively; REJECT takes precedence (§3 Step 4)."""
reject = _REJECT_PATTERN.search(text)
if reject:
reason = (reject.group(1) or "").strip()
return CheckerVerdict(decision="reject", reason=reason)
if _APPROVE_PATTERN.search(text):
return CheckerVerdict(decision="approve", reason=None)
return CheckerVerdict(decision="absent", reason=None)
def apply_checker_gate(
outcome: ValidatedProposal | Rejection, verdict: CheckerVerdict
) -> ValidatedProposal | Rejection:
"""Opt-in-reject gate (fail-open): only an explicit REJECT overrides (§3 Step 4).
A validator rejection stands regardless of the checker; APPROVE or a
missing/unparseable marker never blocks. An explicit REJECT turns an
otherwise-validated outcome into a rejection whose reason is prefixed
with the checker's reason.
"""
if isinstance(outcome, Rejection):
return outcome
if verdict.decision != "reject":
return outcome
reason = verdict.reason or "checker rejected the reasoning"
return Rejection(reason=f"{reason} (checker REJECT overrode a validated outcome)")
# --- Step 5: refine, informed and bounded ----------------------------------------------------
@dataclass(frozen=True)
class CandidateRun:
"""The refinement loop's result: last proposal, its outcome, attempts used."""
proposal: SavingsProposal
outcome: ValidatedProposal | Rejection
attempts: int
def _informed_prompt(base_prompt: str, reason: str) -> str:
# Only the most recent rejection REASON crosses attempts — never an
# accumulated history, never the prior proposal JSON (§3 Step 5).
return (
f"{base_prompt}\n\n"
"The previous attempt was rejected by the deterministic validator. "
"Revise the candidate to address this falsification:\n"
f"{reason}"
)
def run_candidate_loop(
client: ModelClient,
base_prompt: str,
*,
meter: BudgetMeter,
max_attempts: int = 3,
default_project_id: str | None = None,
) -> CandidateRun:
"""Generate → validate, informed by the last rejection reason (§3 Steps 2+5).
Attempt 1 uses the unchanged base prompt; each later attempt appends the
previous rejection reason VERBATIM as a revision instruction. Bounded by
``max_attempts`` and the meter (a round tick between attempts, §8). The
only per-attempt falsifier is the deterministic validator.
"""
if max_attempts <= 0:
raise ValueError(f"max_attempts must be positive, got {max_attempts}")
reason: str | None = None
for attempt in range(1, max_attempts + 1):
prompt = base_prompt if reason is None else _informed_prompt(base_prompt, reason)
proposal = generate_candidate(
client, prompt, meter=meter, default_project_id=default_project_id
)
outcome = validate_proposal(proposal)
if isinstance(outcome, ValidatedProposal):
return CandidateRun(proposal=proposal, outcome=outcome, attempts=attempt)
reason = outcome.reason
if attempt < max_attempts:
meter.charge_round()
return CandidateRun(proposal=proposal, outcome=outcome, attempts=max_attempts)
# --- The run: debate → generation → both falsifiers, recorded separately --------------------
@dataclass(frozen=True)
class RunResult:
"""One project run (§3 Step 6): a validated proposal or a TYPED rejection.
``validator_decision`` mirrors the DETERMINISTIC VALIDATOR only, stamped
before any checker override (§9); ``checker_decision`` is the second
falsifier's own result field — the two are never conflated.
"""
outcome: ValidatedProposal | Rejection
validator_decision: Literal["validated", "rejected"]
checker_decision: CheckerDecision
attempts: int
proposal: SavingsProposal
def _generation_prompt(context: str, converged_reasoning: str) -> str:
# §3 Step 3: the debate's converged proposer output feeds generation.
return (
f"{context}\n\n"
f"Converged reasoning from the maker-checker debate:\n{converged_reasoning}\n\n"
"Reply with exactly one candidate measure as a JSON object with the "
"fields: project_id, measure, affected_items (list of {code, quantity, "
"unit_cost}), claimed_saving_nok, and optionally assumptions."
)
def run_project(
client: ModelClient,
context: str,
*,
meter: BudgetMeter,
max_debate_rounds: int,
max_attempts: int = 3,
default_project_id: str | None = None,
) -> RunResult:
"""Run steps 2–5 for one project: debate, generate, validate, gate (§3)."""
debate = run_debate(client, context, max_rounds=max_debate_rounds, meter=meter)
run = run_candidate_loop(
client,
_generation_prompt(context, debate.proposer_output),
meter=meter,
max_attempts=max_attempts,
default_project_id=default_project_id,
)
verdict = parse_checker_verdict(debate.checker_last)
# §9: stamped from the validator's outcome BEFORE the checker override.
validator_decision = stamp_validator_decision(run.outcome)
return RunResult(
outcome=apply_checker_gate(run.outcome, verdict),
validator_decision=validator_decision,
checker_decision=verdict.decision,
attempts=run.attempts,
proposal=run.proposal,
)

53
tests/_scripted.py Normal file
View file

@ -0,0 +1,53 @@
"""SCRIPTED model-client stand-in for the offline test suite.
Honesty rule (method-spec §1): this is a scripted stand-in, NOT a model. It
replays canned replies (or computes them via a prompt-sensitive script
function) deterministically, records every call, and never touches a network
or an API key. The ONE genuine model run in the programme is S10.
"""
from __future__ import annotations
from typing import Callable
from portfolio_optimiser_claude.loop import ModelReply
ScriptFn = Callable[[str, str], ModelReply]
class ScriptedClient:
"""Deterministic ``ModelClient`` stand-in — canned replies, recorded calls.
Either ``replies`` (a FIFO of ``ModelReply``) or ``script`` (a function of
``(role, prompt)`` — prompt-sensitive, for load-bearing flip proofs) must
be given. Every call is recorded as ``(role, prompt)`` in ``calls``.
"""
def __init__(
self,
replies: list[ModelReply] | None = None,
script: ScriptFn | None = None,
) -> None:
if (replies is None) == (script is None):
raise ValueError("give exactly one of 'replies' or 'script'")
self._replies = list(replies) if replies is not None else None
self._script = script
self.calls: list[tuple[str, str]] = []
def complete(self, prompt: str, *, role: str) -> ModelReply:
self.calls.append((role, prompt))
if self._script is not None:
return self._script(role, prompt)
assert self._replies is not None
if not self._replies:
raise AssertionError("scripted client exhausted: no reply left for this call")
return self._replies.pop(0)
def prompts(self, role: str) -> list[str]:
"""The recorded prompts sent to ``role``, in call order."""
return [prompt for r, prompt in self.calls if r == role]
def reply(text: str, usage_tokens: int | None = 10) -> ModelReply:
"""Shorthand for a scripted ``ModelReply`` with a small default usage."""
return ModelReply(text=text, usage_tokens=usage_tokens)

75
tests/test_budget.py Normal file
View file

@ -0,0 +1,75 @@
"""The budget meter (method-spec §8) — never an unbounded loop, anywhere.
Normative behaviours proved here: token accounting comes from provider-reported
usage only (a missing usage fails CLOSED, never silently stops counting);
crossing a cap raises a STRUCTURED stop event carrying the breached kind, the
limit, and the observed value — never a silent hang.
"""
from __future__ import annotations
import pytest
from portfolio_optimiser_claude.budget import BudgetExceeded, BudgetMeter, UsageAccountingError
from portfolio_optimiser_claude.contracts import TerminationContract
def _meter(max_rounds: int = 5, max_tokens: int = 100) -> BudgetMeter:
return BudgetMeter(TerminationContract(max_rounds=max_rounds, max_tokens=max_tokens))
class TestTokenAccounting:
def test_accumulates_provider_reported_usage(self) -> None:
meter = _meter(max_tokens=100)
meter.charge_tokens(30)
meter.charge_tokens(20)
assert meter.tokens_used == 50
def test_missing_usage_fails_closed(self) -> None:
# §8: on counting paths, a response missing usage MUST fail closed —
# an error, not a silently-uncounted call.
meter = _meter()
with pytest.raises(UsageAccountingError):
meter.charge_tokens(None)
assert meter.tokens_used == 0
def test_reaching_the_cap_exactly_does_not_stop(self) -> None:
meter = _meter(max_tokens=100)
meter.charge_tokens(100)
assert meter.tokens_used == 100
def test_crossing_the_token_cap_raises_structured_stop(self) -> None:
meter = _meter(max_tokens=100)
meter.charge_tokens(90)
with pytest.raises(BudgetExceeded) as exc_info:
meter.charge_tokens(20)
stop = exc_info.value
assert stop.kind == "tokens"
assert stop.limit == 100
assert stop.observed == 110
class TestRoundAccounting:
def test_round_ticks_accumulate(self) -> None:
meter = _meter(max_rounds=5)
meter.charge_round()
meter.charge_round()
assert meter.rounds_used == 2
def test_crossing_the_round_cap_raises_structured_stop(self) -> None:
meter = _meter(max_rounds=2)
meter.charge_round()
meter.charge_round()
with pytest.raises(BudgetExceeded) as exc_info:
meter.charge_round()
stop = exc_info.value
assert stop.kind == "rounds"
assert stop.limit == 2
assert stop.observed == 3
def test_caps_come_from_the_startup_contract() -> None:
# §8/§10: the termination contract already refuses non-positive caps at
# construction — the meter builds on that, never on loose ints.
with pytest.raises(Exception):
TerminationContract(max_rounds=0, max_tokens=100)

162
tests/test_checker_gate_loadbearing.py Normal file
View file

@ -0,0 +1,162 @@
"""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

210
tests/test_loop.py Normal file
View file

@ -0,0 +1,210 @@
"""Steps 2 and 3 of the loop (method-spec §3): generation + maker-checker debate.
Step 2 — a reply that fails to parse into the typed IR is retried BLIND (same
prompt), never silently accepted or repaired downstream, bounded by the budget
meter (§8; round ticks between attempts). Step 3 — the debate is round-capped
with a turn-count safety net above it, state is fresh per run, and the checker
is INSTRUCTED to end with exactly one verdict line. Verdict parsing is
case-insensitive, the reject marker takes precedence, trailing text is the
reason, and a missing marker parses as absent (fail-open input to the gate).
"""
from __future__ import annotations
import json
import pytest
from _scripted import ScriptedClient, reply
from portfolio_optimiser_claude.budget import BudgetExceeded, BudgetMeter
from portfolio_optimiser_claude.contracts import TerminationContract
from portfolio_optimiser_claude.loop import (
generate_candidate,
parse_checker_verdict,
run_debate,
)
# One affected item: 100 × 1000 = 100_000 NOK total; nominal feasible 30_000;
# no assumptions band, so every Monte Carlo sample is 30_000 and p90 == 30_000.
VALID_PROPOSAL = {
"project_id": "p1",
"measure": "led-retrofit",
"affected_items": [{"code": "E01", "quantity": 100, "unit_cost": 1000}],
"claimed_saving_nok": 25000,
}
def _meter(max_rounds: int = 50, max_tokens: int = 10_000) -> BudgetMeter:
return BudgetMeter(TerminationContract(max_rounds=max_rounds, max_tokens=max_tokens))
class TestGenerateCandidate:
def test_valid_reply_parses_into_the_typed_ir(self) -> None:
client = ScriptedClient(replies=[reply(json.dumps(VALID_PROPOSAL))])
meter = _meter()
proposal = generate_candidate(client, "base prompt", meter=meter)
assert proposal.claimed_saving_nok == 25000
assert meter.tokens_used == 10
def test_malformed_reply_is_retried_blind_with_the_same_prompt(self) -> None:
client = ScriptedClient(
replies=[reply("not json at all"), reply(json.dumps(VALID_PROPOSAL))]
)
proposal = generate_candidate(client, "base prompt", meter=_meter())
assert proposal.measure == "led-retrofit"
# Blind retry: the SAME prompt, unchanged — never a repair instruction.
assert client.prompts("proposer") == ["base prompt", "base prompt"]
def test_schema_invalid_json_is_retried_never_repaired(self) -> None:
# Claim above the items' own total is a schema error (§7.1) — the value
# must never exist; the loop retries, it does not clamp or repair.
overclaim = dict(VALID_PROPOSAL, claimed_saving_nok=999_999)
client = ScriptedClient(
replies=[reply(json.dumps(overclaim)), reply(json.dumps(VALID_PROPOSAL))]
)
proposal = generate_candidate(client, "base prompt", meter=_meter())
assert proposal.claimed_saving_nok == 25000
def test_parse_retries_charge_round_ticks(self) -> None:
# §8: round ticks are charged between attempts so the meter also
# bounds parse-retries.
client = ScriptedClient(
replies=[reply("garbage"), reply("garbage"), reply(json.dumps(VALID_PROPOSAL))]
)
meter = _meter()
generate_candidate(client, "base prompt", meter=meter)
assert meter.rounds_used == 2
def test_endless_garbage_is_stopped_by_the_meter(self) -> None:
client = ScriptedClient(script=lambda role, prompt: reply("garbage"))
with pytest.raises(BudgetExceeded) as exc_info:
generate_candidate(client, "base prompt", meter=_meter(max_rounds=3))
assert exc_info.value.kind == "rounds"
def test_project_id_may_be_defaulted_from_the_project(self) -> None:
omitted = {k: v for k, v in VALID_PROPOSAL.items() if k != "project_id"}
client = ScriptedClient(replies=[reply(json.dumps(omitted))])
proposal = generate_candidate(
client, "base prompt", meter=_meter(), default_project_id="p-default"
)
assert proposal.project_id == "p-default"
def test_present_project_id_is_never_overwritten_by_the_default(self) -> None:
client = ScriptedClient(replies=[reply(json.dumps(VALID_PROPOSAL))])
proposal = generate_candidate(
client, "base prompt", meter=_meter(), default_project_id="p-default"
)
assert proposal.project_id == "p1"
def test_missing_usage_on_the_counting_path_fails_closed(self) -> None:
client = ScriptedClient(replies=[reply(json.dumps(VALID_PROPOSAL), usage_tokens=None)])
with pytest.raises(Exception, match="usage"):
generate_candidate(client, "base prompt", meter=_meter())
class TestRunDebate:
def test_converges_when_the_checker_approves(self) -> None:
client = ScriptedClient(replies=[reply("reasoning v1"), reply("holds. VERDICT: APPROVE")])
debate = run_debate(client, "context", max_rounds=3, meter=_meter())
assert debate.rounds == 1
assert debate.proposer_output == "reasoning v1"
assert "VERDICT: APPROVE" in debate.checker_last
def test_round_cap_bounds_a_never_approving_debate(self) -> None:
client = ScriptedClient(
script=lambda role, prompt: reply(
"VERDICT: REJECT - weak numbers" if role == "checker" else "reasoning"
)
)
debate = run_debate(client, "context", max_rounds=2, meter=_meter())
assert debate.rounds == 2
assert len(client.prompts("proposer")) == 2
assert len(client.prompts("checker")) == 2
assert "REJECT" in debate.checker_last
def test_checker_is_instructed_to_end_with_the_verdict_line(self) -> None:
# §3 Step 3: the checker MUST be instructed to end its reply with
# exactly one verdict line, both marker forms spelled out.
client = ScriptedClient(replies=[reply("reasoning"), reply("VERDICT: APPROVE")])
run_debate(client, "context", max_rounds=1, meter=_meter())
checker_prompt = client.prompts("checker")[0]
assert "VERDICT: APPROVE" in checker_prompt
assert "VERDICT: REJECT - <short reason>" in checker_prompt
def test_checker_critique_reaches_the_next_proposer_turn(self) -> None:
client = ScriptedClient(
replies=[
reply("reasoning v1"),
reply("VERDICT: REJECT - unit costs are stale"),
reply("reasoning v2"),
reply("VERDICT: APPROVE"),
]
)
debate = run_debate(client, "context", max_rounds=3, meter=_meter())
assert debate.rounds == 2
assert "unit costs are stale" in client.prompts("proposer")[1]
assert debate.proposer_output == "reasoning v2"
def test_debate_state_is_fresh_per_run(self) -> None:
# §3 Step 3: no conversation state may survive from one run into the
# next — the second run's opening proposer prompt carries nothing from
# the first run's transcript.
client = ScriptedClient(
replies=[
reply("FIRST-RUN-MARKER reasoning"),
reply("VERDICT: REJECT - FIRST-RUN-CRITIQUE"),
reply("more reasoning"),
reply("VERDICT: APPROVE"),
reply("second-run reasoning"),
reply("VERDICT: APPROVE"),
]
)
run_debate(client, "context A", max_rounds=2, meter=_meter())
run_debate(client, "context B", max_rounds=2, meter=_meter())
second_run_opening = client.prompts("proposer")[2]
assert "FIRST-RUN-MARKER" not in second_run_opening
assert "FIRST-RUN-CRITIQUE" not in second_run_opening
def test_every_turn_is_charged_on_the_meter(self) -> None:
client = ScriptedClient(
replies=[reply("reasoning", usage_tokens=7), reply("VERDICT: APPROVE", usage_tokens=5)]
)
meter = _meter()
run_debate(client, "context", max_rounds=1, meter=meter)
assert meter.tokens_used == 12
assert meter.rounds_used == 1
def test_the_turn_safety_net_sits_above_the_round_cap(self) -> None:
# §3 Step 3 / §8: an additional turn-count termination safety net
# ABOVE the round cap — it must never fire within a round-capped run,
# and must refuse a turn count beyond it.
from portfolio_optimiser_claude.loop import check_turn_safety_net
check_turn_safety_net(turns=2 * 3, max_rounds=3) # within: no raise
with pytest.raises(RuntimeError):
check_turn_safety_net(turns=2 * 3 + 3, max_rounds=3)
class TestParseCheckerVerdict:
def test_approve_is_parsed_case_insensitively(self) -> None:
verdict = parse_checker_verdict("the numbers hold.\nverdict: approve")
assert verdict.decision == "approve"
def test_reject_carries_the_trailing_text_as_reason(self) -> None:
verdict = parse_checker_verdict("VERDICT: REJECT - savings claim is double-counted")
assert verdict.decision == "reject"
assert verdict.reason == "savings claim is double-counted"
def test_the_reject_marker_takes_precedence(self) -> None:
verdict = parse_checker_verdict(
"VERDICT: APPROVE was my first instinct, but no.\nVERDICT: REJECT - stale baseline"
)
assert verdict.decision == "reject"
assert verdict.reason == "stale baseline"
def test_missing_marker_parses_as_absent(self) -> None:
verdict = parse_checker_verdict("looks fine to me")
assert verdict.decision == "absent"
def test_empty_text_parses_as_absent(self) -> None:
assert parse_checker_verdict("").decision == "absent"

119
tests/test_step5_refine_loadbearing.py Normal file
View file

@ -0,0 +1,119 @@
"""Load-bearing: informed, bounded refinement (method-spec §3 Step 5, §11).
RED-conditions proved here: the prior rejection reason no longer appears
VERBATIM in the next prompt, or the outcome never flips. The flip proof uses a
prompt-SENSITIVE scripted stand-in (honesty rule §1): it returns a feasible
proposal ONLY when the full rejection reason is present in the prompt — so the
test goes red the moment the informed block is detached. Bounds proved: only
the MOST RECENT reason is carried (never accumulated history), never the prior
proposal JSON, attempt 1 uses the unchanged base prompt, and the loop stops at
``max_attempts``.
"""
from __future__ import annotations
import json
import pytest
from _scripted import ScriptedClient, reply
from portfolio_optimiser_claude.budget import BudgetMeter
from portfolio_optimiser_claude.contracts import TerminationContract
from portfolio_optimiser_claude.ir import SavingsProposal
from portfolio_optimiser_claude.loop import ModelReply, run_candidate_loop
from portfolio_optimiser_claude.validator import Rejection, ValidatedProposal, validate_proposal
# 100 × 1000 = 100_000 NOK total; nominal feasible 30_000 == p90 (no band).
GOOD = {
"project_id": "p1",
"measure": "led-retrofit",
"affected_items": [{"code": "E01", "quantity": 100, "unit_cost": 1000}],
"claimed_saving_nok": 25000,
}
BAD_90K = dict(GOOD, claimed_saving_nok=90000) # IR-valid, above p90 → rejected
BAD_80K = dict(GOOD, claimed_saving_nok=80000) # a SECOND distinct rejection reason
def _rejection_reason(raw: dict[str, object]) -> str:
outcome = validate_proposal(SavingsProposal.model_validate(raw))
assert isinstance(outcome, Rejection)
return outcome.reason
def _meter() -> BudgetMeter:
return BudgetMeter(TerminationContract(max_rounds=50, max_tokens=10_000))
class TestInformedRefinement:
def test_the_outcome_flips_because_the_reason_reaches_the_prompt(self) -> None:
# THE load-bearing flip: the stand-in addresses the falsification ONLY
# if the full rejection reason appears verbatim — red at detach.
reason_90k = _rejection_reason(BAD_90K)
def script(role: str, prompt: str) -> ModelReply:
if reason_90k in prompt:
return reply(json.dumps(GOOD))
return reply(json.dumps(BAD_90K))
client = ScriptedClient(script=script)
result = run_candidate_loop(client, "base prompt", meter=_meter(), max_attempts=3)
assert isinstance(result.outcome, ValidatedProposal)
assert result.attempts == 2
assert reason_90k in client.prompts("proposer")[1]
def test_attempt_one_uses_the_unchanged_base_prompt(self) -> None:
client = ScriptedClient(replies=[reply(json.dumps(GOOD))])
run_candidate_loop(client, "base prompt", meter=_meter(), max_attempts=3)
assert client.prompts("proposer")[0] == "base prompt"
def test_only_the_most_recent_reason_is_carried(self) -> None:
# §3 Step 5: never an accumulated history — bounded prompt growth.
reason_90k = _rejection_reason(BAD_90K)
reason_80k = _rejection_reason(BAD_80K)
client = ScriptedClient(
replies=[
reply(json.dumps(BAD_90K)),
reply(json.dumps(BAD_80K)),
reply(json.dumps(GOOD)),
]
)
result = run_candidate_loop(client, "base prompt", meter=_meter(), max_attempts=3)
assert isinstance(result.outcome, ValidatedProposal)
third_prompt = client.prompts("proposer")[2]
assert reason_80k in third_prompt
assert reason_90k not in third_prompt
def test_the_prior_proposal_json_is_never_carried(self) -> None:
# The model must address the falsification, not parrot the rejected
# candidate — only the REASON crosses attempts.
client = ScriptedClient(replies=[reply(json.dumps(BAD_90K)), reply(json.dumps(GOOD))])
run_candidate_loop(client, "base prompt", meter=_meter(), max_attempts=3)
second_prompt = client.prompts("proposer")[1]
assert json.dumps(BAD_90K) not in second_prompt
assert '"affected_items"' not in second_prompt
def test_the_loop_stops_at_max_attempts(self) -> None:
client = ScriptedClient(script=lambda role, prompt: reply(json.dumps(BAD_90K)))
result = run_candidate_loop(client, "base prompt", meter=_meter(), max_attempts=3)
assert isinstance(result.outcome, Rejection)
assert result.attempts == 3
assert len(client.prompts("proposer")) == 3
def test_max_attempts_must_be_positive(self) -> None:
client = ScriptedClient(replies=[reply(json.dumps(GOOD))])
with pytest.raises(ValueError):
run_candidate_loop(client, "base prompt", meter=_meter(), max_attempts=0)
def test_round_ticks_are_charged_between_attempts(self) -> None:
client = ScriptedClient(replies=[reply(json.dumps(BAD_90K)), reply(json.dumps(GOOD))])
meter = _meter()
run_candidate_loop(client, "base prompt", meter=meter, max_attempts=3)
assert meter.rounds_used == 1
def test_a_never_validating_run_yields_a_typed_rejection_not_a_bare_failure(self) -> None:
# §3 Step 6: the outcome is either the validated proposal or a TYPED
# rejection with its reason — never a bare failure.
client = ScriptedClient(script=lambda role, prompt: reply(json.dumps(BAD_90K)))
result = run_candidate_loop(client, "base prompt", meter=_meter(), max_attempts=2)
assert isinstance(result.outcome, Rejection)
assert "exceeds the optimistic feasible bound" in result.outcome.reason