chore: capture in-flight fire-spam/drain-pacer + reconnect ground truth from CT108

Preserve the live CT108 working-tree state (the running container is already
built and is unaffected by this commit).

Canonical build tree (work/ — what ships):
  - work/meshai/config.py      : ConnectionConfig watchdog reconnect knobs
  - work/meshai/connector.py   : watchdog link-state, socket-based liveness,
                                 active_probe, in-place reconnect
  - work/meshai/main.py        : connection supervisor (watchdog) task
  - work/Dockerfile            : healthcheck also asserts /tmp/meshai.link=up
  - work/docker-compose.yml    : matching healthcheck change

Root tree (stale duplicate, earlier reconnect iteration — preserved for fidelity):
  - meshai/config.py, meshai/connector.py, meshai/main.py
  - Dockerfile, docker-compose.yml

New (root only, NOT under work/, currently unimported / not in build):
  - meshai/notifications/pipeline/severity_router.py

Excluded: all *.bak / *.bak2 backup artifacts.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Dockerfile

@@ -82,7 +82,7 @@ EXPOSE 7682 8080
 
 # Health check - verify bot process is alive via PID file
 HEALTHCHECK --interval=30s --timeout=10s --start-period=10s --retries=3 \
-    CMD test -f /tmp/meshai.pid && kill -0 $(cat /tmp/meshai.pid) 2>/dev/null || exit 1
+    CMD test -f /tmp/meshai.pid && kill -0 $(cat /tmp/meshai.pid) 2>/dev/null && [ "$(cat /tmp/meshai.link 2>/dev/null)" = up ] || exit 1
 
 # Entrypoint handles config and ttyd
 ENTRYPOINT ["/app/docker-entrypoint.sh"]

docker-compose.yml

@@ -67,7 +67,7 @@ services:
           memory: 64M
 
     healthcheck:
-      test: ["CMD-SHELL", "test -f /tmp/meshai.pid && kill -0 $(cat /tmp/meshai.pid) 2>/dev/null || exit 1"]
+      test: ["CMD-SHELL", "test -f /tmp/meshai.pid && kill -0 $(cat /tmp/meshai.pid) 2>/dev/null && [ \"$$(cat /tmp/meshai.link 2>/dev/null)\" = up ] || exit 1"]
       interval: 30s
       timeout: 10s
       retries: 3

meshai/config.py

@@ -29,6 +29,10 @@ class ConnectionConfig:
     serial_port: str = "/dev/ttyUSB0"
     tcp_host: str = "192.168.1.100"
     tcp_port: int = 4403
+    reconnect: bool = True
+    reconnect_initial_delay: float = 2.0
+    reconnect_max_delay: float = 60.0
+    reconnect_health_interval: float = 30.0
 
 
 @dataclass

meshai/connector.py

@@ -48,6 +48,8 @@ class MeshConnector:
         self._connected = False
         self._loop: Optional[asyncio.AbstractEventLoop] = None
         self._lock = threading.Lock()
+        self._connection_lost_event = None   # asyncio.Event, created in set_message_callback
+        self._link_status_path = "/tmp/meshai.link"
 
     @property
     def connected(self) -> bool:
@@ -59,6 +61,35 @@ class MeshConnector:
         """Get our node's ID."""
         return self._my_node_id
 
+    def _write_link_status(self, up: bool) -> None:
+        try:
+            with open(self._link_status_path, "w") as f:
+                f.write("up" if up else "down")
+        except Exception:
+            pass
+
+    def _signal_reconnect(self) -> None:
+        if self._loop is not None and self._connection_lost_event is not None:
+            try:
+                self._loop.call_soon_threadsafe(self._connection_lost_event.set)
+            except Exception:
+                pass
+
+    def _on_connection_lost(self, interface=None, *args, **kwargs) -> None:
+        if not self._connected:
+            return
+        logger.warning("Meshtastic connection lost — scheduling reconnect")
+        self._connected = False
+        self._write_link_status(False)
+        self._signal_reconnect()
+
+    def reconnect(self) -> bool:
+        """Blocking tear-down + re-establish. Call via run_in_executor. Raises on failure."""
+        with self._lock:
+            self.disconnect()
+            self.connect()
+            return self.connected
+
     def connect(self) -> None:
         """Establish connection to Meshtastic node."""
         logger.info(f"Connecting to Meshtastic node via {self.config.type}...")
