diff --git a/work/meshai/adapter_config/defaults.py b/work/meshai/adapter_config/defaults.py index 31f7ec9..925a170 100644 --- a/work/meshai/adapter_config/defaults.py +++ b/work/meshai/adapter_config/defaults.py @@ -330,6 +330,18 @@ 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 (env/store.py::_ingest_fires). + # The cold-start seed is now gated on a per-process FIRST-POLL flag, not a + # wall-clock window: the FIRST fires poll after boot silent-seeds EVERY fire + # present (regardless of age, no broadcast); a fire only broadcasts "New" if + # it first appears on a LATER poll. cold_start_grace_seconds is retained + # (GUI-visible / reserved) but no longer gates the fire seed; the old + # new_ignition_max_age_seconds knob was removed with the 48h age window. + ("fires", "cold_start_grace_seconds"): { + "default": 120, + "type": "int", + "description": "Reserved boot-grace window (seconds). The native fire cold-start silent-seed is now gated on the first fires poll rather than this window, so it no longer affects fire seeding; retained for GUI/back-compat. 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 diff --git a/work/meshai/env/fires.py b/work/meshai/env/fires.py index bddc42a..e8e4665 100644 --- a/work/meshai/env/fires.py +++ b/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')}") diff --git a/work/meshai/env/store.py b/work/meshai/env/store.py index e1c9e93..d9337c8 100644 --- a/work/meshai/env/store.py +++ b/work/meshai/env/store.py @@ -79,6 +79,17 @@ class EnvironmentalStore: self._seen: dict[str, set] = {} # source -> set of item keys self._seeded: set[str] = set() # sources past their first non-empty poll + # Native WFIGS cold-start silent-seed gate (see _ingest_fires). The + # FIRST fires poll after boot treats every fire present as already-known + # (seed silently, no broadcast) so a fresh deploy never dumps the active- + # fire backlog to the mesh. Gated on this per-process flag rather than a + # wall-clock boot grace so it holds no matter how late the first + # successful WFIGS fetch lands (e.g. a failed first fetch pushes the + # first real poll ~10min out, well past any grace window). Flag flips + # only after a non-empty fires ingest; a fire that first appears on a + # LATER poll is a genuine ignition and broadcasts "New". + self._fires_seeded: bool = False + # Create adapter instances with error isolation self._register_adapter("nws", config.nws, ".nws", "NWSAlertsAdapter", lambda cfg: (cfg, self._coverage_for("nws"))) @@ -276,6 +287,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 EVERY current + # fire on the first poll (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 +325,127 @@ 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. On the FIRST fires poll after boot + (``self._fires_seeded`` is False), EVERY first-sight fire (no + ``fires`` row) is treated as a pre-existing active fire — + regardless of age. 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 (a fresh deploy with an empty ``fires`` table + must NOT broadcast every fire discovered in the last 48h). A fire + only broadcasts "New" if it FIRST appears on a LATER poll (a + genuine ignition since startup). The shared ``gating.fire.decide`` + is untouched. This is gated on the per-process first-poll flag, not + a wall-clock boot grace, so it holds no matter how late the first + successful WFIGS fetch lands. + + 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 + + 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 + + # Cold-start gate: is this the FIRST fires poll since boot? Captured + # once for the whole batch, before the flag is flipped below, so every + # fire in the initial full-state sweep is silent-seeded together. + cold_start = not self._fires_seeded + + events = adapter.get_events() + for evt in 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 + + # (1) Cold-start silent-seed — EVERY first-sight fire on the + # first poll, regardless of age; seed as already-broadcast, no + # emit. + if first_sight and cold_start: + 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", "?")) + + # First fires poll complete: later polls broadcast genuine ignitions. + # `_ingest_fires` is only reached when the adapter's tick() reported a + # change, which for the atomic WFIGS fetch means a non-empty batch on + # the first successful poll (a 0-fire fetch is `changed=False` and never + # ingests) — so this flip only ever happens on a real full-state sweep. + if events: + self._fires_seeded = True + def _seed_from_persistent(self) -> None: """Pre-seed ``self._seen`` from the durable hazard tables at startup. @@ -449,10 +592,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, diff --git a/work/meshai/notifications/cutover.py b/work/meshai/notifications/cutover.py index 2349196..7604610 100644 --- a/work/meshai/notifications/cutover.py +++ b/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: diff --git a/work/meshai/notifications/renderers/composer.py b/work/meshai/notifications/renderers/composer.py index 9ac1121..248b318 100644 --- a/work/meshai/notifications/renderers/composer.py +++ b/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: diff --git a/work/tests/test_fire_native_growth.py b/work/tests/test_fire_native_growth.py new file mode 100644 index 0000000..79ef819 --- /dev/null +++ b/work/tests/test_fire_native_growth.py @@ -0,0 +1,311 @@ +"""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: ANY first-sight fire on the FIRST poll (undated OR + recently declared) -> NO broadcast, `fires` row seeded already-broadcast. + The first full-state sweep after boot must produce ZERO broadcasts. + 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. Later-poll ignition: a fire absent at boot that FIRST appears on a later + poll (source already seeded) -> 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, on the FIRST fires poll (source not yet seeded). + adapter.set_batch([_raw_fire(acres=2410, contained=10, declared=None)]) + store._ingest("nifc", adapter) + + assert captured == [], "cold-start must broadcast NOTHING" + assert store._fires_seeded is True, "first non-empty poll marks fires seeded" + 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 + + +# ── 1b. cold-start seeds a RECENTLY-declared fire silently too (no 48h dump) ── +def test_cold_start_recent_fire_still_silent(env): + conn, _clk = env + store, adapter, captured = _make_store() + # Declared just 1h ago — under the OLD 48h window this first-sight fire + # would have broadcast New. On cold start it must now seed SILENT anyway: + # a fresh deploy must never dump fires discovered in the last 48h. + adapter.set_batch([_raw_fire(name="FRESH", irwin="IRWIN-FRESH-9", + acres=120, contained=0, + declared=_NOW - 3600)]) + store._ingest("nifc", adapter) + + assert captured == [], "cold-start must seed EVERY fire silent, even recent" + row = conn.execute( + "SELECT last_broadcast_acres, last_broadcast_at FROM fires " + "WHERE irwin_id=?", ("IRWIN-FRESH-9",)).fetchone() + assert row is not None and row["last_broadcast_acres"] == 120 + assert row["last_broadcast_at"] is not None + assert store._fires_seeded is True + + +# ── 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. later-poll ignition -> New, and no re-spam on the next unchanged poll ── +def test_later_poll_ignition_new_then_no_respam(env): + conn, clk = env + store, adapter, captured = _make_store() + + # First (cold-start) poll: an existing fire present at boot -> silent-seed. + # This marks the fires source seeded; NOTHING broadcasts. + adapter.set_batch([_raw_fire(acres=2410, contained=10)]) + store._ingest("nifc", adapter) + assert captured == [], "cold-start poll must broadcast nothing" + assert store._fires_seeded is True + + # Later poll (well after any grace window): a NEW fire that was NOT present + # at boot -> genuine ignition -> New broadcast. + clk.t = _NOW + 20 * 60 + captured.clear() + fresh = dict(name="FRESH", irwin="IRWIN-FRESH-9", acres=120, contained=0, + declared=_NOW - 3600) + adapter.set_batch([_raw_fire(acres=2410, contained=10), + _raw_fire(**fresh)]) + store._ingest("nifc", adapter) + assert len(captured) == 1, "new fire on a later poll must broadcast New" + ev = captured[0] + assert ev.data.get("irwin_id") == "IRWIN-FRESH-9" + 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(acres=2410, contained=10), + _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"