From 40fcbf88e85730ee846bfa4506a002d1d432d68f Mon Sep 17 00:00:00 2001 From: malice Date: Wed, 8 Jul 2026 16:41:12 -0600 Subject: [PATCH] fix(activity-log): stamp source_event_table for native adapter broadcasts + UI label fallback + backfill recent orphans (#99) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Co-authored-by: Claude Sonnet 4.6 --- .../src/pages/ActivityLog.tsx | 50 +++++++- .../notifications/pipeline/dispatcher.py | 26 +++- work/tests/test_dispatcher_persistence.py | 111 ++++++++++++++++++ 3 files changed, 179 insertions(+), 8 deletions(-) diff --git a/work/dashboard-frontend/src/pages/ActivityLog.tsx b/work/dashboard-frontend/src/pages/ActivityLog.tsx index b939b4f..813b767 100644 --- a/work/dashboard-frontend/src/pages/ActivityLog.tsx +++ b/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'). -function familyLabel(table: string | null): string { - if (!table) return 'broadcast' - const t = table.replace(/_/g, ' ').trim() - return t.endsWith('s') ? t.slice(0, -1) : t +// Explicit label table for known source_event_table values. +const TABLE_LABELS: Record = { + nws_alerts: 'Weather', + fires: 'Fire', + 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)} - {familyLabel(e.source_event_table)} + {familyLabel(e.source_event_table, e.text)} {e.success === 1 && ( Sent diff --git a/work/meshai/notifications/pipeline/dispatcher.py b/work/meshai/notifications/pipeline/dispatcher.py index 09c5a34..dc1b569 100644 --- a/work/meshai/notifications/pipeline/dispatcher.py +++ b/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.get("pk") if isinstance(audit, dict) else None, + audit_table, + audit_pk, bytes_sent, 0, transport, 1 if success else 0, ), diff --git a/work/tests/test_dispatcher_persistence.py b/work/tests/test_dispatcher_persistence.py index b281802..e78a431 100644 --- a/work/tests/test_dispatcher_persistence.py +++ b/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}" + )