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 <mj@k7zvx.com>
Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

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:

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),

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;

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"

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",
     }