fix(activity-log): stamp source_event_table for native adapter broadcasts + UI label fallback + backfill recent orphans (#99)

Backend: add _SOURCE_TO_TABLE class constant in Dispatcher mapping event.source
("nws", "nifc", "wzdx", "traffic", "511") to canonical audit table names.
_post_broadcast_commit now falls back to this map when _broadcast_audit is
absent/None, so native env adapter sends (nws.py, fires.py, wzdx.py,
roads511.py, traffic.py) write a non-NULL source_event_table instead of NULL.
Existing _broadcast_audit paths (Central handlers, scheduled broadcasts) are
unchanged.

Frontend: replace naive familyLabel() string transform with explicit
TABLE_LABELS lookup (10 known tables → friendly names) plus a TEXT_HINTS
emoji-prefix heuristic for legacy NULL-source rows, so historical orphan rows
still display a meaningful label before/after backfill.

Tests: three new unit tests in test_dispatcher_persistence.py covering the
fallback path (nws→nws_alerts), the full _SOURCE_TO_TABLE map, and that
explicit _broadcast_audit is never overridden by the fallback.

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

work/dashboard-frontend/src/pages/ActivityLog.tsx

@@ -40,11 +40,49 @@ function channelLabel(channel: string | number | null): string {
   return channel.startsWith('#') ? channel : `#${channel}`
 }
 
-// Type/family tag derived from source_event_table (e.g. 'fires' -> 'fire').
+// Explicit label table for known source_event_table values.
-function familyLabel(table: string | null): string {
+const TABLE_LABELS: Record<string, string> = {
-  if (!table) return 'broadcast'
+  nws_alerts:                 'Weather',
-  const t = table.replace(/_/g, ' ').trim()
+  fires:                      'Fire',
-  return t.endsWith('s') ? t.slice(0, -1) : t
+  fire_digest_broadcasts:     'Fire digest',
+  satpass_events:             'Satellite',
+  band_conditions_broadcasts: 'Band',
+  traffic_events:             'Traffic',
+  quake_events:               'Quake',
+  swpc_events:                'Space Wx',
+  gauge_readings:             'Hydro',
+  event_log:                  'Avalanche',
+}
+
+// Text-prefix / emoji heuristics for legacy NULL-source rows.
+// Derived from actual formatter outputs:
+//   fires.py / firms.py  → "🔥 …"
+//   work_zone.py (wzdx/511/traffic) → "🚧 …"
+//   avalanche.py         → "⛷ …"
+//   hydro.py             → "🌊 …"
+//   swpc.py              → "🧲 …" | "☀️ …"
+//   quake.py             → "🌐 …"
+const TEXT_HINTS: Array<[string, string]> = [
+  ['🔥', 'Fire'],
+  ['🚧', 'Traffic'],
+  ['⛷', 'Avalanche'],
+  ['🌊', 'Hydro'],
+  ['🧲', 'Space Wx'],
+  ['☀️', 'Space Wx'],
+  ['🌐', 'Quake'],
+]
+
+// Type/family tag derived from source_event_table, with text-prefix fallback
+// for legacy NULL-source rows (native adapter broadcasts before the fix).
+function familyLabel(table: string | null, text?: string | null): string {
+  if (table) return TABLE_LABELS[table] ?? table.replace(/_/g, ' ').replace(/s$/, '')
+  // NULL source_event_table: try emoji prefix heuristic on the broadcast text.
+  if (text) {
+    for (const [prefix, label] of TEXT_HINTS) {
+      if (text.startsWith(prefix)) return label
+    }
+  }
+  return 'broadcast'
 }
 
 // --- component -------------------------------------------------------------
@@ -193,7 +231,7 @@ export default function ActivityLog() {
                           {channelLabel(e.channel)}
                         </span>
                         <span className="text-xs px-2 py-0.5 rounded-full bg-[#f59e0b]/10 text-[#f59e0b]">
-                          {familyLabel(e.source_event_table)}
+                          {familyLabel(e.source_event_table, e.text)}
                         </span>
                         {e.success === 1 && (
                           <span className="text-xs text-green-500">Sent</span>

work/meshai/notifications/pipeline/dispatcher.py

@@ -906,6 +906,17 @@ class Dispatcher:
         {"mesh_broadcast", "meshcore_broadcast", "mesh_dm", "meshcore_dm"}
     )
 
+    # Fallback: map event.source → canonical audit table for native env
+    # adapters that do not stamp _broadcast_audit on their events.
+    # Only include mappings confirmed against the adapters + schema.
+    _SOURCE_TO_TABLE: dict = {
+        "nws":     "nws_alerts",
+        "nifc":    "fires",
+        "wzdx":    "traffic_events",
+        "traffic": "traffic_events",
+        "511":     "traffic_events",
+    }
+
     def _post_broadcast_commit(self, event, payload, rule, ch_type: str,
                                *, success: bool = True) -> None:
         """Persistence side-effects of a per-mesh broadcast delivery.
@@ -934,6 +945,17 @@ class Dispatcher:
         # --- Audit row (always, for any mesh delivery attempt) ---
         if ch_type in self._MESH_CH_TYPES:
             audit = data.get("_broadcast_audit") if data else None
+            # Resolve audit table/pk: use handler-stamped _broadcast_audit when
+            # present; otherwise fall back to _SOURCE_TO_TABLE keyed by
+            # event.source (native env adapters don't stamp _broadcast_audit).
+            if isinstance(audit, dict):
+                audit_table = audit.get("table")
+                audit_pk = audit.get("pk")
+            else:
+                audit_table = self._SOURCE_TO_TABLE.get(
+                    getattr(event, "source", "") or ""
+                )
+                audit_pk = None
             try:
                 from meshai.persistence import get_db
                 conn = get_db()
@@ -947,8 +969,8 @@ class Dispatcher:
                     "VALUES (?,?,?,?,?,?,?,?,?,?)",
                     (
                         int(committed_at), recipient, channel, text,
-                        audit.get("table") if isinstance(audit, dict) else None,
+                        audit_table,
-                        audit.get("pk") if isinstance(audit, dict) else None,
+                        audit_pk,
                         bytes_sent, 0,
                         transport, 1 if success else 0,
                     ),

work/tests/test_dispatcher_persistence.py

@@ -501,3 +501,114 @@ def test_dispatcher_construct_when_db_unavailable(monkeypatch, tmp_path):
     d = Dispatcher(cfg, factory)   # must not raise
     assert d._stale_dropped == 0
     assert d._first_event_at is None
+
+
+# ============================================================================
+# _SOURCE_TO_TABLE fallback — native adapter events without _broadcast_audit
+# ============================================================================
+
+def test_source_to_table_fallback_stamps_audit_row(db_path):
+    """_post_broadcast_commit must write source_event_table via _SOURCE_TO_TABLE
+    when the event has no _broadcast_audit.  Covers the core Fix 1 regression
+    guard: event.source='nws' → source_event_table='nws_alerts'.
+    """
+    from unittest.mock import MagicMock
+    from meshai.notifications.events import make_event
+    from meshai.notifications.pipeline.dispatcher import Dispatcher
+    from meshai.persistence import get_db
+
+    cfg = _build_config(cold_start_grace=0)
+    factory, _ = _mk_channel_factory()
+    d = Dispatcher(cfg, factory)
+
+    # Build a native NWS event — no _broadcast_audit, no data dict at all.
+    ev = make_event(
+        source="nws",
+        category="weather_alert",
+        severity="priority",
+        region="US-ID",
+        title="⚠️ Winter Weather Advisory",
+        lat=43.6, lon=-116.2,
+    )
+
+    # Fake a minimal rule + payload for the internal commit call.
+    rule = MagicMock()
+    rule.broadcast_channel = 1
+    rule.delivery_types = ["mesh_broadcast"]
+
+    payload = MagicMock()
+    payload.message = "⚠️ Winter Weather Advisory — test"
+
+    # Call _post_broadcast_commit directly — does NOT transmit anything.
+    d._post_broadcast_commit(ev, payload, rule, "mesh_broadcast", success=True)
+
+    conn = get_db()
+    row = conn.execute(
+        "SELECT source_event_table, source_event_pk, transport "
+        "FROM mesh_broadcasts_out ORDER BY id DESC LIMIT 1"
+    ).fetchone()
+    assert row is not None, "No audit row was written"
+    assert row["source_event_table"] == "nws_alerts", (
+        f"Expected 'nws_alerts', got {row['source_event_table']!r}"
+    )
+    assert row["source_event_pk"] is None, "pk should be NULL for fallback path"
+
+
+def test_source_to_table_fallback_all_native_sources(db_path):
+    """Verify _SOURCE_TO_TABLE covers all five native adapter sources."""
+    from meshai.notifications.pipeline.dispatcher import Dispatcher
+    expected = {
+        "nws":     "nws_alerts",
+        "nifc":    "fires",
+        "wzdx":    "traffic_events",
+        "traffic": "traffic_events",
+        "511":     "traffic_events",
+    }
+    cfg = _build_config()
+    factory, _ = _mk_channel_factory()
+    d = Dispatcher(cfg, factory)
+    assert d._SOURCE_TO_TABLE == expected, (
+        f"Map mismatch: {d._SOURCE_TO_TABLE!r}"
+    )
+
+
+def test_broadcast_audit_present_not_overridden(db_path):
+    """When _broadcast_audit IS stamped, _post_broadcast_commit must use its
+    table/pk, NOT the _SOURCE_TO_TABLE fallback."""
+    from unittest.mock import MagicMock
+    from meshai.notifications.events import make_event
+    from meshai.notifications.pipeline.dispatcher import Dispatcher
+    from meshai.persistence import get_db
+
+    cfg = _build_config(cold_start_grace=0)
+    factory, _ = _mk_channel_factory()
+    d = Dispatcher(cfg, factory)
+
+    ev = make_event(
+        source="nws",
+        category="weather_alert",
+        severity="priority",
+        region="US-ID",
+        title="⚠️ Test explicit audit",
+        lat=43.6, lon=-116.2,
+    )
+    # Stamp explicit audit (as Central handlers do).
+    ev.data["_broadcast_audit"] = {"table": "nws_alerts", "pk": "CAP-XYZ-001"}
+
+    rule = MagicMock()
+    rule.broadcast_channel = 1
+    rule.delivery_types = ["mesh_broadcast"]
+    payload = MagicMock()
+    payload.message = "⚠️ Test explicit audit"
+
+    d._post_broadcast_commit(ev, payload, rule, "mesh_broadcast", success=True)
+
+    conn = get_db()
+    row = conn.execute(
+        "SELECT source_event_table, source_event_pk "
+        "FROM mesh_broadcasts_out ORDER BY id DESC LIMIT 1"
+    ).fetchone()
+    assert row["source_event_table"] == "nws_alerts"
+    assert row["source_event_pk"] == "CAP-XYZ-001", (
+        f"Expected pk='CAP-XYZ-001', got {row['source_event_pk']!r}"
+    )