The gap, found by the mutation sweep of 2026-07-25: verdict_dir=args.verdict_dir → None in run.py's execute_portfolio call left the suite 603/603 GREEN. The flag was wired but not guarded — the inner merge (test_portfolio_learning_ loadbearing.py), the argparse refusal (--verdict-dir without --portfolio) and the README↔--help sync all stay green under that mutation, so none of them covered the forwarding itself. One load-bearing test, no production code. It authors an expert verdict into a tmp portfolio inbox — keyed on the bundle's own codes + measure type so it ranks into the fold, with a distinct saving so its id cannot collide with the bundle's seed — drives main(["--portfolio", …, "--verdict-dir", X]) with the scripted client, and asserts the verdict's id AND a marker token (present nowhere in the bundle) reach the proposer prompt. Detach proof (mutation restored from a COPY, never git checkout): the wire mutated to None → 1 failed, 603 passed, and the failure is this test alone. Restored → 604 passed, ruff format left 71 files unchanged, ruff check + mypy --strict clean. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MQu2xxwedckjU56byu1aUG
444 lines
18 KiB
Python
444 lines
18 KiB
Python
"""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 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 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
|
||
|
||
|
||
# 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 == []
|
||
|
||
|
||
# --- 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"
|