open/portfolio-optimiser-claude
1
0
Fork 0
Code Issues Releases 1 Activity

feat(portfolio): K10 — notification/notifier seam, opt-in webhook egress (parity row 23) [skip-docs]

Browse source 
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
This commit is contained in:
Kjell Tore Guttormsen 2026-07-24 20:16:56 +02:00
commit a2acfc0f98
6 changed files with 863 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

257
tests/test_notify_loadbearing.py Normal file
View file

@ -0,0 +1,257 @@
"""Notification delivery — LOAD-BEARING (S5.2-analog; §8; §11; K10).
The seam this file keeps alive: a run/HITL pass can DELIVER a notification
(console, file, or webhook) WITHOUT breaking the no-silent-egress invariant. A
webhook — the one transport that leaves the machine — fires ONLY behind an
explicit per-run opt-in flag (mirroring ingest-spec §8: "the flag is a run
argument, never a config field — the config cannot grant itself network
access"). The transport is INJECTED: the suite passes a canned transport, so no
socket is ever opened; the real transport lives behind one clearly-marked seam
function that the suite never calls.
Three detached seams proven RED here:
* Opt-in gate: build a webhook notifier WITHOUT the opt-in flag → it MUST refuse
fail-fast (``EgressNotPermitted``) and its transport MUST stay unfired. Detach
point: drop the ``if not egress_opt_in: raise`` guard → the webhook constructs
and its transport fires on ``notify`` → RED.
* Payload structure: the canned transport receives the STRUCTURED event
(``event`` / ``summary`` / ``fields`` as deterministic JSON), not a prose blob.
Detach point: change the encoded shape → RED.
* No socket outside the seam: an AST grep-guard proves ``notify.py`` carries no
network import ANYWHERE except inside ``default_webhook_transport`` (the one
injectable seam). Detach point: hoist the ``urllib`` import to module scope (or
add any ``socket``/``httpx`` path elsewhere) → RED.
Key assumption (stack-local, no shared spec): the payload shape is this stack's
own — divergence from the MAF sibling is accepted and documented in the module
docstring. Pinned here by asserting the exact structured payload.
"""
from __future__ import annotations
import ast
import json
from pathlib import Path
import pytest
from portfolio_optimiser_claude.notify import (
ConsoleNotifier,
EgressNotPermitted,
FileNotifier,
Notification,
NotifyConfig,
WebhookNotifier,
build_notifiers,
emit,
notify_config_from_args,
)
SRC_PKG = Path(__file__).resolve().parents[1] / "src" / "portfolio_optimiser_claude"
_NETWORK_MODULES = {"socket", "urllib", "http", "https", "requests", "httpx", "anthropic", "ssl"}
_SEAM_FUNC = "default_webhook_transport"
class _CannedTransport:
"""A canned webhook transport — records calls, opens NO socket (§11 suite)."""
def __init__(self) -> None:
self.calls: list[tuple[str, bytes]] = []
def send(self, url: str, payload: bytes) -> None:
self.calls.append((url, payload))
def _event() -> Notification:
return Notification(
event="run.completed",
summary="run r-001: validated",
fields={"run_id": "r-001", "decision": "validated", "attempts": 1},
)
# --- the opt-in gate (LOAD-BEARING: webhook egress is opt-in per run) ------------------------
class TestWebhookOptInGate:
"""§8 mirror: the webhook refuses fail-fast WITHOUT the per-run opt-in flag."""
def test_webhook_without_optin_refuses_and_never_fires(self) -> None:
# LOAD-BEARING (the opt-in seam). Detach point: drop the guard in
# WebhookNotifier.__init__ → construction succeeds → transport fires on
# notify → this RED. Restore from the implemented copy, never git checkout.
canned = _CannedTransport()
with pytest.raises(EgressNotPermitted):
WebhookNotifier("https://hooks.example/x", canned.send, egress_opt_in=False)
assert canned.calls == [] # nothing left the machine
def test_build_notifiers_refuses_webhook_without_optin(self) -> None:
# The run-argument seam: the CLI flag threads the opt-in into build; a
# webhook_url without the flag is refused BEFORE any transport exists.
canned = _CannedTransport()
with pytest.raises(EgressNotPermitted):
build_notifiers(
NotifyConfig(webhook_url="https://hooks.example/x", webhook_egress_opt_in=False),
webhook_transport=canned.send,
)
assert canned.calls == []
def test_optin_webhook_constructs_and_fires(self) -> None:
canned = _CannedTransport()
notifier = WebhookNotifier("https://hooks.example/x", canned.send, egress_opt_in=True)
notifier.notify(_event())
assert len(canned.calls) == 1
# --- the payload structure (LOAD-BEARING: structured, deterministic) -------------------------
class TestPayloadStructure:
"""The canned transport receives the structured event, not a prose blob."""
def test_canned_transport_receives_structured_payload(self) -> None:
# LOAD-BEARING (payload shape, stack-local). Detach point: change the
# encoded shape (e.g. drop ``fields`` or emit prose) → RED.
canned = _CannedTransport()
notifiers = build_notifiers(
NotifyConfig(webhook_url="https://hooks.example/x", webhook_egress_opt_in=True),
webhook_transport=canned.send,
)
emit(notifiers, _event())
assert len(canned.calls) == 1
url, payload = canned.calls[0]
assert url == "https://hooks.example/x"
assert json.loads(payload) == {
"event": "run.completed",
"summary": "run r-001: validated",
"fields": {"run_id": "r-001", "decision": "validated", "attempts": 1},
}
def test_payload_is_byte_deterministic(self) -> None:
# Same event → identical bytes (sorted keys) — a notification is not a
# source of run-to-run drift.
a, b = _CannedTransport(), _CannedTransport()
WebhookNotifier("https://x", a.send, egress_opt_in=True).notify(_event())
WebhookNotifier("https://x", b.send, egress_opt_in=True).notify(_event())
assert a.calls[0][1] == b.calls[0][1]
# --- no socket outside the injectable seam (LOAD-BEARING: AST grep-guard) --------------------
def _import_nodes(tree: ast.AST) -> list[ast.stmt]:
return [n for n in ast.walk(tree) if isinstance(n, (ast.Import, ast.ImportFrom))]
def _modules_of(node: ast.stmt) -> set[str]:
if isinstance(node, ast.Import):
return {alias.name.split(".")[0] for alias in node.names}
if isinstance(node, ast.ImportFrom) and node.module:
return {node.module.split(".")[0]}
return set()
class TestNoSocketOutsideSeam:
"""LOAD-BEARING (§11): the ONLY network path in notify.py is the seam function."""
def test_no_network_import_outside_the_seam(self) -> None:
# Detach point: hoist the urllib import out of default_webhook_transport
# to module scope, or add any socket/httpx path elsewhere → an import
# node OUTSIDE the seam references a network module → RED.
tree = ast.parse((SRC_PKG / "notify.py").read_text("utf-8"))
seam = next(
n for n in ast.walk(tree) if isinstance(n, ast.FunctionDef) and n.name == _SEAM_FUNC
)
seam_node_ids = {id(n) for n in _import_nodes(seam)}
outside: set[str] = set()
for node in _import_nodes(tree):
if id(node) not in seam_node_ids:
outside |= _modules_of(node) & _NETWORK_MODULES
assert outside == set()
def test_the_seam_is_real_not_hollow(self) -> None:
# Green-but-dead guard: the seam MUST actually carry the network import
# it isolates — deleting it (so the guard passes vacuously) is RED here.
tree = ast.parse((SRC_PKG / "notify.py").read_text("utf-8"))
seam = next(
n for n in ast.walk(tree) if isinstance(n, ast.FunctionDef) and n.name == _SEAM_FUNC
)
seam_modules: set[str] = set()
for node in _import_nodes(seam):
seam_modules |= _modules_of(node)
assert "urllib" in seam_modules
# --- the local notifiers (console, file — no network) ----------------------------------------
class TestLocalNotifiers:
"""console and file notifiers deliver locally — never a socket."""
def test_console_notifier_prints_event_and_summary(
self, capsys: pytest.CaptureFixture[str]
) -> None:
ConsoleNotifier().notify(_event())
out = capsys.readouterr().out
assert "run.completed" in out
assert "run r-001: validated" in out
def test_file_notifier_writes_deterministic_json(self, tmp_path: Path) -> None:
FileNotifier(tmp_path).notify(_event())
written = tmp_path / "run.completed.json"
assert written.is_file()
body = json.loads(written.read_text("utf-8"))
assert body["event"] == "run.completed"
assert body["fields"]["run_id"] == "r-001"
# deterministic house JSON: sorted keys, trailing newline
assert written.read_text("utf-8").endswith("}\n")
def test_build_notifiers_composes_the_configured_set(self, tmp_path: Path) -> None:
notifiers = build_notifiers(NotifyConfig(console=True, file_dir=tmp_path))
assert [type(n).__name__ for n in notifiers] == ["ConsoleNotifier", "FileNotifier"]
def test_no_config_builds_nothing(self) -> None:
assert build_notifiers(NotifyConfig()) == []
# --- the CLI-arg mapping (shared by run.py and hitl.py) --------------------------------------
class TestNotifyConfigFromArgs:
"""notify_config_from_args maps the shared CLI flags into a NotifyConfig."""
def test_flags_map_into_config(self, tmp_path: Path) -> None:
import argparse
from portfolio_optimiser_claude.notify import add_notify_args
parser = argparse.ArgumentParser()
add_notify_args(parser)
args = parser.parse_args(
[
"--notify-console",
"--notify-file",
str(tmp_path),
"--notify-webhook",
"https://x",
"--allow-webhook-egress",
]
)
cfg = notify_config_from_args(args)
assert cfg == NotifyConfig(
console=True,
file_dir=tmp_path,
webhook_url="https://x",
webhook_egress_opt_in=True,
)
def test_defaults_are_all_off(self) -> None:
import argparse
from portfolio_optimiser_claude.notify import add_notify_args
parser = argparse.ArgumentParser()
add_notify_args(parser)
cfg = notify_config_from_args(parser.parse_args([]))
assert cfg == NotifyConfig()

245
tests/test_notify_seam_loadbearing.py Normal file
View file

@ -0,0 +1,245 @@
"""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