fix(gating): firms decide() spotting/halt severity key, issue #121

gating/firms.py's decide() stamped a plain "severity" key in the
data_patch for the wildfire_spotting and wildfire_halted broadcast
paths. central/consumer.py only ever promotes data["_severity_override"]
onto Event.severity -- the plain key is a silent no-op, the same class
of bug as #118 (fixed for firms_handler.py's own inline stamps in
PR #120). Currently inert (MESHAI_CUTOVER_CATEGORIES is unset by
default), but the moment wildfire_spotting/wildfire_halted are cut
over, spotting would silently stop being "immediate".

- gating/firms.py: both data_patch sites now use _severity_override.
  Checked the other gating modules (fire.py, avalanche.py, swpc.py,
  quake.py, nws.py) -- all already use _severity_override correctly;
  firms.py was the only one with the plain-key mistake.
- Fixed the stale module docstring claiming the unattributed-hotspot
  cluster path "is DEAD" -- it has been live since d479ca53 (#73); the
  stale comment directly caused a bogus bug report against production.
- test_firms_refactor.py: updated two existing tests that had codified
  the buggy plain-"severity" behavior as expected, and added
  TestCutoverSeverityReachesEvent, which drives the real cutover path
  end-to-end through CentralConsumer._normalize and asserts the
  emitted Event's severity (immediate for spotting, routine for halt).
  Verified both new tests fail against the unfixed decider and pass
  against the fix.

Full suite: 20 failed, 2242 passed, 72 skipped (vs. origin/main
baseline 20 failed, 2240 passed, 72 skipped -- same 20 pre-existing
failures, confirmed identical with this change stashed out; +2 passed
are the new regression tests).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

work/meshai/notifications/gating/firms.py

@@ -14,10 +14,11 @@ Three broadcast categories are handled, discriminated by an internal
     firms_spotting  -> wildfire_spotting (pixel outside prior perimeter, cooldown)
     firms_halt      -> wildfire_halted   (fire idle >= halt_minimum_seconds)
 
-The unattributed-hotspot cluster path is DEAD (an unconditional ``return None``
-at the top of ``_maybe_emit_cluster``) and is NOT represented here.  Native
-``env/fires.py`` hotspot broadcasts (new_ignition / wildfire_hotspot) are a
-deferred follow-up and are not migrated.
+The unattributed-hotspot cluster path is LIVE (shipped in ``d479ca53`` / #73)
+and is NOT represented here -- it is not one of the three broadcast
+categories this decider covers. Native ``env/fires.py`` hotspot broadcasts
+(new_ignition / wildfire_hotspot) are a deferred follow-up and are not
+migrated.
 
 Tier-b deferral (intended behavior change, validated by gate-sequence, NOT
 golden byte-parity):
@@ -123,10 +124,12 @@ def decide(data: dict, *, source: str, now: float) -> GateResult:
                 return GateResult(broadcast=False, lifecycle="cooldown",
                                   reason="inside spotting cooldown")
         patch = {
-            # Legacy stamps (verbatim on the not-cutover path): plain severity,
-            # NOT _severity_override.
+            # Legacy stamps (verbatim on the not-cutover path). issue #121:
+            # must be _severity_override -- consumer.py only ever honors that
+            # key; a plain "severity" key here is a silent no-op (same class
+            # of bug as #118, fixed for firms_handler.py in PR #120).
             "category": "wildfire_spotting",
-            "severity": "immediate",
+            "_severity_override": "immediate",
             # Render hints for the firms formatter (cutover path).
             "dist_mi": data.get("dist_mi"),
             "direction": data.get("direction"),
@@ -165,9 +168,12 @@ def decide(data: dict, *, source: str, now: float) -> GateResult:
         name = row["incident_name"] or "(unnamed fire)"
         hours = max(0, int((float(now) - float(row["last_pass_at"])) / 3600.0))
         patch = {
-            # Legacy stamps (verbatim on the not-cutover path).
+            # Legacy stamps (verbatim on the not-cutover path). issue #121:
+            # must be _severity_override -- consumer.py only ever honors that
+            # key; a plain "severity" key here is a silent no-op (same class
+            # of bug as #118, fixed for firms_handler.py in PR #120).
             "category": "wildfire_halted",
-            "severity": "routine",
+            "_severity_override": "routine",
             # Render hints + the identity the handler needs for the wire/latch.
             "incident_name": name,
             "hours": hours,

work/tests/test_firms_refactor.py

@@ -245,13 +245,15 @@ class TestSpottingDecideSequence:
         d.update(over)
         return d
 
-    def test_first_broadcasts_with_plain_severity(self):
+    def test_first_broadcasts_with_severity_override(self):
+        # issue #121: must be _severity_override -- consumer.py only ever
+        # honors that key, a plain "severity" key is a silent no-op.
         _seed_fire(irwin_id="ID-S", lat=43.0, lon=-115.0)
         gr = firms_decide(self._in(), source="firms", now=1_000_000.0)
         assert gr.broadcast is True
         assert gr.data_patch["category"] == "wildfire_spotting"
-        assert gr.data_patch["severity"] == "immediate"
-        assert "_severity_override" not in gr.data_patch
+        assert gr.data_patch["_severity_override"] == "immediate"
+        assert "severity" not in gr.data_patch
         assert gr.commit is not None
 
     def test_latch_not_burned_without_commit(self):
@@ -279,12 +281,15 @@ class TestSpottingDecideSequence:
 
 class TestHaltDecideSequence:
     def test_first_broadcasts_with_hours_and_severity(self):
+        # issue #121: must be _severity_override -- consumer.py only ever
+        # honors that key, a plain "severity" key is a silent no-op.
         now = 1780768800.0
         _seed_stale_fire("ID-H", now_epoch=now, idle_hours=14)
         gr = firms_decide({"_kind": "firms_halt"}, source="firms", now=now)
         assert gr.broadcast is True
         assert gr.data_patch["category"] == "wildfire_halted"
-        assert gr.data_patch["severity"] == "routine"
+        assert gr.data_patch["_severity_override"] == "routine"
+        assert "severity" not in gr.data_patch
         assert gr.data_patch["hours"] == 14
         assert gr.data_patch["incident_name"] == "Cold Fire"
         assert gr.data_patch["irwin_id"] == "ID-H"
@@ -444,3 +449,84 @@ class TestClusterBelowThreshold:
             frp=20.0, data=data, now=1780747200, this_pixel_id=1)
         assert out is None
         assert data == {}
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 6. issue #121 — cutover severity must reach the Event (regression guard)
+# ─────────────────────────────────────────────────────────────────────────────
+# Spotting and halt used to stamp a plain "severity" key in data_patch.
+# central/consumer.py only ever promotes data["_severity_override"] onto
+# Event.severity (see consumer.py's issue #118 comment) -- the plain key was
+# a silent no-op, the same class of bug fixed for firms_handler.py's own
+# inline stamps in PR #120. These drive the REAL cutover path end-to-end
+# through CentralConsumer._normalize (the actual production entry point,
+# adapter=="firms" dispatch) and assert the emitted Event's severity, so a
+# regression back to a plain "severity" key fails loudly instead of silently.
+
+class TestCutoverSeverityReachesEvent:
+    def _consumer(self):
+        from unittest.mock import MagicMock
+        from meshai.config import Config
+        from meshai.central.consumer import CentralConsumer
+        cfg = Config()
+        cfg.notifications.cold_start_grace_seconds = 0
+        return CentralConsumer(cfg.environmental, MagicMock())
+
+    def test_spotting_cutover_event_severity_is_immediate(self, monkeypatch):
+        _cutover(monkeypatch, "wildfire_spotting")
+        try:
+            _seed_pass_a_hex_then_close("ID-SEV-S", 43.0, -115.0)
+            sp_lat, sp_lon = _offset_mi(43.0, -115.0,
+                                        north_mi=2.0 / math.sqrt(2),
+                                        east_mi=2.0 / math.sqrt(2))
+            env = _envelope(lat=sp_lat, lon=sp_lon, acq_time="1800")
+            env["id"] = "spot-sev-test"
+            env["data"]["id"] = "spot-sev-test"
+            env["data"]["geo"] = {"centroid": [sp_lon, sp_lat]}
+
+            event = self._consumer()._normalize(_SUBJECT, env)
+            assert event is not None, "expected a broadcast Event, got None"
+            assert event.severity == "immediate", (
+                f"issue #121 regression: expected 'immediate', got "
+                f"{event.severity!r} -- gating.firms.decide's spotting "
+                f"data_patch must use _severity_override, not the plain "
+                f"'severity' key"
+            )
+        finally:
+            from meshai.notifications.cutover import _clear_cache
+            _clear_cache()
+
+    def test_halt_cutover_event_severity_is_routine(self, monkeypatch):
+        _cutover(monkeypatch, "wildfire_halted")
+        try:
+            from meshai.central import firms_handler as _fh
+            fixed_now = 1780768800
+            monkeypatch.setattr(_fh.time, "time", lambda: float(fixed_now))
+            _seed_stale_fire("ID-SEV-H", now_epoch=fixed_now, idle_hours=14)
+
+            # Any unrelated pixel arrival opportunistically triggers the halt
+            # scan (_maybe_emit_halt runs on every pixel as a fallback); use
+            # one far away so it isn't attributed to (and doesn't grow) the
+            # stale fire instead.
+            env = _envelope(lat=10.0, lon=10.0, acq_time="1800")
+            env["id"] = "halt-sev-test"
+            env["data"]["id"] = "halt-sev-test"
+            env["data"]["geo"] = {"centroid": [10.0, 10.0]}
+            # Raw envelope severity maps to "immediate" (>= immediate_min=3),
+            # deliberately NOT "routine" -- so the assertion below can only
+            # pass if the halt data_patch's _severity_override actually
+            # overrides it down to "routine". A plain "severity" key (the
+            # bug) would silently leave this at "immediate" instead.
+            env["data"]["severity"] = 3
+
+            event = self._consumer()._normalize(_SUBJECT, env)
+            assert event is not None, "expected a broadcast Event, got None"
+            assert event.severity == "routine", (
+                f"issue #121 regression: expected 'routine', got "
+                f"{event.severity!r} -- gating.firms.decide's halt "
+                f"data_patch must use _severity_override, not the plain "
+                f"'severity' key"
+            )
+        finally:
+            from meshai.notifications.cutover import _clear_cache
+            _clear_cache()