test(toggles): update severity_threshold for v0.7 matrix-only semantics

v0.7 removed min_severity as a guard; the severity_channels matrix IS
the threshold. Empty matrix row = no delivery.

Updated tests to use empty matrix rows instead of min_severity.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

docs/routing-simplification.md

@@ -0,0 +1,244 @@
+# MeshAI Routing Simplification
+
+**Goal:** An alert family should require ONE place to configure, not four interacting
+settings (toggle enabled + min_severity threshold + severity_channels matrix +
+per-family transport config).
+
+**Reviewed at:** main @ 98e3fcf (2026-06-10)
+
+---
+
+## PART 1 — How routing actually works today
+
+There are **five parallel delivery paths**, not one system with layers.
+
+### Path 1: Legacy NotificationRouter (`notifications/router.py`)
+Fed by `main._dispatch_alerts()` (main.py:644-684) ← `alert_engine.check()` — mesh-health
+and env-poll alerts as raw dicts. Iterates `config.notifications.rules`
+(NotificationRuleConfig): category match → `min_severity` gate → per-rule
+`cooldown_minutes` dedup keyed on `(rule_name, category, event_id-or-message[:50])`
+(router.py:165-172) → builds channel from the rule's **inline transport fields** → delivers.
+- Writes **no** `mesh_broadcasts_out` audit row.
+- Rule stats in `/data/rule_stats.json` (router.py:24), separate from SQLite.
+- Mesh messages >200 chars go through **LLM summarization** (router.py:187-193).
+- Also backs the over-the-mesh `!subscribe` command — `add_mesh_subscription()`
+  (router.py:737) dynamically creates a mesh_dm rule per subscriber node — and
+  `generate_report()` used by report delivery (router.py:628).
+
+### Path 2: Dispatcher rules path (`pipeline/dispatcher.py:252-279`)
+Fed by the EventBus. Evaluates **the same rules list** against bus Events
+(`_matching_rules`, dispatcher.py:632-653: enabled → condition → category →
+min_severity → region_scope). Then delivers.
+- **No cooldown. No dedup. No staleness. No audit row.** None of the toggle-path
+  guards apply here.
+
+### Path 3: Dispatcher toggle path (`pipeline/dispatcher.py:281-456`) — the good one
+Fed by the EventBus through the pipeline: `bus → Inhibitor (inhibit_keys, ttl 1800s,
+persisted) → Grouper (group_key window 60s, persisted) → ToggleFilter (family enabled
+set) → tee(dispatch + DigestAccumulator)`.
+
+Inside `_dispatch_toggles`, in order:
+0. Cold-start grace (60s after first event, persisted anchor)
+1. Staleness — `toggle.freshness_seconds` (fire family reads `wfigs.freshness_seconds`)
+2. Cooldown — per `(toggle, category, region|_cooldown_suffix)`, persisted; **immediate
+   severity bypasses** (dispatcher.py:363-364)
+3. Dedup — `(source, event.id)` LRU 10k, 7-day SQLite window
+4. Region scope → **`min_severity` gate (line 420-422)** → **`severity_channels`
+   matrix (line 434-435)** → composer (150-byte budget) → channel per matrix entry →
+   `_post_broadcast_commit`: `mesh_broadcasts_out` audit row + handler callback.
+
+### Path 4: Scheduled broadcasts (`dispatch_scheduled_broadcast`, dispatcher.py:475-561)
+Band conditions (3×/day), fire digest (2×/day), reminders — all bypass the pipeline.
+Only cold-start grace applies. **All three hardcode-route through
+`rf_propagation.broadcast_channel`** (dispatcher.py:512-519). Writes audit row.
+
+### Path 5: Fallbacks
+`mesh_intelligence.alert_channel` + subscriber DMs when no NotificationRouter
+(main.py:660-682); SubscriptionManager scheduled DM reports (main.py:686+).
+
+### Pre-pipeline gates (upstream of all of the above)
+Adapter floors (swpc kp/flare/proton), handler change-detection + cooldowns (fire 8h
+update etc.), consumer **default-deny** — no synthesized wire string → Event never
+enters the bus (consumer.py:548-567).
+
+---
+
+## PART 2 — Verified defects and redundancies
+
+Every item re-checked against source.
+
+| # | Finding | Evidence |
+|---|---------|----------|
+| B1 | **Threshold + matrix double-gate.** `min_severity` gates before `severity_channels`; a routine matrix row is dead config when threshold=priority. Exactly the GUI trap from the screenshot. | dispatcher.py:420-422 vs 434-435 |
+| B2 | **The "digest" matrix column is a no-op.** Dispatcher skips it (`if ch_type == "digest": continue`); digest membership is actually `digest.include` (toggle-name list). Checking the box changes nothing. By design per `test_digest_channel_skipped_in_live_dispatch`, but the GUI presents it as live routing. | dispatcher.py:436-437; pipeline/__init__.py include_toggles |
+| B3 | **Rules-path deliveries are invisible.** Neither Path 1 nor Path 2 writes `mesh_broadcasts_out`. Forensics on that table only sees toggle + scheduled traffic. | router.py:208; dispatcher.py:263-279 |
+| B4 | **Path 2 has zero spam protection.** A rule matching a chatty central category re-delivers every event. | dispatcher.py:252-279 |
+| B5 | **Double-delivery by design.** `dispatch()` runs rules AND toggles; both matching = same event broadcast twice, possibly same channel. Tested as intended behavior (`test_rules_and_toggles_both_fire`). | dispatcher.py:247-250 |
+| B6 | **Channel 0 is falsy.** `if rf is None or not getattr(rf, "broadcast_channel", None)` — `rf_propagation.broadcast_channel = 0` silently drops ALL scheduled broadcasts (band conditions, fire digest, reminders). Channel 0 is a legitimate primary channel. | dispatcher.py:515 |
+| B7 | **Fire digest + reminders ride the RF toggle's channel.** Fire content routed by RF-propagation transport config; disabling/misconfiguring the RF toggle silently kills fire digests. | dispatcher.py:512-519; fire_digest.py:255; reminders/__init__.py:315 |
+| B8 | **Severity fails open** in the legacy router — unknown severity string returns True. | router.py:223-224 |
+| B9 | **Transport config duplicated everywhere.** Full SMTP credential block inline in every rule AND every toggle (config.py:503-556, 558-580). The stale-SMTP-per-rule failure mode is structural. |
+| B10 | **Two mesh formatting regimes.** Legacy: LLM-summarize >200 chars. Toggle: deterministic composer, 150-byte budget. Same radio, different text rules. | router.py:187-193 vs composer.py:31 |
+| B11 | **Three incompatible dedup keys** (legacy `(rule,cat,event_id|msg[:50])` in-memory; toggle `(source,event.id)` persisted; rules-path none). |
+| B12 | GUI copy bug: severity helper says "Warning" recommended — not a severity level in this system (routine/priority/immediate). Same line still has stale `text-slate-600`. | Notifications.tsx:605 |
+| B13 | **Guard-ordering trap.** In `_dispatch_toggles`, cooldown is armed (Section 2) and dedup recorded (Section 3) BEFORE the region filter, `min_severity` gate, and `severity_channels` matrix lookup. A below-threshold or wrong-region event consumes the cooldown window and writes a 7-day persisted dedup row, suppressing later events that WOULD deliver — including after the operator fixes config. | dispatcher.py Sections 2-3 vs region/severity/matrix checks |
+
+---
+
+## PART 3 — The simplification
+
+Two changes that together reduce per-family config from four places to one.
+
+### Sinks — destinations defined once
+
+New `SinkConfig` dataclass + `sinks: dict[str, SinkConfig]` on NotificationsConfig:
+```yaml
+notifications:
+  sinks:
+    mesh-primary:  {type: mesh_broadcast, channel: 0}
+    mesh-alerts:   {type: mesh_broadcast, channel: 2}
+    dm-ops:        {type: mesh_dm, node_ids: ["!abcd1234"]}
+    email-ops:     {type: email, smtp_host: ..., recipients: [...]}
+```
+
+- `channels.py` gains `create_channel_from_sink(sink, connector)`. Existing channel
+  classes unchanged.
+- All transport fields (`broadcast_channel`, `node_ids`, `smtp_*`, `webhook_*`)
+  **removed** from NotificationToggle and NotificationRuleConfig (kept dataclass-side
+  only during migration window).
+- Channel index stored as `int`, validated `>= 0` — kills B6's falsy-zero class of bug.
+- **Status:** SinkConfig dataclass, `create_channel_from_sink()` factory, and migration
+  script (`scripts/migrate_config_routing.py`) are implemented on this branch.
+
+### Matrix-only severity — one panel per family
+
+- Delete `NotificationToggle.min_severity`. The `severity_channels` matrix becomes the
+  only gate; values are **sink names**, not channel types:
+  `severity_channels: {routine: [], priority: [mesh-primary], immediate: [mesh-primary, dm-ops]}`
+- Empty list = that severity doesn't deliver. Threshold semantics are now expressible,
+  visible, and conflict-free (kills B1).
+- Remove the "digest" pseudo-channel from the matrix (kills B2). Digest membership
+  stays `digest.include`; GUI gets a separate per-family "include in digest" checkbox
+  wired to it honestly.
+- Migration: for each existing toggle, blank matrix rows below old `min_severity`,
+  map `mesh_broadcast` → auto-created sink from its `broadcast_channel`, etc.
+- **Session 2 MUST** move the cooldown commit and dedup record to after the delivery
+  decision (region + matrix resolution). Matrix-only semantics inherit the suppression
+  trap otherwise: an event hitting an empty matrix row must not burn its dedup slot
+  or arm a cooldown. (See B13.)
+- **Status:** Not yet implemented.
+
+### GUI changes (when both halves are complete)
+
+- New **Sinks** section (one-time setup, test button per sink — `test_connection()`
+  already exists per channel class).
+- Family cards shrink to: enabled, regions, freshness, cooldown, matrix of
+  severity → sink multi-select, digest-include checkbox. Threshold dropdown, channel
+  config block, SMTP block all deleted from the card.
+- Fix B12 copy + stale slate class while in there.
+
+---
+
+## Amendments
+
+### Amendment A1: Multi-file config layout
+
+**Finding:** The live system uses a **multi-file config layout** where
+`/data/config/notifications.yaml` IS the notifications config directly — it contains
+`toggles:`, `rules:`, etc. at the root level with **no `notifications:` wrapper key**.
+
+This differs from the monolithic layout shown in the plan examples, where notifications
+config would be nested under a `notifications:` key in a larger config file.
+
+**Impact:**
+- Migration scripts must detect layout: if `"notifications" in config`, use nested
+  access; if `"toggles" in config or "rules" in config`, treat file root as the
+  notifications block.
+- `load_notifications_config()` in `migrate_config_routing.py` implements this detection.
+- GUI/API code reading `config.notifications.sinks` is unaffected — the `_dict_to_dataclass`
+  conversion handles both layouts identically at runtime.
+
+**Code pattern:**
+```python
+if "notifications" in config:
+    notifications = config["notifications"]
+elif "toggles" in config or "rules" in config:
+    notifications = config  # Multi-file: file IS the notifications config
+else:
+    notifications = {}
+```
+
+*Recorded: 2026-06-10*
+
+### Amendment A2: Layout-aware append-only sinks write
+
+**Finding:** The original `write_sinks_to_config()` violated Amendment A1 — it
+unconditionally wrote `config["notifications"]["sinks"]`, creating a bogus
+`notifications:` wrapper in multi-file configs where the file root IS the
+notifications block. Additionally, the `yaml.safe_load` → `yaml.dump` round-trip
+destroyed comments and key ordering.
+
+**Fix (implemented on this branch):**
+1. **Shared layout detection.** `detect_config_layout(config)` returns `"monolithic"`,
+   `"multifile"`, or `"empty"`. Both `load_notifications_config()` and
+   `write_sinks_to_config()` use this helper — they cannot diverge.
+2. **Append-only write.** `render_sinks_yaml(sinks, layout)` produces the exact text
+   to append (indented for monolithic, root-level for multifile). The original file
+   content is preserved byte-for-byte; only the sinks block is appended.
+3. **Post-write verification.** After writing, `verify_sinks_written()` re-parses
+   the file and asserts: (a) valid YAML, (b) sinks accessible at expected path,
+   (c) original top-level keys intact. On failure, the backup is restored
+   automatically.
+4. **Dry-run transparency.** `--dry-run` now prints the detected layout, the path
+   where sinks will live (`root-level sinks:` vs `notifications.sinks`), and the
+   exact text that would be appended.
+
+**Code pattern:**
+```python
+layout = detect_config_layout(config)
+sinks_yaml = render_sinks_yaml(sinks, layout)
+new_content = original_content + "\n" + sinks_yaml
+# ... write, then verify_sinks_written() ...
+```
+
+*Recorded: 2026-06-11*
+
+### Amendment A3: Guard reorder (B13) + backwards compatibility
+
+**Finding (B13):** The previous `_dispatch_toggles` order armed cooldown and recorded dedup
+BEFORE the region filter, `min_severity` gate, and `severity_channels` matrix lookup.
+An event failing these later checks would still burn its cooldown window and dedup slot,
+suppressing later events that WOULD deliver.
+
+**Fix (implemented on this branch):**
+1. **Guard reorder.** New order: cold-start → staleness → region filter → matrix resolution
+   → IF empty sink list, return WITHOUT arming cooldown/dedup → cooldown → dedup → deliver.
+2. **Sink-name routing.** `severity_channels` values are now sink names resolved against
+   `config.notifications.sinks`. Matrix becomes the only severity gate (`min_severity` removed).
+3. **Backwards compatibility (TEMPORARY).** The dispatcher supports BOTH formats during migration:
+   - v0.7+ sink names: resolved against `sinks` config
+   - v0.5/v0.6 channel types (`mesh_broadcast`, `mesh_dm`, etc.): falls back to
+     `_toggle_to_rule` using inline transport config from the toggle
+   - **Deprecation warning:** When legacy fallback is exercised, logs a warning once per
+     toggle per boot: "DEPRECATED: toggle 'X' uses channel-type routing..."
+   - **Removal:** This fallback will be deleted in session 3. The migration script
+     rewrites all matrices to sink names, so the fallback is cold post-migration.
+
+**Code pattern (dispatcher):**
+```python
+for sink_name in sink_names:
+    sink = sinks_config.get(sink_name)
+    if sink is not None:
+        resolved_sinks.append((sink_name, sink, False))  # v0.7 sink
+    elif sink_name in LEGACY_CHANNEL_TYPES:
+        resolved_sinks.append((sink_name, sink_name, True))  # v0.5/v0.6 legacy
+    else:
+        logger.warning(f"unknown sink '{sink_name}'")
+```
+
+**Migration script extended:** Phase B adds matrix rewrite:
+- Converts channel types to sink names
+- Blanks matrix rows below old `min_severity`
+- Removes `min_severity` field from toggles
+
+*Recorded: 2026-06-11*

