"""Live-run drill — LOAD-BEARING (K8; method-spec §8; comparison protocol §4 pt 3). The seam this file keeps alive: ``--live-dry-run`` BUILDS everything a real live run would (contracts fail-fast → compose → client construction → preflight) and CAPTURES the run-config + preflight artifacts to the outbox, then STOPS before the first model call. A future operator-gated live run (the M2-analog) is thus fully rigged and rehearsed offline — without one model call, without a key. Detach proof (the 0-calls seam): remove the dry-run branch from ``main`` so it falls through to ``execute_run`` → the injected call-counting client's ``complete`` fires → ``calls`` is non-empty (and the empty-reply stand-in raises) → red. Detach proof (the capture seam): drop the artifact write → the outbox lacks the run_id-named pair → red. No credential and no network are needed: the drill constructs the client (the verified key-free SDK premise) and the call-counting stand-in guarantees the boundary. The env is monkeypatched so the preflight verdict is deterministic regardless of the operator's ambient shell. """ from __future__ import annotations import json from pathlib import Path from typing import Callable import pytest from _scripted import ScriptedClient from portfolio_optimiser_claude.contracts import Contracts, load_contracts from portfolio_optimiser_claude.loop import ModelClient from portfolio_optimiser_claude.run import main BUNDLE = Path(__file__).resolve().parents[1] / "shared" / "examples" / "bygg-energi-mikro" RUN_ID = "dryrun-001" ClientFactory = Callable[[Contracts, float], ModelClient] def _counting_factory() -> tuple[ClientFactory, list[ScriptedClient]]: """A factory whose clients record every call and carry NO replies. An empty reply list means any ``complete`` both records the call and raises — so a detached dry-run (one that reaches the loop) fails loudly, and a correct dry-run leaves ``calls`` empty. """ created: list[ScriptedClient] = [] def factory(contracts: Contracts, max_budget_usd_per_call: float) -> ModelClient: client = ScriptedClient(replies=[]) created.append(client) return client return factory, created def _clear_credentials(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) monkeypatch.delenv("CLAUDE_CODE_OAUTH_TOKEN", raising=False) def _set_credential(monkeypatch: pytest.MonkeyPatch) -> None: # A non-placeholder form; the preflight never validates it online, so this is # not a real key and never leaves the process (the counting client blocks any # call). It only exercises the clear-preflight branch. _clear_credentials(monkeypatch) monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-drill-not-a-real-key") class TestDryRunStopsBeforeFirstCall: """The boundary: the drill builds everything but never calls the model.""" def test_zero_model_calls(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: _clear_credentials(monkeypatch) factory, created = _counting_factory() main( [ "--bundle", str(BUNDLE), "--outbox", str(tmp_path / "outbox"), "--run-id", RUN_ID, "--live-dry-run", ], client_factory=factory, ) # The client was constructed (the drill builds the client), but never called. (client,) = created assert client.calls == [] def test_requires_outbox_and_run_id(self, tmp_path: Path) -> None: factory, _ = _counting_factory() # No --outbox / --run-id: the run_id-named artifacts have nowhere to go. with pytest.raises(SystemExit): main( ["--bundle", str(BUNDLE), "--live-dry-run"], client_factory=factory, ) class TestDryRunArtifactCapture: """The captured set (run-config + preflight) is complete and deterministic.""" def test_captures_runconfig_and_preflight( self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: _clear_credentials(monkeypatch) outbox = tmp_path / "outbox" factory, _ = _counting_factory() main( [ "--bundle", str(BUNDLE), "--outbox", str(outbox), "--run-id", RUN_ID, "--live-dry-run", ], client_factory=factory, ) runconfig = json.loads((outbox / f"{RUN_ID}-runconfig.json").read_text("utf-8")) preflight = json.loads((outbox / f"{RUN_ID}-preflight.json").read_text("utf-8")) # §4 pt 3: model-id per role the loop calls, parameters, caps — no wall-clock. assert runconfig["run_id"] == RUN_ID assert runconfig["profile"] == "anthropic" assert runconfig["models"]["proposer"] == "claude-haiku-4-5-20251001" assert runconfig["models"]["checker"] == "claude-haiku-4-5-20251001" assert runconfig["caps"]["max_rounds"] == 12 assert runconfig["caps"]["max_tokens"] == 150_000 assert runconfig["caps"]["max_budget_usd_per_call"] == 0.25 assert "date" not in runconfig # determinism: date is stamped at report time # Preflight result captured (no credential here → credential refusal recorded). assert preflight["run_id"] == RUN_ID assert preflight["clear"] is False assert any(r["check"] == "credential" for r in preflight["refusals"]) def test_artifacts_are_byte_deterministic( self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: _clear_credentials(monkeypatch) first = tmp_path / "a" second = tmp_path / "b" for outbox in (first, second): factory, _ = _counting_factory() main( [ "--bundle", str(BUNDLE), "--outbox", str(outbox), "--run-id", RUN_ID, "--live-dry-run", ], client_factory=factory, ) for name in (f"{RUN_ID}-runconfig.json", f"{RUN_ID}-preflight.json"): assert (first / name).read_bytes() == (second / name).read_bytes() class TestDryRunPreflightGate: """Exit code reflects go-live readiness; capture happens either way.""" def test_clear_preflight_exits_zero( self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: _set_credential(monkeypatch) outbox = tmp_path / "outbox" factory, created = _counting_factory() code = main( [ "--bundle", str(BUNDLE), "--outbox", str(outbox), "--run-id", RUN_ID, "--live-dry-run", ], client_factory=factory, ) assert code == 0 preflight = json.loads((outbox / f"{RUN_ID}-preflight.json").read_text("utf-8")) assert preflight["clear"] is True assert preflight["refusals"] == [] (client,) = created assert client.calls == [] # still zero calls def test_refused_preflight_captures_but_exits_nonzero( self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: _clear_credentials(monkeypatch) outbox = tmp_path / "outbox" factory, created = _counting_factory() code = main( [ "--bundle", str(BUNDLE), "--outbox", str(outbox), "--run-id", RUN_ID, "--live-dry-run", ], client_factory=factory, ) assert code != 0 # refused: the rig is not clear to go live # ...yet the artifacts are captured and no model call was made. assert (outbox / f"{RUN_ID}-runconfig.json").is_file() assert (outbox / f"{RUN_ID}-preflight.json").is_file() (client,) = created assert client.calls == [] class TestKeyFreeConstruction: """K8 key premise: the SDK client constructs with no credential (no call).""" def test_default_factory_constructs_without_credential( self, monkeypatch: pytest.MonkeyPatch ) -> None: _clear_credentials(monkeypatch) from portfolio_optimiser_claude.run import default_client_factory from portfolio_optimiser_claude.sdk_client import SdkModelClient contracts = load_contracts( data_source={"docs_dir": str(BUNDLE), "top_k": 3}, termination={"max_rounds": 1, "max_tokens": 1}, feedback={"decision": "approved", "rationale": "startup shape check (§10)"}, ) client = default_client_factory(contracts, 0.25) assert isinstance(client, SdkModelClient)