fix(region-routing): audit every broadcast (matrix+toggle); preserve cell channel through send queue; fix /api/channels for composite transport (#94)

Defect A (audit gap): _post_broadcast_commit only wrote a mesh_broadcasts_out row
when event.data["_broadcast_audit"] was a dict. Native traffic/weather/roads events
from native adapters never set that key, so every matrix-dispatched send was
invisible in the audit table even when the dispatcher logged success. Fix: write
the audit row for every mesh delivery attempt (ch_type in _MESH_CH_TYPES),
unconditionally. source_event_table/source_event_pk come from _broadcast_audit when
present, else NULL (best-effort). The early-return on empty data is preserved only
for the _on_broadcast_committed callback, not for the audit write.

Defect B (channel routing): full trace of the send path confirms the channel IS
correctly threaded from the matrix cell through _toggle_to_rule (mt_override) →
create_channel(channel_index=rule.broadcast_channel) → MeshBroadcastChannel
(self._channel) → send_message_async(channel=self._channel) → CompositeTransport
→ MeshtasticTransport send_queue job closure → _blocking_mt_send(channel) →
sendText(channelIndex=channel). No code bug: the correct channel index reaches the
radio. The missing audit rows (Defect A) prevented confirming this from the DB.

/api/channels fix: the endpoint read connector._interface which does not exist on
CompositeTransport (only on bare MeshtasticTransport), so it always returned []
when MeshCore was also configured. Fix: detect CompositeTransport and route to
meshtastic_child()._interface instead.

Tests added: matrix send without _broadcast_audit writes audit row with correct
channel+transport+success; failed delivery writes success=0 row; matrix cell
channel index reaches _blocking_mt_send end-to-end (queue path exercised);
/api/channels returns real channel list via CompositeTransport.

Co-authored-by: Matt Johnson <mj@k7zvx.com>
Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

work/meshai/dashboard/api/mesh_routes.py

@@ -367,14 +367,28 @@ async def get_edges(request: Request):
 
 @router.get("/channels")
 async def get_channels(request: Request):
-    """Get radio channels from the connected Meshtastic interface."""
+    """Get radio channels from the connected Meshtastic interface.
+
+    Works with both a bare MeshtasticTransport (connector._interface) and a
+    CompositeTransport (connector.meshtastic_child()._interface). The previous
+    implementation read ``connector._interface`` directly, which does not exist
+    on CompositeTransport, causing it to always return [] when MeshCore is also
+    configured.
+    """
     connector = getattr(request.app.state, "connector", None)
 
     if not connector or not connector.connected:
         return []
 
     try:
-        interface = connector._interface
+        # Resolve the Meshtastic child for CompositeTransport; fall back to the
+        # connector itself for bare MeshtasticTransport.
+        from meshai.transport.composite_transport import CompositeTransport
+        if isinstance(connector, CompositeTransport):
+            mt = connector.meshtastic_child()
+        else:
+            mt = connector
+        interface = getattr(mt, "_interface", None) if mt is not None else None
         if not interface or not hasattr(interface, "localNode"):
             return []
 

work/meshai/notifications/pipeline/dispatcher.py

@@ -878,17 +878,26 @@ class Dispatcher:
         # Unknown / non-mesh: leave transport NULL, fall back to legacy channel.
         return None, getattr(rule, "broadcast_channel", None), "broadcast"
 
+    # Mesh channel types that get a mesh_broadcasts_out audit row.
+    _MESH_CH_TYPES = frozenset(
+        {"mesh_broadcast", "meshcore_broadcast", "mesh_dm", "meshcore_dm"}
+    )
+
     def _post_broadcast_commit(self, event, payload, rule, ch_type: str,
                                *, success: bool = True) -> None:
         """Persistence side-effects of a per-mesh broadcast delivery.
 
         Called ONCE PER MESH CHANNEL (one per delivery_type family), so a
         broadcast that fans to both meshes writes TWO mesh_broadcasts_out
-        rows -- each carrying its own `transport` + `success` flag. The row
+        rows -- each carrying its own `transport` + `success` flag.
-        is written whenever the handler signalled it wants an audit trail
+
-        via `event.data["_broadcast_audit"]`, REGARDLESS of success, so a
+        A row is written for EVERY mesh delivery attempt (ch_type in
-        skip/failure (e.g. MeshCore channel-not-found -> deliver()==False)
+        _MESH_CH_TYPES), REGARDLESS of success and REGARDLESS of whether
-        is still visible as success=0.
+        the handler stamped `_broadcast_audit` on the event. Handlers that
+        do stamp it supply `source_event_table`/`source_event_pk`; native
+        events that do not have those fields NULL (best-effort). This makes
+        region-routed traffic/weather/roads sends visible in the audit even
+        though those native adapters do not set `_broadcast_audit`.
 
         The handler-supplied `_on_broadcast_committed` callback (which
         refreshes last_broadcast_* bookkeeping) fires ONLY when the send
@@ -897,12 +906,11 @@ class Dispatcher:
         dispatch for sibling toggles.
         """
         data = getattr(event, "data", None) or {}
-        if not data:
-            return
         committed_at = time.time()
 
-        audit = data.get("_broadcast_audit")
+        # --- Audit row (always, for any mesh delivery attempt) ---
-        if isinstance(audit, dict):
+        if ch_type in self._MESH_CH_TYPES:
+            audit = data.get("_broadcast_audit") if data else None
             try:
                 from meshai.persistence import get_db
                 conn = get_db()
@@ -916,7 +924,8 @@ class Dispatcher:
                     "VALUES (?,?,?,?,?,?,?,?,?,?)",
                     (
                         int(committed_at), recipient, channel, text,
-                        audit.get("table"), audit.get("pk"),
+                        audit.get("table") if isinstance(audit, dict) else None,
+                        audit.get("pk") if isinstance(audit, dict) else None,
                         bytes_sent, 0,
                         transport, 1 if success else 0,
                     ),
@@ -924,10 +933,14 @@ class Dispatcher:
             except Exception:
                 self._logger.exception(
                     "post-broadcast: mesh_broadcasts_out insert failed "
-                    "(table=%s pk=%s)",
+                    "(ch_type=%s event_id=%s)",
-                    audit.get("table"), audit.get("pk"),
+                    ch_type, getattr(event, "id", None),
                 )
 
+        # --- Handler callback (only on success, only when data present) ---
+        if not data:
+            return
+
         if not success:
             # A failed/skipped send is audited above but must NOT arm the
             # handler's last_broadcast_* bookkeeping.

work/tests/test_region_routing_dispatcher.py

@@ -438,3 +438,229 @@ def test_restart_dedup_key_matches_boot_restore_form():
         "matrix must dedup against the restored 2-tuple key; a re-broadcast "
         "here is the restart-flood bug"
     )
