From 48af45cdf6454a7491f8e052149db1e09d4cabb8 Mon Sep 17 00:00:00 2001 From: Matt Johnson Date: Sat, 11 Jul 2026 22:55:59 +0000 Subject: [PATCH] fix(firms): repair the FIRMS fire-fusion Event contract (issues #117-#119) 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: Claude Opus 4.8 (1M context) --- work/meshai/central/consumer.py | 39 +- work/meshai/central/firms_handler.py | 14 +- work/meshai/notifications/pipeline/pacer.py | 57 ++- work/tests/test_fire_tracker_phase1.py | 2 +- work/tests/test_fire_tracker_phase2.py | 2 +- work/tests/test_fire_tracker_phase3.py | 2 +- work/tests/test_firms_cluster_f3.py | 2 +- .../tests/test_firms_fusion_event_contract.py | 383 ++++++++++++++++++ work/tests/test_firms_native_fusion.py | 6 +- work/tests/test_firms_refactor.py | 4 +- 10 files changed, 479 insertions(+), 32 deletions(-) create mode 100644 work/tests/test_firms_fusion_event_contract.py diff --git a/work/meshai/central/consumer.py b/work/meshai/central/consumer.py index 1992a07..9594a9e 100644 --- a/work/meshai/central/consumer.py +++ b/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.data or {}).get("_severity_override") == "priority"): + and event.source in ("fires", "wfigs", "firms") + and event.severity in ("priority", "immediate")): self._pacer.enqueue(event) else: self._bus.emit(event) diff --git a/work/meshai/central/firms_handler.py b/work/meshai/central/firms_handler.py index e868622..7353df7 100644 --- a/work/meshai/central/firms_handler.py +++ b/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. - data["severity"] = "priority" + # issue #118: `_severity_override` is the ONLY key consumer.py honors + # 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 diff --git a/work/meshai/notifications/pipeline/pacer.py b/work/meshai/notifications/pipeline/pacer.py index 16039b2..d96c3a7 100644 --- a/work/meshai/notifications/pipeline/pacer.py +++ b/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.""" - self._queue.put_nowait(event) - logger.debug("pacer: enqueued event source=%s category=%s (pending=%d)", - event.source, event.category, self._queue.qsize()) + """Non-blocking enqueue. Safe to call from sync code in the same loop. + + Issue #119: "immediate"-severity events go to the HEAD of the queue + 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: - event = await self._queue.get() - except asyncio.CancelledError: - return + while not self._queue: + self._not_empty.clear() + try: + 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) diff --git a/work/tests/test_fire_tracker_phase1.py b/work/tests/test_fire_tracker_phase1.py index 71dad76..0a4f77d 100644 --- a/work/tests/test_fire_tracker_phase1.py +++ b/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); diff --git a/work/tests/test_fire_tracker_phase2.py b/work/tests/test_fire_tracker_phase2.py index dd5c429..53a2e06 100644 --- a/work/tests/test_fire_tracker_phase2.py +++ b/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( diff --git a/work/tests/test_fire_tracker_phase3.py b/work/tests/test_fire_tracker_phase3.py index 25b1122..75f883b 100644 --- a/work/tests/test_fire_tracker_phase3.py +++ b/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( diff --git a/work/tests/test_firms_cluster_f3.py b/work/tests/test_firms_cluster_f3.py index 49c2b6d..da99f98 100644 --- a/work/tests/test_firms_cluster_f3.py +++ b/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, diff --git a/work/tests/test_firms_fusion_event_contract.py b/work/tests/test_firms_fusion_event_contract.py new file mode 100644 index 0000000..09531a3 --- /dev/null +++ b/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() diff --git a/work/tests/test_firms_native_fusion.py b/work/tests/test_firms_native_fusion.py index 344947f..b86cf91 100644 --- a/work/tests/test_firms_native_fusion.py +++ b/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" # ═════════════════════════════════════════════════════════════════════════════ diff --git a/work/tests/test_firms_refactor.py b/work/tests/test_firms_refactor.py index 790bf52..6f471cf 100644 --- a/work/tests/test_firms_refactor.py +++ b/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]