diff --git a/work/tests/test_fire_age_gate.py b/work/tests/test_fire_age_gate.py index b376f8a..02a6fed 100644 --- a/work/tests/test_fire_age_gate.py +++ b/work/tests/test_fire_age_gate.py @@ -99,14 +99,21 @@ def test_custom_knob_respected(): # --------------------------------------------------------------------------- -# Handler-path: New is suppressed, Update still emits. +# Decider-path: New is suppressed, Update still emits. # -# These exercise the real Case (i)/(ii)/(iii) paths in handle_wfigs against -# the isolated tmp DB seeded by conftest. +# These exercise the real Case (i)/(ii)/(iii) paths in gating.fire.decide() +# (the LIVE decider) against the isolated tmp DB seeded by conftest. +# +# chore/ripout-2dii: previously drove these through handle_wfigs (the dead +# Central NATS-envelope entrypoint); that entrypoint has been removed from +# meshai.env.fire_render (zero live production callers). decide() is the +# SAME decision logic the native WFIGS adapter (env/fires.py -> +# env/store.py::_emit_event) uses live -- see tests/test_fire_native_growth.py +# for full end-to-end coverage of the age-gate through that real entrypoint. # --------------------------------------------------------------------------- -def _normalized(*, irwin_id, declared_at_epoch, acres=250.0, contained=0): +def _canonical(*, irwin_id, declared_at_epoch, acres=250.0, contained=0): return { "_kind": "wfigs_incident", "irwin_id": irwin_id, @@ -120,58 +127,39 @@ def _normalized(*, irwin_id, declared_at_epoch, acres=250.0, contained=0): } -def _envelope(): - return { - "data": {"adapter": "wfigs", "category": "wildfire_incident", - "severity": "priority"} - } - - -def test_handler_new_path_suppresses_old_fire(): - """Case (i): a brand-new fire whose declared_at is ~45d old is INSERTed - but the 'New' broadcast is suppressed (wire is None), and the New-path - category tag is NOT applied.""" - from meshai.env.fire_render import handle_wfigs - from meshai.persistence import get_db +def test_decider_new_path_suppresses_old_fire(): + """Case (i): a brand-new fire whose declared_at is ~45d old suppresses + the 'New' broadcast (gate.broadcast is False), and the New-path category + tag is NOT applied.""" + from meshai.notifications.gating.fire import decide as fire_decide now = int(time.time()) declared = now - _45D # OTR 11 style - data = {} - wire = handle_wfigs(_normalized(irwin_id="ID-OTR-11", declared_at_epoch=declared), - _envelope(), - subject="central.fire.incident.id", - data=data, now=now) - assert wire is None, f"old fire should be silenced, got wire={wire!r}" - assert data.get("category") != "wildfire_declared" - # The row is still INSERTed (so genuine future Updates work). - row = get_db().execute( - "SELECT irwin_id, last_broadcast_at FROM fires WHERE irwin_id=?", - ("ID-OTR-11",)).fetchone() - assert row is not None - assert row["last_broadcast_at"] is None + gate = fire_decide(_canonical(irwin_id="ID-OTR-11", declared_at_epoch=declared), + source="wfigs", now=float(now)) + assert gate.broadcast is False, f"old fire should be silenced, got {gate!r}" + assert "category" not in gate.data_patch -def test_handler_new_path_announces_recent_fire(): +def test_decider_new_path_announces_recent_fire(): """Case (i): a recent fire (~5d) DOES broadcast 'New' and tags wildfire_declared.""" - from meshai.env.fire_render import handle_wfigs + from meshai.notifications.gating.fire import decide as fire_decide now = int(time.time()) declared = now - _5D - data = {} - wire = handle_wfigs(_normalized(irwin_id="ID-RECENT-1", declared_at_epoch=declared), - _envelope(), - subject="central.fire.incident.id", - data=data, now=now) - assert wire is not None and "New" in wire - assert data.get("category") == "wildfire_declared" + gate = fire_decide(_canonical(irwin_id="ID-RECENT-1", declared_at_epoch=declared), + source="wfigs", now=float(now)) + assert gate.broadcast is True + assert gate.lifecycle == "new" + assert gate.data_patch.get("category") == "wildfire_declared" -def test_handler_update_path_still_emits_for_old_fire(): +def test_decider_update_path_still_emits_for_old_fire(): """An already-broadcast OLD fire that grows acreage still emits an 'Update' (Case (iii) is NOT gated -- genuine old-but-active fires keep getting containment/acreage updates).""" - from meshai.env.fire_render import handle_wfigs + from meshai.notifications.gating.fire import decide as fire_decide from meshai.persistence import get_db now = int(time.time()) @@ -185,12 +173,10 @@ def test_handler_update_path_still_emits_for_old_fire(): ("ID-OLD-ACTIVE", "Old Town Road", 250.0, 0, 42.93, -114.45, declared, now - 30000, now - 30000, 250.0, 0), ) - data = {} - wire = handle_wfigs( - _normalized(irwin_id="ID-OLD-ACTIVE", declared_at_epoch=declared, - acres=900.0, contained=20), - _envelope(), subject="central.fire.incident.id", - data=data, now=now) - assert wire is not None and "Update" in wire, \ - f"old-but-active fire Update must still emit, got {wire!r}" - assert data.get("category") != "wildfire_declared" + gate = fire_decide( + _canonical(irwin_id="ID-OLD-ACTIVE", declared_at_epoch=declared, + acres=900.0, contained=20), + source="wfigs", now=float(now)) + assert gate.broadcast is True and gate.lifecycle == "update", \ + f"old-but-active fire Update must still emit, got {gate!r}" + assert "category" not in gate.data_patch diff --git a/work/tests/test_fire_refactor.py b/work/tests/test_fire_refactor.py index b7359c5..5d31bf1 100644 --- a/work/tests/test_fire_refactor.py +++ b/work/tests/test_fire_refactor.py @@ -8,14 +8,28 @@ hazard, mirroring test_hydro_refactor.py / test_quake_refactor.py: growth ((+delta) size line), a movement-line case, an anchor-line case, and the wildfire_closed all-clear. -2. Gate-sequence parity: an explicit `now`-timeline of WFIGS events driven - through the NEW gating.fire.decide() matches the OLD handle_wfigs - broadcast/suppress behavior AND the data-dict stamps - (category / _severity_override / _dedup_suffix / _cooldown_suffix). +2. Gate-sequence: an explicit `now`-timeline of WFIGS events driven through + `gating.fire.decide()` (the LIVE decider -- see + tests/test_fire_native_growth.py for its full end-to-end native-adapter + coverage) exercises the New/cooldown/Update/closed lifecycle and the + data-dict stamps (category / _severity_override / _dedup_suffix / + _cooldown_suffix) `decide()` hands the dispatcher. -3. Registration: the three explicit categories the wfigs_handler emits - (wildfire_declared / wildfire_incident / wildfire_closed) resolve to the - fire formatter + decider, and FIRMS categories do NOT. +3. Registration: the three explicit categories the WFIGS decider/formatter + pair emits (wildfire_declared / wildfire_incident / wildfire_closed) + resolve to the fire formatter + decider, and FIRMS categories do NOT. + +chore/ripout-2dii: `handle_wfigs` (the dead Central NATS-envelope entrypoint +this file used purely as a byte-identity driver) has been REMOVED from +`meshai.env.fire_render` -- zero live production callers. `fire_format` IS +live (reached for wildfire_declared/wildfire_incident via the native WFIGS +adapter `env/fires.py` -> `env/store.py::_emit_event`, forced onto +`gating.fire.decide` + this formatter via `cutover.NATIVE_ALWAYS_DECIDE` +independent of any env var -- see `notifications/renderers/composer.py:343-347` +and `tests/test_fire_native_growth.py`), so this file's fire_format coverage +stays green; only the handle_wfigs-as-oracle plumbing was replaced with +direct `_wfigs_render` calls (also live -- see env/fire_fusion.py's FIRMS +growth path) and direct `fire_decide()` driving. """ from __future__ import annotations @@ -25,7 +39,6 @@ from meshai.notifications.formatters._budget import budget_for from meshai.env.fire_render import ( _build_canonical, _render as _wfigs_render, - handle_wfigs, ) from meshai.notifications.formatters.fire import format as fire_format from meshai.notifications.gating.fire import decide as fire_decide @@ -70,6 +83,33 @@ class _FakeEvent: self.category = category +def _write_fire_state(conn, *, irwin_id, name, acres, contained_pct, + lat, lon, county, state, declared_at_epoch=None, + now): + """Unconditional current_* state write, mirroring the LIVE native path's + own upsert (env/store.py::_ingest_fires INSERT/UPDATE current_acres / + current_contained_pct) -- the same shape handle_wfigs used to do inline + before it was deleted (chore/ripout-2dii).""" + row = conn.execute( + "SELECT irwin_id FROM fires WHERE irwin_id=?", (irwin_id,)).fetchone() + if row is None: + 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, name, acres, contained_pct, lat, lon, county, state, + declared_at_epoch, 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_pct, lat, lon, now, irwin_id), + ) + + # ───────────────────────────────────────────────────────────────────────────── # 1. Golden byte-identical — formatter reproduces _render()/all-clear exactly # ───────────────────────────────────────────────────────────────────────────── @@ -145,92 +185,63 @@ class TestFormatterGolden: assert "Moving" not in new def test_wildfire_closed_all_clear(self, mem_db): - # Drive the real handler to produce the legacy all-clear wire, then - # assert the formatter reproduces it byte-for-byte from event.data. - env = _make_active_envelope(geocoder_city="Burley") - n0 = _normalize_wfigs(env) - data0 = {} - handle_wfigs(n0, env, env["subject"], data=data0, now=1_000_000) - data0["_on_broadcast_committed"](float(1_000_000)) # arm last_broadcast_* + # Drive the LIVE tombstone decision (gating.fire.decide) directly -- + # handle_wfigs (the dead Central entrypoint that used to wrap this) is + # gone (chore/ripout-2dii). State is written the same unconditional + # shape the live native path uses (_write_fire_state). + irwin_id = "IRWIN-CLOSED-1" + _write_fire_state( + mem_db, irwin_id=irwin_id, name="Cache Peak Fire", + acres=1847.0, contained_pct=23, lat=42.197, lon=-113.710, + county="Cassia", state="ID", now=1_000_000) + gate0 = fire_decide( + {"_kind": "wfigs_incident", "irwin_id": irwin_id, + "incident_name": "Cache Peak Fire", "acres": 1847.0, + "contained_pct": 23, "declared_at_epoch": None, + "lat": 42.197, "lon": -113.710, "county": "Cassia", "state": "ID"}, + source="wfigs", now=1_000_000.0) + assert gate0.broadcast is True + gate0.commit(1_000_000.0) # arm last_broadcast_* (fire "reached mesh") - tomb = _make_tombstone() - data_t = {} - old_wire = handle_wfigs(_normalize_wfigs(tomb), tomb, tomb["subject"], - data=data_t, now=2_000_000) - assert old_wire is not None - assert old_wire.startswith("✅ Cache Peak Fire — contained & closed") - assert data_t["category"] == "wildfire_closed" + gate_t = fire_decide( + {"_kind": "wfigs_tombstone", "irwin_id": irwin_id}, + source="wfigs", now=2_000_000.0) + assert gate_t.broadcast is True + assert gate_t.data_patch["category"] == "wildfire_closed" + assert gate_t.data_patch["incident_name"] == "Cache Peak Fire" - # Reconstruct the canonical fields the decider's data_patch supplies to - # the formatter for the closed wire, and render. - closed_data = { - "category": "wildfire_closed", - "incident_name": "Cache Peak Fire", - "acres": 1847.0, - "contained_pct": 23, - "lat": env["data"]["data"]["latitude"], - "lon": env["data"]["data"]["longitude"], - "county": "Cassia", - "state": "ID", - } - new_wire = fire_format(_FakeEvent(closed_data, category="wildfire_closed"), - now=_AT, budget=budget_for("wfigs")) - assert_byte_identical(new_wire, old_wire) + new_wire = fire_format( + _FakeEvent(gate_t.data_patch, category="wildfire_closed"), + now=_AT, budget=budget_for("wfigs")) + assert new_wire.startswith("✅ Cache Peak Fire — contained & closed") + # Golden literal (captured from the live fire_format all-clear branch; + # this is the SAME format string handle_wfigs used to build inline + # before its removal -- see notifications/formatters/fire.py::_render_allclear). + assert new_wire == "✅ Cache Peak Fire — contained & closed\n1,847 ac | 23% contained | 24 mi S of Burley" # ───────────────────────────────────────────────────────────────────────────── -# 2. Gate-sequence parity — new decide() vs old handle_wfigs across a lifecycle +# 2. Gate-sequence — decide() drives New/cooldown/Update/closed # ───────────────────────────────────────────────────────────────────────────── class TestGateSequenceParity: - """A full fire lifecycle agrees between decide() and handle_wfigs().""" + """decide() correctly negotiates a full fire lifecycle end to end. + + New/Update/cooldown/suppress behavior through the REAL native entrypoint + (env/fires.py -> env/store.py -> compose_mesh_message) is covered more + faithfully by tests/test_fire_native_growth.py; this class focuses on the + parts that file doesn't reach -- the tombstone/closed lifecycle step (no + live producer currently emits `_kind=wfigs_tombstone`, but the decider + + formatter branch is live/shipped code and deserves regression coverage) + -- plus a defense-in-depth pass over the whole sequence via decide() + directly (no handle_wfigs; that entrypoint is gone). + """ def _decide(self, env, now): n = _normalize_wfigs(env) canonical = _build_canonical(n, n["_kind"]) return fire_decide(canonical, source="wfigs", now=float(now)) - def _step(self, env, now, *, expect_broadcast, expect_lifecycle): - """Assert decide() and handle_wfigs() agree at one timeline step. - - decide() (called first) only READS state, so it sees the same pre-write - row the handler's internal decide() sees. On broadcast the handler's - legacy (not-cutover) stamps must equal decide()'s data_patch, and we arm - last_broadcast_* via the commit callback (simulating the dispatcher). - """ - n = _normalize_wfigs(env) - gate = self._decide(env, now) - assert gate.broadcast is expect_broadcast, ( - f"decide broadcast {gate.broadcast} != {expect_broadcast} " - f"@ {now} ({expect_lifecycle})") - assert gate.lifecycle == expect_lifecycle, ( - f"decide lifecycle {gate.lifecycle} != {expect_lifecycle} @ {now}") - - data = {} - wire = handle_wfigs(n, env, env["subject"], data=data, now=now) - assert (wire is not None) is expect_broadcast - - if expect_broadcast: - # Handler's stamped keys (legacy path) match decide()'s data_patch. - for k in ("_severity_override", "_dedup_suffix", "_cooldown_suffix"): - assert data.get(k) == gate.data_patch.get(k), ( - f"stamp {k}: handler={data.get(k)!r} " - f"decide={gate.data_patch.get(k)!r}") - if expect_lifecycle == "new": - assert data.get("category") == "wildfire_declared" - assert gate.data_patch.get("category") == "wildfire_declared" - elif expect_lifecycle == "update": - # Update keeps the envelope-derived wildfire_incident: no override. - assert "category" not in data - assert "category" not in gate.data_patch - elif expect_lifecycle == "closed": - assert data.get("category") == "wildfire_closed" - assert gate.data_patch.get("category") == "wildfire_closed" - assert data.get("_severity_override") == "priority" - # Arm last_broadcast_* for the next cooldown check. - data["_on_broadcast_committed"](float(now)) - return wire - def test_full_lifecycle(self, mem_db): irwin = _IRWIN_A base = 1_800_000_000 @@ -242,22 +253,47 @@ class TestGateSequenceParity: daily_acres=acres, pct_contained=pct, fire_discovery_dt_ms=disc_ms) + def _step(env, now, *, expect_broadcast, expect_lifecycle): + n = _normalize_wfigs(env) + gate = self._decide(env, now) + assert gate.broadcast is expect_broadcast, ( + f"decide broadcast {gate.broadcast} != {expect_broadcast} " + f"@ {now} ({expect_lifecycle})") + assert gate.lifecycle == expect_lifecycle, ( + f"decide lifecycle {gate.lifecycle} != {expect_lifecycle} @ {now}") + # Unconditional state write, mirroring the live native path. + _write_fire_state( + mem_db, irwin_id=irwin, name=n.get("incident_name"), + acres=n.get("acres"), contained_pct=n.get("contained_pct"), + lat=n.get("lat"), lon=n.get("lon"), county=n.get("county"), + state=n.get("state"), declared_at_epoch=n.get("declared_at_epoch"), + now=now) + if expect_broadcast: + if expect_lifecycle == "new": + assert gate.data_patch.get("category") == "wildfire_declared" + elif expect_lifecycle == "update": + assert "category" not in gate.data_patch + assert gate.data_patch.get("_severity_override") == "priority" + gate.commit(float(now)) + # [0] first sight → New broadcast - self._step(_active(250.0, 0), base, - expect_broadcast=True, expect_lifecycle="new") + _step(_active(250.0, 0), base, + expect_broadcast=True, expect_lifecycle="new") # [1] small growth 1h later (inside 8h cooldown) → suppress - self._step(_active(300.0, 0), base + 3600, - expect_broadcast=False, expect_lifecycle="cooldown") + _step(_active(300.0, 0), base + 3600, + expect_broadcast=False, expect_lifecycle="cooldown") # [2] growth after cooldown (8h) → Update - self._step(_active(500.0, 0), base + 28800, - expect_broadcast=True, expect_lifecycle="update") + _step(_active(500.0, 0), base + 28800, + expect_broadcast=True, expect_lifecycle="update") # [3] containment change after another cooldown → Update - self._step(_active(500.0, 40), base + 28800 * 2, - expect_broadcast=True, expect_lifecycle="update") + _step(_active(500.0, 40), base + 28800 * 2, + expect_broadcast=True, expect_lifecycle="update") # [4] tombstone → all-clear (fire was broadcast earlier) - tomb = _make_tombstone(irwin_id=irwin) - self._step(tomb, base + 100000, - expect_broadcast=True, expect_lifecycle="closed") + gate_t = fire_decide({"_kind": "wfigs_tombstone", "irwin_id": irwin}, + source="wfigs", now=float(base + 100000)) + assert gate_t.broadcast is True + assert gate_t.lifecycle == "closed" + assert gate_t.data_patch["category"] == "wildfire_closed" def test_dedup_suffix_tracks_state(self, mem_db): """_dedup_suffix carries the acres|contained that justified the @@ -278,10 +314,6 @@ class TestGateSequenceParity: gate = self._decide(tomb, 1_000_000) assert gate.broadcast is False assert gate.lifecycle == "suppress" - # Handler agrees: returns None. - out = handle_wfigs(_normalize_wfigs(tomb), tomb, tomb["subject"], - data={}, now=1_000_000) - assert out is None def test_perimeter_never_broadcasts(self, mem_db): from tests.test_wfigs_handler import _make_perimeter @@ -314,10 +346,6 @@ class TestRegistration: def test_firms_categories_not_captured(self, cat): # These native FIRMS categories remain deferred; they must NOT resolve # to the fire formatter/decider via the "fire" toggle family fallback. - # (Phase-3c migrated the FIRMS FUSION categories wildfire_growth / - # wildfire_spotting / wildfire_halted — covered in test_firms_refactor.py - # — but growth reuses the fire FORMATTER while keeping its OWN firms - # DECIDER, so it is intentionally not asserted here.) from meshai.notifications.formatters import get_formatter from meshai.notifications.gating import get_decider assert get_formatter(cat) is not fire_format diff --git a/work/tests/test_fire_tracker_phase1.py b/work/tests/test_fire_tracker_phase1.py index 5d0ab4a..67bd11d 100644 --- a/work/tests/test_fire_tracker_phase1.py +++ b/work/tests/test_fire_tracker_phase1.py @@ -50,29 +50,28 @@ def _seed_fire(*, irwin_id, lat, lon, name="Stub Fire", state="ID"): ) -def _envelope(*, lat, lon, acq_date="2026-06-06", acq_time="1200", - frp=15.0, satellite="N20", conf="high"): - """Build a Central FIRMS envelope shaped like the real Central feed.""" - return { - "data": { - "adapter": "firms", - "category": "wildfire_hotspot", - "severity": "routine", - "data": { - "latitude": lat, - "longitude": lon, - "frp": frp, - "bright_ti4": 320.5, - "satellite": satellite, - "instrument": "VIIRS", - "confidence": conf, - "acq_date": acq_date, - "acq_time": acq_time, - "daynight": "D", - "version": "2.0NRT", - }, - } - } +def _pixel(*, lat, lon, acq_date="2026-06-06", acq_time="1200", + frp=15.0, satellite="N20", conf="high"): + """Build a canonical FIRMS pixel dict for the LIVE ingest_hotspot_pixel + entrypoint (mirrors tests/test_firms_native_fusion.py's `_pixel`).""" + import datetime as _dt + acq_epoch = int(_dt.datetime.strptime( + f"{acq_date} {str(acq_time).zfill(4)}", "%Y-%m-%d %H%M" + ).replace(tzinfo=_dt.timezone.utc).timestamp()) + return {"lat": lat, "lon": lon, "frp": frp, "confidence": conf, + "brightness": 320.5, "satellite": satellite, "acq_epoch": acq_epoch} + + +def _ingest(pixel, *, now): + """Feed one pixel through the LIVE native entrypoint (chore/ripout-2dii: + the dead Central handle_firms wrapper this file used to drive through is + gone) and return the (wire, data) of the first broadcast it produced (or + (None, {})).""" + from meshai.env.fire_fusion import ingest_hotspot_pixel + broadcasts = ingest_hotspot_pixel(pixel, now=now) + if broadcasts: + return broadcasts[0] + return None, {} # --------------------------------------------------------------------------- @@ -81,7 +80,6 @@ def _envelope(*, lat, lon, acq_date="2026-06-06", acq_time="1200", def test_pixel_within_radius_attributes_to_fire(): - from meshai.env.fire_fusion import handle_firms from meshai.persistence import get_db # Cache Peak Fire stub @ 42.118, -113.643. @@ -89,9 +87,8 @@ def test_pixel_within_radius_attributes_to_fire(): lat=42.118, lon=-113.643) # FIRMS pixel ~0.2 mi NE of the anchor -- well inside default 5 mi. - env = _envelope(lat=42.121, lon=-113.640, frp=18.0) - wire = handle_firms(env, subject="central.fire.hotspot.N20.high.us.id", - data={}, now=1780728000) + wire, _data = _ingest(_pixel(lat=42.121, lon=-113.640, frp=18.0), + now=1780728000) # Attribution is silent (return None); the wire only fires on cluster. assert wire is None @@ -120,7 +117,6 @@ def test_pixel_within_radius_attributes_to_fire(): def test_centroid_recomputes_as_median_across_passes(): """A second attributed pixel updates the centroid to the median, not just the latest pixel's coords.""" - from meshai.env.fire_fusion import handle_firms from meshai.persistence import get_db _seed_fire(irwin_id="ID-TEST-002", @@ -135,12 +131,9 @@ def test_centroid_recomputes_as_median_across_passes(): # arithmetic mean) survives the Phase 2 semantic shift. coords = [(42.001, -113.001), (42.002, -113.002), (42.003, -113.003)] for i, (la, lo) in enumerate(coords): - env = _envelope(lat=la, lon=lo, - acq_date="2026-06-06", - acq_time=f"12{i * 10:02d}") # 12:00, 12:10, 12:20 - handle_firms(env, - subject="central.fire.hotspot.N20.high.us.id", - data={}, now=1780728000 + i * 600) + _ingest(_pixel(lat=la, lon=lo, acq_date="2026-06-06", + acq_time=f"12{i * 10:02d}"), # 12:00, 12:10, 12:20 + now=1780728000 + i * 600) fire = get_db().execute( "SELECT current_centroid_lat, current_centroid_lon " @@ -157,16 +150,14 @@ def test_centroid_recomputes_as_median_across_passes(): def test_pixel_outside_radius_stays_unattributed(): - from meshai.env.fire_fusion import handle_firms from meshai.persistence import get_db _seed_fire(irwin_id="ID-TEST-003", lat=42.000, lon=-113.000) # Pixel ~50 mi away -- comfortably outside the 5 mi default. - env = _envelope(lat=42.700, lon=-113.000, frp=10.0) - wire = handle_firms(env, subject="central.fire.hotspot.N20.high.us.id", - data={}, now=1780728000) + wire, _data = _ingest(_pixel(lat=42.700, lon=-113.000, frp=10.0), + now=1780728000) # No attribution AND below the 3-pixel cluster threshold -> no wire. assert wire is None @@ -196,7 +187,6 @@ def _hhmm(h, m=0): def test_three_unattributed_pixels_fire_cluster_once(): - from meshai.env.fire_fusion import handle_firms from meshai.persistence import get_db # No fires seeded -- everything is unattributed. @@ -209,12 +199,9 @@ def test_three_unattributed_pixels_fire_cluster_once(): ] wires: list[str | None] = [] for la, lo, t, frp in pixels: - env = _envelope(lat=la, lon=lo, acq_time=t, frp=frp) - data = {} - wires.append(handle_firms( - env, subject="central.fire.hotspot.N20.high.unknown", - data=data, now=1780728000, - )) + wire, data = _ingest(_pixel(lat=la, lon=lo, acq_time=t, frp=frp), + now=1780728000) + wires.append(wire) if wires[-1] is not None: # The handler must have tagged data with the cluster category. assert data.get("category") == "unattributed_hotspot_cluster" @@ -235,7 +222,6 @@ def test_three_unattributed_pixels_fire_cluster_once(): def test_fourth_pixel_in_same_cluster_does_not_refire(): - from meshai.env.fire_fusion import handle_firms from meshai.persistence import get_db base_lat, base_lon = 43.500, -114.500 @@ -245,18 +231,12 @@ def test_fourth_pixel_in_same_cluster_does_not_refire(): (base_lat + 0.001, base_lon + 0.001, "1210"), (base_lat - 0.001, base_lon - 0.002, "1220"), ]): - env = _envelope(lat=la, lon=lo, acq_time=t) - handle_firms(env, - subject="central.fire.hotspot.N20.high.unknown", - data={}, now=1780728000 + i) + _ingest(_pixel(lat=la, lon=lo, acq_time=t), now=1780728000 + i) # A 4th pixel inside the same cluster footprint. - env = _envelope(lat=base_lat + 0.0005, lon=base_lon - 0.0005, - acq_time="1230") - data4 = {} - wire = handle_firms(env, - subject="central.fire.hotspot.N20.high.unknown", - data=data4, now=1780728100) + wire, data4 = _ingest( + _pixel(lat=base_lat + 0.0005, lon=base_lon - 0.0005, acq_time="1230"), + now=1780728100) # The existing 3 members already have cluster_broadcast_at stamped, # so they don't count toward the new cluster query (the SQL filter # is `cluster_broadcast_at IS NULL`). The 4th pixel alone fails @@ -279,8 +259,6 @@ def test_fifth_pixel_after_time_window_can_form_new_cluster(): no nearby unstamped pixels to count, so it stays silent -- but if we then ingest TWO more nearby pixels (also outside the original window), we should fire a NEW cluster.""" - from meshai.env.fire_fusion import handle_firms - base_lat, base_lon = 43.500, -114.500 # First cluster at 12:00..12:20 -> fires + stamps all 3. for i, (la, lo, t) in enumerate([ @@ -288,10 +266,7 @@ def test_fifth_pixel_after_time_window_can_form_new_cluster(): (base_lat + 0.001, base_lon + 0.001, "1210"), (base_lat - 0.001, base_lon - 0.002, "1220"), ]): - env = _envelope(lat=la, lon=lo, acq_time=t) - handle_firms(env, - subject="central.fire.hotspot.N20.high.unknown", - data={}, now=1780728000 + i) + _ingest(_pixel(lat=la, lon=lo, acq_time=t), now=1780728000 + i) # Three NEW pixels at 14:00..14:20 -- well past the 60 min window # from the first cluster (which ended at 12:20 acq time). @@ -302,11 +277,9 @@ def test_fifth_pixel_after_time_window_can_form_new_cluster(): (base2_lat + 0.001, base2_lon + 0.001, "1410"), (base2_lat - 0.001, base2_lon - 0.002, "1420"), ]): - env = _envelope(lat=la, lon=lo, acq_time=t) - wires2.append(handle_firms( - env, subject="central.fire.hotspot.N20.high.unknown", - data={}, now=1780728000 + 7200 + i, - )) + wire, _data = _ingest(_pixel(lat=la, lon=lo, acq_time=t), + now=1780728000 + 7200 + i) + wires2.append(wire) # F3: the first cluster's 3 members are stamped (excluded from the query), # so the second batch forms an independent NEW cluster -- one wire on its # 3rd pixel. @@ -321,9 +294,12 @@ def test_fifth_pixel_after_time_window_can_form_new_cluster(): def test_wfigs_first_sight_tags_wildfire_declared(): - from meshai.env.fire_render import handle_wfigs + # chore/ripout-2dii: drives the LIVE decider (gating.fire.decide) directly + # -- handle_wfigs (the dead Central entrypoint this used to route through) + # is gone. Same canonical schema env/fires.py::to_event builds. + from meshai.notifications.gating.fire import decide as fire_decide - normalized = { + canonical = { "_kind": "wfigs_incident", "irwin_id": "ID-NEW-001", "incident_name": "Pine Gulch", @@ -334,24 +310,17 @@ def test_wfigs_first_sight_tags_wildfire_declared(): "county": "Twin Falls", "state": "ID", "declared_at_epoch": 1780728000, } - envelope = { - "data": {"adapter": "wfigs", "category": "wildfire_incident", - "severity": "priority"} - } - data = {} - wire = handle_wfigs(normalized, envelope, - subject="central.fire.incident.id", - data=data, now=1780728000) - assert wire is not None - assert "New" in wire - assert data.get("category") == "wildfire_declared", \ - f"expected wildfire_declared, got data={data!r}" + gate = fire_decide(canonical, source="wfigs", now=1780728000.0) + assert gate.broadcast is True + assert gate.lifecycle == "new" + assert gate.data_patch.get("category") == "wildfire_declared", \ + f"expected wildfire_declared, got data_patch={gate.data_patch!r}" def test_wfigs_update_does_not_retag_wildfire_declared(): """After a row exists AND has been broadcast, an acres-grew Update must NOT carry the wildfire_declared category.""" - from meshai.env.fire_render import handle_wfigs + from meshai.notifications.gating.fire import decide as fire_decide from meshai.persistence import get_db # Pre-existing row that has already been broadcast. @@ -364,7 +333,7 @@ def test_wfigs_update_does_not_retag_wildfire_declared(): ("ID-UPD-001", "Pine Gulch", 250.0, 0, 42.93, -114.45, 1780728000, 1780728000, 250.0, 0), ) - normalized = { + canonical = { "_kind": "wfigs_incident", "irwin_id": "ID-UPD-001", "incident_name": "Pine Gulch", @@ -374,19 +343,12 @@ def test_wfigs_update_does_not_retag_wildfire_declared(): "lat": 42.93, "lon": -114.45, "county": "Twin Falls", "state": "ID", } - envelope = { - "data": {"adapter": "wfigs", "category": "wildfire_incident", - "severity": "priority"} - } - data = {} # 8h cooldown clear: now > 28800s after last_broadcast_at. - wire = handle_wfigs(normalized, envelope, - subject="central.fire.incident.id", - data=data, now=1780728000 + 30000) - assert wire is not None - assert "Update" in wire + gate = fire_decide(canonical, source="wfigs", now=1780728000.0 + 30000) + assert gate.broadcast is True + assert gate.lifecycle == "update" # Update branch must NOT re-tag with wildfire_declared. - assert data.get("category") != "wildfire_declared" + assert "category" not in gate.data_patch # --------------------------------------------------------------------------- diff --git a/work/tests/test_fire_tracker_phase2.py b/work/tests/test_fire_tracker_phase2.py index 9fbe862..f6a498a 100644 --- a/work/tests/test_fire_tracker_phase2.py +++ b/work/tests/test_fire_tracker_phase2.py @@ -47,22 +47,28 @@ def _seed_fire(*, irwin_id, lat, lon, name="Stub Fire"): ) -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", - }, - } - } +def _pixel(*, lat, lon, acq_date="2026-06-06", acq_time="1200", + frp=20.0, satellite="N20"): + """Build a canonical FIRMS pixel dict for the LIVE ingest_hotspot_pixel + entrypoint (mirrors tests/test_firms_native_fusion.py's `_pixel`).""" + import datetime as _dt + acq_epoch = int(_dt.datetime.strptime( + f"{acq_date} {str(acq_time).zfill(4)}", "%Y-%m-%d %H%M" + ).replace(tzinfo=_dt.timezone.utc).timestamp()) + return {"lat": lat, "lon": lon, "frp": frp, "confidence": "high", + "brightness": 320.0, "satellite": satellite, "acq_epoch": acq_epoch} + + +def _ingest(pixel, *, now): + """Feed one pixel through the LIVE native entrypoint (chore/ripout-2dii: + the dead Central handle_firms wrapper this file used to drive through is + gone) and return the (wire, data) of the first broadcast it produced (or + (None, {})).""" + from meshai.env.fire_fusion import ingest_hotspot_pixel + broadcasts = ingest_hotspot_pixel(pixel, now=now) + if broadcasts: + return broadcasts[0] + return None, {} # --------------------------------------------------------------------------- @@ -73,7 +79,6 @@ def _envelope(*, lat, lon, acq_date="2026-06-06", acq_time="1200", def test_two_pass_drift_emits_growth_with_direction_and_speed(): """Pass 1 (N20 bucket A), pass 2 (N20 bucket B, ~6h later, centroid 1.0 mi N of pass 1). Drift should be ~1.0 mi N, broadcast must fire.""" - from meshai.env.fire_fusion import handle_firms from meshai.persistence import get_db _seed_fire(irwin_id="ID-GROWTH-001", @@ -87,26 +92,19 @@ def test_two_pass_drift_emits_growth_with_direction_and_speed(): # Pass A pixels (5 pixels tightly clustered around (42.000, -114.000)). for i in range(5): - env = _envelope( + _ingest(_pixel( lat=pass_a_lat + 0.0001 * i, lon=-114.000 + 0.0001 * (i - 2), acq_date="2026-06-06", acq_time=f"{12:02d}{0 + i:02d}", frp=20.0 + i, satellite="N20", - ) - handle_firms(env, subject="central.fire.hotspot.N20.high.us.id", - data={}, now=1780747200 + i) + ), now=1780747200 + i) # First pixel of pass B fires the growth broadcast. - env_b_first = _envelope( + wire, data_b = _ingest(_pixel( lat=pass_b_lat, lon=-114.000, acq_date="2026-06-06", acq_time="1800", frp=22.0, satellite="N20", - ) - data_b = {} - wire = handle_firms( - env_b_first, subject="central.fire.hotspot.N20.high.us.id", - data=data_b, now=1780768800, - ) + ), now=1780768800) assert wire is not None, "pass-B boundary should fire growth broadcast" assert "Moving N" in wire, f"expected N direction, got: {wire}" assert data_b.get("category") == "wildfire_growth" @@ -139,7 +137,6 @@ def test_two_pass_drift_emits_growth_with_direction_and_speed(): def test_drift_below_threshold_does_not_emit_growth(): """0.3 mi drift between consecutive passes -- below the 0.5 mi default -- must NOT broadcast wildfire_growth.""" - from meshai.env.fire_fusion import handle_firms from meshai.persistence import get_db _seed_fire(irwin_id="ID-DRIFT-001", @@ -148,19 +145,15 @@ def test_drift_below_threshold_does_not_emit_growth(): # Pass A: 3 pixels. for i in range(3): - env = _envelope(lat=43.000 + 0.0001 * i, lon=-115.000, - acq_time=f"{12:02d}{i:02d}", - frp=15.0, satellite="N20") - handle_firms(env, subject="central.fire.hotspot.N20.high.us.id", - data={}, now=1780747200 + i) + _ingest(_pixel(lat=43.000 + 0.0001 * i, lon=-115.000, + acq_time=f"{12:02d}{i:02d}", + frp=15.0, satellite="N20"), now=1780747200 + i) # Pass B: 0.3 mi N (below threshold). pass_b_lat = 43.000 + (0.3 / 69.0) - env_b = _envelope(lat=pass_b_lat, lon=-115.000, - acq_time="1800", frp=15.0, satellite="N20") - data_b = {} - wire = handle_firms(env_b, subject="central.fire.hotspot.N20.high.us.id", - data=data_b, now=1780768800) + wire, data_b = _ingest(_pixel(lat=pass_b_lat, lon=-115.000, + acq_time="1800", frp=15.0, satellite="N20"), + now=1780768800) assert wire is None, f"sub-threshold drift should NOT broadcast: {wire}" assert data_b.get("category") != "wildfire_growth" # The pass row still exists with the (sub-threshold) drift recorded. @@ -181,7 +174,7 @@ def test_drift_below_threshold_does_not_emit_growth(): def test_halt_detector_fires_once_after_12h_idle(): """Fire with last_pass_at 14h ago + no new pixels in that fire triggers halt on the next FIRMS pixel arrival (for any fire).""" - from meshai.env.fire_fusion import handle_firms, _maybe_emit_halt + from meshai.env.fire_fusion import _maybe_emit_halt from meshai.persistence import get_db now_epoch = 1780768800 # 2026-06-06 18:00 UTC @@ -339,7 +332,6 @@ def test_pass_row_aggregates_match_member_pixels(): """5 pixels attributed in the same pass yield ONE fire_passes row with pixel_count=5, total_frp = sum, pass_started_at = min(acq), pass_ended_at = max(acq), centroid = median.""" - from meshai.env.fire_fusion import handle_firms from meshai.persistence import get_db _seed_fire(irwin_id="ID-AGG-001", @@ -354,10 +346,8 @@ def test_pass_row_aggregates_match_member_pixels(): (42.004, -114.004, "1220", 50.0), ] for la, lo, t, frp in pixels: - env = _envelope(lat=la, lon=lo, acq_time=t, frp=frp, - satellite="N20") - handle_firms(env, subject="central.fire.hotspot.N20.high.us.id", - data={}, now=1780747200) + _ingest(_pixel(lat=la, lon=lo, acq_time=t, frp=frp, satellite="N20"), + now=1780747200) row = get_db().execute( "SELECT pixel_count, total_frp, pass_centroid_lat, " diff --git a/work/tests/test_fire_tracker_phase3.py b/work/tests/test_fire_tracker_phase3.py index 2b4d0a2..894db73 100644 --- a/work/tests/test_fire_tracker_phase3.py +++ b/work/tests/test_fire_tracker_phase3.py @@ -32,22 +32,28 @@ def _seed_fire(*, irwin_id, lat, lon, name="Stub Fire"): ) -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", - }, - } - } +def _pixel(*, lat, lon, acq_date="2026-06-06", acq_time="1200", + frp=20.0, satellite="N20"): + """Build a canonical FIRMS pixel dict for the LIVE ingest_hotspot_pixel + entrypoint (mirrors tests/test_firms_native_fusion.py's `_pixel`).""" + import datetime as _dt + acq_epoch = int(_dt.datetime.strptime( + f"{acq_date} {str(acq_time).zfill(4)}", "%Y-%m-%d %H%M" + ).replace(tzinfo=_dt.timezone.utc).timestamp()) + return {"lat": lat, "lon": lon, "frp": frp, "confidence": "high", + "brightness": 320.0, "satellite": satellite, "acq_epoch": acq_epoch} + + +def _ingest(pixel, *, now): + """Feed one pixel through the LIVE native entrypoint (chore/ripout-2dii: + the dead Central handle_firms wrapper this file used to drive through is + gone) and return the (wire, data) of the first broadcast it produced (or + (None, {})).""" + from meshai.env.fire_fusion import ingest_hotspot_pixel + broadcasts = ingest_hotspot_pixel(pixel, now=now) + if broadcasts: + return broadcasts[0] + return None, {} # Useful constant: 1 mi in latitude degrees. @@ -71,7 +77,6 @@ def test_pass_close_stamps_perimeter_geojson(): """Pass A: 6 pixels in a hex around the seeded center. First pixel of pass B triggers boundary close -> perimeter_geojson written for pass A as a closed GeoJSON Polygon.""" - from meshai.env.fire_fusion import handle_firms from meshai.persistence import get_db center_lat, center_lon = 42.500, -114.500 @@ -83,9 +88,8 @@ def test_pass_close_stamps_perimeter_geojson(): angle = i * math.pi / 3 la = center_lat + 0.001 * math.sin(angle) lo = center_lon + 0.001 * math.cos(angle) - env = _envelope(lat=la, lon=lo, acq_time=f"12{i * 2:02d}") - handle_firms(env, subject="central.fire.hotspot.N20.high.us.id", - data={}, now=1780747200 + i) + _ingest(_pixel(lat=la, lon=lo, acq_time=f"12{i * 2:02d}"), + now=1780747200 + i) # First pass B pixel 1 mi N. Within the 5 mi spread radius -> the # pixel attributes to the seeded fire -> boundary detected -> @@ -93,9 +97,7 @@ def test_pass_close_stamps_perimeter_geojson(): # spotting threshold so this also avoids spotting noise in the # test (we just want perimeter_geojson to materialize). far_lat, far_lon = _offset_mi(center_lat, center_lon, north_mi=1.0) - env_b = _envelope(lat=far_lat, lon=far_lon, acq_time="1800") - handle_firms(env_b, subject="central.fire.hotspot.N20.high.us.id", - data={}, now=1780768800) + _ingest(_pixel(lat=far_lat, lon=far_lon, acq_time="1800"), now=1780768800) row = get_db().execute( "SELECT perimeter_geojson FROM fire_passes WHERE irwin_id=? " @@ -121,7 +123,6 @@ def _seed_pass_a_hex_then_close(*, irwin_id, center_lat, center_lon, start_now=1780747200): """Helper: seed a fire + 6 hex-vertex pass A pixels. Caller follows up with a pass B pixel to trigger boundary close + perimeter write.""" - from meshai.env.fire_fusion import handle_firms _seed_fire(irwin_id=irwin_id, lat=center_lat, lon=center_lon, name=irwin_id) for i in range(6): @@ -130,15 +131,13 @@ def _seed_pass_a_hex_then_close(*, irwin_id, center_lat, center_lon, la = center_lat + (0.5 / _MI_PER_DEG_LAT) * math.sin(angle) cos_lat = math.cos(math.radians(center_lat)) lo = center_lon + (0.5 / (_MI_PER_DEG_LAT * cos_lat)) * math.cos(angle) - env = _envelope(lat=la, lon=lo, acq_time=f"12{i * 2:02d}") - handle_firms(env, subject="central.fire.hotspot.N20.high.us.id", - data={}, now=start_now + i) + _ingest(_pixel(lat=la, lon=lo, acq_time=f"12{i * 2:02d}"), + now=start_now + i) def test_pixel_2mi_ne_of_perimeter_emits_spotting(): """Pass B pixel 2 mi NE of pass A's perimeter centroid fires wildfire_spotting with the correct distance + direction.""" - from meshai.env.fire_fusion import handle_firms from meshai.persistence import get_db center_lat, center_lon = 43.000, -115.000 @@ -151,11 +150,8 @@ def test_pixel_2mi_ne_of_perimeter_emits_spotting(): center_lat, center_lon, north_mi=2.0 / math.sqrt(2), east_mi=2.0 / math.sqrt(2), ) - env_b = _envelope(lat=sp_lat, lon=sp_lon, acq_time="1800") - data = {} - wire = handle_firms(env_b, - subject="central.fire.hotspot.N20.high.us.id", - data=data, now=1780768800) + wire, data = _ingest(_pixel(lat=sp_lat, lon=sp_lon, acq_time="1800"), + now=1780768800) assert wire is not None assert wire.startswith("🔥 Possible spotting ") assert "NE of ID-SPOT-002 perimeter" in wire @@ -188,8 +184,6 @@ def test_pixel_2mi_ne_of_perimeter_emits_spotting(): def test_pixel_inside_perimeter_no_spotting(): - from meshai.env.fire_fusion import handle_firms - center_lat, center_lon = 43.500, -114.500 _seed_pass_a_hex_then_close(irwin_id="ID-SPOT-003", center_lat=center_lat, @@ -197,22 +191,15 @@ def test_pixel_inside_perimeter_no_spotting(): # Close the perimeter with one boundary-only pixel ~10 mi away. far_lat, far_lon = _offset_mi(center_lat, center_lon, north_mi=10.0) - env_close = _envelope(lat=far_lat, lon=far_lon, acq_time="1800") - handle_firms(env_close, - subject="central.fire.hotspot.N20.high.us.id", - data={}, now=1780768800) + _ingest(_pixel(lat=far_lat, lon=far_lon, acq_time="1800"), now=1780768800) # Now ingest a pixel 0.1 mi NE of center -- well inside the 0.5 mi # radius hex. inside_lat, inside_lon = _offset_mi( center_lat, center_lon, north_mi=0.05, east_mi=0.05, ) - env_inside = _envelope(lat=inside_lat, lon=inside_lon, - acq_time="1810") - data = {} - wire = handle_firms(env_inside, - subject="central.fire.hotspot.N20.high.us.id", - data=data, now=1780768900) + wire, data = _ingest(_pixel(lat=inside_lat, lon=inside_lon, + acq_time="1810"), now=1780768900) assert wire is None or "spotting" not in (wire or "") assert data.get("category") != "wildfire_spotting" @@ -223,8 +210,6 @@ def test_pixel_inside_perimeter_no_spotting(): def test_second_spotting_within_cooldown_suppressed(): - from meshai.env.fire_fusion import handle_firms - center_lat, center_lon = 44.000, -116.000 _seed_pass_a_hex_then_close(irwin_id="ID-SPOT-004", center_lat=center_lat, @@ -232,10 +217,8 @@ def test_second_spotting_within_cooldown_suppressed(): # First spotting pixel 2 mi N. sp1_lat, sp1_lon = _offset_mi(center_lat, center_lon, north_mi=2.0) - env1 = _envelope(lat=sp1_lat, lon=sp1_lon, acq_time="1800") - wire1 = handle_firms(env1, - subject="central.fire.hotspot.N20.high.us.id", - data={}, now=1780768800) + wire1, _data1 = _ingest(_pixel(lat=sp1_lat, lon=sp1_lon, acq_time="1800"), + now=1780768800) assert wire1 is not None and "spotting" in wire1 # Second spotting candidate 30 min later, 2 mi SE (different @@ -244,10 +227,8 @@ def test_second_spotting_within_cooldown_suppressed(): center_lat, center_lon, north_mi=-2.0 / math.sqrt(2), east_mi=2.0 / math.sqrt(2), ) - env2 = _envelope(lat=sp2_lat, lon=sp2_lon, acq_time="1830") - wire2 = handle_firms(env2, - subject="central.fire.hotspot.N20.high.us.id", - data={}, now=1780768800 + 1800) + wire2, _data2 = _ingest(_pixel(lat=sp2_lat, lon=sp2_lon, acq_time="1830"), + now=1780768800 + 1800) assert wire2 is None, f"second spotting in cooldown should suppress: {wire2}" @@ -257,26 +238,20 @@ def test_second_spotting_within_cooldown_suppressed(): def test_spotting_refires_after_cooldown(): - from meshai.env.fire_fusion import handle_firms - center_lat, center_lon = 44.500, -116.500 _seed_pass_a_hex_then_close(irwin_id="ID-SPOT-005", center_lat=center_lat, center_lon=center_lon) sp1_lat, sp1_lon = _offset_mi(center_lat, center_lon, north_mi=2.0) - env1 = _envelope(lat=sp1_lat, lon=sp1_lon, acq_time="1800") - wire1 = handle_firms(env1, - subject="central.fire.hotspot.N20.high.us.id", - data={}, now=1780768800) + wire1, _data1 = _ingest(_pixel(lat=sp1_lat, lon=sp1_lon, acq_time="1800"), + now=1780768800) assert wire1 is not None # 70 minutes later -> past the 1h cooldown -> next spotting fires. sp2_lat, sp2_lon = _offset_mi(center_lat, center_lon, north_mi=-2.0) - env2 = _envelope(lat=sp2_lat, lon=sp2_lon, acq_time="1910") - wire2 = handle_firms(env2, - subject="central.fire.hotspot.N20.high.us.id", - data={}, now=1780768800 + 70 * 60) + wire2, _data2 = _ingest(_pixel(lat=sp2_lat, lon=sp2_lon, acq_time="1910"), + now=1780768800 + 70 * 60) assert wire2 is not None and "spotting" in wire2 diff --git a/work/tests/test_firms_handler.py b/work/tests/test_firms_handler.py index 7f685ba..bf5e451 100644 --- a/work/tests/test_firms_handler.py +++ b/work/tests/test_firms_handler.py @@ -1,17 +1,27 @@ -"""Tests for v0.6-1 FIRMS handler (storage-only). +"""Tests for the shared FIRMS pixel-ingest core (`_ingest_pixel_core`) + +`_parse_acq_epoch` -- the parts of the retired v0.6-1 FIRMS handler that are +still LIVE, shared by the native `env/firms.py` adapter via +`ingest_hotspot_pixel`. -The handler never returns a wire string (storage-only contract). All -assertions check side effects on `firms_pixels` + `event_log`. Per the -audit doc finding #2 the handler must close the v0.5.13 silent-drop on -`central.fire.hotspot.>` envelopes. +chore/ripout-2dii: `handle_firms` (the dead Central NATS-envelope entrypoint +this file used to test -- confidence/FRP/bbox filtering, envelope field +extraction, missing-coords/missing-acq-time drops, non-firms-adapter guard, +event_log accounting) has been REMOVED from `meshai.env.fire_fusion` -- zero +live production callers, and no live equivalent (the native adapter's own +confidence/bbox filtering happens upstream in `env/firms.py`'s CSV fetch, a +different code path entirely; see `tests/test_adapter_firms.py`). Those tests +were deleted with it. -Envelope shape sourced from firms-investigation.md sampling (250 envelopes -2026-05-28..06-04, all VIIRS). +What remains: the dedup behavior of `_ingest_pixel_core` (INSERT OR IGNORE on +a meters-quantized dedup key -- genuinely shared/live, exercised by BOTH the +native adapter and formerly by handle_firms) and `_parse_acq_epoch`'s +int/short/zero-padded acq_time parsing (also shared/live -- env/firms.py +imports and calls it directly). Both rewritten to drive +`ingest_hotspot_pixel` directly instead of the dead handler. """ import pytest -from meshai.env import fire_fusion as firms_handler -from meshai.env.fire_fusion import handle_firms +from meshai.env.fire_fusion import ingest_hotspot_pixel, _parse_acq_epoch from meshai.persistence import close_thread_connection, init_db from meshai.persistence import db as persistence_db @@ -31,319 +41,88 @@ def mem_db(monkeypatch, tmp_path): persistence_db._initialised.discard(db_path) -def _firms_env(*, - lat=42.19664, lon=-113.70981, frp=135.93, - confidence="high", satellite="N", - acq_date="2026-05-28", acq_time="1949", - bright_ti4=367.0, daynight="D", - envelope_id="firms_001", - category="fire.hotspot.viirs", - severity=3, - region="unknown", - extras=None): - """Build a FIRMS CloudEvents envelope matching the live wire shape. - - Default fixture is SAMPLE B from firms-investigation.md (Cache Peak - high-confidence 135 MW fire). Override individual fields for filter - coverage. - """ - payload = { - "id": envelope_id, - "latitude": lat, - "longitude": lon, - "frp": frp, - "confidence": confidence, - "satellite": satellite, - "instrument": "VIIRS", - "acq_date": acq_date, - "acq_time": acq_time, - "bright_ti4": bright_ti4, - "daynight": daynight, - "version": "2.0NRT", - "_enriched": {"geocoder": {"landclass": "Cache Peak Roadless Area", - "elevation_m": 2151.9}}, - } - if extras: payload.update(extras) - return { - "subject": f"central.fire.hotspot.viirs_snpp.{confidence}.{region}", - "id": envelope_id, - "data": { - "id": envelope_id, - "adapter": "firms", - "category": category, - "severity": severity, - "time": "2026-05-28T19:49:00Z", - "geo": {"centroid": [lon, lat]}, - "data": payload, - }, - } +def _pixel(*, lat=42.19664, lon=-113.70981, frp=135.93, confidence="high", + satellite="N", acq_date="2026-05-28", acq_time="1949", + brightness=367.0): + acq_epoch = _parse_acq_epoch(acq_date, acq_time) + return {"lat": lat, "lon": lon, "frp": frp, "confidence": confidence, + "brightness": brightness, "satellite": satellite, + "acq_epoch": acq_epoch} def _row_count(mem_db, table): return mem_db.execute(f"SELECT COUNT(*) AS n FROM {table}").fetchone()["n"] -def _last_event_log(mem_db): - r = mem_db.execute( - "SELECT * FROM event_log ORDER BY id DESC LIMIT 1" - ).fetchone() - return dict(r) if r else None - - # ============================================================================ -# Happy path: high-confidence pixel stored +# _parse_acq_epoch -- acq_time format quirks (shared with env/firms.py) # ============================================================================ -def test_high_confidence_pixel_persisted(mem_db): - env = _firms_env() - out = handle_firms(env, env["subject"], data={}, now=1_780_660_000) +def test_acq_time_int_accepted(): + """FIRMS sometimes publishes acq_time as int 2013 rather than '2013'.""" + assert _parse_acq_epoch("2026-05-28", 1949) is not None + assert _parse_acq_epoch("2026-05-28", 1949) == _parse_acq_epoch("2026-05-28", "1949") - assert out is None, "storage-only handler must never return a wire string" - assert _row_count(mem_db, "firms_pixels") == 1 - row = mem_db.execute("SELECT * FROM firms_pixels").fetchone() - assert row["lat"] == pytest.approx(42.19664) - assert row["lon"] == pytest.approx(-113.70981) - assert row["frp"] == pytest.approx(135.93) - assert row["confidence"] == "high" - assert row["satellite"] == "N" - assert row["brightness"] == pytest.approx(367.0) - assert row["irwin_id"] is None # unattached; v0.6 fire-tracker fills later - - # acq_time should be the parsed UTC epoch of 2026-05-28 19:49Z. +def test_short_acq_time_zero_padded(): + """acq_time '49' (early-morning pass) must zero-pad to '0049'.""" + epoch = _parse_acq_epoch("2026-05-28", "49") + assert epoch is not None import datetime as _dt - expected = int(_dt.datetime(2026, 5, 28, 19, 49, - tzinfo=_dt.timezone.utc).timestamp()) - assert row["acq_time"] == expected - - # event_log: handled=1, table_name set, table_pk = inserted rowid. - log = _last_event_log(mem_db) - assert log["source"] == "firms" - assert log["handled"] == 1 - assert log["table_name"] == "firms_pixels" - assert log["table_pk"] == str(row["id"]) + dt = _dt.datetime.fromtimestamp(epoch, tz=_dt.timezone.utc) + assert (dt.hour, dt.minute) == (0, 49) -def test_nominal_confidence_pixel_persisted_under_default_floor(mem_db): - """Default FIRMS_CONFIDENCE_FLOOR='low' must accept nominal/high/low.""" - env = _firms_env(confidence="nominal", envelope_id="firms_002") - out = handle_firms(env, env["subject"], data={}, now=1_780_660_000) - assert out is None - assert _row_count(mem_db, "firms_pixels") == 1 - - -def test_low_confidence_pixel_persisted_under_default_floor(mem_db): - env = _firms_env(confidence="low", envelope_id="firms_003") - out = handle_firms(env, env["subject"], data={}, now=1_780_660_000) - assert out is None - assert _row_count(mem_db, "firms_pixels") == 1 +def test_missing_acq_time_returns_none(): + assert _parse_acq_epoch(None, None) is None + assert _parse_acq_epoch("2026-05-28", None) is None # ============================================================================ -# Confidence floor when bumped up -# ============================================================================ - - -def test_low_confidence_dropped_when_floor_is_nominal(mem_db, monkeypatch): - monkeypatch.setattr(firms_handler, "FIRMS_CONFIDENCE_FLOOR", "nominal") - env = _firms_env(confidence="low", envelope_id="firms_lo") - out = handle_firms(env, env["subject"], data={}, now=1_780_660_000) - assert out is None - assert _row_count(mem_db, "firms_pixels") == 0 - - log = _last_event_log(mem_db) - assert log["handled"] == 0 - assert log["category"].endswith("|below_confidence_floor") - - -def test_unknown_confidence_value_dropped(mem_db): - env = _firms_env(confidence="bogus", envelope_id="firms_bog") - out = handle_firms(env, env["subject"], data={}, now=1_780_660_000) - assert out is None - assert _row_count(mem_db, "firms_pixels") == 0 - log = _last_event_log(mem_db) - assert log["category"].endswith("|below_confidence_floor") - - -# ============================================================================ -# FRP floor -# ============================================================================ - - -def test_low_frp_dropped_when_floor_set(mem_db, monkeypatch): - monkeypatch.setattr(firms_handler, "FIRMS_FRP_FLOOR", 5.0) - env = _firms_env(frp=1.4, confidence="high", envelope_id="firms_frp_low") - out = handle_firms(env, env["subject"], data={}, now=1_780_660_000) - assert out is None - assert _row_count(mem_db, "firms_pixels") == 0 - log = _last_event_log(mem_db) - assert log["category"].endswith("|below_frp_floor") - - -def test_frp_at_floor_stored(mem_db, monkeypatch): - monkeypatch.setattr(firms_handler, "FIRMS_FRP_FLOOR", 5.0) - env = _firms_env(frp=5.0, confidence="high", envelope_id="firms_frp_at") - out = handle_firms(env, env["subject"], data={}, now=1_780_660_000) - assert out is None - assert _row_count(mem_db, "firms_pixels") == 1 - - -def test_missing_frp_dropped_when_floor_set(mem_db, monkeypatch): - monkeypatch.setattr(firms_handler, "FIRMS_FRP_FLOOR", 5.0) - env = _firms_env(confidence="high", envelope_id="firms_no_frp", - extras={"frp": None}) - out = handle_firms(env, env["subject"], data={}, now=1_780_660_000) - assert out is None - assert _row_count(mem_db, "firms_pixels") == 0 - - -def test_missing_frp_stored_when_floor_zero(mem_db): - """Default FIRMS_FRP_FLOOR=0: missing FRP still stores (null in column).""" - env = _firms_env(confidence="high", envelope_id="firms_no_frp_2", - extras={"frp": None}) - out = handle_firms(env, env["subject"], data={}, now=1_780_660_000) - assert out is None - assert _row_count(mem_db, "firms_pixels") == 1 - row = mem_db.execute("SELECT * FROM firms_pixels").fetchone() - assert row["frp"] is None - - -# ============================================================================ -# Bbox filter (default None = pass-through; explicit bbox tested both sides) -# ============================================================================ - - -def test_bbox_none_passes_through(mem_db): - env = _firms_env(lat=51.0, lon=10.0, # Germany - envelope_id="firms_eu") - out = handle_firms(env, env["subject"], data={}, now=1_780_660_000) - assert out is None - assert _row_count(mem_db, "firms_pixels") == 1 - - -def test_bbox_drops_outside(mem_db, monkeypatch): - # Idaho-ish bbox. - monkeypatch.setattr(firms_handler, "FIRMS_BBOX_OPTIONAL", - (42.0, -117.5, 49.0, -111.0)) - env = _firms_env(lat=51.0, lon=10.0, envelope_id="firms_out") - out = handle_firms(env, env["subject"], data={}, now=1_780_660_000) - assert out is None - assert _row_count(mem_db, "firms_pixels") == 0 - log = _last_event_log(mem_db) - assert log["category"].endswith("|outside_bbox") - - -def test_bbox_keeps_inside(mem_db, monkeypatch): - monkeypatch.setattr(firms_handler, "FIRMS_BBOX_OPTIONAL", - (42.0, -117.5, 49.0, -111.0)) - env = _firms_env(envelope_id="firms_in") # Cache Peak is inside - out = handle_firms(env, env["subject"], data={}, now=1_780_660_000) - assert out is None - assert _row_count(mem_db, "firms_pixels") == 1 - - -# ============================================================================ -# Dedup: same satellite pixel observation arriving twice = no-op +# Dedup: same satellite pixel observation arriving twice = no-op. +# `_ingest_pixel_core`'s meters-quantized dedup_key + INSERT OR IGNORE is the +# LIVE, source-agnostic core shared by ingest_hotspot_pixel. # ============================================================================ def test_dedup_same_pixel_idempotent(mem_db): - env = _firms_env(envelope_id="dup_001") - handle_firms(env, env["subject"], data={}, now=1_780_660_000) - handle_firms(env, env["subject"], data={}, now=1_780_660_001) - + p = _pixel() + ingest_hotspot_pixel(p, now=1_780_660_000) + ingest_hotspot_pixel(p, now=1_780_660_001) assert _row_count(mem_db, "firms_pixels") == 1, "OR IGNORE collapses dup" - # Two event_log rows; second is dedup_hit. - rows = mem_db.execute( - "SELECT category, table_name, handled FROM event_log " - "WHERE source='firms' ORDER BY id" - ).fetchall() - assert len(rows) == 2 - assert rows[0]["table_name"] == "firms_pixels" - assert rows[1]["table_name"] is None - assert rows[1]["category"].endswith("|dedup_hit") def test_dedup_collapses_lat_lon_float_noise(mem_db): """Same coord with sub-1m float noise must hit the same dedup key. - 5-decimal rounding => differences in the 6th+ decimal are absorbed.""" - e1 = _firms_env(lat=42.196641234567, lon=-113.709810000001, - envelope_id="dup_a") - e2 = _firms_env(lat=42.196641111111, lon=-113.709810999999, - envelope_id="dup_b") - handle_firms(e1, e1["subject"], data={}, now=1_780_660_000) - handle_firms(e2, e2["subject"], data={}, now=1_780_660_001) + Meters-based quantization absorbs differences well under 1 pixel.""" + p1 = _pixel(lat=42.196641234567, lon=-113.709810000001) + p2 = _pixel(lat=42.196641111111, lon=-113.709810999999) + ingest_hotspot_pixel(p1, now=1_780_660_000) + ingest_hotspot_pixel(p2, now=1_780_660_001) assert _row_count(mem_db, "firms_pixels") == 1 def test_dedup_different_satellite_stored_separately(mem_db): """Same coord + acq_time but different satellite is 2 distinct observations.""" - e1 = _firms_env(satellite="N", envelope_id="sat_n") - e2 = _firms_env(satellite="N20", envelope_id="sat_n20") - handle_firms(e1, e1["subject"], data={}, now=1_780_660_000) - handle_firms(e2, e2["subject"], data={}, now=1_780_660_001) + ingest_hotspot_pixel(_pixel(satellite="N"), now=1_780_660_000) + ingest_hotspot_pixel(_pixel(satellite="N20"), now=1_780_660_001) assert _row_count(mem_db, "firms_pixels") == 2 def test_dedup_different_acq_time_stored_separately(mem_db): """Same pixel observed on two passes 12h apart -> 2 rows.""" - e1 = _firms_env(acq_time="0700", envelope_id="t1") - e2 = _firms_env(acq_time="1900", envelope_id="t2") - handle_firms(e1, e1["subject"], data={}, now=1_780_660_000) - handle_firms(e2, e2["subject"], data={}, now=1_780_660_001) + ingest_hotspot_pixel(_pixel(acq_time="0700"), now=1_780_660_000) + ingest_hotspot_pixel(_pixel(acq_time="1900"), now=1_780_660_001) assert _row_count(mem_db, "firms_pixels") == 2 -# ============================================================================ -# Bad / missing inputs -# ============================================================================ - - -def test_missing_coords_dropped(mem_db): - env = _firms_env(envelope_id="no_coords", - extras={"latitude": None, "longitude": None}) - out = handle_firms(env, env["subject"], data={}, now=1_780_660_000) - assert out is None - assert _row_count(mem_db, "firms_pixels") == 0 - log = _last_event_log(mem_db) - assert log["category"].endswith("|missing_coords") - - -def test_missing_acq_time_dropped(mem_db): - env = _firms_env(envelope_id="no_acq", - extras={"acq_date": None, "acq_time": None}) - out = handle_firms(env, env["subject"], data={}, now=1_780_660_000) - assert out is None - assert _row_count(mem_db, "firms_pixels") == 0 - log = _last_event_log(mem_db) - assert log["category"].endswith("|missing_acq_time") - - -def test_non_firms_adapter_passes_through(mem_db): - """Defense in depth: handler must early-return on non-firms envelopes.""" - env = _firms_env(envelope_id="wrong_adapter") - env["data"]["adapter"] = "nws" - out = handle_firms(env, env["subject"], data={}, now=1_780_660_000) - assert out is None - assert _row_count(mem_db, "firms_pixels") == 0 - assert _row_count(mem_db, "event_log") == 0 - - -def test_acq_time_int_accepted(mem_db): - """FIRMS sometimes publishes acq_time as int 2013 rather than '2013'.""" - env = _firms_env(envelope_id="int_acq", - extras={"acq_time": 1949}) - out = handle_firms(env, env["subject"], data={}, now=1_780_660_000) - assert out is None +def test_missing_frp_stored(mem_db): + """A pixel with no FRP still stores (null in column) -- ingest_hotspot_pixel + never filters on FRP (that was handle_firms-specific, now gone).""" + p = _pixel() + p["frp"] = None + ingest_hotspot_pixel(p, now=1_780_660_000) assert _row_count(mem_db, "firms_pixels") == 1 - - -def test_short_acq_time_zero_padded(mem_db): - """acq_time '49' (early-morning pass) must zero-pad to '0049'.""" - env = _firms_env(envelope_id="short_acq", - extras={"acq_time": "49"}) - out = handle_firms(env, env["subject"], data={}, now=1_780_660_000) - assert out is None - assert _row_count(mem_db, "firms_pixels") == 1 - + row = mem_db.execute("SELECT * FROM firms_pixels").fetchone() + assert row["frp"] is None diff --git a/work/tests/test_firms_native_fusion.py b/work/tests/test_firms_native_fusion.py index d57709c..bab495a 100644 --- a/work/tests/test_firms_native_fusion.py +++ b/work/tests/test_firms_native_fusion.py @@ -380,68 +380,10 @@ class TestAdapterTickFusion: 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.env.fire_fusion 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.env.fire_fusion 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" +# chore/ripout-2dii: section 3 ("Central-path guard — extraction did not +# change handle_firms") REMOVED. `handle_firms` (the dead Central +# NATS-envelope entrypoint it drove) has been deleted from +# `meshai.env.fire_fusion` -- zero live production callers, and this section +# existed purely to prove that entrypoint was unchanged by the extraction. +# The shared core it guarded (_ingest_pixel_core) is exercised directly by +# every test above via `ingest_hotspot_pixel` (the live native entrypoint). diff --git a/work/tests/test_firms_refactor.py b/work/tests/test_firms_refactor.py index e866f31..8483d8f 100644 --- a/work/tests/test_firms_refactor.py +++ b/work/tests/test_firms_refactor.py @@ -5,24 +5,42 @@ fusion broadcast categories — wildfire_growth / wildfire_spotting / wildfire_halted — mirroring test_fire_refactor.py (WFIGS) and test_hydro_refactor.py: -1. Registration: the three categories resolve to the right formatter/decider; - wildfire_growth reuses the FIRE formatter but keeps its OWN firms decider; +1. Registration: wildfire_spotting/wildfire_halted resolve to the firms + formatter+decider; wildfire_growth is NOT formatter-registered (its wire + is always precomposed -- see below) but DOES keep its own firms decider; still-deferred native FIRMS categories do NOT resolve. 2. Gate-sequence + deferred-latch (the tier-b validation): a `now`-timeline - driven through the NEW gating.firms.decide() reproduces the OLD handle_firms - broadcast/suppress + stamp behavior, AND the latch is now DEFERRED — a - decision does NOT burn the latch until commit() fires (simulating delivery), - after which the next decision suppresses. + driven through gating.firms.decide() exercises broadcast/suppress + stamp + behavior, AND the latch is DEFERRED — a decision does NOT burn the latch + until commit() fires (simulating delivery), after which the next decision + suppresses. -3. Golden byte-identity: the cutover formatter path reproduces the legacy inline - wire byte-for-byte for growth (via fire.py), spotting, and halt (via firms.py). - -4. Not-cutover parity: with no category cut over, handle_firms keeps the legacy - eager-latch + stamps VERBATIM (byte-identical live behavior). +3. Golden byte-identity: the cutover formatter path reproduces the legacy + inline wire byte-for-byte for spotting and halt (via firms.py). 5. The unattributed_hotspot_cluster path is curated: below cluster_min_pixels it stays silent (returns None) -- F3 enabled the real broadcast path. + +chore/ripout-2dii: + - `handle_firms` (the dead Central NATS-envelope entrypoint this file used + as a driver) has been REMOVED from `meshai.env.fire_fusion` -- zero live + production callers. Driver helpers now call `ingest_hotspot_pixel` + directly (the LIVE native entrypoint -- same underlying + `_ingest_pixel_core` engine, so behavior is identical; see + `tests/test_firms_native_fusion.py` for the primary native-path coverage). + - The dead `is_cutover("wildfire_growth")` NEW-PATH in + `env.fire_fusion._handle_pass_boundary` has been removed (it stamped + hints for a formatter that was never invoked live for this category -- + wildfire_growth events are always fully precomposed, and the category is + not in `cutover.NATIVE_ALWAYS_DECIDE`). `test_growth_wire_reuses_fire_formatter` + and the whole `TestNotCutoverLegacyVerbatim` class (fully redundant with + `tests/test_firms_native_fusion.py`'s `TestIngestGrowth` / `TestIngestSpotting` + / `TestIngestHalt`, which already assert the same not-cutover stamps + + eager-latch behavior through the live `ingest_hotspot_pixel` entrypoint) + were removed accordingly. `wildfire_growth`'s formatter registration was + also removed (dead -- never reachable live); `TestRegistration` below now + asserts it resolves to `None`. """ from __future__ import annotations @@ -32,7 +50,6 @@ import uuid import pytest from meshai.notifications.formatters._budget import budget_for -from meshai.notifications.formatters.fire import format as fire_format from meshai.notifications.formatters.firms import format as firms_format from meshai.notifications.gating.firms import decide as firms_decide from tests.harness.goldens import assert_byte_identical @@ -59,16 +76,6 @@ def _isolate_db(tmp_path, monkeypatch): pdb._initialised.discard(db_path) -@pytest.fixture -def _no_cutover(monkeypatch): - """Default deploy state: nothing cut over → handler runs legacy verbatim.""" - monkeypatch.delenv("MESHAI_CUTOVER_CATEGORIES", raising=False) - from meshai.notifications.cutover import _clear_cache - _clear_cache() - yield - _clear_cache() - - def _cutover(monkeypatch, *cats): monkeypatch.setenv("MESHAI_CUTOVER_CATEGORIES", ",".join(cats)) from meshai.notifications.cutover import _clear_cache @@ -95,56 +102,39 @@ def _seed_fire(*, irwin_id, lat, lon, name="Stub Fire", **cols): conn.execute(f"INSERT INTO fires({keys}) VALUES ({ph})", tuple(base.values())) -def _envelope(*, lat, lon, acq_date="2026-06-06", acq_time="1200", - frp=20.0, satellite="N20"): - return { - "data": { - "adapter": "firms", - "category": "wildfire_hotspot", - "severity": "routine", - "data": { - "latitude": lat, "longitude": lon, "frp": frp, - "bright_ti4": 320.0, "satellite": satellite, - "instrument": "VIIRS", "confidence": "high", - "acq_date": acq_date, "acq_time": acq_time, - "daynight": "D", "version": "2.0NRT", - }, - } - } +def _pixel(*, lat, lon, acq_date="2026-06-06", acq_time="1200", + frp=20.0, satellite="N20", confidence="high", brightness=320.0): + """Build a canonical FIRMS pixel dict for the LIVE ingest_hotspot_pixel + entrypoint (mirrors tests/test_firms_native_fusion.py's `_pixel`).""" + import datetime as _dt + acq_epoch = int(_dt.datetime.strptime( + f"{acq_date} {str(acq_time).zfill(4)}", "%Y-%m-%d %H%M" + ).replace(tzinfo=_dt.timezone.utc).timestamp()) + return {"lat": lat, "lon": lon, "frp": frp, "confidence": confidence, + "brightness": brightness, "satellite": satellite, + "acq_epoch": acq_epoch} -_SUBJECT = "central.fire.hotspot.N20.high.us.id" - - -def _drive_two_pass_growth(irwin_id, center_lat, center_lon): - """Seed a fire + pass A (5 px) + first pass-B pixel 1 mi N. Returns the - (wire, data) from the boundary pixel that fires wildfire_growth.""" - from meshai.env.fire_fusion import handle_firms - _seed_fire(irwin_id=irwin_id, lat=center_lat, lon=center_lon, name="Pine Gulch") - for i in range(5): - env = _envelope(lat=center_lat + 0.0001 * i, - lon=center_lon + 0.0001 * (i - 2), - acq_date="2026-06-06", acq_time=f"12{i:02d}", - frp=20.0 + i) - handle_firms(env, subject=_SUBJECT, data={}, now=1780747200 + i) - pass_b_lat = center_lat + (1.0 / _MI_PER_DEG_LAT) - env_b = _envelope(lat=pass_b_lat, lon=center_lon, acq_time="1800", frp=22.0) - data = {} - wire = handle_firms(env_b, subject=_SUBJECT, data=data, now=1780768800) - return wire, data +def _ingest(pixel, *, now): + """Feed one pixel through the LIVE native entrypoint and return the + (wire, data) of the first broadcast it produced (or (None, {})).""" + from meshai.env.fire_fusion import ingest_hotspot_pixel + broadcasts = ingest_hotspot_pixel(pixel, now=now) + if broadcasts: + return broadcasts[0] + return None, {} def _seed_pass_a_hex_then_close(irwin_id, center_lat, center_lon, start_now=1780747200): - from meshai.env.fire_fusion import handle_firms _seed_fire(irwin_id=irwin_id, lat=center_lat, lon=center_lon, name=irwin_id) for i in range(6): angle = i * math.pi / 3 la = center_lat + (0.5 / _MI_PER_DEG_LAT) * math.sin(angle) cos_lat = math.cos(math.radians(center_lat)) lo = center_lon + (0.5 / (_MI_PER_DEG_LAT * cos_lat)) * math.cos(angle) - env = _envelope(lat=la, lon=lo, acq_time=f"12{i * 2:02d}") - handle_firms(env, subject=_SUBJECT, data={}, now=start_now + i) + _ingest(_pixel(lat=la, lon=lo, acq_time=f"12{i * 2:02d}"), + now=start_now + i) def _offset_mi(lat, lon, north_mi, east_mi): @@ -156,15 +146,11 @@ def _offset_mi(lat, lon, north_mi, east_mi): def _drive_spotting(irwin_id, center_lat, center_lon, now=1780768800): """Seed hex pass A + closed perimeter, then a pass-B pixel 2 mi NE that fires wildfire_spotting. Returns (wire, data).""" - from meshai.env.fire_fusion import handle_firms _seed_pass_a_hex_then_close(irwin_id, center_lat, center_lon) sp_lat, sp_lon = _offset_mi(center_lat, center_lon, north_mi=2.0 / math.sqrt(2), east_mi=2.0 / math.sqrt(2)) - env_b = _envelope(lat=sp_lat, lon=sp_lon, acq_time="1800") - data = {} - wire = handle_firms(env_b, subject=_SUBJECT, data=data, now=now) - return wire, data + return _ingest(_pixel(lat=sp_lat, lon=sp_lon, acq_time="1800"), now=now) def _seed_stale_fire(irwin_id, *, now_epoch, idle_hours=14, name="Cold Fire"): @@ -183,9 +169,17 @@ def _seed_stale_fire(irwin_id, *, now_epoch, idle_hours=14, name="Cold Fire"): # ───────────────────────────────────────────────────────────────────────────── class TestRegistration: - def test_growth_formatter_is_fire(self): + def test_growth_formatter_not_registered(self): + """wildfire_growth is intentionally UNREGISTERED (chore/ripout-2dii): + its wire is always precomposed by env.fire_fusion._handle_pass_boundary + -> env.fire_render._render, and the category is excluded from + cutover.NATIVE_ALWAYS_DECIDE, so compose_mesh_message's + formatter-invocation branch can never reach a registered formatter + for it live -- the precomposed-title bypass always wins first. A + registration reappearing here would be dead weight (or worse, a sign + the precomposed bypass stopped covering this category); guard it.""" from meshai.notifications.formatters import get_formatter - assert get_formatter("wildfire_growth") is fire_format + assert get_formatter("wildfire_growth") is None @pytest.mark.parametrize("cat", ["wildfire_spotting", "wildfire_halted"]) def test_spotting_halt_formatter_is_firms(self, cat): @@ -327,20 +321,6 @@ class TestHaltDecideSequence: # ───────────────────────────────────────────────────────────────────────────── class TestFormatterGolden: - def test_growth_wire_reuses_fire_formatter(self, monkeypatch): - # Drive the real growth broadcast under cutover: handle_firms returns the - # inline _render() wire AND populates data with the fire formatter hints. - _cutover(monkeypatch, "wildfire_growth") - try: - wire, data = _drive_two_pass_growth("ID-GG", 42.0, -114.0) - assert wire is not None and wire.startswith("🔥 Pine Gulch") - rendered = fire_format(_FakeEvent(data, category="wildfire_growth"), - now=0.0, budget=budget_for("wfigs")) - assert_byte_identical(rendered, wire) - finally: - from meshai.notifications.cutover import _clear_cache - _clear_cache() - def test_spotting_wire_golden(self, monkeypatch): _cutover(monkeypatch, "wildfire_spotting") try: @@ -388,49 +368,13 @@ class TestFormatterGolden: # ───────────────────────────────────────────────────────────────────────────── -# 4. Not-cutover parity — legacy eager-latch + stamps preserved VERBATIM +# 4. Not-cutover parity (legacy eager-latch + stamps) — REMOVED chore/ripout-2dii: +# fully redundant with tests/test_firms_native_fusion.py's TestIngestGrowth / +# TestIngestSpotting / TestIngestHalt, which already assert the same +# category/_severity_override/_cooldown_suffix stamps + eager-latch behavior +# through the LIVE ingest_hotspot_pixel entrypoint directly. # ───────────────────────────────────────────────────────────────────────────── -class TestNotCutoverLegacyVerbatim: - def test_growth_stamps_and_no_latch(self, _no_cutover): - wire, data = _drive_two_pass_growth("ID-GN", 42.0, -114.0) - assert wire is not None and wire.startswith("🔥 Pine Gulch") - assert "Moving N" in wire - assert data["category"] == "wildfire_growth" - assert data["_severity_override"] == "immediate" - assert data["_cooldown_suffix"] == "ID-GN" - # Legacy growth never attached a deferred commit. - assert "_on_broadcast_committed" not in data - - def test_spotting_eager_latch_stamped(self, _no_cutover): - from meshai.persistence import get_db - wire, data = _drive_spotting("ID-SN", 43.0, -115.0, now=1780768800) - assert wire is not None and "spotting" in wire - assert data["category"] == "wildfire_spotting" - assert data["_severity_override"] == "immediate" - # Legacy path stamps the latch EAGERLY with the handler `now`. - latch = get_db().execute( - "SELECT last_spotting_broadcast_at FROM fires WHERE irwin_id=?", - ("ID-SN",)).fetchone()[0] - assert latch == 1780768800.0 - assert "_on_broadcast_committed" not in data - - def test_halt_eager_latch_stamped(self, _no_cutover): - from meshai.env.fire_fusion import _maybe_emit_halt - from meshai.persistence import get_db - now = 1780768800 - _seed_stale_fire("ID-HN", now_epoch=now, idle_hours=14) - data = {} - wire = _maybe_emit_halt(get_db(), data=data, now=now) - assert wire == "🔥 Cold Fire no growth in 14h" - assert data["category"] == "wildfire_halted" - assert data["_severity_override"] == "routine" - latch = get_db().execute( - "SELECT halt_broadcast_at FROM fires WHERE irwin_id=?", - ("ID-HN",)).fetchone()[0] - assert latch == float(now) - assert "_on_broadcast_committed" not in data - # ───────────────────────────────────────────────────────────────────────────── # 5. Cluster path: below-threshold is silent (F3 enabled the path; a lone diff --git a/work/tests/test_tail_followups.py b/work/tests/test_tail_followups.py index 3d7c206..6f87d36 100644 --- a/work/tests/test_tail_followups.py +++ b/work/tests/test_tail_followups.py @@ -277,28 +277,13 @@ def test_fires_has_tombstoned_at_column(): assert "tombstoned_at" in cols -def test_wfigs_tombstone_stamps_column(): - """A tombstone envelope sets fires.tombstoned_at.""" - from meshai.env.fire_render import handle_wfigs - conn = get_db() - # Seed an active fire row. - irwin = "TOMB-1" - now = int(time.time()) - conn.execute( - "INSERT INTO fires(irwin_id, incident_name, incident_type, " - "current_acres, current_contained_pct, lat, lon, county, state, " - "declared_at, last_event_at) " - "VALUES (?,?,?,?,?,?,?,?,?,?,?)", - (irwin, "Test", "WF", 100, 10, 43.6, -116.2, "Ada", "ID", now - 3600, now), - ) - n = {"_kind": "wfigs_tombstone", "irwin_id": irwin} - envelope = {"data": {"adapter": "fires", "category": "fire.incident.removed", - "id": irwin}} - handle_wfigs(n, envelope, "central.fire.incident.removed.id", - data=None, now=now) - row = conn.execute("SELECT tombstoned_at FROM fires WHERE irwin_id=?", - (irwin,)).fetchone() - assert row["tombstoned_at"] is not None +# chore/ripout-2dii: test_wfigs_tombstone_stamps_column REMOVED. It asserted +# handle_wfigs's OWN inline `UPDATE fires SET tombstoned_at=...` write -- a +# dead-entrypoint-only side effect (handle_wfigs is gone, zero live +# production callers; the native WFIGS path -- env/fires.py -- never emits a +# `_kind=wfigs_tombstone` event, so nothing in the live system currently +# stamps tombstoned_at). The column itself remains covered by +# test_fires_has_tombstoned_at_column above. def _enable_wfigs_reminders(): diff --git a/work/tests/test_tombstone_broadcast.py b/work/tests/test_tombstone_broadcast.py index 85544b7..713a537 100644 --- a/work/tests/test_tombstone_broadcast.py +++ b/work/tests/test_tombstone_broadcast.py @@ -16,6 +16,14 @@ broadcasts back through the normal pipeline guards (Grouper, cooldown). This file's expectations were written before that downgrade and never updated -- "immediate" here would be reverting a deliberate, documented incident fix. + +chore/ripout-2dii: `handle_wfigs` (the dead Central NATS-envelope entrypoint +T1/T2 used to drive) has been REMOVED from `meshai.env.fire_render` -- zero +live production callers. T1/T2 now drive `gating.fire.decide` directly (the +LIVE, shared decider -- `_kind="wfigs_tombstone"` is its all-clear branch, +reused by the shared fire formatter). `test_commit_callback_flips_handled` +(which asserted handle_wfigs's OWN event_log-row flip on commit -- a +Central-only concept the native path never used) was deleted with it. """ from __future__ import annotations @@ -23,7 +31,7 @@ import time import pytest -from meshai.env.fire_render import handle_wfigs +from meshai.notifications.gating.fire import decide as fire_decide from meshai.notifications.env_reporter import EnvReporter from meshai.persistence import get_db @@ -49,7 +57,12 @@ def _seed_fire(conn, *, irwin_id, name, acres, contained=None, class TestTombstoneSeverityAndCommitHandles: - """T1: tombstone branch sets priority severity and attaches commit handles.""" + """T1: tombstone branch sets priority severity and attaches commit handles. + + Drives gating.fire.decide() directly (chore/ripout-2dii: handle_wfigs, the + dead entrypoint this used to wrap, is gone). decide()'s data_patch carries + the same stamps the dispatcher reads off event.data. + """ def test_severity_is_priority(self): conn = get_db() @@ -58,20 +71,14 @@ class TestTombstoneSeverityAndCommitHandles: acres=500, contained=80, last_broadcast_at=now - 3600) - data = {} - wire = handle_wfigs( - normalized={"_kind": "wfigs_tombstone", "irwin_id": "FIRE-001"}, - envelope={"data": {"category": "wildfire", "severity": "immediate"}}, - subject="wfigs.tombstone", - data=data, - now=now, - ) - assert wire is not None, "tombstone should produce wire for previously-broadcast fire" + gate = fire_decide({"_kind": "wfigs_tombstone", "irwin_id": "FIRE-001"}, + source="wfigs", now=float(now)) + assert gate.broadcast is True, "tombstone should broadcast for previously-broadcast fire" # See module docstring: fire severity was deliberately downgraded # from "immediate" to "priority" (commit 2f677e85) so fire # broadcasts flow through the normal Grouper/cooldown guards. - assert data.get("_severity_override") == "priority", ( - f"expected priority, got {data.get('_severity_override')}") + assert gate.data_patch.get("_severity_override") == "priority", ( + f"expected priority, got {gate.data_patch.get('_severity_override')}") def test_commit_handles_attached(self): conn = get_db() @@ -80,19 +87,11 @@ class TestTombstoneSeverityAndCommitHandles: acres=1000, contained=95, last_broadcast_at=now - 7200) - data = {} - wire = handle_wfigs( - normalized={"_kind": "wfigs_tombstone", "irwin_id": "FIRE-002"}, - envelope={"data": {"category": "wildfire", "severity": "immediate"}}, - subject="wfigs.tombstone", - data=data, - now=now, - ) - assert wire is not None - assert "_on_broadcast_committed" in data, "commit callback missing" - assert "_broadcast_audit" in data, "broadcast audit descriptor missing" - assert "_cooldown_suffix" in data, "cooldown suffix missing" - assert data["_cooldown_suffix"] == "FIRE-002" + gate = fire_decide({"_kind": "wfigs_tombstone", "irwin_id": "FIRE-002"}, + source="wfigs", now=float(now)) + assert gate.broadcast is True + assert callable(gate.commit), "commit callback missing" + assert gate.data_patch.get("_cooldown_suffix") == "FIRE-002" def test_dedup_suffix_is_closed(self): conn = get_db() @@ -101,61 +100,32 @@ class TestTombstoneSeverityAndCommitHandles: acres=200, contained=100, last_broadcast_at=now - 600) - data = {} - wire = handle_wfigs( - normalized={"_kind": "wfigs_tombstone", "irwin_id": "FIRE-003"}, - envelope={"data": {"category": "wildfire", "severity": "immediate"}}, - subject="wfigs.tombstone", - data=data, - now=now, - ) - assert wire is not None - assert data.get("_dedup_suffix") == "closed", ( - f"expected 'closed', got {data.get('_dedup_suffix')}") - - def test_commit_callback_flips_handled(self): - """The commit callback should flip event_log.handled to 1.""" - conn = get_db() - now = int(time.time()) - _seed_fire(conn, irwin_id="FIRE-004", name="Callback Fire", - acres=300, contained=50, - last_broadcast_at=now - 1800) - - data = {} - wire = handle_wfigs( - normalized={"_kind": "wfigs_tombstone", "irwin_id": "FIRE-004"}, - envelope={"data": {"category": "wildfire", "severity": "immediate"}}, - subject="wfigs.tombstone", - data=data, - now=now, - ) - assert wire is not None - assert "_on_broadcast_committed" in data - - # Before commit: event_log row should have handled=0 - row_before = conn.execute( - "SELECT id, handled FROM event_log WHERE event_id_external='FIRE-004' " - "ORDER BY id DESC LIMIT 1" - ).fetchone() - assert row_before is not None, "event_log row should exist" - assert row_before["handled"] == 0, "should be unhandled before commit" - - # Fire the commit callback - data["_on_broadcast_committed"](float(now)) - - # After commit: handled should be 1 - row_after = conn.execute( - "SELECT handled FROM event_log WHERE id=?", - (row_before["id"],) - ).fetchone() - assert row_after["handled"] == 1, "should be handled=1 after commit callback" + gate = fire_decide({"_kind": "wfigs_tombstone", "irwin_id": "FIRE-003"}, + source="wfigs", now=float(now)) + assert gate.broadcast is True + assert gate.data_patch.get("_dedup_suffix") == "closed", ( + f"expected 'closed', got {gate.data_patch.get('_dedup_suffix')}") class TestTombstoneAfterNewBroadcast: - """T2: closure dispatches when a New broadcast went out earlier.""" + """T2: closure dispatches when a New broadcast went out earlier. + + Drives gating.fire.decide() directly (chore/ripout-2dii: handle_wfigs is + gone). The wire itself is rendered by the shared, LIVE fire formatter + (notifications/formatters/fire.py::format) from decide()'s data_patch -- + the same contract the native WFIGS path uses. + """ def test_closure_wire_after_prior_broadcast(self): """Fire that was broadcast 10 min ago gets a closure wire on tombstone.""" + from meshai.notifications.formatters.fire import format as fire_format + from meshai.notifications.formatters._budget import budget_for + + class _FakeEvent: + def __init__(self, data, category=None): + self.data = data + self.category = category + conn = get_db() now = int(time.time()) _seed_fire(conn, irwin_id="IA-1", name="IA 1", @@ -163,22 +133,19 @@ class TestTombstoneAfterNewBroadcast: last_broadcast_at=now - 600, # 10 min ago last_event_at=now - 600) - data = {} - wire = handle_wfigs( - normalized={"_kind": "wfigs_tombstone", "irwin_id": "IA-1"}, - envelope={"data": {"category": "wildfire", "severity": "immediate"}}, - subject="wfigs.tombstone", - data=data, - now=now, - ) - assert wire is not None, "tombstone should produce wire" - assert "✅" in wire, "closure wire should contain checkmark" - assert "IA 1" in wire, "closure wire should name the fire" - assert data["category"] == "wildfire_closed" + gate = fire_decide({"_kind": "wfigs_tombstone", "irwin_id": "IA-1"}, + source="wfigs", now=float(now)) + assert gate.broadcast is True, "tombstone should broadcast" + assert gate.data_patch["category"] == "wildfire_closed" # See module docstring: downgraded from "immediate" to "priority" # by commit 2f677e85 to prevent Grouper/cooldown bypass. - assert data["_severity_override"] == "priority" - assert callable(data.get("_on_broadcast_committed")) + assert gate.data_patch["_severity_override"] == "priority" + assert callable(gate.commit) + + wire = fire_format(_FakeEvent(gate.data_patch, category="wildfire_closed"), + now=0.0, budget=budget_for("wfigs")) + assert "✅" in wire, "closure wire should contain checkmark" + assert "IA 1" in wire, "closure wire should name the fire" def test_no_wire_when_never_broadcast(self): """Fire that was never broadcast should NOT get a closure wire.""" @@ -189,36 +156,9 @@ class TestTombstoneAfterNewBroadcast: last_broadcast_at=None, # never broadcast last_event_at=now - 600) - data = {} - wire = handle_wfigs( - normalized={"_kind": "wfigs_tombstone", "irwin_id": "SILENT-1"}, - envelope={"data": {"category": "wildfire", "severity": "immediate"}}, - subject="wfigs.tombstone", - data=data, - now=now, - ) - assert wire is None, "no closure wire for never-broadcast fire" - - def test_tombstoned_at_stamped(self): - """Tombstone should stamp tombstoned_at on the fires row.""" - conn = get_db() - now = int(time.time()) - _seed_fire(conn, irwin_id="STAMP-1", name="Stamp Fire", - acres=10, contained=50, - last_broadcast_at=now - 3600) - - handle_wfigs( - normalized={"_kind": "wfigs_tombstone", "irwin_id": "STAMP-1"}, - envelope={"data": {"category": "wildfire", "severity": "immediate"}}, - subject="wfigs.tombstone", - data={}, - now=now, - ) - row = conn.execute( - "SELECT tombstoned_at FROM fires WHERE irwin_id='STAMP-1'" - ).fetchone() - assert row is not None - assert row["tombstoned_at"] == now + gate = fire_decide({"_kind": "wfigs_tombstone", "irwin_id": "SILENT-1"}, + source="wfigs", now=float(now)) + assert gate.broadcast is False, "no closure wire for never-broadcast fire" class TestEnvSummaryExcludesContainedTombstoned: diff --git a/work/tests/test_wfigs_handler.py b/work/tests/test_wfigs_handler.py index 4e3b4f6..c81728b 100644 --- a/work/tests/test_wfigs_handler.py +++ b/work/tests/test_wfigs_handler.py @@ -1,41 +1,41 @@ -"""Tests for env/fire_render.py's handle_wfigs() -- WFIGS persistence wire-up. +"""Tests for env/fire_render.py's shared WFIGS wire-render + anchor helpers. -Covers: - (a) parse clean active-incident envelope (all fields populated) - (b) acres fallback chain: top-null -> raw.DiscoveryAcres used - (c) acres absent at every level -> renders "N/A" - (d) IncidentName="IA 1" placeholder passes through verbatim - (e) tombstone subject -> handler returns None + event_log row handled=0 - (f) perimeter subject -> handler returns None + event_log row handled=0 - (g) NEW IRWIN -> "New:" prefix + fires INSERT + mesh_broadcasts_out audit row - (h) known IRWIN no change -> drop silently, last_broadcast_* unchanged - (i) known IRWIN acres up but <8h elapsed -> drop, last_broadcast_* unchanged - (j) known IRWIN acres up + >=8h elapsed -> "Update:" prefix + audit row +chore/ripout-2dii: `handle_wfigs` (the dead Central NATS-envelope entrypoint) +has been REMOVED from `meshai.env.fire_render` -- it had zero live production +callers (Central's consumer that drove it is gone). Tests that ONLY exercised +`handle_wfigs`'s own envelope-parsing / New-Update-cooldown decision / audit-row +contract (with no live equivalent -- that decision now lives in the LIVE +`gating.fire.decide`, already covered end-to-end via the real native adapter in +`tests/test_fire_native_growth.py`) were deleted alongside it. + +What remains here: (k) location anchor priority: geocoder.city > nearest_town > landclass > county + -- `_location_anchor` is shared, LIVE code (used by `_render`, which is + called directly on the FIRMS `wildfire_growth` path). Rewritten to call + `_render` directly instead of routing through the dead handler. + - "size unknown" / "containment unknown" missing-acres rendering -- same + live-`_render` rationale, rewritten to call `_render` directly. + - budget-fit worst case / date-only discovery -- already called `_render` + directly (never used handle_wfigs); unchanged. -handle_wfigs (env/fire_render.py) has no live production caller -- Central's -NATS consumer that drove it is gone -- but it remains the parity-tested -legacy contract for the WFIGS wire format (see that module's docstring), and -`_location_anchor`, which it shares with the LIVE `_render` (used on the -FIRMS fire-growth path), is real regression-tested surface here. It expects -its input pre-shaped into a flat "normalized" dict; that shaping used to be -done by the WFIGS dispatch branch of the deleted Central-envelope adapter- -normalizer module's `normalize()` + its `_parse_wfigs_incidents` helper. -`_normalize_wfigs()` below is a verbatim-logic local replica of that -dispatch, scoped to the wfigs_incidents/wfigs_perimeters envelope shapes the -fixtures in this file build -- it exists so this test file has no dependency -on that (now fully removed) module. +The envelope builders (`_make_active_envelope`, `_make_tombstone`, +`_make_perimeter`, `_normalize_wfigs`) are KEPT: they are shared fixture +infrastructure imported by other test files (test_fire_refactor.py, +test_fire_tracker_phase1.py, test_fire_age_gate.py) to build canonical dicts +for the LIVE `gating.fire.decide` / `notifications.formatters.fire.format`. +`_normalize_wfigs` is a verbatim-logic local replica of the WFIGS dispatch +branch of the deleted Central-envelope adapter-normalizer module's +`normalize()` + `_parse_wfigs_incidents` helper (that module is already gone +from production; this replica exists purely so fixture-building has no +dependency on it). """ -import os import time from typing import Any, Optional import pytest from meshai.env.fire_render import ( - WFIGS_BROADCAST_COOLDOWN_S, - handle_wfigs, _render as _wfigs_render, ) from meshai.persistence import close_thread_connection, init_db @@ -302,275 +302,44 @@ def _make_perimeter(irwin_id=_IRWIN_A, state="ID", county="Cassia", # ============================================================================ -# (a) parse a clean active-incident envelope with all fields +# Missing-acres rendering -- LIVE `_render` behavior, called directly +# (formerly driven through the dead handle_wfigs). # ============================================================================ -def test_a_parse_clean_active_envelope(mem_db, no_photon): - env = _make_active_envelope() - n = _normalize_wfigs(env) - assert n is not None - assert n["_kind"] == "wfigs_incident" - assert n["irwin_id"] == _IRWIN_A - assert n["incident_name"] == "Cache Peak Fire" - assert n["incident_type"] == "wildfire" - assert n["acres"] == 1847.0 - assert n["contained_pct"] == 23 - assert n["county"] == "Cassia" - assert n["state"] == "ID" - # FireDiscoveryDateTime epoch-ms -> epoch-s conversion - assert n["declared_at_epoch"] == 1780529163 -# ============================================================================ -# (b) null top-level acres -> raw.DiscoveryAcres fallback used -# ============================================================================ -def test_b_acres_fallback_to_raw_discovery_acres(mem_db, no_photon): - env = _make_active_envelope(daily_acres=None, pct_contained=None, - raw_discovery_acres=0.1, - raw_pct_contained=0) - n = _normalize_wfigs(env) - assert n["acres"] == 0.1 - assert n["contained_pct"] == 0 - - -# ============================================================================ -# (c) no acres anywhere -> renders "N/A" -# ============================================================================ -def test_c_acres_missing_renders_na(mem_db, no_photon): +def test_acres_missing_renders_na(mem_db, no_photon): env = _make_active_envelope(name="IA 7", daily_acres=None, pct_contained=None, irwin_id=_IRWIN_C, landclass="Sawtooth National Forest") - wire = handle_wfigs(_normalize_wfigs(env), env, env["subject"], now=1_000_000) + n = _normalize_wfigs(env) + wire = _wfigs_render(n, prefix="New") assert wire is not None assert "size unknown" in wire assert "containment unknown" in wire -# ============================================================================ -# (d) "IA 1" placeholder name passes through verbatim -# ============================================================================ -def test_d_ia_placeholder_passthrough(mem_db, no_photon): +def test_ia_placeholder_passthrough(mem_db, no_photon): env = _make_active_envelope(name="IA 1", county="Elmore", daily_acres=None, pct_contained=None, landclass="Sawtooth National Forest", irwin_id=_IRWIN_B) - wire = handle_wfigs(_normalize_wfigs(env), env, env["subject"], now=1_000_000) + n = _normalize_wfigs(env) + wire = _wfigs_render(n, prefix="New") assert wire is not None assert "IA 1" in wire # ============================================================================ -# (e) tombstone subject -> handler returns None + event_log handled=0 -# ============================================================================ -def test_e_tombstone_returns_none_and_logs(mem_db, no_photon): - env = _make_tombstone() - n = _normalize_wfigs(env) - assert n["_kind"] == "wfigs_tombstone" - out = handle_wfigs(n, env, env["subject"], now=2_000_000) - assert out is None - row = mem_db.execute( - "SELECT source, category, handled, table_name, table_pk, nats_subject " - "FROM event_log WHERE event_id_external=?", (_IRWIN_A,)).fetchone() - assert row is not None - assert row["source"] == "wfigs_incidents" - assert row["category"] == "fire.incident.removed" - assert row["handled"] == 0 - assert row["table_name"] is None - assert row["table_pk"] == _IRWIN_A - assert row["nats_subject"] == "central.fire.incident.removed.id" - # No row in fires. - n_fires = mem_db.execute("SELECT COUNT(*) AS n FROM fires").fetchone()["n"] - assert n_fires == 0 - - -# ============================================================================ -# (f) perimeter subject -> same as tombstone -# ============================================================================ -def test_f_perimeter_returns_none_and_logs(mem_db, no_photon): - env = _make_perimeter() - n = _normalize_wfigs(env) - assert n["_kind"] == "wfigs_perimeter" - out = handle_wfigs(n, env, env["subject"], now=3_000_000) - assert out is None - row = mem_db.execute( - "SELECT source, handled FROM event_log WHERE event_id_external=?", - (_IRWIN_A,)).fetchone() - assert row is not None - assert row["source"] == "wfigs_perimeters" - assert row["handled"] == 0 - n_fires = mem_db.execute("SELECT COUNT(*) AS n FROM fires").fetchone()["n"] - assert n_fires == 0 - - -# ============================================================================ -# (g) NEW IRWIN -> "New:" prefix + fires INSERT + mesh_broadcasts_out audit -# ============================================================================ -def test_g_new_irwin_inserts_and_broadcasts(mem_db, no_photon): - env = _make_active_envelope(geocoder_city="Burley") # avoids Photon path - now = 5_000_000 - data = {} - wire = handle_wfigs(_normalize_wfigs(env), env, env["subject"], - data=data, now=now) - assert wire is not None - assert wire.startswith("🔥 Cache Peak Fire — New") - assert "Burley" in wire - assert "1,847 ac" in wire - assert "containment 23%" in wire - # Budget-fit rework: no unique-fire-id line, no bold markdown. - assert "ID:" not in wire - assert "**" not in wire - - # v0.5.8b: handler INSERTs the fires row with last_broadcast_*=NULL, - # then attaches a commit callback. The dispatcher fires the callback - # on successful broadcast; we simulate that here. - fr_pre = mem_db.execute( - "SELECT last_broadcast_at FROM fires WHERE irwin_id=?", - (_IRWIN_A,)).fetchone() - assert fr_pre["last_broadcast_at"] is None - data["_on_broadcast_committed"](float(now)) - fr = mem_db.execute( - "SELECT current_acres, last_broadcast_at, last_broadcast_acres, " - "last_broadcast_contained, last_event_at " - "FROM fires WHERE irwin_id=?", (_IRWIN_A,)).fetchone() - assert fr is not None - assert fr["current_acres"] == 1847.0 - assert fr["last_broadcast_at"] == now - assert fr["last_broadcast_acres"] == 1847.0 - assert fr["last_broadcast_contained"] == 23 - assert fr["last_event_at"] == now - - # event_log row logged with handled=1. - el = mem_db.execute( - "SELECT handled, table_name, table_pk FROM event_log " - "WHERE event_id_external=?", (_IRWIN_A,)).fetchone() - assert el["handled"] == 1 - assert el["table_name"] == "fires" - assert el["table_pk"] == _IRWIN_A - - # v0.5.8b: mesh_broadcasts_out is inserted by the dispatcher - # (test_cold_start_grace covers that path). The handler only signals - # via data["_broadcast_audit"] that an audit row is wanted. - assert data["_broadcast_audit"] == {"table": "fires", "pk": _IRWIN_A} - - -# ============================================================================ -# (h) known IRWIN no-change -> drop silently, last_broadcast_* unchanged -# ============================================================================ -def test_h_known_irwin_no_change_drops(mem_db, no_photon): - # Use wall-clock-adjacent timestamps so _cleanup_stale_fires doesn't - # delete the row (it uses real time.time() with a 7d cutoff internally). - # Anchor the fire's discovery date 2d before that `now` so it stays - # inside the 14d fire age-gate -- otherwise the static fixture date - # (2026-06-03) is now stale vs wall-clock and the first-sight "New" - # broadcast this test depends on would be (correctly) suppressed. - first_now = int(time.time()) - env = _make_active_envelope(geocoder_city="Burley", - fire_discovery_dt_ms=(first_now - 2 * 86400) * 1000) - data0 = {} - handle_wfigs(_normalize_wfigs(env), env, env["subject"], - data=data0, now=first_now) - # v0.5.8b: dispatcher commit closes the broadcast. - data0["_on_broadcast_committed"](float(first_now)) - - # Re-publish the same incident exactly 30 min later: same acres + contained. - later = first_now + 1800 - out = handle_wfigs(_normalize_wfigs(env), env, env["subject"], now=later) - assert out is None - - fr = mem_db.execute( - "SELECT last_broadcast_at, last_broadcast_acres, last_broadcast_contained, " - "last_event_at FROM fires WHERE irwin_id=?", (_IRWIN_A,)).fetchone() - # last_broadcast_* unchanged from the original. - assert fr["last_broadcast_at"] == first_now - assert fr["last_broadcast_acres"] == 1847.0 - assert fr["last_broadcast_contained"] == 23 - # last_event_at was refreshed. - assert fr["last_event_at"] == later - - # v0.5.8b: mesh_broadcasts_out is inserted by the dispatcher, not the - # handler -- this test never invokes a real dispatcher, so count is 0. - cnt = mem_db.execute( - "SELECT COUNT(*) AS n FROM mesh_broadcasts_out WHERE source_event_pk=?", - (_IRWIN_A,)).fetchone()["n"] - assert cnt == 0 - - -# ============================================================================ -# (i) known IRWIN acres up but <8h elapsed -> drop, last_broadcast_* unchanged -# ============================================================================ -def test_i_known_irwin_change_inside_cooldown_drops(mem_db, no_photon): - # Wall-clock `now` so _cleanup_stale_fires (real time.time(), 7d cutoff) - # keeps the row; anchor discovery 2d earlier so the fire stays inside - # the 14d fire age-gate (the static fixture date is now stale vs - # wall-clock and would suppress the first-sight "New" broadcast). - _base = int(time.time()) - env_initial = _make_active_envelope( - geocoder_city="Burley", - fire_discovery_dt_ms=(_base - 2 * 86400) * 1000) - data0 = {} - handle_wfigs(_normalize_wfigs(env_initial), env_initial, - env_initial["subject"], data=data0, now=_base) - data0["_on_broadcast_committed"](float(_base)) - - # Bigger fire, but only 4h later -- inside cooldown. Same discovery - # date as the initial envelope (same fire, still inside the age-gate). - env_grown = _make_active_envelope( - geocoder_city="Burley", daily_acres=3000.0, pct_contained=23, - fire_discovery_dt_ms=(_base - 2 * 86400) * 1000) - later = _base + 4 * 3600 - out = handle_wfigs(_normalize_wfigs(env_grown), env_grown, - env_grown["subject"], now=later) - assert out is None - - fr = mem_db.execute( - "SELECT last_broadcast_at, last_broadcast_acres, last_broadcast_contained, " - "current_acres FROM fires WHERE irwin_id=?", (_IRWIN_A,)).fetchone() - assert fr["last_broadcast_at"] == _base - assert fr["last_broadcast_acres"] == 1847.0 - assert fr["last_broadcast_contained"] == 23 - # current_acres was refreshed to the new value. - assert fr["current_acres"] == 3000.0 - - -# ============================================================================ -# (j) known IRWIN acres up AND >=8h elapsed -> "Update:" + audit row -# ============================================================================ -def test_j_known_irwin_change_after_cooldown_broadcasts(mem_db, no_photon): - env_initial = _make_active_envelope(geocoder_city="Burley") - data_j0 = {} - handle_wfigs(_normalize_wfigs(env_initial), env_initial, - env_initial["subject"], data=data_j0, now=5_000_000) - data_j0["_on_broadcast_committed"](float(5_000_000)) - - env_grown = _make_active_envelope(geocoder_city="Burley", - daily_acres=3000.0, pct_contained=35) - later = 5_000_000 + WFIGS_BROADCAST_COOLDOWN_S - data2 = {} - out = handle_wfigs(_normalize_wfigs(env_grown), env_grown, - env_grown["subject"], data=data2, now=later) - assert out is not None - assert out.startswith("🔥 Cache Peak Fire — Update") - assert "3,000 ac" in out - assert "containment 35%" in out - - # Simulate dispatcher commit. - data2["_on_broadcast_committed"](float(later)) - fr = mem_db.execute( - "SELECT last_broadcast_at, last_broadcast_acres, last_broadcast_contained " - "FROM fires WHERE irwin_id=?", (_IRWIN_A,)).fetchone() - assert fr["last_broadcast_at"] == later - assert fr["last_broadcast_acres"] == 3000.0 - assert fr["last_broadcast_contained"] == 35 - - -# ============================================================================ -# (k) location anchor priority -- city > nearest_town > landclass > county +# location anchor priority -- city > nearest_town > landclass > county +# `_location_anchor` is shared LIVE code (used by `_render`). # ============================================================================ def test_k_anchor_geocoder_city_wins(mem_db, no_photon): env = _make_active_envelope(geocoder_city="Twin Falls", landclass="Sawtooth NF", county="Cassia") - wire = handle_wfigs(_normalize_wfigs(env), env, env["subject"], now=1) + n = _normalize_wfigs(env) + wire = _wfigs_render(n, prefix="New") assert "Twin Falls" in wire assert "Sawtooth NF" not in wire assert "Cassia Co" not in wire @@ -586,8 +355,9 @@ def test_k_anchor_falls_to_nearest_town(monkeypatch, mem_db): env = _make_active_envelope(geocoder_city=None, landclass="Sawtooth NF", county="Cassia") - wire = handle_wfigs(_normalize_wfigs(env), env, env["subject"], now=1) - # Handler now resolves anchor via town_anchors table (Burley @ 42.536, -113.793) + n = _normalize_wfigs(env) + wire = _wfigs_render(n, prefix="New") + # Resolves anchor via town_anchors table (Burley @ 42.536, -113.793) assert "Burley" in wire @@ -599,8 +369,9 @@ def test_k_anchor_falls_to_landclass(monkeypatch, mem_db): env = _make_active_envelope(geocoder_city=None, landclass="Sawtooth National Forest", county="Cassia") - wire = handle_wfigs(_normalize_wfigs(env), env, env["subject"], now=1) - # Handler resolves nearest town from town_anchors table, overriding landclass + n = _normalize_wfigs(env) + wire = _wfigs_render(n, prefix="New") + # Resolves nearest town from town_anchors table, overriding landclass assert "Burley" in wire @@ -611,8 +382,9 @@ def test_k_anchor_falls_to_county(monkeypatch, mem_db): ) env = _make_active_envelope(geocoder_city=None, landclass=None, county="Cassia", state="ID") - wire = handle_wfigs(_normalize_wfigs(env), env, env["subject"], now=1) - # Handler resolves nearest town from town_anchors table + n = _normalize_wfigs(env) + wire = _wfigs_render(n, prefix="New") + # Resolves nearest town from town_anchors table assert "Burley" in wire @@ -623,138 +395,16 @@ def test_k_anchor_nearest_town_under_one_mile_says_near(monkeypatch, mem_db): lambda lat, lon, max_distance_mi=50.0: fake, ) env = _make_active_envelope(geocoder_city=None) - wire = handle_wfigs(_normalize_wfigs(env), env, env["subject"], now=1) - # Handler resolves anchor via town_anchors; exact format depends on distance - assert "Burley" in wire - - -# ============================================================================ -# v0.5.8b refactor -- New:/Update: prefix survives cold-start drops -# ============================================================================ - - -def _run_handler_only(env, data=None, now=None): - """Run normalize + handler WITHOUT invoking any commit callback. - Simulates the dispatcher dropping the broadcast (grace/cooldown/stale) - after the handler has already written persistence rows.""" n = _normalize_wfigs(env) - if data is None: - data = {} - wire = handle_wfigs(n, env, env["subject"], data=data, now=now) - return wire, data - - -def _commit(data, committed_at): - """Simulate the dispatcher invoking the handler's post-commit callback.""" - cb = data.get("_on_broadcast_committed") - assert callable(cb), "handler must attach _on_broadcast_committed" - cb(committed_at) - - -def test_e_cold_start_then_resume_still_new(mem_db, no_photon): - """Cold-start drop scenario: first pass writes fires + event_log but - dispatcher drops the broadcast (we skip the callback). Second pass on - the SAME IRWIN must still produce "New:" because last_broadcast_at is - still NULL -- it really is the first delivery for that fire. - """ - env = _make_active_envelope(geocoder_city="Burley") - - # Pass 1: handler runs, but the dispatcher drops the broadcast (we - # mimic that by not calling the commit callback). - wire1, data1 = _run_handler_only(env, now=10_000) - assert wire1.startswith("🔥 Cache Peak Fire — New") - fr = mem_db.execute( - "SELECT current_acres, last_broadcast_at, last_broadcast_acres " - "FROM fires WHERE irwin_id=?", (_IRWIN_A,)).fetchone() - assert fr is not None - assert fr["current_acres"] == 1847.0 - assert fr["last_broadcast_at"] is None - assert fr["last_broadcast_acres"] is None - - # Pass 2: same envelope 5 minutes later (still pre-broadcast). - wire2, data2 = _run_handler_only(env, now=10_300) - assert wire2.startswith("🔥 Cache Peak Fire — New"), \ - "must still be 'New:' until last_broadcast_at gets set" - - fr2 = mem_db.execute( - "SELECT current_acres, last_broadcast_at, last_event_at " - "FROM fires WHERE irwin_id=?", (_IRWIN_A,)).fetchone() - # last_event_at advanced; last_broadcast_at still NULL. - assert fr2["last_event_at"] == 10_300 - assert fr2["last_broadcast_at"] is None - - -def test_f_commit_callback_updates_last_broadcast(mem_db, no_photon): - """After the dispatcher calls the callback, last_broadcast_* reflect - the committed timestamp + the acres/containment of THIS broadcast.""" - env = _make_active_envelope(geocoder_city="Burley") - wire, data = _run_handler_only(env, now=20_000) - assert wire is not None - - _commit(data, committed_at=20_005.0) - - fr = mem_db.execute( - "SELECT last_broadcast_at, last_broadcast_acres, last_broadcast_contained " - "FROM fires WHERE irwin_id=?", (_IRWIN_A,)).fetchone() - assert fr["last_broadcast_at"] == 20_005 - assert fr["last_broadcast_acres"] == 1847.0 - assert fr["last_broadcast_contained"] == 23 - - # Third pass: same IRWIN, no growth, no callback (cooldown applies). - # Handler must return None this time because last_broadcast_at IS NOT NULL - # and the change-detection gates report no change. - env_same = _make_active_envelope(geocoder_city="Burley") - wire3, _ = _run_handler_only(env_same, now=20_010) - assert wire3 is None - - -def test_g_callback_not_called_means_last_broadcast_stays_null(mem_db, no_photon): - """If dispatcher drops for any reason (grace, staleness, cooldown, - dedup) the callback is not invoked -- last_broadcast_* stays NULL and - the next successful broadcast emits 'New:' (not 'Update:'). This is - the inverse of test_e from the persistence-row side.""" - env = _make_active_envelope(geocoder_city="Burley") - wire, data = _run_handler_only(env, now=30_000) - assert wire is not None - # No _commit() call. - fr = mem_db.execute( - "SELECT last_broadcast_at FROM fires WHERE irwin_id=?", - (_IRWIN_A,)).fetchone() - assert fr["last_broadcast_at"] is None - - -def test_h_no_audit_row_inserted_when_handler_skips_commit(mem_db, no_photon): - """The handler no longer writes mesh_broadcasts_out -- the dispatcher - inserts it via `_broadcast_audit`. Until the dispatcher calls _commit, - there should be zero rows in mesh_broadcasts_out even though fires - has the new row.""" - env = _make_active_envelope(geocoder_city="Burley") - wire, data = _run_handler_only(env, now=40_000) - assert wire is not None - n = mem_db.execute( - "SELECT COUNT(*) AS n FROM mesh_broadcasts_out").fetchone()["n"] - assert n == 0 - # The handler signalled the dispatcher SHOULD insert an audit row. - audit = data["_broadcast_audit"] - assert audit == {"table": "fires", "pk": _IRWIN_A} - - -def test_h_handler_attaches_audit_descriptor_and_callback(mem_db, no_photon): - """Sanity: every active wire-string return must come with the two - dispatcher hooks attached.""" - env = _make_active_envelope(geocoder_city="Burley", irwin_id=_IRWIN_B) - data = {} - wire = handle_wfigs(_normalize_wfigs(env), env, env["subject"], - data=data, now=50_000) - assert wire is not None - assert callable(data["_on_broadcast_committed"]) - assert data["_broadcast_audit"]["table"] == "fires" - assert data["_broadcast_audit"]["pk"] == _IRWIN_B + wire = _wfigs_render(n, prefix="New") + # Anchor resolved via town_anchors; exact format depends on distance + assert "Burley" in wire # ============================================================================ # Budget-fit worst case: longest plausible fire payload fits 140 chars, with # no `ID:` line, no `**` markdown, and discovery rendered DATE-ONLY. +# (LIVE `_render`, called directly -- never used handle_wfigs.) # ============================================================================ @@ -798,17 +448,3 @@ def test_wfigs_discovery_is_date_only(): # no time-of-day (colon in an H:MM would appear as ":3" etc.) assert "2:30" not in wire and "PM" not in wire and "AM" not in wire assert "ID:" not in wire - - -# ============================================================================ -# A new-fire envelope produces a per-fire wfigs broadcast. -# ============================================================================ - - -def test_per_fire_wfigs_broadcasts_new_fire(mem_db, no_photon): - env = _make_active_envelope(geocoder_city="Burley") - data = {} - wire = handle_wfigs(_normalize_wfigs(env), env, env["subject"], - data=data, now=6_000_000) - assert wire is not None - assert wire.startswith("🔥 Cache Peak Fire — New")