diff --git a/Dockerfile b/Dockerfile index 85626a5..96896e1 100644 --- a/Dockerfile +++ b/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"] diff --git a/docker-compose.yml b/docker-compose.yml index 69fce55..5c2ad24 100644 --- a/docker-compose.yml +++ b/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 diff --git a/meshai/config.py b/meshai/config.py index 7a95b0a..f1edfe7 100644 --- a/meshai/config.py +++ b/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 diff --git a/meshai/connector.py b/meshai/connector.py index c0d8d4e..a7d3e19 100644 --- a/meshai/connector.py +++ b/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 - diff --git a/meshai/main.py b/meshai/main.py index 45ba167..f2c7105 100644 --- a/meshai/main.py +++ b/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) diff --git a/meshai/notifications/pipeline/severity_router.py b/meshai/notifications/pipeline/severity_router.py new file mode 100644 index 0000000..a91fffa --- /dev/null +++ b/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) diff --git a/work/Dockerfile b/work/Dockerfile index a2a867f..0dec6df 100644 --- a/work/Dockerfile +++ b/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"] diff --git a/work/docker-compose.yml b/work/docker-compose.yml index 69fce55..cc7ca64 100644 --- a/work/docker-compose.yml +++ b/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 diff --git a/work/meshai/config.py b/work/meshai/config.py index 7a95b0a..acfda69 100644 --- a/work/meshai/config.py +++ b/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 diff --git a/work/meshai/connector.py b/work/meshai/connector.py index c0d8d4e..e805b75 100644 --- a/work/meshai/connector.py +++ b/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, diff --git a/work/meshai/main.py b/work/meshai/main.py index 45ba167..af0c068 100644 --- a/work/meshai/main.py +++ b/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()