Kjell Tore Guttormsen
a2acfc0f98
S5.2-analog. New notify.py: Notifier protocol + console/file/webhook sinks. The webhook (the one transport that leaves the machine) fires ONLY behind an explicit per-run opt-in flag (--allow-webhook-egress), mirroring ingest-spec §8 (the flag is a run argument, never a config field). Transport is injected — canned in the suite (NULL socket), real transport behind one seam function default_webhook_transport; an AST grep-guard proves no network path exists outside that seam. run.py (both outcomes — a budget stop notifies too) and hitl.py (read-only preserved) share the same opt-in-gated CLI seam, refusing a webhook-without-opt-in before any spend. Payload shape is stack-local (no shared notification spec; divergence documented). Two new load-bearing test files (18 tests): opt-in gate + payload structure + the grep-guard + run/hitl emit wiring + run-level opt-in threading, each detach-proven RED. 544 -> 562 green, full gate clean (ruff+format+mypy strict, 26 src files). README sync (test count x2 + notify.py module note + load-bearing omtale). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RiTwaKLesgcwXx2mDviqpt
245 lines
8.7 KiB
Python
245 lines
8.7 KiB
Python
"""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
|