feat(fire): route native WFIGS through the Phase-3 growth decider (fix updates)

Completes the Phase-3 fire migration for the native adapter. env/fires.py now
emits canonical data (_kind/irwin_id/declared_at/acres/contained), native
fires bypass the received-delta gate and run the shared gating.fire.decide +
fire formatter (forward-only growth + containment + 8h cooldown + deferred
commit), and a native-only cold-start pre-pass silent-seeds old/known fires so
no backlog spam. Fixes growth/containment silence (MORA) and revives the
fires-table-backed reminders/digest. Reuses the existing decider — no dup.

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

work/meshai/adapter_config/defaults.py

@@ -330,6 +330,24 @@ REGISTRY: dict[tuple[str, str], dict[str, Any]] = {
         "type": "float",
         "description": "Default attribution radius for FIRMS hotspot -> fire matching, miles. Per-fire override in fires.spread_radius_mi.",
     },
+    # Native WFIGS cold-start silent-seed knobs (env/store.py::_ingest_fires).
+    # new_ignition_max_age_seconds: a FIRST-SIGHT fire older than this at boot is
+    # treated as a pre-existing active fire and silent-seeded (no broadcast); a
+    # first-sight fire younger than this is a genuine fresh ignition and the
+    # decider announces it (New). Default 48h.
+    ("fires", "new_ignition_max_age_seconds"): {
+        "default": 172800,
+        "type": "int",
+        "description": "Max discovery age (seconds) for a first-sight native fire to count as a fresh ignition (announced New). Older/undated first-sights within the boot-grace window are silent-seeded instead. Default 48h.",
+    },
+    # cold_start_grace_seconds: how long after process boot the silent-seed
+    # pre-pass stays active. After this window, first-sights fall through to
+    # normal decider behavior. Default 120s (covers the first fire poll).
+    ("fires", "cold_start_grace_seconds"): {
+        "default": 120,
+        "type": "int",
+        "description": "Seconds after boot during which first-sight old/known native fires are silent-seeded (no backlog dump). After this window first-sights follow normal decider behavior. Default 120s.",
+    },
     # v0.7-fire-2 -- growth + halt detection thresholds.
     # growth_drift_threshold_mi: a per-pass centroid drift of at least
     # this many miles fires wildfire_growth. 0.5 mi matches the design

work/meshai/env/fires.py

