fix(pipeline): let wildfire_spotting skip the grouper (category-scoped)

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: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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
 

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