matt/meshai
1
0
Fork 0
mirror of https://github.com/zvx-echo6/meshai.git synced 2026-08-26 17:31:34 +00:00
Code Issues Projects Releases Packages Wiki Activity

refactor(phase3c): migrate FIRMS fusion broadcasts to formatter+decider

Browse source 
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) <noreply@anthropic.com>
This commit is contained in:
Matt Johnson 2026-07-05 05:18:27 +00:00
commit c7a2cd626a
7 changed files with 896 additions and 58 deletions
Download patch file Download diff file Expand all files Collapse all files

173
work/meshai/central/firms_handler.py
View file

@ -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

17
work/meshai/notifications/formatters/__init__.py
View file

@ -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)

53
work/meshai/notifications/formatters/firms.py Normal file
View file

@ -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"
)

15
work/meshai/notifications/gating/__init__.py
View file

@ -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)

241
work/meshai/notifications/gating/firms.py Normal file
View file

@ -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

10
work/tests/test_fire_refactor.py
View file

@ -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

443
work/tests/test_firms_refactor.py Normal file
View file

@ -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 == {}