From 099cec783cbd03b02d080c9346770a7fa69f3309 Mon Sep 17 00:00:00 2001 From: malice Date: Wed, 8 Jul 2026 14:42:20 -0600 Subject: [PATCH] fix(native): promote decider severity/category onto Event + make silent severity-floor drop observable (counter+log) (#97) CHANGE 1 (store.py): after applying gate.data_patch into event.data, promote _severity_override and category keys onto event.severity / event.category. Previously, decider overrides (e.g. fire: "priority" on every New/Update) landed only in event.data, leaving event.severity at the adapter's raw value ("routine" for fires >=25 km from an anchor). This silently failed the toggle/matrix min_severity floor. Native and Central now share identical broadcast decisions at the shared choke point. CHANGE 2 (dispatcher.py + v27 migration): both the toggle-path and matrix per-cell severity-floor drop paths now emit a WARN log and increment a new persisted counter (severity_floor_dropped) in dispatcher_state, following the existing drop-counter pattern exactly. v27.sql adds the column; the counter restores on restart and appears in dispatch_stats(). Co-authored-by: Matt Johnson Co-authored-by: Claude Sonnet 4.6 --- work/meshai/env/store.py | 16 + .../notifications/pipeline/dispatcher.py | 35 +- work/meshai/persistence/migrations/v27.sql | 13 + ...est_native_severity_promotion_and_floor.py | 380 ++++++++++++++++++ work/tests/test_v052_dispatcher.py | 2 +- 5 files changed, 439 insertions(+), 7 deletions(-) create mode 100644 work/meshai/persistence/migrations/v27.sql create mode 100644 work/tests/test_native_severity_promotion_and_floor.py diff --git a/work/meshai/env/store.py b/work/meshai/env/store.py index 17df10f..7e9ac0f 100644 --- a/work/meshai/env/store.py +++ b/work/meshai/env/store.py @@ -727,6 +727,22 @@ class EnvironmentalStore: return # Apply data_patch into event.data event.data.update(gate.data_patch) + # Promote decider overrides onto the Event itself, mirroring the + # Central path (central/consumer.py). Deciders stamp + # _severity_override (fire: "priority" on every New/Update) and, + # for fire New/tombstone, a category override. Merging them into + # event.data alone left event.severity at the adapter's value + # ("routine" for fires >=25 km from an anchor), which silently + # failed the toggle/matrix min_severity floors. Native and Central + # must share identical broadcast decisions. satpass/firms already + # self-consume this in-adapter; this fixes it for every other + # cut-over category at the shared choke point. + _sev_override = gate.data_patch.get("_severity_override") + if _sev_override: + event.severity = _sev_override + _cat_override = gate.data_patch.get("category") + if _cat_override: + event.category = _cat_override if gate.commit is not None: event.data["_on_broadcast_committed"] = gate.commit except Exception as _gate_exc: diff --git a/work/meshai/notifications/pipeline/dispatcher.py b/work/meshai/notifications/pipeline/dispatcher.py index 589455f..09c5a34 100644 --- a/work/meshai/notifications/pipeline/dispatcher.py +++ b/work/meshai/notifications/pipeline/dispatcher.py @@ -83,6 +83,7 @@ class Dispatcher: # from config so it can be tuned at runtime via /api/config PUT. self._first_event_at: Optional[float] = None self._cold_start_dropped = 0 + self._severity_floor_dropped = 0 # (toggle.name, category, region) -> last-fire wall-clock seconds self._toggle_cooldown: dict[tuple[str, str, str], float] = {} # Insertion-ordered (source, event.id) -> sentinel; evict oldest at cap. @@ -118,7 +119,8 @@ class Dispatcher: try: row = conn.execute( "SELECT cold_start_anchor, stale_dropped, cooldown_dropped, " - "dedup_dropped, cold_start_dropped FROM dispatcher_state WHERE id=1" + "dedup_dropped, cold_start_dropped, severity_floor_dropped " + "FROM dispatcher_state WHERE id=1" ).fetchone() except Exception: self._logger.debug( @@ -131,12 +133,13 @@ class Dispatcher: self._cooldown_dropped = int(row["cooldown_dropped"] or 0) self._dedup_dropped = int(row["dedup_dropped"] or 0) self._cold_start_dropped = int(row["cold_start_dropped"] or 0) + self._severity_floor_dropped = int(row["severity_floor_dropped"] or 0) self._logger.info( "dispatcher state restored: cold_start_anchor=%s " - "stale=%d cooldown=%d dedup=%d cold_start=%d", + "stale=%d cooldown=%d dedup=%d cold_start=%d sev_floor=%d", self._first_event_at, self._stale_dropped, self._cooldown_dropped, self._dedup_dropped, - self._cold_start_dropped, + self._cold_start_dropped, self._severity_floor_dropped, ) # Cooldowns: every row restored verbatim (the in-memory prune @@ -180,10 +183,10 @@ class Dispatcher: conn.execute( "UPDATE dispatcher_state SET cold_start_anchor=?, " "stale_dropped=?, cooldown_dropped=?, dedup_dropped=?, " - "cold_start_dropped=?, updated_at=? WHERE id=1", + "cold_start_dropped=?, severity_floor_dropped=?, updated_at=? WHERE id=1", (self._first_event_at, self._stale_dropped, self._cooldown_dropped, self._dedup_dropped, - self._cold_start_dropped, time.time()), + self._cold_start_dropped, self._severity_floor_dropped, time.time()), ) except Exception: self._logger.exception( @@ -409,6 +412,16 @@ class Dispatcher: _floor = ((_cell.get("min_severity") if isinstance(_cell, dict) else getattr(_cell, "min_severity", None)) or "routine") if event_rank < self.SEVERITY_RANK.get(_floor, 0): + self._logger.warning( + "severity-floor drop: source=%s category=%s id=%s severity=%s " + "< floor=%s (matrix ch=%s region=%s)", + event.source, event.category, event.id, event.severity, + _floor, + _cell.get("mt") if isinstance(_cell, dict) else getattr(_cell, "mt", "?"), + _mr, + ) + self._severity_floor_dropped += 1 + self._persist_state() continue # below per-cell floor for this region _mt = _cell.get("mt") if isinstance(_cell, dict) else getattr(_cell, "mt", None) _mc = _cell.get("mc") if isinstance(_cell, dict) else getattr(_cell, "mc", None) @@ -554,7 +567,16 @@ class Dispatcher: if not (set(regions) & ev_regions): return event_rank = self.SEVERITY_RANK.get(event.severity, 0) - if event_rank < self.SEVERITY_RANK.get(getattr(tog, "min_severity", "routine"), 0): + _tog_floor = getattr(tog, "min_severity", "routine") + if event_rank < self.SEVERITY_RANK.get(_tog_floor, 0): + self._logger.warning( + "severity-floor drop: source=%s category=%s id=%s severity=%s " + "< floor=%s (toggle=%s)", + event.source, event.category, event.id, event.severity, + _tog_floor, fam, + ) + self._severity_floor_dropped += 1 + self._persist_state() return # v0.16 (Integration C1) — destinations vs inline routing. # If the toggle references reusable NotificationDestinations, deliver @@ -735,6 +757,7 @@ class Dispatcher: "cooldown_dropped": self._cooldown_dropped, "dedup_dropped": self._dedup_dropped, "cold_start_dropped": self._cold_start_dropped, + "severity_floor_dropped": self._severity_floor_dropped, "cold_start_anchor_at": self._first_event_at, "cooldown_keys": len(self._toggle_cooldown), "dedup_lru_size": len(self._dedup_lru), diff --git a/work/meshai/persistence/migrations/v27.sql b/work/meshai/persistence/migrations/v27.sql new file mode 100644 index 0000000..9f04990 --- /dev/null +++ b/work/meshai/persistence/migrations/v27.sql @@ -0,0 +1,13 @@ +-- v27 add severity_floor_dropped counter to dispatcher_state. +-- +-- Prior to this migration the two severity-floor drop paths in +-- pipeline/dispatcher.py (toggle path + matrix per-cell path) silently +-- returned / continued with no log and no counter. v27 wires a WARN log +-- and this cumulative counter so floor-drops are observable in ops/health +-- and in the LLM context (env_reporter). The counter follows the same +-- write-through / restore pattern as the existing drop counters. +-- +-- The migration runner (persistence/db.py) applies each version exactly +-- once (gated on schema_meta.version), so a plain ADD COLUMN is safe. + +ALTER TABLE dispatcher_state ADD COLUMN severity_floor_dropped INTEGER NOT NULL DEFAULT 0; diff --git a/work/tests/test_native_severity_promotion_and_floor.py b/work/tests/test_native_severity_promotion_and_floor.py new file mode 100644 index 0000000..476e9a3 --- /dev/null +++ b/work/tests/test_native_severity_promotion_and_floor.py @@ -0,0 +1,380 @@ +"""Unit tests for native severity/category promotion (CHANGE 1) and +severity-floor drop observability (CHANGE 2). + +CHANGE 1 — store._emit_event() must promote decider data_patch keys + _severity_override and category onto event.severity / event.category so the + toggle/matrix min_severity floor sees the correct value, not the adapter's + raw value. + +CHANGE 2 — Dispatcher must: + (a) emit a WARN log when an event is dropped by the severity floor, and + (b) increment + persist the severity_floor_dropped counter. + Both the toggle path and the matrix path are tested. +""" +from __future__ import annotations + +import asyncio +import logging +from unittest.mock import MagicMock, AsyncMock, patch + +import pytest + +from meshai.config import ( + Config, EnvironmentalConfig, NotificationToggle, +) +from meshai.notifications.events import make_event, Event +from meshai.notifications.gating.base import GateResult +from meshai.notifications.pipeline.dispatcher import Dispatcher +from meshai.notifications.pipeline.bus import EventBus +from meshai.env.store import EnvironmentalStore +from meshai.persistence import close_thread_connection, init_db +from meshai.persistence import db as persistence_db + + +# ───────────────────────────────────────────────────────────────────────────── +# Shared fixtures +# ───────────────────────────────────────────────────────────────────────────── + + +@pytest.fixture +def db_path(tmp_path, monkeypatch): + """Isolated SQLite DB for each test.""" + p = str(tmp_path / "sev-promo-test.sqlite") + monkeypatch.setenv("MESHAI_DB_PATH", p) + persistence_db._initialised.clear() + close_thread_connection() + try: + from meshai.adapter_config import adapter_config as _ac + _ac.invalidate() + except Exception: + pass + init_db() + yield p + close_thread_connection() + persistence_db._initialised.discard(p) + + +# ───────────────────────────────────────────────────────────────────────────── +# CHANGE 1 — Promotion tests +# ───────────────────────────────────────────────────────────────────────────── + + +class _FakeAdapter: + """Minimal adapter stub for _emit_event tests.""" + + def __init__(self, source="wfigs", category="wildfire_incident", + severity="routine"): + self._source = source + self._category = category + self._severity = severity + + def to_event(self, raw_evt: dict) -> Event: + return make_event( + source=self._source, + category=self._category, + severity=self._severity, + title="Test fire", + ) + + +def _make_store_with_bus(): + """Build an EnvironmentalStore (no real adapters) + capture bus.""" + bus = EventBus() + captured: list[Event] = [] + bus.subscribe(lambda e: captured.append(e)) + store = EnvironmentalStore(EnvironmentalConfig(), event_bus=bus) + return store, captured + + +class TestSeverityPromotion: + """_emit_event promotes _severity_override + category onto the Event.""" + + def test_promotion_applied_when_override_present(self, db_path): + """A decider patch with _severity_override='priority' and + category='wildfire_declared' must be stamped onto the emitted Event.""" + store, captured = _make_store_with_bus() + adapter = _FakeAdapter(severity="routine") + raw_evt = {"source": "wfigs", "event_id": "fire_001"} + + gate = GateResult( + broadcast=True, + data_patch={ + "_severity_override": "priority", + "category": "wildfire_declared", + }, + ) + fake_decider = lambda data, *, source, now: gate + + with patch("meshai.notifications.gating.get_decider", + return_value=fake_decider), \ + patch("meshai.notifications.cutover.is_cutover", + return_value=False), \ + patch("meshai.notifications.cutover.NATIVE_ALWAYS_DECIDE", + {"wildfire_incident"}): + store._emit_event(adapter, raw_evt) + + assert len(captured) == 1, "expected exactly one emitted event" + ev = captured[0] + assert ev.severity == "priority", ( + f"severity not promoted: got {ev.severity!r}" + ) + assert ev.category == "wildfire_declared", ( + f"category not promoted: got {ev.category!r}" + ) + + def test_no_promotion_without_override(self, db_path): + """An event whose decider patch carries NO _severity_override or + category keys must leave event.severity and event.category unchanged.""" + store, captured = _make_store_with_bus() + adapter = _FakeAdapter(severity="routine", category="wildfire_incident") + raw_evt = {"source": "wfigs", "event_id": "fire_002"} + + gate = GateResult( + broadcast=True, + data_patch={"some_other_key": "value"}, # no override keys + ) + fake_decider = lambda data, *, source, now: gate + + with patch("meshai.notifications.gating.get_decider", + return_value=fake_decider), \ + patch("meshai.notifications.cutover.is_cutover", + return_value=False), \ + patch("meshai.notifications.cutover.NATIVE_ALWAYS_DECIDE", + {"wildfire_incident"}): + store._emit_event(adapter, raw_evt) + + assert len(captured) == 1 + ev = captured[0] + assert ev.severity == "routine", ( + f"severity was unexpectedly changed: got {ev.severity!r}" + ) + assert ev.category == "wildfire_incident", ( + f"category was unexpectedly changed: got {ev.category!r}" + ) + + def test_no_promotion_when_no_decider(self, db_path): + """Without a registered decider the event passes through unchanged.""" + store, captured = _make_store_with_bus() + adapter = _FakeAdapter(severity="routine", category="wildfire_incident") + raw_evt = {"source": "wfigs", "event_id": "fire_003"} + + with patch("meshai.notifications.gating.get_decider", + return_value=None): + store._emit_event(adapter, raw_evt) + + assert len(captured) == 1 + ev = captured[0] + assert ev.severity == "routine" + assert ev.category == "wildfire_incident" + + +# ───────────────────────────────────────────────────────────────────────────── +# CHANGE 2 — Floor observability: toggle path +# ───────────────────────────────────────────────────────────────────────────── + + +def _build_dispatcher_config(*, toggle_name="fire", min_severity="priority", + cold_start_grace=0): + """Config with one toggle, no region filter, floor at min_severity.""" + cfg = Config() + cfg.notifications.cold_start_grace_seconds = cold_start_grace + cfg.notifications.toggles = { + toggle_name: NotificationToggle( + name=toggle_name, enabled=True, + min_severity=min_severity, + freshness_seconds=0, # disabled — avoids stale-drop + cooldown_seconds=0, + severity_channels={ + "routine": ["mesh_broadcast"], + "priority": ["mesh_broadcast"], + "immediate": ["mesh_broadcast"], + }, + broadcast_channel=1, + ), + } + return cfg + + +def _mk_channel_factory(): + ch = MagicMock() + ch.deliver = AsyncMock(return_value=True) + return lambda rule, connector: ch + + +class TestToggleFloorObservability: + """Toggle-path severity-floor drop emits WARN and increments counter.""" + + def test_floor_drop_increments_counter(self, db_path, caplog): + """Event severity='routine' below toggle min_severity='priority' + must increment severity_floor_dropped and NOT deliver.""" + cfg = _build_dispatcher_config(min_severity="priority") + d = Dispatcher(cfg, _mk_channel_factory()) + assert d._severity_floor_dropped == 0 + + ev = make_event( + source="wfigs", category="wildfire_incident", + severity="routine", + title="Below floor test", + ) + with caplog.at_level(logging.WARNING, logger="meshai.pipeline.dispatcher"): + asyncio.run(d._dispatch_toggles(ev)) + + assert d._severity_floor_dropped == 1, ( + f"counter not incremented; got {d._severity_floor_dropped}" + ) + + def test_floor_drop_emits_warn_log(self, db_path, caplog): + """WARN log must fire when event is below the toggle min_severity floor.""" + cfg = _build_dispatcher_config(min_severity="priority") + d = Dispatcher(cfg, _mk_channel_factory()) + + ev = make_event( + source="wfigs", category="wildfire_incident", + severity="routine", + title="Below floor warn test", + ) + with caplog.at_level(logging.WARNING, logger="meshai.pipeline.dispatcher"): + asyncio.run(d._dispatch_toggles(ev)) + + warn_records = [ + r for r in caplog.records + if r.levelno == logging.WARNING and "severity-floor drop" in r.message + ] + assert warn_records, ( + "expected at least one 'severity-floor drop' WARNING log; got none. " + f"Records: {[r.message for r in caplog.records]}" + ) + + def test_floor_drop_persists_to_db(self, db_path): + """Counter write-through: severity_floor_dropped must land in dispatcher_state.""" + cfg = _build_dispatcher_config(min_severity="priority") + d = Dispatcher(cfg, _mk_channel_factory()) + + ev = make_event( + source="wfigs", category="wildfire_incident", severity="routine", + title="DB persist test", + ) + asyncio.run(d._dispatch_toggles(ev)) + assert d._severity_floor_dropped == 1 + + conn = persistence_db.get_db(db_path) + row = conn.execute( + "SELECT severity_floor_dropped FROM dispatcher_state" + ).fetchone() + assert row is not None + assert row["severity_floor_dropped"] == 1 + + def test_above_floor_does_not_increment(self, db_path): + """Event at or above the floor must NOT increment severity_floor_dropped.""" + cfg = _build_dispatcher_config(min_severity="routine") + d = Dispatcher(cfg, _mk_channel_factory()) + + ev = make_event( + source="wfigs", category="wildfire_incident", severity="routine", + title="At-floor pass-through", + ) + asyncio.run(d._dispatch_toggles(ev)) + assert d._severity_floor_dropped == 0, ( + "floor drop counter should stay 0 when event meets the floor" + ) + + def test_floor_drop_counter_survives_restart(self, db_path): + """Counter value restored on a fresh Dispatcher reading from same DB.""" + cfg = _build_dispatcher_config(min_severity="priority") + d1 = Dispatcher(cfg, _mk_channel_factory()) + + ev = make_event( + source="wfigs", category="wildfire_incident", severity="routine", + title="Restart test", + ) + asyncio.run(d1._dispatch_toggles(ev)) + assert d1._severity_floor_dropped == 1 + + # Fresh Dispatcher, same DB. + d2 = Dispatcher(cfg, _mk_channel_factory()) + assert d2._severity_floor_dropped == 1, ( + f"counter not restored after restart; got {d2._severity_floor_dropped}" + ) + + def test_dispatch_stats_includes_floor_dropped(self, db_path): + """dispatch_stats() exposes severity_floor_dropped.""" + cfg = _build_dispatcher_config(min_severity="priority") + d = Dispatcher(cfg, _mk_channel_factory()) + + ev = make_event( + source="wfigs", category="wildfire_incident", severity="routine", + title="Stats test", + ) + asyncio.run(d._dispatch_toggles(ev)) + stats = d.dispatch_stats() + assert "severity_floor_dropped" in stats + assert stats["severity_floor_dropped"] == 1 + + +# ───────────────────────────────────────────────────────────────────────────── +# CHANGE 2 — Floor observability: matrix path +# ───────────────────────────────────────────────────────────────────────────── + + +class TestMatrixFloorObservability: + """Matrix per-cell severity floor drop emits WARN and increments counter.""" + + def _cfg_with_matrix(self, *, cell_min_severity="priority") -> Config: + """Config with a fire toggle + enabled region_routes matrix. + The matrix cell for US-ID has min_severity=cell_min_severity. + The toggle-level min_severity is 'routine' so the toggle floor + passes; only the matrix per-cell floor may drop the event. + """ + from meshai.config import RegionRouteMatrix + cfg = Config() + cfg.notifications.cold_start_grace_seconds = 0 + tog = NotificationToggle( + name="fire", enabled=True, + min_severity="routine", # toggle floor passes; matrix cell drops + freshness_seconds=0, + cooldown_seconds=0, + severity_channels={ + "routine": ["mesh_broadcast"], + "priority": ["mesh_broadcast"], + }, + broadcast_channel=1, + ) + cfg.notifications.toggles = {"fire": tog} + # Matrix: enabled, one cell for fire/US-ID + cfg.notifications.region_routes = RegionRouteMatrix( + enabled=True, + cells={ + "fire": { + "US-ID": { + "enabled": True, + "min_severity": cell_min_severity, + "mt": "1", + "mc": "", + } + } + }, + ) + return cfg + + def test_matrix_floor_drop_increments_counter(self, db_path, caplog): + """Matrix per-cell floor drop (routine below priority) must + increment severity_floor_dropped.""" + cfg = self._cfg_with_matrix(cell_min_severity="priority") + d = Dispatcher(cfg, _mk_channel_factory()) + + ev = make_event( + source="wfigs", category="wildfire_incident", + severity="routine", region="US-ID", + title="Matrix floor test", + ) + with caplog.at_level(logging.WARNING, logger="meshai.pipeline.dispatcher"): + asyncio.run(d._dispatch_toggles(ev)) + + assert d._severity_floor_dropped >= 1, ( + f"matrix floor drop not counted; got {d._severity_floor_dropped}" + ) + warn_records = [ + r for r in caplog.records + if r.levelno == logging.WARNING and "severity-floor drop" in r.message + ] + assert warn_records, "expected 'severity-floor drop' WARNING from matrix path" diff --git a/work/tests/test_v052_dispatcher.py b/work/tests/test_v052_dispatcher.py index 4e6d9ee..9652b81 100644 --- a/work/tests/test_v052_dispatcher.py +++ b/work/tests/test_v052_dispatcher.py @@ -358,6 +358,6 @@ def test_dispatch_stats_exposes_all_counters(): stats = d.dispatch_stats() assert set(stats.keys()) == { "stale_dropped", "cooldown_dropped", "dedup_dropped", - "cold_start_dropped", "cold_start_anchor_at", + "cold_start_dropped", "severity_floor_dropped", "cold_start_anchor_at", "cooldown_keys", "dedup_lru_size", }