fix(notifications): test shows live data, not just canned examples

- Test always shows current data for the rule's feed categories
- RF rules show live SFI/Kp/R/S/G and ducting conditions
- Weather rules show active NWS alert count and headlines
- Fire rules show active fire/hotspot count
- Stream rules show current gauge readings
- Mesh rules show current health score and infra status
- Send Current Conditions delivers live snapshot through channel
- Send Test Alert delivers example through channel
- Send Live Alert available when real conditions match

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

meshai/commands/roads_cmd.py

@@ -0,0 +1,74 @@
+"""Road conditions command."""
+
+from .base import CommandContext, CommandHandler
+
+
+class RoadsCommand(CommandHandler):
+    """Show traffic flow and road conditions."""
+
+    aliases = ["traffic", "highways"]
+
+    def __init__(self, env_store):
+        self._env_store = env_store
+        self._name = "roads"
+
+    @property
+    def name(self) -> str:
+        return self._name
+
+    @name.setter
+    def name(self, value: str):
+        self._name = value
+
+    @property
+    def description(self) -> str:
+        return "Show traffic flow and road conditions"
+
+    @property
+    def usage(self) -> str:
+        return "!roads"
+
+    async def execute(self, args: str, context: CommandContext) -> str:
+        if not self._env_store:
+            return "Environmental feeds not configured."
+
+        traffic_events = self._env_store.get_active(source="traffic")
+        road_events = self._env_store.get_active(source="511")
+
+        if not traffic_events and not road_events:
+            return "No traffic or road data available. Check if sources are configured."
+
+        lines = []
+
+        # Traffic flow from TomTom
+        if traffic_events:
+            lines.append("Traffic Flow:")
+            for event in traffic_events:
+                props = event.get("properties", {})
+                corridor = props.get("corridor", "Unknown")
+                current = props.get("currentSpeed", 0)
+                free_flow = props.get("freeFlowSpeed", 0)
+                ratio = props.get("speedRatio", 1.0)
+                closure = props.get("roadClosure", False)
+
+                if closure:
+                    lines.append(f"  {corridor}: CLOSED")
+                else:
+                    pct = int(ratio * 100)
+                    lines.append(f"  {corridor}: {int(current)}mph ({pct}% of {int(free_flow)}mph)")
+
+        # 511 road events
+        if road_events:
+            if traffic_events:
+                lines.append("")  # Separator
+            lines.append("Road Events:")
+            for event in road_events:
+                event_type = event.get("event_type", "Event")
+                headline = event.get("headline", "")[:80]
+                props = event.get("properties", {})
+                is_closure = props.get("is_closure", False)
+
+                icon = "X" if is_closure else "-"
+                lines.append(f"  {icon} {headline}")
+
+        return "\n".join(lines) if lines else "No road conditions data."

meshai/commands/streams_cmd.py

@@ -0,0 +1,73 @@
+"""Stream gauge command."""
+
+from .base import CommandContext, CommandHandler
+
+
+class StreamsCommand(CommandHandler):
+    """Show current stream gauge readings."""
+
+    aliases = ["gauges", "rivers"]
+
+    def __init__(self, env_store):
+        self._env_store = env_store
+        self._name = "streams"
+
+    @property
+    def name(self) -> str:
+        return self._name
+
+    @name.setter
+    def name(self, value: str):
+        self._name = value
+
+    @property
+    def description(self) -> str:
+        return "Show stream gauge readings"
+
+    @property
+    def usage(self) -> str:
+        return "!streams"
+
+    async def execute(self, args: str, context: CommandContext) -> str:
+        if not self._env_store:
+            return "Environmental feeds not configured."
+
+        events = self._env_store.get_active(source="usgs")
+
+        if not events:
+            return "No stream gauge data available. Check if USGS sites are configured."
+
+        lines = []
+
+        # Group by site
+        sites = {}
+        for event in events:
+            props = event.get("properties", {})
+            site_id = props.get("site_id", "")
+            site_name = props.get("site_name", "Unknown")
+
+            if site_id not in sites:
+                sites[site_id] = {"name": site_name, "readings": []}
+
+            param = props.get("parameter", "")
+            value = props.get("value", 0)
+            unit = props.get("unit", "")
+
+            sites[site_id]["readings"].append((param, value, unit))
+
+        for site_id, data in sites.items():
+            name = data["name"]
+            readings = data["readings"]
+
+            # Format readings
+            parts = []
+            for param, value, unit in readings:
+                if "flow" in param.lower() or unit == "ft3/s":
+                    parts.append(f"{value:,.0f} {unit}")
+                else:
+                    parts.append(f"{value:.1f} {unit}")
+
+            reading_str = ", ".join(parts)
+            lines.append(f"{name}: {reading_str}")
+
+        return "\n".join(lines) if lines else "No stream gauge readings."

meshai/commands/subscribe.py

@@ -8,6 +8,7 @@ if TYPE_CHECKING:
     from ..mesh_data_store import MeshDataStore
     from ..mesh_reporter import MeshReporter
     from ..subscriptions import SubscriptionManager
+    from ..notifications.router import NotificationRouter
 
 
 class SubCommand(CommandHandler):
@@ -15,7 +16,7 @@ class SubCommand(CommandHandler):
 
     name = "sub"
     description = "Subscribe to reports or alerts"
-    usage = "!sub daily|weekly|alerts [time] [day] [scope]"
+    usage = "!sub daily|weekly|alerts|<category> [time] [day] [scope]"
     aliases = ["subscribe"]
 
     def __init__(
@@ -23,23 +24,35 @@ class SubCommand(CommandHandler):
         subscription_manager: "SubscriptionManager" = None,
         mesh_reporter: "MeshReporter" = None,
         data_store: "MeshDataStore" = None,
+        notification_router: "NotificationRouter" = None,
     ):
         self._sub_manager = subscription_manager
         self._reporter = mesh_reporter
         self._data_store = data_store
+        self._notification_router = notification_router
 
     async def execute(self, args: str, context: CommandContext) -> str:
         """Handle subscription command."""
-        if not self._sub_manager:
-            return "Subscriptions not available."
-
         parts = args.strip().split()
+
+        # No args - show available alert categories
         if not parts:
-            return self._usage_help()
+            return self._show_categories()
 
         sub_type = parts[0].lower()
+
+        # Check if it's a category subscription
+        if self._notification_router:
+            from ..notifications.categories import ALERT_CATEGORIES
+            if sub_type in ALERT_CATEGORIES or sub_type == "all":
+                return self._handle_category_subscription(sub_type, context)
+
+        # Legacy subscription types
         if sub_type not in ("daily", "weekly", "alerts"):
-            return f"Invalid type '{sub_type}'. Use: daily, weekly, or alerts"
+            return self._show_categories()
+
+        if not self._sub_manager:
+            return "Subscriptions not available."
 
         try:
             if sub_type == "daily":
@@ -51,15 +64,55 @@ class SubCommand(CommandHandler):
         except ValueError as e:
             return f"Error: {e}"
 
+    def _show_categories(self) -> str:
+        """Show available alert categories."""
+        try:
+            from ..notifications.categories import ALERT_CATEGORIES
+        except ImportError:
+            return self._usage_help()
+
+        lines = ["Available alert categories:"]
+        for cat_id, cat_info in ALERT_CATEGORIES.items():
+            lines.append(f"  {cat_id} - {cat_info['description']}")
+        lines.append("")
+        lines.append("Usage:")
+        lines.append("  !sub <category>  - subscribe to a category")
+        lines.append("  !sub all         - subscribe to all alerts")
+        lines.append("  !sub alerts      - legacy mesh-wide alerts")
+
+        return "\n".join(lines)
+
+    def _handle_category_subscription(self, category: str, context: CommandContext) -> str:
+        """Handle category-based alert subscription."""
+        node_id = self._get_user_id(context)
+
+        if category == "all":
+            categories = []  # Empty = all categories
+        else:
+            categories = [category]
+
+        # Add subscription via notification router
+        rule_name = self._notification_router.add_mesh_subscription(
+            node_id=node_id,
+            categories=categories,
+        )
+
+        if category == "all":
+            return "Subscribed to all alert categories. Use !unsub to remove."
+        else:
+            from ..notifications.categories import get_category
+            cat_info = get_category(category)
+            return f"Subscribed to {cat_info['name']} alerts. Use !unsub {category} to remove."
+
     def _usage_help(self) -> str:
         """Return usage help."""
         return """Usage:
 !sub daily 1830              - daily mesh report at 6:30 PM
 !sub daily 1830 region SCID  - daily region report
-!sub daily 1830 node MHR     - daily node report
 !sub weekly 0800 sun         - weekly digest Sunday 8 AM
-!sub alerts                  - mesh-wide alerts
-!sub alerts region SCID      - alerts for a region"""
+!sub alerts                  - mesh-wide alerts (legacy)
+!sub <category>              - subscribe to alert category
+!sub all                     - subscribe to all alerts"""
 
     def _handle_daily(self, args: list, context: CommandContext) -> str:
         """Handle daily subscription."""
@@ -68,11 +121,9 @@ class SubCommand(CommandHandler):
 
         schedule_time = args[0]
         scope_type, scope_value = self._parse_scope(args[1:])
-
-        # Validate scope
         scope_value = self._validate_scope(scope_type, scope_value)
 
-        result = self._sub_manager.add(
+        self._sub_manager.add(
             user_id=self._get_user_id(context),
             sub_type="daily",
             schedule_time=schedule_time,
@@ -92,11 +143,9 @@ class SubCommand(CommandHandler):
         schedule_time = args[0]
         schedule_day = args[1].lower()
         scope_type, scope_value = self._parse_scope(args[2:])
-
-        # Validate scope
         scope_value = self._validate_scope(scope_type, scope_value)
 
-        result = self._sub_manager.add(
+        self._sub_manager.add(
             user_id=self._get_user_id(context),
             sub_type="weekly",
             schedule_time=schedule_time,
@@ -111,13 +160,11 @@ class SubCommand(CommandHandler):
         return f"Subscribed: weekly {scope_desc}report at {time_fmt} {day_fmt}"
 
     def _handle_alerts(self, args: list, context: CommandContext) -> str:
-        """Handle alerts subscription."""
+        """Handle alerts subscription (legacy)."""
         scope_type, scope_value = self._parse_scope(args)
-
-        # Validate scope
         scope_value = self._validate_scope(scope_type, scope_value)
 
-        result = self._sub_manager.add(
+        self._sub_manager.add(
             user_id=self._get_user_id(context),
             sub_type="alerts",
             scope_type=scope_type,
@@ -128,15 +175,10 @@ class SubCommand(CommandHandler):
         return f"Subscribed: alerts for {scope_desc.strip() or 'mesh'}"
 
     def _parse_scope(self, args: list) -> tuple[str, str]:
-        """Parse scope from remaining args.
-
-        Returns:
-            (scope_type, scope_value) tuple
-        """
+        """Parse scope from remaining args."""
         if not args:
             return "mesh", None
 
-        # Look for 'region' or 'node' keyword
         scope_type = "mesh"
         scope_value = None
 
@@ -144,26 +186,17 @@ class SubCommand(CommandHandler):
             arg_lower = arg.lower()
             if arg_lower == "region":
                 scope_type = "region"
-                # Everything after 'region' is the region name
                 scope_value = " ".join(args[i + 1:]) if i + 1 < len(args) else None
                 break
             elif arg_lower == "node":
                 scope_type = "node"
-                # Next arg is the node identifier
                 scope_value = args[i + 1] if i + 1 < len(args) else None
                 break
 
         return scope_type, scope_value
 
     def _validate_scope(self, scope_type: str, scope_value: str) -> str:
-        """Validate and resolve scope value.
-
-        Returns:
-            Resolved scope_value (e.g., full region name)
-
-        Raises:
-            ValueError: If scope not found
-        """
+        """Validate and resolve scope value."""
         if scope_type == "mesh":
             return None
 
@@ -172,14 +205,9 @@ class SubCommand(CommandHandler):
 
         if scope_type == "region" and self._reporter:
             region = self._reporter._find_region(scope_value)
-            if not region:
-                # List available regions
-                health = self._reporter.health_engine.mesh_health
-                if health:
-                    available = [r.name for r in health.regions if r.node_ids]
-                    return scope_value  # Use as-is, will fail at delivery if invalid
-                raise ValueError(f"Region '{scope_value}' not found")
-            return region.name  # Return canonical name
+            if region:
+                return region.name
+            return scope_value
 
         if scope_type == "node" and self._reporter:
             node = self._reporter._find_node(scope_value)
@@ -191,7 +219,6 @@ class SubCommand(CommandHandler):
 
     def _get_user_id(self, context: CommandContext) -> str:
         """Extract user ID from context."""
-        # sender_id is like "!abcd1234" - convert to node_num
         sender_id = context.sender_id
         if sender_id.startswith("!"):
             return str(int(sender_id[1:], 16))
@@ -217,26 +244,40 @@ class UnsubCommand(CommandHandler):
 
     name = "unsub"
     description = "Remove subscription(s)"
-    usage = "!unsub daily|weekly|alerts|all"
+    usage = "!unsub daily|weekly|alerts|<category>|all"
     aliases = ["unsubscribe"]
 
-    def __init__(self, subscription_manager: "SubscriptionManager" = None):
+    def __init__(
+        self,
+        subscription_manager: "SubscriptionManager" = None,
+        notification_router: "NotificationRouter" = None,
+    ):
         self._sub_manager = subscription_manager
+        self._notification_router = notification_router
 
     async def execute(self, args: str, context: CommandContext) -> str:
         """Handle unsubscribe command."""
-        if not self._sub_manager:
-            return "Subscriptions not available."
-
         sub_type = args.strip().lower() if args else None
 
         if not sub_type:
-            return "Usage: !unsub daily|weekly|alerts|all"
-
-        if sub_type not in ("daily", "weekly", "alerts", "all"):
-            return f"Invalid type '{sub_type}'. Use: daily, weekly, alerts, or all"
+            return "Usage: !unsub daily|weekly|alerts|<category>|all"
 
         user_id = self._get_user_id(context)
+
+        # Check if it's a category unsubscription
+        if self._notification_router:
+            from ..notifications.categories import ALERT_CATEGORIES
+            if sub_type in ALERT_CATEGORIES or sub_type == "all":
+                self._notification_router.remove_mesh_subscription(user_id)
+                return "Removed alert subscriptions"
+
+        # Legacy subscription types
+        if not self._sub_manager:
+            return "Subscriptions not available."
+
+        if sub_type not in ("daily", "weekly", "alerts", "all"):
+            return f"Invalid type '{sub_type}'. Use: daily, weekly, alerts, <category>, or all"
+
         removed = self._sub_manager.remove(user_id, sub_type if sub_type != "all" else None)
 
         if removed == 0:
@@ -260,26 +301,44 @@ class MySubsCommand(CommandHandler):
     name = "mysubs"
     description = "List your subscriptions"
     usage = "!mysubs"
-    aliases = ["subs"]
+    aliases = ["subs", "subscriptions"]
 
-    def __init__(self, subscription_manager: "SubscriptionManager" = None):
+    def __init__(
+        self,
+        subscription_manager: "SubscriptionManager" = None,
+        notification_router: "NotificationRouter" = None,
+    ):
         self._sub_manager = subscription_manager
+        self._notification_router = notification_router
 
     async def execute(self, args: str, context: CommandContext) -> str:
         """List user's subscriptions."""
