fix(fires): wire FirePacer into the native fire broadcast path

Native fire adapters (env/fires.py source="nifc", env/firms.py
source="firms") emitted straight to the EventBus from
EnvironmentalStore._emit_event with no rate limiting of their own.
FirePacer was only ever attached to CentralConsumer (main.py), which
never runs in the actual production deployment (central.enabled=False,
all adapters feed_source=native) -- so the <=1/60s throttle + immediate
head-of-line behavior fixed for Central in #120 (issue #119) was
completely inert in production. A poll that produces several distinct
fires/clusters at once (a lightning outbreak, or several tracked fires
crossing a satellite-pass boundary together) would dump all of them on
the mesh back-to-back instead of at the intended cadence.

_emit_event() now routes fire-family Events (source in
{"nifc","firms"}, severity in {"priority","immediate"}) through an
attached FirePacer, mirroring the exact gate CentralConsumer._handle
applies. main.py attaches the same FirePacer instance to env_store
right after constructing it. "routine"-severity fire events, non-fire
native adapters, and the Central path are all unaffected; a paced event
cannot re-enter either gate (native vs Central are mutually exclusive
per feed_source), so nothing can be paced twice.

Added tests/test_native_fire_pacer.py covering: native fire events
route through the pacer, an immediate event jumps an already-queued
priority queue with nothing dropped, routine-severity fire events and
non-fire native events are NOT paced, and the no-pacer-attached
fallback is unchanged.

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

work/meshai/env/store.py

@@ -30,6 +30,21 @@ def _key_eid(source: str, event_id) -> str:
     return f"{source}{_SEP}eid:{event_id}"
 
 
+# ── FirePacer routing (native path) ──────────────────────────────────────────
+# The exact Event.source strings the NATIVE fire adapters mint (verified
+# against the adapter code, not assumed): env/fires.py's WFIGS incident
+# adapter stamps source="nifc" (NOT "fires"/"wfigs" -- those are the Central
+# NATS subject/handler names); env/firms.py's fusion broadcasts always stamp
+# source="firms" regardless of fusion kind (growth/spotting/halt/cluster) --
+# "firms_fusion" only ever appears as the raw internal event dict's bookkeeping
+# key, never as the emitted Event.source. Mirrors the gate
+# central/consumer.py:_handle applies to the Central path (issue #119) so a
+# native and a Central deployment throttle identically; "routine" fire events
+# (e.g. firms halt) are intentionally excluded, exactly as on the Central path.
+_FIRE_PACER_SOURCES = frozenset({"nifc", "firms"})
+_FIRE_PACER_SEVERITIES = frozenset({"priority", "immediate"})
+
+
 class EnvironmentalStore:
     """Cache and tick-driver for all environmental feed adapters."""
 
@@ -50,6 +65,12 @@ class EnvironmentalStore:
         self._failed_adapters = {}  # name -> last_error string
         self._events = {}  # (source, event_id) -> event dict
         self._event_bus = event_bus  # Pipeline EventBus for emission
+        # FirePacer, injected post-construction from main.py (mirrors
+        # central/consumer.py's self._pacer) once the pacer exists -- the
+        # store is built earlier in _init_components(), before start()
+        # creates the pacer. None until attached; _emit_event() falls back
+        # to emitting straight to the bus when unset (pre-existing behavior).
+        self._fire_pacer = None
         self._swpc_status = {}  # Kp/SFI/scales snapshot
         self._ducting_status = {}  # tropo ducting assessment
         self._mesh_zones = config.nws_zones or []
@@ -960,6 +981,26 @@ class EnvironmentalStore:
                 )
                 return  # default-deny on decider error
 
