"""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. **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 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.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 main 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 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_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 == [] # --- 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"