+
+
+# ============================================================ Defect A tests
+# Every matrix broadcast attempt writes a mesh_broadcasts_out row even when
+# the event has no _broadcast_audit stamp (native traffic/weather events).
+
+
+def test_matrix_broadcast_writes_audit_row_without_broadcast_audit():
+    """Defect A fix: a native event without _broadcast_audit must still create
+    a mesh_broadcasts_out row recording the channel, transport, and success."""
+    from meshai.persistence import get_db
+
+    cfg = _base_cfg(fam="roads", cooldown_s=0, min_severity="routine")
+    cfg.notifications.region_routes = _rr(cells={
+        "roads": {
+            "SW Idaho": {"mt": 3, "mc": None, "min_severity": "routine", "enabled": True},
+        }
+    })
+
+    # Build a native-style event with NO _broadcast_audit on its data.
+    from meshai.notifications.events import make_event
+    ev = make_event(
+        source="wzdx", category="work_zone",
+        severity="immediate", title="Lane closure on I-84",
+    )
+    ev.region = "SW Idaho"
+    # No ev.data["_broadcast_audit"] set — this is the native adapter case.
+    assert "_broadcast_audit" not in (ev.data or {}), \
+        "precondition: native event must not have _broadcast_audit"
+
+    d, rec = _make_dispatcher(cfg)
+    asyncio.run(d.dispatch(ev))
+
+    assert len(rec) == 1, "delivery should have succeeded"
+
+    # Defect A fix: the audit row must now exist even without _broadcast_audit.
+    conn = get_db()
+    rows = conn.execute(
+        "SELECT channel, transport, success FROM mesh_broadcasts_out"
+    ).fetchall()
+    assert len(rows) == 1, (
+        f"expected 1 mesh_broadcasts_out row (got {len(rows)}); "
+        "Defect A: native events without _broadcast_audit were never audited"
+    )
+    row = rows[0]
+    assert row["channel"] == 3, \
+        f"audit row must record the matrix cell's channel (3), got {row['channel']}"
+    assert row["transport"] == "meshtastic", \
+        f"audit row transport must be 'meshtastic', got {row['transport']}"
+    assert row["success"] == 1, \
+        f"audit row success must be 1 (delivery succeeded), got {row['success']}"
+
+
+def test_matrix_broadcast_audit_row_on_failure():
+    """Defect A fix: a failed matrix delivery must still write a success=0 row."""
+    from meshai.persistence import get_db
+
+    cfg = _base_cfg(fam="roads", cooldown_s=0, min_severity="routine")
+    cfg.notifications.region_routes = _rr(cells={
+        "roads": {
+            "SW Idaho": {"mt": 5, "mc": None, "min_severity": "routine", "enabled": True},
+        }
+    })
+
+    from meshai.notifications.events import make_event
+    ev = make_event(
+        source="itd_511", category="road_closure",
+        severity="immediate", title="Road closed",
+    )
+    ev.region = "SW Idaho"
+
+    # Dispatcher that always returns False from deliver().
+    d, rec = _make_dispatcher(cfg, succeed=False)
+    asyncio.run(d.dispatch(ev))
+
+    conn = get_db()
+    rows = conn.execute(
+        "SELECT channel, transport, success FROM mesh_broadcasts_out"
+    ).fetchall()
+    assert len(rows) == 1, "failed delivery must still create an audit row"
+    assert rows[0]["channel"] == 5
+    assert rows[0]["success"] == 0, "failed delivery must record success=0"
+
+
+# ============================================================ Defect B tests
+# The matrix cell's channel index must survive the full dispatcher→channel→
+# connector→send_queue→blocking_send chain unmodified.
+
+
+def test_matrix_cell_channel_reaches_connector_send():
+    """Defect B verification: the channel index from the matrix cell (3) must
+    be the exact value passed to connector.send_message_async, not the toggle
+    default (0) or any other value.
+
+    Uses a real MeshtasticTransport (queue path) with a patched
+    _blocking_mt_send so no radio is required. Asserts that the channel arg
+    arriving at _blocking_mt_send equals the matrix cell's mt value.
+    """
+    import asyncio
+    from unittest.mock import patch
+    from meshai.config import Config, RegionRouteMatrix, ConnectionConfig
+    from meshai.connector import MeshtasticTransport
+    from meshai.notifications.channels import create_channel
+    from meshai.notifications.pipeline.dispatcher import Dispatcher
+    from meshai.transport.send_queue import RadioSendQueue
+
+    # Build a MeshtasticTransport with a real send queue so the full
+    # enqueue → drain → _blocking_mt_send path is exercised.
+    conn_cfg = ConnectionConfig(meshtastic_send_pacing_seconds=0.05)
+    mt = MeshtasticTransport(conn_cfg)
+    mt._connected = True  # satisfy connected check inside deliver
+
+    loop = asyncio.new_event_loop()
+    pacing_fn = lambda: max(0.25, getattr(conn_cfg, "meshtastic_send_pacing_seconds", 2.0))
+    mt._mt_queue = RadioSendQueue(pacing_fn=pacing_fn)
+    mt._mt_queue.start(loop)
+    mt._loop = loop
+
+    sent_channels: list = []
+
+    def _fake_blocking_send(text, destination, channel):
+        sent_channels.append(channel)
+        return True
+
+    cfg = Config()
+    cfg.notifications.rules = []
+    cfg.notifications.cold_start_grace_seconds = 0
+    t = cfg.notifications.toggles["roads"]
+    t.enabled = True
+    t.min_severity = "routine"
+    t.severity_channels = {"routine": ["mesh_broadcast"], "immediate": ["mesh_broadcast"]}
+    t.freshness_seconds = 0
+    t.cooldown_seconds = 0
+    t.broadcast_channel = 0   # toggle default — must NOT reach the radio
+    cfg.notifications.region_routes = RegionRouteMatrix(
+        enabled=True,
+        cells={"roads": {"SW Idaho": {"mt": 3, "mc": None,
+                                       "min_severity": "routine", "enabled": True}}},
+    )
+
+    from meshai.notifications.events import make_event
+    ev = make_event(source="wzdx", category="work_zone",
+                    severity="immediate", title="Test")
+    ev.region = "SW Idaho"
+
+    with patch.object(mt, "_blocking_mt_send", side_effect=_fake_blocking_send):
+        d = Dispatcher(cfg, create_channel, connector=mt)
+        loop.run_until_complete(d.dispatch(ev))
+
+    loop.run_until_complete(mt._mt_queue.stop())
+    loop.close()
+
+    assert sent_channels, "send must have been called at least once"
+    assert all(ch == 3 for ch in sent_channels), (
+        f"ALL sends must use channel 3 (the matrix cell's mt value), "
+        f"got channels: {sent_channels}. "
+        f"If any value is 0 it means the toggle default leaked through instead "
+        f"of the matrix cell's channel."
+    )
+
+
+# ============================================================ /api/channels test
+# GET /api/channels must work on a CompositeTransport (not just bare MT).
+
+
+def test_get_channels_composite_transport():
+    """Defect: /api/channels read connector._interface which does not exist on
+    CompositeTransport → always returned []. Fix: resolve the meshtastic child
+    first via connector.meshtastic_child()."""
+    from unittest.mock import MagicMock
+    from fastapi import FastAPI
+    from fastapi.testclient import TestClient
+    from meshai.dashboard.api.mesh_routes import router
+    from meshai.transport.composite_transport import CompositeTransport
+    from meshai.connector import MeshtasticTransport
+    from meshai.config import ConnectionConfig
+
+    # Build a real CompositeTransport wrapping a fake MeshtasticTransport.
+    mt = MagicMock(spec=MeshtasticTransport)
+    mt.transport_name = "meshtastic"
+    mt.connected = True
+
+    # Fake localNode with three channels.
+    def _make_ch(idx, name, role):
+        ch = MagicMock()
+        ch.index = idx
+        ch.role = role
+        s = MagicMock()
+        s.name = name
+        ch.settings = s
+        return ch
+
+    fake_node = MagicMock()
+    fake_node.channels = [
+        _make_ch(0, "LongFast", 1),   # PRIMARY
+        _make_ch(1, "", 0),            # DISABLED
+        _make_ch(3, "SWI Alerts", 2), # SECONDARY
+    ]
+    fake_interface = MagicMock()
+    fake_interface.localNode = fake_node
+    mt._interface = fake_interface
+
+    composite = CompositeTransport([mt])
+
+    app = FastAPI()
+    app.include_router(router, prefix="/api")
+    app.state.connector = composite
+
+    client = TestClient(app)
+    r = client.get("/api/channels")
+    assert r.status_code == 200
+    data = r.json()
+
+    assert len(data) == 3, f"expected 3 channel entries, got {data}"
+    by_idx = {c["index"]: c for c in data}
+
+    assert by_idx[0]["name"] == "LongFast"
+    assert by_idx[0]["role"] == "PRIMARY"
+    assert by_idx[0]["enabled"] is True
+
+    assert by_idx[1]["role"] == "DISABLED"
+    assert by_idx[1]["enabled"] is False
+
+    assert by_idx[3]["name"] == "SWI Alerts"
+    assert by_idx[3]["role"] == "SECONDARY"
+    assert by_idx[3]["enabled"] is True