@@ -58,7 +58,8 @@ class NICFFiresAdapter:
         """
         out_fields = (
             "attr_IncidentName,attr_IncidentSize,attr_PercentContained,"
-            "attr_FireDiscoveryDateTime,attr_POOState,poly_GISAcres"
+            "attr_FireDiscoveryDateTime,attr_POOState,poly_GISAcres,"
+            "attr_IrwinID,attr_UniqueFireIdentifier"
         )
         if self._coverage is not None:
             params = {
@@ -160,15 +161,36 @@ class NICFFiresAdapter:
             else:
                 event_id_state = self._state
 
+            event_id = f"nifc_{name.replace(' ', '_').lower()}_{event_id_state}"
+
+            # Canonical WFIGS identity + discovery date the Phase-3 fire decider
+            # (notifications/gating/fire.py::decide) reads off Event.data. Prefer
+            # the real IRWIN GUID; fall back to the unique fire id, then to the
+            # stable adapter event_id so a fire is always gate-able by the fires
+            # table even when WFIGS omits the id fields.
+            irwin_id = (
+                props.get("attr_IrwinID")
+                or props.get("attr_UniqueFireIdentifier")
+                or event_id
+            )
+            declared_at_epoch = self._parse_discovery_epoch(
+                props.get("attr_FireDiscoveryDateTime"))
+
             event = {
                 "source": "nifc",
-                "event_id": f"nifc_{name.replace(' ', '_').lower()}_{event_id_state}",
+                "event_id": event_id,
                 "event_type": "Wildfire",
                 "severity": severity,
                 "headline": headline,
                 "name": name,
                 "acres": acres,
                 "pct_contained": pct_contained,
+                # Canonical keys the decider consumes (mirrored into Event.data
+                # by to_event) + reused by store cold-start seeding.
+                "irwin_id": irwin_id,
+                "contained_pct": pct_contained,
+                "declared_at_epoch": declared_at_epoch,
+                "county": None,  # WFIGS perimeter layer carries no county field
                 "lat": lat,
                 "lon": lon,
                 "distance_km": distance_km,
@@ -200,6 +222,27 @@ class NICFFiresAdapter:
 
         return changed
 
+    @staticmethod
+    def _parse_discovery_epoch(raw) -> Optional[int]:
+        """WFIGS FireDiscoveryDateTime (epoch MILLISECONDS) -> epoch seconds.
+
+        Guards None/blank/garbage; WFIGS reports ms since epoch, so values with
+        13+ digits are divided by 1000. Returns None when unparseable so the
+        decider's age-gate fails OPEN (announces) exactly as for a dateless fire.
+        """
+        if raw is None or raw == "":
+            return None
+        try:
+            val = int(float(raw))
+        except (TypeError, ValueError):
+            return None
+        if val <= 0:
+            return None
+        # >= ~ year 2001 in ms (1e12) -> treat as milliseconds.
+        if val >= 1_000_000_000_000:
+            val //= 1000
+        return val
+
     def _compute_centroid(self, geom) -> tuple:
         """Compute centroid from GeoJSON geometry."""
         if not geom:
@@ -325,6 +368,23 @@ class NICFFiresAdapter:
             # sole inhibit_key lets the pipeline Inhibitor suppress lower-severity
             # re-emissions while a higher-severity one is active (severity tiering
             # delegated to the Inhibitor).
+            # Canonical WFIGS schema on Event.data — the exact keys the Phase-3
+            # decider (gating/fire.py::decide) and formatter (formatters/fire.py)
+            # read. `_kind="wfigs_incident"` routes decide() into the active-fire
+            # New/Update/suppress state machine (without it, decide suppresses).
+            data = {
+                "_kind": "wfigs_incident",
+                "irwin_id": evt.get("irwin_id"),
+                "incident_name": name,
+                "acres": acres,
+                "contained_pct": evt.get("contained_pct"),
+                "declared_at_epoch": evt.get("declared_at_epoch"),
+                "lat": lat,
+                "lon": lon,
+                "county": evt.get("county"),
+                "state": evt.get("state"),
+            }
+
             return make_event(
                 source="nifc",
                 category="wildfire_incident",
@@ -337,6 +397,7 @@ class NICFFiresAdapter:
                 lon=lon,
                 group_key=event_id,
                 inhibit_keys=[event_id],
+                data=data,
             )
         except Exception:
             logger.exception(f"NIFC to_event failed for evt: {evt.get('event_id')}")

work/meshai/env/store.py

@@ -44,6 +44,11 @@ class EnvironmentalStore:
         self._adapters = {}  # name -> adapter instance
         self._failed_adapters = {}  # name -> last_error string
         self._events = {}  # (source, event_id) -> event dict
+        # Process boot epoch — the native fire cold-start silent-seed window is
+        # measured from here (see _ingest_fires). clock.now() is the pipeline's
+        # monkeypatchable time seam so tests can freeze/override it.
+        from meshai.notifications import clock as _clock
+        self._boot_at = _clock.now()
         self._event_bus = event_bus  # Pipeline EventBus for emission
         self._swpc_status = {}  # Kp/SFI/scales snapshot
         self._ducting_status = {}  # tropo ducting assessment
@@ -276,6 +281,17 @@ class EnvironmentalStore:
                 self._events[key] = evt
                 touched.add(evt.get("source") or name)
                 self._delta_emit(name, adapter, evt)
+        elif name == "nifc":
+            # Native WFIGS fires DELIBERATELY bypass the received-delta `_seen`
+            # gate: a fire's growth updates are repeat sightings of the SAME
+            # event_id that `_delta_emit` would wrongly suppress. The DECIDER
+            # (gating.fire.decide, backed by the fires table + 8h cooldown) is
+            # the gate instead. `_ingest_fires` also silent-seeds old/known
+            # fires at cold start so boot never dumps a backlog.
+            for evt in adapter.get_events():
+                key = (evt["source"], evt["event_id"])
+                self._events[key] = evt
+            self._ingest_fires(adapter)
         elif name == "avalanche":
             # Avalanche: re-emit on danger_level rise (Update:) not just new
             # events. The rise is a legitimate CONTENT change, so it passes
@@ -303,6 +319,119 @@ class EnvironmentalStore:
         # seeded — the leak-proof invariant behind the wzdx staging fix.
         self._seeded.update(touched)
 
+    def _ingest_fires(self, adapter) -> None:
+        """Native WFIGS fire ingest — Phase-3 growth-decider path.
+
+        For each polled fire (already recorded in ``self._events`` by the
+        caller):
+
+          1. COLD-START silent-seed. A FIRST-SIGHT fire (no ``fires`` row) that
+             is old/undated AND seen inside the boot-grace window is a
+             pre-existing active fire, not a fresh ignition. We INSERT a
+             ``fires`` row stamped as ALREADY broadcast (``last_broadcast_*`` =
+             current) and emit NOTHING, so the decider treats later polls as
+             Update and only fires on real growth — no boot backlog dump. The
+             shared ``gating.fire.decide`` is untouched.
+
+          2. Unconditional current-state write (mirrors the Central
+             ``wfigs_handler``): INSERT (``last_broadcast_*`` NULL) on first
+             sight else UPDATE ``current_*``. This GUARANTEES a row exists so
+             the decider's deferred ``commit`` UPSERT
+             (``WHERE irwin_id=?``) persists ``last_broadcast_*`` — without it a
+             genuine New would never latch and would re-broadcast every poll.
+
+          3. Decider path via ``_emit_event``: runs ``gating.fire.decide``,
+             applies ``gate.data_patch`` (``_dedup_suffix`` / ``_severity_override``
+             / render hints) and arms ``gate.commit`` onto the emitted Event.
+        """
+        from meshai.notifications import clock as _clock
+        from meshai.adapter_config import adapter_config
+
+        now = _clock.now()
+        try:
+            from meshai.persistence import get_db
+            conn = get_db()
+        except Exception as e:
+            logger.warning("nifc fire ingest skipped (DB unavailable): %s", e)
+            return
+
+        try:
+            grace = int(adapter_config.fires.cold_start_grace_seconds)
+        except Exception:
+            grace = 120
+        try:
+            max_age = int(adapter_config.fires.new_ignition_max_age_seconds)
+        except Exception:
+            max_age = 172800
+        within_boot_grace = (now - self._boot_at) < grace
+
+        for evt in adapter.get_events():
+            try:
+                irwin_id = evt.get("irwin_id")
+                if not irwin_id:
+                    continue  # no stable identity -> not gate-able via fires
+                acres = evt.get("acres")
+                contained = evt.get("contained_pct")
+                declared = evt.get("declared_at_epoch")
+
+                row = conn.execute(
+                    "SELECT last_broadcast_at FROM fires WHERE irwin_id=?",
+                    (irwin_id,)).fetchone()
+                first_sight = row is None
+
+                old_or_undated = (
+                    declared is None or (now - int(declared)) >= max_age)
+
+                # (1) Cold-start silent-seed — seed as already-broadcast, no emit.
+                if first_sight and within_boot_grace and old_or_undated:
+                    conn.execute(
+                        "INSERT INTO fires(irwin_id, incident_name, "
+                        "current_acres, current_contained_pct, lat, lon, "
+                        "county, state, declared_at, last_event_at, "
+                        "last_broadcast_at, first_broadcast_at, "
+                        "last_broadcast_acres, last_broadcast_contained) "
+                        "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
+                        (irwin_id, evt.get("name"), acres, contained,
+                         evt.get("lat"), evt.get("lon"), evt.get("county"),
+                         evt.get("state"), declared, int(now),
+                         int(now), int(now), acres, contained),
+                    )
+                    logger.info(
+                        "nifc cold-start silent-seed irwin=%s acres=%s "
+                        "(seeded already-broadcast, no mesh)", irwin_id, acres)
+                    continue
+
+                # (2) Unconditional current-state write so commit can latch.
+                if first_sight:
+                    conn.execute(
+                        "INSERT INTO fires(irwin_id, incident_name, "
+                        "current_acres, current_contained_pct, lat, lon, "
+                        "county, state, declared_at, last_event_at, "
+                        "last_broadcast_at, last_broadcast_acres, "
+                        "last_broadcast_contained) "
+                        "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)",
+                        (irwin_id, evt.get("name"), acres, contained,
+                         evt.get("lat"), evt.get("lon"), evt.get("county"),
+                         evt.get("state"), declared, int(now),
+                         None, None, None),
+                    )
+                else:
+                    conn.execute(
+                        "UPDATE fires SET current_acres=?, "
+                        "current_contained_pct=?, lat=COALESCE(?, lat), "
+                        "lon=COALESCE(?, lon), last_event_at=? "
+                        "WHERE irwin_id=?",
+                        (acres, contained, evt.get("lat"), evt.get("lon"),
+                         int(now), irwin_id),
+                    )
+
+                # (3) Run the decider + emit (or suppress) via the shared path.
+                if self._event_bus is not None and hasattr(adapter, "to_event"):
+                    self._emit_event(adapter, evt)
+            except Exception:
+                logger.exception(
+                    "nifc fire ingest failed for %s", evt.get("event_id", "?"))
+
     def _seed_from_persistent(self) -> None:
         """Pre-seed ``self._seen`` from the durable hazard tables at startup.
 