-        if not self._sub_manager:
-            return "Subscriptions not available."
-
         user_id = self._get_user_id(context)
-        subs = self._sub_manager.get_user_subs(user_id)
+        lines = []
 
-        if not subs:
+        # Check notification router subscriptions
+        if self._notification_router:
+            categories = self._notification_router.get_node_subscriptions(user_id)
+            if categories:
+                if categories == ["all"]:
+                    lines.append("Alert subscriptions: all categories")
+                else:
+                    lines.append(f"Alert subscriptions: {', '.join(categories)}")
+
+        # Check legacy subscriptions
+        if self._sub_manager:
+            subs = self._sub_manager.get_user_subs(user_id)
+            if subs:
+                if not lines:
+                    lines.append("Your subscriptions:")
+                else:
+                    lines.append("\nScheduled reports:")
+                for i, sub in enumerate(subs, 1):
+                    lines.append(f"  {i}. {self._format_sub(sub)}")
+
+        if not lines:
             return "No active subscriptions. Use !sub to subscribe."
 
-        lines = ["Your subscriptions:"]
-        for i, sub in enumerate(subs, 1):
-            lines.append(f"  {i}. {self._format_sub(sub)}")
-
         return "\n".join(lines)
 
     def _format_sub(self, sub: dict) -> str:
@@ -301,7 +360,7 @@ class MySubsCommand(CommandHandler):
             time_str = self._format_time(sub.get("schedule_time", "0000"))
             day_str = (sub.get("schedule_day") or "").capitalize()
             return f"Weekly {scope_desc}report at {time_str} {day_str}"
-        else:  # alerts
+        else:
             return f"Alerts for {scope_desc.strip() or 'mesh'}"
 
     def _format_time(self, hhmm: str) -> str:

meshai/dashboard/api/notification_routes.py

@@ -0,0 +1,113 @@
+"""Notification API routes."""
+
+from fastapi import APIRouter, Request, HTTPException
+from pydantic import BaseModel
+from typing import Optional
+
+router = APIRouter(prefix="/notifications", tags=["notifications"])
+
+
+class ChannelCreate(BaseModel):
+    """Channel creation request."""
+    id: str
+    type: str
+    enabled: bool = True
+    channel_index: int = 0
+    node_ids: list[str] = []
+    smtp_host: str = ""
+    smtp_port: int = 587
+    smtp_user: str = ""
+    smtp_password: str = ""
+    smtp_tls: bool = True
+    from_address: str = ""
+    recipients: list[str] = []
+    url: str = ""
+    headers: dict = {}
+
+
+class RuleCreate(BaseModel):
+    """Rule creation request."""
+    name: str
+    categories: list[str] = []
+    min_severity: str = "warning"
+    channel_ids: list[str] = []
+    override_quiet: bool = False
+
+
+class QuietHoursUpdate(BaseModel):
+    """Quiet hours update request."""
+    start: str
+    end: str
+
+
+@router.get("/categories")
+async def get_categories():
+    """Get all alert categories with descriptions."""
+    try:
+        from ...notifications.categories import list_categories
+        return list_categories()
+    except ImportError:
+        return []
+
+
+@router.get("/channels")
+async def get_channels(request: Request):
+    """Get configured notification channels."""
+    notification_router = getattr(request.app.state, "notification_router", None)
+    if not notification_router:
+        return []
+    return notification_router.get_channels()
+
+
+@router.post("/channels")
+async def create_channel(request: Request, channel: ChannelCreate):
+    """Create a new notification channel."""
+    # This would require runtime config modification
+    # For now, return not implemented
+    raise HTTPException(status_code=501, detail="Channel creation requires config file edit")
+
+
+@router.post("/channels/{channel_id}/test")
+async def test_channel(request: Request, channel_id: str):
+    """Send a test alert to a channel."""
+    notification_router = getattr(request.app.state, "notification_router", None)
+    if not notification_router:
+        raise HTTPException(status_code=404, detail="Notification router not configured")
+
+    success, message = await notification_router.test_channel(channel_id)
+    return {"success": success, "message": message}
+
+
+@router.get("/rules")
+async def get_rules(request: Request):
+    """Get configured notification rules."""
+    notification_router = getattr(request.app.state, "notification_router", None)
+    if not notification_router:
+        return []
+    return notification_router.get_rules()
+
+
+@router.post("/rules")
+async def create_rule(request: Request, rule: RuleCreate):
+    """Create a new notification rule."""
+    # This would require runtime config modification
+    raise HTTPException(status_code=501, detail="Rule creation requires config file edit")
+
+
+@router.get("/quiet-hours")
+async def get_quiet_hours(request: Request):
+    """Get quiet hours configuration."""
+    config = getattr(request.app.state, "config", None)
+    if not config or not hasattr(config, "notifications"):
+        return {"start": "22:00", "end": "06:00"}
+    return {
+        "start": config.notifications.quiet_hours_start,
+        "end": config.notifications.quiet_hours_end,
+    }
+
+
+@router.put("/quiet-hours")
+async def update_quiet_hours(request: Request, quiet_hours: QuietHoursUpdate):
+    """Update quiet hours configuration."""
+    # This would require runtime config modification
+    raise HTTPException(status_code=501, detail="Quiet hours update requires config file edit")

meshai/notifications/__init__.py

@@ -0,0 +1,6 @@
+"""Notification system for MeshAI alerts."""
+
+from .categories import ALERT_CATEGORIES, get_category, list_categories
+from .router import NotificationRouter
+
+__all__ = ["ALERT_CATEGORIES", "get_category", "list_categories", "NotificationRouter"]

meshai/notifications/categories.py

@@ -0,0 +1,157 @@
+"""Alert category registry.
+
+Defines all alertable conditions with human-readable names and descriptions.
+"""
+
+ALERT_CATEGORIES = {
+    # Infrastructure alerts
+    "infra_offline": {
+        "name": "Infrastructure Offline",
+        "description": "An infrastructure node stopped responding",
+        "default_severity": "warning",
+    },
+    "critical_node_down": {
+        "name": "Critical Node Down",
+        "description": "A node marked as critical went offline",
+        "default_severity": "critical",
+    },
+    "infra_recovery": {
+        "name": "Infrastructure Recovery",
+        "description": "An infrastructure node came back online",
+        "default_severity": "info",
+    },
+    "new_router": {
+        "name": "New Router",
+        "description": "A new router appeared on the mesh",
+        "default_severity": "info",
+    },
+
+    # Power alerts
+    "battery_warning": {
+        "name": "Battery Warning",
+        "description": "Infrastructure node battery below warning threshold",
+        "default_severity": "warning",
+    },
+    "battery_critical": {
+        "name": "Battery Critical",
+        "description": "Infrastructure node battery below critical threshold",
+        "default_severity": "critical",
+    },
+    "battery_emergency": {
+        "name": "Battery Emergency",
+        "description": "Infrastructure node battery critically low",
+        "default_severity": "emergency",
+    },
+    "battery_trend": {
+        "name": "Battery Declining",
+        "description": "Battery showing declining trend over 7 days",
+        "default_severity": "warning",
+    },
+    "power_source_change": {
+        "name": "Power Source Change",
+        "description": "Node switched from USB to battery (possible outage)",
+        "default_severity": "warning",
+    },
+    "solar_not_charging": {
+        "name": "Solar Not Charging",
+        "description": "Solar panel not charging during daylight hours",
+        "default_severity": "warning",
+    },
+
+    # Utilization alerts
+    "sustained_high_util": {
+        "name": "High Utilization",
+        "description": "Channel utilization elevated for extended period",
+        "default_severity": "warning",
+    },
+    "packet_flood": {
+        "name": "Packet Flood",
+        "description": "Node sending excessive packets",
+        "default_severity": "warning",
+    },
+
+    # Coverage alerts
+    "infra_single_gateway": {
+        "name": "Single Gateway",
+        "description": "Infrastructure node dropped to single gateway coverage",
+        "default_severity": "warning",
+    },
+    "feeder_offline": {
+        "name": "Feeder Offline",
+        "description": "A feeder gateway stopped responding",
+        "default_severity": "warning",
+    },
+    "region_total_blackout": {
+        "name": "Region Blackout",
+        "description": "All infrastructure in a region is offline",
+        "default_severity": "emergency",
+    },
+
+    # Health score alerts
+    "mesh_score_low": {
+        "name": "Mesh Health Low",
+        "description": "Overall mesh health score below threshold",
+        "default_severity": "warning",
+    },
+    "region_score_low": {
+        "name": "Region Health Low",
+        "description": "A region's health score below threshold",
+        "default_severity": "warning",
+    },
+
+    # Environmental alerts
+    "weather_warning": {
+        "name": "Severe Weather",
+        "description": "NWS warning or advisory for mesh area",
+        "default_severity": "warning",
+    },
+    "hf_blackout": {
+        "name": "HF Radio Blackout",
+        "description": "R3+ solar event degrading HF propagation",
+        "default_severity": "warning",
+    },
+    "tropospheric_ducting": {
+        "name": "Tropospheric Ducting",
+        "description": "Atmospheric conditions extending VHF/UHF range",
+        "default_severity": "info",
+    },
+    "wildfire_proximity": {
+        "name": "Fire Near Mesh",
+        "description": "Wildfire detected within configured distance",
+        "default_severity": "warning",
+    },
+    "new_ignition": {
+        "name": "New Fire Ignition",
+        "description": "Satellite hotspot not matching any known fire",
+        "default_severity": "warning",
+    },
+    "flood_warning": {
+        "name": "Flood Warning",
+        "description": "Stream gauge exceeds flood threshold",
+        "default_severity": "warning",
+    },
+    "road_closure": {
+        "name": "Road Closure",
+        "description": "Full road closure on monitored corridor",
+        "default_severity": "warning",
+    },
+}
+
+
+def get_category(category_id: str) -> dict:
+    """Get category info by ID, with fallback for unknown categories."""
+    if category_id in ALERT_CATEGORIES:
+        return ALERT_CATEGORIES[category_id]
+    return {
+        "name": category_id.replace("_", " ").title(),
+        "description": f"Alert type: {category_id}",
+        "default_severity": "info",
+    }
+
+
+def list_categories() -> list[dict]:
+    """List all categories with their IDs."""
+    return [
+        {"id": cat_id, **cat_info}
+        for cat_id, cat_info in ALERT_CATEGORIES.items()
+    ]

meshai/notifications/channels.py

