"""Notify seam in run.py / hitl.py — LOAD-BEARING (S5.2-analog; §8; §11; K10). The seam this file keeps alive: BOTH deliverable entrances (``run.py`` and the read-only ``hitl.py``) can emit a notification behind the SAME opt-in-gated CLI seam, and a webhook without the explicit per-run opt-in flag refuses fail-fast BEFORE any spend (run) / BEFORE any transport fires (hitl). No socket is opened in the suite — the transport is injected (canned). Detach proofs: * Opt-in threaded, not hardcoded: ``--notify-webhook URL`` WITHOUT ``--allow-webhook-egress`` → refuse fail-fast; the scripted client is never constructed (run: no spend) and the canned transport stays empty. Detach point: hardcode the opt-in true (or drop the gate) → the webhook builds and fires → RED. * Emit wired: an opted-in run/hitl pass delivers the STRUCTURED event to the injected transport. Detach point: drop the ``emit`` call → the canned transport gets nothing → RED. * hitl stays read-only: notification does NOT write any of the three layers — a before/after byte snapshot of outbox+inbox is unchanged even while emitting. """ from __future__ import annotations import json from pathlib import Path import pytest from _scripted import ScriptedClient, reply from portfolio_optimiser_claude.contracts import Contracts from portfolio_optimiser_claude.hitl import main as hitl_main from portfolio_optimiser_claude.ir import AffectedItem, SavingsProposal, load_validator_input from portfolio_optimiser_claude.loop import ModelClient, RunResult from portfolio_optimiser_claude.outbox import persist_outbox from portfolio_optimiser_claude.provenance import Citation, Provenance from portfolio_optimiser_claude.run import main as run_main from portfolio_optimiser_claude.validator import ValidatedProposal BUNDLE = Path(__file__).resolve().parents[1] / "shared" / "examples" / "bygg-energi-mikro" class _CannedTransport: def __init__(self) -> None: self.calls: list[tuple[str, bytes]] = [] def send(self, url: str, payload: bytes) -> None: self.calls.append((url, payload)) def _scripted_factory(replies: list[object]) -> tuple[object, list[ScriptedClient]]: created: list[ScriptedClient] = [] def factory(contracts: Contracts, max_budget_usd_per_call: float) -> ModelClient: client = ScriptedClient(replies=list(replies)) # type: ignore[arg-type] created.append(client) return client return factory, created def _happy_replies() -> list[object]: return [ reply("debate reasoning"), reply("VERDICT: APPROVE"), reply(json.dumps(load_validator_input(BUNDLE).model_dump())), ] # --- run.py notify seam ---------------------------------------------------------------------- class TestRunNotifySeam: def test_webhook_without_optin_refuses_before_any_spend(self, tmp_path: Path) -> None: # LOAD-BEARING (opt-in threaded). Detach: hardcode opt-in true → the # webhook builds and the run proceeds → this RED. The client is never # constructed, so no model call (and no spend) rides on the refusal. canned = _CannedTransport() factory, created = _scripted_factory(_happy_replies()) with pytest.raises(SystemExit): run_main( [ "--bundle", str(BUNDLE), "--out", str(tmp_path / "out"), "--notify-webhook", "https://hooks.example/x", ], client_factory=factory, notifier_transport=canned.send, ) assert created == [] # refused before the client was constructed → no spend assert canned.calls == [] def test_completion_emits_structured_event(self, tmp_path: Path) -> None: # LOAD-BEARING (emit wired). Detach: drop the completion emit → the # canned transport gets nothing → RED. canned = _CannedTransport() factory, _ = _scripted_factory(_happy_replies()) code = run_main( [ "--bundle", str(BUNDLE), "--out", str(tmp_path / "out"), "--notify-webhook", "https://hooks.example/x", "--allow-webhook-egress", ], client_factory=factory, notifier_transport=canned.send, ) assert code == 0 assert len(canned.calls) == 1 body = json.loads(canned.calls[0][1]) assert body["event"] == "run.completed" assert body["fields"]["validator_decision"] == "validated" def test_budget_stop_emits_stopped_event(self, tmp_path: Path) -> None: # A budget stop is a run outcome, not an absence of one — it notifies too. canned = _CannedTransport() factory, _ = _scripted_factory([reply("debate reasoning", usage_tokens=10)]) code = run_main( [ "--bundle", str(BUNDLE), "--out", str(tmp_path / "out"), "--max-tokens", "5", "--notify-webhook", "https://hooks.example/x", "--allow-webhook-egress", ], client_factory=factory, notifier_transport=canned.send, ) assert code == 3 assert len(canned.calls) == 1 body = json.loads(canned.calls[0][1]) assert body["event"] == "run.stopped" assert body["fields"]["kind"] == "tokens" # --- hitl.py notify seam (read-only preserved) ----------------------------------------------- def _proposal() -> SavingsProposal: return SavingsProposal( project_id="bygg-kontor-nord", measure="LED-retrofit", affected_items=[AffectedItem(code="EL-01", quantity=100, unit_cost=250.0)], claimed_saving_nok=20000.0, ) def _run(proposal: SavingsProposal) -> RunResult: return RunResult( outcome=ValidatedProposal( validates=True, claimed_saving_nok=20000.0, nominal_feasible=25000.0, p10=18000.0, p50=22000.0, p90=27000.0, ), validator_decision="validated", checker_decision="approve", attempts=1, proposal=proposal, ) def _provenance() -> Provenance: return Provenance( citations=[Citation(file="index.md", span="chars 0-5", snippet="Bygg-")], model="claude-haiku-4-5-20251001", role="proposer", validator_decision="validated", tokens_used=1234, ) def _snapshot(root: Path) -> dict[str, bytes]: if not root.exists(): return {} return { str(p.relative_to(root)): p.read_bytes() for p in sorted(root.rglob("*")) if p.is_file() } def _seed_pending(tmp_path: Path) -> tuple[Path, Path]: outbox, inbox = tmp_path / "outbox", tmp_path / "inbox" persist_outbox(outbox, run=_run(_proposal()), provenance=_provenance(), run_id="r-001") return outbox, inbox class TestHitlNotifySeam: def test_pending_webhook_without_optin_refuses(self, tmp_path: Path) -> None: canned = _CannedTransport() outbox, inbox = _seed_pending(tmp_path) before_out = _snapshot(outbox) with pytest.raises(SystemExit): hitl_main( [ "pending", "--outbox", str(outbox), "--inbox", str(inbox), "--notify-webhook", "https://hooks.example/x", ], notifier_transport=canned.send, ) assert canned.calls == [] assert _snapshot(outbox) == before_out # read-only preserved def test_pending_emits_count_and_stays_read_only(self, tmp_path: Path) -> None: # LOAD-BEARING (emit wired + read-only). Detach: drop the emit → canned # empty → RED. The three layers are never written even while notifying. canned = _CannedTransport() outbox, inbox = _seed_pending(tmp_path) before_out, before_in = _snapshot(outbox), _snapshot(inbox) code = hitl_main( [ "pending", "--outbox", str(outbox), "--inbox", str(inbox), "--notify-webhook", "https://hooks.example/x", "--allow-webhook-egress", ], notifier_transport=canned.send, ) assert code == 0 assert len(canned.calls) == 1 body = json.loads(canned.calls[0][1]) assert body["event"] == "hitl.pending" assert body["fields"]["pending"] == 1 assert _snapshot(outbox) == before_out assert _snapshot(inbox) == before_in