fix(firms): repair the FIRMS fire-fusion Event contract (issues #117-#119) (#120)

Three independent bugs kept firms_handler's growth/spotting/halt/cluster
fusion decisions from reaching a correct mesh Event:

- #117: consumer._normalize() computed `category` from the raw Central
  category BEFORE the per-adapter handler ran and never re-read
  data["category"] afterward, so every firms_handler category stamp was a
  silent no-op. Now re-read post-dispatch, validated against the known
  category registry (unrecognized overrides are logged and ignored).

- #118: consumer.py only ever honors data["_severity_override"], but
  firms_handler's halt/spotting/cluster sites stamped the plain
  data["severity"] key instead (only growth used the right key). Switched
  all three sites to `_severity_override` for one consistent contract.
  This is severity plumbing only -- it does not change which events fire.

- #119: FirePacer's gate only matched source in ("fires","wfigs") at
  severity=="priority", so FIRMS fusion broadcasts (source="firms",
  growth/spotting at "immediate") never reached the pacer. Broadened the
  gate to cover "firms" + {"priority","immediate"}, and gave FirePacer
  head-of-line insertion so an "immediate" event is never stuck behind
  already-queued "priority" events. Still unbounded/never-drops.

Cluster detection is left exactly as main ships it: live, always on, no
toggle (PR #73's curated new-fire cluster broadcasts with cold-start
silent-seeding). Only its severity-override key changes, under #118.

Updated existing tests that asserted the old (buggy) data["severity"]
contract, and added tests/test_firms_fusion_event_contract.py covering
all three fixes end-to-end through consumer._normalize()/_handle() and
FirePacer directly.

Co-authored-by: Matt Johnson <mj@k7zvx.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

work/meshai/central/consumer.py

@@ -21,7 +21,7 @@ from typing import Optional
 from meshai.adapter_config import adapter_config
 
 from meshai.notifications.events import Event, make_event
-from meshai.notifications.categories import get_category
+from meshai.notifications.categories import get_category, ALERT_CATEGORIES
 
 logger = logging.getLogger("meshai.central.consumer")
 
@@ -645,7 +645,30 @@ class CentralConsumer:
         source = CENTRAL_ADAPTER_TO_SOURCE.get(raw_adapter, raw_adapter)
         if source != raw_adapter:
             logger.debug("Central adapter %r -> meshai source %r", raw_adapter, source)
-        # v0.6-3c: use handler severity override if present
+
+        # v0.7-fire-fusion (issue #117): a per-adapter handler (e.g.
+        # firms_handler's growth / cluster / halt / spotting fusion) may stamp
+        # a fully-resolved category onto data["category"] AFTER `category` was
+        # computed above from the raw Central category string. Re-read it
+        # here, now that the handler has run, so the override actually reaches
+        # make_event() -- previously this local `category` was captured before
+        # dispatch and the handler's stamp was a silent no-op. An unrecognized
+        # override is logged and the original mapped category is kept, rather
+        # than trusting an arbitrary string into the category taxonomy.
+        cat_override = data.get("category") if isinstance(data, dict) else None
+        if cat_override and cat_override != category:
+            if cat_override in ALERT_CATEGORIES:
+                category = cat_override
+            else:
+                logger.warning(
+                    "consumer: handler set unrecognized category override %r "
+                    "for adapter=%s; keeping %r",
+                    cat_override, raw_adapter, category)
+
+        # v0.6-3c / issue #118: use handler severity override if present.
+        # `_severity_override` is the ONLY key honored here -- handlers must
+        # stamp `_severity_override`, not the plain `severity` key, or their
+        # severity choice silently never reaches the Event.
         sev_override = data.get("_severity_override") if isinstance(data, dict) else None
         return make_event(
             source=source,
@@ -690,10 +713,16 @@ class CentralConsumer:
             if irwin_id and event.source in ("fires", "wfigs"):
                 self._drain_irwin_ids.add(irwin_id)
         elif self._bus is not None:
-            # Normal mode: route fire events through pacer, others direct
+            # Normal mode: route fire events through pacer, others direct.
+            # Issue #119: FIRMS-synthesized broadcasts (growth/spotting/halt/
+            # cluster) carry source="firms" and growth+spotting use severity
+            # "immediate" -- neither was covered by the old "fires"/"wfigs" +
+            # "priority"-only gate, so FIRMS fusion events skipped the pacer
+            # entirely. Broaden to every fire-family source at priority OR
+            # immediate severity so nothing bypasses the <=1/min throttle.
             if (self._pacer is not None
-                    and event.source in ("fires", "wfigs")
+                    and event.source in ("fires", "wfigs", "firms")
-                    and (event.data or {}).get("_severity_override") == "priority"):
+                    and event.severity in ("priority", "immediate")):
                 self._pacer.enqueue(event)
             else:
                 self._bus.emit(event)

work/meshai/central/firms_handler.py

@@ -676,8 +676,10 @@ def _maybe_emit_cluster(conn, *, lat, lon, acq_epoch, frp, data, now,
     # broadcast under unattributed_hotspot_cluster (priority, fire toggle).
     if isinstance(data, dict):
         data["category"] = "unattributed_hotspot_cluster"
-        # Set severity to priority so downstream rules see the right tier.
+        # issue #118: `_severity_override` is the ONLY key consumer.py honors
-        data["severity"] = "priority"
+        # when building the Event -- the plain "severity" key here was a
+        # silent no-op (the Event fell back to map_severity() instead).
+        data["_severity_override"] = "priority"
 
     return _render_cluster_wire(
         n=len(members), radius_mi=radius_mi,
@@ -1039,7 +1041,9 @@ def _maybe_emit_halt(conn, *, data, now, seed=False):
                 (float(now), irwin_id),
             )
             data["category"] = "wildfire_halted"
-            data["severity"] = "routine"
+            # issue #118: `_severity_override` is the ONLY key consumer.py
+            # honors -- the plain "severity" key here was a silent no-op.
+            data["_severity_override"] = "routine"
     return wire
 
 
@@ -1227,7 +1231,9 @@ def _check_spotting(conn, *, irwin_id, pixel_lat, pixel_lon,
                 (float(now), irwin_id),
             )
             data["category"] = "wildfire_spotting"
-            data["severity"] = "immediate"
+            # issue #118: `_severity_override` is the ONLY key consumer.py
+            # honors -- the plain "severity" key here was a silent no-op.
+            data["_severity_override"] = "immediate"
 
     return wire
 

work/meshai/notifications/pipeline/pacer.py

@@ -8,11 +8,19 @@ succession.
 The queue is in-memory only — no persistence. On restart, the drain mode in
 consumer.py re-evaluates from DB state, so queued events lost on shutdown
 are re-derived naturally.
+
+Issue #119 (head-of-line priority): an "immediate"-severity event (e.g. a
+FIRMS wildfire_spotting broadcast) is enqueued at the HEAD of the FIFO
+instead of the tail, so it is never stuck waiting behind lower-urgency
+"priority" events already queued. "priority" events still append at the
+tail (plain FIFO among themselves). This is a pure ordering change --
+the queue remains unbounded and nothing is ever dropped.
 """
 
 from __future__ import annotations
 
 import asyncio
+import collections
 import logging
 from typing import Optional
 
@@ -20,7 +28,8 @@ logger = logging.getLogger(__name__)
 
 
 class FirePacer:
-    """Unbounded FIFO queue that drains events to the bus at a fixed rate."""
+    """Unbounded FIFO (with immediate-severity head-of-line) queue that
+    drains events to the bus at a fixed rate."""
 
     def __init__(self, bus, interval_seconds: float = 60.0):
         """Args:
@@ -29,30 +38,50 @@ class FirePacer:
         """
         self._bus = bus
         self._interval = interval_seconds
-        self._queue: asyncio.Queue = asyncio.Queue()
+        # A plain deque + asyncio.Event, rather than asyncio.Queue, so an
+        # "immediate" event can jump the line via appendleft() -- asyncio.Queue
+        # only exposes tail-append (put_nowait). Never bounded, never drops.
+        self._queue: collections.deque = collections.deque()
+        self._not_empty = asyncio.Event()
         self._task: Optional[asyncio.Task] = None
 
     def enqueue(self, event) -> None:
-        """Non-blocking enqueue. Safe to call from sync code in the same loop."""
+        """Non-blocking enqueue. Safe to call from sync code in the same loop.
-        self._queue.put_nowait(event)
+
-        logger.debug("pacer: enqueued event source=%s category=%s (pending=%d)",
+        Issue #119: "immediate"-severity events go to the HEAD of the queue
-                      event.source, event.category, self._queue.qsize())
+        so they broadcast before any already-queued "priority" event; every
+        other severity appends at the tail as before.
+        """
+        if getattr(event, "severity", None) == "immediate":
+            self._queue.appendleft(event)
+            logger.debug(
+                "pacer: enqueued (head-of-line, immediate) event source=%s "
+                "category=%s (pending=%d)",
+                event.source, event.category, len(self._queue))
+        else:
+            self._queue.append(event)
+            logger.debug("pacer: enqueued event source=%s category=%s (pending=%d)",
+                          event.source, event.category, len(self._queue))
+        self._not_empty.set()
 
     async def _drain_loop(self) -> None:
         """Pop one event, emit it, sleep interval, repeat."""
         while True:
-            try:
+            while not self._queue:
-                event = await self._queue.get()
+                self._not_empty.clear()
-            except asyncio.CancelledError:
+                try:
-                return
+                    await self._not_empty.wait()
+                except asyncio.CancelledError:
+                    return
+            event = self._queue.popleft()
             try:
                 self._bus.emit(event)
                 logger.info("pacer: emitted event source=%s category=%s (remaining=%d)",
-                            event.source, event.category, self._queue.qsize())
+                            event.source, event.category, len(self._queue))
             except Exception:
                 logger.exception("pacer: bus.emit() failed for event source=%s",
                                  event.source)
-            if self._queue.empty():
+            if not self._queue:
                 # No point sleeping when nothing is queued — wait for next put
                 continue
             try:
@@ -76,7 +105,7 @@ class FirePacer:
             except asyncio.CancelledError:
                 pass
             self._task = None
-            remaining = self._queue.qsize()
+            remaining = len(self._queue)
             if remaining:
                 logger.warning("pacer: stopped with %d events still queued", remaining)
             else:
@@ -84,4 +113,4 @@ class FirePacer:
 
     def pending_count(self) -> int:
         """Number of events waiting in the queue."""
-        return self._queue.qsize()
+        return len(self._queue)

work/tests/test_fire_tracker_phase1.py

@@ -218,7 +218,7 @@ def test_three_unattributed_pixels_fire_cluster_once():
         if wires[-1] is not None:
             # The handler must have tagged data with the cluster category.
             assert data.get("category") == "unattributed_hotspot_cluster"
-            assert data.get("severity") == "priority"
+            assert data.get("_severity_override") == "priority"
 
     # F3: cluster detection is ENABLED. The 3rd pixel reaches cluster_min_pixels
     # (3) within 1 mi / 60 min, so exactly ONE cluster wire fires (on pixel 3);

work/tests/test_fire_tracker_phase2.py

@@ -202,7 +202,7 @@ def test_halt_detector_fires_once_after_12h_idle():
     assert "Cold Fire" in wire
     assert "no growth in 14h" in wire
     assert data.get("category") == "wildfire_halted"
-    assert data.get("severity") == "routine"
+    assert data.get("_severity_override") == "routine"
 
     # halt_broadcast_at stamped.
     halt_at = conn.execute(

work/tests/test_fire_tracker_phase3.py

@@ -172,7 +172,7 @@ def test_pixel_2mi_ne_of_perimeter_emits_spotting():
     # routing; verification REPORTS only quote the wire (per the
     # feedback-no-event-metadata-in-reports memory rule).
     assert data["category"] == "wildfire_spotting"
-    assert data["severity"] == "immediate"
+    assert data["_severity_override"] == "immediate"
 
     # Latch stamped.
     fire = get_db().execute(

work/tests/test_firms_cluster_f3.py

@@ -125,7 +125,7 @@ def test_cluster_broadcasts_after_seed_then_refire_silent():
     assert wire.startswith("🔥 Possible new fire:")
     assert "3 hotspots within 1 mi" in wire
     assert data["category"] == "unattributed_hotspot_cluster"
-    assert data["severity"] == "priority"
+    assert data["_severity_override"] == "priority"
 
     # A 4th pixel inside the just-broadcast cluster -> silent (members stamped).
     refire = _feed(_pixel(lat=n2_lat + 0.0005, lon=n2_lon - 0.0005,

work/tests/test_firms_fusion_event_contract.py

@@ -0,0 +1,383 @@
+"""Regression tests for the FIRMS fire-fusion Event contract (issues #117-#119).
+
+Three independent bugs in the path from firms_handler's growth / spotting /
+halt / cluster fusion decisions to the actual meshai Event that reaches the
+dispatcher + pacer:
+
+  #117 category overrides dead: consumer._normalize() computed `category`
+       from the raw Central category BEFORE the per-adapter handler ran, and
+       the final make_event() call never re-read data["category"], so every
+       firms_handler category stamp (wildfire_growth / wildfire_halted /
+       wildfire_spotting / unattributed_hotspot_cluster) was a silent no-op.
+
+  #118 severity overrides dead for 3 of 4 fusion kinds: consumer.py only ever
+       honored data["_severity_override"], but firms_handler's halt / spotting
+       / cluster sites stamped the plain data["severity"] key instead (only
+       growth used the correct key), so those events fell back to whatever
+       map_severity(inner.get("severity")) produced from the raw envelope.
+
+  #119 FirePacer didn't cover FIRMS: the pacer gate only matched
+       source in ("fires", "wfigs") and severity == "priority", but FIRMS
+       fusion broadcasts carry source="firms" and growth/spotting are
+       severity="immediate" -- so none of them were ever paced. The fix
+       broadens the gate to source="firms" + {"priority","immediate"}, and
+       adds head-of-line insertion so an "immediate" event is never stuck
+       behind already-queued "priority" events.
+
+Cluster detection itself is a deliberate, always-on feature of main (PR #73:
+curated new-fire cluster broadcasts, cold-start silent-seeded). Nothing here
+enables or disables it -- the cluster case below only asserts that its
+severity override reaches the Event (the #118 fix).
+
+Sections:
+  A. Category overrides survive to the emitted Event (growth/halt/spotting).
+  B. Severity overrides survive to the emitted Event, using the shared
+     `_severity_override` contract (spotting=immediate, halt=routine,
+     cluster=priority).
+  C. FirePacer routes FIRMS broadcasts; immediate jumps the queue; nothing
+     is ever dropped.
+"""
+from __future__ import annotations
+
+import asyncio
+import math
+import time
+import uuid
+
+import pytest
+
+from meshai.config import Config
+from meshai.central.consumer import CentralConsumer
+from meshai.notifications.events import make_event
+from meshai.notifications.pipeline.pacer import FirePacer
+from meshai.persistence import close_thread_connection, init_db
+from meshai.persistence import db as persistence_db
+
+_MI_PER_DEG_LAT = 69.0
+_SUBJECT = "central.fire.hotspot.N20.high.us.id"
+
+
+# ── isolation ────────────────────────────────────────────────────────────────
+
+@pytest.fixture(autouse=True)
+def _isolate_db(tmp_path, monkeypatch):
+    db_path = str(tmp_path / f"meshai-{uuid.uuid4().hex}.sqlite")
+    monkeypatch.setenv("MESHAI_DB_PATH", db_path)
+    persistence_db._initialised.clear()
+    close_thread_connection()
+    init_db()
+    try:
+        from meshai.adapter_config import adapter_config as _ac
+        _ac.invalidate()
+    except Exception:
+        pass
+    yield db_path
+    close_thread_connection()
+    persistence_db._initialised.discard(db_path)
+
+
+@pytest.fixture(autouse=True)
+def _no_cutover(monkeypatch):
+    """Default deploy state: nothing cut over -> firms_handler's legacy
+    stamps are what actually reach `data`."""
+    monkeypatch.delenv("MESHAI_CUTOVER_CATEGORIES", raising=False)
+    from meshai.notifications.cutover import _clear_cache
+    _clear_cache()
+    yield
+    _clear_cache()
+
+
+# firms_handler.handle_firms defaults `now` to real wall-clock time
+# (int(time.time())) whenever it is called without an explicit `now`
+# kwarg -- which is exactly how consumer._normalize()/_handle() call it (no
+# `now` is threaded through from the envelope). The growth/spotting/halt
+# fixtures below use fixed 2026-06-06 acq_date/acq_time values (matching the
+# rest of the FIRMS test suite), so pin the wall clock to a fixed reference
+# in the same window; otherwise the halt detector opportunistically fires on
+# every pixel once the real host clock is weeks past the canned acq_times.
+_FIXED_NOW = 1780768800.0  # 2026-06-06 18:00 UTC
+
+
+@pytest.fixture(autouse=True)
+def _fixed_clock(monkeypatch):
+    monkeypatch.setattr("time.time", lambda: _FIXED_NOW)
+    yield
+
+
+@pytest.fixture
+def consumer():
+    """CentralConsumer with a mocked bus, mirroring
+    test_consumer_default_deny.py's `consumer` fixture."""
+    from unittest.mock import MagicMock
+    cfg = Config()
+    cfg.notifications.cold_start_grace_seconds = 0
+    bus = MagicMock()
+    c = CentralConsumer(cfg.environmental, bus)
+    return c, bus
+
+
+# ── envelope + fire-seeding helpers (mirror test_firms_refactor.py) ─────────
+
+def _seed_fire(*, irwin_id, lat, lon, name="Stub Fire", **cols):
+    from meshai.persistence import get_db
+    conn = get_db()
+    base = {"irwin_id": irwin_id, "incident_name": name, "lat": lat, "lon": lon,
+            "last_event_at": int(time.time())}
+    base.update(cols)
+    keys = ",".join(base)
+    ph = ",".join("?" * len(base))
+    conn.execute(f"INSERT INTO fires({keys}) VALUES ({ph})", tuple(base.values()))
+
+
+def _envelope(*, lat, lon, acq_date="2026-06-06", acq_time="1200",
+              frp=20.0, satellite="N20", eid=None):
+    eid = eid or f"firms-{lat}-{lon}-{acq_time}"
+    return {
+        "id": f"env_{eid}",
+        "data": {
+            "id": eid,
+            "adapter": "firms",
+            "category": "wildfire_hotspot",
+            "severity": "routine",
+            "geo": {"primary_region": "US-ID"},
+            "data": {
+                "latitude": lat, "longitude": lon, "frp": frp,
+                "bright_ti4": 320.0, "satellite": satellite,
+                "instrument": "VIIRS", "confidence": "high",
+                "acq_date": acq_date, "acq_time": acq_time,
+                "daynight": "D", "version": "2.0NRT",
+            },
+        },
+    }
+
+
+def _offset_mi(lat, lon, north_mi, east_mi):
+    dlat = north_mi / _MI_PER_DEG_LAT
+    dlon = east_mi / (_MI_PER_DEG_LAT * math.cos(math.radians(lat)))
+    return lat + dlat, lon + dlon
+
+
+# ═════════════════════════════════════════════════════════════════════════════
+# A. Category overrides reach the emitted Event (issue #117)
+# ═════════════════════════════════════════════════════════════════════════════
+
+class TestCategoryOverrideReachesEvent:
+    def test_growth_event_category_is_wildfire_growth(self, consumer):
+        c, _bus = consumer
+        center_lat, center_lon = 42.0, -114.0
+        _seed_fire(irwin_id="ID-CAT-G", lat=center_lat, lon=center_lon,
+                   name="Pine Gulch")
+        # Pass A: 5 pixels build the baseline (no broadcast yet).
+        for i in range(5):
+            evt = c._normalize(_SUBJECT, _envelope(
+                lat=center_lat + 0.0001 * i, lon=center_lon + 0.0001 * (i - 2),
+                acq_time=f"12{i:02d}", frp=20.0 + i, eid=f"ga{i}"))
+            assert evt is None, "pass-A pixels must not broadcast"
+        # Pass B: 1 mi N, later pass bucket -> growth boundary.
+        pass_b_lat = center_lat + (1.0 / _MI_PER_DEG_LAT)
+        evt = c._normalize(_SUBJECT, _envelope(
+            lat=pass_b_lat, lon=center_lon, acq_time="1800", frp=22.0, eid="gb"))
+        assert evt is not None
+        assert evt.category == "wildfire_growth", (
+            "category override must survive to the Event, not the generic "
+            "wildfire_hotspot/wildfire_incident fallback")
+        assert evt.source == "firms"
+
+    def test_halt_event_category_is_wildfire_halted(self, consumer):
+        c, _bus = consumer
+        now = 1780768800
+        idle_at = now - 14 * 3600
+        from meshai.persistence import get_db
+        get_db().execute(
+            "INSERT INTO fires(irwin_id, incident_name, lat, lon, "
+            "last_event_at, last_pass_id, last_pass_at) VALUES (?,?,?,?,?,?,?)",
+            ("ID-CAT-H", "Cold Fire", 42.5, -114.5, int(idle_at),
+             "N20-329627", float(idle_at)))
+        # A fresh unattributed pixel far away triggers the opportunistic
+        # halt detector for the idle fire.
+        evt = c._normalize(_SUBJECT, _envelope(
+            lat=45.0, lon=-118.0, acq_time="1800", eid="halt1"))
+        assert evt is not None
+        assert evt.category == "wildfire_halted"
+
+    def test_spotting_event_category_is_wildfire_spotting(self, consumer):
+        c, _bus = consumer
+        center_lat, center_lon = 43.0, -115.0
+        _seed_fire(irwin_id="ID-CAT-S", lat=center_lat, lon=center_lon,
+                   name="Spot Fire")
+        for i in range(6):
+            angle = i * math.pi / 3
+            la = center_lat + (0.5 / _MI_PER_DEG_LAT) * math.sin(angle)
+            cos_lat = math.cos(math.radians(center_lat))
+            lo = center_lon + (0.5 / (_MI_PER_DEG_LAT * cos_lat)) * math.cos(angle)
+            evt = c._normalize(_SUBJECT, _envelope(
+                lat=la, lon=lo, acq_time=f"12{i * 2:02d}", eid=f"sa{i}"))
+            assert evt is None
+        sp_lat, sp_lon = _offset_mi(center_lat, center_lon,
+                                    north_mi=2.0 / math.sqrt(2),
+                                    east_mi=2.0 / math.sqrt(2))
+        evt = c._normalize(_SUBJECT, _envelope(
+            lat=sp_lat, lon=sp_lon, acq_time="1800", eid="sb"))
+        assert evt is not None
+        assert evt.category == "wildfire_spotting"
+
+
+# ═════════════════════════════════════════════════════════════════════════════
+# B. Severity overrides reach the emitted Event via `_severity_override`
+#    (issue #118)
+# ═════════════════════════════════════════════════════════════════════════════
+
+class TestSeverityOverrideReachesEvent:
+    def test_spotting_event_severity_is_immediate(self, consumer):
+        c, _bus = consumer
+        center_lat, center_lon = 44.0, -116.0
+        _seed_fire(irwin_id="ID-SEV-S", lat=center_lat, lon=center_lon)
+        for i in range(6):
+            angle = i * math.pi / 3
+            la = center_lat + (0.5 / _MI_PER_DEG_LAT) * math.sin(angle)
+            cos_lat = math.cos(math.radians(center_lat))
+            lo = center_lon + (0.5 / (_MI_PER_DEG_LAT * cos_lat)) * math.cos(angle)
+            c._normalize(_SUBJECT, _envelope(
+                lat=la, lon=lo, acq_time=f"12{i * 2:02d}", eid=f"ssa{i}"))
+        sp_lat, sp_lon = _offset_mi(center_lat, center_lon,
+                                    north_mi=2.0 / math.sqrt(2),
+                                    east_mi=2.0 / math.sqrt(2))
+        evt = c._normalize(_SUBJECT, _envelope(
+            lat=sp_lat, lon=sp_lon, acq_time="1800", eid="ssb"))
+        assert evt is not None
+        assert evt.severity == "immediate", (
+            "spotting must reach the pacer/dispatcher as immediate severity, "
+            "not fall back to map_severity() of the raw envelope")
+
+    def test_halt_event_severity_is_routine(self, consumer):
+        c, _bus = consumer
+        now = 1780768800
+        idle_at = now - 14 * 3600
+        from meshai.persistence import get_db
+        get_db().execute(
+            "INSERT INTO fires(irwin_id, incident_name, lat, lon, "
+            "last_event_at, last_pass_id, last_pass_at) VALUES (?,?,?,?,?,?,?)",
+            ("ID-SEV-H", "Cold Fire", 42.5, -114.5, int(idle_at),
+             "N20-329627", float(idle_at)))
+        evt = c._normalize(_SUBJECT, _envelope(
+            lat=45.0, lon=-118.0, acq_time="1800", eid="sevhalt"))
+        assert evt is not None
+        assert evt.severity == "routine"
+
+    def test_cluster_event_severity_is_priority(self, consumer):
+        c, _bus = consumer
+        base_lat, base_lon = 43.500, -114.500
+        pixels = [
+            (base_lat, base_lon, "1200"),
+            (base_lat + 0.001, base_lon + 0.001, "1210"),
+            (base_lat - 0.001, base_lon - 0.002, "1220"),
+        ]
+        events = []
+        for i, (la, lo, t) in enumerate(pixels):
+            evt = c._normalize("central.fire.hotspot.N20.high.unknown", _envelope(
+                lat=la, lon=lo, acq_time=t, eid=f"clu{i}"))
+            if evt is not None:
+                events.append(evt)
+        assert len(events) == 1, f"expected exactly one cluster event: {events}"
+        assert events[0].category == "unattributed_hotspot_cluster"
+        assert events[0].severity == "priority"
+
+
+# ═════════════════════════════════════════════════════════════════════════════
+# C. FirePacer covers FIRMS; immediate jumps the queue; nothing is dropped
+#    (issue #119)
+# ═════════════════════════════════════════════════════════════════════════════
+
+class TestPacerCoversFirms:
+    def test_firms_growth_broadcast_routes_through_pacer(self, consumer):
+        """A real FIRMS growth broadcast (source=firms, severity=immediate)
+        must be handed to the pacer, not emitted straight to the bus."""
+        from unittest.mock import MagicMock
+        c, bus = consumer
+        pacer = MagicMock()
+        c._pacer = pacer
+
+        center_lat, center_lon = 42.0, -114.0
+        _seed_fire(irwin_id="ID-PACE-G", lat=center_lat, lon=center_lon,
+                   name="Pine Gulch")
+        for i in range(5):
+            c._handle(_SUBJECT, _raw(_envelope(
+                lat=center_lat + 0.0001 * i, lon=center_lon + 0.0001 * (i - 2),
+                acq_time=f"12{i:02d}", frp=20.0 + i, eid=f"pga{i}")))
+        pass_b_lat = center_lat + (1.0 / _MI_PER_DEG_LAT)
+        event = c._handle(_SUBJECT, _raw(_envelope(
+            lat=pass_b_lat, lon=center_lon, acq_time="1800", frp=22.0,
+            eid="pgb")))
+
+        assert event is not None
+        assert event.category == "wildfire_growth"
+        pacer.enqueue.assert_called_once_with(event)
+        bus.emit.assert_not_called()
+
+    def test_immediate_event_emitted_before_already_queued_priority_events(self):
+        """Two 'priority' events are queued first; a later 'immediate' event
+        must still be emitted BEFORE them (head-of-line), not after."""
+        emitted = []
+
+        class _FakeBus:
+            def emit(self, event):
+                emitted.append(event)
+
+        pacer = FirePacer(_FakeBus(), interval_seconds=0.01)
+
+        p1 = make_event(source="fires", category="wildfire_incident",
+                         severity="priority", title="priority-1")
+        p2 = make_event(source="fires", category="wildfire_incident",
+                         severity="priority", title="priority-2")
+        imm = make_event(source="firms", category="wildfire_spotting",
+                          severity="immediate", title="immediate-1")
+
+        pacer.enqueue(p1)
+        pacer.enqueue(p2)
+        pacer.enqueue(imm)  # must jump ahead of p1/p2
+
+        async def _drive():
+            await pacer.start()
+            await asyncio.sleep(0.2)
+            await pacer.stop()
+
+        asyncio.run(_drive())
+
+        assert [e.title for e in emitted] == [
+            "immediate-1", "priority-1", "priority-2"]
+
+    def test_pacer_never_drops_events(self):
+        """Rapid-fire enqueue of many events (mixed severities) -- the
+        unbounded FIFO must eventually deliver every single one."""
+        emitted = []
+
+        class _FakeBus:
+            def emit(self, event):
+                emitted.append(event)
+
+        pacer = FirePacer(_FakeBus(), interval_seconds=0.001)
+
+        total = 25
+        for i in range(total):
+            sev = "immediate" if i % 5 == 0 else "priority"
+            pacer.enqueue(make_event(
+                source="firms", category="wildfire_growth",
+                severity=sev, title=f"evt-{i}"))
+        assert pacer.pending_count() == total
+
+        async def _drive():
+            await pacer.start()
+            # Generous wait: interval is 1ms, 25 events, allow real margin.
+            await asyncio.sleep(1.0)
+            await pacer.stop()
+
+        asyncio.run(_drive())
+
+        assert len(emitted) == total, (
+            f"pacer must never drop events: expected {total}, got {len(emitted)}")
+        assert pacer.pending_count() == 0
+
+
+def _raw(envelope: dict) -> bytes:
+    import json
+    return json.dumps(envelope).encode()

work/tests/test_firms_native_fusion.py

@@ -169,7 +169,7 @@ class TestIngestSpotting:
         assert len(out) == 1
         wire, data = out[0]
         assert data["category"] == "wildfire_spotting"
-        assert data["severity"] == "immediate"
+        assert data["_severity_override"] == "immediate"
         assert wire.startswith("🔥 Possible spotting ")
         # Eager latch stamped during ingest (not-cutover legacy path).
         from meshai.persistence import get_db
@@ -197,7 +197,7 @@ class TestIngestHalt:
         assert len(out) == 1
         wire, data = out[0]
         assert data["category"] == "wildfire_halted"
-        assert data["severity"] == "routine"
+        assert data["_severity_override"] == "routine"
         assert wire == "🔥 Cold Fire no growth in 14h"
         latch = get_db().execute(
             "SELECT halt_broadcast_at FROM fires WHERE irwin_id=?",
@@ -230,7 +230,7 @@ class TestIngestNeverRaw:
         for wire, data in produced:
             assert wire.startswith("🔥 Possible new fire:")
             assert data["category"] == "unattributed_hotspot_cluster"
-            assert data["severity"] == "priority"
+            assert data["_severity_override"] == "priority"
 
 
 # ═════════════════════════════════════════════════════════════════════════════

work/tests/test_firms_refactor.py

@@ -402,7 +402,7 @@ class TestNotCutoverLegacyVerbatim:
         wire, data = _drive_spotting("ID-SN", 43.0, -115.0, now=1780768800)
         assert wire is not None and "spotting" in wire
         assert data["category"] == "wildfire_spotting"
-        assert data["severity"] == "immediate"
+        assert data["_severity_override"] == "immediate"
         # Legacy path stamps the latch EAGERLY with the handler `now`.
         latch = get_db().execute(
             "SELECT last_spotting_broadcast_at FROM fires WHERE irwin_id=?",
@@ -419,7 +419,7 @@ class TestNotCutoverLegacyVerbatim:
         wire = _maybe_emit_halt(get_db(), data=data, now=now)
         assert wire == "🔥 Cold Fire no growth in 14h"
         assert data["category"] == "wildfire_halted"
-        assert data["severity"] == "routine"
+        assert data["_severity_override"] == "routine"
         latch = get_db().execute(
             "SELECT halt_broadcast_at FROM fires WHERE irwin_id=?",
             ("ID-HN",)).fetchone()[0]