@@ -0,0 +1,308 @@
+"""Notification channel implementations."""
+
+import asyncio
+import logging
+import smtplib
+import ssl
+import time
+from abc import ABC, abstractmethod
+from email.mime.text import MIMEText
+from email.mime.multipart import MIMEMultipart
+from typing import Optional, TYPE_CHECKING
+
+import httpx
+
+if TYPE_CHECKING:
+    from ..connector import MeshConnector
+
+logger = logging.getLogger(__name__)
+
+
+class NotificationChannel(ABC):
+    """Base class for notification delivery channels."""
+
+    channel_type: str = "base"
+
+    @abstractmethod
+    async def deliver(self, alert: dict, rule: dict) -> bool:
+        """Send alert. Returns True on success."""
+        raise NotImplementedError
+
+    @abstractmethod
+    async def test(self) -> tuple[bool, str]:
+        """Send test message. Returns (success, message)."""
+        raise NotImplementedError
+
+
+class MeshBroadcastChannel(NotificationChannel):
+    """Post alert to mesh channel."""
+
+    channel_type = "mesh_broadcast"
+
+    def __init__(self, connector: "MeshConnector", channel_index: int = 0):
+        self._connector = connector
+        self._channel = channel_index
+
+    async def deliver(self, alert: dict, rule: dict) -> bool:
+        """Send alert to mesh channel."""
+        if not self._connector:
+            logger.warning("No mesh connector available")
+            return False
+
+        try:
+            message = alert.get("message", "")
+            self._connector.send_message(
+                text=message,
+                destination=None,
+                channel=self._channel,
+            )
+            logger.info("Broadcast alert to channel %d", self._channel)
+            return True
+        except Exception as e:
+            logger.error("Failed to broadcast alert: %s", e)
+            return False
+
+    async def test(self) -> tuple[bool, str]:
+        """Send test broadcast."""
+        try:
+            self._connector.send_message(
+                text="[TEST] MeshAI notification system test",
+                destination=None,
+                channel=self._channel,
+            )
+            return True, "Test message sent to channel %d" % self._channel
+        except Exception as e:
+            return False, "Failed to send test: %s" % e
+
+
+class MeshDMChannel(NotificationChannel):
+    """DM alert to specific node IDs."""
+
+    channel_type = "mesh_dm"
+
+    def __init__(self, connector: "MeshConnector", node_ids: list[str]):
+        self._connector = connector
+        self._node_ids = node_ids
+
+    async def deliver(self, alert: dict, rule: dict) -> bool:
+        """Send alert via DM to configured nodes."""
+        if not self._connector:
+            return False
+
+        message = alert.get("message", "")
+        success = True
+
+        for node_id in self._node_ids:
+            try:
+                dest = int(node_id) if node_id.isdigit() else node_id
+                self._connector.send_message(text=message, destination=dest, channel=0)
+            except Exception as e:
+                logger.error("Failed to DM %s: %s", node_id, e)
+                success = False
+
+        return success
+
+    async def test(self) -> tuple[bool, str]:
+        """Send test DM to all configured nodes."""
+        if not self._node_ids:
+            return False, "No node IDs configured"
+        try:
+            for node_id in self._node_ids:
+                dest = int(node_id) if node_id.isdigit() else node_id
+                self._connector.send_message(
+                    text="[TEST] MeshAI notification test",
+                    destination=dest,
+                    channel=0,
+                )
+            return True, "Test DMs sent to %d nodes" % len(self._node_ids)
+        except Exception as e:
+            return False, "Failed to send test DMs: %s" % e
+
+
+class EmailChannel(NotificationChannel):
+    """Send alert via SMTP email."""
+
+    channel_type = "email"
+
+    def __init__(
+        self,
+        smtp_host: str,
+        smtp_port: int,
+        smtp_user: str,
+        smtp_password: str,
+        smtp_tls: bool,
+        from_address: str,
+        recipients: list[str],
+    ):
+        self._host = smtp_host
+        self._port = smtp_port
+        self._user = smtp_user
+        self._password = smtp_password
+        self._tls = smtp_tls
+        self._from = from_address
+        self._recipients = recipients
+
+    async def deliver(self, alert: dict, rule: dict) -> bool:
+        """Send alert via email."""
+        if not self._recipients:
+            return False
+
+        alert_type = alert.get("type", "alert")
+        severity = alert.get("severity", "info").upper()
+        message = alert.get("message", "")
+        subject = "[MeshAI %s] %s" % (severity, alert_type.replace("_", " ").title())
+        body = "MeshAI Alert\n\nType: %s\nSeverity: %s\nTime: %s\n\n%s\n\n---\nAutomated message from MeshAI." % (
+            alert_type, severity, time.strftime("%Y-%m-%d %H:%M:%S"), message
+        )
+
+        try:
+            loop = asyncio.get_event_loop()
+            await loop.run_in_executor(None, self._send_email, subject, body)
+            return True
+        except Exception as e:
+            logger.error("Failed to send email: %s", e)
+            return False
+
+    def _send_email(self, subject: str, body: str):
+        msg = MIMEMultipart()
+        msg["From"] = self._from
+        msg["To"] = ", ".join(self._recipients)
+        msg["Subject"] = subject
+        msg.attach(MIMEText(body, "plain"))
+
+        if self._tls:
+            context = ssl.create_default_context()
+            with smtplib.SMTP(self._host, self._port) as server:
+                server.starttls(context=context)
+                if self._user and self._password:
+                    server.login(self._user, self._password)
+                server.sendmail(self._from, self._recipients, msg.as_string())
+        else:
+            with smtplib.SMTP(self._host, self._port) as server:
+                if self._user and self._password:
+                    server.login(self._user, self._password)
+                server.sendmail(self._from, self._recipients, msg.as_string())
+
+    async def test(self) -> tuple[bool, str]:
+        try:
+            loop = asyncio.get_event_loop()
+            await loop.run_in_executor(
+                None,
+                self._send_email,
+                "[MeshAI TEST] Notification Test",
+                "Test message from MeshAI.",
+            )
+            return True, "Test email sent to %d recipients" % len(self._recipients)
+        except Exception as e:
+            return False, "Failed to send test email: %s" % e
+
+
+class WebhookChannel(NotificationChannel):
+    """POST alert JSON to a URL."""
+
+    channel_type = "webhook"
+
+    def __init__(self, url: str, headers: Optional[dict] = None):
+        self._url = url
+        self._headers = headers or {}
+
+    async def deliver(self, alert: dict, rule: dict) -> bool:
+        """POST alert to webhook URL."""
+        payload = {
+            "type": alert.get("type"),
+            "severity": alert.get("severity", "info"),
+            "message": alert.get("message", ""),
+            "timestamp": time.time(),
+            "node_name": alert.get("node_name"),
+            "region": alert.get("region"),
+        }
+
+        # Discord/Slack format
+        if "discord.com" in self._url or "slack.com" in self._url:
+            severity = alert.get("severity", "info")
+            color = {
+                "emergency": 0xFF0000,
+                "critical": 0xFF4444,
+                "warning": 0xFFAA00,
+                "info": 0x0099FF,
+            }.get(severity, 0x888888)
+            payload = {
+                "embeds": [{
+                    "title": "MeshAI: %s" % alert.get("type", "unknown"),
+                    "description": alert.get("message", ""),
+                    "color": color,
+                }]
+            }
+
+        # ntfy format
+        elif "ntfy" in self._url:
+            headers = {
+                **self._headers,
+                "Title": "MeshAI: %s" % alert.get("type", "alert"),
+                "Priority": "3",
+            }
+            try:
+                async with httpx.AsyncClient() as client:
+                    resp = await client.post(
+                        self._url,
+                        content=alert.get("message", ""),
+                        headers=headers,
+                        timeout=10,
+                    )
+                    return resp.status_code < 400
+            except Exception as e:
+                logger.error("Webhook failed: %s", e)
+                return False
+
+        try:
+            async with httpx.AsyncClient() as client:
+                resp = await client.post(
+                    self._url,
+                    json=payload,
+                    headers={"Content-Type": "application/json", **self._headers},
+                    timeout=10,
+                )
+                return resp.status_code < 400
+        except Exception as e:
+            logger.error("Webhook failed: %s", e)
+            return False
+
+    async def test(self) -> tuple[bool, str]:
+        test_alert = {"type": "test", "severity": "info", "message": "MeshAI test message"}
+        success = await self.deliver(test_alert, {})
+        if success:
+            return True, "Test sent to %s" % self._url
+        return False, "Webhook failed"
+
+
+def create_channel(config: dict, connector=None) -> NotificationChannel:
+    """Create a channel instance from config."""
+    channel_type = config.get("type", "")
+
+    if channel_type == "mesh_broadcast":
+        return MeshBroadcastChannel(
+            connector=connector,
+            channel_index=config.get("channel_index", 0),
+        )
+    elif channel_type == "mesh_dm":
+        return MeshDMChannel(
+            connector=connector,
+            node_ids=config.get("node_ids", []),
+        )
+    elif channel_type == "email":
+        return EmailChannel(
+            smtp_host=config.get("smtp_host", ""),
+            smtp_port=config.get("smtp_port", 587),
+            smtp_user=config.get("smtp_user", ""),
+            smtp_password=config.get("smtp_password", ""),
+            smtp_tls=config.get("smtp_tls", True),
+            from_address=config.get("from_address", ""),
+            recipients=config.get("recipients", []),
+        )
+    elif channel_type == "webhook":
+        return WebhookChannel(
+            url=config.get("url", ""),
+            headers=config.get("headers", {}),
+        )
+    else:
+        raise ValueError("Unknown channel type: %s" % channel_type)

meshai/notifications/router.py

