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

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