feat(firms): source-agnostic fire-fusion — native FIRMS feeds growth/spotting/halt

Closes the last standalone gap. Extract ingest_hotspot_pixel(pixel, *, now)
from firms_handler so the FIRMS attribution/fusion engine (firms_pixels ->
_attribute_or_cluster -> fire_pixels/fire_passes/centroid -> growth/spotting/
halt) is source-agnostic. Both the Central NATS path and native env/firms.py
drive one identical engine.

- shared _ingest_pixel_core(conn, ...) called by both ingest_hotspot_pixel
  and handle_firms; Central path byte-identical (its tests pass unchanged)
- env/firms.py _fetch() feeds each fetched pixel into ingest_hotspot_pixel;
  DB-level dedup makes re-fetched pixels no-ops (no double count); to_event()
  returns None for raw hotspots, precomposed Event for fusion outputs
  (wildfire_growth/spotting/halted via the Phase-3c formatters/gating)
- raw hotspots / new_ignition / cluster NEVER broadcast (cluster stays dead)

FLIP NOTE: keep wildfire_growth/spotting/halted OUT of cutover — native
emits precomposed with gating done inside the engine; cutover would re-run
the _kind-keyed decider on data lacking _kind and suppress.

10 new tests; Central firms/fire-tracker suites unchanged; full suite
10 failed/1682 passed (baseline 10, +10 new).

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

work/meshai/central/firms_handler.py

@@ -200,7 +200,7 @@ def handle_firms(envelope: dict, subject: str,
                     table_name=None, table_pk=None)
         return None
 
-    # ---- persist (INSERT OR IGNORE via v4.sql unique partial index) ------
+    # ---- persist + attribute + fuse (source-agnostic core) --------------
 
     satellite = d.get("satellite") or ""
     brightness_raw = d.get("bright_ti4") if d.get("bright_ti4") is not None \
