refactor(phase3b): migrate WFIGS fire to formatter+decider (tier-a)

Move wfigs wire rendering into notifications/formatters/fire.py and the
full fire state machine into notifications/gating/fire.py, behind the
registry, NO cutover. handle_wfigs builds a canonical dict, calls
decide(), keeps the inline fires INSERT/UPDATE of current_* and the
tombstoned_at stamp unconditional, then branches on is_cutover(...) —
legacy _attach_commit_handles/all-clear path stays byte-identical while
the new path bakes in shadow.

Reproduces every legacy stamp through GateResult.data_patch:
- forward-only acres/containment growth + 8h cooldown gating
- tombstone wildfire_closed all-clear (row exists AND last_broadcast_at
  IS NOT NULL) with _severity_override="priority", _dedup_suffix="closed"
- growth _dedup_suffix=f"{acres}|{contained_pct}", _cooldown_suffix=irwin_id
- idempotent commit UPSERT of fires(last_broadcast_*) + event_log flip
- full _location_anchor fallback chain (geocoder_city -> resolve_anchor
  -> landclass -> county -> state) preserved in the formatter

Registered under the three explicit categories (wildfire_declared,
wildfire_incident, wildfire_closed) rather than the `fire` toggle, so the
family-fallback does NOT capture the still-deferred FIRMS categories
(wildfire_hotspot/new_ignition/wildfire_growth); a registration test
asserts those resolve elsewhere.

Native env/fires.py deferred (missing IRWIN/cause/landclass, no tombstone
concept); non-cutover so store._emit_event won't run it.

tier-a: 19 new golden+gate-sequence tests; wfigs handler 23/23 preserved;
suite at 34-failure baseline (1571 passed).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

work/meshai/central/wfigs_handler.py

@@ -83,16 +83,55 @@ def _fire_too_old_to_announce(declared_at_epoch, now) -> bool:
 # ---------- public entry --------------------------------------------------
 
 
+def _build_canonical(normalized: dict, kind: str) -> dict:
+    """Flat canonical dict consumed by gating.fire.decide + formatters.fire.
+
+    Carries the render-ready WFIGS fields the decider (gating decision) and the
+    formatter (wire) read.  ``_kind`` routes the decider between the active-fire
+    state machine, the tombstone all-clear, and the never-broadcast perimeter.
+    """
+    return {
+        "_kind": kind,
+        "irwin_id": normalized.get("irwin_id"),
+        "incident_name": normalized.get("incident_name"),
+        "incident_type": normalized.get("incident_type"),
+        "acres": normalized.get("acres"),
+        "contained_pct": normalized.get("contained_pct"),
+        "fire_cause": normalized.get("fire_cause"),
+        "declared_at_epoch": normalized.get("declared_at_epoch"),
+        "unique_fire_id": normalized.get("unique_fire_id"),
+        "lat": normalized.get("lat"),
+        "lon": normalized.get("lon"),
+        "county": normalized.get("county"),
+        "state": normalized.get("state"),
+        "landclass": normalized.get("landclass"),
+        "geocoder_city": normalized.get("geocoder_city"),
+    }
+
+
 def handle_wfigs(normalized: dict, envelope: dict, subject: str,
                   data: Optional[dict] = None,
                   now: Optional[int] = None) -> Optional[str]:
     """Route a normalized WFIGS dict through persistence + change-detection.
 
+    Phase-3b refactor: the broadcast DECISION (New/Update/suppress + cooldown +
+    age-gate + all-clear eligibility) now lives in
+    ``meshai.notifications.gating.fire.decide``.  This handler keeps ownership
+    of the inline ``fires`` INSERT/UPDATE of ``current_*`` (unconditional state
+    write), the ``tombstoned_at`` stamp, the ``event_log`` row + handled flip,
+    and the wire it returns (mirror quake/nwis).
+
     `data` is the mutable dict the caller (consumer._normalize) is composing
     into the Event. When a broadcast should fire, the handler attaches an
     `_on_broadcast_committed` callback and `_broadcast_audit` descriptor to
     it; the dispatcher invokes both AFTER a successful deliver().
 
+    Cutover gate: when the emitted category is explicitly cut over, the handler
+    writes ``gate.data_patch`` + wraps ``gate.commit`` (new path live);
+    otherwise it keeps the legacy ``_attach_commit_handles`` / all-clear
+    stamping VERBATIM so the live broadcast stays byte-for-byte identical while
+    the new formatter+decider bake in shadow.
+
     Returns a wire string when a broadcast should fire, None otherwise.
     """
     if not isinstance(normalized, dict):
@@ -114,6 +153,11 @@ def handle_wfigs(normalized: dict, envelope: dict, subject: str,
                           "deferring to default pipeline")
         return None
 
