From 74a5fa44d4bb8f1cf4c05f0463cbef39dde2663d Mon Sep 17 00:00:00 2001 From: malice Date: Tue, 14 Jul 2026 10:37:07 -0600 Subject: [PATCH] fix(pipeline): let wildfire_spotting skip the grouper (category-scoped) (#131) wildfire_spotting is the most urgent signal in the system (a fire throwing embers past its own containment line), but every fire event carries a group_key (= event_id), so the Grouper held spotting for the full grouper_window_seconds (60s live) before it could even reach the dispatcher -- a ~60s latency FLOOR, and ~120s+ once FirePacer queuing is added on top. Add a module-level _NEVER_COALESCE_CATEGORIES frozenset and bypass the coalescing window for the categories in it. Currently: wildfire_spotting only. This is safe because spotting is already rate-limited AT THE SOURCE: a per-fire 1h cooldown (adapter_config.fires.spotting_cooldown_seconds, default 3600, latched on fires.last_spotting_broadcast_at per irwin_id in gating/firms.py) gates spotting DETECTION itself, so N active fires yield at most N spotting alerts per hour. FirePacer (60s interval, with head-of-line ordering for immediate severity) and the dispatcher's per-(toggle, category, region) cooldown still apply downstream. The bypass is scoped by CATEGORY, never by severity. Commit 85d48ce3 deliberately removed a severity == "immediate" bypass from this exact spot because ALL fire events carry _severity_override="immediate", so a severity bypass exempts the entire fire family from rate control. The comment on the constant spells that out so it does not get re-added. Tests: spotting with a group_key passes straight through; wildfire_growth and wildfire_incident at immediate severity are STILL held (proving no severity bypass crept back in); spotting with no group_key still passes through. PR #129's test_immediate_severity_is_also_coalesced_no_bypass is untouched and still passes. Co-authored-by: Matt Johnson Co-authored-by: Claude Opus 4.8 (1M context) --- work/meshai/notifications/pipeline/grouper.py | 38 ++++++++- work/tests/test_pipeline_grouper.py | 85 +++++++++++++++++++ 2 files changed, 122 insertions(+), 1 deletion(-) diff --git a/work/meshai/notifications/pipeline/grouper.py b/work/meshai/notifications/pipeline/grouper.py index ce42a12..26ce187 100644 --- a/work/meshai/notifications/pipeline/grouper.py +++ b/work/meshai/notifications/pipeline/grouper.py @@ -17,6 +17,39 @@ from typing import Callable from meshai.notifications.events import Event +# Categories that skip the coalescing window entirely, regardless of +# group_key. This is a CATEGORY allowlist, not a severity one -- keep +# it that way. +# +# wildfire_spotting is exempt because it is a life-safety signal (a +# fire throwing embers past its own containment line) where the +# grouper's hold window (adapter_config.pipeline.grouper_window_seconds, +# 60s live) plus FirePacer queuing would add a ~60-120s latency floor +# on the single most urgent alert in the system. It is safe to bypass +# because spotting is ALREADY rate-limited at the source: a per-fire +# 1-hour cooldown gates spotting *detection* itself +# (adapter_config.fires.spotting_cooldown_seconds, default 3600s, +# latched on fires.last_spotting_broadcast_at per irwin_id in +# gating/firms.py). N active fires therefore yield at most N spotting +# alerts per hour, independent of this bypass. The FirePacer (60s +# interval, head-of-line ordering for "immediate" severity) and the +# dispatcher's per-(toggle, CATEGORY, region) cooldown still apply +# downstream, so nothing here is unbounded. +# +# DO NOT add a severity-based bypass here (e.g. "if event.severity == +# 'immediate'"). Commit 85d48ce3 ("fix(fire): remove immediate-severity +# exemption from grouper + cooldown") deliberately deleted exactly that +# check: EVERY fire event carries _severity_override="immediate", so a +# severity bypass exempts the entire fire family (growth, containment, +# declared, halted -- not just spotting) from both the grouper window +# AND effectively defeats rate control, reopening the fire-broadcast- +# spam hole on a public radio mesh. This has already been misdiagnosed +# and "fixed" that wrong way twice. Any future urgency-driven bypass +# MUST be scoped by event.category (added to this frozenset), never by +# event.severity. +_NEVER_COALESCE_CATEGORIES = frozenset({"wildfire_spotting"}) + + class Grouper: """Coalesce same-group_key events inside a window.""" @@ -96,11 +129,14 @@ class Grouper: """Process an event. Events without group_key pass through immediately. + Events whose category is in _NEVER_COALESCE_CATEGORIES also pass + through immediately, group_key or not (see module docstring for + why -- category-scoped only, never severity-scoped). Events with group_key are held, replacing any prior held event with the same group_key. The held event is emitted later via tick(). """ - if not event.group_key: + if not event.group_key or event.category in _NEVER_COALESCE_CATEGORIES: self._next(event) return diff --git a/work/tests/test_pipeline_grouper.py b/work/tests/test_pipeline_grouper.py index b8a1061..02a4d60 100644 --- a/work/tests/test_pipeline_grouper.py +++ b/work/tests/test_pipeline_grouper.py @@ -94,3 +94,88 @@ def test_priority_is_also_coalesced_not_bypassed(): g.handle(_ev("priority")) assert rec.received == [] assert g.held_count() == 1 + + +def test_wildfire_spotting_bypasses_the_grouper_even_with_group_key(): + """wildfire_spotting is a CATEGORY-scoped bypass (owner-approved): it + skips the coalescing window entirely, even though it carries a + group_key that would otherwise hold it. This is the one narrow + exemption -- see _NEVER_COALESCE_CATEGORIES in grouper.py for why + it's safe (source-side 1h per-fire cooldown) and why it must stay + category-scoped, not severity-scoped.""" + rec = Recorder() + g = Grouper(next_handler=rec.handle, window_seconds=60.0) + ev = make_event( + source="firms", + category="wildfire_spotting", + severity="immediate", + title="test spotting", + lat=42.6, + lon=-114.5, + group_key="fire-irwin-123", + ) + g.handle(ev) + # Passed straight through -- NOT held for the coalescing window. + assert len(rec.received) == 1 + assert rec.received[0].category == "wildfire_spotting" + assert g.held_count() == 0 + + +def test_wildfire_growth_immediate_is_still_held_no_severity_bypass_crept_in(): + """A same-severity, same-fire-family event of a DIFFERENT category + (wildfire_growth, not wildfire_spotting) must still be coalesced. + This proves the new bypass is scoped to category and did not + accidentally reintroduce a severity-based bypass (the exact bug + commit 85d48ce3 removed).""" + rec = Recorder() + g = Grouper(next_handler=rec.handle, window_seconds=60.0) + ev = make_event( + source="wfigs", + category="wildfire_growth", + severity="immediate", + title="test growth", + lat=42.6, + lon=-114.5, + group_key="fire-irwin-123", + ) + g.handle(ev) + assert rec.received == [] + assert g.held_count() == 1 + + +def test_wildfire_incident_immediate_is_still_held_no_severity_bypass_crept_in(): + """Same as above for wildfire_incident: only wildfire_spotting bypasses, + every other fire category (even at immediate severity) is coalesced.""" + rec = Recorder() + g = Grouper(next_handler=rec.handle, window_seconds=60.0) + ev = make_event( + source="wfigs", + category="wildfire_incident", + severity="immediate", + title="test incident", + lat=42.6, + lon=-114.5, + group_key="fire-irwin-123", + ) + g.handle(ev) + assert rec.received == [] + assert g.held_count() == 1 + + +def test_wildfire_spotting_with_no_group_key_still_passes_through(): + """Existing no-group_key behavior is preserved for spotting too -- + the bypass condition is an `or`, not a replacement.""" + rec = Recorder() + g = Grouper(next_handler=rec.handle, window_seconds=60.0) + ev = make_event( + source="firms", + category="wildfire_spotting", + severity="immediate", + title="test spotting no key", + lat=42.6, + lon=-114.5, + group_key=None, + ) + g.handle(ev) + assert len(rec.received) == 1 + assert g.held_count() == 0