open/portfolio-optimiser-claude
1
0
Fork 0
Code Issues Releases 1 Activity
portfolio-optimiser-claude/tests/test_cli_paritet_loadbearing.py
Kjell Tore Guttormsen fc4a536e09 fix(run): classify portfolio-mode flags by allowlist so a new flag fails closed 
The portfolio entrance refused unsupported flags from a hard-coded BLOCKLIST:
--inbox, --out, --outbox, --run-id, --value-report, --live-dry-run. That
construction fails OPEN. A flag added to the parser later and forgotten in the
list is accepted, does nothing, and says nothing — the operator's flag is a
claim the run does not back (§1). MAF's report mode already used an allowlist;
the divergence was raised as an open question and the operator decided it this
session in favour of fail-closed.

unsupported_flags_given() now reports every flag GIVEN that the allowlist does
not name. "Given" is measured against the parser's own default, so it needs no
knowledge of which flags exist — that is what keeps it correct for flags added
after it was written, including store_true switches.

Load-bearing (§11), detach-proven twice (before and after ruff format, restored
from a copy): swapping the membership test back to a hard-coded refusal list
turns test_a_flag_nobody_classified_is_refused RED, while every CLI-level
refusal test stays green — they only exercise flags a blocklist already names,
so they do not cover this seam. The other direction is covered too: a run
passing all fourteen honoured flags still exits 0, and the allowlist entries
are checked against the CLI's own --help so a rename cannot leave a dead entry.

612 -> 624 passed, ruff + mypy --strict clean. README states the allowlist.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01MQu2xxwedckjU56byu1aUG
2026-07-25 15:33:30 +02:00