+    from meshai.notifications.cutover import is_cutover
+    from meshai.notifications.gating.fire import decide as _gate_decide
+
+    canonical = _build_canonical(normalized, kind)
+
     if kind in ("wfigs_tombstone", "wfigs_perimeter"):
         source = "wfigs_incidents" if kind == "wfigs_tombstone" else "wfigs_perimeters"
         log_id = _log_event_returning_id(
@@ -124,6 +168,7 @@ def handle_wfigs(normalized: dict, envelope: dict, subject: str,
         # v0.6-tail item 4: tombstone branch stamps fires.tombstoned_at so
         # the ReminderScheduler stops re-broadcasting the closed fire.
         # Only the tombstone kind closes the fire; perimeter polls don t.
+        # UNCONDITIONAL state write — stays inline, mirror the original.
         if kind == "wfigs_tombstone" and irwin_id:
             try:
                 conn.execute(
@@ -136,13 +181,17 @@ def handle_wfigs(normalized: dict, envelope: dict, subject: str,
 
         # All-clear broadcast: only fires that previously made it to mesh
         # get a closure message. Silent for fires that were never broadcast.
+        # The DECISION (row exists AND last_broadcast_at NOT NULL) is delegated
+        # to the decider; the WIRE is still rendered here from the fire row for
+        # byte-identity, and the not-cutover data stamping stays verbatim.
         if kind == "wfigs_tombstone" and irwin_id:
-            fire_row = conn.execute(
-                "SELECT incident_name, current_acres, current_contained_pct, "
-                "last_broadcast_at, county, state, lat, lon "
-                "FROM fires WHERE irwin_id = ?", (irwin_id,)
-            ).fetchone()
-            if fire_row is not None and fire_row["last_broadcast_at"] is not None:
+            gate = _gate_decide(canonical, source="wfigs", now=float(now))
+            if gate.broadcast:
+                fire_row = conn.execute(
+                    "SELECT incident_name, current_acres, current_contained_pct, "
+                    "last_broadcast_at, county, state, lat, lon "
+                    "FROM fires WHERE irwin_id = ?", (irwin_id,)
+                ).fetchone()
                 name = fire_row["incident_name"] or "(unnamed fire)"
                 # Build line 2 parts
                 parts = []
@@ -163,15 +212,36 @@ def handle_wfigs(normalized: dict, envelope: dict, subject: str,
                     lines.append(" | ".join(parts))
                 wire = "\n".join(lines)
                 if isinstance(data, dict):
-                    data["category"] = "wildfire_closed"
-                    data["_severity_override"] = "priority"
-                _attach_commit_handles(
-                    data, irwin_id=irwin_id,
-                    acres=fire_row["current_acres"],
-                    contained_pct=fire_row["current_contained_pct"],
-                    event_log_row_id=log_id)
-                if isinstance(data, dict):
-                    data["_dedup_suffix"] = "closed"
+                    if is_cutover("wildfire_closed"):
+                        # NEW PATH: formatter re-renders from data_patch fields.
+                        data.update(gate.data_patch)
+                        data["_broadcast_audit"] = {"table": "fires", "pk": irwin_id}
+                        _raw_commit = gate.commit
+                        _log_row_id = log_id
+
+                        def _on_commit(committed_at: float) -> None:
+                            if _raw_commit is not None:
+                                _raw_commit(committed_at)
+                            if _log_row_id is not None:
+                                try:
+                                    get_db().execute(
+                                        "UPDATE event_log SET handled=1 WHERE id=?",
+                                        (int(_log_row_id),))
+                                except Exception:
+                                    logger.exception(
+                                        "wfigs closed commit: event_log update failed")
+
+                        data["_on_broadcast_committed"] = _on_commit
+                    else:
+                        # LEGACY verbatim (byte-for-byte identical live output).
+                        data["category"] = "wildfire_closed"
+                        data["_severity_override"] = "priority"
+                        _attach_commit_handles(
+                            data, irwin_id=irwin_id,
+                            acres=fire_row["current_acres"],
+                            contained_pct=fire_row["current_contained_pct"],
+                            event_log_row_id=log_id)
+                        data["_dedup_suffix"] = "closed"
                 return wire
 
         return None
@@ -187,15 +257,19 @@ def handle_wfigs(normalized: dict, envelope: dict, subject: str,
         subject=subject, handled=0,
         table_name="fires", table_pk=irwin_id)
 
-    row = conn.execute(
-        "SELECT current_acres, current_contained_pct, last_broadcast_at, "
-        "last_broadcast_acres, last_broadcast_contained "
-        "FROM fires WHERE irwin_id = ?", (irwin_id,)).fetchone()
-
     acres = normalized.get("acres")
     contained_pct = normalized.get("contained_pct")
 
-    # ---- (i) row missing -- INSERT, mark "New", but DO NOT set last_broadcast_*
+    # Delegate the New/Update/suppress + age-gate + cooldown decision. The
+    # decider READS the fires row (pre-write) exactly as the original inline
+    # branch did; the inline INSERT/UPDATE below stays handler-owned.
+    gate = _gate_decide(canonical, source="wfigs", now=float(now))
+
+    # ---- inline state write (UNCONDITIONAL) -- INSERT or UPDATE current_*.
+    # Re-read the row (same pre-write state the decider saw) to pick the write.
+    row = conn.execute(
+        "SELECT last_broadcast_at FROM fires WHERE irwin_id = ?",
+        (irwin_id,)).fetchone()
     if row is None:
         conn.execute(
             "INSERT INTO fires(irwin_id, incident_name, incident_type, "
@@ -217,27 +291,7 @@ def handle_wfigs(normalized: dict, envelope: dict, subject: str,
                 None, None, None,  # last_broadcast_* explicitly NULL
             ),
         )
-        # Step 3 age-gate: keep the INSERT (so genuine future Updates work) but
-        # suppress the "New" broadcast for fires whose declared_at is too old.
-        if _fire_too_old_to_announce(normalized.get("declared_at_epoch"), now):
-            return None
-        wire = _render(normalized, prefix="New")
-        # v0.7-fire-tracker-1: tag first-sight broadcasts with the new
-        # wildfire_declared category so the dispatcher rules them apart
-        # from acres/containment updates (wildfire_incident).
-        if isinstance(data, dict):
-            data["category"] = "wildfire_declared"
-        # v0.6-3c: severity override for fire broadcasts (downgraded from
-        # immediate to priority to prevent cooldown/grouper bypass)
-        if isinstance(data, dict):
-            data["_severity_override"] = "priority"
-        _attach_commit_handles(data, irwin_id=irwin_id,
-                                 acres=acres, contained_pct=contained_pct,
-                                 event_log_row_id=log_id)
-        return wire
-
-    # ---- (ii) row exists but never broadcast -- UPDATE current_*, prefix="New"
-    if row["last_broadcast_at"] is None:
+    else:
         conn.execute(
             "UPDATE fires SET current_acres=?, current_contained_pct=?, "
             "lat=COALESCE(?, lat), lon=COALESCE(?, lon), last_event_at=? "
@@ -245,72 +299,57 @@ def handle_wfigs(normalized: dict, envelope: dict, subject: str,
             (acres, contained_pct, normalized.get("lat"),
              normalized.get("lon"), now, irwin_id),
         )
-        # Step 3 age-gate: keep the UPDATE but suppress the "New" broadcast for
-        # fires whose declared_at is too old (closed/stale-fire resurrection).
-        if _fire_too_old_to_announce(normalized.get("declared_at_epoch"), now):
-            return None
-        wire = _render(normalized, prefix="New")
-        # v0.7-fire-tracker-1: case-(ii) is also first-sight as far as
-        # broadcast history goes -- the row exists because some prior
-        # handler call ran but no actual broadcast went out.
-        if isinstance(data, dict):
-            data["category"] = "wildfire_declared"
-        # v0.6-3c: severity override for fire broadcasts (downgraded from
-        # immediate to priority to prevent cooldown/grouper bypass)
-        if isinstance(data, dict):
+
+    if not gate.broadcast:
+        # Case-(iii) suppress (no forward change OR inside cooldown) ran the
+        # stale-fire cleanup in the original; the age-gate suppress did not.
+        if gate.lifecycle == "cooldown":
+            _cleanup_stale_fires(conn)
+        return None
+
+    # ---- broadcast: render the wire (back-compat) + stamp data.
+    prefix = "Update" if gate.lifecycle == "update" else "New"
+    wire = _render(normalized, prefix=prefix,
+                    last_bcast_acres=gate.data_patch.get("last_bcast_acres"),
+                    last_bcast_contained=gate.data_patch.get("last_bcast_contained"))
+
+    # Cutover gate keys on the category THIS broadcast carries: New first-sight
+    # is wildfire_declared, growth Update stays wildfire_incident.
+    _cut = (is_cutover("wildfire_incident") if gate.lifecycle == "update"
+            else is_cutover("wildfire_declared"))
+    if isinstance(data, dict):
+        if _cut:
+            # NEW PATH: formatter re-renders from canonical + data_patch.
+            data.update(canonical)
+            data.update(gate.data_patch)
+            data["_broadcast_audit"] = {"table": "fires", "pk": irwin_id}
+            _raw_commit = gate.commit
+            _log_row_id = log_id
+
+            def _on_commit(committed_at: float) -> None:
+                if _raw_commit is not None:
+                    _raw_commit(committed_at)
+                if _log_row_id is not None:
+                    try:
+                        get_db().execute(
+                            "UPDATE event_log SET handled=1 WHERE id=?",
+                            (int(_log_row_id),))
+                    except Exception:
+                        logger.exception(
+                            "wfigs commit: event_log update failed")
+
+            data["_on_broadcast_committed"] = _on_commit
+        else:
+            # LEGACY verbatim (byte-for-byte identical live output).
+            # New first-sight tags wildfire_declared; Update keeps the
+            # envelope-derived wildfire_incident (no category override).
+            if gate.lifecycle != "update":
+                data["category"] = "wildfire_declared"
             data["_severity_override"] = "priority"
-        _attach_commit_handles(data, irwin_id=irwin_id,
-                                 acres=acres, contained_pct=contained_pct,
-                                 event_log_row_id=log_id)
-        return wire
-
-    # ---- (iii) row exists AND already broadcast -- gate on change + 8h cooldown
-    conn.execute(
-        "UPDATE fires SET current_acres=?, current_contained_pct=?, "
-        "lat=COALESCE(?, lat), lon=COALESCE(?, lon), last_event_at=? "
-        "WHERE irwin_id=?",
-        (acres, contained_pct, normalized.get("lat"),
-         normalized.get("lon"), now, irwin_id),
-    )
-
-    last_bcast_at = row["last_broadcast_at"]
-    last_bcast_acres = row["last_broadcast_acres"]
-    last_bcast_contained = row["last_broadcast_contained"]
-
-    # Forward-only change detection: more acres or higher containment counts.
-    # Downward revisions and unchanged values do not warrant re-broadcast.
-    # v0.6-3b: each axis can be silenced via adapter_config toggles.
-    changed_acres = (
-        bool(adapter_config.wfigs.broadcast_on_acres)
-        and acres is not None
-        and (last_bcast_acres is None or acres > last_bcast_acres)
-    )
-    changed_contained = (
-        bool(adapter_config.wfigs.broadcast_on_contained)
-        and contained_pct is not None
-        and (last_bcast_contained is None or contained_pct > last_bcast_contained)
-    )
-    cooldown_s = int(adapter_config.wfigs.cooldown_seconds)
-    eight_hours_passed = (
-        last_bcast_at is None
-        or (now - int(last_bcast_at) >= cooldown_s)
-    )
-
-    if (changed_acres or changed_contained) and eight_hours_passed:
-        wire = _render(normalized, prefix="Update",
-                        last_bcast_acres=last_bcast_acres,
-                        last_bcast_contained=last_bcast_contained)
-        # v0.6-3c: severity override for fire updates (downgraded from
-        # immediate to priority to prevent cooldown/grouper bypass)
-        if isinstance(data, dict):
-            data["_severity_override"] = "priority"
-        _attach_commit_handles(data, irwin_id=irwin_id,
-                                 acres=acres, contained_pct=contained_pct,
-                                 event_log_row_id=log_id)
-        return wire
-
-    _cleanup_stale_fires(conn)
-    return None
+            _attach_commit_handles(data, irwin_id=irwin_id,
+                                     acres=acres, contained_pct=contained_pct,
+                                     event_log_row_id=log_id)
+    return wire
 
 
 # ---------- commit-callback factory ---------------------------------------

work/meshai/notifications/formatters/__init__.py

@@ -94,3 +94,17 @@ register("traffic_congestion", _incident_fmt_mod.format)
 # stream_high_water) are deferred — see gating/__init__.py note.
 from meshai.notifications.formatters import hydro as _hydro_fmt_mod  # noqa: E402,F401
 register("stream_flow", _hydro_fmt_mod.format)
+
+# Phase-3b: WFIGS wildfire. The Central wfigs_handler emits THREE explicit
+# categories, all under toggle "fire": `wildfire_declared` (first-sight New,
+# cases i/ii), `wildfire_incident` (growth/containment Update, case iii — the
+# envelope-derived category with no override), and `wildfire_closed` (tombstone
+# all-clear). One formatter handles all three; event.category selects the wire
+# shape (wildfire_closed -> all-clear, else incident/growth). Registered under
+# the explicit strings — NOT the "fire" toggle — so FIRMS categories
+# (wildfire_hotspot / new_ignition / unattributed_hotspot_cluster /
+# wildfire_growth) are NOT captured by the family fallback. FIRMS is deferred.
+from meshai.notifications.formatters import fire as _fire_fmt_mod  # noqa: E402,F401
+register("wildfire_declared", _fire_fmt_mod.format)
+register("wildfire_incident", _fire_fmt_mod.format)
+register("wildfire_closed",   _fire_fmt_mod.format)

work/meshai/notifications/formatters/fire.py

@@ -0,0 +1,207 @@
+"""WFIGS wildfire formatter — Phase-3b migration.
+
+Reproduces BOTH legacy WFIGS wire shapes byte-identically, reading the canonical
+schema the Central path writes into event.data on broadcast:
+
+  (a) Active incident / growth — mirrors ``wfigs_handler._render`` exactly:
+        Line 1: 🔥 {name} — {New|Update}
+        Line 2: {acres} ac{ (+delta)} · containment {pct}%
+        Line 3: {movement line}  OR  {anchor}
+        Line 4: Cause: {cause} · Discovered {Mon D}   (either/both/neither)
+
+  (b) All-clear ("contained & closed") — mirrors the tombstone branch exactly:
+        Line 1: ✅ {name} — contained & closed
+        Line 2: {acres} ac | {pct}% contained | {anchor}   (only present parts)
+
+Branch selection (event.data):
+    category == "wildfire_closed"  OR  _kind == "wfigs_tombstone"  -> all-clear
+    otherwise                                                      -> incident
+
+Canonical schema consumed (active incident):
+    incident_name, acres, contained_pct, fire_cause, declared_at_epoch,
+    unique_fire_id, lat, lon, county, state, landclass, geocoder_city,
+    movement (FIRMS-injected {direction, speed_mph}, else absent/None),
+    is_update, last_bcast_acres, last_bcast_contained  (decider render hints)
+
+Canonical schema consumed (all-clear):
+    incident_name, acres, contained_pct, lat, lon, county, state
+
+Anchor resolution (``_fire_anchor``) reproduces ``wfigs_handler._location_anchor``
+tier-for-tier: geocoder_city → curated town_anchors / Photon nearest_town (via
+the shared ``resolve_anchor`` helper, re-formatted to the legacy string) →
+landclass → "{county} Co {state}" → state → "(location unknown)".  The extra
+fallback tiers around ``resolve_anchor`` are required for byte-identity because
+``resolve_anchor`` covers only the town step.
+
+Time contract: ``now`` is accepted but unused (the discovery date is rendered
+from ``declared_at_epoch`` in a fixed UTC-6 offset, exactly as ``_render``).
+``budget`` is injected — the caller supplies ``budget_for("wfigs")``.
+"""
+from __future__ import annotations
+
+import datetime as _dt
+import logging
+from typing import TYPE_CHECKING, Optional
+
+from meshai.adapter_config import adapter_config
+from meshai.notifications.formatters._anchor import resolve_anchor
+from meshai.notifications.formatters._budget import fit_to_budget
+
+if TYPE_CHECKING:
+    from meshai.notifications.events import Event
+
+logger = logging.getLogger(__name__)
+
+
+def _fire_anchor(d: dict) -> str:
+    """Byte-identical replica of ``wfigs_handler._location_anchor``.
+
+    geocoder.city > nearest town (curated town_anchors, then Photon) >
+    landclass > "{county} Co {state}" > state > "(location unknown)".
+    """
+    city = d.get("geocoder_city")
+    if city:
+        return str(city)
+
+    lat = d.get("lat")
+    lon = d.get("lon")
+    if isinstance(lat, (int, float)) and isinstance(lon, (int, float)):
+        try:
+            max_mi = float(adapter_config.wfigs.anchor_max_mi)
+        except Exception:
+            max_mi = 100.0
+        try:
+            res = resolve_anchor(lat, lon, max_mi=max_mi)
+        except Exception:
+            logger.debug("fire anchor: resolve_anchor failed; falling through")
+            res = None
+        if res and res.get("town"):
+            # Legacy applies .title() in BOTH the town_anchors and nearest_town
+            # arms; resolve_anchor titles only the town_anchors arm, so title
+            # here unconditionally (idempotent for already-titled names).
+            town = str(res["town"]).title()
+            dist = res.get("distance_mi")
+            bearing = res.get("bearing")
+            if isinstance(dist, (int, float)):
+                if dist < 1:
+                    return f"near {town}"
+                return f"{int(round(dist))} mi {bearing or ''} of {town}".strip()
+            return f"near {town}"
+
+    landclass = d.get("landclass")
+    if landclass:
+        return str(landclass)
+
+    county = d.get("county")
+    state = d.get("state")
+    if county and state:
+        return f"{county} Co {state}"
+    if state:
+        return str(state)
+    return "(location unknown)"
+
+
+def _render_incident(d: dict, budget: int) -> str:
+    """Byte-identical replica of ``wfigs_handler._render`` (active incident)."""
+    name = d.get("incident_name") or "(unnamed)"
+    acres = d.get("acres")
+    contained_pct = d.get("contained_pct")
+    cause = d.get("fire_cause")
+    declared_at_epoch = d.get("declared_at_epoch")
+    movement = d.get("movement")
+    is_update = bool(d.get("is_update"))
+    last_bcast_acres = d.get("last_bcast_acres")
+    prefix = "Update" if is_update else "New"
+    anchor = _fire_anchor(d)
+
+    lines: list[str] = []
+
+    # Line 1: header
+    lines.append(f"🔥 {name} — {prefix}")
+
+    # Line 2: size / containment with delta (plain text — no bold markdown).
+    acres_str = f"{int(acres):,} ac" if acres is not None else "size unknown"
+    delta_str = ""
+    if (prefix == "Update" and last_bcast_acres is not None
+            and acres is not None and acres > last_bcast_acres):
+        delta_str = f" (+{int(acres - last_bcast_acres):,})"
+    contained_str = (f"containment {int(contained_pct)}%"
+                     if contained_pct is not None else "containment unknown")
+    lines.append(f"{acres_str}{delta_str} · {contained_str}")
+
+    # Line 3: movement or plain anchor.
+    if (isinstance(movement, dict)
+            and movement.get("direction") and movement.get("speed_mph") is not None):
+        lines.append(f"Moving {movement['direction']} {movement['speed_mph']:.1f} mi/h · {anchor}")
+    else:
+        lines.append(f"{anchor}")
+
+    # Line 4: cause / discovered (DATE ONLY — no time-of-day).
+    cause_part = cause if cause else None
+    disc_part = None
+    if declared_at_epoch is not None:
+        try:
+            dt = _dt.datetime.fromtimestamp(
+                declared_at_epoch,
+                tz=_dt.timezone(_dt.timedelta(hours=-6)))
+            disc_part = dt.strftime("%b %-d")
+        except Exception:
+            pass
+    if cause_part and disc_part:
+        lines.append(f"Cause: {cause_part} · Discovered {disc_part}")
+    elif cause_part:
+        lines.append(f"Cause: {cause_part}")
+    elif disc_part:
+        lines.append(f"Discovered {disc_part}")
+
+    return fit_to_budget("\n".join(lines), budget)
+
+
+def _render_allclear(d: dict, budget: int) -> str:
+    """Byte-identical replica of the ``wfigs_handler`` tombstone all-clear wire."""
+    name = d.get("incident_name") or "(unnamed fire)"
+    parts: list[str] = []
+    acres = d.get("acres")
+    contained_pct = d.get("contained_pct")
+    if acres is not None:
+        parts.append(f"{int(acres):,} ac")
+    if contained_pct is not None:
+        parts.append(f"{int(contained_pct)}% contained")
+    # Location via the same anchor chain, but the tombstone branch feeds ONLY
+    # {lat, lon, county, state} (no geocoder_city / landclass).
+    loc_dict = {
+        "lat": d.get("lat"), "lon": d.get("lon"),
+        "county": d.get("county"), "state": d.get("state"),
+    }
+    anchor = _fire_anchor(loc_dict)
+    if anchor and anchor != "(location unknown)":
+        parts.append(anchor)
+    lines = [f"✅ {name} — contained & closed"]
+    if parts:
+        lines.append(" | ".join(parts))
+    return fit_to_budget("\n".join(lines), budget)
+
+
+def format(event: "Event", *, now: float, budget: int) -> str:
+    """Render the WFIGS wire string from canonical event.data.
+
+    Args:
+        event:  Pipeline Event — reads from event.data (canonical schema).
+        now:    Frozen-clock epoch (seam; not used in current rendering).
+        budget: Mesh-packet character budget (from budget_for("wfigs")).
+
+    Returns:
+        UTF-8 string fitting within *budget* characters.
+    """
+    d = event.data or {}
+    category = None
+    try:
+        category = event.category
+    except Exception:
+        category = None
+    if category is None:
+        category = d.get("category")
+
+    if category == "wildfire_closed" or d.get("_allclear") or d.get("_kind") == "wfigs_tombstone":
+        return _render_allclear(d, budget)
+    return _render_incident(d, budget)

work/meshai/notifications/gating/__init__.py

@@ -86,3 +86,17 @@ register("traffic_congestion", _incident_gate_mod.decide)
 # NOT cut over, store._emit_event's native decider hook won't run it.
 from meshai.notifications.gating import hydro as _hydro_gate_mod  # noqa: E402,F401
 register("stream_flow", _hydro_gate_mod.decide)
+
+# Phase-3b: WFIGS wildfire. Three explicit categories the wfigs_handler emits:
+# `wildfire_declared` (New), `wildfire_incident` (growth Update), and
+# `wildfire_closed` (tombstone all-clear). One decider handles all three; it
+# keys off canonical `_kind` (wfigs_incident / wfigs_tombstone / wfigs_perimeter)
+# to route New/Update/suppress vs the all-clear eligibility. Registered under
+# explicit strings (not the "fire" toggle) so FIRMS categories are untouched —
+# FIRMS + native env/fires.py are a deferred follow-up (env/fires.py lacks the
+# IRWIN/FireCause/landclass fields and has no tombstone concept, and since fire
+# is NOT cut over, store._emit_event's native decider hook won't run it).
+from meshai.notifications.gating import fire as _fire_gate_mod  # noqa: E402,F401
+register("wildfire_declared", _fire_gate_mod.decide)
+register("wildfire_incident", _fire_gate_mod.decide)
+register("wildfire_closed",   _fire_gate_mod.decide)

work/meshai/notifications/gating/fire.py

@@ -0,0 +1,256 @@
+"""WFIGS wildfire gating decider — Phase-3b migration.
+
+Moves the broadcast DECISION out of ``central.wfigs_handler.handle_wfigs`` (the
+New-vs-Update-vs-suppress state machine, the forward-only acres/containment
+growth gate + 8h cooldown, the stale-fire age-gate, and the tombstone all-clear
+eligibility) into a source-agnostic decider, mirroring quake/hydro.
+
+The Central handler keeps ownership of:
+  * the ``fires`` INSERT / UPDATE of ``current_*`` (unconditional state write),
+  * the ``tombstoned_at`` stamp,
+  * the ``event_log`` row + its handled flip,
+  * the wire rendering (``_render`` / all-clear) it returns for back-compat.
+
+This decider owns ONLY:
+  * the read of the ``fires`` row for the gating decision,
+  * the growth/cooldown/age math,
+  * the ``data_patch`` stamps every broadcast carries
+    (``category`` override, ``_severity_override``, ``_dedup_suffix``,
+    ``_cooldown_suffix``, and the render hints ``is_update`` /
+    ``last_bcast_acres`` / ``last_bcast_contained``),
+  * the deferred ``commit`` closure that UPSERTs
+    ``fires(last_broadcast_at, first_broadcast_at, last_broadcast_acres,
+    last_broadcast_contained)`` on confirmed delivery.
+
+Canonical ``data`` dict consumed (built by the handler from the normalized
+WFIGS dict):
+    _kind, irwin_id, incident_name, incident_type, acres, contained_pct,
+    fire_cause, declared_at_epoch, unique_fire_id, lat, lon, county, state,
+    landclass, geocoder_city
+
+Three ``_kind`` values are handled:
+    wfigs_incident   -> active-fire New/Update/suppress state machine
+    wfigs_tombstone  -> all-clear ("contained & closed") eligibility
+    wfigs_perimeter  -> never broadcasts (suppress)
+
+Lifecycle labels (drive handler-side side effects):
+    "new"      first-sight broadcast (INSERT or row-exists-never-broadcast)
+    "update"   forward-only growth/containment broadcast after cooldown
+    "closed"   tombstone all-clear broadcast
+    "cooldown" case-(iii) suppress (no change OR inside cooldown) — the handler
+               runs ``_cleanup_stale_fires`` for THIS suppress only
+    "suppress" every other suppress (age-gate, never-broadcast tombstone,
+               perimeter, unknown) — no cleanup
+"""
+from __future__ import annotations
+
+import logging
+from typing import Optional
+
+from meshai.adapter_config import adapter_config
+from meshai.notifications.gating.base import GateResult
+from meshai.persistence import get_db
+
+logger = logging.getLogger(__name__)
+
+
+def _fire_too_old_to_announce(declared_at_epoch, now) -> bool:
+    """Stale/closed-fire resurrection guard for first-announce ("New") only.
+
+    Verbatim copy of ``wfigs_handler._fire_too_old_to_announce``: re-reads the
+    knob at the use-site (GUI-live), fails OPEN (announces) when the gate is
+    disabled or the fire has no discovery date.
+    """
+    try:
+        max_age = int(adapter_config.wfigs.max_declare_age_seconds)
+    except Exception:
+        return False
+    if max_age <= 0 or declared_at_epoch is None:
+        return False
+    return (now - int(declared_at_epoch)) >= max_age
+
+
+def decide(data: dict, *, source: str, now: float) -> GateResult:
+    """Broadcast decision for a normalized WFIGS event.
+
+    Parameters
+    ----------
+    data:
+        Canonical WFIGS dict (see module docstring for schema).
+    source:
+        Adapter source name, e.g. "wfigs".
+    now:
+        Current epoch (from clock.now()) — determinism seam.
+
+    Returns
+    -------
+    GateResult with broadcast/lifecycle/data_patch/commit populated.  The
+    handler still performs the inline ``fires`` INSERT/UPDATE and renders the
+    wire; ``data_patch`` supplies the stamps + render hints, ``commit`` arms
+    the ``last_broadcast_*`` columns on confirmed delivery.
+    """
+    kind = data.get("_kind")
+    irwin_id = data.get("irwin_id")
+
+    try:
+        conn = get_db()
+    except Exception:
+        logger.exception("fire decide: persistence unavailable")
+        return GateResult(broadcast=False, lifecycle="suppress",
+                          reason="persistence unavailable")
+
+    # ── Tombstone all-clear ─────────────────────────────────────────────────
+    if kind == "wfigs_tombstone":
+        if not irwin_id:
+            return GateResult(broadcast=False, lifecycle="suppress",
+                              reason="tombstone without irwin_id")
+        fire_row = conn.execute(
+            "SELECT incident_name, current_acres, current_contained_pct, "
+            "last_broadcast_at, county, state, lat, lon "
+            "FROM fires WHERE irwin_id = ?", (irwin_id,)
+        ).fetchone()
+        # Only fires that previously reached the mesh get a closure message.
+        if fire_row is None or fire_row["last_broadcast_at"] is None:
+            return GateResult(broadcast=False, lifecycle="suppress",
+                              reason="tombstone for never-broadcast fire")
+        acres = fire_row["current_acres"]
+        contained_pct = fire_row["current_contained_pct"]
+        patch: dict = {
+            "category": "wildfire_closed",
+            "_severity_override": "priority",
+            # legacy _attach_commit_handles sets _cooldown_suffix=irwin_id then
+            # the handler overwrites _dedup_suffix with "closed".
+            "_cooldown_suffix": irwin_id,
+            "_dedup_suffix": "closed",
+            # render hints for the all-clear formatter (from the fire ROW, not
+            # the tombstone payload): name/size/loc — no geocoder_city/landclass
+            # so the anchor falls to resolve_anchor -> county exactly as legacy.
+            "_allclear": True,
+            "incident_name": fire_row["incident_name"],
+            "acres": acres,
+            "contained_pct": contained_pct,
+            "lat": fire_row["lat"],
+            "lon": fire_row["lon"],
+            "county": fire_row["county"],
+            "state": fire_row["state"],
+        }
+        return GateResult(
+            broadcast=True, lifecycle="closed",
+            reason=f"all-clear irwin={irwin_id}",
+            data_patch=patch,
+            commit=_make_commit(irwin_id, acres, contained_pct),
+        )
+
+    # ── Perimeter / unknown never broadcast ─────────────────────────────────
+    if kind != "wfigs_incident":
+        return GateResult(broadcast=False, lifecycle="suppress",
+                          reason=f"non-broadcast kind {kind}")
+
+    # ── Active incident state machine ───────────────────────────────────────
+    acres = data.get("acres")
+    contained_pct = data.get("contained_pct")
+    declared_at_epoch = data.get("declared_at_epoch")
+
+    row = conn.execute(
+        "SELECT current_acres, current_contained_pct, last_broadcast_at, "
+        "last_broadcast_acres, last_broadcast_contained "
+        "FROM fires WHERE irwin_id = ?", (irwin_id,)).fetchone()
+
+    # Stamps every active broadcast (New + Update) carries. category override
+    # is added only for New (Update keeps the envelope-derived wildfire_incident).
+    def _broadcast_patch(*, is_update: bool,
+                          last_bcast_acres=None, last_bcast_contained=None,
+                          new_category: bool) -> dict:
+        p = {
+            "_severity_override": "priority",
+            "_dedup_suffix": f"{acres}|{contained_pct}",
+            "_cooldown_suffix": irwin_id,
+            "is_update": is_update,
+            "last_bcast_acres": last_bcast_acres,
+            "last_bcast_contained": last_bcast_contained,
+        }
+        if new_category:
+            p["category"] = "wildfire_declared"
+        return p
+
+    # (i) row missing  &  (ii) row exists but never broadcast -> "New".
+    if row is None or row["last_broadcast_at"] is None:
+        # Age-gate suppresses only the "New" BROADCAST; the handler still writes
+        # current_* unconditionally. Suppress carries NO stamps (legacy returns
+        # None before setting category/severity) and does NOT trigger cleanup.
+        if _fire_too_old_to_announce(declared_at_epoch, now):
+            return GateResult(broadcast=False, lifecycle="suppress",
+                              reason="new fire too old to announce")
+        return GateResult(
+            broadcast=True, lifecycle="new",
+            reason="first sighting",
+            data_patch=_broadcast_patch(is_update=False, new_category=True),
+            commit=_make_commit(irwin_id, acres, contained_pct),
+        )
+
+    # (iii) row exists AND already broadcast -> forward-only growth + cooldown.
+    last_bcast_at = row["last_broadcast_at"]
+    last_bcast_acres = row["last_broadcast_acres"]
+    last_bcast_contained = row["last_broadcast_contained"]
+
+    changed_acres = (
+        bool(adapter_config.wfigs.broadcast_on_acres)
+        and acres is not None
+        and (last_bcast_acres is None or acres > last_bcast_acres)
+    )
+    changed_contained = (
+        bool(adapter_config.wfigs.broadcast_on_contained)
+        and contained_pct is not None
+        and (last_bcast_contained is None or contained_pct > last_bcast_contained)
+    )
+    cooldown_s = int(adapter_config.wfigs.cooldown_seconds)
+    eight_hours_passed = (
+        last_bcast_at is None
+        or (now - int(last_bcast_at) >= cooldown_s)
+    )
+
+    if (changed_acres or changed_contained) and eight_hours_passed:
+        return GateResult(
+            broadcast=True, lifecycle="update",
+            reason="forward-only growth after cooldown",
+            data_patch=_broadcast_patch(
+                is_update=True,
+                last_bcast_acres=last_bcast_acres,
+                last_bcast_contained=last_bcast_contained,
+                new_category=False),
+            commit=_make_commit(irwin_id, acres, contained_pct),
+        )
+
+    # No change OR inside cooldown -> suppress; handler runs _cleanup_stale_fires
+    # for this case only (legacy did the cleanup in the case-(iii) suppress arm).
+    return GateResult(broadcast=False, lifecycle="cooldown",
+                      reason="no forward change or inside cooldown")
+
+
+def _make_commit(irwin_id: str, acres, contained_pct):
+    """Build the deferred commit closure: idempotent UPSERT of the
+    ``fires`` broadcast-state columns.  Mirrors
+    ``wfigs_handler._attach_commit_handles._on_commit`` exactly (minus the
+    event_log flip, which the handler wraps back on)."""
+
+    def _commit(committed_at: float) -> None:
+        try:
+            conn = get_db()
+        except Exception:
+            logger.exception(
+                "fire commit: persistence unavailable; last_broadcast_* not "
+                "updated for irwin=%s", irwin_id)
+            return
+        try:
+            conn.execute(
+                "UPDATE fires SET last_broadcast_at=?, "
+                "first_broadcast_at=COALESCE(first_broadcast_at, ?), "
+                "last_broadcast_acres=?, last_broadcast_contained=? "
+                "WHERE irwin_id=?",
+                (int(committed_at), int(committed_at), acres, contained_pct,
+                 irwin_id),
+            )
+        except Exception:
+            logger.exception("fire commit: fires UPSERT failed irwin=%s", irwin_id)
+
+    return _commit

work/tests/test_fire_refactor.py

@@ -0,0 +1,320 @@
+"""Phase-3b WFIGS wildfire refactor tests.
+
+Verifies the source-agnostic formatter+decider migration for the wildfire
+hazard, mirroring test_hydro_refactor.py / test_quake_refactor.py:
+
+1. Golden byte-identical: formatters.fire.format() reproduces the legacy
+   wfigs_handler._render() wire exactly for a New incident, an Update-with-
+   growth ((+delta) size line), a movement-line case, an anchor-line case, and
+   the wildfire_closed all-clear.
+
+2. Gate-sequence parity: an explicit `now`-timeline of WFIGS events driven
+   through the NEW gating.fire.decide() matches the OLD handle_wfigs
+   broadcast/suppress behavior AND the data-dict stamps
+   (category / _severity_override / _dedup_suffix / _cooldown_suffix).
+
+3. Registration: the three explicit categories the wfigs_handler emits
+   (wildfire_declared / wildfire_incident / wildfire_closed) resolve to the
+   fire formatter + decider, and FIRMS categories do NOT.
+"""
+from __future__ import annotations
+
+import pytest
+
+from meshai import central_normalizer as cn
+from meshai.central.budget import budget_for
+from meshai.central.wfigs_handler import (
+    _build_canonical,
+    _render as _wfigs_render,
+    handle_wfigs,
+)
+from meshai.notifications.formatters.fire import format as fire_format
+from meshai.notifications.gating.fire import decide as fire_decide
+from meshai.persistence import close_thread_connection, init_db
+from meshai.persistence import db as persistence_db
+from tests.harness.goldens import assert_byte_identical
+from tests.test_wfigs_handler import (
+    _IRWIN_A,
+    _make_active_envelope,
+    _make_tombstone,
+)
+
+_AT = 1_800_000_000.0  # pinned epoch (unused by fire render; kept for parity)
+
+
+@pytest.fixture
+def mem_db(monkeypatch, tmp_path):
+    db_path = str(tmp_path / "fire-refactor-test.sqlite")
+    monkeypatch.setenv("MESHAI_DB_PATH", db_path)
+    persistence_db._initialised.clear()
+    close_thread_connection()
+    conn = init_db()
+    try:
+        from meshai.adapter_config import adapter_config as _ac
+        _ac.invalidate()
+    except Exception:
+        pass
+    try:
+        from meshai.central import wfigs_handler as _wh
+        _wh._last_cleanup = 0
+    except Exception:
+        pass
+    yield conn
+    close_thread_connection()
+    persistence_db._initialised.discard(db_path)
+
+
+class _FakeEvent:
+    def __init__(self, data, category=None):
+        self.data = data
+        self.category = category
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 1. Golden byte-identical — formatter reproduces _render()/all-clear exactly
+# ─────────────────────────────────────────────────────────────────────────────
+
+class TestFormatterGolden:
+    """formatters.fire.format() == wfigs_handler._render() for the same inputs."""
+
+    def _budget(self) -> int:
+        return budget_for("wfigs")
+
+    def _incident_data(self, **over) -> dict:
+        d = {
+            "incident_name": "Cache Peak Fire",
+            "acres": 1847.0,
+            "contained_pct": 23,
+            "fire_cause": "Lightning",
+            "declared_at_epoch": 1_781_204_400,
+            "unique_fire_id": "2026-IDSCF-000987",
+            "geocoder_city": "Burley",  # short-circuits anchor (no DB/Photon)
+            "lat": 42.197,
+            "lon": -113.710,
+            "county": "Cassia",
+            "state": "ID",
+            "landclass": None,
+        }
+        d.update(over)
+        return d
+
+    def test_new_incident(self, mem_db):
+        n = self._incident_data()
+        old = _wfigs_render(n, prefix="New")
+        new = fire_format(_FakeEvent({**n, "is_update": False}),
+                          now=_AT, budget=self._budget())
+        assert_byte_identical(new, old)
+        assert new.startswith("🔥 Cache Peak Fire — New")
+        assert "1,847 ac" in new
+        assert "containment 23%" in new
+
+    def test_update_with_growth_delta(self, mem_db):
+        n = self._incident_data(acres=3000.0, contained_pct=35)
+        old = _wfigs_render(n, prefix="Update",
+                            last_bcast_acres=1847.0, last_bcast_contained=23)
+        new = fire_format(
+            _FakeEvent({**n, "is_update": True,
+                        "last_bcast_acres": 1847.0, "last_bcast_contained": 23}),
+            now=_AT, budget=self._budget())
+        assert_byte_identical(new, old)
+        assert new.startswith("🔥 Cache Peak Fire — Update")
+        assert "3,000 ac (+1,153)" in new  # delta line
+        assert "containment 35%" in new
+
+    def test_movement_line(self, mem_db):
+        # movement is FIRMS-injected; the formatter must read it from event.data
+        # exactly as _render(movement=...) does today.
+        mv = {"direction": "NE", "speed_mph": 1.2}
+        n = self._incident_data()
+        old = _wfigs_render(n, prefix="New", movement=mv)
+        new = fire_format(_FakeEvent({**n, "is_update": False, "movement": mv}),
+                          now=_AT, budget=self._budget())
+        assert_byte_identical(new, old)
+        assert "Moving NE 1.2 mi/h" in new
+
+    def test_anchor_line(self, mem_db):
+        # No geocoder_city → line 3 is resolved via the shared resolve_anchor +
+        # the legacy fallback tiers. Both paths hit the same seeded town_anchors
+        # table, so the wire must stay byte-identical.
+        n = self._incident_data(geocoder_city=None)
+        old = _wfigs_render(n, prefix="New")
+        new = fire_format(_FakeEvent({**n, "is_update": False}),
+                          now=_AT, budget=self._budget())
+        assert_byte_identical(new, old)
+        # line 3 is NOT a movement line and NOT a raw city name
+        assert "Moving" not in new
+
+    def test_wildfire_closed_all_clear(self, mem_db):
+        # Drive the real handler to produce the legacy all-clear wire, then
+        # assert the formatter reproduces it byte-for-byte from event.data.
+        env = _make_active_envelope(geocoder_city="Burley")
+        n0 = cn.normalize(env)
+        data0 = {}
+        handle_wfigs(n0, env, env["subject"], data=data0, now=1_000_000)
+        data0["_on_broadcast_committed"](float(1_000_000))  # arm last_broadcast_*
+
+        tomb = _make_tombstone()
+        data_t = {}
+        old_wire = handle_wfigs(cn.normalize(tomb), tomb, tomb["subject"],
+                                data=data_t, now=2_000_000)
+        assert old_wire is not None
+        assert old_wire.startswith("✅ Cache Peak Fire — contained & closed")
+        assert data_t["category"] == "wildfire_closed"
+
+        # Reconstruct the canonical fields the decider's data_patch supplies to
+        # the formatter for the closed wire, and render.
+        closed_data = {
+            "category": "wildfire_closed",
+            "incident_name": "Cache Peak Fire",
+            "acres": 1847.0,
+            "contained_pct": 23,
+            "lat": env["data"]["data"]["latitude"],
+            "lon": env["data"]["data"]["longitude"],
+            "county": "Cassia",
+            "state": "ID",
+        }
+        new_wire = fire_format(_FakeEvent(closed_data, category="wildfire_closed"),
+                               now=_AT, budget=budget_for("wfigs"))
+        assert_byte_identical(new_wire, old_wire)
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 2. Gate-sequence parity — new decide() vs old handle_wfigs across a lifecycle
+# ─────────────────────────────────────────────────────────────────────────────
+
+class TestGateSequenceParity:
+    """A full fire lifecycle agrees between decide() and handle_wfigs()."""
+
+    def _decide(self, env, now):
+        n = cn.normalize(env)
+        canonical = _build_canonical(n, n["_kind"])
+        return fire_decide(canonical, source="wfigs", now=float(now))
+
+    def _step(self, env, now, *, expect_broadcast, expect_lifecycle):
+        """Assert decide() and handle_wfigs() agree at one timeline step.
+
+        decide() (called first) only READS state, so it sees the same pre-write
+        row the handler's internal decide() sees.  On broadcast the handler's
+        legacy (not-cutover) stamps must equal decide()'s data_patch, and we arm
+        last_broadcast_* via the commit callback (simulating the dispatcher).
+        """
+        n = cn.normalize(env)
+        gate = self._decide(env, now)
+        assert gate.broadcast is expect_broadcast, (
+            f"decide broadcast {gate.broadcast} != {expect_broadcast} "
+            f"@ {now} ({expect_lifecycle})")
+        assert gate.lifecycle == expect_lifecycle, (
+            f"decide lifecycle {gate.lifecycle} != {expect_lifecycle} @ {now}")
+
+        data = {}
+        wire = handle_wfigs(n, env, env["subject"], data=data, now=now)
+        assert (wire is not None) is expect_broadcast
+
+        if expect_broadcast:
+            # Handler's stamped keys (legacy path) match decide()'s data_patch.
+            for k in ("_severity_override", "_dedup_suffix", "_cooldown_suffix"):
+                assert data.get(k) == gate.data_patch.get(k), (
+                    f"stamp {k}: handler={data.get(k)!r} "
+                    f"decide={gate.data_patch.get(k)!r}")
+            if expect_lifecycle == "new":
+                assert data.get("category") == "wildfire_declared"
+                assert gate.data_patch.get("category") == "wildfire_declared"
+            elif expect_lifecycle == "update":
+                # Update keeps the envelope-derived wildfire_incident: no override.
+                assert "category" not in data
+                assert "category" not in gate.data_patch
+            elif expect_lifecycle == "closed":
+                assert data.get("category") == "wildfire_closed"
+                assert gate.data_patch.get("category") == "wildfire_closed"
+            assert data.get("_severity_override") == "priority"
+            # Arm last_broadcast_* for the next cooldown check.
+            data["_on_broadcast_committed"](float(now))
+        return wire
+
+    def test_full_lifecycle(self, mem_db):
+        irwin = _IRWIN_A
+        base = 1_800_000_000
+        disc_ms = (base - 3600) * 1000  # discovered 1h before first sight (fresh)
+
+        def _active(acres, pct, subject_n="a"):
+            return _make_active_envelope(
+                irwin_id=irwin, geocoder_city="Burley",
+                daily_acres=acres, pct_contained=pct,
+                fire_discovery_dt_ms=disc_ms)
+
+        # [0] first sight → New broadcast
+        self._step(_active(250.0, 0), base,
+                   expect_broadcast=True, expect_lifecycle="new")
+        # [1] small growth 1h later (inside 8h cooldown) → suppress
+        self._step(_active(300.0, 0), base + 3600,
+                   expect_broadcast=False, expect_lifecycle="cooldown")
+        # [2] growth after cooldown (8h) → Update
+        self._step(_active(500.0, 0), base + 28800,
+                   expect_broadcast=True, expect_lifecycle="update")
+        # [3] containment change after another cooldown → Update
+        self._step(_active(500.0, 40), base + 28800 * 2,
+                   expect_broadcast=True, expect_lifecycle="update")
+        # [4] tombstone → all-clear (fire was broadcast earlier)
+        tomb = _make_tombstone(irwin_id=irwin)
+        self._step(tomb, base + 100000,
+                   expect_broadcast=True, expect_lifecycle="closed")
+
+    def test_dedup_suffix_tracks_state(self, mem_db):
+        """_dedup_suffix carries the acres|contained that justified the
+        broadcast (so unchanged re-deliveries dedup but genuine updates pass)."""
+        base = 1_800_000_000
+        disc_ms = (base - 3600) * 1000
+        env_new = _make_active_envelope(
+            irwin_id=_IRWIN_A, geocoder_city="Burley",
+            daily_acres=250.0, pct_contained=0, fire_discovery_dt_ms=disc_ms)
+        gate = self._decide(env_new, base)
+        n = cn.normalize(env_new)
+        assert gate.data_patch["_dedup_suffix"] == f"{n['acres']}|{n['contained_pct']}"
+        assert gate.data_patch["_cooldown_suffix"] == _IRWIN_A
+
+    def test_never_broadcast_tombstone_suppressed(self, mem_db):
+        """A tombstone for a fire that never reached the mesh is silent."""
+        tomb = _make_tombstone(irwin_id=_IRWIN_A)
+        gate = self._decide(tomb, 1_000_000)
+        assert gate.broadcast is False
+        assert gate.lifecycle == "suppress"
+        # Handler agrees: returns None.
+        out = handle_wfigs(cn.normalize(tomb), tomb, tomb["subject"],
+                           data={}, now=1_000_000)
+        assert out is None
+
+    def test_perimeter_never_broadcasts(self, mem_db):
+        from tests.test_wfigs_handler import _make_perimeter
+        per = _make_perimeter(irwin_id=_IRWIN_A)
+        gate = self._decide(per, 1_000_000)
+        assert gate.broadcast is False
+        assert gate.lifecycle == "suppress"
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 3. Registration — the three explicit categories resolve; FIRMS does not
+# ─────────────────────────────────────────────────────────────────────────────
+
+class TestRegistration:
+    @pytest.mark.parametrize(
+        "cat", ["wildfire_declared", "wildfire_incident", "wildfire_closed"])
+    def test_formatter_registered(self, cat):
+        from meshai.notifications.formatters import get_formatter
+        assert get_formatter(cat) is fire_format
+
+    @pytest.mark.parametrize(
+        "cat", ["wildfire_declared", "wildfire_incident", "wildfire_closed"])
+    def test_decider_registered(self, cat):
+        from meshai.notifications.gating import get_decider
+        assert get_decider(cat) is fire_decide
+
+    @pytest.mark.parametrize(
+        "cat", ["wildfire_hotspot", "new_ignition",
+                "unattributed_hotspot_cluster", "wildfire_growth"])
+    def test_firms_categories_not_captured(self, cat):
+        # FIRMS is deferred; its categories must NOT resolve to the fire
+        # formatter/decider via the "fire" toggle family fallback.
+        from meshai.notifications.formatters import get_formatter
+        from meshai.notifications.gating import get_decider
+        assert get_formatter(cat) is not fire_format
+        assert get_decider(cat) is not fire_decide

work/tests/test_formatter_scaffold.py

@@ -18,7 +18,8 @@ from meshai.notifications.renderers.composer import compose_mesh_message
     # earthquake_event removed: Phase-1 registers formatters.quake for it.
     # weather_warning/weather_statement removed: Phase-2 registers formatters.nws.
     # road_closure/work_zone/road_incident/traffic_congestion removed: Phase-2 registers formatters.incident.
-    "wildfire_incident",
+    # wildfire_declared/wildfire_incident/wildfire_closed removed: Phase-3b registers formatters.fire.
+    "wildfire_hotspot",  # FIRMS — still deferred (not migrated)
     "battery_critical",
 ])
 def test_get_formatter_returns_none_while_registry_empty(category):

work/tests/test_shadow_inert.py

@@ -142,11 +142,13 @@ class TestShadowInertWhenNoDecider:
 class TestShadowRenderInertWhenNoFormatter:
     """MESHAI_SHADOW_CATEGORIES set but no formatter registered → still no-op.
 
-    Note: earthquake_event has a formatter in Phase 1+; this class uses
-    wildfire_incident which remains un-migrated and has no formatter entry.
+    Note: earthquake_event has a formatter in Phase 1+; Phase-3b migrated the
+    WFIGS categories (wildfire_declared / wildfire_incident / wildfire_closed),
+    so this class uses wildfire_hotspot — a FIRMS category that remains
+    un-migrated (deferred) and has no formatter entry.
     """
 
-    _CATEGORY = "wildfire_incident"
+    _CATEGORY = "wildfire_hotspot"
 
     def setup_method(self):
         os.environ["MESHAI_SHADOW_CATEGORIES"] = self._CATEGORY