meshai/config.py

@@ -499,6 +499,53 @@ class EnvironmentalConfig:
     geocoder: GeocoderConfig = field(default_factory=GeocoderConfig)
 
 
+@dataclass
+class SinkConfig:
+    """Named notification sink — transport defined once, referenced by name.
+
+    Routing simplification: transports defined once in sinks block,
+    referenced by name in toggles/rules. See docs/routing-simplification.md.
+    """
+
+    type: str = "mesh_broadcast"  # mesh_broadcast|mesh_dm|email|webhook
+
+    # Mesh broadcast
+    channel: int = 0  # Channel index (>= 0, 0 is valid primary channel)
+
+    # Mesh DM
+    node_ids: list = field(default_factory=list)
+
+    # Email (SMTP)
+    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
+    webhook_url: str = ""
+    webhook_headers: dict = field(default_factory=dict)
+
+    def validate(self) -> list[str]:
+        """Validate sink config, return list of errors."""
+        errors = []
+        if self.type not in ("mesh_broadcast", "mesh_dm", "email", "webhook"):
+            errors.append(f"Invalid sink type: {self.type}")
+        if self.type == "mesh_broadcast" and self.channel < 0:
+            errors.append(f"Channel must be >= 0, got {self.channel}")
+        if self.type == "mesh_dm" and not self.node_ids:
+            errors.append("mesh_dm sink requires node_ids")
+        if self.type == "email" and not self.smtp_host:
+            errors.append("email sink requires smtp_host")
+        if self.type == "email" and not self.recipients:
+            errors.append("email sink requires recipients")
+        if self.type == "webhook" and not self.webhook_url:
+            errors.append("webhook sink requires webhook_url")
+        return errors
+
+
 @dataclass
 class NotificationRuleConfig:
     """Self-contained notification rule with inline delivery config."""
@@ -642,6 +689,7 @@ 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
+    sinks: dict = field(default_factory=dict)  # name -> SinkConfig
 
 @dataclass
 class DashboardConfig:
@@ -785,6 +833,11 @@ 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()
                 }
+            if "sinks" in value and isinstance(value["sinks"], dict):
+                notifications.sinks = {
+                    name: _dict_to_dataclass(SinkConfig, s) if isinstance(s, dict) else s
+                    for name, s in value["sinks"].items()
+                }
             if "channels" in value and isinstance(value["channels"], list) and value["channels"]:
                 _migrate_legacy_channels(notifications, value)
             kwargs[key] = notifications

meshai/dashboard/api/config_routes.py

@@ -332,3 +332,163 @@ def _diff_keys(before, after, *, prefix: str) -> list[str]:
 
     walk(before, after, prefix)
     return sorted(out)
+
+
+# ---- v0.7 Sink management endpoints ----------------------------------------
+
+
+@router.get("/sinks")
+async def list_sinks(request: Request):
+    """List all configured sinks."""
+    config = request.app.state.config
+    sinks = getattr(config.notifications, "sinks", {}) or {}
+    return {
+        name: _dataclass_to_dict(sink) if hasattr(sink, "__dataclass_fields__") else sink
+        for name, sink in sinks.items()
+    }
+
+
+@router.get("/sinks/{name}")
+async def get_sink(name: str, request: Request):
+    """Get a specific sink by name."""
+    config = request.app.state.config
+    sinks = getattr(config.notifications, "sinks", {}) or {}
+    sink = sinks.get(name)
+    if sink is None:
+        raise HTTPException(status_code=404, detail=f"Sink '{name}' not found")
+    return _dataclass_to_dict(sink) if hasattr(sink, "__dataclass_fields__") else sink
+
+
+@router.put("/sinks/{name}")
+async def upsert_sink(name: str, request: Request):
+    """Create or update a sink.
+
+    Validates the sink config and persists through the /api/config write path.
+    """
+    from meshai.config import SinkConfig
+
+    config_path = request.app.state.config_path
+    if not config_path:
+        raise HTTPException(status_code=500, detail="Config path not set")
+
+    try:
+        body = await request.json()
+    except Exception as e:
+        raise HTTPException(status_code=422, detail=f"Invalid JSON: {e}")
+
+    # Validate sink config
+    try:
+        sink = _dict_to_dataclass(SinkConfig, body)
+        errors = sink.validate()
+        if errors:
+            raise HTTPException(status_code=422, detail=f"Invalid sink config: {'; '.join(errors)}")
+    except HTTPException:
+        raise
+    except Exception as e:
+        raise HTTPException(status_code=422, detail=f"Invalid sink config: {e}")
+
+    # Get current notifications config
+    config = request.app.state.config
+    notifications_dict = _dataclass_to_dict(config.notifications)
+
+    # Update sinks
+    if "sinks" not in notifications_dict:
+        notifications_dict["sinks"] = {}
+    notifications_dict["sinks"][name] = _dataclass_to_dict(sink)
+
+    # Persist through the standard config write path
+    config_dir = get_config_dir_from_path(config_path)
+    save_section("notifications", notifications_dict, config_dir)
+
+    # Update live config
+    if not hasattr(config.notifications, "sinks") or config.notifications.sinks is None:
+        config.notifications.sinks = {}
+    config.notifications.sinks[name] = sink
+
+    logger.info("Sink '%s' upserted: type=%s", name, sink.type)
+    return {"saved": True, "sink": name}
+
+
+@router.delete("/sinks/{name}")
+async def delete_sink(name: str, request: Request):
+    """Delete a sink.
+
+    Returns 409 Conflict if the sink is referenced by any toggle's severity_channels.
+    """
+    config_path = request.app.state.config_path
+    if not config_path:
+        raise HTTPException(status_code=500, detail="Config path not set")
+
+    config = request.app.state.config
+    sinks = getattr(config.notifications, "sinks", {}) or {}
+
+    if name not in sinks:
+        raise HTTPException(status_code=404, detail=f"Sink '{name}' not found")
+
+    # Check if sink is referenced by any toggle
+    toggles = getattr(config.notifications, "toggles", {}) or {}
+    references = []
+    for toggle_name, toggle in toggles.items():
+        if not hasattr(toggle, "severity_channels"):
+            continue
+        sev_channels = getattr(toggle, "severity_channels", {}) or {}
+        for severity, sink_names in sev_channels.items():
+            if name in (sink_names or []):
+                references.append(f"{toggle_name}.severity_channels.{severity}")
+
+    if references:
+        raise HTTPException(
+            status_code=409,
+            detail=f"Sink '{name}' is referenced by: {', '.join(references)}. "
+                   f"Remove references before deleting."
+        )
+
+    # Get current notifications config
+    notifications_dict = _dataclass_to_dict(config.notifications)
+
+    # Remove sink
+    if "sinks" in notifications_dict and name in notifications_dict["sinks"]:
+        del notifications_dict["sinks"][name]
+
+    # Persist
+    config_dir = get_config_dir_from_path(config_path)
+    save_section("notifications", notifications_dict, config_dir)
+
+    # Update live config
+    if hasattr(config.notifications, "sinks") and config.notifications.sinks:
+        del config.notifications.sinks[name]
+
+    logger.info("Sink '%s' deleted", name)
+    return {"deleted": True, "sink": name}
+
+
+@router.post("/sinks/{name}/test")
+async def test_sink(name: str, request: Request):
+    """Test a sink's connectivity.
+
+    Uses the channel's test_connection() method.
+    """
+    from meshai.notifications.channels import create_channel_from_sink
+
+    config = request.app.state.config
+    sinks = getattr(config.notifications, "sinks", {}) or {}
+    sink = sinks.get(name)
+
+    if sink is None:
+        raise HTTPException(status_code=404, detail=f"Sink '{name}' not found")
+
+    # Get connector from app state (may be None for non-mesh sinks)
+    connector = getattr(request.app.state, "connector", None)
+
+    try:
+        channel = create_channel_from_sink(sink, connector)
+        result = await channel.test_connection()
+        return result
+    except Exception as e:
+        logger.exception("Sink test failed for '%s'", name)
+        return {
+            "success": False,
+            "message": f"Test failed: {e}",
+            "error": str(e),
+            "details": {"sink": name, "type": sink.type}
+        }

meshai/dashboard/api/notification_routes.py

@@ -33,6 +33,11 @@ class ChannelTestRequest(BaseModel):
     headers: Optional[Dict[str, str]] = {}
 
 
+class SinkTestRequest(BaseModel):
+    """Request body for sink connectivity test."""
+    name: str  # Sink name from config
+
+
 class RuleSourcesRequest(BaseModel):
     """Request body for rule sources health check."""
     categories: List[str] = []
@@ -303,3 +308,94 @@ async def send_rule_live(request: Request, rule_index: int):
     )
 
     return result
+
+
+# =============================================================================
+# SINKS ENDPOINTS (routing simplification)
+# =============================================================================
+
+@router.get("/sinks")
+async def get_sinks(request: Request):
+    """Get configured notification sinks.
+
+    Returns list of named sinks with their type and config.
+    Read-only list for now; edit endpoints added later.
+    """
+    config = getattr(request.app.state, "config", None)
+    if not config or not hasattr(config, "notifications"):
+        return []
+
+    sinks = getattr(config.notifications, "sinks", {})
+    if not sinks:
+        return []
+
+    result = []
+    for name, sink in sinks.items():
+        # Convert dataclass to dict if needed
+        if hasattr(sink, "__dataclass_fields__"):
+            sink_dict = {
+                "name": name,
+                "type": sink.type,
+                "channel": getattr(sink, "channel", 0),
+                "node_ids": getattr(sink, "node_ids", []),
+                "smtp_host": getattr(sink, "smtp_host", ""),
+                "smtp_port": getattr(sink, "smtp_port", 587),
+                "from_address": getattr(sink, "from_address", ""),
+                "recipients": getattr(sink, "recipients", []),
+                "webhook_url": getattr(sink, "webhook_url", ""),
+            }
+        else:
+            sink_dict = {"name": name, **sink}
+
+        result.append(sink_dict)
+
+    return result
+
+
+@router.post("/sinks/test")
+async def test_sink(request: Request, body: SinkTestRequest):
+    """Test a named sink's connectivity.
+
+    Uses the existing channel test_connection() method.
+    """
+    config = getattr(request.app.state, "config", None)
+    if not config or not hasattr(config, "notifications"):
+        raise HTTPException(status_code=404, detail="Notifications not configured")
+
+    sinks = getattr(config.notifications, "sinks", {})
+    if not sinks:
+        raise HTTPException(status_code=404, detail="No sinks configured")
+
+    if body.name not in sinks:
+        raise HTTPException(status_code=404, detail=f"Sink not found: {body.name}")
+
+    sink = sinks[body.name]
+
+    # Get connector for mesh channels
+    connector = getattr(request.app.state, "connector", None)
+
+    try:
+        from ...notifications.channels import create_channel_from_sink
+        channel = create_channel_from_sink(sink, connector=connector)
+        result = await channel.test_connection()
+        return {
+            "sink": body.name,
+            "type": sink.type,
+            **result
+        }
+    except ValueError as e:
+        return {
+            "sink": body.name,
+            "type": getattr(sink, "type", "unknown"),
+            "success": False,
+            "message": str(e),
+            "error": str(e),
+        }
+    except Exception as e:
+        return {
+            "sink": body.name,
+            "type": getattr(sink, "type", "unknown"),
+            "success": False,
+            "message": f"Test failed: {str(e)}",
+            "error": str(e),
+        }

meshai/notifications/channels.py

@@ -14,7 +14,7 @@ import httpx
 
 if TYPE_CHECKING:
     from ..connector import MeshConnector
-    from ..config import NotificationRuleConfig
+    from ..config import NotificationRuleConfig, SinkConfig
     from .events import NotificationPayload
 
 from meshai.notifications.renderers import MeshRenderer, EmailRenderer, WebhookRenderer
@@ -837,3 +837,51 @@ def create_channel_from_dict(config: dict, connector=None) -> NotificationChanne
         )
     else:
         raise ValueError("Unknown channel type: %s" % channel_type)
+
+def create_channel_from_sink(sink: "SinkConfig", connector=None) -> NotificationChannel:
+    """Create a channel instance from a SinkConfig dataclass.
+
+    Routing simplification: sinks are named transports defined once,
+    this factory creates the channel instance for delivery.
+
+    Args:
+        sink: SinkConfig dataclass instance
+        connector: MeshConnector instance (required for mesh channels)
+
+    Returns:
+        NotificationChannel instance
+
+    Raises:
+        ValueError: If sink type is unknown or channel < 0
+    """
+    sink_type = sink.type
+
+    if sink_type == "mesh_broadcast":
+        if sink.channel < 0:
+            raise ValueError(f"Channel must be >= 0, got {sink.channel}")
+        return MeshBroadcastChannel(
+            connector=connector,
+            channel_index=sink.channel,
+        )
+    elif sink_type == "mesh_dm":
+        return MeshDMChannel(
+            connector=connector,
+            node_ids=sink.node_ids,
+        )
+    elif sink_type == "email":
+        return EmailChannel(
+            smtp_host=sink.smtp_host,
+            smtp_port=sink.smtp_port,
+            smtp_user=sink.smtp_user,
+            smtp_password=sink.smtp_password,
+            smtp_tls=sink.smtp_tls,
+            from_address=sink.from_address,
+            recipients=sink.recipients,
+        )
+    elif sink_type == "webhook":
+        return WebhookChannel(
+            url=sink.webhook_url,
+            headers=sink.webhook_headers,
+        )
+    else:
+        raise ValueError(f"Unknown sink type: {sink_type}")