@@ -449,10 +578,16 @@ class EnvironmentalStore:
             # dry-run comparison for the bake period.
             try:
                 from meshai.notifications.gating import get_decider
-                from meshai.notifications.cutover import is_cutover
+                from meshai.notifications.cutover import is_cutover, NATIVE_ALWAYS_DECIDE
                 from meshai.notifications import clock as _clock
                 decider = get_decider(event.category)
-                if decider is not None and is_cutover(event.category):
+                # Native WFIGS fire categories always run the decider — it IS
+                # the fire gate (fires table + cooldown), so it can't hang on the
+                # shadow-bake env var. Central is off, so this cannot affect any
+                # Central render. All other categories stay env-gated (is_cutover).
+                if decider is not None and (
+                        is_cutover(event.category)
+                        or event.category in NATIVE_ALWAYS_DECIDE):
                     if event.data is None:
                         event.data = {}
                     gate = decider(event.data, source=event.source,

work/meshai/notifications/cutover.py

@@ -30,6 +30,20 @@ from __future__ import annotations
 import functools
 import os
 
+# ── Native-adapter always-decide/always-render set ───────────────────────────
+# The native WFIGS fire path (env/fires.py) must run the shared gating decider
+# AND the shared fire formatter DETERMINISTICALLY, independent of the shadow-bake
+# MESHAI_CUTOVER_CATEGORIES env var: the decider IS the fire gate (fires table +
+# 8h cooldown), so it cannot be left to an operator toggle. These categories are
+# emitted ONLY by the native fire adapter in this standalone deployment (Central
+# is off, feed_source=native), so forcing decider+formatter here has no
+# Central-render impact. Consumed by store._emit_event (decider gate) and
+# renderers.composer.compose_mesh_message (formatter gate) so native fires take
+# the exact same single render path a cut-over category would.
+NATIVE_ALWAYS_DECIDE = frozenset(
+    {"wildfire_incident", "wildfire_declared", "wildfire_closed"}
+)
+
 
 @functools.lru_cache(maxsize=1)
 def _load_cutover_categories() -> frozenset:

work/meshai/notifications/renderers/composer.py

@@ -334,10 +334,16 @@ def compose_mesh_message(event: Event) -> str:
     # the formatter is used only by shadow_render (dry-run comparison) and by
     # direct unit tests — compose_mesh_message falls through to the legacy path.
     from meshai.notifications.formatters import get_formatter
-    from meshai.notifications.cutover import is_cutover
+    from meshai.notifications.cutover import is_cutover, NATIVE_ALWAYS_DECIDE
     from meshai.notifications import clock
     fmt = get_formatter(event.category)
-    if fmt is not None and is_cutover(event.category):
+    # Native WFIGS fire categories always render via the shared fire formatter
+    # (same single render path a cut-over category takes), independent of the
+    # shadow-bake env var. Central is off in this deployment, so these categories
+    # are emitted only by the native fire adapter — no Central-render impact.
+    if fmt is not None and (
+            is_cutover(event.category)
+            or event.category in NATIVE_ALWAYS_DECIDE):
         try:
             return fmt(event, now=clock.now(), budget=_resolve_budget(event))
         except Exception:

work/tests/test_fire_native_growth.py

@@ -0,0 +1,273 @@
+"""Native WFIGS fire -> Phase-3 growth-decider path (F1).
+
+These tests cover the native fire migration: env/fires.py now emits canonical
+`data` the shared `gating.fire.decide` reads, env/store.py routes native fires
+through the decider (bypassing the received-delta `_seen` gate) with an
+unconditional `fires` state-write + a cold-start silent-seed pre-pass, and the
+composer renders them via the shared fire formatter. The shared decider and
+formatter are REUSED unchanged.
+
+Scenarios (mirror the gate-sequence intent of test_fire_refactor.py):
+  1. Cold-start silent-seed: old/undated first-sight within boot grace -> NO
+     broadcast, `fires` row seeded already-broadcast.
+  2. Growth: seeded MORA (last_bcast 2410 @ now-9h), poll 3000 -> Update; after
+     the deferred commit runs, last_broadcast_acres == 3000.
+  3. Containment rise past cooldown -> Update.
+  4. Unchanged OR within cooldown -> suppress (no emit).
+  5. Fresh ignition (first-sight, declared within 48h) -> New; does NOT re-spam
+     on the next unchanged poll (the state-write latched last_broadcast_*).
+  6. Native event carries _dedup_suffix + _severity_override +
+     _on_broadcast_committed onto the emitted Event; to_event stamps canonical
+     `data`; the composer renders it via the fire formatter with no env var.
+"""
+from __future__ import annotations
+
+import pytest
+
+from meshai.config import EnvironmentalConfig, NICFFiresConfig
+from meshai.env.fires import NICFFiresAdapter
+from meshai.env.store import EnvironmentalStore
+from meshai.notifications.pipeline.bus import EventBus
+from meshai.notifications.renderers.composer import compose_mesh_message
+from meshai.persistence import close_thread_connection, init_db
+from meshai.persistence import db as persistence_db
+
+_NOW = 1_800_000_000.0  # pinned base epoch
+_IRWIN = "IRWIN-MORA-0001"
+
+
+class _Clock:
+    """Mutable time seam so a test can advance between polls."""
+
+    def __init__(self, t: float):
+        self.t = t
+
+    def now(self) -> float:
+        return self.t
+
+
+@pytest.fixture
+def env(monkeypatch, tmp_path):
+    db_path = str(tmp_path / "fire-native.sqlite")
+    monkeypatch.setenv("MESHAI_DB_PATH", db_path)
+    persistence_db._initialised.clear()
+    close_thread_connection()
+    conn = init_db()
+    from meshai.adapter_config import adapter_config as _ac
+    _ac.invalidate()
+    clk = _Clock(_NOW)
+    monkeypatch.setattr("meshai.notifications.clock.now", clk.now)
+    yield conn, clk
+    close_thread_connection()
+    persistence_db._initialised.discard(db_path)
+
+
+class _FakeFires:
+    """Native NIFC stand-in: controllable batch, REAL to_event mapping."""
+
+    def __init__(self):
+        self._batch: list = []
+        self._real = NICFFiresAdapter(NICFFiresConfig())
+
+    def set_batch(self, evts: list) -> None:
+        self._batch = evts
+
+    def tick(self) -> bool:
+        return True
+
+    def get_events(self) -> list:
+        return list(self._batch)
+
+    def to_event(self, evt: dict):
+        return self._real.to_event(evt)
+
+
+def _raw_fire(*, name="MORA", irwin=_IRWIN, acres=2410, contained=10,
+              declared=None, lat=44.0, lon=-115.0, state="US-ID") -> dict:
+    """Mirror the raw internal dict env/fires.py::_fetch builds per fire."""
+    eid = f"nifc_{name.replace(' ', '_').lower()}_{state}"
+    return {
+        "source": "nifc",
+        "event_id": eid,
+        "event_type": "Wildfire",
+        "name": name,
+        "irwin_id": irwin,
+        "acres": acres,
+        "pct_contained": contained,
+        "contained_pct": contained,
+        "declared_at_epoch": declared,
+        "county": None,
+        "lat": lat,
+        "lon": lon,
+        "distance_km": None,
+        "nearest_anchor": None,
+        "severity": "routine",
+        "state": state,
+        "fetched_at": _NOW,
+        "expires": _NOW + 21600,
+    }
+
+
+def _make_store():
+    bus = EventBus()
+    captured: list = []
+    bus.subscribe(lambda e: captured.append(e))
+    store = EnvironmentalStore(EnvironmentalConfig(), event_bus=bus)
+    adapter = _FakeFires()
+    store._adapters["nifc"] = adapter
+    return store, adapter, captured
+
+
+def _seed_row(conn, *, acres, contained, last_bcast_at):
+    """Manually insert an already-broadcast fires row (skips cold-start)."""
+    conn.execute(
+        "INSERT INTO fires(irwin_id, incident_name, current_acres, "
+        "current_contained_pct, lat, lon, state, last_event_at, "
+        "last_broadcast_at, last_broadcast_acres, last_broadcast_contained) "
+        "VALUES (?,?,?,?,?,?,?,?,?,?,?)",
+        (_IRWIN, "MORA", acres, contained, 44.0, -115.0, "US-ID",
+         int(_NOW), int(last_bcast_at), acres, contained),
+    )
+
+
+# ── 1. cold-start silent-seed ────────────────────────────────────────────────
+def test_cold_start_silent_seed_no_broadcast(env):
+    conn, _clk = env
+    store, adapter, captured = _make_store()
+    # First-sight, undated, within boot grace (_boot_at == now).
+    adapter.set_batch([_raw_fire(acres=2410, contained=10, declared=None)])
+    store._ingest("nifc", adapter)
+
+    assert captured == [], "cold-start must broadcast NOTHING"
+    row = conn.execute(
+        "SELECT last_broadcast_acres, last_broadcast_at, "
+        "last_broadcast_contained FROM fires WHERE irwin_id=?",
+        (_IRWIN,)).fetchone()
+    assert row is not None, "cold-start must seed a fires row"
+    assert row["last_broadcast_acres"] == 2410  # seeded == current
+    assert row["last_broadcast_contained"] == 10
+    assert row["last_broadcast_at"] is not None
+
+
+# ── 2. growth after cooldown -> Update + commit latches ─────────────────────
+def test_growth_update_and_commit(env):
+    conn, clk = env
+    store, adapter, captured = _make_store()
+    _seed_row(conn, acres=2410, contained=10, last_bcast_at=_NOW - 9 * 3600)
+
+    adapter.set_batch([_raw_fire(acres=3000, contained=10)])
+    store._ingest("nifc", adapter)
+
+    assert len(captured) == 1, "growth past cooldown must broadcast Update"
+    ev = captured[0]
+    assert ev.data.get("is_update") is True
+    assert ev.data.get("last_bcast_acres") == 2410
+
+    commit = ev.data.get("_on_broadcast_committed")
+    assert callable(commit), "Update must arm the deferred commit"
+    commit(clk.now())
+    row = conn.execute(
+        "SELECT last_broadcast_acres FROM fires WHERE irwin_id=?",
+        (_IRWIN,)).fetchone()
+    assert row["last_broadcast_acres"] == 3000, "commit must latch new acres"
+
+
+# ── 3. containment rise past cooldown -> Update ─────────────────────────────
+def test_containment_rise_update(env):
+    conn, _clk = env
+    store, adapter, captured = _make_store()
+    _seed_row(conn, acres=2410, contained=10, last_bcast_at=_NOW - 9 * 3600)
+
+    adapter.set_batch([_raw_fire(acres=2410, contained=55)])
+    store._ingest("nifc", adapter)
+
+    assert len(captured) == 1, "containment rise past cooldown must broadcast"
+    assert captured[0].data.get("is_update") is True
+
+
+# ── 4. unchanged OR within cooldown -> suppress ─────────────────────────────
+def test_within_cooldown_suppresses(env):
+    conn, _clk = env
+    store, adapter, captured = _make_store()
+    # Growth present but last broadcast only 1h ago -> inside 8h cooldown.
+    _seed_row(conn, acres=2410, contained=10, last_bcast_at=_NOW - 3600)
+
+    adapter.set_batch([_raw_fire(acres=3000, contained=10)])
+    store._ingest("nifc", adapter)
+
+    assert captured == [], "growth inside cooldown must be suppressed"
+
+
+def test_no_change_suppresses(env):
+    conn, _clk = env
+    store, adapter, captured = _make_store()
+    _seed_row(conn, acres=2410, contained=10, last_bcast_at=_NOW - 9 * 3600)
+
+    adapter.set_batch([_raw_fire(acres=2410, contained=10)])
+    store._ingest("nifc", adapter)
+
+    assert captured == [], "no forward change must be suppressed"
+
+
+# ── 5. fresh ignition -> New, and no re-spam on the next unchanged poll ──────
+def test_fresh_ignition_new_then_no_respam(env):
+    conn, clk = env
+    store, adapter, captured = _make_store()
+    fresh = dict(name="FRESH", irwin="IRWIN-FRESH-9", acres=120, contained=0,
+                 declared=_NOW - 3600)  # 1h old -> genuine fresh ignition
+
+    adapter.set_batch([_raw_fire(**fresh)])
+    store._ingest("nifc", adapter)
+    assert len(captured) == 1, "fresh ignition must broadcast New"
+    ev = captured[0]
+    assert ev.data.get("category") == "wildfire_declared"
+    assert ev.data.get("is_update") is False
+    # Latch the New broadcast.
+    ev.data["_on_broadcast_committed"](clk.now())
+
+    # Next poll, unchanged, even well past cooldown -> must NOT re-broadcast.
+    clk.t = _NOW + 10 * 3600
+    captured.clear()
+    adapter.set_batch([_raw_fire(**fresh)])
+    store._ingest("nifc", adapter)
+    assert captured == [], "latched New must not re-spam on unchanged re-poll"
+
+
+# ── 6. stamps + canonical data + formatter render ───────────────────────────
+def test_event_carries_stamps_and_renders_via_formatter(env):
+    conn, _clk = env
+    store, adapter, captured = _make_store()
+    _seed_row(conn, acres=2410, contained=10, last_bcast_at=_NOW - 9 * 3600)
+
+    adapter.set_batch([_raw_fire(acres=3000, contained=20)])
+    store._ingest("nifc", adapter)
+    assert len(captured) == 1
+    ev = captured[0]
+
+    # decider stamps present
+    assert ev.data.get("_dedup_suffix") == "3000|20"
+    assert ev.data.get("_severity_override") == "priority"
+    assert callable(ev.data.get("_on_broadcast_committed"))
+    # canonical data from to_event
+    assert ev.data.get("_kind") == "wfigs_incident"
+    assert ev.data.get("irwin_id") == _IRWIN
+    assert ev.data.get("acres") == 3000
+    assert ev.data.get("contained_pct") == 20
+
+    # composer renders native fire via the shared fire formatter (no env var):
+    wire = compose_mesh_message(ev)
+    assert "MORA" in wire and "Update" in wire and wire.startswith("\U0001f525")
+
+
+def test_to_event_stamps_canonical_data(env):
+    _conn, _clk = env
+    adapter = _FakeFires()
+    ev = adapter.to_event(_raw_fire(acres=555, contained=30, declared=_NOW))
+    assert ev is not None
+    assert ev.category == "wildfire_incident"
+    assert ev.data["_kind"] == "wfigs_incident"
+    assert ev.data["irwin_id"] == _IRWIN
+    assert ev.data["acres"] == 555
+    assert ev.data["contained_pct"] == 30
+    assert ev.data["declared_at_epoch"] == _NOW
+    assert ev.data["lat"] == 44.0 and ev.data["state"] == "US-ID"