From d1f78a08368a770857587dec6caf576bb3a5a2ef Mon Sep 17 00:00:00 2001 From: Matt Johnson Date: Sun, 12 Jul 2026 01:09:10 +0000 Subject: [PATCH] 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 --- work/meshai/env/store.py | 41 ++++++ work/meshai/main.py | 11 ++ work/tests/test_native_fire_pacer.py | 201 +++++++++++++++++++++++++++ 3 files changed, 253 insertions(+) create mode 100644 work/tests/test_native_fire_pacer.py diff --git a/work/meshai/env/store.py b/work/meshai/env/store.py index 818c26a..58672a3 100644 --- a/work/meshai/env/store.py +++ b/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", diff --git a/work/meshai/main.py b/work/meshai/main.py index ea9eab1..fbc5b1d 100644 --- a/work/meshai/main.py +++ b/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 diff --git a/work/tests/test_native_fire_pacer.py b/work/tests/test_native_fire_pacer.py new file mode 100644 index 0000000..15f6efb --- /dev/null +++ b/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"]