From 9701511754dd998a15b0e0919724fc04e47aa252 Mon Sep 17 00:00:00 2001 From: malice Date: Tue, 7 Jul 2026 01:51:41 -0600 Subject: [PATCH] feat(notifications): reusable delivery destinations (additive, inline fallback) (#83) Add NotificationDestination + config.notifications.destinations and a `destinations` reference list on toggles/rules. When a toggle/rule references destinations, delivery resolves from the shared destination; when empty, the existing inline-field delivery path runs UNCHANGED (zero regression). Lets email/webhook/mesh-channel be defined once and reused, de-duplicating the delivery config. UI to follow (C2). Co-authored-by: Matt Johnson Co-authored-by: Claude Opus 4.8 (1M context) --- work/meshai/config.py | 158 ++++++++ .../notifications/pipeline/dispatcher.py | 126 ++++++- work/tests/test_destinations_c1.py | 348 ++++++++++++++++++ 3 files changed, 612 insertions(+), 20 deletions(-) create mode 100644 work/tests/test_destinations_c1.py diff --git a/work/meshai/config.py b/work/meshai/config.py index 681dd39..13c87bd 100644 --- a/work/meshai/config.py +++ b/work/meshai/config.py @@ -592,6 +592,12 @@ class NotificationRuleConfig: webhook_url: str = "" webhook_headers: dict = field(default_factory=dict) + # Integration C1: names of shared NotificationDestinations. When NON-EMPTY, + # the dispatcher delivers via the resolved destinations instead of the inline + # fields above. When EMPTY (default / all pre-C1 rules) the inline path runs + # unchanged. Inline fields are retained as the fallback and never deleted. + destinations: list = field(default_factory=list) + # Behavior cooldown_minutes: int = 10 @@ -630,6 +636,9 @@ class NotificationToggle: recipients: list = field(default_factory=list) webhook_url: str = "" webhook_headers: dict = field(default_factory=dict) + # Integration C1: names of shared NotificationDestinations. NON-EMPTY => + # deliver via resolved destinations; EMPTY => existing inline path unchanged. + destinations: list = field(default_factory=list) TOGGLE_FAMILIES = [ @@ -698,6 +707,45 @@ class DigestConfig: include: list[str] = field(default_factory=list) # Toggle names to include (empty = default set) +@dataclass +class NotificationDestination: + """A named, reusable delivery target (Integration C1). + + Delivery config used to be DUPLICATED inline on every ``NotificationToggle`` + and ``NotificationRuleConfig`` (an operator configured the same SMTP/webhook + twice). A ``NotificationDestination`` lets a delivery target be defined ONCE + under ``NotificationsConfig.destinations`` and referenced by name from + toggles/rules via their ``destinations`` list. The field set mirrors the + inline delivery fields exactly, so a destination maps to ``create_channel`` + identically to an inline rule (only the fields relevant to ``type`` are used). + """ + + name: str = "" # id, referenced by toggles/rules + type: str = "mesh_broadcast" # mesh_broadcast|meshcore_broadcast|mesh_dm|meshcore_dm|email|webhook|digest + + # Mesh broadcast fields + broadcast_channel: Optional[int] = None + # Per-family MeshCore channel NAME on the companion; None = not on MeshCore. + meshcore_channel: Optional[str] = None + + # DM fields + node_ids: list = field(default_factory=list) + meshcore_dm_contacts: list = field(default_factory=list) + + # Email fields + smtp_host: str = "" + smtp_port: int = 587 + smtp_user: str = "" + smtp_password: str = "" + smtp_tls: bool = True + from_address: str = "" + recipients: list = field(default_factory=list) + + # Webhook fields + webhook_url: str = "" + webhook_headers: dict = field(default_factory=dict) + + @dataclass class NotificationsConfig: """Notification system settings.""" @@ -720,6 +768,108 @@ class NotificationsConfig: toggles: dict = field(default_factory=_default_toggles) # family -> NotificationToggle digest: DigestConfig = field(default_factory=DigestConfig) rules: list = field(default_factory=list) # List of NotificationRuleConfig + # Integration C1: named, reusable delivery targets (name -> NotificationDestination). + # Toggles/rules reference these by name via their own `destinations` list; + # empty reference list => the inline delivery fields are used (unchanged). + destinations: dict = field(default_factory=dict) + + +# Fields copied verbatim from an inline (toggle/rule) delivery config onto a +# NotificationDestination of a given type. Used only by synthesize_destinations. +_DEST_TYPE_FIELDS = { + "mesh_broadcast": ("broadcast_channel",), + "meshcore_broadcast": ("meshcore_channel",), + "mesh_dm": ("node_ids",), + "meshcore_dm": ("meshcore_dm_contacts",), + "email": ("smtp_host", "smtp_port", "smtp_user", "smtp_password", + "smtp_tls", "from_address", "recipients"), + "webhook": ("webhook_url", "webhook_headers"), +} + + +def synthesize_destinations(config) -> None: + """Populate ``config.notifications.destinations`` from the inline delivery + fields on toggles/rules, and set each entry's ``destinations`` reference + list -- a DISPLAY/opt-in convenience for the C2 UI ("convert to + destinations"). + + IMPORTANT — this is NOT wired into the dispatcher and MUST NOT be + auto-run as part of load/save. It is intentionally decoupled because it + is NOT guaranteed byte-identical for toggles: a toggle's per-severity + ``severity_channels`` routing (different channel TYPES at different + severities) collapses into a flat destination set, and the dispatcher's + destination path fires all referenced destinations subject only to the + ``min_severity`` floor (see dispatcher._dispatch_toggles). Only invoke it + from the UI on an explicit operator action, and only where that flattening + is acceptable. Default delivery stays on the inline fallback path. + + Idempotent: entries that already reference destinations are left untouched; + identical inline delivery configs are de-duplicated to a shared destination. + """ + notif = getattr(config, "notifications", None) + if notif is None: + return + registry = getattr(notif, "destinations", None) + if not isinstance(registry, dict): + registry = {} + notif.destinations = registry + + # signature -> destination name, so identical configs share one destination. + sig_to_name: dict = {} + for nm, dest in registry.items(): + sig_to_name.setdefault(_dest_signature(dest), nm) + + def _intern(dtype: str, source) -> str: + """Create-or-find a destination of dtype from source's inline fields.""" + dest = NotificationDestination(type=dtype) + for f in _DEST_TYPE_FIELDS.get(dtype, ()): # copy only relevant fields + setattr(dest, f, getattr(source, f, getattr(dest, f))) + sig = _dest_signature(dest) + if sig in sig_to_name: + return sig_to_name[sig] + name = f"{dtype}_{len([n for n in registry if registry[n].type == dtype]) + 1}" + while name in registry: + name += "_x" + dest.name = name + registry[name] = dest + sig_to_name[sig] = name + return name + + # Toggles: union of channel types across all severity rows (minus digest). + for tog in (getattr(notif, "toggles", None) or {}).values(): + if getattr(tog, "destinations", None): + continue + types = [] + for row in (getattr(tog, "severity_channels", None) or {}).values(): + for t in row: + if t != "digest" and t not in types: + types.append(t) + refs = [_intern(t, tog) for t in types if t in _DEST_TYPE_FIELDS] + if refs: + tog.destinations = refs + + # Rules: single delivery type. + for rule in (getattr(notif, "rules", None) or []): + if getattr(rule, "destinations", None): + continue + dtype = getattr(rule, "delivery_type", "") + if dtype in _DEST_TYPE_FIELDS: + rule.destinations = [_intern(dtype, rule)] + + +def _dest_signature(dest) -> tuple: + """Hashable identity of a destination's delivery-relevant fields.""" + dtype = getattr(dest, "type", "") + parts = [dtype] + for f in _DEST_TYPE_FIELDS.get(dtype, ()): + v = getattr(dest, f, None) + if isinstance(v, list): + v = tuple(v) + elif isinstance(v, dict): + v = tuple(sorted(v.items())) + parts.append((f, v)) + return tuple(parts) + @dataclass class DashboardConfig: @@ -993,6 +1143,14 @@ def _dict_to_dataclass(cls, data: dict): name: _dict_to_dataclass(NotificationToggle, t) if isinstance(t, dict) else t for name, t in value["toggles"].items() } + # Integration C1: destinations is a dict of name -> NotificationDestination. + # field_type is a bare `dict`, so the generic nested-dataclass handler + # never coerces it -- do it explicitly here (mirrors toggles above). + if "destinations" in value and isinstance(value["destinations"], dict): + notifications.destinations = { + name: _dict_to_dataclass(NotificationDestination, d) if isinstance(d, dict) else d + for name, d in value["destinations"].items() + } if "channels" in value and isinstance(value["channels"], list) and value["channels"]: _migrate_legacy_channels(notifications, value) kwargs[key] = notifications diff --git a/work/meshai/notifications/pipeline/dispatcher.py b/work/meshai/notifications/pipeline/dispatcher.py index 846cee2..ff791d2 100644 --- a/work/meshai/notifications/pipeline/dispatcher.py +++ b/work/meshai/notifications/pipeline/dispatcher.py @@ -261,22 +261,35 @@ class Dispatcher: ) return for rule in rules: - try: - channel = self._channel_factory(rule, self._connector) - payload = make_payload_from_event(event) - success = await channel.deliver(payload, rule) - if success: - self._logger.info( - f"Dispatched event {event.id} via {rule.delivery_type}" + # v0.16 (Integration C1): a rule that references reusable + # destinations fans out to each resolved destination; a rule with + # EMPTY `destinations` delivers via its own inline delivery_type + + # fields exactly as before (drule IS rule -> byte-identical). + if getattr(rule, "destinations", None): + # Opted in: fan out to resolved destinations only (unknown names + # are skipped; no silent fallback to the rule's inline fields). + dests = self._resolve_destinations(rule.destinations) + delivery = [self._destination_to_rule(d, event) for d in dests + if getattr(d, "type", "") != "digest"] + else: + delivery = [rule] + for drule in delivery: + try: + channel = self._channel_factory(drule, self._connector) + payload = make_payload_from_event(event) + success = await channel.deliver(payload, drule) + if success: + self._logger.info( + f"Dispatched event {event.id} via {drule.delivery_type}" + ) + else: + self._logger.warning( + f"Channel delivery returned False for rule {rule.name}" + ) + except Exception: + self._logger.exception( + f"Channel delivery failed for rule {rule.name}" ) - else: - self._logger.warning( - f"Channel delivery returned False for rule {rule.name}" - ) - except Exception: - self._logger.exception( - f"Channel delivery failed for rule {rule.name}" - ) async def _dispatch_toggles(self, event: Event) -> None: """Route an event through its family master-toggle (parallel to rules). @@ -373,9 +386,30 @@ class Dispatcher: event_rank = self.SEVERITY_RANK.get(event.severity, 0) if event_rank < self.SEVERITY_RANK.get(getattr(tog, "min_severity", "routine"), 0): return - sev_channels = getattr(tog, "severity_channels", None) or {} - ch_types = [c for c in sev_channels.get(event.severity, []) if c != "digest"] - if not ch_types: + # v0.16 (Integration C1) — destinations vs inline routing. + # If the toggle references reusable NotificationDestinations, deliver + # via those resolved destinations (each carries its own delivery type), + # gated ONLY by the region scope + min_severity floor already applied + # above. The severity_channels matrix is intentionally BYPASSED on the + # destination path: a shared destination is the single source of truth + # for its delivery type + fields. When `destinations` is EMPTY (all + # pre-C1 config) the existing severity_channels -> inline-field path + # runs completely unchanged (zero regression). + # Opt-in is decided by the presence of a reference list, NOT by whether + # it resolves: a toggle that references destinations but whose names are + # unknown delivers NOTHING (it does not silently fall back to the inline + # fields, which would broadcast stale/duplicate config unexpectedly). + if getattr(tog, "destinations", None): + dests = self._resolve_destinations(tog.destinations) + # digest-typed destinations belong to the digest scheduler, not the + # live broadcast path (mirrors the inline "digest" exclusion below). + delivery_plan = [("dest", d) for d in dests + if getattr(d, "type", "") != "digest"] + else: + sev_channels = getattr(tog, "severity_channels", None) or {} + ch_types = [c for c in sev_channels.get(event.severity, []) if c != "digest"] + delivery_plan = [("toggle", ct) for ct in ch_types] + if not delivery_plan: return # ---------- Section 3 — per-toggle cooldown (check only) ---------- @@ -454,11 +488,16 @@ class Dispatcher: pass delivered_any = False - for ch_type in ch_types: + for _kind, _item in delivery_plan: rule = None payload = None try: - rule = self._toggle_to_rule(tog, ch_type, event) + if _kind == "dest": + ch_type = getattr(_item, "type", "") + rule = self._destination_to_rule(_item, event) + else: + ch_type = _item + rule = self._toggle_to_rule(tog, ch_type, event) channel = self._channel_factory(rule, self._connector) if friendly is not None and ch_type in ( "mesh_broadcast", "mesh_dm", "meshcore_broadcast", "meshcore_dm" @@ -733,6 +772,53 @@ class Dispatcher: "post-broadcast: handler commit-callback raised" ) + def _resolve_destinations(self, names) -> list: + """Resolve destination NAMES -> NotificationDestination objects via + config.notifications.destinations (Integration C1). + + Returns [] for an empty/None input so callers fall back to the inline + delivery path unchanged. Unknown names are skipped with a warning (never + raised) so a dangling reference degrades gracefully rather than dropping + the whole broadcast. + """ + if not names: + return [] + registry = getattr(self._config.notifications, "destinations", None) + if not isinstance(registry, dict) or not registry: + self._logger.warning( + "dispatcher: destinations referenced (%s) but none configured", names) + return [] + out = [] + for nm in names: + dest = registry.get(nm) + if dest is None: + self._logger.warning( + "dispatcher: unknown destination %r referenced; skipping", nm) + continue + out.append(dest) + return out + + def _destination_to_rule(self, dest, event: Event): + """Synthesize a NotificationRuleConfig from a shared destination's + fields (mirrors _toggle_to_rule) so it maps to create_channel identically + to an inline rule.""" + from meshai.config import NotificationRuleConfig + return NotificationRuleConfig( + name=f"dest:{getattr(dest, 'name', '')}", + enabled=True, trigger_type="condition", + delivery_type=getattr(dest, "type", ""), + broadcast_channel=(getattr(dest, "broadcast_channel", None) or 0), + meshcore_channel=getattr(dest, "meshcore_channel", None), + node_ids=list(getattr(dest, "node_ids", []) or []), + meshcore_dm_contacts=list(getattr(dest, "meshcore_dm_contacts", []) or []), + smtp_host=getattr(dest, "smtp_host", ""), smtp_port=getattr(dest, "smtp_port", 587), + smtp_user=getattr(dest, "smtp_user", ""), smtp_password=getattr(dest, "smtp_password", ""), + smtp_tls=getattr(dest, "smtp_tls", True), from_address=getattr(dest, "from_address", ""), + recipients=list(getattr(dest, "recipients", []) or []), + webhook_url=getattr(dest, "webhook_url", ""), + webhook_headers=dict(getattr(dest, "webhook_headers", {}) or {}), + ) + def _toggle_to_rule(self, tog, ch_type: str, event: Event): from meshai.config import NotificationRuleConfig return NotificationRuleConfig( diff --git a/work/tests/test_destinations_c1.py b/work/tests/test_destinations_c1.py new file mode 100644 index 0000000..8beb8e2 --- /dev/null +++ b/work/tests/test_destinations_c1.py @@ -0,0 +1,348 @@ +"""Integration C1 — reusable NotificationDestinations (additive, inline fallback). + +The delivery config was DUPLICATED inline on every NotificationToggle and +NotificationRuleConfig. C1 introduces a shared, named NotificationDestination +that toggles/rules reference by name. The safety contract is ADDITIVE + +regression-free: + + * EMPTY `destinations` (all pre-C1 config) -> the existing inline-field + delivery path runs BYTE-IDENTICALLY (proven here against _toggle_to_rule / + the rule object itself). + * NON-EMPTY `destinations` -> delivery resolves from the shared destination's + fields, gated only by region scope + the min_severity floor. The + severity_channels matrix is intentionally BYPASSED on the destination path. +""" + +import asyncio +import dataclasses + +import pytest + +from meshai.config import ( + Config, + NotificationDestination, + NotificationRuleConfig, + load_config, + save_config, + synthesize_destinations, +) +from meshai.notifications.pipeline.dispatcher import Dispatcher +from meshai.notifications.events import make_event + + +class RecChannel: + """Records the FULL rule object handed to create_channel/deliver so tests + can assert delivery params are byte-identical.""" + + def __init__(self, rule, rec, succeed=True): + self._rule = rule + self._rec = rec + self._succeed = succeed + + async def deliver(self, payload, rule): + self._rec.append({ + "delivery_type": rule.delivery_type, + "rule": rule, + "message": payload.message, + }) + return self._succeed + + +def _disp(cfg, succeed=True): + rec: list = [] + d = Dispatcher(cfg, + lambda rule, conn: RecChannel(rule, rec, succeed), + connector=None) + return d, rec + + +def _base_cfg(): + cfg = Config() + cfg.notifications.rules = [] + cfg.notifications.cold_start_grace_seconds = 0 + return cfg + + +def _weather_ev(eid="ev-1", severity="priority"): + ev = make_event(source="nws", category="weather.alert.severe", + severity=severity, title="t") + ev.id = eid + return ev + + +def _rule_fields(rule): + """Delivery-relevant subset for byte-identical comparison.""" + return { + k: getattr(rule, k) for k in ( + "delivery_type", "broadcast_channel", "meshcore_channel", + "node_ids", "meshcore_dm_contacts", "smtp_host", "smtp_port", + "smtp_user", "smtp_password", "smtp_tls", "from_address", + "recipients", "webhook_url", "webhook_headers", + ) + } + + +# --------------------------------------------------------------- inline path + + +def test_inline_toggle_delivery_is_byte_identical(): + """A toggle with inline delivery fields + EMPTY destinations must deliver + via the EXACT same synthesized rule params as the unchanged inline path + (_toggle_to_rule).""" + cfg = _base_cfg() + t = cfg.notifications.toggles["weather"] + t.enabled = True + t.min_severity = "routine" + t.cooldown_seconds = 0 + t.severity_channels = {"priority": ["mesh_broadcast", "email", "webhook"]} + t.broadcast_channel = 3 + t.smtp_host = "smtp.example.com" + t.smtp_port = 2525 + t.smtp_user = "u" + t.smtp_password = "p" + t.from_address = "alerts@example.com" + t.recipients = ["ops@example.com", "oncall@example.com"] + t.webhook_url = "https://hook.example.com/x" + t.webhook_headers = {"X-Token": "abc"} + assert t.destinations == [] # opted out -> inline fallback + + d, rec = _disp(cfg) + ev = _weather_ev(severity="priority") + asyncio.run(d.dispatch(ev)) + + # One delivery per channel type in the matrix row, same order. + assert [r["delivery_type"] for r in rec] == \ + ["mesh_broadcast", "email", "webhook"] + + # Each delivered rule is byte-identical to the unchanged _toggle_to_rule + # construction -> proves the inline path routes exactly as before. + for ct, r in zip(["mesh_broadcast", "email", "webhook"], rec): + expected = _rule_fields(d._toggle_to_rule(t, ct, ev)) + assert _rule_fields(r["rule"]) == expected + + +def test_inline_rule_delivery_unchanged_when_no_destinations(): + """A condition rule with inline delivery + EMPTY destinations delivers via + the rule object itself (drule IS rule).""" + cfg = _base_cfg() + rule = NotificationRuleConfig( + name="ops-email", enabled=True, trigger_type="condition", + categories=["weather.alert.severe"], min_severity="routine", + delivery_type="email", smtp_host="smtp.example.com", + recipients=["ops@example.com"], from_address="a@example.com", + ) + assert rule.destinations == [] + cfg.notifications.rules = [rule] + + d, rec = _disp(cfg) + asyncio.run(d.dispatch(_weather_ev(severity="immediate"))) + assert len(rec) == 1 + # The delivered rule is the SAME object -> byte-identical, zero regression. + assert rec[0]["rule"] is rule + + +# ----------------------------------------------------------- destination path + + +def test_toggle_destination_path_uses_destination_params(): + """A toggle referencing a destination delivers via the DESTINATION's fields, + not the toggle's inline fields (inline values differ to prove it).""" + cfg = _base_cfg() + cfg.notifications.destinations = { + "email_ops": NotificationDestination( + name="email_ops", type="email", + smtp_host="dest-smtp.example.com", smtp_port=465, smtp_user="du", + smtp_password="dp", from_address="dest@example.com", + recipients=["dest-ops@example.com"], + ), + } + t = cfg.notifications.toggles["weather"] + t.enabled = True + t.min_severity = "routine" + t.cooldown_seconds = 0 + t.destinations = ["email_ops"] + # Divergent inline values that MUST be ignored on the destination path. + t.severity_channels = {"priority": ["mesh_broadcast"]} + t.smtp_host = "INLINE-should-not-be-used" + t.recipients = ["inline@example.com"] + + d, rec = _disp(cfg) + asyncio.run(d.dispatch(_weather_ev(severity="priority"))) + + assert len(rec) == 1 + r = rec[0]["rule"] + assert r.delivery_type == "email" + assert r.smtp_host == "dest-smtp.example.com" + assert r.smtp_port == 465 + assert r.recipients == ["dest-ops@example.com"] + assert r.from_address == "dest@example.com" + + +def test_toggle_destinations_bypass_severity_channels_matrix(): + """The destination path is gated by the min_severity FLOOR only; an empty + severity_channels matrix does NOT suppress it (matrix is bypassed).""" + cfg = _base_cfg() + cfg.notifications.destinations = { + "mesh_a": NotificationDestination( + name="mesh_a", type="mesh_broadcast", broadcast_channel=7), + } + t = cfg.notifications.toggles["weather"] + t.enabled = True + t.min_severity = "priority" + t.cooldown_seconds = 0 + t.destinations = ["mesh_a"] + t.severity_channels = {} # empty matrix would kill the inline path + + d, rec = _disp(cfg) + # Above the floor -> fires despite empty matrix. + asyncio.run(d.dispatch(_weather_ev(eid="a", severity="priority"))) + assert len(rec) == 1 + assert rec[0]["rule"].delivery_type == "mesh_broadcast" + assert rec[0]["rule"].broadcast_channel == 7 + + +def test_toggle_destinations_respect_min_severity_floor(): + """Below the floor, the destination path does not fire (floor still gates).""" + cfg = _base_cfg() + cfg.notifications.destinations = { + "mesh_a": NotificationDestination( + name="mesh_a", type="mesh_broadcast", broadcast_channel=7), + } + t = cfg.notifications.toggles["weather"] + t.enabled = True + t.min_severity = "immediate" + t.cooldown_seconds = 0 + t.destinations = ["mesh_a"] + t.severity_channels = {} + + d, rec = _disp(cfg) + asyncio.run(d.dispatch(_weather_ev(severity="priority"))) # below floor + assert rec == [] + + +def test_toggle_multiple_destinations_all_fire(): + """A toggle may fan out to several destinations of mixed types; digest-typed + destinations are skipped on the live path.""" + cfg = _base_cfg() + cfg.notifications.destinations = { + "mesh_a": NotificationDestination( + name="mesh_a", type="mesh_broadcast", broadcast_channel=1), + "email_ops": NotificationDestination( + name="email_ops", type="email", smtp_host="s", recipients=["r"]), + "daily": NotificationDestination(name="daily", type="digest"), + } + t = cfg.notifications.toggles["weather"] + t.enabled = True + t.min_severity = "routine" + t.cooldown_seconds = 0 + t.destinations = ["mesh_a", "email_ops", "daily"] + + d, rec = _disp(cfg) + asyncio.run(d.dispatch(_weather_ev(severity="priority"))) + # digest is excluded from the live broadcast path. + assert sorted(r["delivery_type"] for r in rec) == ["email", "mesh_broadcast"] + + +def test_unknown_destination_reference_is_skipped_not_fatal(): + """A dangling destination name degrades gracefully (skipped, no crash) and, + when it leaves nothing to deliver, drops rather than falling back to inline.""" + cfg = _base_cfg() + cfg.notifications.destinations = {} # registry empty + t = cfg.notifications.toggles["weather"] + t.enabled = True + t.min_severity = "routine" + t.cooldown_seconds = 0 + t.destinations = ["nope"] + t.broadcast_channel = 9 # inline present but not used + + d, rec = _disp(cfg) + asyncio.run(d.dispatch(_weather_ev(severity="priority"))) + assert rec == [] + + +def test_rule_destination_fanout(): + """A condition rule referencing destinations fans out to each; the rule's + own inline delivery_type is superseded.""" + cfg = _base_cfg() + cfg.notifications.destinations = { + "mesh_a": NotificationDestination( + name="mesh_a", type="mesh_broadcast", broadcast_channel=2), + "email_ops": NotificationDestination( + name="email_ops", type="email", smtp_host="s", recipients=["r"]), + } + rule = NotificationRuleConfig( + name="fanout", enabled=True, trigger_type="condition", + categories=["weather.alert.severe"], min_severity="routine", + delivery_type="webhook", webhook_url="https://inline-not-used", + destinations=["mesh_a", "email_ops"], + ) + cfg.notifications.rules = [rule] + + d, rec = _disp(cfg) + asyncio.run(d.dispatch(_weather_ev(severity="immediate"))) + assert sorted(r["delivery_type"] for r in rec) == ["email", "mesh_broadcast"] + assert all(r["rule"] is not rule for r in rec) # synthesized, not inline + + +# ------------------------------------------------------------------ round-trip + + +def test_destinations_roundtrip_save_load(tmp_path): + """destinations dict + toggle.destinations + rule.destinations survive a + YAML save/load cycle as real dataclasses / lists.""" + cfg = _base_cfg() + cfg.notifications.destinations = { + "email_ops": NotificationDestination( + name="email_ops", type="email", smtp_host="smtp.example.com", + smtp_port=465, recipients=["a@example.com", "b@example.com"], + webhook_headers={}, + ), + "mesh_a": NotificationDestination( + name="mesh_a", type="mesh_broadcast", broadcast_channel=4), + } + cfg.notifications.toggles["weather"].destinations = ["email_ops", "mesh_a"] + cfg.notifications.rules = [NotificationRuleConfig( + name="r1", destinations=["email_ops"])] + + p = tmp_path / "config.yaml" + save_config(cfg, p) + loaded = load_config(p) + + dests = loaded.notifications.destinations + assert set(dests) == {"email_ops", "mesh_a"} + assert isinstance(dests["email_ops"], NotificationDestination) + assert dests["email_ops"].smtp_port == 465 + assert dests["email_ops"].recipients == ["a@example.com", "b@example.com"] + assert dests["mesh_a"].broadcast_channel == 4 + assert loaded.notifications.toggles["weather"].destinations == \ + ["email_ops", "mesh_a"] + assert loaded.notifications.rules[0].destinations == ["email_ops"] + + +# ----------------------------------------------- synthesize_destinations helper + + +def test_synthesize_destinations_populates_refs_and_dedups(): + """The C2 migration helper creates named destinations from inline fields and + de-duplicates identical configs. It is NOT auto-run by delivery.""" + cfg = _base_cfg() + # Two rules with an IDENTICAL email delivery -> should share one destination. + common = dict(delivery_type="email", smtp_host="s", from_address="f", + recipients=["r"]) + cfg.notifications.rules = [ + NotificationRuleConfig(name="r1", **common), + NotificationRuleConfig(name="r2", **common), + ] + for t in cfg.notifications.toggles.values(): + t.severity_channels = {} # keep toggle synthesis empty for this test + + synthesize_destinations(cfg) + + assert cfg.notifications.rules[0].destinations + # Same inline config -> same shared destination name. + assert cfg.notifications.rules[0].destinations == \ + cfg.notifications.rules[1].destinations + ref = cfg.notifications.rules[0].destinations[0] + dest = cfg.notifications.destinations[ref] + assert dest.type == "email" + assert dest.recipients == ["r"]