+            # Native fire-family events get the same <=1/60s pacing (with
+            # head-of-line for "immediate" severity) that the Central path
+            # applies via CentralConsumer._handle (issue #119) -- see the
+            # _FIRE_PACER_SOURCES/_FIRE_PACER_SEVERITIES gate above. Only ONE
+            # of {pacer, direct bus.emit} ever runs for a given event: this
+            # gate only executes on the NATIVE ingest path (an adapter whose
+            # feed_source=="native"), and the pacer's own drain loop calls
+            # bus.emit() directly -- never back through this method or
+            # through CentralConsumer._handle -- so a paced event cannot
+            # re-enter this gate and be paced twice.
+            fire_pacer = getattr(self, "_fire_pacer", None)
+            if (fire_pacer is not None
+                    and event.source in _FIRE_PACER_SOURCES
+                    and event.severity in _FIRE_PACER_SEVERITIES):
+                fire_pacer.enqueue(event)
+                logger.info(
+                    "Queued %s event %s (%s) on FirePacer",
+                    event.source, event.id, event.category,
+                )
+                return
             self._event_bus.emit(event)
             logger.info(
                 "Emitted %s event %s (%s) to pipeline bus",

work/meshai/main.py

@@ -131,6 +131,17 @@ class MeshAI:
             self._fire_pacer = FirePacer(bus=self.event_bus, interval_seconds=60.0)
             await self._fire_pacer.start()
 
+            # Native fire adapters (env/fires.py "nifc", env/firms.py "firms")
+            # emit straight to the EventBus from EnvironmentalStore._emit_event
+            # with no rate limiting of their own -- attach the SAME pacer so a
+            # burst of distinct fires/clusters in one poll (e.g. a lightning
+            # outbreak, or several tracked fires crossing a satellite-pass
+            # boundary together) gets the identical <=1/60s throttle the
+            # Central path gets, instead of flooding the mesh essentially
+            # back-to-back. See env/store.py's _FIRE_PACER_SOURCES gate.
+            if self.env_store is not None:
+                self.env_store._fire_pacer = self._fire_pacer
+
             from .central.consumer import CentralConsumer
             self._central_consumer = CentralConsumer(self.config.environmental, self.event_bus)
             self._central_consumer._pacer = self._fire_pacer

work/tests/test_native_fire_pacer.py

@@ -0,0 +1,201 @@
+"""Native fire-family events must be paced too, not just Central ones.
+
+FirePacer (notifications/pipeline/pacer.py) was wired ONLY into the Central
+NATS consumer path (central/consumer.py._handle routes fire-family events to
+`self._pacer.enqueue(event)` instead of `self._bus.emit(event)` -- see issue
+#119). In an all-native deployment (central.enabled=False, the actual
+production configuration on CT 108) that consumer never receives any NATS
+message, so the pacer sat completely unused: every native fire adapter
+(env/fires.py, source="nifc"; env/firms.py, source="firms") emitted straight
+to the EventBus from EnvironmentalStore._emit_event with no rate limiting at
+all. A FIRMS poll (or a WFIGS poll) that produces several distinct
+fires/clusters at once -- a lightning outbreak forming multiple new-fire
+clusters, or several already-tracked fires crossing a satellite-pass boundary
+in the same fetch -- would dump all of them onto the mesh essentially
+back-to-back instead of at the intended <=1/60s cadence.
+
+These tests exercise the fix: store._emit_event() now routes fire-family
+Events (source in {"nifc", "firms"}, severity in {"priority", "immediate"})
+through an attached FirePacer, mirroring the exact gate CentralConsumer._handle
+applies, and leaves everything else (other native adapters, "routine"-severity
+fire events, and the case where no pacer is attached at all) unchanged.
+"""
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from meshai.config import EnvironmentalConfig
+from meshai.env.store import EnvironmentalStore
+from meshai.notifications.cutover import _clear_cache
+from meshai.notifications.events import make_event
+from meshai.notifications.pipeline.bus import EventBus
+from meshai.notifications.pipeline.pacer import FirePacer
+
+
+@pytest.fixture(autouse=True)
+def _no_cutover(monkeypatch):
+    """None of the categories used below are meant to hit the real gating
+    deciders (this file is only exercising the pacer-routing gate); clearing
+    cutover keeps store._emit_event's decider hook a no-op regardless of
+    what earlier tests left in the environment / lru_cache."""
+    monkeypatch.delenv("MESHAI_CUTOVER_CATEGORIES", raising=False)
+    _clear_cache()
+    yield
+    _clear_cache()
+
+
+class _StubAdapter:
+    """Minimal adapter stand-in: to_event() always returns the same
+    caller-controlled Event, independent of the raw dict passed in."""
+
+    def __init__(self, source: str, severity: str, category: str):
+        self._source = source
+        self._severity = severity
+        self._category = category
+
+    def to_event(self, raw_evt: dict):
+        eid = raw_evt["event_id"]
+        return make_event(
+            source=self._source,
+            category=self._category,
+            severity=self._severity,
+            title=eid,
+            summary=eid,
+            group_key=eid,
+        )
+
+
+def _make_store():
+    bus = EventBus()
+    captured: list = []
+    bus.subscribe(lambda e: captured.append(e))
+    store = EnvironmentalStore(EnvironmentalConfig(), event_bus=bus)
+    return store, bus, captured
+
+
+class _FakePacer:
+    """Records enqueue() calls without any real draining."""
+
+    def __init__(self):
+        self.calls: list = []
+
+    def enqueue(self, event) -> None:
+        self.calls.append(event)
+
+
+def test_native_firms_immediate_event_routes_through_pacer():
+    """source=firms, severity=immediate (a FIRMS growth/spotting broadcast)
+    with a pacer attached must be enqueued on the pacer, and must NOT also
+    reach the bus directly (no double-delivery)."""
+    store, bus, captured = _make_store()
+    pacer = _FakePacer()
+    store._fire_pacer = pacer
+    adapter = _StubAdapter(source="firms", severity="immediate",
+                            category="wildfire_growth")
+
+    store._emit_event(adapter, {"event_id": "g1"})
+
+    assert len(pacer.calls) == 1
+    assert pacer.calls[0].source == "firms"
+    assert pacer.calls[0].severity == "immediate"
+    assert captured == [], "must not ALSO be emitted straight to the bus"
+
+
+def test_native_nifc_priority_event_routes_through_pacer():
+    """source=nifc (WFIGS incident), severity=priority is also paced."""
+    store, bus, captured = _make_store()
+    pacer = _FakePacer()
+    store._fire_pacer = pacer
+    adapter = _StubAdapter(source="nifc", severity="priority",
+                            category="unattributed_hotspot_cluster")
+
+    store._emit_event(adapter, {"event_id": "n1"})
+
+    assert len(pacer.calls) == 1
+    assert pacer.calls[0].source == "nifc"
+    assert captured == []
+
+
+def test_native_fire_routine_severity_not_paced():
+    """A fire-family event at "routine" severity (e.g. a FIRMS halt) is
+    excluded from pacing -- exactly like the Central-path gate, which only
+    covers priority/immediate."""
+    store, bus, captured = _make_store()
+    pacer = _FakePacer()
+    store._fire_pacer = pacer
+    adapter = _StubAdapter(source="firms", severity="routine",
+                            category="wildfire_halted")
+
+    store._emit_event(adapter, {"event_id": "h1"})
+
+    assert pacer.calls == []
+    assert len(captured) == 1
+    assert captured[0].severity == "routine"
+
+
+def test_non_fire_native_event_never_paced():
+    """A non-fire native adapter (e.g. roads511) at immediate severity must
+    go straight to the bus even with a pacer attached -- pacing is
+    fire-family-only, keyed on event.source."""
+    store, bus, captured = _make_store()
+    pacer = _FakePacer()
+    store._fire_pacer = pacer
+    adapter = _StubAdapter(source="roads511", severity="immediate",
+                            category="road_closure")
+
+    store._emit_event(adapter, {"event_id": "r1"})
+
+    assert pacer.calls == []
+    assert len(captured) == 1
+    assert captured[0].source == "roads511"
+
+
+def test_no_pacer_attached_falls_back_to_direct_emit():
+    """Pre-existing behavior is unaffected when no pacer is attached (e.g.
+    notifications disabled, or the brief startup window in main.py before
+    the pacer is constructed and wired in)."""
+    store, bus, captured = _make_store()
+    assert store._fire_pacer is None
+    adapter = _StubAdapter(source="firms", severity="immediate",
+                            category="wildfire_growth")
+
+    store._emit_event(adapter, {"event_id": "g1"})
+
+    assert len(captured) == 1
+    assert captured[0].source == "firms"
+
+
+def test_native_immediate_event_jumps_ahead_of_already_queued_priority_events():
+    """End-to-end through a REAL FirePacer: two native "priority" fires
+    queued first must not block a later native "immediate" one (head-of-line,
+    issue #119's fix -- now reachable from the native ingest path), and
+    nothing is ever dropped."""
+    store, bus, emitted = _make_store()
+    pacer = FirePacer(bus=bus, interval_seconds=0.01)
+    store._fire_pacer = pacer
+
+    p1 = _StubAdapter(source="nifc", severity="priority",
+                       category="unattributed_hotspot_cluster")
+    p2 = _StubAdapter(source="nifc", severity="priority",
+                       category="unattributed_hotspot_cluster")
+    imm = _StubAdapter(source="firms", severity="immediate",
+                        category="wildfire_spotting")
+
+    store._emit_event(p1, {"event_id": "priority-1"})
+    store._emit_event(p2, {"event_id": "priority-2"})
+    store._emit_event(imm, {"event_id": "immediate-1"})
+
+    assert pacer.pending_count() == 3
+    assert emitted == []
+
+    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"]