feat(s52): fail-fast NotifierConfig + build_notifier factory (egress gate threaded)

This commit is contained in:
Kjell Tore Guttormsen 2026-07-16 19:51:57 +02:00
commit 1267f6c3eb
2 changed files with 64 additions and 1 deletions

View file

@ -6,7 +6,7 @@ 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): pure stdlib. ``Verdict`` is imported ONLY under ``TYPE_CHECKING`` and
**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
@ -23,6 +23,8 @@ 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, model_validator
if TYPE_CHECKING: # verdicts imports agent_framework — keep it out of the runtime import graph
from portfolio_optimiser.verdicts import Verdict
@ -128,3 +130,40 @@ class WebhookNotifier:
# STRICTER than ingest.py:314: the message NEVER carries the url — a Slack/Teams
# webhook URL embeds the receiver secret (the original cause stays chained).
raise NotifyError("webhook POST failed") from exc
# --- 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)."""
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 == "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}")