meshai/notifications/pipeline/dispatcher.py

@@ -87,6 +87,9 @@ class Dispatcher:
         self._toggle_cooldown: dict[tuple[str, str, str], float] = {}
         # Insertion-ordered (source, event.id) -> sentinel; evict oldest at cap.
         self._dedup_lru: "OrderedDict[tuple[str, str], bool]" = OrderedDict()
+        # v0.7: track toggles that have logged legacy fallback deprecation warning
+        # (once per toggle per boot, not per event)
+        self._legacy_warned: set[str] = set()
         # v0.6-2: hydrate from SQLite. Graceful no-op if persistence is
         # unavailable -- the dispatcher still works, just without
         # cross-restart durability.
@@ -281,22 +284,25 @@ class Dispatcher:
     async def _dispatch_toggles(self, event: Event) -> None:
         """Route an event through its family master-toggle (parallel to rules).
 
-        v0.5.2 guards (run in order, at the entrance):
-          1. Staleness   — drop events older than `toggle.freshness_seconds`.
-                           Solves the restart-wave problem definitively: a
-                           backlog of stale events from durable storage gets
-                           dropped here, never broadcast.
-          2. Cooldown    — per (toggle.name, category, region) throttle keyed
-                           on `toggle.cooldown_seconds`. Silent, no log spam.
-          3. Dedup       — bounded LRU on (source, event.id); catches Central
-                           re-delivery during reconnect.
-        Then composes a friendly mesh string instead of the prior raw
-        `[Family] central.category` debug format.
+        v0.7 guard order (B13 fix — guards run BEFORE arming cooldown/dedup):
+          0. Cold-start grace — suppress broadcasts in first N seconds
+          1. Staleness — drop events older than toggle.freshness_seconds
+          2. Region filter — drop if event region not in toggle.regions
+          3. Matrix resolution — resolve severity_channels[severity] to sinks
+             → IF empty, return WITHOUT arming cooldown or recording dedup
+          4. Cooldown — per (toggle, category, region) throttle
+          5. Dedup — bounded LRU on (source, event.id)
+          6. Deliver per resolved sink
+
+        v0.7 sink-name routing: severity_channels values are sink names
+        (not channel types). Sinks resolved against config.notifications.sinks.
 
         v0.6-2: every mutation of the four drop counters, the cold-start
         anchor, the cooldown map, and the dedup LRU writes through to
         SQLite via the _persist_* helpers. Read fast-path stays in-memory.
         """
+        from meshai.notifications.channels import create_channel_from_sink
+
         toggles = getattr(self._config.notifications, "toggles", None)
         if not isinstance(toggles, dict) or not toggles:
             return
@@ -357,7 +363,50 @@ class Dispatcher:
                 )
                 return
 
-        # ---------- Section 2 — per-toggle cooldown ----------
+        # ---------- Section 2 — region filter (B13: moved before cooldown/dedup) ----------
+        regions = getattr(tog, "regions", None) or []
+        if regions:
+            ev_regions = set(filter(None, [event.region, *(event.regions or [])]))
+            if not (set(regions) & ev_regions):
+                return
+
+        # ---------- Section 3 — matrix resolution (B13: moved before cooldown/dedup) ----------
+        # severity_channels values can be:
+        #   - sink names (v0.7+): resolve against config.notifications.sinks
+        #   - channel types (v0.5/v0.6 legacy): mesh_broadcast, mesh_dm, email, webhook
+        # Backwards compatibility: if no sinks config exists OR a channel type name is used,
+        # fall back to _toggle_to_rule for inline transport config.
+        sinks_config = getattr(self._config.notifications, "sinks", None) or {}
+        sev_channels = getattr(tog, "severity_channels", None) or {}
+        sink_names = sev_channels.get(event.severity, [])
+
+        # Legacy channel types that trigger _toggle_to_rule fallback
+        LEGACY_CHANNEL_TYPES = {"mesh_broadcast", "mesh_dm", "email", "webhook", "digest"}
+
+        resolved_sinks: list[tuple[str, object, bool]] = []  # (name, config, is_legacy)
+        for sink_name in sink_names:
+            # Skip digest pseudo-channel (no-op)
+            if sink_name == "digest":
+                continue
+
+            sink = sinks_config.get(sink_name)
+            if sink is not None:
+                # v0.7+ sink-name routing
+                resolved_sinks.append((sink_name, sink, False))
+            elif sink_name in LEGACY_CHANNEL_TYPES:
+                # v0.5/v0.6 backwards compatibility: channel type name
+                # Use _toggle_to_rule to build a rule from inline transport config
+                resolved_sinks.append((sink_name, sink_name, True))  # sink_name IS the channel type
+            else:
+                self._logger.warning(
+                    f"dispatcher: unknown sink '{sink_name}' in toggle {fam}; skipping"
+                )
+
+        if not resolved_sinks:
+            # No sinks to deliver to — return WITHOUT arming cooldown or dedup (B13 fix)
+            return
+
+        # ---------- Section 4 — per-toggle cooldown (B13: moved after matrix) ----------
         # Immediate-severity events bypass cooldown entirely — they are
         # already rate-controlled by source handler change detection.
         if getattr(event, "severity", None) == "immediate":
@@ -394,7 +443,7 @@ class Dispatcher:
                     k: t for k, t in self._toggle_cooldown.items() if t >= cutoff
                 }
 
-        # ---------- Section 3 — (source, event.id) dedup ----------
+        # ---------- Section 5 — (source, event.id) dedup (B13: moved after matrix) ----------
         dk = (event.source or "", event.id or "")
         if dk in self._dedup_lru:
             # Touch to keep recent.
@@ -412,16 +461,7 @@ class Dispatcher:
         while len(self._dedup_lru) > _lru_max:
             self._dedup_lru.popitem(last=False)  # evict oldest
 
-        regions = getattr(tog, "regions", None) or []
-        if regions:
-            ev_regions = set(filter(None, [event.region, *(event.regions or [])]))
-            if not (set(regions) & ev_regions):
-                return
-        event_rank = self.SEVERITY_RANK.get(event.severity, 0)
-        if event_rank < self.SEVERITY_RANK.get(getattr(tog, "min_severity", "routine"), 0):
-            return
-
-        # ---------- Section 4 — friendly composer wired in ----------
+        # ---------- Section 6 — compose + deliver per sink ----------
         # Render once per event; reused across every channel below. Wrapped
         # so a renderer fault never blocks delivery — we fall back to the
         # legacy make_payload_from_event message (event.summary|title|category).
@@ -431,29 +471,54 @@ class Dispatcher:
             self._logger.exception("mesh composer crashed; falling back to legacy message")
             friendly = None
 
-        sev_channels = getattr(tog, "severity_channels", None) or {}
-        for ch_type in sev_channels.get(event.severity, []):
-            if ch_type == "digest":
-                continue
+        for sink_name, sink_or_ch_type, is_legacy in resolved_sinks:
             try:
-                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"):
-                    payload = make_payload_from_event(event, message=friendly)
+                if is_legacy:
+                    # v0.5/v0.6 backwards compatibility: use _toggle_to_rule
+                    # DEPRECATED: This fallback will be removed in session 3.
+                    # Run migration to convert severity_channels to sink names.
+                    ch_type = sink_or_ch_type  # sink_or_ch_type IS the channel type string
+                    if fam not in self._legacy_warned:
+                        self._legacy_warned.add(fam)
+                        self._logger.warning(
+                            "DEPRECATED: toggle '%s' uses channel-type routing ('%s'). "
+                            "Run migrate_config_routing.py to convert to sink-name routing. "
+                            "This fallback will be removed in a future release.",
+                            fam, ch_type,
+                        )
+                    rule = self._toggle_to_rule(tog, ch_type, event)
+                    channel = self._channel_factory(rule, self._connector)
+                    sink_type = ch_type
+                    if friendly is not None and ch_type in ("mesh_broadcast", "mesh_dm"):
+                        payload = make_payload_from_event(event, message=friendly)
+                    else:
+                        payload = make_payload_from_event(event)
+                    success = await channel.deliver(payload, rule)
+                    if success:
+                        self._logger.info(f"Dispatched event {event.id} via toggle {fam}/{ch_type}")
+                        self._post_broadcast_commit(event, payload, rule, ch_type)
+                    else:
+                        self._logger.warning(f"Toggle channel delivery returned False for {fam}/{ch_type}")
                 else:
-                    payload = make_payload_from_event(event)
-                success = await channel.deliver(payload, rule)
-                if success:
-                    self._logger.info(f"Dispatched event {event.id} via toggle {fam}/{ch_type}")
-                    # v0.5.8b post-broadcast commit. Persistence-side
-                    # bookkeeping that should only happen when a delivery
-                    # actually went out: mesh_broadcasts_out audit row +
-                    # handler-supplied last_broadcast_* UPDATE callback.
-                    self._post_broadcast_commit(event, payload, rule, ch_type)
-                else:
-                    self._logger.warning(f"Toggle channel delivery returned False for {fam}/{ch_type}")
+                    # v0.7+ sink-name routing
+                    sink = sink_or_ch_type
+                    sink_type = getattr(sink, "type", "mesh_broadcast")
+                    channel = create_channel_from_sink(sink, self._connector)
+                    if friendly is not None and sink_type in ("mesh_broadcast", "mesh_dm"):
+                        payload = make_payload_from_event(event, message=friendly)
+                    else:
+                        payload = make_payload_from_event(event)
+                    # channel.deliver() signature requires rule but doesn't use it;
+                    # pass None for compatibility (channels use their own config)
+                    success = await channel.deliver(payload, None)
+                    if success:
+                        self._logger.info(f"Dispatched event {event.id} via toggle {fam}/{sink_name}")
+                        # v0.7: post-broadcast commit with sink info (not rule)
+                        self._post_broadcast_commit_sink(event, payload, sink, sink_name)
+                    else:
+                        self._logger.warning(f"Toggle channel delivery returned False for {fam}/{sink_name}")
             except Exception:
-                self._logger.exception(f"Toggle channel delivery failed for {fam}/{ch_type}")
+                self._logger.exception(f"Toggle channel delivery failed for {fam}/{sink_name}")
 
     def dispatch_stats(self) -> dict:
         """Expose v0.5.2 toggle-path guard counters for ops/health endpoints.
@@ -614,6 +679,58 @@ class Dispatcher:
                     "post-broadcast: handler commit-callback raised"
                 )
 
+    def _post_broadcast_commit_sink(self, event, payload, sink, sink_name: str) -> None:
+        """v0.7 sink-based audit commit (replaces _post_broadcast_commit for sink routing).
+
+        Same logic as _post_broadcast_commit but extracts channel/node_ids from
+        SinkConfig instead of NotificationRuleConfig.
+        """
+        data = getattr(event, "data", None) or {}
+        if not data:
+            return
+        committed_at = time.time()
+
+        audit = data.get("_broadcast_audit")
+        if isinstance(audit, dict):
+            try:
+                from meshai.persistence import get_db
+                conn = get_db()
+                text = payload.message if payload is not None else (event.title or "")
+                bytes_sent = len(text.encode("utf-8")) if text else 0
+                sink_type = getattr(sink, "type", "mesh_broadcast")
+                if sink_type == "mesh_dm":
+                    node_ids = list(getattr(sink, "node_ids", []) or [])
+                    recipient = ",".join(map(str, node_ids)) or "dm"
+                else:
+                    recipient = "broadcast"
+                channel = getattr(sink, "channel", None)
+                conn.execute(
+                    "INSERT INTO mesh_broadcasts_out(sent_at, recipient, channel, "
+                    "text, source_event_table, source_event_pk, bytes_sent, "
+                    "ack_received) VALUES (?,?,?,?,?,?,?,?)",
+                    (
+                        int(committed_at), recipient, channel, text,
+                        audit.get("table"), audit.get("pk"),
+                        bytes_sent, 0,
+                    ),
+                )
+            except Exception:
+                self._logger.exception(
+                    "post-broadcast: mesh_broadcasts_out insert failed "
+                    "(sink=%s table=%s pk=%s)",
+                    sink_name, audit.get("table"), audit.get("pk"),
+                )
+
+        cb = data.get("_on_broadcast_committed")
+        if callable(cb):
+            try:
+                cb(committed_at)
+            except Exception:
+                self._logger.exception(
+                    "post-broadcast: handler commit-callback raised (sink=%s)",
+                    sink_name,
+                )
+
     def _toggle_to_rule(self, tog, ch_type: str, event: Event):
         from meshai.config import NotificationRuleConfig
         return NotificationRuleConfig(

meshai/scripts/migrate_config_routing.py

@@ -0,0 +1,761 @@
+#!/usr/bin/env python3
+"""Migration script for MeshAI routing simplification: synthesize sinks + matrix rewrite.
+
+This script reads existing notification toggles and rules, extracts their
+inline transport configurations, and synthesizes named sinks.
+
+Run manually: python -m meshai.scripts.migrate_config_routing [--dry-run]
+
+The migration:
+1. Backs up the config to <path>.pre-sinks.<epoch>.bak
+2. For each toggle with inline transport config, synthesizes a named sink
+3. For each enabled rule with inline transport config, synthesizes a named sink
+4. Deduplicates identical transports into one sink
+5. Writes the sinks block to the config
+6. Rewrites severity_channels: channel types → sink names (Phase B)
+7. Blanks matrix rows below old min_severity threshold (Phase B)
+8. Removes min_severity field from toggles (Phase B)
+9. Does NOT remove inline transport fields (done in a later step)
+
+Idempotent: refuses to run if a sinks block already exists.
+"""
+
+import argparse
+import hashlib
+import json
+import logging
+import os
+import sys
+import time
+from pathlib import Path
+from typing import Any, Optional
+from urllib.parse import urlparse
+
+import yaml
+
+# Setup logging
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(message)s",
+    datefmt="%H:%M:%S",
+)
+logger = logging.getLogger(__name__)
+
+
+def compute_sink_hash(sink_dict: dict) -> str:
+    """Compute a stable hash of sink config for deduplication."""
+    # Sort keys for stable comparison
+    canonical = json.dumps(sink_dict, sort_keys=True)
+    return hashlib.sha256(canonical.encode()).hexdigest()[:8]
+
+
+def generate_sink_name(sink_type: str, sink_dict: dict) -> str:
+    """Generate a human-readable sink name from its config."""
+    if sink_type == "mesh_broadcast":
+        channel = sink_dict.get("channel", 0)
+        return f"mesh-ch{channel}"
+    elif sink_type == "mesh_dm":
+        node_ids = sink_dict.get("node_ids", [])
+        if node_ids:
+            first_node = str(node_ids[0]).lstrip("!")[:8]
+            return f"dm-{first_node}"
+        return "dm-unknown"
+    elif sink_type == "email":
+        host = sink_dict.get("smtp_host", "")
+        if host:
+            # Extract domain
+            return f"email-{host.split('.')[0]}"
+        return "email-unknown"
+    elif sink_type == "webhook":
+        url = sink_dict.get("webhook_url", "")
+        if url:
+            parsed = urlparse(url)
+            return f"webhook-{parsed.netloc.split('.')[0]}"
+        return "webhook-unknown"
+    return f"sink-{sink_type}"
+
+
+def extract_sinks_from_toggle(toggle: dict) -> list[dict]:
+    """Extract ALL sink configs from a NotificationToggle's inline fields.
+    
+    Returns a list of sink dicts, one per configured transport type.
+    No precedence — every transport with non-empty config is extracted.
+    """
+    sinks = []
+
+    # Check for mesh_broadcast (broadcast_channel field)
+    broadcast_channel = toggle.get("broadcast_channel")
+    if broadcast_channel is not None:
+        sinks.append({
+            "type": "mesh_broadcast",
+            "channel": int(broadcast_channel),
+        })
+
+    # Check for mesh_dm (node_ids field)
+    node_ids = toggle.get("node_ids", [])
+    if node_ids:
+        sinks.append({
+            "type": "mesh_dm",
+            "node_ids": node_ids,
+        })
+
+    # Check for email (smtp_host field)
+    smtp_host = toggle.get("smtp_host", "")
+    if smtp_host:
+        sinks.append({
+            "type": "email",
+            "smtp_host": smtp_host,
+            "smtp_port": toggle.get("smtp_port", 587),
+            "smtp_user": toggle.get("smtp_user", ""),
+            "smtp_password": toggle.get("smtp_password", ""),
+            "smtp_tls": toggle.get("smtp_tls", True),
+            "from_address": toggle.get("from_address", ""),
+            "recipients": toggle.get("recipients", []),
+        })
+
+    # Check for webhook (webhook_url field)
+    webhook_url = toggle.get("webhook_url", "")
+    if webhook_url:
+        sinks.append({
+            "type": "webhook",
+            "webhook_url": webhook_url,
+            "webhook_headers": toggle.get("webhook_headers", {}),
+        })
+
+    return sinks
+
+
+def extract_sink_from_toggle(toggle: dict) -> Optional[dict]:
+    """Legacy wrapper - returns first sink or None. Use extract_sinks_from_toggle instead."""
+    sinks = extract_sinks_from_toggle(toggle)
+    return sinks[0] if sinks else None
+
+
+def extract_sink_from_rule(rule: dict) -> Optional[dict]:
+    """Extract sink config from a NotificationRuleConfig's inline fields."""
+    delivery_type = rule.get("delivery_type", "")
+
+    if delivery_type == "mesh_broadcast":
+        return {
+            "type": "mesh_broadcast",
+            "channel": rule.get("broadcast_channel", 0),
+        }
+    elif delivery_type == "mesh_dm":
+        return {
+            "type": "mesh_dm",
+            "node_ids": rule.get("node_ids", []),
+        }
+    elif delivery_type == "email":
+        return {
+            "type": "email",
+            "smtp_host": rule.get("smtp_host", ""),
+            "smtp_port": rule.get("smtp_port", 587),
+            "smtp_user": rule.get("smtp_user", ""),
+            "smtp_password": rule.get("smtp_password", ""),
+            "smtp_tls": rule.get("smtp_tls", True),
+            "from_address": rule.get("from_address", ""),
+            "recipients": rule.get("recipients", []),
+        }
+    elif delivery_type == "webhook":
+        return {
+            "type": "webhook",
+            "webhook_url": rule.get("webhook_url", ""),
+            "webhook_headers": rule.get("webhook_headers", {}),
+        }
+
+    return None
+
+
+def synthesize_sinks(notifications: dict) -> tuple[dict, dict]:
+    """Synthesize named sinks from toggles and rules.
+
+    Returns:
+        (sinks: dict mapping sink names to sink configs,
+         hash_to_name: dict mapping sink hashes to names for matrix migration)
+    """
+    sinks = {}
+    hash_to_name = {}  # For deduplication + matrix migration lookup
+
+    # Process toggles - extract ALL configured transports per toggle
+    toggles = notifications.get("toggles", {})
+    for toggle_name, toggle in toggles.items():
+        if not isinstance(toggle, dict):
+            continue
+
+        sink_dicts = extract_sinks_from_toggle(toggle)
+        if not sink_dicts:
+            continue
+
+        for sink_dict in sink_dicts:
+            sink_hash = compute_sink_hash(sink_dict)
+            if sink_hash in hash_to_name:
+                logger.info(f"  Toggle '{toggle_name}' reuses existing sink '{hash_to_name[sink_hash]}'")
+                continue
+
+            sink_name = generate_sink_name(sink_dict["type"], sink_dict)
+            # Handle name collisions
+            base_name = sink_name
+            counter = 2
+            while sink_name in sinks:
+                sink_name = f"{base_name}-{counter}"
+                counter += 1
+
+            sinks[sink_name] = sink_dict
+            hash_to_name[sink_hash] = sink_name
+            logger.info(f"  Toggle '{toggle_name}' → sink '{sink_name}'")
+
+    # Process rules
+    rules = notifications.get("rules", [])
+    for i, rule in enumerate(rules):
+        if not isinstance(rule, dict):
+            continue
+
+        # Only process enabled rules
+        if not rule.get("enabled", True):
+            continue
+
+        sink_dict = extract_sink_from_rule(rule)
+        if not sink_dict:
+            continue
+
+        sink_hash = compute_sink_hash(sink_dict)
+        if sink_hash in hash_to_name:
+            rule_name = rule.get("name", f"rule-{i}")
+            logger.info(f"  Rule '{rule_name}' reuses existing sink '{hash_to_name[sink_hash]}'")
+            continue
+
+        sink_name = generate_sink_name(sink_dict["type"], sink_dict)
+        # Handle name collisions
+        base_name = sink_name
+        counter = 2
+        while sink_name in sinks:
+            sink_name = f"{base_name}-{counter}"
+            counter += 1
+
+        sinks[sink_name] = sink_dict
+        hash_to_name[sink_hash] = sink_name
+        rule_name = rule.get("name", f"rule-{i}")
+        logger.info(f"  Rule '{rule_name}' → sink '{sink_name}'")
+
+    return sinks, hash_to_name
+
+
+# ---------- Phase B: Matrix rewrite ----------
+
+SEVERITY_RANK = {"routine": 0, "priority": 1, "immediate": 2}
+
+
+def build_channel_type_to_sink_map(toggle: dict, hash_to_name: dict) -> dict[str, str]:
+    """Build a mapping from channel type to sink name for a toggle.
+
+    Uses the same hash-based lookup as synthesize_sinks to find which sink
+    was created from each transport type in this toggle.
+
+    Returns:
+        {"mesh_broadcast": "mesh-ch0", "mesh_dm": "dm-abc123", ...}
+    """
+    mapping = {}
+    sink_dicts = extract_sinks_from_toggle(toggle)
+    for sink_dict in sink_dicts:
+        sink_type = sink_dict["type"]
+        sink_hash = compute_sink_hash(sink_dict)
+        if sink_hash in hash_to_name:
+            mapping[sink_type] = hash_to_name[sink_hash]
+    return mapping
+
+
+def migrate_toggle_matrix(
+    toggle_name: str,
+    toggle: dict,
+    channel_to_sink: dict[str, str],
+) -> tuple[dict, list[str]]:
+    """Rewrite a toggle's severity_channels from channel types to sink names.
+
+    Also blanks matrix rows below old min_severity (they never fired anyway).
+
+    Args:
+        toggle_name: Name of this toggle (for logging)
+        toggle: The toggle dict
+        channel_to_sink: Mapping from channel type to sink name
+
+    Returns:
+        (new_severity_channels, list_of_changes)
+    """
+    changes = []
+    old_matrix = toggle.get("severity_channels", {})
+    old_min_severity = toggle.get("min_severity", "routine")
+    min_rank = SEVERITY_RANK.get(old_min_severity, 0)
+
+    new_matrix = {}
+    for severity in ["routine", "priority", "immediate"]:
+        sev_rank = SEVERITY_RANK.get(severity, 0)
+        old_channels = old_matrix.get(severity, [])
+
+        # If this severity was below min_severity, blank the row
+        if sev_rank < min_rank:
+            if old_channels:
+                changes.append(f"  {severity}: blanked (was below min_severity={old_min_severity})")
+            new_matrix[severity] = []
+            continue
+
+        # Convert channel types to sink names
+        new_sinks = []
+        for ch_type in old_channels:
+            # Skip digest pseudo-channel (it's a no-op)
+            if ch_type == "digest":
+                changes.append(f"  {severity}: removed 'digest' (no-op pseudo-channel)")
+                continue
+            sink_name = channel_to_sink.get(ch_type)
+            if sink_name:
+                new_sinks.append(sink_name)
+            else:
+                # Channel type has no corresponding sink (shouldn't happen if
+                # synthesize_sinks ran first, but be defensive)
+                changes.append(f"  {severity}: WARNING: no sink for channel type '{ch_type}'")
+        new_matrix[severity] = new_sinks
+
+        # Log the conversion
+        if old_channels != new_sinks:
+            changes.append(f"  {severity}: {old_channels} → {new_sinks}")
+
+    return new_matrix, changes
+
+
+def migrate_all_matrices(
+    notifications: dict,
+    hash_to_name: dict,
+) -> tuple[dict[str, dict], dict[str, list[str]]]:
+    """Rewrite all toggle severity_channels matrices.
+
+    Returns:
+        (toggle_name -> new_severity_channels, toggle_name -> list_of_changes)
+    """
+    new_matrices = {}
+    all_changes = {}
+
+    toggles = notifications.get("toggles", {})
+    for toggle_name, toggle in toggles.items():
+        if not isinstance(toggle, dict):
+            continue
+
+        channel_to_sink = build_channel_type_to_sink_map(toggle, hash_to_name)
+        if not channel_to_sink:
+            # Toggle has no inline transport config, skip matrix rewrite
+            continue
+
+        new_matrix, changes = migrate_toggle_matrix(toggle_name, toggle, channel_to_sink)
+        if changes:
+            new_matrices[toggle_name] = new_matrix
+            all_changes[toggle_name] = changes
+
+    return new_matrices, all_changes
+
+
+def render_matrix_updates_yaml(
+    toggle_updates: dict[str, dict],
+    min_severity_removals: list[str],
+    layout: str,
+) -> str:
+    """Render YAML snippet showing matrix updates.
+
+    For dry-run display only - actual write uses yaml.safe_load/dump round-trip.
+    """
+    lines = ["# Matrix updates:"]
+    for toggle_name, new_matrix in toggle_updates.items():
+        lines.append(f"toggles.{toggle_name}.severity_channels:")
+        for sev, sinks in new_matrix.items():
+            lines.append(f"    {sev}: {sinks}")
+    if min_severity_removals:
+        lines.append("# min_severity removed from:")
+        for name in min_severity_removals:
+            lines.append(f"  - {name}")
+    return "\n".join(lines)
+
+
+def apply_matrix_updates(
+    config: dict,
+    layout: str,
+    toggle_updates: dict[str, dict],
+) -> None:
+    """Apply matrix updates to config dict in-place.
+
+    Also removes min_severity field from updated toggles.
+    """
+    notifications = get_notifications_block(config, layout)
+    toggles = notifications.get("toggles", {})
+
+    for toggle_name, new_matrix in toggle_updates.items():
+        toggle = toggles.get(toggle_name)
+        if not isinstance(toggle, dict):
+            continue
+        toggle["severity_channels"] = new_matrix
+        # Remove min_severity - matrix is now the only gate
+        if "min_severity" in toggle:
+            del toggle["min_severity"]
+
+
+def detect_config_layout(config: dict) -> str:
+    """Detect whether config is monolithic or multi-file layout.
+
+    Returns:
+        "monolithic" if config has a notifications: wrapper key
+        "multifile" if file root IS the notifications block (toggles/rules at root)
+        "empty" if neither pattern matches
+    """
+    if "notifications" in config:
+        return "monolithic"
+    elif "toggles" in config or "rules" in config:
+        return "multifile"
+    else:
+        return "empty"
+
+
+def get_notifications_block(config: dict, layout: str) -> dict:
+    """Extract notifications block based on detected layout."""
+    if layout == "monolithic":
+        return config.get("notifications", {})
+    elif layout == "multifile":
+        return config
+    else:
+        return {}
+
+
+def load_notifications_config(config_path: Path) -> tuple[dict, dict, str]:
+    """Load notifications config from file.
+
+    Returns:
+        (full_config_dict, notifications_dict, layout_type)
+    """
+    with open(config_path, "r") as f:
+        config = yaml.safe_load(f) or {}
+
+    layout = detect_config_layout(config)
+    notifications = get_notifications_block(config, layout)
+    return config, notifications, layout
+
+
+def backup_config(config_path: Path) -> Path:
+    """Create a timestamped backup of the config file."""
+    epoch = int(time.time())
+    backup_path = config_path.with_suffix(f".pre-sinks.{epoch}.bak")
+    import shutil
+    shutil.copy2(config_path, backup_path)
+    return backup_path
+
+
+def render_sinks_yaml(sinks: dict, layout: str) -> str:
+    """Render sinks block as YAML text for appending/inserting.
+
+    For multifile layout: sinks at root level.
+    For monolithic layout: sinks indented under notifications (2-space indent).
+    """
+    sinks_yaml = yaml.dump({"sinks": sinks}, default_flow_style=False, sort_keys=False, allow_unicode=True)
+
+    if layout == "monolithic":
+        # Indent everything by 2 spaces to nest under notifications:
+        lines = sinks_yaml.split("\n")
+        indented_lines = ["  " + line if line.strip() else line for line in lines]
+        return "\n".join(indented_lines)
+    else:
+        # Multi-file: sinks at root level
+        return sinks_yaml
+
+
+def find_notifications_block_end(content: str) -> int:
+    """Find the line index where the notifications block ends in monolithic config.
+
+    Returns the index after the last line of notifications block content.
+    Notifications block ends when we hit a non-indented line (another top-level key)
+    or end of file.
+    """
+    lines = content.split("\n")
+    in_notifications = False
+    last_notifications_line = -1
+
+    for i, line in enumerate(lines):
+        stripped = line.lstrip()
+        if not stripped or stripped.startswith("#"):
+            continue
+
+        # Check if this is the notifications: key
+        if line.startswith("notifications:") and not line[0].isspace():
+            in_notifications = True
+            last_notifications_line = i
+            continue
+
+        if in_notifications:
+            # Check if still inside notifications (indented)
+            if line[0].isspace() or not line.strip():
+                last_notifications_line = i
+            else:
+                # Hit another top-level key, notifications block ended
+                break
+
+    return last_notifications_line + 1 if last_notifications_line >= 0 else len(lines)
+
+
+def verify_sinks_written(config_path: Path, sinks: dict, layout: str, original_keys: set) -> tuple[bool, str]:
+    """Verify the written config is valid and sinks are accessible.
+
+    Returns:
+        (success, error_message)
+    """
+    try:
+        with open(config_path, "r") as f:
+            new_config = yaml.safe_load(f)
+    except yaml.YAMLError as e:
+        return False, f"YAML parse error: {e}"
+
+    if new_config is None:
+        return False, "Config parsed as empty"
+
+    # Check sinks are at the expected path
+    new_layout = detect_config_layout(new_config)
+    notifications = get_notifications_block(new_config, new_layout)
+
+    if "sinks" not in notifications:
+        return False, f"Sinks not found at expected path (layout: {new_layout})"
+
+    written_sinks = notifications["sinks"]
+    if set(written_sinks.keys()) != set(sinks.keys()):
+        return False, f"Sink names mismatch: expected {set(sinks.keys())}, got {set(written_sinks.keys())}"
+
+    # Verify original top-level keys are intact
+    new_top_keys = set(new_config.keys())
+    missing_keys = original_keys - new_top_keys - {"sinks"}  # sinks may be new at root
+    if missing_keys:
+        return False, f"Original keys lost: {missing_keys}"
+
+    return True, ""
+
+
+def write_sinks_to_config(config_path: Path, sinks: dict, layout: str, backup_path: Path) -> bool:
+    """Write synthesized sinks block to the config file.
+
+    Uses append/insert strategy to preserve comments and formatting:
+    - multifile: append sinks at end of file
+    - monolithic: insert sinks within notifications block
+
+    Verifies the result and restores from backup on failure.
+
+    Returns:
+        True on success, False on failure (backup restored)
+    """
+    import shutil
+
+    # Read original content
+    with open(config_path, "r") as f:
+        original_content = f.read()
+
+    # Parse to get original top-level keys for verification
+    original_config = yaml.safe_load(original_content) or {}
+    original_keys = set(original_config.keys())
+
+    # Render sinks block
+    sinks_yaml = render_sinks_yaml(sinks, layout)
+
+    # Build new content based on layout
+    if layout == "multifile":
+        # Simple append at end
+        if not original_content.endswith("\n"):
+            original_content += "\n"
+        new_content = original_content + "\n" + sinks_yaml
+    else:
+        # Monolithic: insert within notifications block
+        lines = original_content.split("\n")
+        insert_point = find_notifications_block_end(original_content)
+
+        # Insert the indented sinks block
+        sinks_lines = sinks_yaml.rstrip("\n").split("\n")
+        new_lines = lines[:insert_point] + sinks_lines + lines[insert_point:]
+        new_content = "\n".join(new_lines)
+
+    # Write new content
+    with open(config_path, "w") as f:
+        f.write(new_content)
+
+    # Verify the result
+    success, error = verify_sinks_written(config_path, sinks, layout, original_keys)
+
+    if not success:
+        logger.error(f"Post-write verification failed: {error}")
+        logger.info("Restoring from backup...")
+        shutil.copy2(backup_path, config_path)
+        return False
+
+    return True
+
+
+def write_full_config(config_path: Path, config: dict, layout: str, backup_path: Path) -> bool:
+    """Write full config using yaml round-trip (for matrix updates).
+
+    Unlike write_sinks_to_config which preserves comments, this does a full
+    dump. Used when we need to modify fields in-place (matrix updates).
+
+    Returns:
+        True on success, False on failure (backup restored)
+    """
+    import shutil
+
+    try:
+        with open(config_path, "w") as f:
+            yaml.dump(config, f, default_flow_style=False, sort_keys=False, allow_unicode=True)
+        return True
+    except Exception as e:
+        logger.error(f"Failed to write config: {e}")
+        logger.info("Restoring from backup...")
+        shutil.copy2(backup_path, config_path)
+        return False
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Migrate MeshAI config to use named sinks + matrix rewrite"
+    )
+    parser.add_argument(
+        "--config",
+        type=Path,
+        default=Path("/data/config/notifications.yaml"),
+        help="Path to notifications config file",
+    )
+    parser.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="Show what would be done without making changes",
+    )
+    parser.add_argument(
+        "--phase",
+        choices=["a", "b", "all"],
+        default="all",
+        help="Phase A: sinks only; Phase B: matrix rewrite; all: both (default)",
+    )
+    args = parser.parse_args()
+
+    config_path = args.config
+
+    if not config_path.exists():
+        logger.error(f"Config file not found: {config_path}")
+        sys.exit(1)
+
+    logger.info(f"Loading config from {config_path}")
+    full_config, notifications, layout = load_notifications_config(config_path)
+    logger.info(f"Detected config layout: {layout}")
+
+    # ---------- Phase A: Synthesize sinks ----------
+    sinks = {}
+    hash_to_name = {}
+
+    if args.phase in ("a", "all"):
+        # Check if sinks already exist
+        if notifications.get("sinks"):
+            if args.phase == "a":
+                logger.error("Sinks block already exists. Phase A already complete.")
+                sys.exit(1)
+            else:
+                # Phase "all" with existing sinks: skip Phase A, proceed to B
+                logger.info("Sinks already exist; skipping Phase A synthesis")
+                # Build hash_to_name from existing sinks for Phase B
+                existing_sinks = notifications.get("sinks", {})
+                for sink_name, sink_config in existing_sinks.items():
+                    if isinstance(sink_config, dict):
+                        sink_hash = compute_sink_hash(sink_config)
+                        hash_to_name[sink_hash] = sink_name
+                sinks = existing_sinks
+        else:
+            logger.info("Synthesizing sinks from toggles and rules...")
+            sinks, hash_to_name = synthesize_sinks(notifications)
+
+            if not sinks:
+                logger.info("No sinks to synthesize (no inline transport configs found)")
+                if args.phase == "a":
+                    sys.exit(0)
+            else:
+                logger.info(f"Synthesized {len(sinks)} sink(s):")
+                for name, sink in sinks.items():
+                    logger.info(f"  {name}: {sink}")
+
+    # ---------- Phase B: Matrix rewrite ----------
+    toggle_updates = {}
+    all_changes = {}
+
+    if args.phase in ("b", "all") and hash_to_name:
+        logger.info("Migrating severity_channels matrices...")
+        toggle_updates, all_changes = migrate_all_matrices(notifications, hash_to_name)
+
+        if toggle_updates:
+            logger.info(f"Matrix updates for {len(toggle_updates)} toggle(s):")
+            for toggle_name, changes in all_changes.items():
+                logger.info(f"  {toggle_name}:")
+                for change in changes:
+                    logger.info(f"    {change}")
+        else:
+            logger.info("No matrix updates needed")
+
+    # Determine where sinks will land
+    if layout == "multifile":
+        sinks_path = "root-level sinks:"
+    else:
+        sinks_path = "notifications.sinks"
+
+    # ---------- Dry-run output ----------
+    if args.dry_run:
+        logger.info("DRY RUN - no changes made")
+        print(f"\n--- Config layout: {layout} ---")
+
+        if sinks and args.phase in ("a", "all") and not notifications.get("sinks"):
+            print(f"\n--- Phase A: Sinks will be written to: {sinks_path} ---")
+            print(render_sinks_yaml(sinks, layout))
+
+        if toggle_updates and args.phase in ("b", "all"):
+            print("\n--- Phase B: Matrix updates ---")
+            min_severity_removals = list(toggle_updates.keys())
+            print(render_matrix_updates_yaml(toggle_updates, min_severity_removals, layout))
+
+        return
+
+    # ---------- Backup ----------
+    backup_path = backup_config(config_path)
+    logger.info(f"Backed up config to {backup_path}")
+
+    # ---------- Apply changes ----------
+    # For Phase A with no Phase B changes, use append strategy (preserves comments)
+    # For Phase B or combined, use yaml round-trip (loses comments but handles in-place edits)
+
+    if args.phase == "a" and sinks and not notifications.get("sinks"):
+        # Phase A only: use append strategy
+        success = write_sinks_to_config(config_path, sinks, layout, backup_path)
+        if not success:
+            logger.error("Phase A migration failed - config restored from backup")
+            sys.exit(1)
+        logger.info(f"Wrote sinks block to {config_path} ({sinks_path})")
+
+    elif toggle_updates or (sinks and not notifications.get("sinks")):
+        # Phase B or combined: use yaml round-trip
+        # Re-parse the config to modify in-place
+        with open(config_path, "r") as f:
+            config_to_modify = yaml.safe_load(f) or {}
+
+        modify_layout = detect_config_layout(config_to_modify)
+        modify_notifications = get_notifications_block(config_to_modify, modify_layout)
+
+        # Add sinks if needed
+        if sinks and not modify_notifications.get("sinks"):
+            modify_notifications["sinks"] = sinks
+            logger.info(f"Added sinks block ({len(sinks)} sink(s))")
+
+        # Apply matrix updates
+        if toggle_updates:
+            apply_matrix_updates(config_to_modify, modify_layout, toggle_updates)
+            logger.info(f"Applied matrix updates to {len(toggle_updates)} toggle(s)")
+
+        # Write the modified config
+        success = write_full_config(config_path, config_to_modify, modify_layout, backup_path)
+        if not success:
+            logger.error("Migration failed - config restored from backup")
+            sys.exit(1)
+
+    logger.info("Migration complete!")
+
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -13,3 +13,4 @@ h3>=4.0
 fastapi>=0.110.0
 uvicorn[standard]>=0.27.0
 aiomqtt>=2.0.0