@@ -0,0 +1,266 @@
+"""Notification router - matches alerts to rules and delivers via channels."""
+
+import logging
+import time
+from datetime import datetime
+from typing import Optional, TYPE_CHECKING
+
+from .channels import create_channel, NotificationChannel
+from .summarizer import MessageSummarizer
+
+if TYPE_CHECKING:
+    from ..connector import MeshConnector
+
+logger = logging.getLogger(__name__)
+
+# Severity levels in order
+SEVERITY_ORDER = ["info", "advisory", "watch", "warning", "critical", "emergency"]
+
+
+class NotificationRouter:
+    """Routes alerts through matching rules to notification channels."""
+
+    def __init__(
+        self,
+        config,
+        connector: Optional["MeshConnector"] = None,
+        llm_backend=None,
+        timezone: str = "America/Boise",
+    ):
+        self._channels: dict[str, NotificationChannel] = {}
+        self._rules: list[dict] = []
+        self._quiet_start = getattr(config, "quiet_hours_start", "22:00")
+        self._quiet_end = getattr(config, "quiet_hours_end", "06:00")
+        self._timezone = timezone
+        self._dedup_window = getattr(config, "dedup_seconds", 600)
+        self._recent: dict[tuple, float] = {}  # (category, event_key) -> last_sent_time
+        self._summarizer = MessageSummarizer(llm_backend) if llm_backend else None
+        self._connector = connector
+
+        # Create channel instances from config
+        channels_config = getattr(config, "channels", [])
+        for ch_config in channels_config:
+            if hasattr(ch_config, "__dict__"):
+                ch_dict = {k: v for k, v in ch_config.__dict__.items() if not k.startswith("_")}
+            else:
+                ch_dict = ch_config
+
+            if not ch_dict.get("enabled", True):
+                continue
+
+            channel_id = ch_dict.get("id", "")
+            if not channel_id:
+                continue
+
+            try:
+                channel = create_channel(ch_dict, connector)
+                self._channels[channel_id] = channel
+                logger.debug("Created notification channel: %s (%s)", channel_id, ch_dict.get("type"))
+            except Exception as e:
+                logger.warning("Failed to create channel %s: %s", channel_id, e)
+
+        # Load rules
+        rules_config = getattr(config, "rules", [])
+        for rule in rules_config:
+            if hasattr(rule, "__dict__"):
+                rule_dict = {k: v for k, v in rule.__dict__.items() if not k.startswith("_")}
+            else:
+                rule_dict = rule
+            self._rules.append(rule_dict)
+
+        logger.info(
+            "Notification router initialized: %d channels, %d rules",
+            len(self._channels),
+            len(self._rules),
+        )
+
+    async def process_alert(self, alert: dict) -> bool:
+        """Route an alert through matching rules to channels.
+
+        Returns True if alert was delivered to at least one channel.
+        """
+        category = alert.get("type", "")
+        severity = alert.get("severity", "info")
+        delivered = False
+
+        for rule in self._rules:
+            # Check category match
+            rule_categories = rule.get("categories", [])
+            if rule_categories and category not in rule_categories:
+                continue
+
+            # Check severity threshold
+            min_severity = rule.get("min_severity", "info")
+            if not self._severity_meets(severity, min_severity):
+                continue
+
+            # Check quiet hours (emergencies and criticals override)
+            if self._in_quiet_hours() and severity not in ("emergency", "critical"):
+                if not rule.get("override_quiet", False):
+                    continue
+
+            # Check dedup
+            event_id = alert.get("event_id", alert.get("message", "")[:50])
+            dedup_key = (category, event_id)
+            now = time.time()
+            if dedup_key in self._recent:
+                if now - self._recent[dedup_key] < self._dedup_window:
+                    logger.debug("Skipping duplicate alert: %s", category)
+                    continue
+            self._recent[dedup_key] = now
+
+            # Deliver to each channel in the rule
+            channel_ids = rule.get("channel_ids", [])
+            for channel_id in channel_ids:
+                channel = self._channels.get(channel_id)
+                if not channel:
+                    continue
+
+                try:
+                    # Summarize for mesh channels if over 200 chars
+                    delivery_alert = alert
+                    message = alert.get("message", "")
+                    if channel.channel_type in ("mesh_broadcast", "mesh_dm"):
+                        if len(message) > 200:
+                            if self._summarizer:
+                                summary = await self._summarizer.summarize(message, max_chars=195)
+                                delivery_alert = {**alert, "message": summary}
+                            else:
+                                delivery_alert = {**alert, "message": message[:195] + "..."}
+
+                    success = await channel.deliver(delivery_alert, rule)
+                    if success:
+                        delivered = True
+                        logger.info(
+                            "Alert delivered via %s: %s",
+                            channel_id,
+                            category,
+                        )
+                except Exception as e:
+                    logger.warning("Channel %s delivery failed: %s", channel_id, e)
+
+        return delivered
+
+    def _severity_meets(self, actual: str, required: str) -> bool:
+        """Check if actual severity meets or exceeds required severity."""
+        try:
+            actual_idx = SEVERITY_ORDER.index(actual.lower())
+            required_idx = SEVERITY_ORDER.index(required.lower())
+            return actual_idx >= required_idx
+        except ValueError:
+            return True  # Unknown severity, allow through
+
+    def _in_quiet_hours(self) -> bool:
+        """Check if current time is within quiet hours."""
+        try:
+            from zoneinfo import ZoneInfo
+            tz = ZoneInfo(self._timezone)
+            now = datetime.now(tz)
+            current_time = now.strftime("%H:%M")
+
+            start = self._quiet_start
+            end = self._quiet_end
+
+            if start <= end:
+                # Simple range (e.g., 01:00 to 06:00)
+                return start <= current_time <= end
+            else:
+                # Crosses midnight (e.g., 22:00 to 06:00)
+                return current_time >= start or current_time <= end
+        except Exception:
+            return False
+
+    def get_channels(self) -> list[dict]:
+        """Get list of configured channels."""
+        return [
+            {"id": ch_id, "type": ch.channel_type}
+            for ch_id, ch in self._channels.items()
+        ]
+
+    def get_rules(self) -> list[dict]:
+        """Get list of configured rules."""
+        return self._rules
+
+    async def test_channel(self, channel_id: str) -> tuple[bool, str]:
+        """Send a test alert to a specific channel."""
+        channel = self._channels.get(channel_id)
+        if not channel:
+            return False, "Channel not found: %s" % channel_id
+        return await channel.test()
+
+    def add_mesh_subscription(
+        self,
+        node_id: str,
+        categories: list[str],
+        rule_name: Optional[str] = None,
+    ) -> str:
+        """Add a mesh DM subscription for a node.
+
+        Creates a channel and rule for the node to receive alerts.
+        Returns the rule name.
+        """
+        # Create channel ID
+        channel_id = "mesh_dm_%s" % node_id
+
+        # Create channel if it doesn't exist
+        if channel_id not in self._channels:
+            from .channels import MeshDMChannel
+            channel = MeshDMChannel(
+                connector=self._connector,
+                node_ids=[node_id],
+            )
+            self._channels[channel_id] = channel
+
+        # Create rule
+        if not rule_name:
+            rule_name = "sub_%s" % node_id
+
+        # Check if rule already exists
+        for rule in self._rules:
+            if rule.get("name") == rule_name:
+                # Update existing rule
+                rule["categories"] = categories if categories else []
+                rule["channel_ids"] = [channel_id]
+                return rule_name
+
+        # Add new rule
+        self._rules.append({
+            "name": rule_name,
+            "categories": categories if categories else [],  # Empty = all
+            "min_severity": "warning",
+            "channel_ids": [channel_id],
+            "override_quiet": False,
+        })
+
+        return rule_name
+
+    def remove_mesh_subscription(self, node_id: str) -> bool:
+        """Remove a mesh subscription for a node."""
+        channel_id = "mesh_dm_%s" % node_id
+        rule_name = "sub_%s" % node_id
+
+        # Remove channel
+        if channel_id in self._channels:
+            del self._channels[channel_id]
+
+        # Remove rule
+        self._rules = [r for r in self._rules if r.get("name") != rule_name]
+
+        return True
+
+    def get_node_subscriptions(self, node_id: str) -> list[str]:
+        """Get categories a node is subscribed to."""
+        rule_name = "sub_%s" % node_id
+        for rule in self._rules:
+            if rule.get("name") == rule_name:
+                categories = rule.get("categories", [])
+                return categories if categories else ["all"]
+        return []
+
+    def cleanup_recent(self, max_age: int = 3600):
+        """Clean up old entries from recent alerts cache."""
+        now = time.time()
+        self._recent = {
+            k: v for k, v in self._recent.items()
+            if now - v < max_age
+        }

meshai/notifications/summarizer.py

@@ -0,0 +1,64 @@
+"""Message summarizer for mesh delivery."""
+
+import logging
+from typing import Optional, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from ..backends import LLMBackend
+
+logger = logging.getLogger(__name__)
+
+
+class MessageSummarizer:
+    """Summarizes long messages for mesh delivery.
+
+    Only used when:
+    - Delivering to mesh channels (broadcast or DM)
+    - Message exceeds max_chars (default 200)
+    - LLM backend is available
+
+    Email and webhook channels receive full messages.
+    """
+
+    def __init__(self, llm_backend: Optional["LLMBackend"] = None):
+        self._llm = llm_backend
+
+    async def summarize(self, message: str, max_chars: int = 195) -> str:
+        """Summarize a message to fit within max_chars.
+
+        Args:
+            message: Original message text
+            max_chars: Maximum characters for summary
+
+        Returns:
+            Summarized message, or truncated original if LLM unavailable
+        """
+        if len(message) <= max_chars:
+            return message
+
+        if not self._llm:
+            return message[:max_chars - 3] + "..."
+
+        prompt = (
+            "Summarize this alert in under %d characters. "
+            "Keep severity, location, and key facts. No preamble, just the summary:\n\n%s"
+            % (max_chars, message)
+        )
+
+        try:
+            # Use the LLM to generate a summary
+            response = await self._llm.generate(
+                prompt,
+                system_prompt="You are a concise alert summarizer. Output only the summary, no explanation.",
+                max_tokens=100,
+            )
+            summary = response.strip()
+
+            # Ensure it fits
+            if len(summary) <= max_chars:
+                return summary
+            return summary[:max_chars - 3] + "..."
+
+        except Exception as e:
+            logger.debug("LLM summarization failed: %s", e)
+            return message[:max_chars - 3] + "..."