@@ -210,6 +210,69 @@ def handle_firms(envelope: dict, subject: str,
     except (TypeError, ValueError):
         brightness = None
 
+    # The INSERT-OR-IGNORE into firms_pixels + attribution + growth/spotting/
+    # halt fusion now live in the shared _ingest_pixel_core so the native
+    # env/firms.py adapter feeds the SAME engine. handle_firms keeps the
+    # envelope-specific concerns (field extraction, filtering, event_log
+    # accounting) here so the Central path stays byte-identical. `data` is the
+    # consumer's mutable Event.data dict; core stamps it in place on a fusion
+    # broadcast exactly as the old inline path did.
+    stored, rowid, wire = _ingest_pixel_core(
+        conn, lat=lat, lon=lon, acq_epoch=acq_epoch, frp=frp,
+        confidence=conf, brightness=brightness, satellite=satellite,
+        data=data, now=now,
+    )
+
+    # event_log row regardless of dedup outcome -- both "stored" and
+    # "dedup-hit" count as "handled" for accounting; the suffix tells them
+    # apart for ops grep. (Written after the core call: event_log is a
+    # distinct table with its own rowid sequence, so ordering it after the
+    # attribution INSERTs leaves the row content + relative order identical.)
+    cat_tag = category_raw if stored else category_raw + "|dedup_hit"
+    _log_event(conn, now=now, source="firms", category=cat_tag,
+                severity_word=severity_word,
+                event_id_external=event_id_external,
+                subject=subject, handled=1,
+                table_name="firms_pixels" if stored else None,
+                table_pk=(str(rowid) if stored else None))
+
+    # Dedup hits skip broadcast -- the original insert already had its chance.
+    if not stored:
+        return None
+    return wire
+
+
+# ============================================================================
+# Source-agnostic pixel ingest (Central + native env/firms.py share this)
+# ============================================================================
+
+
+def _ingest_pixel_core(conn, *, lat, lon, acq_epoch, frp, confidence,
+                        brightness, satellite, data, now):
+    """INSERT one canonical hotspot pixel + run attribution/fusion.
+
+    This is the source-agnostic heart of the fire-fusion engine, extracted so
+    the Central NATS handler (``handle_firms``) and the native FIRMS adapter
+    (``env/firms.py``) drive IDENTICAL attribution/fusion from a bare pixel.
+
+    Steps (exactly what the old inline ``handle_firms`` tail did per pixel):
+      1. Compute the v7 meters-quantized ``dedup_key`` and ``INSERT OR IGNORE``
+         into ``firms_pixels`` (unique on ``dedup_key, acq_time, satellite``).
+         A dedup hit is a no-op -- the original insert already ran attribution.
+      2. For a NEWLY-stored pixel, run ``_attribute_or_cluster`` -> writes
+         ``fire_pixels`` / ``fire_passes`` / ``fires`` centroid+cursor and runs
+         the growth / spotting / halt fusion (honoring the Phase-3c deferred
+         latches). The dead unattributed-cluster path stays dead -- NO
+         raw-hotspot / cluster broadcast is ever produced here.
+
+    Determinism: ``now`` is threaded explicitly; there is no hidden clock read.
+
+    Returns ``(stored, rowid, wire)`` where ``wire`` is the fusion broadcast
+    string (or ``None``). ``data`` is mutated in place with the Phase-3c
+    broadcast stamps when a fusion wire is produced.
+    """
+    satellite = satellite or ""
+
     # v0.6-3b: dedup_key from meters-based quantization (v7 schema).
     dedup_distance_m = float(adapter_config.firms.dedup_distance_m)
     if dedup_distance_m > 0:
@@ -224,37 +287,83 @@ def handle_firms(envelope: dict, subject: str,
         "frp, confidence, satellite, brightness, dedup_key) "
         "VALUES (?,?,?,?,?,?,?,?,?)",
         (None, lat, lon, acq_epoch, frp,
-         (str(conf) if conf is not None else None),
+         (str(confidence) if confidence is not None else None),
          satellite, brightness, dedup_key),
     )
     stored = cur.rowcount > 0
-
-    # event_log row regardless of dedup outcome -- both "stored" and
-    # "dedup-hit" count as "handled" for accounting; the suffix tells them
-    # apart for ops grep.
-    handled = 1 if stored else 1   # dedup hit is still handled-success
-    cat_tag = category_raw if stored else category_raw + "|dedup_hit"
-    _log_event(conn, now=now, source="firms", category=cat_tag,
-                severity_word=severity_word,
-                event_id_external=event_id_external,
-                subject=subject, handled=handled,
-                table_name="firms_pixels" if stored else None,
-                table_pk=(str(cur.lastrowid) if stored else None))
-
-    # ---- v0.7-fire-tracker-1: attribution + cluster -------------------
-    # Dedup hits skip attribution -- the original insert already had its
-    # chance. Only newly-stored pixels run through here.
+    rowid = int(cur.lastrowid)
     if not stored:
-        return None
+        return (False, rowid, None)
 
-    return _attribute_or_cluster(
+    wire = _attribute_or_cluster(
         conn,
-        pixel_row_id=int(cur.lastrowid),
+        pixel_row_id=rowid,
         lat=lat, lon=lon,
         acq_epoch=acq_epoch,
         frp=frp, satellite=satellite,
         data=data, now=now,
     )
+    return (True, rowid, wire)
+
+
+def ingest_hotspot_pixel(pixel: dict, *, now) -> list[tuple[str, dict]]:
+    """Source-agnostic entrypoint: ingest ONE canonical FIRMS pixel into the
+    fire-fusion engine and return the fusion broadcasts it produced.
+
+    ``pixel`` is the canonical FIRMS schema::
+
+        {lat, lon, frp, confidence, brightness, satellite, acq_epoch}
+
+    (``lat``/``lon``/``acq_epoch`` required; the rest optional.) This INSERTs
+    the pixel into ``firms_pixels`` (dedup-safe), runs attribution, and runs the
+    growth / spotting / halt fusion -- the same pipeline the Central handler
+    runs -- honoring the Phase-3c deferred latches.
+
+    Returns a list of ``(wire, data)`` fusion broadcasts, where ``wire`` is the
+    mesh text and ``data`` carries the Phase-3c category/severity stamps (and,
+    under cutover, the deferred commit hook). A dedup hit or a pixel that
+    triggers no fusion returns ``[]``. This NEVER returns a raw-hotspot,
+    ``wildfire_hotspot``, ``new_ignition``, or cluster broadcast -- the only
+    outputs are ``wildfire_growth`` / ``wildfire_spotting`` / ``wildfire_halted``
+    (the cluster path is dead). Callers must therefore NEVER broadcast the raw
+    pixel itself -- only these returned fusion wires.
+
+    Determinism: ``now`` (epoch) is required and threaded straight through.
+    """
+    try:
+        conn = get_db()
+    except Exception:
+        logger.exception("ingest_hotspot_pixel: persistence unavailable; dropping")
+        return []
+
+    lat = pixel.get("lat")
+    lon = pixel.get("lon")
+    acq_epoch = pixel.get("acq_epoch")
+    if not (isinstance(lat, (int, float)) and isinstance(lon, (int, float))):
+        return []
+    if acq_epoch is None:
+        return []
+
+    try:
+        frp = float(pixel["frp"]) if pixel.get("frp") is not None else None
+    except (TypeError, ValueError):
+        frp = None
+    try:
+        brightness = float(pixel["brightness"]) \
+            if pixel.get("brightness") is not None else None
+    except (TypeError, ValueError):
+        brightness = None
+
+    data: dict = {}
+    _stored, _rowid, wire = _ingest_pixel_core(
+        conn,
+        lat=float(lat), lon=float(lon), acq_epoch=int(acq_epoch),
+        frp=frp, confidence=pixel.get("confidence"), brightness=brightness,
+        satellite=pixel.get("satellite") or "", data=data, now=now,
+    )
+    if wire is None:
+        return []
+    return [(wire, data)]
 
 
 # ============================================================================

work/meshai/env/firms.py

@@ -37,6 +37,7 @@ class FIRMSAdapter:
 
         self._last_tick = 0.0
         self._events = []
+        self._fusion_events = []  # fire-fusion broadcasts (growth/spotting/halt)
         self._consecutive_errors = 0
         self._last_error = None
         self._is_loaded = False
@@ -119,10 +120,19 @@ class FIRMSAdapter:
         # Parse CSV response
         new_events = self._parse_csv(csv_data)
 
-        # Check if data changed
+        # Feed every fetched pixel into the SHARED source-agnostic fusion engine
+        # (central.firms_handler.ingest_hotspot_pixel): attribution + growth /
+        # spotting / halt. DB-level dedup (firms_pixels unique key) makes this
+        # idempotent across ticks, so re-fetched pixels never double-count.
+        # NEVER broadcasts raw hotspots -- only the fusion wires it returns.
+        self._fusion_events = self._run_fusion(new_events)
+
+        # Check if data changed. Fusion output also counts as "changed" so the
+        # store ingests + emits the broadcast on this tick (store dedups its own
+        # emission by event_id, so an already-emitted fusion won't re-fire).
         old_ids = {e["event_id"] for e in self._events}
         new_ids = {e["event_id"] for e in new_events}
-        changed = old_ids != new_ids
+        changed = (old_ids != new_ids) or bool(self._fusion_events)
 
         self._events = new_events
         self._consecutive_errors = 0
@@ -344,48 +354,163 @@ class FIRMSAdapter:
 
         return (None, None)
 
+    def _satellite_code(self) -> str:
+        """Map the FIRMS source product to the satellite code the fusion
+        engine uses for pass bucketing + the firms_pixels dedup key.
+
+        Consistency across pixels of one fetch is what matters (pass_id groups
+        an overpass); the exact code just needs to be stable per product."""
+        s = (self._source or "").upper()
+        if "NOAA20" in s or "N20" in s or "J1" in s:
+            return "N20"
+        if "NOAA21" in s or "N21" in s or "J2" in s:
+            return "N21"
+        if "SNPP" in s or "SUOMI" in s or "NPP" in s:
+            return "N"
+        if "MODIS" in s:
+            return "MODIS"
+        return self._source or "?"
+
+    def _run_fusion(self, raw_events: list) -> list:
+        """Feed each fetched hotspot pixel into the SHARED attribution/fusion
+        engine and collect the fire-fusion broadcasts.
+
+        Every parsed pixel is mapped to the canonical FIRMS schema and passed to
+        ``central.firms_handler.ingest_hotspot_pixel`` -- the exact same engine
+        the Central NATS handler drives. The engine INSERTs the pixel, runs
+        attribution, and runs the growth / spotting / halt fusion; it returns
+        ONLY fusion wires (never a raw-hotspot / cluster broadcast). DB-level
+        dedup makes re-fetched pixels no-ops, so nothing double-counts.
+        """
+        try:
+            from meshai.central.firms_handler import (
+                ingest_hotspot_pixel, _parse_acq_epoch,
+            )
+        except Exception:
+            logger.exception("FIRMS fusion: handler import failed; skipping")
+            return []
+
+        now = time.time()
+        satellite = self._satellite_code()
+        fusion: list = []
+        for evt in raw_events:
+            props = evt.get("properties", {}) or {}
+            acq_epoch = _parse_acq_epoch(props.get("acq_date"),
+                                         props.get("acq_time"))
+            if acq_epoch is None:
+                continue
+            pixel = {
+                "lat": evt.get("lat"),
+                "lon": evt.get("lon"),
+                "frp": props.get("frp"),
+                "confidence": props.get("confidence"),
+                "brightness": props.get("brightness"),
+                "satellite": satellite,
+                "acq_epoch": acq_epoch,
+            }
+            try:
+                broadcasts = ingest_hotspot_pixel(pixel, now=int(now))
+            except Exception:
+                logger.exception("FIRMS fusion: ingest failed for %s",
+                                 evt.get("event_id"))
+                continue
+            for wire, data in broadcasts:
+                fusion.append(self._make_fusion_event(wire, data, evt, now))
+        return fusion
+
+    def _make_fusion_event(self, wire: str, data: dict, source_evt: dict,
+                           now: float) -> dict:
+        """Wrap a fusion broadcast (wire, data) in an adapter event dict.
+
+        Carries source ``firms_fusion`` (distinct from raw ``firms`` so it never
+        pollutes the hotspot LLM summary / get_active(source="firms")) and a
+        ``_fusion`` payload that ``to_event`` turns into a precomposed Event."""
+        category = data.get("category")
+        severity = (data.get("severity")
+                    or data.get("_severity_override") or "routine")
+        base_id = source_evt.get("event_id") \
+            or f"{source_evt.get('lat')}_{source_evt.get('lon')}"
+        # Precomposed marker: the composer passes event.title through verbatim
+        # (same contract satpass native + the Central consumer use), so the wire
+        # the shared engine produced is what the mesh sees.
+        data["_meshai_precomposed"] = True
+        return {
+            "source": "firms_fusion",
+            "event_id": f"{category}:{base_id}",
+            "event_type": "Fire Fusion",
+            "severity": severity,
+            "headline": wire,
+            "lat": source_evt.get("lat"),
+            "lon": source_evt.get("lon"),
+            "expires": now + 21600,
+            "fetched_at": now,
+            "properties": {"new_ignition": False, "category": category},
+            "_fusion": {"wire": wire, "data": data, "category": category,
+                        "severity": severity},
+        }
+
     def to_event(self, evt: dict) -> Optional["Event"]:
-        """Attribution-only: the native FIRMS adapter NEVER broadcasts hotspots.
+        """Emit ONLY fire-fusion broadcasts; NEVER a raw hotspot.
 
         Firm rule (Matt): "we do NOT broadcast hotspots." A raw satellite
-        thermal pixel -- a single-pixel ``wildfire_hotspot`` or
-        ``new_ignition`` detection -- is noisy and not actionable on its own;
-        broadcasting it would flood the mesh with unattributed heat. This
-        mirrors the Central path (``central/firms_handler.py``), which is
-        storage-only and returns None for every raw pixel. The ONLY FIRMS
-        signals that ever reach the mesh are the fire-tracker FUSION outputs
-        (``wildfire_growth`` / ``wildfire_spotting`` / ``wildfire_halted``),
-        produced by the Central handler's attribution engine -- NOT here.
+        thermal pixel is noisy and not actionable on its own, so a bare
+        ``wildfire_hotspot`` / ``new_ignition`` is NEVER emitted -- this returns
+        None for every raw pixel (identical to the Central storage-only path).
 
-        This method therefore ALWAYS returns None. The neutralization is
-        UNCONDITIONAL (not behind any config flag): the no-hotspots rule is
-        absolute. Previously this emitted ``make_event(category="new_ignition"
-        if new_ignition else "wildfire_hotspot", ...)``, and because
-        ``store._emit_event`` only consults a gating decider for cut-over
-        categories -- and NO decider is registered for the raw hotspot
-        categories (see ``notifications/gating/__init__.py``: "native
-        env/fires.py hotspot broadcasts ... are NOT migrated") -- those Events
-        went straight to the bus and out to the mesh. Returning None removes
-        that broadcast on the native path entirely.
+        What DID change: native FIRMS now feeds the SHARED attribution/fusion
+        engine (see ``_run_fusion`` -> ``firms_handler.ingest_hotspot_pixel``).
+        The fire-tracker FUSION outputs (``wildfire_growth`` /
+        ``wildfire_spotting`` / ``wildfire_halted``) it produces ARE emitted
+        here, as precomposed Events, so the native/standalone deployment gets
+        the same fire signals the Central handler produces. Raw hotspots are
+        distinguished by the absence of a ``_fusion`` payload on the event dict
+        (they carry source ``firms``; fusion dicts carry source
+        ``firms_fusion``); the raw dicts remain available via ``get_events()`` /
+        ``get_new_ignitions()`` for LLM context and health reporting.
 
-        The raw hotspot dicts remain available in-memory via ``get_events()``
-        / ``get_new_ignitions()`` for LLM context and health reporting; they
-        just never become a broadcastable Event.
-
-        STANDALONE GAP (not fixed here): unlike the Central handler, the native
-        adapter has NO fusion wiring -- it does not persist ``firms_pixels`` or
-        run attribution / clustering / pass-boundary growth. Native FIRMS
-        therefore feeds NOTHING into the fire tracker today; the cross-ref to
-        known NIFC fires only sets the (now unused for broadcast)
-        ``new_ignition`` flag on the in-memory dict. Wiring native pixels into
-        the attribution engine so growth/spotting/halt can eventually fire is a
-        separate, larger effort.
+        The fusion categories render via the EXISTING Phase-3c
+        formatters/gating; when they are NOT cut over (the default, and FIRMS's
+        documented state) ``store._emit_event`` emits directly and the decision
+        already made inside the engine stands. (If those categories were cut
+        over, ``store._emit_event`` would re-run the decider on event.data --
+        which lacks the internal ``_kind`` -- and suppress; FIRMS shadow/cutover
+        remains a deferred follow-up, unchanged by this wiring.)
         """
-        return None
+        fusion = evt.get("_fusion")
+        if not fusion:
+            return None  # raw hotspot -- never broadcast
+        try:
+            wire = fusion["wire"]
+            data = fusion["data"]
+            category = fusion["category"]
+            severity = fusion["severity"]
+            event_id = evt.get("event_id")
+            return make_event(
+                source="firms",
+                category=category,
+                severity=severity,
+                title=wire,        # precomposed: composer passes it verbatim
+                summary=wire,
+                lat=evt.get("lat"),
+                lon=evt.get("lon"),
+                timestamp=evt.get("fetched_at"),
+                expires=evt.get("expires"),
+                group_key=event_id,
+                inhibit_keys=[event_id] if event_id else [],
+                data=data,         # carries Phase-3c stamps + precomposed marker
+            )
+        except Exception:
+            logger.exception("FIRMS fusion: to_event failed")
+            return None
 
     def get_events(self) -> list:
-        """Get current hotspot events."""
-        return self._events
+        """Get current hotspot events + fire-fusion broadcasts.
+
+        Raw hotspots (source ``firms``) render None from ``to_event`` and are
+        kept only for LLM context / health; fire-fusion broadcasts (source
+        ``firms_fusion``) render real Events. The store keys on
+        ``(source, event_id)`` and emits each once."""
+        return self._events + self._fusion_events
 
     def get_new_ignitions(self) -> list:
         """Get only potential new ignitions (not near known fires)."""

work/tests/test_firms_native_fusion.py

@@ -0,0 +1,417 @@
+"""Native FIRMS fire-fusion tests — the source-agnostic ingest entrypoint.
+
+Covers the wiring that lets the native ``env/firms.py`` adapter feed
+locally-fetched NASA FIRMS pixels into the SAME attribution/fusion pipeline the
+Central handler uses, via ``central.firms_handler.ingest_hotspot_pixel``:
+
+1. ``ingest_hotspot_pixel`` drives growth / spotting / halt from canonical
+   pixels and returns ``(wire, data)`` fusion broadcasts — and NEVER a raw
+   hotspot / cluster / new_ignition broadcast.
+2. ``env/firms.py`` tick() (CSV fetch monkeypatched) feeds those pixels through
+   the shared engine, emits the fusion Events, and still returns None for raw
+   hotspots.
+3. A guard that the Central ``handle_firms`` path is unchanged by the
+   extraction (storage side-effects + growth wire/stamps identical).
+
+Mirrors the driving style of test_firms_refactor.py / test_fire_tracker_phase*.
+"""
+from __future__ import annotations
+
+import math
+import time
+import uuid
+from datetime import datetime, timezone
+
+import pytest
+
+_MI_PER_DEG_LAT = 69.0
+_FUSION_CATS = {"wildfire_growth", "wildfire_spotting", "wildfire_halted"}
+_RAW_CATS = {"wildfire_hotspot", "new_ignition", "unattributed_hotspot_cluster"}
+
+
+# ── isolation (real-DB, mirrors test_firms_refactor) ─────────────────────────
+
+@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(autouse=True)
+def _no_cutover(monkeypatch):
+    """Default deploy state: nothing cut over -> legacy stamps, direct emit."""
+    monkeypatch.delenv("MESHAI_CUTOVER_CATEGORIES", raising=False)
+    from meshai.notifications.cutover import _clear_cache
+    _clear_cache()
+    yield
+    _clear_cache()
+
+
+# ── helpers ──────────────────────────────────────────────────────────────────
+
+def _seed_fire(*, irwin_id, lat, lon, name="Stub Fire", **cols):
+    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 _acq_epoch(acq_date: str, acq_time: str) -> int:
+    return int(datetime.strptime(f"{acq_date} {acq_time.zfill(4)}",
+                                 "%Y-%m-%d %H%M")
+               .replace(tzinfo=timezone.utc).timestamp())
+
+
+def _pixel(*, lat, lon, acq_date="2026-06-06", acq_time="1200",
+           frp=20.0, confidence="high", brightness=320.0, satellite="N20"):
+    """Build a canonical FIRMS pixel dict."""
+    return {
+        "lat": lat, "lon": lon, "frp": frp, "confidence": confidence,
+        "brightness": brightness, "satellite": satellite,
+        "acq_epoch": _acq_epoch(acq_date, acq_time),
+    }
+
+
+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 _feed(pixel, *, now):
+    from meshai.central.firms_handler import ingest_hotspot_pixel
+    return ingest_hotspot_pixel(pixel, now=now)
+
+
+def _assert_no_raw(broadcasts):
+    """Every returned broadcast must be a fusion category; never a raw one."""
+    for wire, data in broadcasts:
+        cat = data.get("category")
+        assert cat in _FUSION_CATS, f"unexpected non-fusion broadcast: {cat!r}"
+        assert cat not in _RAW_CATS
+
+
+# ═════════════════════════════════════════════════════════════════════════════
+# 1. ingest_hotspot_pixel — growth / spotting / halt
+# ═════════════════════════════════════════════════════════════════════════════
+
+class TestIngestGrowth:
+    def test_two_pass_growth_broadcasts(self):
+        center_lat, center_lon = 42.0, -114.0
+        _seed_fire(irwin_id="ID-NG", lat=center_lat, lon=center_lon,
+                   name="Pine Gulch")
+        # Pass A: 5 pixels around the anchor (acq 12:00-12:04 = one pass bucket).
+        for i in range(5):
+            out = _feed(_pixel(lat=center_lat + 0.0001 * i,
+                               lon=center_lon + 0.0001 * (i - 2),
+                               acq_time=f"12{i:02d}", frp=20.0 + i),
+                        now=1780747200 + i)
+            _assert_no_raw(out)
+            assert out == [], "pass-A pixels must not broadcast (no boundary yet)"
+        # Pass B: one pixel 1 mi north (18:00 = different pass bucket) -> growth.
+        pass_b_lat = center_lat + (1.0 / _MI_PER_DEG_LAT)
+        out = _feed(_pixel(lat=pass_b_lat, lon=center_lon, acq_time="1800",
+                           frp=22.0), now=1780768800)
+        _assert_no_raw(out)
+        assert len(out) == 1
+        wire, data = out[0]
+        assert data["category"] == "wildfire_growth"
+        assert data["_severity_override"] == "immediate"
+        assert wire.startswith("🔥 Pine Gulch")
+        assert "Moving N" in wire
+
+
+class TestIngestSpotting:
+    def _seed_hex_pass_a(self, irwin_id, center_lat, center_lon,
+                         start_now=1780747200):
+        _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)
+            out = _feed(_pixel(lat=la, lon=lo, acq_time=f"12{i * 2:02d}"),
+                        now=start_now + i)
+            _assert_no_raw(out)
+
+    def test_spotting_broadcasts(self):
+        center_lat, center_lon = 43.0, -115.0
+        self._seed_hex_pass_a("ID-NS", center_lat, center_lon)
+        # Pass B pixel ~2 mi NE of the perimeter -> spotting.
+        sp_lat, sp_lon = _offset_mi(center_lat, center_lon,
+                                    north_mi=2.0 / math.sqrt(2),
+                                    east_mi=2.0 / math.sqrt(2))
+        out = _feed(_pixel(lat=sp_lat, lon=sp_lon, acq_time="1800"),
+                    now=1780768800)
+        _assert_no_raw(out)
+        assert len(out) == 1
+        wire, data = out[0]
+        assert data["category"] == "wildfire_spotting"
+        assert data["severity"] == "immediate"
+        assert wire.startswith("🔥 Possible spotting ")
+        # Eager latch stamped during ingest (not-cutover legacy path).
+        from meshai.persistence import get_db
+        latch = get_db().execute(
+            "SELECT last_spotting_broadcast_at FROM fires WHERE irwin_id=?",
+            ("ID-NS",)).fetchone()[0]
+        assert latch == 1780768800.0
+
+
+class TestIngestHalt:
+    def test_halt_broadcasts_for_idle_fire(self):
+        now = 1780768800
+        idle_at = now - 14 * 3600
+        from meshai.persistence import get_db
+        get_db().execute(
+            "INSERT INTO fires(irwin_id, incident_name, lat, lon, "
+            "last_event_at, last_pass_id, last_pass_at) VALUES (?,?,?,?,?,?,?)",
+            ("ID-NH", "Cold Fire", 42.5, -114.5, int(idle_at),
+             "N20-329627", float(idle_at)),
+        )
+        # A fresh pixel FAR from the idle fire -> no attribution, cluster dead,
+        # halt detector fires for the idle fire.
+        out = _feed(_pixel(lat=45.0, lon=-118.0, acq_time="1800"), now=now)
+        _assert_no_raw(out)
+        assert len(out) == 1
+        wire, data = out[0]
+        assert data["category"] == "wildfire_halted"
+        assert data["severity"] == "routine"
+        assert wire == "🔥 Cold Fire no growth in 14h"
+        latch = get_db().execute(
+            "SELECT halt_broadcast_at FROM fires WHERE irwin_id=?",
+            ("ID-NH",)).fetchone()[0]
+        assert latch == float(now)
+
+
+class TestIngestNeverRawOrCluster:
+    def test_lone_pixel_no_fire_yields_nothing(self):
+        # Stored, attributed to nothing, cluster path DEAD, no idle fire ->
+        # returns []. A raw hotspot is NEVER emitted on any path.
+        out = _feed(_pixel(lat=44.4, lon=-116.2, acq_time="1300"),
+                    now=1780750000)
+        assert out == []
+
+    def test_dense_unattributed_cluster_stays_silent(self):
+        # Several unattributed pixels close together would have tripped the old
+        # cluster broadcast; that path is dead -> no broadcast, ever.
+        base_lat, base_lon = 44.0, -116.0
+        produced = []
+        for i in range(6):
+            out = _feed(_pixel(lat=base_lat + 0.001 * i, lon=base_lon,
+                               acq_time=f"13{i:02d}"), now=1780750000 + i)
+            _assert_no_raw(out)
+            produced.extend(out)
+        assert produced == []
+
+
+# ═════════════════════════════════════════════════════════════════════════════
+# 2. env/firms.py tick() — shared engine + raw-hotspot suppression
+# ═════════════════════════════════════════════════════════════════════════════
+
+def _adapter(source="VIIRS_NOAA20_NRT"):
+    from unittest.mock import MagicMock
+    from meshai.env.firms import FIRMSAdapter
+    cfg = MagicMock()
+    cfg.map_key = "test-key"
+    cfg.source = source
+    cfg.bbox = [-117, 42, -114, 44]
+    cfg.day_range = 1
+    cfg.tick_seconds = 0
+    cfg.confidence_min = "nominal"
+    cfg.proximity_km = 10.0
+    a = FIRMSAdapter(cfg, region_anchors=[], fires_adapter=None)
+    a._last_tick = 0.0
+    return a
+
+
+def _csv(rows):
+    header = "latitude,longitude,bright_ti4,confidence,frp,acq_date,acq_time"
+    lines = [header]
+    for r in rows:
+        lines.append("{lat},{lon},{b},{c},{frp},{d},{t}".format(
+            lat=r["lat"], lon=r["lon"], b=r.get("b", 320.0),
+            c=r.get("c", "high"), frp=r.get("frp", 20.0),
+            d=r.get("d", "2026-06-06"), t=r.get("t", "1200")))
+    return "\n".join(lines)
+
+
+class _FakeResp:
+    def __init__(self, text):
+        self._text = text
+    def read(self):
+        return self._text.encode("utf-8")
+    def __enter__(self):
+        return self
+    def __exit__(self, *a):
+        return False
+
+
+def _patch_fetch(monkeypatch, csv_text):
+    monkeypatch.setattr("meshai.env.firms.urlopen",
+                        lambda *a, **k: _FakeResp(csv_text))
+
+
+class TestAdapterTickFusion:
+    def _growth_rows(self, center_lat, center_lon):
+        # Anchor acq times to NOW (native path uses now=time.time()): pass A
+        # ~6h ago, pass B ~now. Both are well inside the 12h halt gate, so the
+        # growing fire never looks idle -- only the growth broadcast fires.
+        # (6h apart => distinct pass buckets => a real pass boundary.)
+        now_dt = datetime.now(timezone.utc)
+        pass_a = now_dt - __import__("datetime").timedelta(hours=6)
+        rows = []
+        for i in range(5):
+            t = pass_a + __import__("datetime").timedelta(minutes=i)
+            rows.append({"lat": center_lat + 0.0001 * i,
+                         "lon": center_lon + 0.0001 * (i - 2),
+                         "d": t.strftime("%Y-%m-%d"), "t": t.strftime("%H%M"),
+                         "frp": 20.0 + i})
+        rows.append({"lat": center_lat + 1.0 / _MI_PER_DEG_LAT,
+                     "lon": center_lon,
+                     "d": now_dt.strftime("%Y-%m-%d"),
+                     "t": now_dt.strftime("%H%M"), "frp": 22.0})
+        return rows
+
+    def test_tick_emits_growth_and_suppresses_raw(self, monkeypatch):
+        center_lat, center_lon = 42.0, -114.0
+        _seed_fire(irwin_id="ID-TG", lat=center_lat, lon=center_lon,
+                   name="Pine Gulch")
+        _patch_fetch(monkeypatch, _csv(self._growth_rows(center_lat, center_lon)))
+
+        a = _adapter()
+        assert a.tick() is True
+
+        evts = a.get_events()
+        raw = [e for e in evts if e.get("source") == "firms"]
+        fusion = [e for e in evts if e.get("source") == "firms_fusion"]
+        # Raw pixels present (LLM context) but every one renders None.
+        assert raw, "raw hotspots should still be cached for LLM context"
+        for r in raw:
+            assert a.to_event(r) is None
+        # Exactly one fusion broadcast, and it renders a real growth Event.
+        assert len(fusion) == 1
+        ev = a.to_event(fusion[0])
+        assert ev is not None
+        assert ev.category == "wildfire_growth"
+        assert ev.summary.startswith("🔥 Pine Gulch")
+        assert ev.data.get("_meshai_precomposed") is True
+
+    def test_tick_emits_through_store_emit_event(self, monkeypatch):
+        """End-to-end: the fusion Event reaches the pipeline bus via
+        store._emit_event; raw hotspots never do."""
+        from unittest.mock import MagicMock
+        from meshai.env.store import EnvironmentalStore
+
+        center_lat, center_lon = 42.0, -114.0
+        _seed_fire(irwin_id="ID-TS", lat=center_lat, lon=center_lon,
+                   name="Pine Gulch")
+        _patch_fetch(monkeypatch, _csv(self._growth_rows(center_lat, center_lon)))
+
+        a = _adapter()
+        bus = MagicMock()
+        store = EnvironmentalStore.__new__(EnvironmentalStore)
+        store._events = {}
+        store._event_bus = bus
+        assert a.tick() is True
+        store._ingest("firms", a)
+
+        emitted = [c.args[0] for c in bus.emit.call_args_list]
+        cats = [e.category for e in emitted]
+        assert cats == ["wildfire_growth"], cats
+        assert emitted[0].summary.startswith("🔥 Pine Gulch")
+
+    def test_tick_with_no_fire_emits_nothing(self, monkeypatch):
+        # Pixels with no seeded fire: attribution misses, cluster dead ->
+        # zero fusion, and raw hotspots still suppressed.
+        _patch_fetch(monkeypatch, _csv([
+            {"lat": 43.5, "lon": -115.5, "t": "1300"},
+            {"lat": 43.6, "lon": -115.4, "t": "1305"},
+        ]))
+        a = _adapter()
+        a.tick()
+        fusion = [e for e in a.get_events() if e.get("source") == "firms_fusion"]
+        assert fusion == []
+
+
+# ═════════════════════════════════════════════════════════════════════════════
+# 3. Central-path guard — extraction did not change handle_firms
+# ═════════════════════════════════════════════════════════════════════════════
+
+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"
+
+
+class TestCentralPathUnchanged:
+    def test_storage_only_pixel_still_stores_and_returns_none(self):
+        from meshai.central.firms_handler import handle_firms
+        from meshai.persistence import get_db
+        env = _envelope(lat=42.19664, lon=-113.70981)
+        out = handle_firms(env, subject=_SUBJECT, data={}, now=1780660000)
+        assert out is None  # no fire seeded -> storage only, no broadcast
+        n = get_db().execute(
+            "SELECT COUNT(*) AS n FROM firms_pixels").fetchone()["n"]
+        assert n == 1
+        log = get_db().execute(
+            "SELECT table_name, table_pk, handled FROM event_log "
+            "ORDER BY id DESC LIMIT 1").fetchone()
+        assert log["handled"] == 1
+        assert log["table_name"] == "firms_pixels"
+        row_id = get_db().execute(
+            "SELECT id FROM firms_pixels").fetchone()["id"]
+        assert log["table_pk"] == str(row_id)
+
+    def test_central_growth_wire_and_stamps_identical(self):
+        """The extracted core produces the SAME growth wire/stamps on the
+        Central envelope path that the inline handler did."""
+        from meshai.central.firms_handler import handle_firms
+        center_lat, center_lon = 42.0, -114.0
+        _seed_fire(irwin_id="ID-CG", lat=center_lat, lon=center_lon,
+                   name="Pine Gulch")
+        for i in range(5):
+            handle_firms(_envelope(lat=center_lat + 0.0001 * i,
+                                   lon=center_lon + 0.0001 * (i - 2),
+                                   acq_time=f"12{i:02d}", frp=20.0 + i),
+                         subject=_SUBJECT, data={}, now=1780747200 + i)
+        data = {}
+        wire = handle_firms(_envelope(lat=center_lat + 1.0 / _MI_PER_DEG_LAT,
+                                      lon=center_lon, acq_time="1800", frp=22.0),
+                            subject=_SUBJECT, data=data, now=1780768800)
+        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-CG"