mirror of
https://github.com/zvx-echo6/meshai.git
synced 2026-08-26 17:31:34 +00:00
feat(routing): MeshCore as first-class delivery types (meshcore_broadcast/dm)
Replace the composite auto-fan with explicit per-mesh delivery types so each family independently controls broadcast/DM per severity on Meshtastic AND MeshCore. mesh_broadcast->Meshtastic only, meshcore_broadcast->MeshCore (by channel name), mesh_dm/meshcore_dm likewise; routing via the existing transport hint. Adds meshcore_dm_contacts. Meshtastic-only configs unchanged. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
24cb6a31df
commit
3b2813cf06
6 changed files with 697 additions and 81 deletions
|
|
@ -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 = ""
|
||||
|
|
|
|||
|
|
@ -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."""
|
||||
|
||||
|
|
@ -773,6 +990,14 @@ 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)
|
||||
|
|
@ -783,15 +1008,32 @@ def create_channel(rule: "NotificationRuleConfig", connector=None) -> Notificati
|
|||
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(
|
||||
|
|
@ -820,15 +1062,17 @@ def create_channel_from_dict(config: dict, connector=None) -> NotificationChanne
|
|||
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(
|
||||
|
|
|
|||
|
|
@ -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,10 +551,28 @@ 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 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")
|
||||
|
|
@ -570,17 +590,21 @@ class Dispatcher:
|
|||
severity="priority", title=text,
|
||||
)
|
||||
ev.data["_meshai_precomposed"] = True
|
||||
rule = self._toggle_to_rule(rf, "mesh_broadcast", ev)
|
||||
|
||||
delivered_any = False
|
||||
for ch_type in ch_types:
|
||||
rule = self._toggle_to_rule(rf, ch_type, 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
|
||||
"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
|
||||
|
|
@ -597,8 +621,8 @@ class Dispatcher:
|
|||
)
|
||||
except Exception:
|
||||
self._logger.exception(
|
||||
"scheduled-broadcast: audit row insert failed")
|
||||
return bool(success)
|
||||
"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", ""),
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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=<broadcast_channel> AND meshcore_channel=<name> 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"
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue