diff --git a/work/meshai/config.py b/work/meshai/config.py index 6c496bd..7df9bea 100644 --- a/work/meshai/config.py +++ b/work/meshai/config.py @@ -559,6 +559,8 @@ class NotificationRuleConfig: # Mesh DM fields node_ids: list = field(default_factory=list) + # MeshCore DM target contacts (names or pubkeys). Parallel to node_ids for Meshtastic. + meshcore_dm_contacts: list = field(default_factory=list) # Email fields smtp_host: str = "" @@ -600,6 +602,8 @@ class NotificationToggle: # Per-family MeshCore channel NAME on the companion; None = not broadcast on MeshCore. meshcore_channel: Optional[str] = None node_ids: list = field(default_factory=list) + # MeshCore DM target contacts (names or pubkeys). Parallel to node_ids for Meshtastic. + meshcore_dm_contacts: list = field(default_factory=list) smtp_host: str = "" smtp_port: int = 587 smtp_user: str = "" diff --git a/work/meshai/notifications/channels.py b/work/meshai/notifications/channels.py index 555d5d3..e99ba76 100644 --- a/work/meshai/notifications/channels.py +++ b/work/meshai/notifications/channels.py @@ -56,22 +56,23 @@ class NotificationChannel(ABC): class MeshBroadcastChannel(NotificationChannel): - """Post alert to mesh channel.""" + """Post alert to Meshtastic channel (explicit Meshtastic-only delivery).""" channel_type = "mesh_broadcast" def __init__(self, connector: "MeshConnector", channel_index: int = 0, - meshcore_channel: Optional[str] = None): + transport: Optional[str] = "meshtastic"): self._connector = connector self._channel = channel_index - # Per-family MeshCore channel NAME (None = MeshCore child skipped - # downstream). Ignored by Meshtastic; behavior-preserving there. - self._meshcore_channel = meshcore_channel + # Transport hint: "meshtastic" for mesh_broadcast; passed to CompositeTransport + # so it routes only to the Meshtastic child. Single-transport implementations + # accept and ignore this kwarg, so behavior is unchanged there. + self._transport = transport _mc = getattr(connector, "max_chars", 200) self._renderer = MeshRenderer(char_limit=_mc if isinstance(_mc, int) else 200) async def deliver(self, alert: "NotificationPayload", rule: "NotificationRuleConfig") -> bool: - """Send alert to mesh channel.""" + """Send alert to Meshtastic channel.""" if not self._connector: logger.warning("No mesh connector available") return False @@ -83,7 +84,7 @@ class MeshBroadcastChannel(NotificationChannel): text=alert.message or "", destination=None, channel=self._channel, - meshcore_channel=self._meshcore_channel, + transport=self._transport, ) logger.info("Broadcast pre-chunked alert to channel %d", self._channel) return True @@ -95,7 +96,7 @@ class MeshBroadcastChannel(NotificationChannel): text=chunk, destination=None, channel=self._channel, - meshcore_channel=self._meshcore_channel, + transport=self._transport, ) logger.info("Broadcast %d chunk(s) to channel %d", len(chunks), self._channel) return True @@ -173,19 +174,127 @@ class MeshBroadcastChannel(NotificationChannel): return False, f"Mesh broadcast failed: {e}" +class MeshCoreBroadcastChannel(NotificationChannel): + """Post alert to a MeshCore channel (explicit MeshCore-only delivery).""" + + channel_type = "meshcore_broadcast" + + def __init__(self, connector: "MeshConnector", meshcore_channel: Optional[str] = None): + self._connector = connector + # Channel NAME on the MeshCore companion (resolved to a slot at send time). + self._meshcore_channel = meshcore_channel + _mc = getattr(connector, "max_chars", 200) + self._renderer = MeshRenderer(char_limit=_mc if isinstance(_mc, int) else 200) + + def _has_meshcore_capability(self) -> bool: + """Return True if the connector can reach a MeshCore transport.""" + # CompositeTransport: check for a child named "meshcore". + by_name = getattr(self._connector, "_by_name", None) + if by_name is not None: + return "meshcore" in by_name + # Single-transport: check for an explicit transport_name tag. + return getattr(self._connector, "transport_name", None) == "meshcore" + + async def deliver(self, alert: "NotificationPayload", rule: "NotificationRuleConfig") -> bool: + """Send alert to MeshCore channel.""" + if not self._connector: + logger.warning("No mesh connector available for meshcore_broadcast") + return False + + if not self._meshcore_channel: + logger.debug("meshcore_broadcast: meshcore_channel not set; skipping") + return False + + if not self._has_meshcore_capability(): + logger.debug( + "meshcore_broadcast: connector has no MeshCore transport; skipping" + ) + return False + + try: + # If payload already has chunk metadata (from digest), use message directly + if alert.chunk_index is not None: + self._connector.send_message( + text=alert.message or "", + destination=None, + meshcore_channel=self._meshcore_channel, + transport="meshcore", + ) + logger.info( + "MeshCore broadcast pre-chunked alert to channel %r", + self._meshcore_channel, + ) + return True + + # Render to chunks for single-event delivery + chunks = self._renderer.render(alert) + for chunk in chunks: + self._connector.send_message( + text=chunk, + destination=None, + meshcore_channel=self._meshcore_channel, + transport="meshcore", + ) + logger.info( + "MeshCore broadcast %d chunk(s) to channel %r", + len(chunks), self._meshcore_channel, + ) + return True + except Exception as e: + logger.error("Failed to MeshCore broadcast alert: %s", e) + return False + + async def test_connection(self) -> dict: + """Test MeshCore channel connectivity.""" + if not self._has_meshcore_capability(): + return { + "success": False, + "message": "No MeshCore transport available", + "error": "Set connection.transport to 'meshcore' or 'both'", + "details": {"meshcore_channel": self._meshcore_channel}, + } + return { + "success": True, + "message": f"MeshCore channel: {self._meshcore_channel}", + "error": "", + "details": {"meshcore_channel": self._meshcore_channel}, + } + + async def deliver_test(self, message: str) -> tuple[bool, str]: + """Deliver a specific test message to the MeshCore channel.""" + if not self._connector: + return False, "Not connected" + if not self._meshcore_channel: + return False, "No MeshCore channel configured" + try: + self._connector.send_message( + text=message, + destination=None, + meshcore_channel=self._meshcore_channel, + transport="meshcore", + ) + return True, f"Sent to MeshCore channel {self._meshcore_channel!r}" + except Exception as e: + return False, f"MeshCore broadcast failed: {e}" + + class MeshDMChannel(NotificationChannel): - """DM alert to specific node IDs.""" + """DM alert to specific Meshtastic node IDs.""" channel_type = "mesh_dm" - def __init__(self, connector: "MeshConnector", node_ids: list[str]): + def __init__(self, connector: "MeshConnector", node_ids: list[str], + transport_hint: Optional[str] = "meshtastic"): self._connector = connector self._node_ids = node_ids + # Explicit transport hint so CompositeTransport routes only to the + # Meshtastic child. Single-transport impls ignore this kwarg. + self._transport_hint = transport_hint _mc = getattr(connector, "max_chars", 200) self._renderer = MeshRenderer(char_limit=_mc if isinstance(_mc, int) else 200) async def deliver(self, alert: "NotificationPayload", rule: "NotificationRuleConfig") -> bool: - """Send alert via DM to configured nodes.""" + """Send alert via DM to configured Meshtastic nodes.""" if not self._connector: return False @@ -201,7 +310,12 @@ class MeshDMChannel(NotificationChannel): for message in messages: try: node_id = str(node_id) - self._connector.send_message(text=message, destination=node_id, channel=0) + self._connector.send_message( + text=message, + destination=node_id, + channel=0, + transport=self._transport_hint, + ) except Exception as e: logger.error("Failed to DM %s: %s", node_id, e) success = False @@ -295,6 +409,109 @@ class MeshDMChannel(NotificationChannel): return False, f"All DMs failed: {'; '.join(errors)}" +class MeshCoreDMChannel(NotificationChannel): + """DM alert to specific MeshCore contacts (names or pubkeys).""" + + channel_type = "meshcore_dm" + + def __init__(self, connector: "MeshConnector", contacts: list): + self._connector = connector + self._contacts = list(contacts) + _mc = getattr(connector, "max_chars", 200) + self._renderer = MeshRenderer(char_limit=_mc if isinstance(_mc, int) else 200) + + def _has_meshcore_capability(self) -> bool: + """Return True if the connector can reach a MeshCore transport.""" + by_name = getattr(self._connector, "_by_name", None) + if by_name is not None: + return "meshcore" in by_name + return getattr(self._connector, "transport_name", None) == "meshcore" + + async def deliver(self, alert: "NotificationPayload", rule: "NotificationRuleConfig") -> bool: + """Send alert via DM to configured MeshCore contacts.""" + if not self._connector: + return False + + if not self._contacts: + logger.debug("meshcore_dm: no contacts configured; skipping") + return False + + if not self._has_meshcore_capability(): + logger.debug( + "meshcore_dm: connector has no MeshCore transport; skipping" + ) + return False + + # If payload already has chunk metadata (from digest), use message directly + if alert.chunk_index is not None: + messages = [alert.message or ""] + else: + messages = self._renderer.render(alert) + + success = True + for contact in self._contacts: + for message in messages: + try: + self._connector.send_message( + text=message, + destination=str(contact), + transport="meshcore", + ) + except Exception as e: + logger.error("Failed to MeshCore DM %s: %s", contact, e) + success = False + + return success + + async def test_connection(self) -> dict: + """Test MeshCore DM connectivity.""" + if not self._has_meshcore_capability(): + return { + "success": False, + "message": "No MeshCore transport available", + "error": "Set connection.transport to 'meshcore' or 'both'", + "details": {"contacts": self._contacts}, + } + if not self._contacts: + return { + "success": False, + "message": "No MeshCore DM contacts configured", + "error": "Add at least one contact to meshcore_dm_contacts", + "details": {"contacts": []}, + } + return { + "success": True, + "message": f"MeshCore DM to {len(self._contacts)} contact(s)", + "error": "", + "details": {"contacts": self._contacts}, + } + + async def deliver_test(self, message: str) -> tuple[bool, str]: + """Deliver a specific test message via MeshCore DM.""" + if not self._connector: + return False, "Not connected" + if not self._contacts: + return False, "No MeshCore DM contacts configured" + success_count = 0 + errors = [] + for contact in self._contacts: + try: + self._connector.send_message( + text=message, + destination=str(contact), + transport="meshcore", + ) + success_count += 1 + except Exception as e: + errors.append(f"{contact}: {e}") + if success_count == len(self._contacts): + return True, f"Sent MeshCore DM to {success_count} contact(s)" + elif success_count > 0: + return True, f"Sent to {success_count}/{len(self._contacts)} contacts. Errors: {'; '.join(errors)}" + else: + return False, f"All MeshCore DMs failed: {'; '.join(errors)}" + + class EmailChannel(NotificationChannel): """Send alert via SMTP email.""" @@ -772,26 +989,51 @@ class WebhookChannel(NotificationChannel): def create_channel(rule: "NotificationRuleConfig", connector=None) -> NotificationChannel: """Create a channel instance from a NotificationRuleConfig. - + + Delivery types and their per-mesh routing: + mesh_broadcast -> Meshtastic ONLY (broadcast_channel; transport="meshtastic") + meshcore_broadcast-> MeshCore ONLY (meshcore_channel NAME; transport="meshcore") + mesh_dm -> Meshtastic DM (node_ids; transport="meshtastic") + meshcore_dm -> MeshCore DM (meshcore_dm_contacts; transport="meshcore") + email -> SMTP email + webhook -> HTTP POST + Args: rule: NotificationRuleConfig with delivery_type and channel settings connector: MeshConnector instance (required for mesh channels) - + Returns: NotificationChannel instance """ delivery_type = rule.delivery_type if delivery_type == "mesh_broadcast": + # Meshtastic-only broadcast: explicit transport hint so CompositeTransport + # routes only to the Meshtastic child and skips MeshCore. return MeshBroadcastChannel( connector=connector, channel_index=rule.broadcast_channel, + transport="meshtastic", + ) + elif delivery_type == "meshcore_broadcast": + # MeshCore-only broadcast: routes to MeshCore child by channel NAME. + return MeshCoreBroadcastChannel( + connector=connector, meshcore_channel=getattr(rule, "meshcore_channel", None), ) elif delivery_type == "mesh_dm": + # Meshtastic-only DM: explicit transport hint so CompositeTransport + # routes only to the Meshtastic child. return MeshDMChannel( connector=connector, node_ids=rule.node_ids, + transport_hint="meshtastic", + ) + elif delivery_type == "meshcore_dm": + # MeshCore-only DM: routes to MeshCore child via contact name/pubkey. + return MeshCoreDMChannel( + connector=connector, + contacts=list(getattr(rule, "meshcore_dm_contacts", []) or []), ) elif delivery_type == "email": return EmailChannel( @@ -814,21 +1056,23 @@ def create_channel(rule: "NotificationRuleConfig", connector=None) -> Notificati def create_channel_from_dict(config: dict, connector=None) -> NotificationChannel: """Create a channel instance from a dict config (legacy interface). - + Used by old router.py and test_channel API. Will be removed in Phase 2.7. """ channel_type = config.get("type", "") if channel_type == "mesh_broadcast": + # Legacy dict configs are Meshtastic-only; no auto-fan. return MeshBroadcastChannel( connector=connector, channel_index=config.get("channel_index", 0), - meshcore_channel=config.get("meshcore_channel"), + transport="meshtastic", ) elif channel_type == "mesh_dm": return MeshDMChannel( connector=connector, node_ids=config.get("node_ids", []), + transport_hint="meshtastic", ) elif channel_type == "email": return EmailChannel( diff --git a/work/meshai/notifications/pipeline/dispatcher.py b/work/meshai/notifications/pipeline/dispatcher.py index a93bfa1..3569087 100644 --- a/work/meshai/notifications/pipeline/dispatcher.py +++ b/work/meshai/notifications/pipeline/dispatcher.py @@ -448,7 +448,9 @@ class Dispatcher: try: rule = self._toggle_to_rule(tog, ch_type, event) channel = self._channel_factory(rule, self._connector) - if friendly is not None and ch_type in ("mesh_broadcast", "mesh_dm"): + if friendly is not None and ch_type in ( + "mesh_broadcast", "mesh_dm", "meshcore_broadcast", "meshcore_dm" + ): payload = make_payload_from_event(event, message=friendly) else: payload = make_payload_from_event(event) @@ -549,15 +551,33 @@ class Dispatcher: source_event_table, source_event_pk) return False - # Route through rf_propagation toggle\'s broadcast_channel. + # Route through rf_propagation toggle\'s configured channels. toggles = getattr(self._config.notifications, "toggles", None) or {} rf = toggles.get("rf_propagation") if isinstance(toggles, dict) else None - if rf is None or not getattr(rf, "broadcast_channel", None): + if rf is None: self._logger.info( - "scheduled-broadcast: rf_propagation channel not " - "configured; dropping") + "scheduled-broadcast: rf_propagation toggle not found; dropping") return False + # Resolve broadcast channel types from the toggle\'s severity_channels for + # "priority" (band-conditions are priority-class RF propagation info). + # Falls back to ["mesh_broadcast"] for old configs without severity_channels. + sev_channels = getattr(rf, "severity_channels", {}) or {} + ch_types = [ + c for c in sev_channels.get("priority", ["mesh_broadcast"]) + if c in ("mesh_broadcast", "meshcore_broadcast") + ] + if not ch_types: + # Backward compat: if severity_channels has no broadcast types, + # use mesh_broadcast when broadcast_channel is configured. + if getattr(rf, "broadcast_channel", None) is not None: + ch_types = ["mesh_broadcast"] + else: + self._logger.info( + "scheduled-broadcast: rf_propagation channel not " + "configured; dropping") + return False + # Build a synthetic Event purely to reuse _toggle_to_rule + the # NotificationPayload constructor. Severity \'priority\' keeps it # out of quiet-hours suppression unless explicitly overridden. @@ -570,35 +590,39 @@ class Dispatcher: severity="priority", title=text, ) ev.data["_meshai_precomposed"] = True - rule = self._toggle_to_rule(rf, "mesh_broadcast", ev) - try: - channel = self._channel_factory(rule, self._connector) - payload = make_payload_from_event(ev, message=text) - success = await channel.deliver(payload, rule) - except Exception: - self._logger.exception( - "scheduled-broadcast: delivery raised; treating as failed") - return False - if success: - # Audit row -- mirrors _post_broadcast_commit for scheduled. + delivered_any = False + for ch_type in ch_types: + rule = self._toggle_to_rule(rf, ch_type, ev) try: - from meshai.persistence import get_db - conn = get_db() - bytes_sent = len(text.encode("utf-8")) if text else 0 - conn.execute( - "INSERT INTO mesh_broadcasts_out(sent_at, recipient, " - "channel, text, source_event_table, source_event_pk, " - "bytes_sent, ack_received) VALUES (?,?,?,?,?,?,?,?)", - (int(time.time()), "broadcast", - rf.broadcast_channel, text, - source_event_table, str(source_event_pk), - bytes_sent, 0), - ) + channel = self._channel_factory(rule, self._connector) + payload = make_payload_from_event(ev, message=text) + success = await channel.deliver(payload, rule) except Exception: self._logger.exception( - "scheduled-broadcast: audit row insert failed") - return bool(success) + "scheduled-broadcast: delivery raised for %s; skipping", ch_type) + continue + + if success: + delivered_any = True + # Audit row -- mirrors _post_broadcast_commit for scheduled. + try: + from meshai.persistence import get_db + conn = get_db() + bytes_sent = len(text.encode("utf-8")) if text else 0 + conn.execute( + "INSERT INTO mesh_broadcasts_out(sent_at, recipient, " + "channel, text, source_event_table, source_event_pk, " + "bytes_sent, ack_received) VALUES (?,?,?,?,?,?,?,?)", + (int(time.time()), "broadcast", + rf.broadcast_channel, text, + source_event_table, str(source_event_pk), + bytes_sent, 0), + ) + except Exception: + self._logger.exception( + "scheduled-broadcast: audit row insert failed for %s", ch_type) + return delivered_any def _post_broadcast_commit(self, event, payload, rule, ch_type: str) -> None: """Persistence side-effects of an actually-successful broadcast. @@ -625,6 +649,9 @@ class Dispatcher: if ch_type == "mesh_dm": node_ids = list(getattr(rule, "node_ids", []) or []) recipient = ",".join(map(str, node_ids)) or "dm" + elif ch_type == "meshcore_dm": + contacts = list(getattr(rule, "meshcore_dm_contacts", []) or []) + recipient = ",".join(map(str, contacts)) or "meshcore_dm" else: recipient = "broadcast" channel = getattr(rule, "broadcast_channel", None) @@ -662,6 +689,7 @@ class Dispatcher: broadcast_channel=(getattr(tog, "broadcast_channel", None) or 0), meshcore_channel=getattr(tog, "meshcore_channel", None), node_ids=list(getattr(tog, "node_ids", []) or []), + meshcore_dm_contacts=list(getattr(tog, "meshcore_dm_contacts", []) or []), smtp_host=getattr(tog, "smtp_host", ""), smtp_port=getattr(tog, "smtp_port", 587), smtp_user=getattr(tog, "smtp_user", ""), smtp_password=getattr(tog, "smtp_password", ""), smtp_tls=getattr(tog, "smtp_tls", True), from_address=getattr(tog, "from_address", ""), diff --git a/work/meshai/transport/composite_transport.py b/work/meshai/transport/composite_transport.py index 572b95e..d6a0c03 100644 --- a/work/meshai/transport/composite_transport.py +++ b/work/meshai/transport/composite_transport.py @@ -224,7 +224,8 @@ class CompositeTransport(MeshTransport): """ if destination is None: # --- Rule 1: broadcast --- - return self._broadcast(text, channel, meshcore_channel=meshcore_channel) + return self._broadcast(text, channel, meshcore_channel=meshcore_channel, + transport=transport) if transport is not None: # --- Rule 2: hinted DM --- @@ -233,18 +234,48 @@ class CompositeTransport(MeshTransport): # --- Rule 3: unhinted DM --- return self._send_unhinted(text, destination, channel) - def _broadcast(self, text: str, channel: int, meshcore_channel: Optional[str] = None) -> bool: + def _broadcast(self, text: str, channel: int, meshcore_channel: Optional[str] = None, + transport: Optional[str] = None) -> bool: """Fan text out to connected children with per-transport channel routing. + If ``transport`` is given, send ONLY to the child whose name matches — + this is the explicit per-mesh delivery path (mesh_broadcast → "meshtastic", + meshcore_broadcast → "meshcore"). If ``transport`` is None, keep the + legacy fan-out: all connected children with per-transport channel routing. + For the Meshtastic child, ``channel`` (Meshtastic channel index) is used. For the MeshCore child: - - ``meshcore_channel`` set → route that channel NAME to MeshCore, - which resolves it to a companion slot at send time. + - ``meshcore_channel`` set → route that channel NAME to MeshCore. - ``meshcore_channel`` is None → skip the MeshCore child entirely (family not configured for MeshCore; no fallback to a default). Returns True if at least one child succeeded. """ + if transport is not None: + # Hinted broadcast: send ONLY to the named child. + child = self._by_name.get(transport) + if child is None: + logger.debug( + "CompositeTransport: broadcast hint %r not found; known: %s", + transport, list(self._by_name), + ) + return False + name = _child_name(child) + if not child.connected: + logger.debug( + "CompositeTransport: hinted broadcast child %r not connected", name + ) + return False + child_channel = meshcore_channel if name == "meshcore" else channel + try: + return child.send_message(text, destination=None, channel=child_channel) + except Exception as exc: + logger.error( + "CompositeTransport: hinted broadcast via %r raised: %s", name, exc + ) + return False + + # No hint: fan to all connected children (backward-compat, no-hint path). any_ok = False for child in self._children: name = _child_name(child) diff --git a/work/tests/test_channel_rendering.py b/work/tests/test_channel_rendering.py index 945127d..20aeac4 100644 --- a/work/tests/test_channel_rendering.py +++ b/work/tests/test_channel_rendering.py @@ -216,15 +216,15 @@ def test_webhook_channel_uses_webhook_renderer(): # ============================================================ # PER-FAMILY MESHCORE ROUTING — end-to-end threading guard -# (regression guard for the broadcast send-path gap) +# Updated for the explicit-per-mesh model (meshcore_broadcast/mesh_broadcast) # ============================================================ -def test_broadcast_threads_meshcore_channel_through_factory(): - """create_channel(rule) -> MeshBroadcastChannel.deliver must pass BOTH - channel= AND meshcore_channel= to send_message. +def test_mesh_broadcast_routes_to_meshtastic_only(): + """mesh_broadcast passes transport='meshtastic' and channel index to + send_message. meshcore_channel is NOT passed (auto-fan removed). - This is the regression guard for the gap where the rule's - meshcore_channel never reached connector.send_message. + Regression guard: before this model, mesh_broadcast also threaded + meshcore_channel through; now it is Meshtastic-only. """ from meshai.config import NotificationRuleConfig from meshai.notifications.channels import create_channel @@ -234,7 +234,7 @@ def test_broadcast_threads_meshcore_channel_through_factory(): name="toggle:fire", delivery_type="mesh_broadcast", broadcast_channel=1, - meshcore_channel="AIDA", + meshcore_channel="AIDA", # present in config but must NOT flow to send_message ) channel = create_channel(rule, mock_connector) @@ -254,54 +254,63 @@ def test_broadcast_threads_meshcore_channel_through_factory(): mock_connector.send_message.assert_called_once() kwargs = mock_connector.send_message.call_args.kwargs assert kwargs.get("channel") == 1 - # The load-bearing assertion: the name was NOT dropped. - assert kwargs.get("meshcore_channel") == "AIDA" + assert kwargs.get("transport") == "meshtastic" + # meshcore_channel must NOT be present (no auto-fan). + assert "meshcore_channel" not in kwargs or kwargs.get("meshcore_channel") is None -def test_broadcast_meshcore_channel_none_passed_through(): - """meshcore_channel=None (family not on MeshCore) => send_message still - receives meshcore_channel=None (MeshCore child skipped downstream).""" +def test_meshcore_broadcast_routes_to_meshcore_only(): + """meshcore_broadcast passes meshcore_channel=name and transport='meshcore' + to send_message. This is the explicit MeshCore-only delivery path.""" from meshai.config import NotificationRuleConfig - from meshai.notifications.channels import create_channel + from meshai.notifications.channels import MeshCoreBroadcastChannel, create_channel + # Simulate a CompositeTransport connector with a meshcore child. mock_connector = MagicMock() + mock_connector._by_name = {"meshcore": MagicMock(), "meshtastic": MagicMock()} + mock_connector.send_message.return_value = True + rule = NotificationRuleConfig( - name="toggle:weather", - delivery_type="mesh_broadcast", - broadcast_channel=0, - meshcore_channel=None, + name="toggle:fire", + delivery_type="meshcore_broadcast", + meshcore_channel="AIDA", ) channel = create_channel(rule, mock_connector) + assert isinstance(channel, MeshCoreBroadcastChannel) payload = NotificationPayload( - message="weather alert", - category="weather_warning", - severity="priority", + message="fire alert", + category="fire", + severity="immediate", timestamp=time.time(), - event_type="weather_warning", + event_type="fire", chunk_index=0, ) assert asyncio.run(channel.deliver(payload, rule)) is True + mock_connector.send_message.assert_called_once() kwargs = mock_connector.send_message.call_args.kwargs - assert kwargs.get("channel") == 0 - assert "meshcore_channel" in kwargs - assert kwargs.get("meshcore_channel") is None + assert kwargs.get("meshcore_channel") == "AIDA" + assert kwargs.get("transport") == "meshcore" + assert kwargs.get("destination") is None def test_broadcast_render_loop_threads_meshcore_channel(): - """Non-prechunked path (renderer loop) also threads meshcore_channel - on every chunk send.""" + """Non-prechunked path (renderer loop) for meshcore_broadcast threads + meshcore_channel on every chunk send.""" from meshai.config import NotificationRuleConfig from meshai.notifications.channels import create_channel mock_connector = MagicMock() + # Connector has a meshcore child so the no-op guard passes. + mock_connector._by_name = {"meshcore": MagicMock(), "meshtastic": MagicMock()} + mock_connector.send_message.return_value = True + rule = NotificationRuleConfig( name="toggle:fire", - delivery_type="mesh_broadcast", - broadcast_channel=2, + delivery_type="meshcore_broadcast", meshcore_channel="AIDA", ) channel = create_channel(rule, mock_connector) @@ -318,5 +327,5 @@ def test_broadcast_render_loop_threads_meshcore_channel(): assert asyncio.run(channel.deliver(payload, rule)) is True assert mock_connector.send_message.call_count >= 2 for call in mock_connector.send_message.call_args_list: - assert call.kwargs.get("channel") == 2 assert call.kwargs.get("meshcore_channel") == "AIDA" + assert call.kwargs.get("transport") == "meshcore" diff --git a/work/tests/test_notification_toggles.py b/work/tests/test_notification_toggles.py index 0a8b407..87d65a6 100644 --- a/work/tests/test_notification_toggles.py +++ b/work/tests/test_notification_toggles.py @@ -1,10 +1,16 @@ -"""v0.5 Section 1: NotificationToggle dispatch routing tests.""" +"""v0.5 Section 1: NotificationToggle dispatch routing tests. + +Also covers the per-mesh delivery type routing introduced in +feat/meshcore-first-class-delivery (meshcore_broadcast, meshcore_dm). +""" import asyncio +from unittest.mock import MagicMock -from meshai.config import Config +from meshai.config import Config, NotificationToggle from meshai.notifications.pipeline.dispatcher import Dispatcher from meshai.notifications.events import make_event +from meshai.notifications.channels import create_channel class RecChannel: @@ -134,3 +140,297 @@ def test_rules_and_toggles_both_fire(): rec = _dispatch(cfg, _ev(severity="priority")) names = {r["name"] for r in rec} assert "legacy" in names and "toggle:weather" in names # parallel paths both fire + + +# ============================================================ +# Per-mesh delivery type routing tests (feat/meshcore-first-class-delivery) +# ============================================================ + +def _wipe_db(): + """Wipe dispatcher persistence so each test is independent.""" + try: + from meshai.persistence import get_db + conn = get_db() + conn.execute("DELETE FROM dispatcher_dedup") + conn.execute("DELETE FROM dispatcher_cooldowns") + conn.execute( + "UPDATE dispatcher_state SET cold_start_anchor=NULL, " + "stale_dropped=0, cooldown_dropped=0, dedup_dropped=0, " + "cold_start_dropped=0 WHERE id=1" + ) + except Exception: + pass + + +def _dispatch_with_connector(cfg, event, connector): + """Dispatch event, using a real connector so send_message calls are captured.""" + _wipe_db() + delivered_rules = [] + + def _factory(rule, conn): + ch = create_channel(rule, connector) + # Wrap to record rule metadata too. + original_deliver = ch.deliver + + async def _record_deliver(payload, r): + result = await original_deliver(payload, r) + delivered_rules.append({ + "delivery_type": r.delivery_type, + "meshcore_channel": getattr(r, "meshcore_channel", None), + "meshcore_dm_contacts": list(getattr(r, "meshcore_dm_contacts", []) or []), + "node_ids": list(getattr(r, "node_ids", []) or []), + }) + return result + + ch.deliver = _record_deliver + return ch + + d = Dispatcher(cfg, _factory, connector=connector) + asyncio.run(d.dispatch(event)) + return delivered_rules + + +def test_meshcore_broadcast_routes_to_meshcore_child_only(): + """meshcore_broadcast in severity_channels → send_message called with + transport='meshcore' and the family's meshcore_channel name. + The Meshtastic child must NOT be called for this type.""" + meshtastic_child = MagicMock() + meshtastic_child.connected = True + meshtastic_child.transport_name = "meshtastic" + meshtastic_child.send_message.return_value = True + + meshcore_child = MagicMock() + meshcore_child.connected = True + meshcore_child.transport_name = "meshcore" + meshcore_child.send_message.return_value = True + + from meshai.transport.composite_transport import CompositeTransport + connector = CompositeTransport([meshtastic_child, meshcore_child]) + # Simulate that the connector has a meshcore child (for capability check in channel). + connector._by_name = {"meshtastic": meshtastic_child, "meshcore": meshcore_child} + + cfg = Config() + cfg.notifications.rules = [] + cfg.notifications.cold_start_grace_seconds = 0 + t = cfg.notifications.toggles["fire"] + t.enabled = True + t.min_severity = "immediate" + t.severity_channels = {"immediate": ["meshcore_broadcast"]} + t.broadcast_channel = 0 + t.meshcore_channel = "AIDA" + + event = make_event( + source="wfigs", category="fire_perimeter", + severity="immediate", title="fire alert", + ) + + rules = _dispatch_with_connector(cfg, event, connector) + assert len(rules) == 1 + assert rules[0]["delivery_type"] == "meshcore_broadcast" + assert rules[0]["meshcore_channel"] == "AIDA" + + # MeshCore child received the call with correct transport+channel. + assert meshcore_child.send_message.called + mc_kwargs = meshcore_child.send_message.call_args.kwargs + assert mc_kwargs.get("destination") is None + assert mc_kwargs.get("channel") == "AIDA" # child receives channel=meshcore_channel + + # Meshtastic child must NOT have been called. + meshtastic_child.send_message.assert_not_called() + + +def test_mesh_broadcast_routes_to_meshtastic_child_only(): + """mesh_broadcast → send_message with transport='meshtastic' and + the Meshtastic channel index. MeshCore child must NOT be called.""" + meshtastic_child = MagicMock() + meshtastic_child.connected = True + meshtastic_child.transport_name = "meshtastic" + meshtastic_child.send_message.return_value = True + + meshcore_child = MagicMock() + meshcore_child.connected = True + meshcore_child.transport_name = "meshcore" + meshcore_child.send_message.return_value = True + + from meshai.transport.composite_transport import CompositeTransport + connector = CompositeTransport([meshtastic_child, meshcore_child]) + connector._by_name = {"meshtastic": meshtastic_child, "meshcore": meshcore_child} + + cfg = Config() + cfg.notifications.rules = [] + cfg.notifications.cold_start_grace_seconds = 0 + t = cfg.notifications.toggles["weather"] + t.enabled = True + t.min_severity = "priority" + t.severity_channels = {"priority": ["mesh_broadcast"]} + t.broadcast_channel = 3 + + event = make_event( + source="nws", category="weather_warning", + severity="priority", title="weather alert", + ) + + rules = _dispatch_with_connector(cfg, event, connector) + assert len(rules) == 1 + assert rules[0]["delivery_type"] == "mesh_broadcast" + + # Meshtastic child received the call. + assert meshtastic_child.send_message.called + mt_kwargs = meshtastic_child.send_message.call_args.kwargs + assert mt_kwargs.get("destination") is None + assert mt_kwargs.get("channel") == 3 + + # MeshCore child must NOT have been called. + meshcore_child.send_message.assert_not_called() + + +def test_meshcore_dm_routes_to_meshcore_contacts(): + """meshcore_dm → connector.send_message called per meshcore_dm_contacts + entry with transport='meshcore'.""" + from meshai.notifications.channels import MeshCoreDMChannel + + mock_connector = MagicMock() + mock_connector._by_name = {"meshcore": MagicMock(), "meshtastic": MagicMock()} + mock_connector.send_message.return_value = True + + from meshai.config import NotificationRuleConfig + import time as _time + from meshai.notifications.events import NotificationPayload + + rule = NotificationRuleConfig( + name="toggle:mesh_health", + delivery_type="meshcore_dm", + meshcore_dm_contacts=["alice", "bob"], + ) + + channel = create_channel(rule, mock_connector) + assert isinstance(channel, MeshCoreDMChannel) + + payload = NotificationPayload( + message="dm alert", + category="mesh_health", + severity="immediate", + timestamp=_time.time(), + chunk_index=0, + ) + + result = asyncio.run(channel.deliver(payload, rule)) + assert result is True + + # One send_message call per contact. + assert mock_connector.send_message.call_count == 2 + destinations = [ + call.kwargs.get("destination") + for call in mock_connector.send_message.call_args_list + ] + assert set(destinations) == {"alice", "bob"} + for call in mock_connector.send_message.call_args_list: + assert call.kwargs.get("transport") == "meshcore" + + +def test_meshcore_broadcast_noop_when_no_meshcore_transport(): + """meshcore_broadcast with transport=meshtastic (no MeshCore child) → + deliver returns False, no exception raised.""" + from meshai.notifications.channels import MeshCoreBroadcastChannel + from meshai.config import NotificationRuleConfig + import time as _time + from meshai.notifications.events import NotificationPayload + + # Connector has NO meshcore child (transport=meshtastic scenario). + mock_connector = MagicMock() + # _by_name exists but has only meshtastic. + mock_connector._by_name = {"meshtastic": MagicMock()} + + rule = NotificationRuleConfig( + name="toggle:fire", + delivery_type="meshcore_broadcast", + meshcore_channel="AIDA", + ) + + channel = create_channel(rule, mock_connector) + assert isinstance(channel, MeshCoreBroadcastChannel) + + payload = NotificationPayload( + message="fire alert", + category="fire", + severity="immediate", + timestamp=_time.time(), + chunk_index=0, + ) + + # Must not raise; returns False (no-op). + result = asyncio.run(channel.deliver(payload, rule)) + assert result is False + # send_message must NOT have been called (no accidental Meshtastic send). + mock_connector.send_message.assert_not_called() + + +def test_config_round_trip_meshcore_fields(): + """NotificationToggle with meshcore types in severity_channels and + meshcore_dm_contacts survives _dataclass_to_dict / _dict_to_dataclass.""" + from meshai.config import _dataclass_to_dict, _dict_to_dataclass, NotificationToggle + + tog = NotificationToggle( + name="fire", + enabled=True, + min_severity="immediate", + severity_channels={ + "priority": ["meshcore_broadcast"], + "immediate": ["mesh_broadcast", "meshcore_broadcast", "meshcore_dm"], + }, + broadcast_channel=1, + meshcore_channel="AIDA", + meshcore_dm_contacts=["alice", "bob"], + node_ids=["!deadbeef"], + ) + + d = _dataclass_to_dict(tog) + assert d["meshcore_dm_contacts"] == ["alice", "bob"] + assert d["meshcore_channel"] == "AIDA" + assert "meshcore_broadcast" in d["severity_channels"]["priority"] + assert "meshcore_dm" in d["severity_channels"]["immediate"] + + restored = _dict_to_dataclass(NotificationToggle, d) + assert restored.meshcore_dm_contacts == ["alice", "bob"] + assert restored.meshcore_channel == "AIDA" + assert "meshcore_broadcast" in restored.severity_channels["priority"] + assert "meshcore_dm" in restored.severity_channels["immediate"] + assert restored.node_ids == ["!deadbeef"] + + +def test_meshtastic_only_config_unchanged(): + """Existing configs with only mesh_broadcast/mesh_dm and + transport=meshtastic behave identically to pre-MeshCore behavior.""" + mock_connector = MagicMock() + # Simulate a plain MeshtasticTransport (no _by_name, transport_name=meshtastic). + mock_connector.transport_name = "meshtastic" + mock_connector.send_message.return_value = True + # No _by_name attribute (not a CompositeTransport). + del mock_connector._by_name + + cfg = Config() + cfg.notifications.rules = [] + cfg.notifications.cold_start_grace_seconds = 0 + t = cfg.notifications.toggles["weather"] + t.enabled = True + t.min_severity = "priority" + t.severity_channels = { + "priority": ["mesh_broadcast"], + "immediate": ["mesh_broadcast", "mesh_dm"], + } + t.broadcast_channel = 0 + t.node_ids = ["!deadbeef"] + + event = make_event( + source="nws", category="weather_warning", + severity="priority", title="weather alert", + ) + rules = _dispatch_with_connector(cfg, event, mock_connector) + assert len(rules) == 1 + assert rules[0]["delivery_type"] == "mesh_broadcast" + + # send_message called with Meshtastic channel and transport hint. + assert mock_connector.send_message.called + kwargs = mock_connector.send_message.call_args.kwargs + assert kwargs.get("transport") == "meshtastic" + assert kwargs.get("channel") == 0