"""S5.2 — Varsling (notification): B11 notify-stub → leverbare Notifiers (roadmap E, målbilde §3). ``run_project`` takes ``notify: Callable[[Verdict], None] | None`` and calls ``notify(verdict)`` after the verdict is captured — this module supplies the deliverable implementations behind that seam (the seam itself stays untouched). ``Notifier`` is the declared contract: a ``__call__``-shaped structural superset of the seam's ``Callable``, so any plain callable still satisfies it. **MAF-free** (D7-portable): stdlib + pydantic. ``Verdict`` is imported ONLY under ``TYPE_CHECKING`` and every runtime access is duck-typed — so importing this module never pulls in ``agent_framework``. Registered in ``tests/test_okf.py``'s ``_MAF_FREE_MODULES`` (direct-import guard ``test_okf_is_maf_free``) and probed transitively by ``tests/test_notify_loadbearing.py::test_notify_transitive_import_graph_is_maf_free``. """ from __future__ import annotations import json import sys from collections.abc import Callable from http.client import HTTPException from pathlib import Path from typing import TYPE_CHECKING, Any, Protocol, TextIO, runtime_checkable from urllib.error import URLError from urllib.request import Request, urlopen from pydantic import BaseModel, ConfigDict, model_validator if TYPE_CHECKING: # verdicts imports agent_framework — keep it out of the runtime import graph from portfolio_optimiser.verdicts import Verdict @runtime_checkable class Notifier(Protocol): """The declared notify contract (B11): structurally a superset of the run-path seam ``Callable[[Verdict], None]`` — ``isinstance(fn, Notifier)`` holds for any plain callable.""" def __call__(self, verdict: Verdict) -> None: ... class NotifyError(RuntimeError): """A notifier failed to deliver (e.g. webhook transport failure).""" class NotifyRefused(Exception): """Fail-closed egress-gate refusal (mirrors ``verdicts.PromotionRefused``).""" class ConsoleNotifier: """One-line verdict summary to a text stream (default stdout). Never receives a URL. The default stream is resolved at CALL time (``None`` → current ``sys.stdout``), so a later ``redirect_stdout``/capture is honored — an import-time ``sys.stdout`` default would bind the original stream object and bypass the redirect.""" def __init__(self, stream: TextIO | None = None) -> None: self._stream = stream def __call__(self, verdict: Verdict) -> None: stream = self._stream if self._stream is not None else sys.stdout print(f"[notify] verdict {verdict.id} decision={verdict.decision}", file=stream) def _verdict_payload(verdict: Verdict) -> dict[str, Any]: """NESTED, byte-deterministic, duck-typed payload — reproduces ``verdicts.verdict_to_dict``'s exact shape (verdicts.py:124-139) WITHOUT importing it (a runtime import would pull ``agent_framework``). Pinned against drift by the ``== verdict_to_dict(v)`` oracle test.""" f = verdict.proposal_features return { "id": verdict.id, "decision": verdict.decision, "rationale": verdict.rationale, "proposal_features": { "affected_codes": sorted(f.affected_codes), "measure_type": f.measure_type, "claimed_saving_nok": f.claimed_saving_nok, "description": f.description, }, } class FileNotifier: """Appends one JSONL line per verdict (byte-deterministic: ``sort_keys`` + LF, no wall-clock — the ``outbox._dump`` idiom). Parent directories are created as needed.""" def __init__(self, path: str) -> None: self._path = Path(path) def __call__(self, verdict: Verdict) -> None: self._path.parent.mkdir(parents=True, exist_ok=True) line = json.dumps(_verdict_payload(verdict), sort_keys=True) + "\n" with self._path.open("a", encoding="utf-8") as handle: handle.write(line) # --- webhook (the ONLY egress point): injectable transport + fail-closed opt-in gate -------------- #: The webhook transport seam: ``(url, json-body) -> None``. Injecting a canned implementation in #: tests keeps the suite socket-free; the default ``_urllib_post`` is the ONLY path that opens a #: socket, and it is reachable ONLY through a gate-passed ``WebhookNotifier``. WebhookPost = Callable[[str, str], None] def _urllib_post(url: str, body: str) -> None: """The stdlib POST — the ONLY socket path in this module (mirrors ``ingest._urllib_get``).""" request = Request( url, data=body.encode("utf-8"), headers={"Content-Type": "application/json"}, method="POST", ) with urlopen(request, timeout=10): pass class WebhookNotifier: """POSTs the verdict payload to a webhook URL — the single egress-bearing notifier. **Fail-closed gate (no-silent-egress):** construction refuses (``NotifyRefused``) unless the caller passes ``allow_egress=True`` explicitly, per run, in code — the kwarg has no default and is NEVER read from config (config cannot self-grant egress; mirrors ingest's ``allow_network`` gate at ``ingest.py:506``).""" def __init__(self, url: str, *, allow_egress: bool, post: WebhookPost = _urllib_post) -> None: if not allow_egress: raise NotifyRefused( "webhook notifier requires explicit per-run opt-in (allow_egress=True)" ) self._url = url self._post = post def __call__(self, verdict: Verdict) -> None: try: self._post(self._url, json.dumps(_verdict_payload(verdict), sort_keys=True)) except (URLError, OSError, ValueError, HTTPException) as exc: # STRICTER than ingest.py:314: NO exception type may carry the url past this point — # a Slack/Teams webhook URL embeds the receiver secret. The tuple covers every # URL-carrier _urllib_post can raise (ValueError incl. UnicodeError; HTTPException # incl. InvalidURL; OSError incl. URLError/HTTPError/TimeoutError). The chain is # severed (`from None`) because __cause__ renders in tracebacks/logging.exception; # the exception type name in the message is the diagnostic substitute for the lost # chain. Deliberately NOT a bare `except Exception`: the transport is injectable, # and a test transport's AssertionError/TypeError (programming errors) must surface. raise NotifyError(f"webhook POST failed ({type(exc).__name__})") from None # --- config + factory (fail-fast, egress never config-grantable) ---------------------------------- class NotifierConfig(BaseModel): """Declarative notifier choice. ``type`` membership is enforced by ``build_notifier`` (unknown type → ``ValueError``); the validator pins the per-type required fields. Deliberately NO ``allow_egress`` field — egress opt-in is a code-level factory kwarg, so config alone can never grant it (mirrors ingest's ``allow_network`` discipline).""" # the url field is secret-bearing (webhook URLs embed the receiver secret), so # validation errors must not echo input (pydantic would render input_value='…secret') model_config = ConfigDict(hide_input_in_errors=True) type: str url: str | None = None path: str | None = None @model_validator(mode="after") def _required_fields_by_type(self) -> NotifierConfig: if self.type == "webhook" and not self.url: raise ValueError("webhook notifier config requires a non-empty url") if self.type == "webhook" and self.url and not self.url.startswith(("http://", "https://")): raise ValueError("webhook notifier url must start with http:// or https://") if self.type == "file" and not self.path: raise ValueError("file notifier config requires a non-empty path") return self def build_notifier(config: NotifierConfig, *, allow_egress: bool = False) -> Notifier: """Type-dispatch factory. The webhook branch threads ``allow_egress`` into ``WebhookNotifier`` — with the default ``False`` the Step-3 gate fires (``NotifyRefused``), so a config file alone still cannot enable egress.""" if config.type == "console": return ConsoleNotifier() if config.type == "file": assert config.path is not None # guaranteed by _required_fields_by_type return FileNotifier(config.path) if config.type == "webhook": assert config.url is not None # guaranteed by _required_fields_by_type return WebhookNotifier(config.url, allow_egress=allow_egress) raise ValueError(f"unknown notifier type: {config.type!r}")