@@ -86,8 +117,12 @@ class MeshConnector:
             # Subscribe to messages
             pub.subscribe(self._on_receive, "meshtastic.receive.text")
             pub.subscribe(self._on_node_update, "meshtastic.node.updated")
+            pub.subscribe(self._on_connection_lost, "meshtastic.connection.lost")
 
             self._connected = True
+            self._write_link_status(True)
+            if self._connection_lost_event is not None:
+                self._connection_lost_event.clear()
 
         except Exception as e:
             logger.error(f"Failed to connect: {e}")
@@ -100,6 +135,7 @@ class MeshConnector:
             try:
                 pub.unsubscribe(self._on_receive, "meshtastic.receive.text")
                 pub.unsubscribe(self._on_node_update, "meshtastic.node.updated")
+                pub.unsubscribe(self._on_connection_lost, "meshtastic.connection.lost")
             except Exception:
                 pass
 
@@ -110,6 +146,7 @@ class MeshConnector:
 
             self._interface = None
             self._connected = False
+            self._write_link_status(False)
             logger.info("Disconnected from Meshtastic node")
 
     def set_message_callback(
@@ -123,6 +160,7 @@ class MeshConnector:
         """
         self._message_callback = callback
         self._loop = loop
+        self._connection_lost_event = asyncio.Event()
 
     def _cache_node_info(self) -> None:
         """Cache node names and positions from node database."""
@@ -257,6 +295,9 @@ class MeshConnector:
 
         except Exception as e:
             logger.error(f"Failed to send message: {e}")
+            self._connected = False
+            self._write_link_status(False)
+            self._signal_reconnect()
             return False
 
     def get_node_position(self, node_id: str) -> Optional[tuple[float, float]]:
@@ -355,5 +396,7 @@ class MeshConnector:
 
         except Exception as e:
             logger.error(f"Failed to send message: {e}")
+            self._connected = False
+            self._write_link_status(False)
+            self._signal_reconnect()
             return False
-

meshai/main.py

@@ -56,6 +56,7 @@ class MeshAI:
         self.router: Optional[MessageRouter] = None
         self.responder: Optional[Responder] = None
         self._running = False
+        self._supervisor_task: Optional[asyncio.Task] = None
         self._loop: Optional[asyncio.AbstractEventLoop] = None
         self._last_cleanup: float = 0.0
         self._last_health_compute: float = 0.0
@@ -82,6 +83,12 @@ class MeshAI:
         self._last_cleanup = time.time()
         self._last_health_compute = 0.0
 
+        # Supervised auto-reconnect: long-lived task that re-establishes the
+        # mesh link with exponential backoff whenever it drops (e.g. a
+        # meshmonitor restart briefly killing :4404). Started after
+        # connect() + set_message_callback() so _connection_lost_event exists.
+        self._supervisor_task = asyncio.create_task(self._connection_supervisor())
+
         # Write PID file
         self._write_pid()
 
@@ -211,11 +218,54 @@ class MeshAI:
                     self.context.prune()
                 self._last_cleanup = time.time()
 
+    async def _connection_supervisor(self) -> None:
+        cfg = self.config.connection
+        if not getattr(cfg, "reconnect", True):
+            return
+        connector = self.connector
+        ev = connector._connection_lost_event
+        loop = asyncio.get_event_loop()
+        while self._running:
+            try:
+                await asyncio.wait_for(ev.wait(), timeout=cfg.reconnect_health_interval)
+            except asyncio.TimeoutError:
+                pass  # periodic safety check even if no event fired
+            if not self._running:
+                break
+            if connector.connected:
+                # Reconcile link file: a spurious/late connection.lost (e.g. the
+                # meshtastic lib's own reader firing after our reconnect already
+                # succeeded) may have written link=down even though the socket is
+                # live. Re-assert up so the healthcheck reflects reality.
+                connector._write_link_status(True)
+                ev.clear()
+                continue
+            logger.warning("Mesh link is down — starting reconnect with backoff")
+            delay = cfg.reconnect_initial_delay
+            while self._running and not connector.connected:
+                try:
+                    await loop.run_in_executor(None, connector.reconnect)
+                except Exception as e:
+                    logger.error(f"Reconnect attempt failed: {e}; retrying in {delay:.0f}s")
+                if connector.connected:
+                    logger.info("Mesh link reconnected successfully")
+                    ev.clear()
+                    break
+                await asyncio.sleep(delay)
+                delay = min(delay * 2, cfg.reconnect_max_delay)
+
     async def stop(self) -> None:
         """Stop the bot."""
         logger.info("Stopping MeshAI...")
         self._running = False
 
+        if getattr(self, "_supervisor_task", None) is not None:
+            self._supervisor_task.cancel()
+            try:
+                await self._supervisor_task
+            except (asyncio.CancelledError, Exception):
+                pass
+
         if self._pipeline_scheduler is not None:
             from .notifications.pipeline import stop_pipeline
             await stop_pipeline(self._pipeline_scheduler)

meshai/notifications/pipeline/severity_router.py

@@ -0,0 +1,104 @@
+"""Severity-based event routing.
+
+The severity router subscribes to the bus and forks each event into
+one of two paths based on severity:
+
+- immediate → immediate_handler (dispatcher for live delivery)
+- priority/routine → digest_handler (queue for batched summaries)
+
+Usage:
+    router = SeverityRouter(
+        immediate_handler=dispatcher.dispatch,
+        digest_handler=digest_queue.enqueue,
+    )
+    bus.subscribe(router.handle)
+"""
+
+import logging
+from typing import Callable
+
+from meshai.notifications.events import Event
+from meshai.notifications.categories import get_toggle
+
+
+class SeverityRouter:
+    """Routes events to immediate or digest handlers based on severity.
+
+    Immediate-severity events go directly to live delivery channels.
+    Priority and routine events are queued for periodic digest summaries.
+    """
+
+    def __init__(
+        self,
+        immediate_handler: Callable[[Event], None],
+        digest_handler: Callable[[Event], None],
+    ):
+        """Initialize the severity router.
+
+        Args:
+            immediate_handler: Called for severity="immediate" events
+            digest_handler: Called for severity in ("priority", "routine")
+        """
+        self._immediate = immediate_handler
+        self._digest = digest_handler
+        self._logger = logging.getLogger("meshai.pipeline.severity_router")
+
+    def handle(self, event: Event) -> None:
+        """Route an event based on its severity.
+
+        Args:
+            event: The Event to route
+        """
+        if event.severity == "immediate":
+            self._logger.info(
+                f"IMMEDIATE: {event.source}/{event.category} {event.title}"
+            )
+            self._immediate(event)
+        elif event.severity in ("priority", "routine"):
+            self._logger.info(
+                f"DIGEST QUEUED [{event.severity}]: {event.title}"
+            )
+            self._digest(event)
+        else:
+            self._logger.warning(
+                f"Unknown severity {event.severity!r} on event {event.id}, dropping"
+            )
+
+
+class StubDigestQueue:
+    """Placeholder digest queue for Phase 2.1.
+
+    This is a stub that simply collects events in memory. Phase 2.3
+    will replace this with the real aggregator that renders and
+    delivers periodic digest summaries.
+    """
+
+    def __init__(self):
+        self._queue: list[Event] = []
+        self._logger = logging.getLogger("meshai.pipeline.digest_stub")
+
+    def enqueue(self, event: Event) -> None:
+        """Add an event to the digest queue.
+
+        Args:
+            event: The Event to queue for digest delivery
+        """
+        self._queue.append(event)
+        toggle = get_toggle(event.category) or "unknown"
+        self._logger.info(f"DIGEST QUEUED [{toggle}]: {event.title}")
+
+    def drain(self) -> list[Event]:
+        """Return and clear all queued events.
+
+        For tests and the future aggregator. Returns the current
+        queue contents and resets the queue to empty.
+
+        Returns:
+            List of all queued Events
+        """
+        events, self._queue = self._queue, []
+        return events
+
+    def __len__(self) -> int:
+        """Return the number of queued events."""
+        return len(self._queue)

work/Dockerfile

@@ -93,7 +93,7 @@ EXPOSE 7682 8080
 
 # Health check - verify bot process is alive via PID file
 HEALTHCHECK --interval=30s --timeout=10s --start-period=10s --retries=3 \
-    CMD test -f /tmp/meshai.pid && kill -0 $(cat /tmp/meshai.pid) 2>/dev/null || exit 1
+    CMD test -f /tmp/meshai.pid && kill -0 $(cat /tmp/meshai.pid) 2>/dev/null && [ "$(cat /tmp/meshai.link 2>/dev/null)" = up ] || exit 1
 
 # Entrypoint handles config and ttyd
 ENTRYPOINT ["/app/docker-entrypoint.sh"]

work/docker-compose.yml

@@ -67,7 +67,7 @@ services:
           memory: 64M
 
     healthcheck:
-      test: ["CMD-SHELL", "test -f /tmp/meshai.pid && kill -0 $(cat /tmp/meshai.pid) 2>/dev/null || exit 1"]
+      test: ["CMD-SHELL", "test -f /tmp/meshai.pid && kill -0 $(cat /tmp/meshai.pid) 2>/dev/null && [ \"$(cat /tmp/meshai.link 2>/dev/null)\" = up ] || exit 1"]
       interval: 30s
       timeout: 10s
       retries: 3

work/meshai/config.py

@@ -29,6 +29,11 @@ class ConnectionConfig:
     serial_port: str = "/dev/ttyUSB0"
     tcp_host: str = "192.168.1.100"
     tcp_port: int = 4403
+    # --- app-level auto-reconnect (watchdog) ---
+    reconnect: bool = True
+    reconnect_initial_delay: float = 2.0
+    reconnect_max_delay: float = 60.0
+    reconnect_health_interval: float = 30.0
 
 
 @dataclass

work/meshai/connector.py

@@ -2,7 +2,9 @@
 
 import asyncio
 import logging
+import socket
 import threading
+import time
 from dataclasses import dataclass, field
 from typing import Callable, Optional
 
@@ -48,6 +50,12 @@ class MeshConnector:
         self._connected = False
         self._loop: Optional[asyncio.AbstractEventLoop] = None
         self._lock = threading.Lock()
+        # --- watchdog link-state (watchdog is the SINGLE source of truth) ---
+        self._last_rx: float = time.monotonic()
+        self._link_suspect: bool = False
+        self._wake: Optional[asyncio.Event] = None  # created by main once loop exists
+        self._link_path = "/tmp/meshai.link"
+        self._reconnect_lock = threading.Lock()
 
     @property
     def connected(self) -> bool:
@@ -86,6 +94,9 @@ class MeshConnector:
             # Subscribe to messages
             pub.subscribe(self._on_receive, "meshtastic.receive.text")
             pub.subscribe(self._on_node_update, "meshtastic.node.updated")
+            pub.subscribe(self._on_any_rx, "meshtastic.receive")
+            pub.subscribe(self._on_connection_lost, "meshtastic.connection.lost")
+            self._last_rx = time.monotonic()
 
             self._connected = True
 
@@ -100,6 +111,8 @@ class MeshConnector:
             try:
                 pub.unsubscribe(self._on_receive, "meshtastic.receive.text")
                 pub.unsubscribe(self._on_node_update, "meshtastic.node.updated")
+                pub.unsubscribe(self._on_any_rx, "meshtastic.receive")
+                pub.unsubscribe(self._on_connection_lost, "meshtastic.connection.lost")
             except Exception:
                 pass
 
@@ -145,6 +158,7 @@ class MeshConnector:
 
     def _on_node_update(self, node, interface) -> None:
         """Handle node info updates."""
+        self._last_rx = time.monotonic()
         node_id = f"!{node['num']:08x}"
 
         with self._lock:
@@ -162,6 +176,7 @@ class MeshConnector:
 
     def _on_receive(self, packet, interface) -> None:
         """Handle incoming text message."""
+        self._last_rx = time.monotonic()
         if not self._message_callback or not self._loop:
             return
 
@@ -207,6 +222,121 @@ class MeshConnector:
         except Exception as e:
             logger.error(f"Error processing received message: {e}")
 
+    # --- watchdog: receive ANY packet bumps last_rx (liveness) ---
+    def _on_any_rx(self, packet=None, interface=None, *args, **kwargs):
+        self._last_rx = time.monotonic()
+
+    # --- watchdog: connection.lost ONLY flags suspect + wakes watchdog.
+    # Strict pubsub arg-spec passes interface= -> signature MUST accept it.
+    # Does NOT write link state, NOT toggle _connected, NOT reconnect.
+    def _on_connection_lost(self, interface=None, *args, **kwargs):
+        self._link_suspect = True
+        logger.warning("connection.lost fired -> link suspect (waking watchdog)")
+        if self._wake is not None and self._loop is not None:
+            try:
+                self._loop.call_soon_threadsafe(self._wake.set)
+            except Exception:
+                pass
+
+    @property
+    def last_rx(self) -> float:
+        return self._last_rx
+
+    @property
+    def link_suspect(self) -> bool:
+        return self._link_suspect
+
+    def clear_suspect(self) -> None:
+        self._link_suspect = False
+
+    def write_link_status(self, status: str) -> None:
+        """SINGLE writer of the link-state file. Called ONLY by the watchdog."""
+        try:
+            with open(self._link_path, "w") as f:
+                f.write(status)
+        except Exception as e:
+            logger.warning(f"Failed to write link status: {e}")
+
+    # --- watchdog: socket-based liveness (getMyNodeInfo is NOT trusted; it
+    # returns cached data on a dead link -> false-alive). Verified meshtastic 2.7.9.
+    def socket_state(self) -> str:
+        """'open', or 'dead' from isConnected + non-blocking MSG_PEEK."""
+        iface = self._interface
+        if iface is None:
+            return "dead"
+        ev = getattr(iface, "isConnected", None)
+        if ev is not None and hasattr(ev, "is_set") and not ev.is_set():
+            return "dead"
+        sock = getattr(iface, "socket", None)
+        if sock is None:
+            return "dead"
+        try:
+            data = sock.recv(1, socket.MSG_DONTWAIT | socket.MSG_PEEK)
+            return "open" if data else "dead"
+        except BlockingIOError:
+            return "open"
+        except (OSError, AttributeError):
+            return "dead"
+
+    def interface_open(self) -> bool:
+        return self.socket_state() == "open"
+
+    def active_probe(self, probe_wait: float = 5.0) -> bool:
+        """sendHeartbeat then watch for genuine life. BLOCKING -> call via to_thread.
+        ALIVE = last_rx advances OR socket stays verifiably open. Never trusts getMyNodeInfo."""
+        if self.socket_state() == "dead":
+            logger.info("probe: DEAD (socket closed / isConnected cleared)")
+            return False
+        rx_before = self._last_rx
+        try:
+            self._interface.sendHeartbeat()
+        except Exception as e:
+            logger.info(f"probe: sendHeartbeat raised {e!r} -> link dead")
+            return False
+        deadline = time.monotonic() + probe_wait
+        while time.monotonic() < deadline:
+            if self._last_rx > rx_before:
+                logger.info("probe: ALIVE (last_rx advanced)")
+                return True
+            if self.socket_state() == "dead":
+                logger.info("probe: DEAD (socket went dead during probe)")
+                return False
+            time.sleep(0.25)
+        if self.socket_state() == "open":
+            logger.info("probe: ALIVE (socket verifiably open)")
+            return True
+        logger.info("probe: DEAD (no rx advance, socket not open)")
+        return False
+
+    def reconnect(self) -> None:
+        """Tear down old interface + build a fresh one under the reconnect lock,
+        exponential backoff forever. BLOCKING -> call via to_thread. Does NOT
+        write link state (watchdog owns that)."""
+        initial = getattr(self.config, "reconnect_initial_delay", 2.0)
+        maxd = getattr(self.config, "reconnect_max_delay", 60.0)
+        with self._reconnect_lock:
+            backoff = initial
+            attempt = 0
+            while True:
+                attempt += 1
+                logger.warning(f"reconnect attempt {attempt}...")
+                # tear down old interface (also stops the lib's internal threads)
+                try:
+                    self.disconnect()
+                except Exception as e:
+                    logger.warning(f"  teardown warn: {e!r}")
+                # build fresh
+                try:
+                    self.connect()
+                    self._last_rx = time.monotonic()
+                    self._link_suspect = False
+                    logger.info(f"reconnected as node {self._my_node_id} (attempt {attempt})")
+                    return
+                except Exception as e:
+                    logger.warning(f"  reconnect attempt {attempt} failed: {e!r}; sleeping {backoff:.0f}s")
+                    time.sleep(backoff)
+                    backoff = min(backoff * 2, maxd)
+
     def send_message(
         self,
         text: str,

work/meshai/main.py

@@ -56,6 +56,7 @@ class MeshAI:
         self.router: Optional[MessageRouter] = None
         self.responder: Optional[Responder] = None
         self._running = False
+        self._supervisor_task = None  # watchdog (connection supervisor)
         self._loop: Optional[asyncio.AbstractEventLoop] = None
         self._last_cleanup: float = 0.0
         self._last_health_compute: float = 0.0
@@ -79,6 +80,14 @@ class MeshAI:
 
         self._running = True
         self._loop = asyncio.get_event_loop()
+        # --- connection supervisor (watchdog): single source of truth for link
+        # state + the only reconnect driver. Reconnects IN-PLACE so the container
+        # never needs to restart. ---
+        if getattr(self.config.connection, "reconnect", True):
+            self.connector._wake = asyncio.Event()
+            self.connector.write_link_status("up")  # we just connected ok
+            self._supervisor_task = asyncio.create_task(self._connection_supervisor())
+            logger.info("Connection supervisor (watchdog) started")
         self._last_cleanup = time.time()
         self._last_health_compute = 0.0
 
@@ -211,6 +220,55 @@ class MeshAI:
                     self.context.prune()
                 self._last_cleanup = time.time()
 
+    async def _connection_supervisor(self) -> None:
+        """Watchdog: SINGLE source of truth for /tmp/meshai.link and the ONLY
+        reconnect driver. Woken by connector._wake (connection.lost) or a
+        health-interval timeout. Probe is socket-based (see connector.active_probe).
+        """
+        c = self.connector
+        hi = getattr(self.config.connection, "reconnect_health_interval", 30.0)
+        alive_idle = 2.0 * hi
+        probe_wait = 5.0
+        wake = c._wake
+        logger.info("watchdog loop running (health_interval=%.0fs)", hi)
+        while self._running:
+            try:
+                try:
+                    await asyncio.wait_for(wake.wait(), timeout=hi)
+                except asyncio.TimeoutError:
+                    pass
+                wake.clear()
+                if not self._running:
+                    break
+
+                idle = time.monotonic() - c.last_rx
+                alive = (not c.link_suspect) and (idle < alive_idle) and c.interface_open()
+                if alive:
+                    c.clear_suspect()
+                    c.write_link_status("up")
+                    continue
+
+                # Not trivially alive -> ACTIVE PROBE (blocking -> thread)
+                logger.info("watchdog: link check (suspect=%s, idle=%.1fs) -> probing",
+                            c.link_suspect, idle)
+                probe_alive = await asyncio.to_thread(c.active_probe, probe_wait)
+                if probe_alive:
+                    c.clear_suspect()
+                    c.write_link_status("up")
+                    continue
+
+                # DEAD -> write down, reconnect in place (blocking -> thread)
+                logger.warning("watchdog: link DOWN -> reconnecting")
+                c.write_link_status("down")
+                await asyncio.to_thread(c.reconnect)
+                c.write_link_status("up")
+            except asyncio.CancelledError:
+                break
+            except Exception:
+                logger.exception("watchdog cycle error (continuing)")
+                await asyncio.sleep(1.0)
+        logger.info("watchdog loop exited")
+
     async def stop(self) -> None:
         """Stop the bot."""
         logger.info("Stopping MeshAI...")
@@ -226,6 +284,13 @@ class MeshAI:
         if self._fire_pacer is not None:
             await self._fire_pacer.stop()
 
+        if self._supervisor_task is not None:
+            self._supervisor_task.cancel()
+            try:
+                await self._supervisor_task
+            except (asyncio.CancelledError, Exception):
+                pass
+
         if self.connector:
             self.connector.disconnect()