581 lines
24 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""CLI parity + documentation honesty — LOAD-BEARING (K12; method-spec §1, §8, §11).
Two seams this file keeps alive.
**The operator drives the whole build from the command line.** The goal
contract (K2) and the portfolio pass (K3) exist as capabilities; K12 is what
makes them REACHABLE. A hard goal already met by the ledger stops the run
BEFORE any model call — the goal bounds achievement where the §8 budget bounds
spend, and a stop is structured output, never a silent one. Detach the goal
check from the entrance → the run proceeds and spends → red. Detach the
portfolio branch → the config's projects never run → red. Stop forwarding
``--verdict-dir`` → the expert inbox the operator named reaches no fold → red:
the flag's refusal path and the README↔``--help`` sync leave that wire uncovered,
so it is asserted here on the fold itself.
**The README claims exactly what the CLI delivers (§1).** Every ``--flag`` the
README documents must exist in the help of a project CLI the README names.
This is the honesty rule in test form: a documented flag that no entrance
offers is a claim the implementation does not back, and it goes red here the
moment the two drift apart.
"""
from __future__ import annotations
import argparse
import contextlib
import importlib
import io
import json
import re
from pathlib import Path
from typing import Callable
import pytest
from _scripted import ScriptedClient, reply
from portfolio_optimiser_claude.contracts import Contracts, FeedbackContract
from portfolio_optimiser_claude.experience import CandidateFeatures
from portfolio_optimiser_claude.inbox import VerdictDocument, write_verdict
from portfolio_optimiser_claude.ir import load_validator_input
from portfolio_optimiser_claude.ledger import SavingsLedger
from portfolio_optimiser_claude.loop import ModelClient, ModelReply
from portfolio_optimiser_claude.run import (
_PORTFOLIO_SUPPORTED_DESTS,
main,
unsupported_flags_given,
)
REPO_ROOT = Path(__file__).resolve().parents[1]
BUNDLE = REPO_ROOT / "shared" / "examples" / "bygg-energi-mikro"
README = REPO_ROOT / "README.md"
ClientFactory = Callable[[Contracts, float], ModelClient]
# --- scripted plumbing (no model, no network — §1) -------------------------------------------
def _validated_replies(runs: int = 1) -> list[ModelReply]:
# The three-turn sequence that drives one project to a VALIDATED outcome,
# repeated once per project the portfolio pass will run.
turns: list[ModelReply] = []
for _ in range(runs):
turns += [
reply("debate reasoning"),
reply("VERDICT: APPROVE"),
reply(json.dumps(load_validator_input(BUNDLE).model_dump())),
]
return turns
def _scripted_factory(runs: int = 1) -> tuple[ClientFactory, list[ScriptedClient]]:
created: list[ScriptedClient] = []
def factory(contracts: Contracts, max_budget_usd_per_call: float) -> ModelClient:
client = ScriptedClient(replies=_validated_replies(runs))
created.append(client)
return client
return factory, created
def _goal_file(tmp_path: Path, *, target_nok: float, mode: str) -> Path:
path = tmp_path / "goal.json"
path.write_text(
json.dumps({"target_nok": target_nok, "mode": mode}), encoding="utf-8", newline="\n"
)
return path
def _ledger_file(tmp_path: Path, *, amount_nok: float) -> Path:
# Realized savings only ever enter the book through the expert gate (K1),
# so the fixture is built the way the ledger itself requires.
ledger = SavingsLedger()
ledger.realize(
project="bygg-kontor-nord",
measure_type="LED-retrofit",
affected_codes=["EL-01"],
amount_nok=amount_nok,
verdict=FeedbackContract(decision="approved", rationale="expert approved (fixture)"),
expert="fixture-expert",
timestamp="2026-07-25T00:00:00Z",
)
path = tmp_path / "ledger.json"
ledger.save(path)
return path
# A token that lives NOWHERE in the bundle — its presence in a prompt can only
# have come through the inbox the operator named with --verdict-dir, never from
# the bundle context or from seeding.
_VERDICT_DIR_MARKER = "K12-CLI-VERDICT-DIR-MARKER"
def _portfolio_inbox(tmp_path: Path) -> tuple[Path, VerdictDocument]:
"""An EXPERT verdict in a portfolio inbox, keyed near the bundle's candidate.
Same codes + measure type as the bundle's proposal (so it ranks into the
fold) with a distinct saving (so its id cannot collide with the bundle's own
seed). The marker rides in the rationale — the fold's learning signal.
"""
candidate = CandidateFeatures.from_proposal(load_validator_input(BUNDLE))
verdict = VerdictDocument.from_candidate(
CandidateFeatures(
affected_codes=candidate.affected_codes,
measure_type=candidate.measure_type,
claimed_saving_nok=candidate.claimed_saving_nok + 3000.0,
),
decision="approved",
rationale=f"prior portfolio verdict — realiseringskorreksjon [{_VERDICT_DIR_MARKER}]",
description="K12 CLI portfolio-inbox fixture (surface text, excluded from ranking)",
)
verdict_dir = tmp_path / "portfolio-inbox"
write_verdict(verdict_dir, verdict)
return verdict_dir, verdict
def _portfolio_file(tmp_path: Path, project_ids: list[str]) -> Path:
path = tmp_path / "portfolio.json"
path.write_text(
json.dumps(
{"projects": [{"project_id": pid, "bundle_dir": str(BUNDLE)} for pid in project_ids]}
),
encoding="utf-8",
newline="\n",
)
return path
# --- the goal seam ---------------------------------------------------------------------------
class TestGoalStopOnTheEntrance:
"""LOAD-BEARING (§11): a hard goal already reached stops the run before any spend."""
def test_hard_goal_reached_stops_before_any_model_call(
self, tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
# Detach point: drop the goal check from main() → the run proceeds and
# spends → exit 0 and a non-empty call log → RED.
factory, created = _scripted_factory()
code = main(
[
"--bundle",
str(BUNDLE),
"--out",
str(tmp_path / "out"),
"--goals",
str(_goal_file(tmp_path, target_nok=100_000.0, mode="hard")),
"--ledger",
str(_ledger_file(tmp_path, amount_nok=150_000.0)),
],
client_factory=factory,
)
assert code == 4
out = capsys.readouterr().out
assert "GOAL REACHED" in out
assert "100000" in out.replace("_", "") and "150000" in out.replace("_", "")
# The stop is BEFORE any spend: no model call was ever made.
assert all(client.calls == [] for client in created)
# A stopped run leaves no run artifacts — it never ran.
assert not (tmp_path / "out" / "proposal.json").exists()
def test_hard_goal_not_reached_runs_normally(self, tmp_path: Path) -> None:
# Control: the same wiring with a target ABOVE the book runs the project.
factory, created = _scripted_factory()
code = main(
[
"--bundle",
str(BUNDLE),
"--out",
str(tmp_path / "out"),
"--goals",
str(_goal_file(tmp_path, target_nok=500_000.0, mode="hard")),
"--ledger",
str(_ledger_file(tmp_path, amount_nok=150_000.0)),
],
client_factory=factory,
)
assert code == 0
assert created and created[0].calls != []
assert (tmp_path / "out" / "proposal.json").is_file()
def test_soft_goal_reached_flags_without_stopping(
self, tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
# A SOFT goal reached is a flag, never a stop (goals.py's own contract,
# preserved across the CLI seam).
factory, created = _scripted_factory()
code = main(
[
"--bundle",
str(BUNDLE),
"--out",
str(tmp_path / "out"),
"--goals",
str(_goal_file(tmp_path, target_nok=100_000.0, mode="soft")),
"--ledger",
str(_ledger_file(tmp_path, amount_nok=150_000.0)),
],
client_factory=factory,
)
assert code == 0
assert "GOAL REACHED (soft)" in capsys.readouterr().out
assert created and created[0].calls != []
def test_goal_without_ledger_reads_an_empty_book(self, tmp_path: Path) -> None:
# An absent ledger is an EMPTY book (0 realized), never a skipped check:
# the goal is evaluated, it is simply not reached.
factory, _ = _scripted_factory()
code = main(
[
"--bundle",
str(BUNDLE),
"--out",
str(tmp_path / "out"),
"--goals",
str(_goal_file(tmp_path, target_nok=1.0, mode="hard")),
],
client_factory=factory,
)
assert code == 0
def test_malformed_goal_is_refused_before_any_spend(self, tmp_path: Path) -> None:
# §10: the goal contract is a startup contract — a percent goal is
# D-E-gated and refuses loudly, before a client is ever constructed.
path = tmp_path / "goal.json"
path.write_text(
json.dumps({"target_nok": 100_000.0, "mode": "hard", "target_percent": 10.0}),
encoding="utf-8",
newline="\n",
)
factory, created = _scripted_factory()
with pytest.raises(SystemExit):
main(
["--bundle", str(BUNDLE), "--out", str(tmp_path / "out"), "--goals", str(path)],
client_factory=factory,
)
assert created == []
# --- the portfolio seam ----------------------------------------------------------------------
class TestPortfolioOnTheEntrance:
"""LOAD-BEARING (§11): the portfolio pass is reachable from the command line."""
def test_portfolio_runs_every_configured_project(
self, tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
# Detach point: drop the portfolio branch from main() → the configured
# projects never run → RED.
factory, created = _scripted_factory(runs=2)
code = main(
["--portfolio", str(_portfolio_file(tmp_path, ["prosjekt-a", "prosjekt-b"]))],
client_factory=factory,
)
assert code == 0
out = capsys.readouterr().out
assert "prosjekt-a" in out and "prosjekt-b" in out
# Both projects genuinely drove the loop: 2 projects × 3 scripted turns.
assert created and len(created[0].calls) == 6
def test_portfolio_and_bundle_are_mutually_exclusive(self, tmp_path: Path) -> None:
factory, created = _scripted_factory()
with pytest.raises(SystemExit):
main(
[
"--bundle",
str(BUNDLE),
"--portfolio",
str(_portfolio_file(tmp_path, ["prosjekt-a"])),
],
client_factory=factory,
)
assert created == []
def test_neither_bundle_nor_portfolio_is_refused(self) -> None:
factory, created = _scripted_factory()
with pytest.raises(SystemExit):
main([], client_factory=factory)
assert created == []
def test_verdict_dir_reaches_every_projects_fold(self, tmp_path: Path) -> None:
# Detach point: stop forwarding args.verdict_dir out of main() (pass None
# to execute_portfolio) → the inbox the operator named never reaches a
# fold → RED. The merge itself is proven inside run_portfolio
# (test_portfolio_learning_loadbearing.py); what this keeps alive is the
# CLI→run_portfolio WIRE — the refusal test below and the README↔--help
# sync both stay green under that mutation, so neither covers it.
verdict_dir, inbox_verdict = _portfolio_inbox(tmp_path)
factory, created = _scripted_factory(runs=1)
code = main(
[
"--portfolio",
str(_portfolio_file(tmp_path, ["prosjekt-a"])),
"--verdict-dir",
str(verdict_dir),
],
client_factory=factory,
)
assert code == 0
assert created
prompts = created[0].prompts("proposer")
assert prompts, "the project must have driven the loop at all"
assert any(_VERDICT_DIR_MARKER in p and inbox_verdict.id in p for p in prompts), (
"the verdict from --verdict-dir (id + marker) must reach the project's fold"
)
def test_verdict_dir_without_portfolio_is_refused(self, tmp_path: Path) -> None:
# --verdict-dir is the PORTFOLIO-level expert inbox; on a single run the
# per-run inbox is --inbox. Accepting it silently would claim a wiring
# that does not exist (§1).
factory, created = _scripted_factory()
with pytest.raises(SystemExit):
main(
[
"--bundle",
str(BUNDLE),
"--verdict-dir",
str(tmp_path / "verdicts"),
],
client_factory=factory,
)
assert created == []
def test_portfolio_refuses_the_run_id_named_flags(self, tmp_path: Path) -> None:
# The portfolio pass persists NOTHING (portfolio.py returns typed results
# and leaves filing to the caller), and the outbox names pairs by run_id.
# Refusing here is the honest alternative to a flag that silently does
# nothing.
factory, created = _scripted_factory()
with pytest.raises(SystemExit):
main(
[
"--portfolio",
str(_portfolio_file(tmp_path, ["prosjekt-a"])),
"--outbox",
str(tmp_path / "outbox"),
"--run-id",
"run-1",
],
client_factory=factory,
)
assert created == []
class TestPortfolioClassifiesByAllowlist:
"""LOAD-BEARING (§1, §11): portfolio mode fails CLOSED on an unclassified flag.
The refusal used to run off a BLOCKLIST — the flags to reject, named one by
one. That construction fails OPEN: a flag added to the parser later and
forgotten in the list is accepted and then silently ignored, which is
exactly the false claim §1 forbids (the operator's flag did nothing and the
CLI said nothing). The allowlist names the flags the portfolio pass HONOURS;
everything else is refused whether or not anyone remembered it.
Detach point: replace the ``dest not in supported`` membership test with a
hard-coded list of flags to refuse → ``test_a_flag_nobody_classified_is_refused``
goes RED, because tomorrow's flag appears in no such list. The CLI-level
refusal tests stay GREEN under that mutation (they only exercise flags a
blocklist already names), so they do not cover this seam.
"""
def _parser_with(self, *flags: str) -> argparse.ArgumentParser:
parser = argparse.ArgumentParser()
for flag in flags:
parser.add_argument(flag, default=None)
return parser
def test_a_flag_nobody_classified_is_refused(self) -> None:
# "--tomorrows-flag" stands in for the next flag someone adds: it is in
# no allowlist and in no blocklist. Fail-closed means it is reported.
parser = self._parser_with("--portfolio", "--tomorrows-flag")
args = parser.parse_args(["--portfolio", "p.json", "--tomorrows-flag", "x"])
assert unsupported_flags_given(parser, args, supported=frozenset({"portfolio"})) == [
"--tomorrows-flag"
]
def test_a_supported_flag_is_not_reported(self) -> None:
parser = self._parser_with("--portfolio", "--ledger")
args = parser.parse_args(["--portfolio", "p.json", "--ledger", "l.json"])
assert (
unsupported_flags_given(parser, args, supported=frozenset({"portfolio", "ledger"}))
== []
)
def test_a_flag_left_at_its_default_was_never_given(self) -> None:
# Absence is not a claim: an unsupported flag the operator did not pass
# must not turn a valid portfolio run into an error.
parser = self._parser_with("--portfolio", "--tomorrows-flag")
args = parser.parse_args(["--portfolio", "p.json"])
assert unsupported_flags_given(parser, args, supported=frozenset({"portfolio"})) == []
def test_a_store_true_flag_counts_as_given_only_when_passed(self) -> None:
# store_true defaults to False, not None — "given" is measured against
# the parser's OWN default, so both directions are correct.
parser = argparse.ArgumentParser()
parser.add_argument("--portfolio", default=None)
parser.add_argument("--tomorrows-switch", action="store_true")
supported = frozenset({"portfolio"})
unset = parser.parse_args(["--portfolio", "p.json"])
assert unsupported_flags_given(parser, unset, supported=supported) == []
given = parser.parse_args(["--portfolio", "p.json", "--tomorrows-switch"])
assert unsupported_flags_given(parser, given, supported=supported) == ["--tomorrows-switch"]
@pytest.mark.parametrize(
"extra",
[
["--out", "out-dir"],
["--inbox", "inbox-dir"],
["--outbox", "outbox-dir"],
["--run-id", "run-1"],
["--value-report", "report.json"],
["--live-dry-run"],
],
ids=["out", "inbox", "outbox", "run-id", "value-report", "live-dry-run"],
)
def test_the_real_cli_refuses_every_unsupported_flag(
self, tmp_path: Path, extra: list[str]
) -> None:
# The wire: main() classifies with the real allowlist, before any client.
factory, created = _scripted_factory()
with pytest.raises(SystemExit):
main(
["--portfolio", str(_portfolio_file(tmp_path, ["prosjekt-a"]))] + extra,
client_factory=factory,
)
assert created == []
def test_the_real_cli_accepts_every_flag_it_honours(self, tmp_path: Path) -> None:
# The other half of fail-closed: the allowlist must not refuse the flags
# the portfolio pass genuinely uses. A goal far above the book does not
# stop the run, so this exercises the full pass.
verdict_dir, _ = _portfolio_inbox(tmp_path)
factory, created = _scripted_factory(runs=1)
code = main(
[
"--portfolio",
str(_portfolio_file(tmp_path, ["prosjekt-a"])),
"--verdict-dir",
str(verdict_dir),
"--ledger",
str(_ledger_file(tmp_path, amount_nok=1000.0)),
"--goals",
str(_goal_file(tmp_path, target_nok=10_000_000.0, mode="hard")),
"--max-rounds",
"12",
"--max-tokens",
"150000",
"--max-budget-usd-per-call",
"0.25",
"--max-debate-rounds",
"3",
"--max-attempts",
"3",
"--top-k",
"3",
"--notify-console",
"--notify-file",
str(tmp_path / "notify"),
],
client_factory=factory,
)
assert code == 0
assert created, "the portfolio pass must have run"
def test_the_allowlist_names_only_flags_the_cli_actually_has(self) -> None:
# A renamed flag would leave a dead entry behind, and the renamed flag
# would start being refused in portfolio mode without anyone saying so.
help_text = _full_help("run")
missing = sorted(
f"--{dest.replace('_', '-')}"
for dest in _PORTFOLIO_SUPPORTED_DESTS
if f"--{dest.replace('_', '-')}" not in help_text
)
assert missing == []
# --- the documentation-honesty seam (§1) -----------------------------------------------------
# Lines about third-party dev tooling are not claims about this framework's CLI.
_FOREIGN_TOOL_MARKERS = ("ruff", "pytest", "mypy", "uv sync")
_FLAG = re.compile(r"--[a-z][a-z0-9-]*")
_MODULE = re.compile(r"portfolio_optimiser_claude\.([a-z_]+)")
_CHOICES = re.compile(r"\{([a-z0-9_,-]+)\}")
def _capture_help(module_name: str, argv: list[str]) -> str:
module = importlib.import_module(f"portfolio_optimiser_claude.{module_name}")
buffer = io.StringIO()
with contextlib.redirect_stdout(buffer), contextlib.suppress(SystemExit):
module.main(argv)
return buffer.getvalue()
def _full_help(module_name: str) -> str:
"""Top-level help plus every subcommand's help (hitl has ``pending``/``route``)."""
text = _capture_help(module_name, ["--help"])
subcommands: set[str] = set()
for match in _CHOICES.finditer(text):
subcommands.update(match.group(1).split(","))
for sub in sorted(subcommands):
text += _capture_help(module_name, [sub, "--help"])
return text
def _readme_documented_modules() -> list[str]:
return sorted(set(_MODULE.findall(README.read_text(encoding="utf-8"))))
def _readme_documented_flags() -> set[str]:
flags: set[str] = set()
for line in README.read_text(encoding="utf-8").splitlines():
if any(marker in line for marker in _FOREIGN_TOOL_MARKERS):
continue
flags.update(_FLAG.findall(line))
return flags
class TestReadmeClaimsMatchTheCli:
"""LOAD-BEARING (§1, §11): the README never documents a flag the CLI lacks."""
def test_every_documented_module_exposes_a_cli(self) -> None:
modules = _readme_documented_modules()
assert modules, "the README documents no entrance — the honesty grep would be vacuous"
for name in modules:
module = importlib.import_module(f"portfolio_optimiser_claude.{name}")
assert callable(getattr(module, "main", None)), (
f"README documents `python -m portfolio_optimiser_claude.{name}` "
"but the module exposes no CLI entrance"
)
def test_every_documented_flag_exists_in_a_documented_cli(self) -> None:
# RED the moment the README claims a flag the code does not offer —
# the drift K12 exists to close, kept closed from here on.
available = "\n".join(_full_help(name) for name in _readme_documented_modules())
assert "--bundle" in available, "help capture is broken — the grep would be vacuous"
undelivered = sorted(flag for flag in _readme_documented_flags() if flag not in available)
assert undelivered == [], (
f"README documents flags no CLI offers: {undelivered} — "
"either wire them or stop claiming them (§1)"
)
def test_the_operator_surfaces_are_all_documented(self) -> None:
# The other direction, bounded to the flags K12 promises the operator
# can drive from the command line: the run entrance's collecting
# surfaces must actually appear in the README.
documented = _readme_documented_flags()
for flag in (
"--bundle",
"--inbox",
"--outbox",
"--verdict-dir",
"--goals",
"--ledger",
"--portfolio",
"--value-report",
):
assert flag in documented, f"{flag} is an operator surface the README never mentions"
Reference in a new issue View git blame Copy permalink