From c7a2cd626a7150f16b4fdb2ee5cb9282f1359dc3 Mon Sep 17 00:00:00 2001 From: Matt Johnson Date: Sun, 5 Jul 2026 05:18:27 +0000 Subject: [PATCH] refactor(phase3c): migrate FIRMS fusion broadcasts to formatter+decider MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Migrate the three FIRMS fire-tracker BROADCAST paths (wildfire_growth, wildfire_spotting, wildfire_halted) behind the registry, NO cutover. Attribution/pass/centroid/perimeter plumbing stays inline; legacy live path is byte-identical. - gating/firms.py decide() discriminates on handler-stamped _kind (firms_growth/firms_spotting/firms_halt) - wildfire_growth reuses formatters/fire.py (verified byte-identical: growth SELECT uses current_* columns _render doesn't read, so the wire is the movement+anchor line with "size/containment unknown" โ€” a latent legacy quirk, reproduced exactly, NOT fixed) - formatters/firms.py renders spotting + halt wires - tier-b (flagged): the eager latch writes (fires.last_spotting_broadcast_at, fires.halt_broadcast_at) move into deferred commit closures โ€” a dropped broadcast no longer burns the latch. Validated by gate-sequence, not golden bytes. Not-cutover live path keeps eager latches verbatim. - FIRMS broadcasts never touch event_log (eager handled=1 at pixel storage) โ†’ no event_log flip to wrap - _maybe_emit_cluster stays dead (unconditional return None) + test Deferred follow-ups (unchanged): env/firms.py native hotspot broadcast neutralization; native canonical emission. 27 new tests; fire-tracker + firms handler suites preserved; full suite at 34-failure baseline (1597 passed). Co-Authored-By: Claude Opus 4.8 (1M context) --- work/meshai/central/firms_handler.py | 173 ++++--- .../notifications/formatters/__init__.py | 17 + work/meshai/notifications/formatters/firms.py | 53 +++ work/meshai/notifications/gating/__init__.py | 15 + work/meshai/notifications/gating/firms.py | 241 ++++++++++ work/tests/test_fire_refactor.py | 10 +- work/tests/test_firms_refactor.py | 443 ++++++++++++++++++ 7 files changed, 895 insertions(+), 57 deletions(-) create mode 100644 work/meshai/notifications/formatters/firms.py create mode 100644 work/meshai/notifications/gating/firms.py create mode 100644 work/tests/test_firms_refactor.py diff --git a/work/meshai/central/firms_handler.py b/work/meshai/central/firms_handler.py index 6f29d3f..be2eb9d 100644 --- a/work/meshai/central/firms_handler.py +++ b/work/meshai/central/firms_handler.py @@ -714,13 +714,25 @@ def _handle_pass_boundary(conn, *, irwin_id, pass_id, lat, lon, if spotting_wire is not None: return spotting_wire - if last_pass_id == pass_id or last_pass_id is None or prev is None: - # No boundary (same pass), or this is the fire's first pass -- - # nothing to compare drift against. - return None + # Phase-3c: the growth broadcast DECISION (pass boundary + drift threshold) + # is delegated to gating.firms.decide; the wire is still rendered inline via + # the WFIGS _render for byte-identity. The boundary predicate reproduces the + # legacy guard (`last_pass_id == pass_id or last_pass_id is None or prev is + # None` -> suppress) exactly. Growth has no latch, so there is nothing to + # defer -- only the data stamping differs between the cutover and legacy + # paths. + from meshai.notifications.cutover import is_cutover + from meshai.notifications.gating.firms import decide as _firms_decide - threshold = float(adapter_config.fires.growth_drift_threshold_mi) - if drift_mi is None or drift_mi < threshold: + boundary_growth = ((last_pass_id != pass_id) and (last_pass_id is not None) + and (prev is not None)) + gate = _firms_decide( + {"_kind": "firms_growth", "irwin_id": irwin_id, + "boundary": boundary_growth, "drift_mi": drift_mi, + "drift_direction": drift_direction, + "drift_mi_per_hour": drift_mi_per_hour}, + source="firms", now=float(now)) + if not gate.broadcast: return None # Drift exceeded the threshold -- emit wildfire_growth via WFIGS renderer. @@ -737,9 +749,25 @@ def _handle_pass_boundary(conn, *, irwin_id, pass_id, lat, lon, normalized = dict(fire) normalized["irwin_id"] = irwin_id if isinstance(data, dict): - data["category"] = "wildfire_growth" - data["_severity_override"] = "immediate" - data["_cooldown_suffix"] = irwin_id + if is_cutover("wildfire_growth"): + # NEW PATH: fire formatter re-renders from these hints. Feed ONLY the + # fields _render actually reads on the growth path (incident_name, + # lat/lon for the anchor, movement, is_update) so the wire matches. + data.update(gate.data_patch) + data["incident_name"] = fire["incident_name"] + data["lat"] = fire["lat"] + data["lon"] = fire["lon"] + data["_broadcast_audit"] = {"table": "fires", "pk": irwin_id} + _raw_commit = gate.commit + if _raw_commit is not None: + def _on_commit(committed_at: float) -> None: + _raw_commit(committed_at) + data["_on_broadcast_committed"] = _on_commit + else: + # LEGACY verbatim (byte-for-byte identical live output). + data["category"] = "wildfire_growth" + data["_severity_override"] = "immediate" + data["_cooldown_suffix"] = irwin_id return _render(normalized, prefix="Update", movement=movement) @@ -769,35 +797,51 @@ def _maybe_emit_halt(conn, *, data, now): """Find one fire matching the halt criteria, latch + broadcast. Returns the wire string when a halt event fires; otherwise None. - Latching is via fires.halt_broadcast_at -- a fire that came back - to life (last_pass_at updated to a fresher value) becomes re- - eligible because we filter on `halt_broadcast_at IS NULL OR - halt_broadcast_at < last_pass_at`. + + Phase-3c: the DECISION (the halt-eligibility SELECT) is delegated to + ``gating.firms.decide``; the wire is still rendered inline here for + byte-identity. On the not-cutover live path the eager + ``fires.halt_broadcast_at`` latch is kept VERBATIM (stamped with the handler + ``now``) so the live broadcast stays byte-for-byte identical; on the cutover + path the latch is deferred into ``gate.commit`` (tier-b change). Halt + re-eligibility is unchanged: a fire whose ``last_pass_at`` advances past + ``halt_broadcast_at`` becomes eligible again. """ - minimum_s = int(adapter_config.fires.halt_minimum_seconds) - cutoff = float(now) - float(minimum_s) - row = conn.execute( - "SELECT irwin_id, incident_name, last_pass_at FROM fires " - "WHERE tombstoned_at IS NULL " - "AND last_pass_at IS NOT NULL AND last_pass_at <= ? " - "AND (halt_broadcast_at IS NULL " - " OR halt_broadcast_at < last_pass_at) " - "ORDER BY last_pass_at ASC LIMIT 1", - (cutoff,), - ).fetchone() - if row is None: + from meshai.notifications.cutover import is_cutover + from meshai.notifications.gating.firms import decide as _firms_decide + + gate = _firms_decide({"_kind": "firms_halt"}, source="firms", now=float(now)) + if not gate.broadcast: return None - conn.execute( - "UPDATE fires SET halt_broadcast_at=? WHERE irwin_id=?", - (float(now), row["irwin_id"]), - ) + p = gate.data_patch + irwin_id = p["irwin_id"] + name = p["incident_name"] + hours = p["hours"] + wire = f"๐Ÿ”ฅ {name} no growth in {hours}h" + if isinstance(data, dict): - data["category"] = "wildfire_halted" - data["severity"] = "routine" - hours = max(0, int((float(now) - float(row["last_pass_at"])) / 3600.0)) - name = row["incident_name"] or "(unnamed fire)" - return f"๐Ÿ”ฅ {name} no growth in {hours}h" + if is_cutover("wildfire_halted"): + # NEW PATH: formatter re-renders from data_patch; latch deferred. + data.update(gate.data_patch) + data["_broadcast_audit"] = {"table": "fires", "pk": irwin_id} + _raw_commit = gate.commit + + def _on_commit(committed_at: float) -> None: + if _raw_commit is not None: + _raw_commit(committed_at) + + data["_on_broadcast_committed"] = _on_commit + else: + # LEGACY verbatim (byte-for-byte identical live output): eager latch + # with the handler `now`, plain category/severity stamps. + conn.execute( + "UPDATE fires SET halt_broadcast_at=? WHERE irwin_id=?", + (float(now), irwin_id), + ) + data["category"] = "wildfire_halted" + data["severity"] = "routine" + return wire def _bearing(lat1: float, lon1: float, @@ -878,7 +922,8 @@ def _check_spotting(conn, *, irwin_id, pixel_lat, pixel_lon, current_pass_id, incident_name, data, now): """Return spotting wire if criteria met, else None.""" threshold_mi = float(adapter_config.fires.spotting_distance_threshold_mi) - cooldown_s = int(adapter_config.fires.spotting_cooldown_seconds) + # Phase-3c: the cooldown gate (and its latch) moved into gating.firms.decide; + # the cooldown seconds are read there. The geometry below stays inline. # Most recent CLOSED pass with a perimeter (i.e. not the current pass). prev = conn.execute( @@ -926,16 +971,6 @@ def _check_spotting(conn, *, irwin_id, pixel_lat, pixel_lon, if dist_mi < threshold_mi: return None - # Cooldown gate. - fires_row = conn.execute( - "SELECT last_spotting_broadcast_at FROM fires WHERE irwin_id=?", - (irwin_id,), - ).fetchone() - if fires_row is not None: - last_ts = fires_row["last_spotting_broadcast_at"] - if last_ts is not None and (float(now) - float(last_ts)) < cooldown_s: - return None - # Direction is FROM the perimeter centroid (the previous pass's # pass_centroid_lat/lon, already on the row) TO this pixel. direction = _direction_8(_bearing( @@ -943,20 +978,50 @@ def _check_spotting(conn, *, irwin_id, pixel_lat, pixel_lon, pixel_lat, pixel_lon, )) - # Stamp the latch + tag data. - conn.execute( - "UPDATE fires SET last_spotting_broadcast_at=? WHERE irwin_id=?", - (float(now), irwin_id), - ) - if isinstance(data, dict): - data["category"] = "wildfire_spotting" - data["severity"] = "immediate" + # Phase-3c: the cooldown DECISION (and its latch) is delegated to + # gating.firms.decide. The wire is still rendered inline for byte-identity. + # On the not-cutover live path the eager fires.last_spotting_broadcast_at + # latch is kept VERBATIM (stamped with the handler `now`); on the cutover + # path the latch is deferred into gate.commit (tier-b change). + from meshai.notifications.cutover import is_cutover + from meshai.notifications.gating.firms import decide as _firms_decide - return ( + gate = _firms_decide( + {"_kind": "firms_spotting", "irwin_id": irwin_id, "dist_mi": dist_mi, + "direction": direction, "incident_name": incident_name}, + source="firms", now=float(now)) + if not gate.broadcast: + return None + + wire = ( f"๐Ÿ”ฅ Possible spotting {dist_mi:.1f} mi {direction} of " f"{incident_name} perimeter" ) + if isinstance(data, dict): + if is_cutover("wildfire_spotting"): + # NEW PATH: formatter re-renders from data_patch; latch deferred. + data.update(gate.data_patch) + data["_broadcast_audit"] = {"table": "fires", "pk": irwin_id} + _raw_commit = gate.commit + + def _on_commit(committed_at: float) -> None: + if _raw_commit is not None: + _raw_commit(committed_at) + + data["_on_broadcast_committed"] = _on_commit + else: + # LEGACY verbatim (byte-for-byte identical live output): eager latch + # with the handler `now`, plain category/severity stamps. + conn.execute( + "UPDATE fires SET last_spotting_broadcast_at=? WHERE irwin_id=?", + (float(now), irwin_id), + ) + data["category"] = "wildfire_spotting" + data["severity"] = "immediate" + + return wire + def _convex_hull(points): """Andrew's monotone-chain convex hull. Returns the hull as a list of diff --git a/work/meshai/notifications/formatters/__init__.py b/work/meshai/notifications/formatters/__init__.py index b65cde6..ce904dc 100644 --- a/work/meshai/notifications/formatters/__init__.py +++ b/work/meshai/notifications/formatters/__init__.py @@ -108,3 +108,20 @@ from meshai.notifications.formatters import fire as _fire_fmt_mod # noqa: E402, register("wildfire_declared", _fire_fmt_mod.format) register("wildfire_incident", _fire_fmt_mod.format) register("wildfire_closed", _fire_fmt_mod.format) + +# Phase-3c: FIRMS fusion broadcasts. Three categories the firms_handler emits, +# all under toggle "fire". `wildfire_growth` REUSES the fire formatter โ€” its +# wire is the WFIGS incident render with a movement dict, and feeding the fire +# formatter only {incident_name, is_update, movement, lat, lon} reproduces the +# legacy _render() call byte-for-byte (the growth SELECT's current_* / declared_at +# columns are NOT read by _render, so size/containment/date render as unknown/ +# absent โ€” matched here by omitting those fields). `wildfire_spotting` and +# `wildfire_halted` have their own terse wires -> formatters/firms.py. Registered +# under the explicit strings (not the "fire" toggle) so the still-deferred FIRMS +# native categories (wildfire_hotspot / new_ignition / unattributed_hotspot_cluster) +# are NOT captured by the family fallback. NOT cut over this phase. +register("wildfire_growth", _fire_fmt_mod.format) + +from meshai.notifications.formatters import firms as _firms_fmt_mod # noqa: E402,F401 +register("wildfire_spotting", _firms_fmt_mod.format) +register("wildfire_halted", _firms_fmt_mod.format) diff --git a/work/meshai/notifications/formatters/firms.py b/work/meshai/notifications/formatters/firms.py new file mode 100644 index 0000000..1352cb5 --- /dev/null +++ b/work/meshai/notifications/formatters/firms.py @@ -0,0 +1,53 @@ +"""FIRMS fire-fusion formatter โ€” Phase-3c migration. + +Reproduces the two FIRMS broadcast wire shapes that are NOT the WFIGS incident +render byte-identically, reading the render hints the decider stamps into +event.data: + + wildfire_spotting -> ๐Ÿ”ฅ Possible spotting {dist:.1f} mi {dir} of {name} perimeter + wildfire_halted -> ๐Ÿ”ฅ {name} no growth in {hours}h + +The third migrated FIRMS category, ``wildfire_growth``, reuses +``formatters/fire.py`` (its wire is the WFIGS incident render with a movement +dict) and is registered there โ€” NOT here. + +Neither legacy wire applied ``fit_to_budget`` (they were terse raw f-strings), +so this formatter deliberately does NOT budget-fit โ€” it returns the raw wire to +stay byte-identical. ``now`` / ``budget`` are accepted for signature parity and +unused (this formatter is clock-free per the Phase-0 purity guard). +""" +from __future__ import annotations + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from meshai.notifications.events import Event + + +def format(event: "Event", *, now: float, budget: int) -> str: + """Render the FIRMS spotting/halt wire from canonical event.data. + + Branch selection is on event.category (falling back to data["category"]). + """ + d = event.data or {} + category = None + try: + category = event.category + except Exception: + category = None + if category is None: + category = d.get("category") + + if category == "wildfire_halted": + name = d.get("incident_name") or "(unnamed fire)" + hours = d.get("hours") + return f"๐Ÿ”ฅ {name} no growth in {hours}h" + + # Default: wildfire_spotting. + dist_mi = d.get("dist_mi") + direction = d.get("direction") + incident_name = d.get("incident_name") + return ( + f"๐Ÿ”ฅ Possible spotting {dist_mi:.1f} mi {direction} of " + f"{incident_name} perimeter" + ) diff --git a/work/meshai/notifications/gating/__init__.py b/work/meshai/notifications/gating/__init__.py index 286074b..65602c8 100644 --- a/work/meshai/notifications/gating/__init__.py +++ b/work/meshai/notifications/gating/__init__.py @@ -100,3 +100,18 @@ from meshai.notifications.gating import fire as _fire_gate_mod # noqa: E402,F40 register("wildfire_declared", _fire_gate_mod.decide) register("wildfire_incident", _fire_gate_mod.decide) register("wildfire_closed", _fire_gate_mod.decide) + +# Phase-3c: FIRMS fusion broadcasts (wildfire_growth / wildfire_spotting / +# wildfire_halted). One decider handles all three; it keys off the handler- +# stamped internal `_kind` (firms_growth / firms_spotting / firms_halt). The +# growth/spotting/halt DECISIONS move here; the eager latch writes +# (last_spotting_broadcast_at / halt_broadcast_at) move into the deferred commit +# closures (an intended tier-b change). All attribution / pass / centroid / +# perimeter plumbing stays INLINE in firms_handler. NOTE: wildfire_growth's +# FORMATTER is fire.py but its DECIDER is firms.py (own decision math). The dead +# unattributed_hotspot_cluster path and native env/fires.py hotspot broadcasts +# are NOT migrated. NOT cut over this phase. +from meshai.notifications.gating import firms as _firms_gate_mod # noqa: E402,F401 +register("wildfire_growth", _firms_gate_mod.decide) +register("wildfire_spotting", _firms_gate_mod.decide) +register("wildfire_halted", _firms_gate_mod.decide) diff --git a/work/meshai/notifications/gating/firms.py b/work/meshai/notifications/gating/firms.py new file mode 100644 index 0000000..01ec65a --- /dev/null +++ b/work/meshai/notifications/gating/firms.py @@ -0,0 +1,241 @@ +"""FIRMS fire-fusion gating decider โ€” Phase-3c migration. + +Moves the three FIRMS *broadcast decisions* out of +``central.firms_handler`` into a source-agnostic decider, mirroring +quake/wfigs. It covers ONLY the decision + the deferred latch; every bit of +attribution / pass-aggregation / centroid / perimeter plumbing STAYS INLINE in +the handler (that data plumbing is analogous to hydro's ``gauge_readings`` +INSERT and does not belong in a decider). + +Three broadcast categories are handled, discriminated by an internal +``_kind`` the handler stamps onto the decider input: + + firms_growth -> wildfire_growth (drift >= threshold at a pass boundary) + firms_spotting -> wildfire_spotting (pixel outside prior perimeter, cooldown) + firms_halt -> wildfire_halted (fire idle >= halt_minimum_seconds) + +The unattributed-hotspot cluster path is DEAD (an unconditional ``return None`` +at the top of ``_maybe_emit_cluster``) and is NOT represented here. Native +``env/fires.py`` hotspot broadcasts (new_ignition / wildfire_hotspot) are a +deferred follow-up and are not migrated. + +Tier-b deferral (intended behavior change, validated by gate-sequence, NOT +golden byte-parity): + The eager latch writes the legacy handler performed *before* returning the + wire โ€” ``fires.last_spotting_broadcast_at`` (spotting) and + ``fires.halt_broadcast_at`` (halt) โ€” move into the deferred ``commit`` + closure. A broadcast that is composed but then dropped by the dispatcher + (cold-start grace / dedup / toggle) therefore no longer burns the latch, so + the next eligible pixel can still fire. ``wildfire_growth`` has no latch + (its re-eligibility is governed by the inline pass cursor), so its + ``commit`` is None. + +The decider input dicts the handler passes: + + firms_growth: {_kind, irwin_id, boundary, drift_mi, + drift_direction, drift_mi_per_hour} + firms_spotting: {_kind, irwin_id, dist_mi, direction, incident_name} + firms_halt: {_kind} +""" +from __future__ import annotations + +import logging + +from meshai.adapter_config import adapter_config +from meshai.notifications.gating.base import GateResult +from meshai.persistence import get_db + +logger = logging.getLogger(__name__) + + +def decide(data: dict, *, source: str, now: float) -> GateResult: + """Broadcast decision for a FIRMS fusion event. + + Parameters + ---------- + data: + Discriminated decider input (see module docstring for the per-kind + schema). ``data["_kind"]`` selects the growth / spotting / halt path. + source: + Adapter source name, e.g. "firms" (accepted for signature parity). + now: + Current epoch (float) โ€” determinism seam. + + Returns + ------- + GateResult with broadcast/lifecycle/data_patch/commit populated. The + handler still renders the wire inline and, on the not-cutover live path, + keeps its eager-latch / stamping VERBATIM; ``data_patch`` supplies the + stamps + render hints and ``commit`` arms the deferred latch on delivery. + """ + if not isinstance(data, dict): + return GateResult(broadcast=False, lifecycle="suppress", + reason="bad decider input") + kind = data.get("_kind") + + # โ”€โ”€ wildfire_growth โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + if kind == "firms_growth": + boundary = bool(data.get("boundary")) + drift_mi = data.get("drift_mi") + threshold = float(adapter_config.fires.growth_drift_threshold_mi) + if (not boundary) or drift_mi is None or drift_mi < threshold: + return GateResult(broadcast=False, lifecycle="suppress", + reason="no pass boundary or drift below threshold") + irwin_id = data.get("irwin_id") + drift_direction = data.get("drift_direction") + drift_mi_per_hour = data.get("drift_mi_per_hour") + patch: dict = { + # Legacy stamps (verbatim on the not-cutover path). + "category": "wildfire_growth", + "_severity_override": "immediate", + "_cooldown_suffix": irwin_id, + # Render hints for the fire formatter (cutover path). The growth + # wire reuses formatters/fire.py's incident render; feeding ONLY + # these fields reproduces the legacy _render() key-mismatch output + # (size unknown ยท containment unknown + movement line) byte-for-byte. + "is_update": True, + "movement": {"direction": drift_direction, + "speed_mph": (drift_mi_per_hour or 0.0)}, + } + # Growth has no latch โ€” re-eligibility is the inline pass cursor. + return GateResult(broadcast=True, lifecycle="growth", + reason="centroid drift >= threshold at pass boundary", + data_patch=patch, commit=None) + + # โ”€โ”€ wildfire_spotting โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + if kind == "firms_spotting": + irwin_id = data.get("irwin_id") + cooldown_s = int(adapter_config.fires.spotting_cooldown_seconds) + try: + conn = get_db() + except Exception: + logger.exception("firms decide (spotting): persistence unavailable") + return GateResult(broadcast=False, lifecycle="suppress", + reason="persistence unavailable") + # Cooldown gate โ€” mirrors _check_spotting's inline read exactly. + row = conn.execute( + "SELECT last_spotting_broadcast_at FROM fires WHERE irwin_id=?", + (irwin_id,), + ).fetchone() + if row is not None: + last_ts = row["last_spotting_broadcast_at"] + if last_ts is not None and (float(now) - float(last_ts)) < cooldown_s: + return GateResult(broadcast=False, lifecycle="cooldown", + reason="inside spotting cooldown") + patch = { + # Legacy stamps (verbatim on the not-cutover path): plain severity, + # NOT _severity_override. + "category": "wildfire_spotting", + "severity": "immediate", + # Render hints for the firms formatter (cutover path). + "dist_mi": data.get("dist_mi"), + "direction": data.get("direction"), + "incident_name": data.get("incident_name"), + } + return GateResult(broadcast=True, lifecycle="spotting", + reason="pixel outside prior perimeter beyond threshold", + data_patch=patch, + commit=_spotting_commit(irwin_id)) + + # โ”€โ”€ wildfire_halted โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + if kind == "firms_halt": + try: + conn = get_db() + except Exception: + logger.exception("firms decide (halt): persistence unavailable") + return GateResult(broadcast=False, lifecycle="suppress", + reason="persistence unavailable") + minimum_s = int(adapter_config.fires.halt_minimum_seconds) + cutoff = float(now) - float(minimum_s) + # Mirror _maybe_emit_halt's SELECT exactly (same ORDER/LIMIT so the + # chosen fire is identical). + row = conn.execute( + "SELECT irwin_id, incident_name, last_pass_at FROM fires " + "WHERE tombstoned_at IS NULL " + "AND last_pass_at IS NOT NULL AND last_pass_at <= ? " + "AND (halt_broadcast_at IS NULL " + " OR halt_broadcast_at < last_pass_at) " + "ORDER BY last_pass_at ASC LIMIT 1", + (cutoff,), + ).fetchone() + if row is None: + return GateResult(broadcast=False, lifecycle="suppress", + reason="no fire meets halt criteria") + irwin_id = row["irwin_id"] + name = row["incident_name"] or "(unnamed fire)" + hours = max(0, int((float(now) - float(row["last_pass_at"])) / 3600.0)) + patch = { + # Legacy stamps (verbatim on the not-cutover path). + "category": "wildfire_halted", + "severity": "routine", + # Render hints + the identity the handler needs for the wire/latch. + "incident_name": name, + "hours": hours, + "irwin_id": irwin_id, + } + return GateResult(broadcast=True, lifecycle="halt", + reason=f"fire idle irwin={irwin_id}", + data_patch=patch, + commit=_halt_commit(irwin_id)) + + # Unknown kind (e.g. the decider was invoked from the shadow/cutover harness + # with a real FIRMS event.data before FIRMS is shadowed) โ€” suppress. FIRMS + # shadow/cutover is a deferred follow-up. + return GateResult(broadcast=False, lifecycle="suppress", + reason=f"unhandled firms kind {kind!r}") + + +def _spotting_commit(irwin_id): + """Deferred latch: stamp fires.last_spotting_broadcast_at on delivery. + + Tier-b change: legacy stamped this eagerly with the handler ``now`` before + returning the wire; here it fires only on confirmed delivery, at the + delivery timestamp. Idempotent (a plain UPDATE), never raises. + """ + def _commit(committed_at: float) -> None: + try: + conn = get_db() + except Exception: + logger.exception( + "firms spotting commit: persistence unavailable; " + "last_spotting_broadcast_at not stamped for irwin=%s", irwin_id) + return + try: + conn.execute( + "UPDATE fires SET last_spotting_broadcast_at=? WHERE irwin_id=?", + (float(committed_at), irwin_id), + ) + except Exception: + logger.exception( + "firms spotting commit: latch UPDATE failed irwin=%s", irwin_id) + + return _commit + + +def _halt_commit(irwin_id): + """Deferred latch: stamp fires.halt_broadcast_at on delivery. + + Tier-b change: legacy stamped this eagerly with the handler ``now``; here it + fires only on confirmed delivery, at the delivery timestamp. The halt + re-eligibility filter (``halt_broadcast_at IS NULL OR halt_broadcast_at < + last_pass_at``) still holds because committed_at > now > last_pass_at. + Idempotent (a plain UPDATE), never raises. + """ + def _commit(committed_at: float) -> None: + try: + conn = get_db() + except Exception: + logger.exception( + "firms halt commit: persistence unavailable; " + "halt_broadcast_at not stamped for irwin=%s", irwin_id) + return + try: + conn.execute( + "UPDATE fires SET halt_broadcast_at=? WHERE irwin_id=?", + (float(committed_at), irwin_id), + ) + except Exception: + logger.exception( + "firms halt commit: latch UPDATE failed irwin=%s", irwin_id) + + return _commit diff --git a/work/tests/test_fire_refactor.py b/work/tests/test_fire_refactor.py index c40c0fa..7201955 100644 --- a/work/tests/test_fire_refactor.py +++ b/work/tests/test_fire_refactor.py @@ -310,10 +310,14 @@ class TestRegistration: @pytest.mark.parametrize( "cat", ["wildfire_hotspot", "new_ignition", - "unattributed_hotspot_cluster", "wildfire_growth"]) + "unattributed_hotspot_cluster"]) def test_firms_categories_not_captured(self, cat): - # FIRMS is deferred; its categories must NOT resolve to the fire - # formatter/decider via the "fire" toggle family fallback. + # 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_firms_refactor.py b/work/tests/test_firms_refactor.py new file mode 100644 index 0000000..933df8f --- /dev/null +++ b/work/tests/test_firms_refactor.py @@ -0,0 +1,443 @@ +"""Phase-3c FIRMS fusion refactor tests. + +Verifies the source-agnostic formatter+decider migration for the THREE FIRMS +fusion broadcast categories โ€” wildfire_growth / wildfire_spotting / +wildfire_halted โ€” mirroring test_fire_refactor.py (WFIGS) and +test_hydro_refactor.py: + +1. Registration: the three categories resolve to the right formatter/decider; + wildfire_growth reuses the FIRE formatter but keeps its OWN firms decider; + still-deferred native FIRMS categories do NOT resolve. + +2. Gate-sequence + deferred-latch (the tier-b validation): a `now`-timeline + driven through the NEW gating.firms.decide() reproduces the OLD handle_firms + broadcast/suppress + stamp behavior, AND the latch is now DEFERRED โ€” a + decision does NOT burn the latch until commit() fires (simulating delivery), + after which the next decision suppresses. + +3. Golden byte-identity: the cutover formatter path reproduces the legacy inline + wire byte-for-byte for growth (via fire.py), spotting, and halt (via firms.py). + +4. Not-cutover parity: with no category cut over, handle_firms keeps the legacy + eager-latch + stamps VERBATIM (byte-identical live behavior). + +5. The unattributed_hotspot_cluster path stays DEAD (returns None). +""" +from __future__ import annotations + +import math +import uuid + +import pytest + +from meshai.central.budget import budget_for +from meshai.notifications.formatters.fire import format as fire_format +from meshai.notifications.formatters.firms import format as firms_format +from meshai.notifications.gating.firms import decide as firms_decide +from tests.harness.goldens import assert_byte_identical + +_MI_PER_DEG_LAT = 69.0 + + +@pytest.fixture(autouse=True) +def _isolate_db(tmp_path, monkeypatch): + db_path = str(tmp_path / f"meshai-{uuid.uuid4().hex}.sqlite") + monkeypatch.setenv("MESHAI_DB_PATH", db_path) + from meshai.persistence import db as pdb + pdb.close_thread_connection() + pdb._initialised.discard(db_path) + from meshai.persistence import init_db + init_db(db_path) + try: + from meshai.adapter_config import adapter_config as _ac + _ac.invalidate() + except Exception: + pass + yield db_path + pdb.close_thread_connection() + pdb._initialised.discard(db_path) + + +@pytest.fixture +def _no_cutover(monkeypatch): + """Default deploy state: nothing cut over โ†’ handler runs legacy verbatim.""" + monkeypatch.delenv("MESHAI_CUTOVER_CATEGORIES", raising=False) + from meshai.notifications.cutover import _clear_cache + _clear_cache() + yield + _clear_cache() + + +def _cutover(monkeypatch, *cats): + monkeypatch.setenv("MESHAI_CUTOVER_CATEGORIES", ",".join(cats)) + from meshai.notifications.cutover import _clear_cache + _clear_cache() + + +class _FakeEvent: + def __init__(self, data, category=None): + self.data = data + self.category = category + + +# โ”€โ”€ shared FIRMS driving helpers (mirror test_fire_tracker_phase2/3) โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + +def _seed_fire(*, irwin_id, lat, lon, name="Stub Fire", **cols): + import time + from meshai.persistence import get_db + conn = get_db() + base = {"irwin_id": irwin_id, "incident_name": name, "lat": lat, "lon": lon, + "last_event_at": int(time.time())} + base.update(cols) + keys = ",".join(base) + ph = ",".join("?" * len(base)) + conn.execute(f"INSERT INTO fires({keys}) VALUES ({ph})", tuple(base.values())) + + +def _envelope(*, lat, lon, acq_date="2026-06-06", acq_time="1200", + frp=20.0, satellite="N20"): + return { + "data": { + "adapter": "firms", + "category": "wildfire_hotspot", + "severity": "routine", + "data": { + "latitude": lat, "longitude": lon, "frp": frp, + "bright_ti4": 320.0, "satellite": satellite, + "instrument": "VIIRS", "confidence": "high", + "acq_date": acq_date, "acq_time": acq_time, + "daynight": "D", "version": "2.0NRT", + }, + } + } + + +_SUBJECT = "central.fire.hotspot.N20.high.us.id" + + +def _drive_two_pass_growth(irwin_id, center_lat, center_lon): + """Seed a fire + pass A (5 px) + first pass-B pixel 1 mi N. Returns the + (wire, data) from the boundary pixel that fires wildfire_growth.""" + from meshai.central.firms_handler import handle_firms + _seed_fire(irwin_id=irwin_id, lat=center_lat, lon=center_lon, name="Pine Gulch") + for i in range(5): + env = _envelope(lat=center_lat + 0.0001 * i, + lon=center_lon + 0.0001 * (i - 2), + acq_date="2026-06-06", acq_time=f"12{i:02d}", + frp=20.0 + i) + handle_firms(env, subject=_SUBJECT, data={}, now=1780747200 + i) + pass_b_lat = center_lat + (1.0 / _MI_PER_DEG_LAT) + env_b = _envelope(lat=pass_b_lat, lon=center_lon, acq_time="1800", frp=22.0) + data = {} + wire = handle_firms(env_b, subject=_SUBJECT, data=data, now=1780768800) + return wire, data + + +def _seed_pass_a_hex_then_close(irwin_id, center_lat, center_lon, + start_now=1780747200): + from meshai.central.firms_handler import handle_firms + _seed_fire(irwin_id=irwin_id, lat=center_lat, lon=center_lon, name=irwin_id) + for i in range(6): + angle = i * math.pi / 3 + la = center_lat + (0.5 / _MI_PER_DEG_LAT) * math.sin(angle) + cos_lat = math.cos(math.radians(center_lat)) + lo = center_lon + (0.5 / (_MI_PER_DEG_LAT * cos_lat)) * math.cos(angle) + env = _envelope(lat=la, lon=lo, acq_time=f"12{i * 2:02d}") + handle_firms(env, subject=_SUBJECT, data={}, now=start_now + i) + + +def _offset_mi(lat, lon, north_mi, east_mi): + dlat = north_mi / _MI_PER_DEG_LAT + dlon = east_mi / (_MI_PER_DEG_LAT * math.cos(math.radians(lat))) + return lat + dlat, lon + dlon + + +def _drive_spotting(irwin_id, center_lat, center_lon, now=1780768800): + """Seed hex pass A + closed perimeter, then a pass-B pixel 2 mi NE that + fires wildfire_spotting. Returns (wire, data).""" + from meshai.central.firms_handler import handle_firms + _seed_pass_a_hex_then_close(irwin_id, center_lat, center_lon) + sp_lat, sp_lon = _offset_mi(center_lat, center_lon, + north_mi=2.0 / math.sqrt(2), + east_mi=2.0 / math.sqrt(2)) + env_b = _envelope(lat=sp_lat, lon=sp_lon, acq_time="1800") + data = {} + wire = handle_firms(env_b, subject=_SUBJECT, data=data, now=now) + return wire, data + + +def _seed_stale_fire(irwin_id, *, now_epoch, idle_hours=14, name="Cold Fire"): + from meshai.persistence import get_db + idle_at = now_epoch - (idle_hours * 3600) + get_db().execute( + "INSERT INTO fires(irwin_id, incident_name, lat, lon, last_event_at, " + "last_pass_id, last_pass_at) VALUES (?,?,?,?,?,?,?)", + (irwin_id, name, 42.5, -114.5, int(idle_at), "N20-329627", + float(idle_at)), + ) + + +# โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ +# 1. Registration +# โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + +class TestRegistration: + def test_growth_formatter_is_fire(self): + from meshai.notifications.formatters import get_formatter + assert get_formatter("wildfire_growth") is fire_format + + @pytest.mark.parametrize("cat", ["wildfire_spotting", "wildfire_halted"]) + def test_spotting_halt_formatter_is_firms(self, cat): + from meshai.notifications.formatters import get_formatter + assert get_formatter(cat) is firms_format + + @pytest.mark.parametrize( + "cat", ["wildfire_growth", "wildfire_spotting", "wildfire_halted"]) + def test_decider_is_firms(self, cat): + from meshai.notifications.gating import get_decider + assert get_decider(cat) is firms_decide + + @pytest.mark.parametrize( + "cat", ["wildfire_hotspot", "new_ignition", + "unattributed_hotspot_cluster"]) + def test_deferred_firms_natives_not_registered(self, cat): + from meshai.notifications.formatters import get_formatter + from meshai.notifications.gating import get_decider + assert get_formatter(cat) is None + assert get_decider(cat) is None + + +# โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ +# 2. Gate-sequence + deferred-latch (tier-b) +# โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + +class TestGrowthDecideSequence: + def _in(self, **over): + d = {"_kind": "firms_growth", "irwin_id": "ID-G", "boundary": True, + "drift_mi": 1.0, "drift_direction": "N", "drift_mi_per_hour": 0.16} + d.update(over) + return d + + def test_boundary_drift_broadcasts_with_stamps(self): + gr = firms_decide(self._in(), source="firms", now=1000.0) + assert gr.broadcast is True + assert gr.data_patch["category"] == "wildfire_growth" + assert gr.data_patch["_severity_override"] == "immediate" + assert gr.data_patch["_cooldown_suffix"] == "ID-G" + # Growth has NO latch to defer. + assert gr.commit is None + + def test_no_boundary_suppresses(self): + gr = firms_decide(self._in(boundary=False), source="firms", now=1000.0) + assert gr.broadcast is False + + def test_sub_threshold_drift_suppresses(self): + # Default growth_drift_threshold_mi is 0.5; 0.3 mi is below. + gr = firms_decide(self._in(drift_mi=0.3), source="firms", now=1000.0) + assert gr.broadcast is False + + +class TestSpottingDecideSequence: + def _in(self, **over): + d = {"_kind": "firms_spotting", "irwin_id": "ID-S", "dist_mi": 2.0, + "direction": "NE", "incident_name": "Pine Gulch"} + d.update(over) + return d + + def test_first_broadcasts_with_plain_severity(self): + _seed_fire(irwin_id="ID-S", lat=43.0, lon=-115.0) + gr = firms_decide(self._in(), source="firms", now=1_000_000.0) + assert gr.broadcast is True + assert gr.data_patch["category"] == "wildfire_spotting" + assert gr.data_patch["severity"] == "immediate" + assert "_severity_override" not in gr.data_patch + assert gr.commit is not None + + def test_latch_not_burned_without_commit(self): + """Tier-b: a decision that is never delivered must NOT burn the latch.""" + _seed_fire(irwin_id="ID-S", lat=43.0, lon=-115.0) + t0 = 1_000_000.0 + gr1 = firms_decide(self._in(), source="firms", now=t0) + assert gr1.broadcast is True + # No commit() โ†’ 10s later a fresh candidate STILL broadcasts. + gr2 = firms_decide(self._in(), source="firms", now=t0 + 10) + assert gr2.broadcast is True, "latch burned without delivery (tier-b broken)" + + def test_commit_then_cooldown_suppresses_then_reopens(self): + _seed_fire(irwin_id="ID-S", lat=43.0, lon=-115.0) + t0 = 1_000_000.0 + gr1 = firms_decide(self._in(), source="firms", now=t0) + gr1.commit(t0) # simulate delivery โ†’ stamps last_spotting_broadcast_at=t0 + # 30 min later (< 1h cooldown) โ†’ suppressed. + gr2 = firms_decide(self._in(), source="firms", now=t0 + 1800) + assert gr2.broadcast is False + # 2h later (> cooldown) โ†’ reopens. + gr3 = firms_decide(self._in(), source="firms", now=t0 + 7200) + assert gr3.broadcast is True + + +class TestHaltDecideSequence: + def test_first_broadcasts_with_hours_and_severity(self): + now = 1780768800.0 + _seed_stale_fire("ID-H", now_epoch=now, idle_hours=14) + gr = firms_decide({"_kind": "firms_halt"}, source="firms", now=now) + assert gr.broadcast is True + assert gr.data_patch["category"] == "wildfire_halted" + assert gr.data_patch["severity"] == "routine" + assert gr.data_patch["hours"] == 14 + assert gr.data_patch["incident_name"] == "Cold Fire" + assert gr.data_patch["irwin_id"] == "ID-H" + assert gr.commit is not None + + def test_latch_not_burned_without_commit(self): + now = 1780768800.0 + _seed_stale_fire("ID-H", now_epoch=now, idle_hours=14) + gr1 = firms_decide({"_kind": "firms_halt"}, source="firms", now=now) + assert gr1.broadcast is True + gr2 = firms_decide({"_kind": "firms_halt"}, source="firms", now=now) + assert gr2.broadcast is True, "latch burned without delivery (tier-b broken)" + + def test_commit_suppresses_then_reeligible_after_new_pass(self): + from meshai.persistence import get_db + now = 1780768800.0 + _seed_stale_fire("ID-H", now_epoch=now, idle_hours=14) + gr1 = firms_decide({"_kind": "firms_halt"}, source="firms", now=now) + gr1.commit(now) # stamps halt_broadcast_at=now + # Same fire no longer eligible (halt_broadcast_at >= last_pass_at). + gr2 = firms_decide({"_kind": "firms_halt"}, source="firms", now=now) + assert gr2.broadcast is False + # Fire reactivates: a new pass advances last_pass_at past the latch. + new_pass_at = now + 100 + get_db().execute("UPDATE fires SET last_pass_at=? WHERE irwin_id=?", + (float(new_pass_at), "ID-H")) + # Evaluate later, when the fire is idle again โ†’ re-eligible. + later = new_pass_at + 14 * 3600 + gr3 = firms_decide({"_kind": "firms_halt"}, source="firms", now=later) + assert gr3.broadcast is True + + +# โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ +# 3. Golden byte-identity โ€” cutover formatter reproduces the legacy inline wire +# โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + +class TestFormatterGolden: + def test_growth_wire_reuses_fire_formatter(self, monkeypatch): + # Drive the real growth broadcast under cutover: handle_firms returns the + # inline _render() wire AND populates data with the fire formatter hints. + _cutover(monkeypatch, "wildfire_growth") + try: + wire, data = _drive_two_pass_growth("ID-GG", 42.0, -114.0) + assert wire is not None and wire.startswith("๐Ÿ”ฅ Pine Gulch") + rendered = fire_format(_FakeEvent(data, category="wildfire_growth"), + now=0.0, budget=budget_for("wfigs")) + assert_byte_identical(rendered, wire) + finally: + from meshai.notifications.cutover import _clear_cache + _clear_cache() + + def test_spotting_wire_golden(self, monkeypatch): + _cutover(monkeypatch, "wildfire_spotting") + try: + wire, data = _drive_spotting("ID-SS", 43.0, -115.0) + assert wire is not None and wire.startswith("๐Ÿ”ฅ Possible spotting ") + rendered = firms_format(_FakeEvent(data, category="wildfire_spotting"), + now=0.0, budget=budget_for("firms")) + assert_byte_identical(rendered, wire) + finally: + from meshai.notifications.cutover import _clear_cache + _clear_cache() + + def test_halt_wire_golden(self, monkeypatch): + from meshai.central.firms_handler import _maybe_emit_halt + from meshai.persistence import get_db + _cutover(monkeypatch, "wildfire_halted") + try: + now = 1780768800 + _seed_stale_fire("ID-HH", now_epoch=now, idle_hours=14) + data = {} + wire = _maybe_emit_halt(get_db(), data=data, now=now) + assert wire is not None and "no growth in 14h" in wire + rendered = firms_format(_FakeEvent(data, category="wildfire_halted"), + now=0.0, budget=budget_for("firms")) + assert_byte_identical(rendered, wire) + finally: + from meshai.notifications.cutover import _clear_cache + _clear_cache() + + def test_spotting_formatter_exact_format(self): + # Pin the exact legacy f-string shape independent of the driver. + wire = firms_format( + _FakeEvent({"dist_mi": 2.34, "direction": "SW", + "incident_name": "Cache Peak"}, + category="wildfire_spotting"), + now=0.0, budget=140) + assert wire == "๐Ÿ”ฅ Possible spotting 2.3 mi SW of Cache Peak perimeter" + + def test_halt_formatter_exact_format(self): + wire = firms_format( + _FakeEvent({"incident_name": "Cache Peak", "hours": 9}, + category="wildfire_halted"), + now=0.0, budget=140) + assert wire == "๐Ÿ”ฅ Cache Peak no growth in 9h" + + +# โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ +# 4. Not-cutover parity โ€” legacy eager-latch + stamps preserved VERBATIM +# โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + +class TestNotCutoverLegacyVerbatim: + def test_growth_stamps_and_no_latch(self, _no_cutover): + wire, data = _drive_two_pass_growth("ID-GN", 42.0, -114.0) + assert wire is not None and wire.startswith("๐Ÿ”ฅ Pine Gulch") + assert "Moving N" in wire + assert data["category"] == "wildfire_growth" + assert data["_severity_override"] == "immediate" + assert data["_cooldown_suffix"] == "ID-GN" + # Legacy growth never attached a deferred commit. + assert "_on_broadcast_committed" not in data + + def test_spotting_eager_latch_stamped(self, _no_cutover): + from meshai.persistence import get_db + wire, data = _drive_spotting("ID-SN", 43.0, -115.0, now=1780768800) + assert wire is not None and "spotting" in wire + assert data["category"] == "wildfire_spotting" + assert data["severity"] == "immediate" + # Legacy path stamps the latch EAGERLY with the handler `now`. + latch = get_db().execute( + "SELECT last_spotting_broadcast_at FROM fires WHERE irwin_id=?", + ("ID-SN",)).fetchone()[0] + assert latch == 1780768800.0 + assert "_on_broadcast_committed" not in data + + def test_halt_eager_latch_stamped(self, _no_cutover): + from meshai.central.firms_handler import _maybe_emit_halt + from meshai.persistence import get_db + now = 1780768800 + _seed_stale_fire("ID-HN", now_epoch=now, idle_hours=14) + data = {} + wire = _maybe_emit_halt(get_db(), data=data, now=now) + assert wire == "๐Ÿ”ฅ Cold Fire no growth in 14h" + assert data["category"] == "wildfire_halted" + assert data["severity"] == "routine" + latch = get_db().execute( + "SELECT halt_broadcast_at FROM fires WHERE irwin_id=?", + ("ID-HN",)).fetchone()[0] + assert latch == float(now) + assert "_on_broadcast_committed" not in data + + +# โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ +# 5. Cluster path stays DEAD +# โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + +class TestClusterDead: + def test_maybe_emit_cluster_returns_none(self): + from meshai.central.firms_handler import _maybe_emit_cluster + from meshai.persistence import get_db + data = {} + out = _maybe_emit_cluster( + get_db(), lat=43.0, lon=-115.0, acq_epoch=1780747200, + frp=20.0, data=data, now=1780747200, this_pixel_id=1) + assert out is None + # The dead path must not tag the data dict either. + assert data == {}