meshai/sources/mqtt_source.py

@@ -0,0 +1,435 @@
+"""MQTT source adapter for Meshtastic broker subscriptions.
+
+Push-based source that subscribes to MQTT topics and decodes
+ServiceEnvelope-wrapped MeshPackets. Provides live node/packet
+data without polling.
+"""
+
+import asyncio
+import logging
+import os
+import time
+from dataclasses import dataclass, field
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+# Port number to name mapping (from portnums_pb2)
+PORTNUM_NAMES = {
+    0: "UNKNOWN_APP",
+    1: "TEXT_MESSAGE_APP",
+    2: "REMOTE_HARDWARE_APP",
+    3: "POSITION_APP",
+    4: "NODEINFO_APP",
+    5: "ROUTING_APP",
+    6: "ADMIN_APP",
+    7: "TEXT_MESSAGE_COMPRESSED_APP",
+    8: "WAYPOINT_APP",
+    9: "AUDIO_APP",
+    10: "DETECTION_SENSOR_APP",
+    11: "ALERT_APP",
+    32: "REPLY_APP",
+    33: "IP_TUNNEL_APP",
+    34: "PAXCOUNTER_APP",
+    64: "SERIAL_APP",
+    65: "STORE_FORWARD_APP",
+    66: "RANGE_TEST_APP",
+    67: "TELEMETRY_APP",
+    68: "ZPS_APP",
+    69: "SIMULATOR_APP",
+    70: "TRACEROUTE_APP",
+    71: "NEIGHBORINFO_APP",
+    72: "ATAK_PLUGIN",
+    73: "MAP_REPORT_APP",
+    74: "POWERSTRESS_APP",
+    256: "PRIVATE_APP",
+    257: "ATAK_FORWARDER",
+}
+
+
+@dataclass
+class MQTTNodeInfo:
+    """Cached node info from MQTT."""
+
+    node_num: int
+    node_id_hex: str = ""
+    short_name: str = ""
+    long_name: str = ""
+    hw_model: str = ""
+    role: int = 0
+    latitude: Optional[float] = None
+    longitude: Optional[float] = None
+    altitude: Optional[float] = None
+    last_heard: float = 0.0
+    battery_percent: Optional[float] = None
+    voltage: Optional[float] = None
+    channel_utilization: Optional[float] = None
+    air_util_tx: Optional[float] = None
+    snr: Optional[float] = None
+    rssi: Optional[int] = None
+    via_mqtt: bool = True
+
+
+@dataclass
+class MQTTPacketInfo:
+    """Packet received from MQTT."""
+
+    packet_id: int
+    from_node: int
+    to_node: int
+    portnum: int
+    portnum_name: str
+    channel: int
+    timestamp: float
+    snr: Optional[float] = None
+    rssi: Optional[int] = None
+    hop_limit: Optional[int] = None
+    hop_start: Optional[int] = None
+    payload_size: int = 0
+    gateway_id: str = ""
+
+
+class MQTTSource:
+    """MQTT source adapter subscribing to Meshtastic broker topics.
+
+    Maintains a subscription loop that processes ServiceEnvelope messages
+    and updates node/packet caches. Unlike poll-based sources, this is
+    push-based and receives data as it arrives.
+    """
+
+    def __init__(
+        self,
+        host: str,
+        port: int = 1883,
+        username: str = "",
+        password: str = "",
+        topic_root: str = "msh/US",
+        use_tls: bool = False,
+        name: str = "mqtt",
+    ):
+        """Initialize MQTT source.
+
+        Args:
+            host: MQTT broker hostname
+            port: MQTT broker port (1883 for plain, 8883 for TLS)
+            username: MQTT username (optional)
+            password: MQTT password (optional, supports ${ENV_VAR})
+            topic_root: Topic root to subscribe to (default: msh/US)
+            use_tls: Enable TLS for connection
+            name: Source name for logging/attribution
+        """
+        self._host = host
+        self._port = port
+        self._username = username
+        self._password = self._resolve_env(password)
+        self._topic_root = topic_root.rstrip("/")
+        self._use_tls = use_tls
+        self._name = name
+
+        # State
+        self._nodes: dict[int, MQTTNodeInfo] = {}
+        self._packets: list[MQTTPacketInfo] = []
+        self._max_packets = 1000  # Ring buffer
+        self._is_connected: bool = False
+        self._is_loaded: bool = False
+        self._last_message: float = 0.0
+        self._last_error: str = ""
+        self._message_count: int = 0
+        self._data_changed: bool = False
+
+        # Subscription task
+        self._task: Optional[asyncio.Task] = None
+        self._stop_event: Optional[asyncio.Event] = None
+
+        # Retry settings
+        self._retry_delay = 5  # Initial retry delay
+        self._max_retry_delay = 300  # Max 5 minutes between retries
+
+    def _resolve_env(self, value: str) -> str:
+        """Resolve ${ENV_VAR} references in value."""
+        if value and value.startswith("${") and value.endswith("}"):
+            env_var = value[2:-1]
+            return os.environ.get(env_var, "")
+        return value
+
+    @property
+    def nodes(self) -> dict[int, MQTTNodeInfo]:
+        """Return cached nodes."""
+        return self._nodes
+
+    @property
+    def packets(self) -> list[dict]:
+        """Return packets as dicts for compatibility."""
+        return [
+            {
+                "packet_id": p.packet_id,
+                "from_node": p.from_node,
+                "to_node": p.to_node,
+                "portnum": p.portnum,
+                "portnum_name": p.portnum_name,
+                "channel": p.channel,
+                "timestamp": p.timestamp,
+                "snr": p.snr,
+                "rssi": p.rssi,
+                "hop_limit": p.hop_limit,
+                "hop_start": p.hop_start,
+                "payload_size": p.payload_size,
+                "gateway_id": p.gateway_id,
+            }
+            for p in self._packets
+        ]
+
+    @property
+    def is_loaded(self) -> bool:
+        """Return True if we have received any data."""
+        return self._is_loaded
+
+    @property
+    def data_changed(self) -> bool:
+        """Return True if data changed since last check, then reset."""
+        changed = self._data_changed
+        self._data_changed = False
+        return changed
+
+    @property
+    def health_status(self) -> dict:
+        """Return health status for dashboard."""
+        return {
+            "name": self._name,
+            "type": "mqtt",
+            "host": self._host,
+            "port": self._port,
+            "topic_root": self._topic_root,
+            "is_connected": self._is_connected,
+            "is_loaded": self._is_loaded,
+            "last_message": self._last_message,
+            "last_error": self._last_error,
+            "message_count": self._message_count,
+            "node_count": len(self._nodes),
+            "packet_count": len(self._packets),
+        }
+
+    async def start(self) -> None:
+        """Start the subscription loop."""
+        if self._task is not None:
+            logger.warning(f"MQTT source '{self._name}' already started")
+            return
+
+        self._stop_event = asyncio.Event()
+        self._task = asyncio.create_task(self._subscription_loop())
+        logger.info(f"Started MQTT source '{self._name}' -> {self._host}:{self._port}")
+
+    async def stop(self) -> None:
+        """Stop the subscription loop."""
+        if self._stop_event:
+            self._stop_event.set()
+        if self._task:
+            self._task.cancel()
+            try:
+                await self._task
+            except asyncio.CancelledError:
+                pass
+            self._task = None
+        self._is_connected = False
+        logger.info(f"Stopped MQTT source '{self._name}'")
+
+    async def _subscription_loop(self) -> None:
+        """Main subscription loop with reconnection logic."""
+        try:
+            import aiomqtt
+        except ImportError:
+            logger.error("aiomqtt not installed. Run: pip install aiomqtt")
+            self._last_error = "aiomqtt not installed"
+            return
+
+        retry_delay = self._retry_delay
+
+        while not self._stop_event.is_set():
+            try:
+                # Build connection kwargs
+                kwargs = {
+                    "hostname": self._host,
+                    "port": self._port,
+                }
+                if self._username:
+                    kwargs["username"] = self._username
+                if self._password:
+                    kwargs["password"] = self._password
+
+                # TLS setup
+                if self._use_tls:
+                    import ssl
+                    tls_context = ssl.create_default_context()
+                    kwargs["tls_context"] = tls_context
+
+                async with aiomqtt.Client(**kwargs) as client:
+                    self._is_connected = True
+                    self._last_error = ""
+                    retry_delay = self._retry_delay  # Reset on successful connect
+                    logger.info(f"MQTT '{self._name}' connected to {self._host}:{self._port}")
+
+                    # Subscribe to all topics under root
+                    # Meshtastic uses: msh/{region}/{channel}/json/{node_id}
+                    # and: msh/{region}/{channel}/!{node_id}
+                    topic = f"{self._topic_root}/#"
+                    await client.subscribe(topic)
+                    logger.info(f"MQTT '{self._name}' subscribed to {topic}")
+
+                    async for message in client.messages:
+                        if self._stop_event.is_set():
+                            break
+                        await self._process_message(message)
+
+            except asyncio.CancelledError:
+                break
+            except Exception as e:
+                self._is_connected = False
+                self._last_error = str(e)
+                logger.warning(f"MQTT '{self._name}' error: {e}. Retrying in {retry_delay}s")
+
+                # Exponential backoff
+                await asyncio.sleep(retry_delay)
+                retry_delay = min(retry_delay * 2, self._max_retry_delay)
+
+    async def _process_message(self, message) -> None:
+        """Process an incoming MQTT message."""
+        try:
+            topic = str(message.topic)
+            payload = message.payload
+
+            # Skip JSON topics (we want binary ServiceEnvelope)
+            if "/json/" in topic:
+                return
+
+            # Skip map reports (stat/ or map/ topics)
+            if "/stat/" in topic or "/map/" in topic:
+                return
+
+            # Parse ServiceEnvelope
+            from meshtastic.protobuf import mqtt_pb2
+
+            envelope = mqtt_pb2.ServiceEnvelope()
+            envelope.ParseFromString(payload)
+
+            if not envelope.packet:
+                return
+
+            packet = envelope.packet
+            gateway_id = envelope.gateway_id or ""
+            channel_id = envelope.channel_id or ""
+
+            # Update stats
+            self._last_message = time.time()
+            self._message_count += 1
+            self._is_loaded = True
+            self._data_changed = True
+
+            # Extract packet info
+            pkt_info = MQTTPacketInfo(
+                packet_id=packet.id,
+                from_node=packet.from_,
+                to_node=packet.to,
+                portnum=packet.decoded.portnum if packet.HasField("decoded") else 0,
+                portnum_name=PORTNUM_NAMES.get(
+                    packet.decoded.portnum if packet.HasField("decoded") else 0,
+                    "UNKNOWN"
+                ),
+                channel=packet.channel,
+                timestamp=time.time(),
+                snr=packet.rx_snr if packet.rx_snr else None,
+                rssi=packet.rx_rssi if packet.rx_rssi else None,
+                hop_limit=packet.hop_limit if packet.hop_limit else None,
+                hop_start=packet.hop_start if packet.hop_start else None,
+                payload_size=len(packet.decoded.payload) if packet.HasField("decoded") else 0,
+                gateway_id=gateway_id,
+            )
+
+            # Add to packet ring buffer
+            self._packets.append(pkt_info)
+            if len(self._packets) > self._max_packets:
+                self._packets = self._packets[-self._max_packets:]
+
+            # Process decoded payload by portnum
+            if packet.HasField("decoded"):
+                await self._process_decoded(packet, gateway_id)
+
+        except Exception as e:
+            logger.debug(f"MQTT message parse error: {e}")
+
+    async def _process_decoded(self, packet, gateway_id: str) -> None:
+        """Process decoded packet payload."""
+        decoded = packet.decoded
+        portnum = decoded.portnum
+        from_node = packet.from_
+
+        # Ensure node exists in cache
+        if from_node not in self._nodes:
+            self._nodes[from_node] = MQTTNodeInfo(
+                node_num=from_node,
+                node_id_hex=f"!{from_node:08x}",
+            )
+
+        node = self._nodes[from_node]
+        node.last_heard = time.time()
+        node.snr = packet.rx_snr if packet.rx_snr else node.snr
+        node.rssi = packet.rx_rssi if packet.rx_rssi else node.rssi
+
+        # NODEINFO_APP (4)
+        if portnum == 4:
+            from meshtastic.protobuf import mesh_pb2
+            user = mesh_pb2.User()
+            try:
+                user.ParseFromString(decoded.payload)
+                node.short_name = user.short_name or node.short_name
+                node.long_name = user.long_name or node.long_name
+                node.hw_model = mesh_pb2.HardwareModel.Name(user.hw_model) if user.hw_model else ""
+                node.role = user.role
+            except Exception:
+                pass
+
+        # POSITION_APP (3)
+        elif portnum == 3:
+            from meshtastic.protobuf import mesh_pb2
+            pos = mesh_pb2.Position()
+            try:
+                pos.ParseFromString(decoded.payload)
+                if pos.latitude_i:
+                    node.latitude = pos.latitude_i * 1e-7
+                if pos.longitude_i:
+                    node.longitude = pos.longitude_i * 1e-7
+                if pos.altitude:
+                    node.altitude = pos.altitude
+            except Exception:
+                pass
+
+        # TELEMETRY_APP (67)
+        elif portnum == 67:
+            from meshtastic.protobuf import telemetry_pb2
+            telem = telemetry_pb2.Telemetry()
+            try:
+                telem.ParseFromString(decoded.payload)
+                if telem.HasField("device_metrics"):
+                    dm = telem.device_metrics
+                    if dm.battery_level and dm.battery_level <= 100:
+                        node.battery_percent = dm.battery_level
+                    if dm.voltage:
+                        node.voltage = dm.voltage
+                    if dm.channel_utilization:
+                        node.channel_utilization = dm.channel_utilization
+                    if dm.air_util_tx:
+                        node.air_util_tx = dm.air_util_tx
+            except Exception:
+                pass
+
+    # Compatibility methods for MeshDataStore integration
+
+    def tick(self) -> Optional[str]:
+        """Tick method for compatibility. MQTT is push-based, not polled.
+
+        Returns None since we do not poll endpoints.
+        """
+        return None
+
+    def maybe_refresh(self) -> bool:
+        """Check if data changed (for legacy compatibility)."""
+        return self.data_changed