refactor(phase3c): migrate FIRMS fusion broadcasts to formatter+decider (#34)

Migrate the three FIRMS fire-tracker BROADCAST paths (wildfire_growth,
wildfire_spotting, wildfire_halted) behind the registry, NO cutover.
Attribution/pass/centroid/perimeter plumbing stays inline; legacy live
path is byte-identical.

- gating/firms.py decide() discriminates on handler-stamped _kind
  (firms_growth/firms_spotting/firms_halt)
- wildfire_growth reuses formatters/fire.py (verified byte-identical:
  growth SELECT uses current_* columns _render doesn't read, so the wire
  is the movement+anchor line with "size/containment unknown" — a latent
  legacy quirk, reproduced exactly, NOT fixed)
- formatters/firms.py renders spotting + halt wires
- tier-b (flagged): the eager latch writes (fires.last_spotting_broadcast_at,
  fires.halt_broadcast_at) move into deferred commit closures — a dropped
  broadcast no longer burns the latch. Validated by gate-sequence, not
  golden bytes. Not-cutover live path keeps eager latches verbatim.
- FIRMS broadcasts never touch event_log (eager handled=1 at pixel
  storage) → no event_log flip to wrap
- _maybe_emit_cluster stays dead (unconditional return None) + test

Deferred follow-ups (unchanged): env/firms.py native hotspot broadcast
neutralization; native canonical emission.

27 new tests; fire-tracker + firms handler suites preserved; full suite
at 34-failure baseline (1597 passed).

Co-authored-by: Matt Johnson <mj@k7zvx.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

work/meshai/central/firms_handler.py

@@ -714,13 +714,25 @@ def _handle_pass_boundary(conn, *, irwin_id, pass_id, lat, lon,
     if spotting_wire is not None:
         return spotting_wire
 
-    if last_pass_id == pass_id or last_pass_id is None or prev is None:
+    # Phase-3c: the growth broadcast DECISION (pass boundary + drift threshold)
-        # No boundary (same pass), or this is the fire's first pass --
+    # is delegated to gating.firms.decide; the wire is still rendered inline via
-        # nothing to compare drift against.
+    # the WFIGS _render for byte-identity. The boundary predicate reproduces the
-        return None
+    # legacy guard (`last_pass_id == pass_id or last_pass_id is None or prev is
+    # None` -> suppress) exactly. Growth has no latch, so there is nothing to
+    # defer -- only the data stamping differs between the cutover and legacy
+    # paths.
+    from meshai.notifications.cutover import is_cutover
+    from meshai.notifications.gating.firms import decide as _firms_decide
 
-    threshold = float(adapter_config.fires.growth_drift_threshold_mi)
+    boundary_growth = ((last_pass_id != pass_id) and (last_pass_id is not None)
-    if drift_mi is None or drift_mi < threshold:
+                       and (prev is not None))
+    gate = _firms_decide(
+        {"_kind": "firms_growth", "irwin_id": irwin_id,
+         "boundary": boundary_growth, "drift_mi": drift_mi,
+         "drift_direction": drift_direction,
+         "drift_mi_per_hour": drift_mi_per_hour},
+        source="firms", now=float(now))
+    if not gate.broadcast:
         return None
 
     # Drift exceeded the threshold -- emit wildfire_growth via WFIGS renderer.
@@ -737,9 +749,25 @@ def _handle_pass_boundary(conn, *, irwin_id, pass_id, lat, lon,
     normalized = dict(fire)
     normalized["irwin_id"] = irwin_id
     if isinstance(data, dict):
-        data["category"] = "wildfire_growth"
+        if is_cutover("wildfire_growth"):
-        data["_severity_override"] = "immediate"
+            # NEW PATH: fire formatter re-renders from these hints. Feed ONLY the
-        data["_cooldown_suffix"] = irwin_id
+            # fields _render actually reads on the growth path (incident_name,
+            # lat/lon for the anchor, movement, is_update) so the wire matches.
+            data.update(gate.data_patch)
+            data["incident_name"] = fire["incident_name"]
+            data["lat"] = fire["lat"]
+            data["lon"] = fire["lon"]
+            data["_broadcast_audit"] = {"table": "fires", "pk": irwin_id}
+            _raw_commit = gate.commit
+            if _raw_commit is not None:
+                def _on_commit(committed_at: float) -> None:
+                    _raw_commit(committed_at)
+                data["_on_broadcast_committed"] = _on_commit
+        else:
+            # LEGACY verbatim (byte-for-byte identical live output).
+            data["category"] = "wildfire_growth"
+            data["_severity_override"] = "immediate"
+            data["_cooldown_suffix"] = irwin_id
     return _render(normalized, prefix="Update", movement=movement)
 
 
@@ -769,35 +797,51 @@ def _maybe_emit_halt(conn, *, data, now):
     """Find one fire matching the halt criteria, latch + broadcast.
 
     Returns the wire string when a halt event fires; otherwise None.
-    Latching is via fires.halt_broadcast_at -- a fire that came back
+
-    to life (last_pass_at updated to a fresher value) becomes re-
+    Phase-3c: the DECISION (the halt-eligibility SELECT) is delegated to
-    eligible because we filter on `halt_broadcast_at IS NULL OR
+    ``gating.firms.decide``; the wire is still rendered inline here for
-    halt_broadcast_at < last_pass_at`.
+    byte-identity.  On the not-cutover live path the eager
+    ``fires.halt_broadcast_at`` latch is kept VERBATIM (stamped with the handler
+    ``now``) so the live broadcast stays byte-for-byte identical; on the cutover
+    path the latch is deferred into ``gate.commit`` (tier-b change).  Halt
+    re-eligibility is unchanged: a fire whose ``last_pass_at`` advances past
+    ``halt_broadcast_at`` becomes eligible again.
     """
-    minimum_s = int(adapter_config.fires.halt_minimum_seconds)
+    from meshai.notifications.cutover import is_cutover
-    cutoff = float(now) - float(minimum_s)
+    from meshai.notifications.gating.firms import decide as _firms_decide
-    row = conn.execute(
+
-        "SELECT irwin_id, incident_name, last_pass_at FROM fires "
+    gate = _firms_decide({"_kind": "firms_halt"}, source="firms", now=float(now))
-        "WHERE tombstoned_at IS NULL "
+    if not gate.broadcast:
-        "AND last_pass_at IS NOT NULL AND last_pass_at <= ? "
-        "AND (halt_broadcast_at IS NULL "
-        "     OR halt_broadcast_at < last_pass_at) "
-        "ORDER BY last_pass_at ASC LIMIT 1",
-        (cutoff,),
-    ).fetchone()
-    if row is None:
         return None
 
-    conn.execute(
+    p = gate.data_patch
-        "UPDATE fires SET halt_broadcast_at=? WHERE irwin_id=?",
+    irwin_id = p["irwin_id"]
-        (float(now), row["irwin_id"]),
+    name = p["incident_name"]
-    )
+    hours = p["hours"]
+    wire = f"🔥 {name} no growth in {hours}h"
+
     if isinstance(data, dict):
-        data["category"] = "wildfire_halted"
+        if is_cutover("wildfire_halted"):
-        data["severity"] = "routine"
+            # NEW PATH: formatter re-renders from data_patch; latch deferred.
-    hours = max(0, int((float(now) - float(row["last_pass_at"])) / 3600.0))
+            data.update(gate.data_patch)
-    name = row["incident_name"] or "(unnamed fire)"
+            data["_broadcast_audit"] = {"table": "fires", "pk": irwin_id}
-    return f"🔥 {name} no growth in {hours}h"
+            _raw_commit = gate.commit
+
+            def _on_commit(committed_at: float) -> None:
+                if _raw_commit is not None:
+                    _raw_commit(committed_at)
+
+            data["_on_broadcast_committed"] = _on_commit
+        else:
+            # LEGACY verbatim (byte-for-byte identical live output): eager latch
+            # with the handler `now`, plain category/severity stamps.
+            conn.execute(
+                "UPDATE fires SET halt_broadcast_at=? WHERE irwin_id=?",
+                (float(now), irwin_id),
+            )
+            data["category"] = "wildfire_halted"
+            data["severity"] = "routine"
+    return wire
 
 
 def _bearing(lat1: float, lon1: float,
@@ -878,7 +922,8 @@ def _check_spotting(conn, *, irwin_id, pixel_lat, pixel_lon,
                       current_pass_id, incident_name, data, now):
     """Return spotting wire if criteria met, else None."""
     threshold_mi = float(adapter_config.fires.spotting_distance_threshold_mi)
-    cooldown_s = int(adapter_config.fires.spotting_cooldown_seconds)
+    # Phase-3c: the cooldown gate (and its latch) moved into gating.firms.decide;
+    # the cooldown seconds are read there. The geometry below stays inline.
 
     # Most recent CLOSED pass with a perimeter (i.e. not the current pass).
     prev = conn.execute(
@@ -926,16 +971,6 @@ def _check_spotting(conn, *, irwin_id, pixel_lat, pixel_lon,
     if dist_mi < threshold_mi:
         return None
 
-    # Cooldown gate.
-    fires_row = conn.execute(
-        "SELECT last_spotting_broadcast_at FROM fires WHERE irwin_id=?",
-        (irwin_id,),
-    ).fetchone()
-    if fires_row is not None:
-        last_ts = fires_row["last_spotting_broadcast_at"]
-        if last_ts is not None and (float(now) - float(last_ts)) < cooldown_s:
-            return None
-
     # Direction is FROM the perimeter centroid (the previous pass's
     # pass_centroid_lat/lon, already on the row) TO this pixel.
     direction = _direction_8(_bearing(
@@ -943,20 +978,50 @@ def _check_spotting(conn, *, irwin_id, pixel_lat, pixel_lon,
         pixel_lat, pixel_lon,
     ))
 
-    # Stamp the latch + tag data.
+    # Phase-3c: the cooldown DECISION (and its latch) is delegated to
-    conn.execute(
+    # gating.firms.decide. The wire is still rendered inline for byte-identity.
-        "UPDATE fires SET last_spotting_broadcast_at=? WHERE irwin_id=?",
+    # On the not-cutover live path the eager fires.last_spotting_broadcast_at
-        (float(now), irwin_id),
+    # latch is kept VERBATIM (stamped with the handler `now`); on the cutover
-    )
+    # path the latch is deferred into gate.commit (tier-b change).
-    if isinstance(data, dict):
+    from meshai.notifications.cutover import is_cutover
-        data["category"] = "wildfire_spotting"
+    from meshai.notifications.gating.firms import decide as _firms_decide
-        data["severity"] = "immediate"
 
-    return (
+    gate = _firms_decide(
+        {"_kind": "firms_spotting", "irwin_id": irwin_id, "dist_mi": dist_mi,
+         "direction": direction, "incident_name": incident_name},
+        source="firms", now=float(now))
+    if not gate.broadcast:
+        return None
+
+    wire = (
         f"🔥 Possible spotting {dist_mi:.1f} mi {direction} of "
         f"{incident_name} perimeter"
     )
 
+    if isinstance(data, dict):
+        if is_cutover("wildfire_spotting"):
+            # NEW PATH: formatter re-renders from data_patch; latch deferred.
+            data.update(gate.data_patch)
+            data["_broadcast_audit"] = {"table": "fires", "pk": irwin_id}
+            _raw_commit = gate.commit
+
+            def _on_commit(committed_at: float) -> None:
+                if _raw_commit is not None:
+                    _raw_commit(committed_at)
+
+            data["_on_broadcast_committed"] = _on_commit
+        else:
+            # LEGACY verbatim (byte-for-byte identical live output): eager latch
+            # with the handler `now`, plain category/severity stamps.
+            conn.execute(
+                "UPDATE fires SET last_spotting_broadcast_at=? WHERE irwin_id=?",
+                (float(now), irwin_id),
+            )
+            data["category"] = "wildfire_spotting"
+            data["severity"] = "immediate"
+
+    return wire
+
 
 def _convex_hull(points):
     """Andrew's monotone-chain convex hull. Returns the hull as a list of

work/meshai/notifications/formatters/__init__.py

@@ -108,3 +108,20 @@ from meshai.notifications.formatters import fire as _fire_fmt_mod  # noqa: E402,
 register("wildfire_declared", _fire_fmt_mod.format)
 register("wildfire_incident", _fire_fmt_mod.format)
 register("wildfire_closed",   _fire_fmt_mod.format)
+
+# Phase-3c: FIRMS fusion broadcasts. Three categories the firms_handler emits,
+# all under toggle "fire". `wildfire_growth` REUSES the fire formatter — its
+# wire is the WFIGS incident render with a movement dict, and feeding the fire
+# formatter only {incident_name, is_update, movement, lat, lon} reproduces the
+# legacy _render() call byte-for-byte (the growth SELECT's current_* / declared_at
+# columns are NOT read by _render, so size/containment/date render as unknown/
+# absent — matched here by omitting those fields). `wildfire_spotting` and
+# `wildfire_halted` have their own terse wires -> formatters/firms.py. Registered
+# under the explicit strings (not the "fire" toggle) so the still-deferred FIRMS
+# native categories (wildfire_hotspot / new_ignition / unattributed_hotspot_cluster)
+# are NOT captured by the family fallback. NOT cut over this phase.
+register("wildfire_growth", _fire_fmt_mod.format)
+
+from meshai.notifications.formatters import firms as _firms_fmt_mod  # noqa: E402,F401
+register("wildfire_spotting", _firms_fmt_mod.format)
+register("wildfire_halted",   _firms_fmt_mod.format)

work/meshai/notifications/formatters/firms.py

@@ -0,0 +1,53 @@
+"""FIRMS fire-fusion formatter — Phase-3c migration.
+
+Reproduces the two FIRMS broadcast wire shapes that are NOT the WFIGS incident
+render byte-identically, reading the render hints the decider stamps into
+event.data:
+
+  wildfire_spotting  ->  🔥 Possible spotting {dist:.1f} mi {dir} of {name} perimeter
+  wildfire_halted    ->  🔥 {name} no growth in {hours}h
+
+The third migrated FIRMS category, ``wildfire_growth``, reuses
+``formatters/fire.py`` (its wire is the WFIGS incident render with a movement
+dict) and is registered there — NOT here.
+
+Neither legacy wire applied ``fit_to_budget`` (they were terse raw f-strings),
+so this formatter deliberately does NOT budget-fit — it returns the raw wire to
+stay byte-identical.  ``now`` / ``budget`` are accepted for signature parity and
+unused (this formatter is clock-free per the Phase-0 purity guard).
+"""
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from meshai.notifications.events import Event
+
+
+def format(event: "Event", *, now: float, budget: int) -> str:
+    """Render the FIRMS spotting/halt wire from canonical event.data.
+
+    Branch selection is on event.category (falling back to data["category"]).
+    """
+    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_halted":
+        name = d.get("incident_name") or "(unnamed fire)"
+        hours = d.get("hours")
+        return f"🔥 {name} no growth in {hours}h"
+
+    # Default: wildfire_spotting.
+    dist_mi = d.get("dist_mi")
+    direction = d.get("direction")
+    incident_name = d.get("incident_name")
+    return (
+        f"🔥 Possible spotting {dist_mi:.1f} mi {direction} of "
+        f"{incident_name} perimeter"
+    )

work/meshai/notifications/gating/__init__.py

@@ -100,3 +100,18 @@ from meshai.notifications.gating import fire as _fire_gate_mod  # noqa: E402,F40
 register("wildfire_declared", _fire_gate_mod.decide)
 register("wildfire_incident", _fire_gate_mod.decide)
 register("wildfire_closed",   _fire_gate_mod.decide)
+
+# Phase-3c: FIRMS fusion broadcasts (wildfire_growth / wildfire_spotting /
+# wildfire_halted). One decider handles all three; it keys off the handler-
+# stamped internal `_kind` (firms_growth / firms_spotting / firms_halt). The
+# growth/spotting/halt DECISIONS move here; the eager latch writes
+# (last_spotting_broadcast_at / halt_broadcast_at) move into the deferred commit
+# closures (an intended tier-b change). All attribution / pass / centroid /
+# perimeter plumbing stays INLINE in firms_handler. NOTE: wildfire_growth's
+# FORMATTER is fire.py but its DECIDER is firms.py (own decision math). The dead
+# unattributed_hotspot_cluster path and native env/fires.py hotspot broadcasts
+# are NOT migrated. NOT cut over this phase.
+from meshai.notifications.gating import firms as _firms_gate_mod  # noqa: E402,F401
+register("wildfire_growth",   _firms_gate_mod.decide)
+register("wildfire_spotting", _firms_gate_mod.decide)
+register("wildfire_halted",   _firms_gate_mod.decide)

work/meshai/notifications/gating/firms.py

@@ -0,0 +1,241 @@
+"""FIRMS fire-fusion gating decider — Phase-3c migration.
+
+Moves the three FIRMS *broadcast decisions* out of
+``central.firms_handler`` into a source-agnostic decider, mirroring
+quake/wfigs.  It covers ONLY the decision + the deferred latch; every bit of
+attribution / pass-aggregation / centroid / perimeter plumbing STAYS INLINE in
+the handler (that data plumbing is analogous to hydro's ``gauge_readings``
+INSERT and does not belong in a decider).
+
+Three broadcast categories are handled, discriminated by an internal
+``_kind`` the handler stamps onto the decider input:
+
+    firms_growth    -> wildfire_growth   (drift >= threshold at a pass boundary)
+    firms_spotting  -> wildfire_spotting (pixel outside prior perimeter, cooldown)
+    firms_halt      -> wildfire_halted   (fire idle >= halt_minimum_seconds)
+
+The unattributed-hotspot cluster path is DEAD (an unconditional ``return None``
+at the top of ``_maybe_emit_cluster``) and is NOT represented here.  Native
+``env/fires.py`` hotspot broadcasts (new_ignition / wildfire_hotspot) are a
+deferred follow-up and are not migrated.
+
+Tier-b deferral (intended behavior change, validated by gate-sequence, NOT
+golden byte-parity):
+    The eager latch writes the legacy handler performed *before* returning the
+    wire — ``fires.last_spotting_broadcast_at`` (spotting) and
+    ``fires.halt_broadcast_at`` (halt) — move into the deferred ``commit``
+    closure.  A broadcast that is composed but then dropped by the dispatcher
+    (cold-start grace / dedup / toggle) therefore no longer burns the latch, so
+    the next eligible pixel can still fire.  ``wildfire_growth`` has no latch
+    (its re-eligibility is governed by the inline pass cursor), so its
+    ``commit`` is None.
+
+The decider input dicts the handler passes:
+
+    firms_growth:   {_kind, irwin_id, boundary, drift_mi,
+                     drift_direction, drift_mi_per_hour}
+    firms_spotting: {_kind, irwin_id, dist_mi, direction, incident_name}
+    firms_halt:     {_kind}
+"""
+from __future__ import annotations
+
+import logging
+
+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 decide(data: dict, *, source: str, now: float) -> GateResult:
+    """Broadcast decision for a FIRMS fusion event.
+
+    Parameters
+    ----------
+    data:
+        Discriminated decider input (see module docstring for the per-kind
+        schema).  ``data["_kind"]`` selects the growth / spotting / halt path.
+    source:
+        Adapter source name, e.g. "firms" (accepted for signature parity).
+    now:
+        Current epoch (float) — determinism seam.
+
+    Returns
+    -------
+    GateResult with broadcast/lifecycle/data_patch/commit populated.  The
+    handler still renders the wire inline and, on the not-cutover live path,
+    keeps its eager-latch / stamping VERBATIM; ``data_patch`` supplies the
+    stamps + render hints and ``commit`` arms the deferred latch on delivery.
+    """
+    if not isinstance(data, dict):
+        return GateResult(broadcast=False, lifecycle="suppress",
+                          reason="bad decider input")
+    kind = data.get("_kind")
+
+    # ── wildfire_growth ─────────────────────────────────────────────────────
+    if kind == "firms_growth":
+        boundary = bool(data.get("boundary"))
+        drift_mi = data.get("drift_mi")
+        threshold = float(adapter_config.fires.growth_drift_threshold_mi)
+        if (not boundary) or drift_mi is None or drift_mi < threshold:
+            return GateResult(broadcast=False, lifecycle="suppress",
+                              reason="no pass boundary or drift below threshold")
+        irwin_id = data.get("irwin_id")
+        drift_direction = data.get("drift_direction")
+        drift_mi_per_hour = data.get("drift_mi_per_hour")
+        patch: dict = {
+            # Legacy stamps (verbatim on the not-cutover path).
+            "category": "wildfire_growth",
+            "_severity_override": "immediate",
+            "_cooldown_suffix": irwin_id,
+            # Render hints for the fire formatter (cutover path). The growth
+            # wire reuses formatters/fire.py's incident render; feeding ONLY
+            # these fields reproduces the legacy _render() key-mismatch output
+            # (size unknown · containment unknown + movement line) byte-for-byte.
+            "is_update": True,
+            "movement": {"direction": drift_direction,
+                         "speed_mph": (drift_mi_per_hour or 0.0)},
+        }
+        # Growth has no latch — re-eligibility is the inline pass cursor.
+        return GateResult(broadcast=True, lifecycle="growth",
+                          reason="centroid drift >= threshold at pass boundary",
+                          data_patch=patch, commit=None)
+
+    # ── wildfire_spotting ───────────────────────────────────────────────────
+    if kind == "firms_spotting":
+        irwin_id = data.get("irwin_id")
+        cooldown_s = int(adapter_config.fires.spotting_cooldown_seconds)
+        try:
+            conn = get_db()
+        except Exception:
+            logger.exception("firms decide (spotting): persistence unavailable")
+            return GateResult(broadcast=False, lifecycle="suppress",
+                              reason="persistence unavailable")
+        # Cooldown gate — mirrors _check_spotting's inline read exactly.
+        row = conn.execute(
+            "SELECT last_spotting_broadcast_at FROM fires WHERE irwin_id=?",
+            (irwin_id,),
+        ).fetchone()
+        if row is not None:
+            last_ts = row["last_spotting_broadcast_at"]
+            if last_ts is not None and (float(now) - float(last_ts)) < cooldown_s:
+                return GateResult(broadcast=False, lifecycle="cooldown",
+                                  reason="inside spotting cooldown")
+        patch = {
+            # Legacy stamps (verbatim on the not-cutover path): plain severity,
+            # NOT _severity_override.
+            "category": "wildfire_spotting",
+            "severity": "immediate",
+            # Render hints for the firms formatter (cutover path).
+            "dist_mi": data.get("dist_mi"),
+            "direction": data.get("direction"),
+            "incident_name": data.get("incident_name"),
+        }
+        return GateResult(broadcast=True, lifecycle="spotting",
+                          reason="pixel outside prior perimeter beyond threshold",
+                          data_patch=patch,
+                          commit=_spotting_commit(irwin_id))
+
+    # ── wildfire_halted ─────────────────────────────────────────────────────
+    if kind == "firms_halt":
+        try:
+            conn = get_db()
+        except Exception:
+            logger.exception("firms decide (halt): persistence unavailable")
+            return GateResult(broadcast=False, lifecycle="suppress",
+                              reason="persistence unavailable")
+        minimum_s = int(adapter_config.fires.halt_minimum_seconds)
+        cutoff = float(now) - float(minimum_s)
+        # Mirror _maybe_emit_halt's SELECT exactly (same ORDER/LIMIT so the
+        # chosen fire is identical).
+        row = conn.execute(
+            "SELECT irwin_id, incident_name, last_pass_at FROM fires "
+            "WHERE tombstoned_at IS NULL "
+            "AND last_pass_at IS NOT NULL AND last_pass_at <= ? "
+            "AND (halt_broadcast_at IS NULL "
+            "     OR halt_broadcast_at < last_pass_at) "
+            "ORDER BY last_pass_at ASC LIMIT 1",
+            (cutoff,),
+        ).fetchone()
+        if row is None:
+            return GateResult(broadcast=False, lifecycle="suppress",
+                              reason="no fire meets halt criteria")
+        irwin_id = row["irwin_id"]
+        name = row["incident_name"] or "(unnamed fire)"
+        hours = max(0, int((float(now) - float(row["last_pass_at"])) / 3600.0))
+        patch = {
+            # Legacy stamps (verbatim on the not-cutover path).
+            "category": "wildfire_halted",
+            "severity": "routine",
+            # Render hints + the identity the handler needs for the wire/latch.
+            "incident_name": name,
+            "hours": hours,
+            "irwin_id": irwin_id,
+        }
+        return GateResult(broadcast=True, lifecycle="halt",
+                          reason=f"fire idle irwin={irwin_id}",
+                          data_patch=patch,
+                          commit=_halt_commit(irwin_id))
+
+    # Unknown kind (e.g. the decider was invoked from the shadow/cutover harness
+    # with a real FIRMS event.data before FIRMS is shadowed) — suppress. FIRMS
+    # shadow/cutover is a deferred follow-up.
+    return GateResult(broadcast=False, lifecycle="suppress",
+                      reason=f"unhandled firms kind {kind!r}")
+
+
+def _spotting_commit(irwin_id):
+    """Deferred latch: stamp fires.last_spotting_broadcast_at on delivery.
+
+    Tier-b change: legacy stamped this eagerly with the handler ``now`` before
+    returning the wire; here it fires only on confirmed delivery, at the
+    delivery timestamp.  Idempotent (a plain UPDATE), never raises.
+    """
+    def _commit(committed_at: float) -> None:
+        try:
+            conn = get_db()
+        except Exception:
+            logger.exception(
+                "firms spotting commit: persistence unavailable; "
+                "last_spotting_broadcast_at not stamped for irwin=%s", irwin_id)
+            return
+        try:
+            conn.execute(
+                "UPDATE fires SET last_spotting_broadcast_at=? WHERE irwin_id=?",
+                (float(committed_at), irwin_id),
+            )
+        except Exception:
+            logger.exception(
+                "firms spotting commit: latch UPDATE failed irwin=%s", irwin_id)
+
+    return _commit
+
+
+def _halt_commit(irwin_id):
+    """Deferred latch: stamp fires.halt_broadcast_at on delivery.
+
+    Tier-b change: legacy stamped this eagerly with the handler ``now``; here it
+    fires only on confirmed delivery, at the delivery timestamp.  The halt
+    re-eligibility filter (``halt_broadcast_at IS NULL OR halt_broadcast_at <
+    last_pass_at``) still holds because committed_at > now > last_pass_at.
+    Idempotent (a plain UPDATE), never raises.
+    """
+    def _commit(committed_at: float) -> None:
+        try:
+            conn = get_db()
+        except Exception:
+            logger.exception(
+                "firms halt commit: persistence unavailable; "
+                "halt_broadcast_at not stamped for irwin=%s", irwin_id)
+            return
+        try:
+            conn.execute(
+                "UPDATE fires SET halt_broadcast_at=? WHERE irwin_id=?",
+                (float(committed_at), irwin_id),
+            )
+        except Exception:
+            logger.exception(
+                "firms halt commit: latch UPDATE failed irwin=%s", irwin_id)
+
+    return _commit

work/tests/test_fire_refactor.py

@@ -310,10 +310,14 @@ class TestRegistration:
 
     @pytest.mark.parametrize(
         "cat", ["wildfire_hotspot", "new_ignition",
-                "unattributed_hotspot_cluster", "wildfire_growth"])
+                "unattributed_hotspot_cluster"])
     def test_firms_categories_not_captured(self, cat):
-        # FIRMS is deferred; its categories must NOT resolve to the fire
+        # These native FIRMS categories remain deferred; they must NOT resolve
-        # formatter/decider via the "fire" toggle family fallback.
+        # to the fire formatter/decider via the "fire" toggle family fallback.
+        # (Phase-3c migrated the FIRMS FUSION categories wildfire_growth /
+        # wildfire_spotting / wildfire_halted — covered in test_firms_refactor.py
+        # — but growth reuses the fire FORMATTER while keeping its OWN firms
+        # DECIDER, so it is intentionally not asserted here.)
         from meshai.notifications.formatters import get_formatter
         from meshai.notifications.gating import get_decider
         assert get_formatter(cat) is not fire_format

work/tests/test_firms_refactor.py

@@ -0,0 +1,443 @@
+"""Phase-3c FIRMS fusion refactor tests.
+
+Verifies the source-agnostic formatter+decider migration for the THREE FIRMS
+fusion broadcast categories — wildfire_growth / wildfire_spotting /
+wildfire_halted — mirroring test_fire_refactor.py (WFIGS) and
+test_hydro_refactor.py:
+
+1. Registration: the three categories resolve to the right formatter/decider;
+   wildfire_growth reuses the FIRE formatter but keeps its OWN firms decider;
+   still-deferred native FIRMS categories do NOT resolve.
+
+2. Gate-sequence + deferred-latch (the tier-b validation): a `now`-timeline
+   driven through the NEW gating.firms.decide() reproduces the OLD handle_firms
+   broadcast/suppress + stamp behavior, AND the latch is now DEFERRED — a
+   decision does NOT burn the latch until commit() fires (simulating delivery),
+   after which the next decision suppresses.
+
+3. Golden byte-identity: the cutover formatter path reproduces the legacy inline
+   wire byte-for-byte for growth (via fire.py), spotting, and halt (via firms.py).
+
+4. Not-cutover parity: with no category cut over, handle_firms keeps the legacy
+   eager-latch + stamps VERBATIM (byte-identical live behavior).
+
+5. The unattributed_hotspot_cluster path stays DEAD (returns None).
+"""
+from __future__ import annotations
+
+import math
+import uuid
+
+import pytest
+
+from meshai.central.budget import budget_for
+from meshai.notifications.formatters.fire import format as fire_format
+from meshai.notifications.formatters.firms import format as firms_format
+from meshai.notifications.gating.firms import decide as firms_decide
+from tests.harness.goldens import assert_byte_identical
+
+_MI_PER_DEG_LAT = 69.0
+
+
+@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)
+    from meshai.persistence import db as pdb
+    pdb.close_thread_connection()
+    pdb._initialised.discard(db_path)
+    from meshai.persistence import init_db
+    init_db(db_path)
+    try:
+        from meshai.adapter_config import adapter_config as _ac
+        _ac.invalidate()
+    except Exception:
+        pass
+    yield db_path
+    pdb.close_thread_connection()
+    pdb._initialised.discard(db_path)
+
+
+@pytest.fixture
+def _no_cutover(monkeypatch):
+    """Default deploy state: nothing cut over → handler runs legacy verbatim."""
+    monkeypatch.delenv("MESHAI_CUTOVER_CATEGORIES", raising=False)
+    from meshai.notifications.cutover import _clear_cache
+    _clear_cache()
+    yield
+    _clear_cache()
+
+
+def _cutover(monkeypatch, *cats):
+    monkeypatch.setenv("MESHAI_CUTOVER_CATEGORIES", ",".join(cats))
+    from meshai.notifications.cutover import _clear_cache
+    _clear_cache()
+
+
+class _FakeEvent:
+    def __init__(self, data, category=None):
+        self.data = data
+        self.category = category
+
+
+# ── shared FIRMS driving helpers (mirror test_fire_tracker_phase2/3) ──────────
+
+def _seed_fire(*, irwin_id, lat, lon, name="Stub Fire", **cols):
+    import time
+    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"):
+    return {
+        "data": {
+            "adapter": "firms",
+            "category": "wildfire_hotspot",
+            "severity": "routine",
+            "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",
+            },
+        }
+    }
+
+
+_SUBJECT = "central.fire.hotspot.N20.high.us.id"
+
+
+def _drive_two_pass_growth(irwin_id, center_lat, center_lon):
+    """Seed a fire + pass A (5 px) + first pass-B pixel 1 mi N. Returns the
+    (wire, data) from the boundary pixel that fires wildfire_growth."""
+    from meshai.central.firms_handler import handle_firms
+    _seed_fire(irwin_id=irwin_id, lat=center_lat, lon=center_lon, name="Pine Gulch")
+    for i in range(5):
+        env = _envelope(lat=center_lat + 0.0001 * i,
+                        lon=center_lon + 0.0001 * (i - 2),
+                        acq_date="2026-06-06", acq_time=f"12{i:02d}",
+                        frp=20.0 + i)
+        handle_firms(env, subject=_SUBJECT, data={}, now=1780747200 + i)
+    pass_b_lat = center_lat + (1.0 / _MI_PER_DEG_LAT)
+    env_b = _envelope(lat=pass_b_lat, lon=center_lon, acq_time="1800", frp=22.0)
+    data = {}
+    wire = handle_firms(env_b, subject=_SUBJECT, data=data, now=1780768800)
+    return wire, data
+
+
+def _seed_pass_a_hex_then_close(irwin_id, center_lat, center_lon,
+                                start_now=1780747200):
+    from meshai.central.firms_handler import handle_firms
+    _seed_fire(irwin_id=irwin_id, lat=center_lat, lon=center_lon, name=irwin_id)
+    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)
+        env = _envelope(lat=la, lon=lo, acq_time=f"12{i * 2:02d}")
+        handle_firms(env, subject=_SUBJECT, data={}, now=start_now + i)
+
+
+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
+
+
+def _drive_spotting(irwin_id, center_lat, center_lon, now=1780768800):
+    """Seed hex pass A + closed perimeter, then a pass-B pixel 2 mi NE that
+    fires wildfire_spotting. Returns (wire, data)."""
+    from meshai.central.firms_handler import handle_firms
+    _seed_pass_a_hex_then_close(irwin_id, center_lat, center_lon)
+    sp_lat, sp_lon = _offset_mi(center_lat, center_lon,
+                                north_mi=2.0 / math.sqrt(2),
+                                east_mi=2.0 / math.sqrt(2))
+    env_b = _envelope(lat=sp_lat, lon=sp_lon, acq_time="1800")
+    data = {}
+    wire = handle_firms(env_b, subject=_SUBJECT, data=data, now=now)
+    return wire, data
+
+
+def _seed_stale_fire(irwin_id, *, now_epoch, idle_hours=14, name="Cold Fire"):
+    from meshai.persistence import get_db
+    idle_at = now_epoch - (idle_hours * 3600)
+    get_db().execute(
+        "INSERT INTO fires(irwin_id, incident_name, lat, lon, last_event_at, "
+        "last_pass_id, last_pass_at) VALUES (?,?,?,?,?,?,?)",
+        (irwin_id, name, 42.5, -114.5, int(idle_at), "N20-329627",
+         float(idle_at)),
+    )
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 1. Registration
+# ─────────────────────────────────────────────────────────────────────────────
+
+class TestRegistration:
+    def test_growth_formatter_is_fire(self):
+        from meshai.notifications.formatters import get_formatter
+        assert get_formatter("wildfire_growth") is fire_format
+
+    @pytest.mark.parametrize("cat", ["wildfire_spotting", "wildfire_halted"])
+    def test_spotting_halt_formatter_is_firms(self, cat):
+        from meshai.notifications.formatters import get_formatter
+        assert get_formatter(cat) is firms_format
+
+    @pytest.mark.parametrize(
+        "cat", ["wildfire_growth", "wildfire_spotting", "wildfire_halted"])
+    def test_decider_is_firms(self, cat):
+        from meshai.notifications.gating import get_decider
+        assert get_decider(cat) is firms_decide
+
+    @pytest.mark.parametrize(
+        "cat", ["wildfire_hotspot", "new_ignition",
+                "unattributed_hotspot_cluster"])
+    def test_deferred_firms_natives_not_registered(self, cat):
+        from meshai.notifications.formatters import get_formatter
+        from meshai.notifications.gating import get_decider
+        assert get_formatter(cat) is None
+        assert get_decider(cat) is None
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 2. Gate-sequence + deferred-latch (tier-b)
+# ─────────────────────────────────────────────────────────────────────────────
+
+class TestGrowthDecideSequence:
+    def _in(self, **over):
+        d = {"_kind": "firms_growth", "irwin_id": "ID-G", "boundary": True,
+             "drift_mi": 1.0, "drift_direction": "N", "drift_mi_per_hour": 0.16}
+        d.update(over)
+        return d
+
+    def test_boundary_drift_broadcasts_with_stamps(self):
+        gr = firms_decide(self._in(), source="firms", now=1000.0)
+        assert gr.broadcast is True
+        assert gr.data_patch["category"] == "wildfire_growth"
+        assert gr.data_patch["_severity_override"] == "immediate"
+        assert gr.data_patch["_cooldown_suffix"] == "ID-G"
+        # Growth has NO latch to defer.
+        assert gr.commit is None
+
+    def test_no_boundary_suppresses(self):
+        gr = firms_decide(self._in(boundary=False), source="firms", now=1000.0)
+        assert gr.broadcast is False
+
+    def test_sub_threshold_drift_suppresses(self):
+        # Default growth_drift_threshold_mi is 0.5; 0.3 mi is below.
+        gr = firms_decide(self._in(drift_mi=0.3), source="firms", now=1000.0)
+        assert gr.broadcast is False
+
+
+class TestSpottingDecideSequence:
+    def _in(self, **over):
+        d = {"_kind": "firms_spotting", "irwin_id": "ID-S", "dist_mi": 2.0,
+             "direction": "NE", "incident_name": "Pine Gulch"}
+        d.update(over)
+        return d
+
+    def test_first_broadcasts_with_plain_severity(self):
+        _seed_fire(irwin_id="ID-S", lat=43.0, lon=-115.0)
+        gr = firms_decide(self._in(), source="firms", now=1_000_000.0)
+        assert gr.broadcast is True
+        assert gr.data_patch["category"] == "wildfire_spotting"
+        assert gr.data_patch["severity"] == "immediate"
+        assert "_severity_override" not in gr.data_patch
+        assert gr.commit is not None
+
+    def test_latch_not_burned_without_commit(self):
+        """Tier-b: a decision that is never delivered must NOT burn the latch."""
+        _seed_fire(irwin_id="ID-S", lat=43.0, lon=-115.0)
+        t0 = 1_000_000.0
+        gr1 = firms_decide(self._in(), source="firms", now=t0)
+        assert gr1.broadcast is True
+        # No commit() → 10s later a fresh candidate STILL broadcasts.
+        gr2 = firms_decide(self._in(), source="firms", now=t0 + 10)
+        assert gr2.broadcast is True, "latch burned without delivery (tier-b broken)"
+
+    def test_commit_then_cooldown_suppresses_then_reopens(self):
+        _seed_fire(irwin_id="ID-S", lat=43.0, lon=-115.0)
+        t0 = 1_000_000.0
+        gr1 = firms_decide(self._in(), source="firms", now=t0)
+        gr1.commit(t0)  # simulate delivery → stamps last_spotting_broadcast_at=t0
+        # 30 min later (< 1h cooldown) → suppressed.
+        gr2 = firms_decide(self._in(), source="firms", now=t0 + 1800)
+        assert gr2.broadcast is False
+        # 2h later (> cooldown) → reopens.
+        gr3 = firms_decide(self._in(), source="firms", now=t0 + 7200)
+        assert gr3.broadcast is True
+
+
+class TestHaltDecideSequence:
+    def test_first_broadcasts_with_hours_and_severity(self):
+        now = 1780768800.0
+        _seed_stale_fire("ID-H", now_epoch=now, idle_hours=14)
+        gr = firms_decide({"_kind": "firms_halt"}, source="firms", now=now)
+        assert gr.broadcast is True
+        assert gr.data_patch["category"] == "wildfire_halted"
+        assert gr.data_patch["severity"] == "routine"
+        assert gr.data_patch["hours"] == 14
+        assert gr.data_patch["incident_name"] == "Cold Fire"
+        assert gr.data_patch["irwin_id"] == "ID-H"
+        assert gr.commit is not None
+
+    def test_latch_not_burned_without_commit(self):
+        now = 1780768800.0
+        _seed_stale_fire("ID-H", now_epoch=now, idle_hours=14)
+        gr1 = firms_decide({"_kind": "firms_halt"}, source="firms", now=now)
+        assert gr1.broadcast is True
+        gr2 = firms_decide({"_kind": "firms_halt"}, source="firms", now=now)
+        assert gr2.broadcast is True, "latch burned without delivery (tier-b broken)"
+
+    def test_commit_suppresses_then_reeligible_after_new_pass(self):
+        from meshai.persistence import get_db
+        now = 1780768800.0
+        _seed_stale_fire("ID-H", now_epoch=now, idle_hours=14)
+        gr1 = firms_decide({"_kind": "firms_halt"}, source="firms", now=now)
+        gr1.commit(now)  # stamps halt_broadcast_at=now
+        # Same fire no longer eligible (halt_broadcast_at >= last_pass_at).
+        gr2 = firms_decide({"_kind": "firms_halt"}, source="firms", now=now)
+        assert gr2.broadcast is False
+        # Fire reactivates: a new pass advances last_pass_at past the latch.
+        new_pass_at = now + 100
+        get_db().execute("UPDATE fires SET last_pass_at=? WHERE irwin_id=?",
+                         (float(new_pass_at), "ID-H"))
+        # Evaluate later, when the fire is idle again → re-eligible.
+        later = new_pass_at + 14 * 3600
+        gr3 = firms_decide({"_kind": "firms_halt"}, source="firms", now=later)
+        assert gr3.broadcast is True
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 3. Golden byte-identity — cutover formatter reproduces the legacy inline wire
+# ─────────────────────────────────────────────────────────────────────────────
+
+class TestFormatterGolden:
+    def test_growth_wire_reuses_fire_formatter(self, monkeypatch):
+        # Drive the real growth broadcast under cutover: handle_firms returns the
+        # inline _render() wire AND populates data with the fire formatter hints.
+        _cutover(monkeypatch, "wildfire_growth")
+        try:
+            wire, data = _drive_two_pass_growth("ID-GG", 42.0, -114.0)
+            assert wire is not None and wire.startswith("🔥 Pine Gulch")
+            rendered = fire_format(_FakeEvent(data, category="wildfire_growth"),
+                                   now=0.0, budget=budget_for("wfigs"))
+            assert_byte_identical(rendered, wire)
+        finally:
+            from meshai.notifications.cutover import _clear_cache
+            _clear_cache()
+
+    def test_spotting_wire_golden(self, monkeypatch):
+        _cutover(monkeypatch, "wildfire_spotting")
+        try:
+            wire, data = _drive_spotting("ID-SS", 43.0, -115.0)
+            assert wire is not None and wire.startswith("🔥 Possible spotting ")
+            rendered = firms_format(_FakeEvent(data, category="wildfire_spotting"),
+                                    now=0.0, budget=budget_for("firms"))
+            assert_byte_identical(rendered, wire)
+        finally:
+            from meshai.notifications.cutover import _clear_cache
+            _clear_cache()
+
+    def test_halt_wire_golden(self, monkeypatch):
+        from meshai.central.firms_handler import _maybe_emit_halt
+        from meshai.persistence import get_db
+        _cutover(monkeypatch, "wildfire_halted")
+        try:
+            now = 1780768800
+            _seed_stale_fire("ID-HH", now_epoch=now, idle_hours=14)
+            data = {}
+            wire = _maybe_emit_halt(get_db(), data=data, now=now)
+            assert wire is not None and "no growth in 14h" in wire
+            rendered = firms_format(_FakeEvent(data, category="wildfire_halted"),
+                                    now=0.0, budget=budget_for("firms"))
+            assert_byte_identical(rendered, wire)
+        finally:
+            from meshai.notifications.cutover import _clear_cache
+            _clear_cache()
+
+    def test_spotting_formatter_exact_format(self):
+        # Pin the exact legacy f-string shape independent of the driver.
+        wire = firms_format(
+            _FakeEvent({"dist_mi": 2.34, "direction": "SW",
+                        "incident_name": "Cache Peak"},
+                       category="wildfire_spotting"),
+            now=0.0, budget=140)
+        assert wire == "🔥 Possible spotting 2.3 mi SW of Cache Peak perimeter"
+
+    def test_halt_formatter_exact_format(self):
+        wire = firms_format(
+            _FakeEvent({"incident_name": "Cache Peak", "hours": 9},
+                       category="wildfire_halted"),
+            now=0.0, budget=140)
+        assert wire == "🔥 Cache Peak no growth in 9h"
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 4. Not-cutover parity — legacy eager-latch + stamps preserved VERBATIM
+# ─────────────────────────────────────────────────────────────────────────────
+
+class TestNotCutoverLegacyVerbatim:
+    def test_growth_stamps_and_no_latch(self, _no_cutover):
+        wire, data = _drive_two_pass_growth("ID-GN", 42.0, -114.0)
+        assert wire is not None and wire.startswith("🔥 Pine Gulch")
+        assert "Moving N" in wire
+        assert data["category"] == "wildfire_growth"
+        assert data["_severity_override"] == "immediate"
+        assert data["_cooldown_suffix"] == "ID-GN"
+        # Legacy growth never attached a deferred commit.
+        assert "_on_broadcast_committed" not in data
+
+    def test_spotting_eager_latch_stamped(self, _no_cutover):
+        from meshai.persistence import get_db
+        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"
+        # Legacy path stamps the latch EAGERLY with the handler `now`.
+        latch = get_db().execute(
+            "SELECT last_spotting_broadcast_at FROM fires WHERE irwin_id=?",
+            ("ID-SN",)).fetchone()[0]
+        assert latch == 1780768800.0
+        assert "_on_broadcast_committed" not in data
+
+    def test_halt_eager_latch_stamped(self, _no_cutover):
+        from meshai.central.firms_handler import _maybe_emit_halt
+        from meshai.persistence import get_db
+        now = 1780768800
+        _seed_stale_fire("ID-HN", now_epoch=now, idle_hours=14)
+        data = {}
+        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"
+        latch = get_db().execute(
+            "SELECT halt_broadcast_at FROM fires WHERE irwin_id=?",
+            ("ID-HN",)).fetchone()[0]
+        assert latch == float(now)
+        assert "_on_broadcast_committed" not in data
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 5. Cluster path stays DEAD
+# ─────────────────────────────────────────────────────────────────────────────
+
+class TestClusterDead:
+    def test_maybe_emit_cluster_returns_none(self):
+        from meshai.central.firms_handler import _maybe_emit_cluster
+        from meshai.persistence import get_db
+        data = {}
+        out = _maybe_emit_cluster(
+            get_db(), lat=43.0, lon=-115.0, acq_epoch=1780747200,
+            frp=20.0, data=data, now=1780747200, this_pixel_id=1)
+        assert out is None
+        # The dead path must not tag the data dict either.
+        assert data == {}