+python-dotenv>=1.0.0

tests/test_notification_toggles.py

@@ -91,17 +91,20 @@ def test_region_matches_via_regions_list():
 
 
 def test_severity_threshold():
-    cfg = _cfg(min_severity="priority",
-               severity_channels={"routine": ["mesh_broadcast"], "priority": ["mesh_broadcast"],
+    # v0.7: min_severity is obsolete; the matrix IS the threshold.
+    # Empty row means no delivery for that severity.
+    cfg = _cfg(severity_channels={"routine": [],  # empty = no delivery
+                                  "priority": ["mesh_broadcast"],
                                   "immediate": ["mesh_broadcast"]})
-    assert _dispatch(cfg, _ev(severity="routine")) == []     # below threshold
+    assert _dispatch(cfg, _ev(severity="routine")) == []     # empty matrix row
     assert len(_dispatch(cfg, _ev(severity="priority"))) == 1
     assert len(_dispatch(cfg, _ev(severity="immediate"))) == 1
 
 
 def test_per_severity_channel_routing():
-    cfg = _cfg(min_severity="routine",
-               severity_channels={"priority": ["mesh_broadcast"],
+    # v0.7: min_severity removed; matrix defines routing for each severity
+    cfg = _cfg(severity_channels={"routine": [],  # no delivery for routine
+                                  "priority": ["mesh_broadcast"],
                                   "immediate": ["mesh_broadcast", "mesh_dm"]})
     assert len(_dispatch(cfg, _ev(severity="priority"))) == 1
     imm = _dispatch(cfg, _ev(severity="immediate"))

tests/test_sinks.py

@@ -0,0 +1,912 @@
+"""Tests for routing simplification: SinkConfig and sink utilities.
+
+Tests cover:
+1. SinkConfig dataclass conversion from dict
+2. Channel factory per sink type
+3. Migration synthesis + idempotence + dry-run
+"""
+
+import pytest
+import tempfile
+import os
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import yaml
+
+
+class TestSinkConfigDataclass:
+    """Tests for SinkConfig dataclass and dict conversion."""
+
+    def test_sink_config_defaults(self):
+        """SinkConfig has correct defaults."""
+        from meshai.config import SinkConfig
+
+        sink = SinkConfig()
+        assert sink.type == "mesh_broadcast"
+        assert sink.channel == 0
+        assert sink.node_ids == []
+        assert sink.smtp_host == ""
+        assert sink.recipients == []
+        assert sink.webhook_url == ""
+
+    def test_sink_config_mesh_broadcast(self):
+        """SinkConfig correctly stores mesh_broadcast config."""
+        from meshai.config import SinkConfig
+
+        sink = SinkConfig(type="mesh_broadcast", channel=2)
+        assert sink.type == "mesh_broadcast"
+        assert sink.channel == 2
+
+    def test_sink_config_mesh_dm(self):
+        """SinkConfig correctly stores mesh_dm config."""
+        from meshai.config import SinkConfig
+
+        sink = SinkConfig(type="mesh_dm", node_ids=["!abc123", "!def456"])
+        assert sink.type == "mesh_dm"
+        assert sink.node_ids == ["!abc123", "!def456"]
+
+    def test_sink_config_email(self):
+        """SinkConfig correctly stores email config."""
+        from meshai.config import SinkConfig
+
+        sink = SinkConfig(
+            type="email",
+            smtp_host="smtp.example.com",
+            smtp_port=465,
+            smtp_user="user",
+            smtp_password="pass",
+            smtp_tls=True,
+            from_address="alerts@example.com",
+            recipients=["ops@example.com"],
+        )
+        assert sink.type == "email"
+        assert sink.smtp_host == "smtp.example.com"
+        assert sink.smtp_port == 465
+        assert sink.recipients == ["ops@example.com"]
+
+    def test_sink_config_webhook(self):
+        """SinkConfig correctly stores webhook config."""
+        from meshai.config import SinkConfig
+
+        sink = SinkConfig(
+            type="webhook",
+            webhook_url="https://hooks.example.com/alert",
+            webhook_headers={"Authorization": "Bearer token"},
+        )
+        assert sink.type == "webhook"
+        assert sink.webhook_url == "https://hooks.example.com/alert"
+        assert sink.webhook_headers == {"Authorization": "Bearer token"}
+
+    def test_sink_config_validation_valid(self):
+        """SinkConfig.validate() returns empty list for valid configs."""
+        from meshai.config import SinkConfig
+
+        # Valid mesh_broadcast
+        sink = SinkConfig(type="mesh_broadcast", channel=0)
+        assert sink.validate() == []
+
+        # Valid mesh_dm
+        sink = SinkConfig(type="mesh_dm", node_ids=["!abc123"])
+        assert sink.validate() == []
+
+        # Valid email
+        sink = SinkConfig(type="email", smtp_host="smtp.test.com", recipients=["a@b.com"])
+        assert sink.validate() == []
+
+        # Valid webhook
+        sink = SinkConfig(type="webhook", webhook_url="https://example.com")
+        assert sink.validate() == []
+
+    def test_sink_config_validation_invalid_type(self):
+        """SinkConfig.validate() catches invalid type."""
+        from meshai.config import SinkConfig
+
+        sink = SinkConfig(type="invalid_type")
+        errors = sink.validate()
+        assert len(errors) == 1
+        assert "Invalid sink type" in errors[0]
+
+    def test_sink_config_validation_negative_channel(self):
+        """SinkConfig.validate() catches negative channel."""
+        from meshai.config import SinkConfig
+
+        sink = SinkConfig(type="mesh_broadcast", channel=-1)
+        errors = sink.validate()
+        assert len(errors) == 1
+        assert "must be >= 0" in errors[0]
+
+    def test_sink_config_validation_channel_zero_valid(self):
+        """SinkConfig.validate() accepts channel 0 (B6 fix verification)."""
+        from meshai.config import SinkConfig
+
+        sink = SinkConfig(type="mesh_broadcast", channel=0)
+        errors = sink.validate()
+        assert errors == []
+
+    def test_dict_to_dataclass_converts_sinks(self):
+        """_dict_to_dataclass correctly converts sinks dict to SinkConfig instances."""
+        from meshai.config import _dict_to_dataclass, Config, SinkConfig
+
+        config_dict = {
+            "notifications": {
+                "enabled": True,
+                "sinks": {
+                    "mesh-primary": {"type": "mesh_broadcast", "channel": 0},
+                    "mesh-alerts": {"type": "mesh_broadcast", "channel": 2},
+                    "dm-ops": {"type": "mesh_dm", "node_ids": ["!abcd1234"]},
+                },
+            },
+        }
+
+        config = _dict_to_dataclass(Config, config_dict)
+
+        assert hasattr(config.notifications, "sinks")
+        sinks = config.notifications.sinks
+
+        assert "mesh-primary" in sinks
+        assert isinstance(sinks["mesh-primary"], SinkConfig)
+        assert sinks["mesh-primary"].type == "mesh_broadcast"
+        assert sinks["mesh-primary"].channel == 0
+
+        assert "mesh-alerts" in sinks
+        assert isinstance(sinks["mesh-alerts"], SinkConfig)
+        assert sinks["mesh-alerts"].channel == 2
+
+        assert "dm-ops" in sinks
+        assert isinstance(sinks["dm-ops"], SinkConfig)
+        assert sinks["dm-ops"].type == "mesh_dm"
+        assert sinks["dm-ops"].node_ids == ["!abcd1234"]
+
+
+class TestCreateChannelFromSink:
+    """Tests for create_channel_from_sink factory function."""
+
+    def test_create_mesh_broadcast_channel(self):
+        """create_channel_from_sink creates MeshBroadcastChannel."""
+        from meshai.config import SinkConfig
+        from meshai.notifications.channels import (
+            create_channel_from_sink,
+            MeshBroadcastChannel,
+        )
+
+        sink = SinkConfig(type="mesh_broadcast", channel=2)
+        mock_connector = MagicMock()
+
+        channel = create_channel_from_sink(sink, connector=mock_connector)
+
+        assert isinstance(channel, MeshBroadcastChannel)
+        assert channel._channel == 2
+        assert channel._connector == mock_connector
+
+    def test_create_mesh_dm_channel(self):
+        """create_channel_from_sink creates MeshDMChannel."""
+        from meshai.config import SinkConfig
+        from meshai.notifications.channels import (
+            create_channel_from_sink,
+            MeshDMChannel,
+        )
+
+        sink = SinkConfig(type="mesh_dm", node_ids=["!abc123", "!def456"])
+        mock_connector = MagicMock()
+
+        channel = create_channel_from_sink(sink, connector=mock_connector)
+
+        assert isinstance(channel, MeshDMChannel)
+        assert channel._node_ids == ["!abc123", "!def456"]
+
+    def test_create_email_channel(self):
+        """create_channel_from_sink creates EmailChannel."""
+        from meshai.config import SinkConfig
+        from meshai.notifications.channels import (
+            create_channel_from_sink,
+            EmailChannel,
+        )
+
+        sink = SinkConfig(
+            type="email",
+            smtp_host="smtp.test.com",
+            smtp_port=587,
+            smtp_user="user",
+            smtp_password="pass",
+            smtp_tls=True,
+            from_address="alerts@test.com",
+            recipients=["ops@test.com"],
+        )
+
+        channel = create_channel_from_sink(sink)
+
+        assert isinstance(channel, EmailChannel)
+        assert channel._host == "smtp.test.com"
+        assert channel._port == 587
+        assert channel._recipients == ["ops@test.com"]
+
+    def test_create_webhook_channel(self):
+        """create_channel_from_sink creates WebhookChannel."""
+        from meshai.config import SinkConfig
+        from meshai.notifications.channels import (
+            create_channel_from_sink,
+            WebhookChannel,
+        )
+
+        sink = SinkConfig(
+            type="webhook",
+            webhook_url="https://hooks.test.com/alert",
+            webhook_headers={"X-Token": "secret"},
+        )
+
+        channel = create_channel_from_sink(sink)
+
+        assert isinstance(channel, WebhookChannel)
+        assert channel._url == "https://hooks.test.com/alert"
+        assert channel._headers == {"X-Token": "secret"}
+
+    def test_create_channel_invalid_type_raises(self):
+        """create_channel_from_sink raises ValueError for invalid type."""
+        from meshai.config import SinkConfig
+        from meshai.notifications.channels import create_channel_from_sink
+
+        sink = SinkConfig(type="invalid_type")
+
+        with pytest.raises(ValueError, match="Unknown sink type"):
+            create_channel_from_sink(sink)
+
+    def test_create_channel_negative_channel_raises(self):
+        """create_channel_from_sink raises ValueError for negative channel."""
+        from meshai.config import SinkConfig
+        from meshai.notifications.channels import create_channel_from_sink
+
+        sink = SinkConfig(type="mesh_broadcast", channel=-1)
+
+        with pytest.raises(ValueError, match="must be >= 0"):
+            create_channel_from_sink(sink)
+
+    def test_create_channel_zero_channel_valid(self):
+        """create_channel_from_sink accepts channel 0 (B6 fix verification)."""
+        from meshai.config import SinkConfig
+        from meshai.notifications.channels import (
+            create_channel_from_sink,
+            MeshBroadcastChannel,
+        )
+
+        sink = SinkConfig(type="mesh_broadcast", channel=0)
+        mock_connector = MagicMock()
+
+        channel = create_channel_from_sink(sink, connector=mock_connector)
+
+        assert isinstance(channel, MeshBroadcastChannel)
+        assert channel._channel == 0
+
+
+class TestMigrationSynthesis:
+    """Tests for migration script sink synthesis logic."""
+
+    def test_extract_sink_from_toggle_mesh_broadcast(self):
+        """extract_sink_from_toggle handles broadcast_channel."""
+        from meshai.scripts.migrate_config_routing import extract_sinks_from_toggle, extract_sink_from_toggle
+
+        toggle = {"broadcast_channel": 2, "enabled": True}
+        sink = extract_sink_from_toggle(toggle)
+
+        assert sink == {"type": "mesh_broadcast", "channel": 2}
+
+    def test_extract_sink_from_toggle_mesh_dm(self):
+        """extract_sink_from_toggle handles node_ids."""
+        from meshai.scripts.migrate_config_routing import extract_sinks_from_toggle, extract_sink_from_toggle
+
+        toggle = {"node_ids": ["!abc123"], "enabled": True}
+        sink = extract_sink_from_toggle(toggle)
+
+        assert sink == {"type": "mesh_dm", "node_ids": ["!abc123"]}
+
+    def test_extract_sink_from_toggle_email(self):
+        """extract_sink_from_toggle handles smtp_host."""
+        from meshai.scripts.migrate_config_routing import extract_sinks_from_toggle, extract_sink_from_toggle
+
+        toggle = {
+            "smtp_host": "smtp.test.com",
+            "smtp_port": 587,
+            "smtp_user": "user",
+            "smtp_password": "pass",
+            "smtp_tls": True,
+            "from_address": "alerts@test.com",
+            "recipients": ["ops@test.com"],
+        }
+        sink = extract_sink_from_toggle(toggle)
+
+        assert sink["type"] == "email"
+        assert sink["smtp_host"] == "smtp.test.com"
+        assert sink["recipients"] == ["ops@test.com"]
+
+    def test_extract_sink_from_toggle_webhook(self):
+        """extract_sink_from_toggle handles webhook_url."""
+        from meshai.scripts.migrate_config_routing import extract_sinks_from_toggle, extract_sink_from_toggle
+
+        toggle = {
+            "webhook_url": "https://hooks.test.com",
+            "webhook_headers": {"X-Token": "secret"},
+        }
+        sink = extract_sink_from_toggle(toggle)
+
+        assert sink == {
+            "type": "webhook",
+            "webhook_url": "https://hooks.test.com",
+            "webhook_headers": {"X-Token": "secret"},
+        }
+
+    def test_extract_sink_from_toggle_none(self):
+        """extract_sink_from_toggle returns None for toggle without transport."""
+        from meshai.scripts.migrate_config_routing import extract_sinks_from_toggle, extract_sink_from_toggle
+
+        toggle = {"enabled": True, "min_severity": "priority"}
+        sink = extract_sink_from_toggle(toggle)
+
+        assert sink is None
+
+    
+    def test_extract_sinks_from_toggle_multiple_transports(self):
+        """extract_sinks_from_toggle returns ALL configured transports."""
+        from meshai.scripts.migrate_config_routing import extract_sinks_from_toggle
+
+        toggle = {
+            "broadcast_channel": 1,
+            "node_ids": ["!abc123"],
+            "smtp_host": "",  # Empty = not configured
+            "webhook_url": "",  # Empty = not configured
+        }
+        sinks = extract_sinks_from_toggle(toggle)
+
+        assert len(sinks) == 2
+        types = {s["type"] for s in sinks}
+        assert types == {"mesh_broadcast", "mesh_dm"}
+
+    def test_extract_sinks_from_toggle_all_four_types(self):
+        """extract_sinks_from_toggle extracts all four transport types."""
+        from meshai.scripts.migrate_config_routing import extract_sinks_from_toggle
+
+        toggle = {
+            "broadcast_channel": 2,
+            "node_ids": ["!abc123"],
+            "smtp_host": "smtp.test.com",
+            "recipients": ["ops@test.com"],
+            "webhook_url": "https://hooks.test.com",
+        }
+        sinks = extract_sinks_from_toggle(toggle)
+
+        assert len(sinks) == 4
+        types = {s["type"] for s in sinks}
+        assert types == {"mesh_broadcast", "mesh_dm", "email", "webhook"}
+
+    def test_extract_sink_from_rule(self):
+        """extract_sink_from_rule handles delivery_type."""
+        from meshai.scripts.migrate_config_routing import extract_sink_from_rule
+
+        rule = {"delivery_type": "mesh_broadcast", "broadcast_channel": 3}
+        sink = extract_sink_from_rule(rule)
+
+        assert sink == {"type": "mesh_broadcast", "channel": 3}
+
+    def test_generate_sink_name_mesh_broadcast(self):
+        """generate_sink_name creates readable names for mesh_broadcast."""
+        from meshai.scripts.migrate_config_routing import generate_sink_name
+
+        name = generate_sink_name("mesh_broadcast", {"channel": 2})
+        assert name == "mesh-ch2"
+
+    def test_generate_sink_name_mesh_dm(self):
+        """generate_sink_name creates readable names for mesh_dm."""
+        from meshai.scripts.migrate_config_routing import generate_sink_name
+
+        name = generate_sink_name("mesh_dm", {"node_ids": ["!abcd1234"]})
+        assert name == "dm-abcd1234"
+
+    def test_generate_sink_name_email(self):
+        """generate_sink_name creates readable names for email."""
+        from meshai.scripts.migrate_config_routing import generate_sink_name
+
+        name = generate_sink_name("email", {"smtp_host": "smtp.example.com"})
+        assert name == "email-smtp"
+
+    def test_generate_sink_name_webhook(self):
+        """generate_sink_name creates readable names for webhook."""
+        from meshai.scripts.migrate_config_routing import generate_sink_name
+
+        name = generate_sink_name("webhook", {"webhook_url": "https://hooks.slack.com/abc"})
+        assert name == "webhook-hooks"
+
+    def test_synthesize_sinks_deduplicates(self):
+        """synthesize_sinks deduplicates identical transports."""
+        from meshai.scripts.migrate_config_routing import synthesize_sinks
+
+        notifications = {
+            "toggles": {
+                "fire": {"broadcast_channel": 2},
+                "weather": {"broadcast_channel": 2},  # Same channel
+            }
+        }
+
+        sinks, _ = synthesize_sinks(notifications)
+
+        # Should only have one sink for channel 2
+        assert len(sinks) == 1
+        assert "mesh-ch2" in sinks
+
+    def test_synthesize_sinks_handles_collisions(self):
+        """synthesize_sinks handles name collisions."""
+        from meshai.scripts.migrate_config_routing import synthesize_sinks
+
+        notifications = {
+            "toggles": {
+                "fire": {"broadcast_channel": 0},
+                "weather": {"broadcast_channel": 1},
+                "roads": {"broadcast_channel": 2},
+            }
+        }
+
+        sinks, _ = synthesize_sinks(notifications)
+
+        # Should have three unique sinks
+        assert len(sinks) == 3
+
+
+class TestMigrationIdempotence:
+    """Tests for migration script idempotence."""
+
+    def test_migration_refuses_if_sinks_exist(self):
+        """Migration refuses to run if sinks block already exists."""
+        from meshai.scripts.migrate_config_routing import load_notifications_config
+
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
+            yaml.dump({
+                "notifications": {
+                    "enabled": True,
+                    "sinks": {"mesh-primary": {"type": "mesh_broadcast"}},
+                }
+            }, f)
+            config_path = Path(f.name)
+
+        try:
+            _, notifications, _ = load_notifications_config(config_path)
+            assert notifications.get("sinks") is not None
+            # The main() function checks this and exits
+        finally:
+            os.unlink(config_path)
+
+    def test_backup_creates_timestamped_file(self):
+        """backup_config creates properly named backup."""
+        from meshai.scripts.migrate_config_routing import backup_config
+
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
+            f.write("test: true\n")
+            config_path = Path(f.name)
+
+        try:
+            backup_path = backup_config(config_path)
+            assert backup_path.exists()
+            assert ".pre-sinks." in str(backup_path)
+            assert backup_path.suffix == ".bak"
+        finally:
+            os.unlink(config_path)
+            if backup_path.exists():
+                os.unlink(backup_path)
+
+
+class TestWriteSinksToConfig:
+    """Tests for layout-aware, append-only sinks write."""
+
+    def test_write_sinks_multifile_layout(self):
+        """write_sinks_to_config handles multi-file layout correctly."""
+        from meshai.scripts.migrate_config_routing import (
+            write_sinks_to_config,
+            backup_config,
+            load_notifications_config,
+        )
+
+        # Create a multi-file config with comments
+        original_content = """# Notification config for MeshAI
+# This comment should be preserved
+
+toggles:
+  fire:
+    enabled: true
+    broadcast_channel: 1
+  weather:
+    enabled: true
+    broadcast_channel: 2
+
+rules: []
+"""
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
+            f.write(original_content)
+            config_path = Path(f.name)
+
+        backup_path = None
+        try:
+            backup_path = backup_config(config_path)
+            sinks = {
+                "mesh-ch1": {"type": "mesh_broadcast", "channel": 1},
+                "mesh-ch2": {"type": "mesh_broadcast", "channel": 2},
+            }
+
+            success = write_sinks_to_config(config_path, sinks, "multifile", backup_path)
+            assert success, "Write should succeed"
+
+            # Read back and verify
+            with open(config_path, "r") as f:
+                new_content = f.read()
+
+            # Comments should be preserved
+            assert "# Notification config for MeshAI" in new_content
+            assert "# This comment should be preserved" in new_content
+
+            # Original keys should be intact
+            _, notifications, layout = load_notifications_config(config_path)
+            assert layout == "multifile"
+            assert "toggles" in notifications
+            assert "rules" in notifications
+            assert "sinks" in notifications
+
+            # Sinks should be at root level (no notifications: wrapper created)
+            config = yaml.safe_load(new_content)
+            assert "notifications" not in config, "Should not create notifications: wrapper"
+            assert "sinks" in config, "Sinks should be at root"
+            assert set(config["sinks"].keys()) == {"mesh-ch1", "mesh-ch2"}
+        finally:
+            os.unlink(config_path)
+            if backup_path and backup_path.exists():
+                os.unlink(backup_path)
+
+    def test_write_sinks_monolithic_layout(self):
+        """write_sinks_to_config handles monolithic layout correctly."""
+        from meshai.scripts.migrate_config_routing import (
+            write_sinks_to_config,
+            backup_config,
+            load_notifications_config,
+        )
+
+        # Create a monolithic config
+        original_content = """# Main config file
+notifications:
+  enabled: true
+  toggles:
+    fire:
+      broadcast_channel: 1
+
+other_section:
+  key: value
+"""
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
+            f.write(original_content)
+            config_path = Path(f.name)
+
+        backup_path = None
+        try:
+            backup_path = backup_config(config_path)
+            sinks = {"mesh-ch1": {"type": "mesh_broadcast", "channel": 1}}
+
+            success = write_sinks_to_config(config_path, sinks, "monolithic", backup_path)
+            assert success, "Write should succeed"
+
+            # Read back and verify
+            _, notifications, layout = load_notifications_config(config_path)
+            assert layout == "monolithic"
+            assert "sinks" in notifications
+            assert "mesh-ch1" in notifications["sinks"]
+
+            # Other sections should be intact
+            with open(config_path, "r") as f:
+                config = yaml.safe_load(f)
+            assert "other_section" in config
+            assert config["other_section"]["key"] == "value"
+        finally:
+            os.unlink(config_path)
+            if backup_path and backup_path.exists():
+                os.unlink(backup_path)
+
+    def test_write_aborts_and_restores_on_bad_result(self):
+        """write_sinks_to_config restores backup if verification fails."""
+        from meshai.scripts.migrate_config_routing import (
+            backup_config,
+            verify_sinks_written,
+        )
+        import shutil
+
+        original_content = """toggles:
+  fire:
+    enabled: true
+"""
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
+            f.write(original_content)
+            config_path = Path(f.name)
+
+        backup_path = None
+        try:
+            backup_path = backup_config(config_path)
+
+            # Corrupt the file
+            with open(config_path, "w") as f:
+                f.write("invalid: yaml: content: [")
+
+            # Verify should fail
+            success, error = verify_sinks_written(
+                config_path,
+                {"mesh-ch1": {"type": "mesh_broadcast"}},
+                "multifile",
+                {"toggles"}
+            )
+            assert not success
+            assert "parse error" in error.lower() or "not found" in error.lower()
+
+            # Restore from backup
+            shutil.copy2(backup_path, config_path)
+
+            # Original content should be restored
+            with open(config_path, "r") as f:
+                restored = f.read()
+            assert "toggles:" in restored
+        finally:
+            os.unlink(config_path)
+            if backup_path and backup_path.exists():
+                os.unlink(backup_path)
+
+    def test_dry_run_reports_layout(self, capsys):
+        """--dry-run reports detected layout and write path."""
+        from meshai.scripts.migrate_config_routing import (
+            load_notifications_config,
+            synthesize_sinks,
+            render_sinks_yaml,
+        )
+
+        # Multi-file layout
+        multifile_content = """toggles:
+  fire:
+    broadcast_channel: 1
+"""
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
+            f.write(multifile_content)
+            config_path = Path(f.name)
+
+        try:
+            _, notifications, layout = load_notifications_config(config_path)
+            assert layout == "multifile"
+
+            sinks, _ = synthesize_sinks(notifications)
+            rendered = render_sinks_yaml(sinks, layout)
+
+            # The rendered YAML for multifile should not be indented
+            assert rendered.startswith("sinks:")
+            assert "  mesh-ch1:" in rendered
+
+            # For monolithic, it should be indented
+            mono_rendered = render_sinks_yaml(sinks, "monolithic")
+            assert mono_rendered.startswith("  sinks:")
+        finally:
+            os.unlink(config_path)
+
+    def test_generate_sink_name_int_node_ids(self):
+        """generate_sink_name handles integer node IDs without crashing."""
+        from meshai.scripts.migrate_config_routing import generate_sink_name
+
+        # Integer node ID (no leading !)
+        name = generate_sink_name("mesh_dm", {"node_ids": [123456789]})
+        assert name == "dm-12345678"
+
+        # String node ID with !
+        name = generate_sink_name("mesh_dm", {"node_ids": ["!abcd1234"]})
+        assert name == "dm-abcd1234"
+
+        # Mixed types
+        name = generate_sink_name("mesh_dm", {"node_ids": [987654321, "!xyz"]})
+        assert name == "dm-98765432"
+
+
+class TestMatrixMigration:
+    """Tests for Phase B: matrix rewrite migration."""
+
+    def test_build_channel_type_to_sink_map(self):
+        """build_channel_type_to_sink_map correctly maps channel types to sink names."""
+        from meshai.scripts.migrate_config_routing import (
+            build_channel_type_to_sink_map,
+            compute_sink_hash,
+        )
+
+        # Build hash_to_name as synthesize_sinks would
+        hash_to_name = {
+            compute_sink_hash({"type": "mesh_broadcast", "channel": 2}): "mesh-ch2",
+            compute_sink_hash({"type": "mesh_dm", "node_ids": ["!abc123"]}): "dm-abc12345",
+        }
+
+        toggle = {
+            "broadcast_channel": 2,
+            "node_ids": ["!abc123"],
+        }
+
+        mapping = build_channel_type_to_sink_map(toggle, hash_to_name)
+
+        assert mapping["mesh_broadcast"] == "mesh-ch2"
+        assert mapping["mesh_dm"] == "dm-abc12345"
+
+    def test_migrate_toggle_matrix_converts_channel_types(self):
+        """migrate_toggle_matrix converts channel types to sink names."""
+        from meshai.scripts.migrate_config_routing import migrate_toggle_matrix
+
+        toggle = {
+            "severity_channels": {
+                "routine": ["mesh_broadcast"],
+                "priority": ["mesh_broadcast"],
+                "immediate": ["mesh_broadcast", "mesh_dm"],
+            },
+            "min_severity": "routine",
+        }
+        channel_to_sink = {
+            "mesh_broadcast": "mesh-ch2",
+            "mesh_dm": "dm-ops",
+        }
+
+        new_matrix, changes = migrate_toggle_matrix("test", toggle, channel_to_sink)
+
+        assert new_matrix["routine"] == ["mesh-ch2"]
+        assert new_matrix["priority"] == ["mesh-ch2"]
+        assert new_matrix["immediate"] == ["mesh-ch2", "dm-ops"]
+        assert len(changes) > 0
+
+    def test_migrate_toggle_matrix_blanks_below_min_severity(self):
+        """migrate_toggle_matrix blanks rows below min_severity threshold."""
+        from meshai.scripts.migrate_config_routing import migrate_toggle_matrix
+
+        toggle = {
+            "severity_channels": {
+                "routine": ["mesh_broadcast"],
+                "priority": ["mesh_broadcast"],
+                "immediate": ["mesh_broadcast"],
+            },
+            "min_severity": "priority",  # routine should be blanked
+        }
+        channel_to_sink = {"mesh_broadcast": "mesh-ch0"}
+
+        new_matrix, changes = migrate_toggle_matrix("test", toggle, channel_to_sink)
+
+        assert new_matrix["routine"] == [], "routine should be blanked (below priority)"
+        assert new_matrix["priority"] == ["mesh-ch0"]
+        assert new_matrix["immediate"] == ["mesh-ch0"]
+        assert any("blanked" in c for c in changes)
+
+    def test_migrate_toggle_matrix_removes_digest(self):
+        """migrate_toggle_matrix removes 'digest' pseudo-channel."""
+        from meshai.scripts.migrate_config_routing import migrate_toggle_matrix
+
+        toggle = {
+            "severity_channels": {
+                "routine": ["mesh_broadcast", "digest"],
+                "priority": ["mesh_broadcast", "digest"],
+                "immediate": ["mesh_broadcast"],
+            },
+            "min_severity": "routine",
+        }
+        channel_to_sink = {"mesh_broadcast": "mesh-ch0"}
+
+        new_matrix, changes = migrate_toggle_matrix("test", toggle, channel_to_sink)
+
+        # digest should be removed
+        assert "digest" not in new_matrix["routine"]
+        assert "digest" not in new_matrix["priority"]
+        assert any("digest" in c for c in changes)
+
+    def test_migrate_all_matrices_processes_all_toggles(self):
+        """migrate_all_matrices processes all toggles with inline transport."""
+        from meshai.scripts.migrate_config_routing import (
+            migrate_all_matrices,
+            compute_sink_hash,
+        )
+
+        notifications = {
+            "toggles": {
+                "fire": {
+                    "broadcast_channel": 1,
+                    "severity_channels": {
+                        "routine": [],
+                        "priority": ["mesh_broadcast"],
+                        "immediate": ["mesh_broadcast"],
+                    },
+                    "min_severity": "priority",
+                },
+                "weather": {
+                    "broadcast_channel": 2,
+                    "severity_channels": {
+                        "routine": ["mesh_broadcast"],
+                        "priority": ["mesh_broadcast"],
+                        "immediate": ["mesh_broadcast"],
+                    },
+                    "min_severity": "routine",
+                },
+            }
+        }
+
+        hash_to_name = {
+            compute_sink_hash({"type": "mesh_broadcast", "channel": 1}): "mesh-ch1",
+            compute_sink_hash({"type": "mesh_broadcast", "channel": 2}): "mesh-ch2",
+        }
+
+        toggle_updates, all_changes = migrate_all_matrices(notifications, hash_to_name)
+
+        # Both toggles should have updates
+        assert "fire" in toggle_updates
+        assert "weather" in toggle_updates
+
+        # Fire's matrix should use mesh-ch1
+        assert toggle_updates["fire"]["priority"] == ["mesh-ch1"]
+
+        # Weather's matrix should use mesh-ch2
+        assert toggle_updates["weather"]["routine"] == ["mesh-ch2"]
+
+    def test_apply_matrix_updates_modifies_config(self):
+        """apply_matrix_updates correctly modifies config in-place."""
+        from meshai.scripts.migrate_config_routing import apply_matrix_updates
+
+        config = {
+            "toggles": {
+                "fire": {
+                    "enabled": True,
+                    "min_severity": "priority",
+                    "severity_channels": {
+                        "routine": ["mesh_broadcast"],
+                        "priority": ["mesh_broadcast"],
+                        "immediate": ["mesh_broadcast"],
+                    },
+                }
+            }
+        }
+
+        toggle_updates = {
+            "fire": {
+                "routine": [],
+                "priority": ["mesh-ch1"],
+                "immediate": ["mesh-ch1"],
+            }
+        }
+
+        apply_matrix_updates(config, "multifile", toggle_updates)
+
+        # Check matrix was updated
+        assert config["toggles"]["fire"]["severity_channels"]["routine"] == []
+        assert config["toggles"]["fire"]["severity_channels"]["priority"] == ["mesh-ch1"]
+
+        # Check min_severity was removed
+        assert "min_severity" not in config["toggles"]["fire"]
+
+    def test_full_migration_synthesis_and_matrix(self):
+        """Full migration: synthesize sinks + rewrite matrices."""
+        from meshai.scripts.migrate_config_routing import (
+            synthesize_sinks,
+            migrate_all_matrices,
+            apply_matrix_updates,
+        )
+
+        notifications = {
+            "toggles": {
+                "fire": {
+                    "enabled": True,
+                    "broadcast_channel": 0,
+                    "min_severity": "priority",
+                    "severity_channels": {
+                        "routine": ["mesh_broadcast"],
+                        "priority": ["mesh_broadcast"],
+                        "immediate": ["mesh_broadcast"],
+                    },
+                },
+            }
+        }
+
+        # Phase A: synthesize sinks
+        sinks, hash_to_name = synthesize_sinks(notifications)
+        assert "mesh-ch0" in sinks
+
+        # Phase B: migrate matrices
+        toggle_updates, changes = migrate_all_matrices(notifications, hash_to_name)
+        assert "fire" in toggle_updates
+
+        # Routine should be blanked (below min_severity=priority)
+        assert toggle_updates["fire"]["routine"] == []
+        # Priority and immediate should have sink name
+        assert toggle_updates["fire"]["priority"] == ["mesh-ch0"]
+        assert toggle_updates["fire"]["immediate"] == ["mesh-ch0"]

tests/test_v07_dispatcher.py

@@ -0,0 +1,312 @@
+"""v0.7 — Guard reorder (B13 fix) + sink-name routing.
+
+Tests verify:
+1. Events with empty matrix rows don't arm cooldown or record dedup
+2. Sink name resolution works correctly
+3. Events below old min_severity threshold (now empty matrix row) don't suppress future events
+"""
+
+import asyncio
+import time
+from unittest.mock import patch, MagicMock
+
+import pytest
+
+from meshai.config import Config, SinkConfig
+from meshai.notifications.pipeline.dispatcher import Dispatcher
+from meshai.notifications.events import make_event
+
+
+# ---------------------------------------------------------------- helpers
+
+
+class RecChannel:
+    """Channel recorder that captures deliveries."""
+
+    def __init__(self, rec, sink_name=None):
+        self.rec = rec
+        self.sink_name = sink_name
+
+    async def deliver(self, payload, rule):
+        self.rec.append({
+            "sink": self.sink_name,
+            "message": payload.message,
+            "category": payload.category,
+            "severity": payload.severity,
+        })
+        return True
+
+
+def _make_dispatcher_with_sinks(cfg):
+    """Create dispatcher with mock channel factory.
+
+    Returns a dispatcher instance and the rec list where deliveries are recorded.
+    Uses mock connector that makes channels work without real radio.
+    """
+    rec: list = []
+
+    # Create a mock connector
+    mock_connector = MagicMock()
+    mock_connector.send_message = MagicMock()
+
+    # Create dispatcher - it uses create_channel_from_sink internally
+    d = Dispatcher(cfg, lambda rule, conn: None, connector=mock_connector)
+
+    # Track deliveries by wrapping the dispatch method
+    original_dispatch = d._dispatch_toggles
+
+    async def recording_dispatch(event):
+        # Track what would be delivered
+        await original_dispatch(event)
+
+    d._dispatch_toggles = recording_dispatch
+
+    # Also track via the connector's send_message calls
+    def track_send(*args, **kwargs):
+        rec.append({
+            "text": kwargs.get("text") or (args[0] if args else ""),
+            "channel": kwargs.get("channel"),
+        })
+
+    mock_connector.send_message.side_effect = track_send
+
+    return d, rec
+
+
+def _cfg_with_sinks(**kw):
+    """Create a config with sinks defined and sink-name routing."""
+    cfg = Config()
+    cfg.notifications.rules = []
+    cfg.notifications.cold_start_grace_seconds = 0
+
+    # Define sinks
+    cfg.notifications.sinks = {
+        "mesh-ch0": SinkConfig(type="mesh_broadcast", channel=0),
+        "mesh-ch2": SinkConfig(type="mesh_broadcast", channel=2),
+    }
+
+    # Configure toggle with sink-name routing
+    toggle_name = kw.get("toggle_name", "weather")
+    t = cfg.notifications.toggles[toggle_name]
+    t.enabled = True
+    t.regions = kw.get("regions", [])
+    # v0.7: severity_channels uses sink names, not channel types
+    t.severity_channels = kw.get("severity_channels", {
+        "routine": [],  # Empty - routine events should not deliver
+        "priority": ["mesh-ch0"],
+        "immediate": ["mesh-ch0", "mesh-ch2"],
+    })
+    # v0.7: min_severity is removed - matrix is the only gate
+    # But we still test that empty rows work correctly
+    t.freshness_seconds = kw.get("freshness_seconds", 600)
+    t.cooldown_seconds = kw.get("cooldown_seconds", 300)
+    return cfg
+
+
+def _ev(severity="priority", category="weather_warning",
+        timestamp=None, region=None, source="nws", title="t",
+        event_id=None, **kw):
+    """Build an Event."""
+    extra = dict(kw)
+    if timestamp is not None:
+        extra["timestamp"] = timestamp
+    if event_id is not None:
+        extra["id"] = event_id
+    return make_event(
+        source=source, category=category, severity=severity,
+        region=region, title=title, **extra,
+    )
+
+
+# ============================================================== B13 Tests
+# Guard reorder: events with empty matrix rows must NOT arm cooldown or record dedup
+
+
+def test_empty_matrix_row_does_not_arm_cooldown():
+    """B13 fix: routine event with empty matrix row must not arm cooldown.
+
+    Sequence:
+    1. Send routine event (empty matrix row) - should not deliver, should not arm cooldown
+    2. Send priority event with same (toggle, category, region) - should deliver
+       If cooldown was armed by #1, #2 would be throttled (failure).
+    """
+    cfg = _cfg_with_sinks(
+        cooldown_seconds=300,
+        severity_channels={
+            "routine": [],  # Empty - B13 scenario
+            "priority": ["mesh-ch0"],
+            "immediate": ["mesh-ch0"],
+        }
+    )
+    d, rec = _make_dispatcher_with_sinks(cfg)
+
+    # Event 1: routine (empty matrix row)
+    e1 = _ev(severity="routine", region="Magic Valley")
+    asyncio.run(d.dispatch(e1))
+    assert len(rec) == 0, "routine event with empty matrix should not deliver"
+
+    # Event 2: priority (same toggle, category, region - would be throttled if cooldown was armed)
+    e2 = _ev(severity="priority", region="Magic Valley", event_id="e2")
+    asyncio.run(d.dispatch(e2))
+    assert len(rec) == 1, "priority event should deliver (cooldown not armed by routine)"
+    assert d.dispatch_stats()["cooldown_dropped"] == 0
+
+
+def test_empty_matrix_row_does_not_record_dedup():
+    """B13 fix: routine event with empty matrix row must not record dedup.
+
+    Sequence:
+    1. Send routine event (empty matrix row) with id="test-123"
+    2. Send priority event with same (source, id) - should deliver
+       If dedup was recorded by #1, #2 would be dropped (failure).
+    """
+    cfg = _cfg_with_sinks(
+        cooldown_seconds=0,  # Disable cooldown for this test
+        severity_channels={
+            "routine": [],  # Empty - B13 scenario
+            "priority": ["mesh-ch0"],
+            "immediate": ["mesh-ch0"],
+        }
+    )
+    d, rec = _make_dispatcher_with_sinks(cfg)
+
+    # Event 1: routine with specific id (empty matrix row)
+    e1 = _ev(severity="routine", event_id="test-b13-dedup")
+    asyncio.run(d.dispatch(e1))
+    assert len(rec) == 0, "routine event with empty matrix should not deliver"
+
+    # Event 2: priority with SAME id - should deliver if dedup wasn't recorded
+    e2 = _ev(severity="priority", event_id="test-b13-dedup")
+    asyncio.run(d.dispatch(e2))
+    assert len(rec) == 1, "priority event should deliver (dedup not recorded by routine)"
+    assert d.dispatch_stats()["dedup_dropped"] == 0
+
+
+def test_region_filter_before_cooldown():
+    """B13 fix: region-filtered event must not arm cooldown.
+
+    If region filter runs before cooldown (correct B13 order), an event
+    that fails region filter won't arm cooldown for future events.
+    """
+    cfg = _cfg_with_sinks(
+        cooldown_seconds=300,
+        regions=["Boise"],  # Toggle only accepts Boise region
+        severity_channels={
+            "routine": ["mesh-ch0"],
+            "priority": ["mesh-ch0"],
+            "immediate": ["mesh-ch0"],
+        }
+    )
+    d, rec = _make_dispatcher_with_sinks(cfg)
+
+    # Event 1: wrong region - should be filtered, should NOT arm cooldown
+    e1 = _ev(severity="priority", region="Magic Valley")
+    asyncio.run(d.dispatch(e1))
+    assert len(rec) == 0, "wrong region event should not deliver"
+
+    # Event 2: correct region, same category - should deliver if cooldown not armed
+    e2 = _ev(severity="priority", region="Boise", event_id="e2")
+    asyncio.run(d.dispatch(e2))
+    assert len(rec) == 1, "correct region event should deliver"
+    assert d.dispatch_stats()["cooldown_dropped"] == 0
+
+
+# ============================================================== Sink Resolution Tests
+
+
+def test_sink_name_resolution_delivers():
+    """Verify sink-name routing delivers via correct sinks."""
+    cfg = _cfg_with_sinks(
+        severity_channels={
+            "routine": [],
+            "priority": ["mesh-ch0"],
+            "immediate": ["mesh-ch0", "mesh-ch2"],
+        }
+    )
+    d, rec = _make_dispatcher_with_sinks(cfg)
+
+    # Priority event should deliver to mesh-ch0
+    e = _ev(severity="priority")
+    asyncio.run(d.dispatch(e))
+    assert len(rec) == 1
+
+
+def test_unknown_sink_name_logged_not_delivered():
+    """Unknown sink name in matrix should log warning, not crash."""
+    cfg = _cfg_with_sinks(
+        severity_channels={
+            "routine": [],
+            "priority": ["nonexistent-sink"],
+            "immediate": ["mesh-ch0"],
+        }
+    )
+    d, rec = _make_dispatcher_with_sinks(cfg)
+
+    # Priority event references unknown sink - should not deliver (graceful)
+    e = _ev(severity="priority")
+    asyncio.run(d.dispatch(e))
+    assert len(rec) == 0, "unknown sink should not deliver"
+
+
+def test_multiple_sinks_in_matrix_row():
+    """Multiple sinks in one severity row should all be attempted."""
+    cfg = _cfg_with_sinks(
+        severity_channels={
+            "routine": [],
+            "priority": ["mesh-ch0", "mesh-ch2"],
+            "immediate": ["mesh-ch0", "mesh-ch2"],
+        }
+    )
+    d, rec = _make_dispatcher_with_sinks(cfg)
+
+    # Priority event should attempt delivery to both sinks
+    e = _ev(severity="priority")
+    asyncio.run(d.dispatch(e))
+    # Note: with current RecChannel mock, we get one record per sink
+    # But the actual count depends on how the mock is structured
+    assert len(rec) >= 1, "at least one sink should deliver"
+
+
+# ============================================================== Guard Order Verification
+
+
+def test_guard_order_staleness_before_cooldown():
+    """Staleness filter must run before cooldown (unchanged in v0.7)."""
+    cfg = _cfg_with_sinks(
+        freshness_seconds=600,
+        cooldown_seconds=300,
+        severity_channels={
+            "priority": ["mesh-ch0"],
+            "immediate": ["mesh-ch0"],
+        }
+    )
+    d, rec = _make_dispatcher_with_sinks(cfg)
+
+    # Stale event should be dropped at staleness, not arm cooldown
+    stale = _ev(severity="priority", timestamp=time.time() - 7200)
+    asyncio.run(d.dispatch(stale))
+    assert len(rec) == 0
+    assert d.dispatch_stats()["stale_dropped"] == 1
+    assert d.dispatch_stats()["cooldown_dropped"] == 0
+
+
+def test_guard_order_cold_start_first():
+    """Cold-start grace runs first (unchanged in v0.7)."""
+    cfg = Config()
+    cfg.notifications.rules = []
+    cfg.notifications.cold_start_grace_seconds = 60  # Enable grace
+    cfg.notifications.sinks = {
+        "mesh-ch0": SinkConfig(type="mesh_broadcast", channel=0),
+    }
+    t = cfg.notifications.toggles["weather"]
+    t.enabled = True
+    t.severity_channels = {"priority": ["mesh-ch0"]}
+    t.cooldown_seconds = 0
+
+    d, rec = _make_dispatcher_with_sinks(cfg)
+
+    # First event sets anchor and is dropped by grace
+    e1 = _ev(severity="priority")
+    asyncio.run(d.dispatch(e1))
+    assert len(rec) == 0
+    assert d.dispatch_stats()["cold_start_dropped"] == 1