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

chore(central-ripout 2d-ii): remove dead fire scraps (NOT the 'rewrite' — there wasn't one) (#166)

Browse source 
* chore(fire-ripout 2dii-a): remove dead wildfire_growth cutover path + handle_firms

wildfire_growth events are always fully precomposed (env/firms.py sets
_meshai_precomposed=True + title=<wire from _render()>) and the category is
not in cutover.NATIVE_ALWAYS_DECIDE, so the registered fire formatter could
never be reached live via compose_mesh_message for this category -- the
precomposed-title bypass always won first. Removes the dead
is_cutover("wildfire_growth") NEW-PATH branch in
fire_fusion._handle_pass_boundary (kept the live legacy leg that calls
_render unconditionally) and the now-unreachable wildfire_growth formatter
registration in notifications/formatters/__init__.py.

Also removes handle_firms, the dead Central NATS-envelope entrypoint (zero
live production callers -- Central's consumer that drove it is gone), plus
its envelope-specific filtering helpers (_confidence_passes, _in_bbox,
_coerce_severity, _log_event, FIRMS_CONFIDENCE_FLOOR/FRP_FLOOR/BBOX_OPTIONAL)
that had no live callers left. The shared, LIVE fusion core
(_ingest_pixel_core, attribution, clustering, growth/spotting/halt) and
_parse_acq_epoch are unaffected -- still the engine behind the native
ingest_hotspot_pixel entrypoint.

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

* chore(fire-ripout 2dii-b): remove dead handle_wfigs entrypoint

handle_wfigs (the dead Central NATS-envelope entrypoint) had zero live
production callers -- Central's consumer that drove it is gone. The LIVE
WFIGS path is env/fires.py (native adapter) -> env/store.py::_emit_event,
forced onto gating.fire.decide + the shared fire formatter via
cutover.NATIVE_ALWAYS_DECIDE, independent of this dead handler.

Removes handle_wfigs and its private-only helpers (_coerce_severity,
_log_event, _log_event_returning_id) that had no callers left. _render
remains -- it is called directly by env.fire_fusion._handle_pass_boundary on
the FIRMS wildfire_growth path, and is used as a byte-identity oracle by
tests against the shared, live fire formatter. _build_canonical,
_attach_commit_handles, _fire_too_old_to_announce, _now, _cleanup_stale_fires
and WFIGS_BROADCAST_COOLDOWN_S are kept -- each has its own direct test
coverage independent of handle_wfigs.

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

* test(fire-ripout 2dii-c): rewire fire tests off the deleted dead entrypoints

handle_firms and handle_wfigs (both removed in the prior two commits) were
used throughout this test suite purely as convenient DRIVERS for live logic
(pixel attribution/clustering/growth/spotting/halt fusion; the WFIGS
New/Update/cooldown/closed decider). Per-file disposition:

Rewired to drive the LIVE native entrypoint directly (no behavior change --
same underlying _ingest_pixel_core / gating.fire.decide engines):
  - test_fire_tracker_phase1/2/3.py: handle_firms -> ingest_hotspot_pixel
  - test_firms_refactor.py: handle_firms driver helpers -> ingest_hotspot_pixel;
    dropped the dead wildfire_growth cutover test + the fully-redundant
    TestNotCutoverLegacyVerbatim class (already covered by
    test_firms_native_fusion.py's TestIngestGrowth/Spotting/Halt against the
    live entrypoint); wildfire_growth formatter registration test now
    asserts None (matches source change).
  - test_wfigs_handler.py: handle_wfigs -> _render (the live WFIGS renderer)
    called directly, for the anchor-priority + missing-acres cases that
    exercise shared/live code (_location_anchor).
  - test_fire_refactor.py: handle_wfigs -> gating.fire.decide() driven
    directly, with state written the same unconditional shape the live
    native path uses; TestGateSequenceParity trimmed to focus on the
    tombstone/closed lifecycle step (not covered by
    test_fire_native_growth.py's native-adapter New/Update/cooldown coverage).
  - test_tombstone_broadcast.py, test_fire_age_gate.py: handle_wfigs ->
    gating.fire.decide() driven directly; same assertions, off the dead path.

Deleted as pure dead-entrypoint contract testing with no live equivalent:
  - test_firms_native_fusion.py::TestCentralPathUnchanged (2 tests) --
    existed only to guard handle_firms's own envelope parsing.
  - test_firms_handler.py: confidence/FRP/bbox filtering, missing-coords,
    non-firms-adapter guard, event_log accounting (all handle_firms-specific;
    the native adapter filters upstream in a different code path). Kept +
    rewired: the shared _ingest_pixel_core dedup behavior and
    _parse_acq_epoch's int/short acq_time parsing (both genuinely live/shared).
  - test_wfigs_handler.py: envelope field-extraction (acres-fallback chain,
    IA-placeholder-as-name), tombstone/perimeter subject -> event_log,
    New/Update/cooldown decision + audit-row wiring (all redundant with
    tests/test_fire_native_growth.py's coverage of gating.fire.decide()
    through the real native adapter).
  - test_tombstone_broadcast.py::test_commit_callback_flips_handled --
    asserted handle_wfigs's own event_log-row flip on commit, a Central-only
    concept the native path never used.
  - test_tail_followups.py::test_wfigs_tombstone_stamps_column -- asserted
    handle_wfigs's own inline tombstoned_at UPDATE; no live producer emits
    _kind=wfigs_tombstone today, so nothing in the live system stamps it.

Both live renderers stay covered: fire_format (wildfire_declared/incident,
via test_fire_refactor.py + test_fire_native_growth.py, untouched) and
_render (wildfire_growth via FIRMS, via test_fire_tracker_phase2.py +
test_firms_native_fusion.py's TestIngestGrowth, both driving the live
ingest_hotspot_pixel entrypoint).

Collection: 2010 -> 1974 tests (net -36: dead-only tests deleted, live
behavior rewired 1:1 or consolidated onto already-existing native-path
coverage). Full suite: 1974 passed, 0 failed.

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

---------

Co-authored-by: Matt Johnson <mj@k7zvx.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
malice 2026-07-17 22:11:00 -06:00 • committed by GitHub
commit 67176d66d5
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
14 changed files with 629 additions and 2026 deletions
Download patch file Download diff file Expand all files Collapse all files

333
work/meshai/env/fire_fusion.py vendored
View file

@ -1,73 +1,29 @@
"""v0.7-fire-tracker-3 FIRMS handler -- storage + attribution + cluster + growth/halt/spotting.
"""v0.7-fire-tracker-3 FIRMS handler -- attribution + cluster + growth/halt/spotting.
Relocated from `meshai.central.firms_handler` during the Central ripout
(central/ handler retirement, chore/ripout-2d). ``ingest_hotspot_pixel`` is
the LIVE fire-fusion engine with one consumer: the native FIRMS adapter
(`meshai.env.firms` -> `_run_fusion`). ``handle_firms`` (the Central
NATS-envelope entrypoint) has no live production caller -- Central's
consumer that drove it is gone -- but it remains the parity-tested legacy
contract (see `tests/test_firms_handler.py`, `tests/test_firms_refactor.py`,
`tests/test_fire_tracker_phase1/2/3.py`) -- kept verbatim, not deleted, per
that coverage.
(`meshai.env.firms` -> `_run_fusion`).
Pre-v0.6-1 the v0.5.13 default-deny gate at consumer._normalize() silently
dropped every `central.fire.hotspot.>` envelope because no per-adapter handler
existed (audit doc v0.6-phase1-audit.md finding #2). The `firms_pixels`
table was created in v0.5.8b (v1.sql:98-111) and has been empty ever since.
chore/ripout-2dii: the dead Central NATS-envelope entrypoint ``handle_firms``
(envelope field-extraction, confidence/FRP/bbox filtering, event_log
accounting) has been REMOVED -- it had zero live production callers (Central's
consumer that drove it is gone) and its own parity tests (`test_firms_handler.py`,
`TestCentralPathUnchanged` in `test_firms_native_fusion.py`). The shared,
LIVE core (`_ingest_pixel_core`, attribution, clustering, growth/spotting/halt
fusion) is unaffected and remains fully covered via `ingest_hotspot_pixel`
(see `test_firms_native_fusion.py`, `test_fire_tracker_phase1/2/3.py`).
This handler closes that gap: every passing FIRMS pixel lands in
`firms_pixels`. No mesh broadcasts are emitted -- FIRMS data is for the
LLM context (commit #5: env_reporter) and for the v0.6 fire-tracker
fusion (per v0.6-design-fire-tracker.md). Returning None from the handler
tells the consumer's default-deny clause "no broadcast", which is exactly
the v0.6-1 contract (memory rule 19).
Subject pattern (Central v0.10.0):
Subject pattern (Central v0.10.0, historical -- kept for context):
central.fire.hotspot.<satellite>.<confidence>.<region>
where <region> is `us.<state>` or `unknown`.
Envelope shape (from firms-investigation.md, 250 envelopes 2026-05-28..06-04):
envelope["data"]["adapter"] == "firms"
envelope["data"]["data"]:
latitude (REAL)
longitude (REAL)
frp (REAL, MW -- fire radiative power)
bright_ti4 (REAL, K -- VIIRS brightness temperature)
bright_ti5 (REAL, K, optional)
satellite ("N" Suomi-NPP | "N20" NOAA-20)
instrument ("VIIRS" so far; MODIS would extend this)
confidence ("nominal" | "high" | "low")
acq_date ("YYYY-MM-DD" UTC)
acq_time ("HHMM" UTC, 4-digit)
daynight ("D" | "N")
version (str)
_enriched.geocoder.city/state/county/landclass/elevation_m
Filtering (hardcoded defaults; commit #3 migrates these to adapter_config
GUI rows per Rule 17. Per Matt's lock: defaults become GUI default values
with no behavior change on first deploy.):
FIRMS_CONFIDENCE_FLOOR = "low" -- rank-based; "low" = store every conf
FIRMS_FRP_FLOOR = 0.0 -- 0 = store every FRP value
FIRMS_BBOX_OPTIONAL = None -- None = no spatial filter
Permissive defaults are intentional: storage is cheap and v0.6 fire-tracker
fusion (FIRMS + WFIGS) needs the full pixel stream to detect unattributed
clusters early. Query-time filtering happens in env_reporter (commit #5).
Dedup:
Unique partial index added in v4.sql on
(round(lat,5), round(lon,5), acq_time, satellite)
Same satellite pixel observation re-published via NATS reconnect /
JetStream replay is a no-op INSERT OR IGNORE. 5 decimals on lat/lon
is ~1.1 m precision -- well inside VIIRS' 375 m pixel.
event_log accounting:
handled=1 -> row inserted into firms_pixels (or dedup-hit -- still
"successfully handled" semantically: we know about it)
handled=0 -> dropped (missing coords / outside bbox / below conf
floor / below FRP floor / missing acq timestamp).
Category is suffixed with "|<reason>" for grep.
"""
from __future__ import annotations
from meshai.adapter_config import adapter_config
@ -75,7 +31,6 @@ from meshai.adapter_config import adapter_config
import json
import logging
import math
import time
from datetime import datetime, timezone
from typing import Any, Optional
@ -86,175 +41,8 @@ logger = logging.getLogger(__name__)
# ============================================================================
# v0.6-3b: all four settings now live in adapter_config.firms. Module-level
# names retained as backward-compat aliases for test monkeypatches; the
# handler reads via adapter_config so a GUI edit takes effect on the next
# envelope without restart.
# ============================================================================
# VIIRS-FIRMS confidence rank table (CODE -- NOAA-defined vocabulary).
_CONFIDENCE_RANK = {"low": 0, "nominal": 1, "high": 2}
# Back-compat aliases for tests that import these names. New code should
# read via adapter_config.firms.<key>.
FIRMS_CONFIDENCE_FLOOR = "low"
FIRMS_FRP_FLOOR = 0.0
FIRMS_BBOX_OPTIONAL: Optional[tuple[float, float, float, float]] = None
# ============================================================================
# Public entry point
# ============================================================================
def handle_firms(envelope: dict, subject: str,
data: Optional[dict] = None,
now: Optional[int] = None) -> Optional[str]:
"""Storage-only FIRMS handler. ALWAYS returns None.
Args:
envelope: CloudEvents envelope from the Central consumer.
subject: NATS subject (`central.fire.hotspot.<sat>.<conf>.<region>`).
data: Mutable Event.data dict (unused -- no broadcast attached).
now: Override current epoch for tests.
Returns:
None unconditionally. The v0.5.13 default-deny clause at
consumer._normalize() interprets None as "no broadcast", which is
the desired contract for storage-only adapters.
"""
if not isinstance(envelope, dict):
return None
inner = envelope.get("data") or {}
if (inner.get("adapter") or "") != "firms":
return None
d = inner.get("data") or {}
now = now if now is not None else int(time.time())
category_raw = inner.get("category") or ""
severity_word = _coerce_severity(inner.get("severity"))
event_id_external = inner.get("id")
try:
conn = get_db()
except Exception:
logger.exception("firms_handler: persistence unavailable; dropping")
return None
# ---- field extraction + validation -----------------------------------
lat = d.get("latitude")
lon = d.get("longitude")
if not (isinstance(lat, (int, float)) and isinstance(lon, (int, float))):
_log_event(conn, now=now, source="firms",
category=category_raw + "|missing_coords",
severity_word=severity_word,
event_id_external=event_id_external,
subject=subject, handled=0,
table_name=None, table_pk=None)
return None
lat = float(lat); lon = float(lon)
# ---- filter: bbox (optional) -----------------------------------------
if not _in_bbox(lat, lon):
_log_event(conn, now=now, source="firms",
category=category_raw + "|outside_bbox",
severity_word=severity_word,
event_id_external=event_id_external,
subject=subject, handled=0,
table_name=None, table_pk=None)
return None
# ---- filter: confidence floor ----------------------------------------
conf = d.get("confidence")
if not _confidence_passes(conf):
_log_event(conn, now=now, source="firms",
category=category_raw + "|below_confidence_floor",
severity_word=severity_word,
event_id_external=event_id_external,
subject=subject, handled=0,
table_name=None, table_pk=None)
return None
# ---- filter: FRP floor (no-op when FIRMS_FRP_FLOOR <= 0) -------------
frp_raw = d.get("frp")
try:
frp = float(frp_raw) if frp_raw is not None else None
except (TypeError, ValueError):
frp = None
import sys as _sys
_this = _sys.modules[__name__]
frp_floor = float(_this.FIRMS_FRP_FLOOR) if _this.FIRMS_FRP_FLOOR > 0 \
else float(adapter_config.firms.frp_floor)
if frp_floor > 0:
if frp is None or frp < frp_floor:
_log_event(conn, now=now, source="firms",
category=category_raw + "|below_frp_floor",
severity_word=severity_word,
event_id_external=event_id_external,
subject=subject, handled=0,
table_name=None, table_pk=None)
return None
# ---- acquisition timestamp (required for dedup key) ------------------
acq_epoch = _parse_acq_epoch(d.get("acq_date"), d.get("acq_time"))
if acq_epoch is None:
_log_event(conn, now=now, source="firms",
category=category_raw + "|missing_acq_time",
severity_word=severity_word,
event_id_external=event_id_external,
subject=subject, handled=0,
table_name=None, table_pk=None)
return None
# ---- persist + attribute + fuse (source-agnostic core) --------------
satellite = d.get("satellite") or ""
brightness_raw = d.get("bright_ti4") if d.get("bright_ti4") is not None \
else d.get("brightness")
try:
brightness = float(brightness_raw) if brightness_raw is not None else None
except (TypeError, ValueError):
brightness = None
# The INSERT-OR-IGNORE into firms_pixels + attribution + growth/spotting/
# halt fusion now live in the shared _ingest_pixel_core so the native
# env/firms.py adapter feeds the SAME engine. handle_firms keeps the
# envelope-specific concerns (field extraction, filtering, event_log
# accounting) here so the Central path stays byte-identical. `data` is the
# consumer's mutable Event.data dict; core stamps it in place on a fusion
# broadcast exactly as the old inline path did.
stored, rowid, wire = _ingest_pixel_core(
conn, lat=lat, lon=lon, acq_epoch=acq_epoch, frp=frp,
confidence=conf, brightness=brightness, satellite=satellite,
data=data, now=now,
)
# event_log row regardless of dedup outcome -- both "stored" and
# "dedup-hit" count as "handled" for accounting; the suffix tells them
# apart for ops grep. (Written after the core call: event_log is a
# distinct table with its own rowid sequence, so ordering it after the
# attribution INSERTs leaves the row content + relative order identical.)
cat_tag = category_raw if stored else category_raw + "|dedup_hit"
_log_event(conn, now=now, source="firms", category=cat_tag,
severity_word=severity_word,
event_id_external=event_id_external,
subject=subject, handled=1,
table_name="firms_pixels" if stored else None,
table_pk=(str(rowid) if stored else None))
# Dedup hits skip broadcast -- the original insert already had its chance.
if not stored:
return None
return wire
# ============================================================================
# Source-agnostic pixel ingest (Central + native env/firms.py share this)
# Source-agnostic pixel ingest (native env/firms.py; formerly also shared
# with the now-deleted Central handle_firms entrypoint)
# ============================================================================
@ -401,41 +189,6 @@ def ingest_hotspot_pixel(pixel: dict, *, now, seed=False) -> list[tuple[str, dic
# ============================================================================
def _confidence_passes(conf: Optional[str]) -> bool:
"""Return True iff `conf` is at or above the configured floor.
v0.6-3b: floor read from adapter_config.firms.confidence_floor; the
module-level FIRMS_CONFIDENCE_FLOOR still wins when explicitly
monkeypatched (so existing tests stay one-line).
"""
if conf is None:
return False
rank = _CONFIDENCE_RANK.get(str(conf).lower())
if rank is None:
return False
import sys
_this = sys.modules[__name__]
if _this.FIRMS_CONFIDENCE_FLOOR != "low":
floor_str = _this.FIRMS_CONFIDENCE_FLOOR
else:
floor_str = str(adapter_config.firms.confidence_floor)
floor = _CONFIDENCE_RANK.get(str(floor_str).lower(), 0)
return rank >= floor
def _in_bbox(lat: float, lon: float) -> bool:
import sys
_this = sys.modules[__name__]
if _this.FIRMS_BBOX_OPTIONAL is not None:
bbox = _this.FIRMS_BBOX_OPTIONAL
else:
bbox = adapter_config.firms.bbox
if bbox is None:
return True
min_lat, min_lon, max_lat, max_lon = bbox
return (min_lat <= lat <= max_lat) and (min_lon <= lon <= max_lon)
def _parse_acq_epoch(date_s: Optional[str],
time_s: Optional[Any]) -> Optional[int]:
"""FIRMS publishes acq_date 'YYYY-MM-DD' + acq_time HHMM (UTC).
@ -454,27 +207,6 @@ def _parse_acq_epoch(date_s: Optional[str],
return None
def _coerce_severity(sev: Any) -> Optional[str]:
if sev is None: return None
if isinstance(sev, str): return sev or None
try: return str(int(sev))
except (TypeError, ValueError): return str(sev)
def _log_event(conn, *, now, source, category, severity_word,
event_id_external, subject, handled,
table_name, table_pk) -> None:
"""event_log writer -- shape matches sibling handlers exactly."""
conn.execute(
"INSERT INTO event_log(received_at, source, category, severity_word, "
"event_id_external, nats_subject, handled, table_name, table_pk) "
"VALUES (?,?,?,?,?,?,?,?,?)",
(now, source, category, severity_word, event_id_external, subject,
int(bool(handled)), table_name, table_pk),
)
# ============================================================================
# v0.7-fire-tracker-1: attribution + unattributed-cluster detection
# ============================================================================
@ -892,12 +624,12 @@ def _handle_pass_boundary(conn, *, irwin_id, pass_id, lat, lon,
# 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
# the WFIGS _render (the live renderer for wildfire_growth -- see
# notifications/renderers/composer.py's `_meshai_precomposed` bypass;
# wildfire_growth is not in NATIVE_ALWAYS_DECIDE and the cutover formatter
# path was never invoked live, so it was removed -- chore/ripout-2dii).
# The boundary predicate reproduces the legacy guard (`last_pass_id ==
# pass_id or last_pass_id is None or prev is None` -> suppress) exactly.
from meshai.notifications.gating.firms import decide as _firms_decide
boundary_growth = ((last_pass_id != pass_id) and (last_pass_id is not None)
@ -934,25 +666,14 @@ def _handle_pass_boundary(conn, *, irwin_id, pass_id, lat, lon,
normalized = dict(fire)
normalized["irwin_id"] = irwin_id
if isinstance(data, dict):
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
# wildfire_growth's only live render path (byte-for-byte identical
# live output). The cutover NEW-PATH (fire formatter re-render) was
# removed -- wildfire_growth is not in NATIVE_ALWAYS_DECIDE and events
# are precomposed, so the formatter was never invoked live for this
# category (chore/ripout-2dii).
data["category"] = "wildfire_growth"
data["_severity_override"] = "immediate"
data["_cooldown_suffix"] = irwin_id
return _render(normalized, prefix="Update", movement=movement)

341
work/meshai/env/fire_render.py vendored
View file

@ -1,36 +1,21 @@
"""WFIGS handler: persistence-backed change-detection + wire renderer.
"""WFIGS wire renderer + shared fire helpers.
Relocated from `meshai.central.wfigs_handler` during the Central ripout
(central/ handler retirement, chore/ripout-2d). ``handle_wfigs`` 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 wildfire wire format
(see `tests/test_fire_refactor.py`, `tests/test_wfigs_handler.py`) -- kept
verbatim, not deleted, per that oracle. ``_render`` IS live: it is imported
by `meshai.env.fire_fusion._handle_pass_boundary` on the FIRMS growth path
(central/ handler retirement, chore/ripout-2d).
chore/ripout-2dii: the dead Central NATS-envelope entrypoint ``handle_wfigs``
(the New:/Update:/tombstone dispatch + event_log accounting + the
`is_cutover` legacy-vs-new branch) has been REMOVED -- it had zero live
production callers (Central's consumer that drove it is gone). ``_render``
IS live: it is called directly by
`meshai.env.fire_fusion._handle_pass_boundary` on the FIRMS growth path
(`env/firms.py` -> `ingest_hotspot_pixel` -> ... -> `_handle_pass_boundary`
-> `_render`) to render the `wildfire_growth` wire.
v0.5.8b refactor: New: vs Update: decision now keys on `last_broadcast_at`,
not on row existence. Cold-start scenarios where the dispatcher drops the
broadcast (cold-start grace, stale filter, cooldown, dedup) leave the fires
row with NULL last_broadcast_at, so the NEXT successful broadcast still
gets the "New:" prefix -- it really is the first delivery for that fire.
Cases (resolved at handler entry):
(i) row missing -> INSERT, prefix="New", return wire
(ii) row exists, last_broadcast_at IS NULL
-> UPDATE current_*, prefix="New",
return wire (never broadcast yet)
(iii) row exists, last_broadcast_at NOT NULL
-> UPDATE current_*, gate on change +
8h cooldown. If pass: prefix="Update",
return wire; else return None.
The last_broadcast_* UPDATE has moved OUT of the handler and INTO a callback
attached to event.data["_on_broadcast_committed"]. The dispatcher calls it
ONLY after a successful broadcast. The mesh_broadcasts_out audit row is now
inserted by the dispatcher (via event.data["_broadcast_audit"]) for the same
reason -- it should only exist for actually-delivered broadcasts.
-> `_render`) to render the `wildfire_growth` wire, and is used as a
byte-identity oracle by `tests/test_fire_refactor.py` /
`tests/test_wfigs_handler.py` against the shared, LIVE fire formatter
(`notifications/formatters/fire.py`, reached for `wildfire_declared` /
`wildfire_incident` via the native WFIGS adapter `env/fires.py` ->
`env/store.py::_emit_event` -> `gating.fire.decide`).
Concurrency: each consumer thread gets its own SQLite connection via
meshai.persistence.get_db() (threading.local pool). Writes are serial
@ -44,7 +29,7 @@ from meshai.notifications.formatters._budget import budget_for, fit_to_budget
import logging
import time
from typing import Any, Optional
from typing import Optional
from meshai.notifications import clock
@ -120,247 +105,18 @@ def _build_canonical(normalized: dict, kind: str) -> dict:
}
def handle_wfigs(normalized: dict, envelope: dict, subject: str,
data: Optional[dict] = None,
now: Optional[int] = None) -> Optional[str]:
"""Route a normalized WFIGS dict through persistence + change-detection.
Phase-3b refactor: the broadcast DECISION (New/Update/suppress + cooldown +
age-gate + all-clear eligibility) now lives in
``meshai.notifications.gating.fire.decide``. This handler keeps ownership
of the inline ``fires`` INSERT/UPDATE of ``current_*`` (unconditional state
write), the ``tombstoned_at`` stamp, the ``event_log`` row + handled flip,
and the wire it returns (mirror quake/nwis).
`data` is the mutable dict the caller (consumer._normalize) is composing
into the Event. When a broadcast should fire, the handler attaches an
`_on_broadcast_committed` callback and `_broadcast_audit` descriptor to
it; the dispatcher invokes both AFTER a successful deliver().
Cutover gate: when the emitted category is explicitly cut over, the handler
writes ``gate.data_patch`` + wraps ``gate.commit`` (new path live);
otherwise it keeps the legacy ``_attach_commit_handles`` / all-clear
stamping VERBATIM so the live broadcast stays byte-for-byte identical while
the new formatter+decider bake in shadow.
Returns a wire string when a broadcast should fire, None otherwise.
"""
if not isinstance(normalized, dict):
return None
kind = normalized.get("_kind")
if kind not in ("wfigs_incident", "wfigs_tombstone", "wfigs_perimeter"):
return None
now = now if now is not None else _now()
inner = envelope.get("data") or {} if isinstance(envelope, dict) else {}
category = inner.get("category") or ""
severity_word = _coerce_severity(inner.get("severity"))
irwin_id = normalized.get("irwin_id")
try:
conn = get_db()
except Exception:
logger.exception("wfigs_handler: persistence unavailable; "
"deferring to default pipeline")
return None
from meshai.notifications.cutover import is_cutover
from meshai.notifications.gating.fire import decide as _gate_decide
canonical = _build_canonical(normalized, kind)
if kind in ("wfigs_tombstone", "wfigs_perimeter"):
source = "wfigs_incidents" if kind == "wfigs_tombstone" else "wfigs_perimeters"
log_id = _log_event_returning_id(
conn, now=now, source=source, category=category,
severity_word=severity_word, irwin_id=irwin_id,
subject=subject, handled=0,
table_name=None, table_pk=irwin_id)
# v0.6-tail item 4: tombstone branch stamps fires.tombstoned_at so
# the ReminderScheduler stops re-broadcasting the closed fire.
# Only the tombstone kind closes the fire; perimeter polls don t.
# UNCONDITIONAL state write — stays inline, mirror the original.
if kind == "wfigs_tombstone" and irwin_id:
try:
conn.execute(
"UPDATE fires SET tombstoned_at=COALESCE(tombstoned_at, ?) "
"WHERE irwin_id=?",
(now, irwin_id),
)
except Exception:
logger.exception("wfigs: tombstoned_at stamp failed irwin=%s", irwin_id)
# All-clear broadcast: only fires that previously made it to mesh
# get a closure message. Silent for fires that were never broadcast.
# The DECISION (row exists AND last_broadcast_at NOT NULL) is delegated
# to the decider; the WIRE is still rendered here from the fire row for
# byte-identity, and the not-cutover data stamping stays verbatim.
if kind == "wfigs_tombstone" and irwin_id:
gate = _gate_decide(canonical, source="wfigs", now=float(now))
if gate.broadcast:
fire_row = conn.execute(
"SELECT incident_name, current_acres, current_contained_pct, "
"last_broadcast_at, county, state, lat, lon "
"FROM fires WHERE irwin_id = ?", (irwin_id,)
).fetchone()
name = fire_row["incident_name"] or "(unnamed fire)"
# Build line 2 parts
parts = []
if fire_row["current_acres"] is not None:
parts.append(f"{int(fire_row['current_acres']):,} ac")
if fire_row["current_contained_pct"] is not None:
parts.append(f"{int(fire_row['current_contained_pct'])}% contained")
# Location via _location_anchor with a minimal normalized dict
loc_dict = {
"lat": fire_row["lat"], "lon": fire_row["lon"],
"county": fire_row["county"], "state": fire_row["state"],
}
anchor = _location_anchor(loc_dict)
if anchor and anchor != "(location unknown)":
parts.append(anchor)
lines = [f"✅ {name} — contained & closed"]
if parts:
lines.append(" | ".join(parts))
wire = "\n".join(lines)
if isinstance(data, dict):
if is_cutover("wildfire_closed"):
# NEW PATH: formatter re-renders from data_patch fields.
data.update(gate.data_patch)
data["_broadcast_audit"] = {"table": "fires", "pk": irwin_id}
_raw_commit = gate.commit
_log_row_id = log_id
def _on_commit(committed_at: float) -> None:
if _raw_commit is not None:
_raw_commit(committed_at)
if _log_row_id is not None:
try:
get_db().execute(
"UPDATE event_log SET handled=1 WHERE id=?",
(int(_log_row_id),))
except Exception:
logger.exception(
"wfigs closed commit: event_log update failed")
data["_on_broadcast_committed"] = _on_commit
else:
# LEGACY verbatim (byte-for-byte identical live output).
data["category"] = "wildfire_closed"
data["_severity_override"] = "priority"
_attach_commit_handles(
data, irwin_id=irwin_id,
acres=fire_row["current_acres"],
contained_pct=fire_row["current_contained_pct"],
event_log_row_id=log_id)
data["_dedup_suffix"] = "closed"
return wire
return None
# ---- active incident ----
# v0.5.8b: log handled=0 initially. The commit callback UPDATEs this
# row to handled=1 if/when the dispatcher actually broadcasts -- if it
# drops (cold-start grace, staleness, cooldown, dedup), the row stays
# handled=0 and we can grep the event_log to find the suppressed events.
log_id = _log_event_returning_id(
conn, now=now, source="wfigs_incidents", category=category,
severity_word=severity_word, irwin_id=irwin_id,
subject=subject, handled=0,
table_name="fires", table_pk=irwin_id)
acres = normalized.get("acres")
contained_pct = normalized.get("contained_pct")
# Delegate the New/Update/suppress + age-gate + cooldown decision. The
# decider READS the fires row (pre-write) exactly as the original inline
# branch did; the inline INSERT/UPDATE below stays handler-owned.
gate = _gate_decide(canonical, source="wfigs", now=float(now))
# ---- inline state write (UNCONDITIONAL) -- INSERT or UPDATE current_*.
# Re-read the row (same pre-write state the decider saw) to pick the write.
row = conn.execute(
"SELECT last_broadcast_at FROM fires WHERE irwin_id = ?",
(irwin_id,)).fetchone()
if row is None:
conn.execute(
"INSERT INTO fires(irwin_id, incident_name, incident_type, "
"current_acres, current_contained_pct, status, lat, lon, "
"county, state, landclass, declared_at, last_event_at, "
"last_broadcast_at, last_broadcast_acres, last_broadcast_contained) "
"VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
(
irwin_id,
normalized.get("incident_name"),
normalized.get("incident_type"),
acres, contained_pct,
None, # status reserved
normalized.get("lat"), normalized.get("lon"),
normalized.get("county"), normalized.get("state"),
normalized.get("landclass"),
normalized.get("declared_at_epoch"),
now, # last_event_at
None, None, None, # last_broadcast_* explicitly NULL
),
)
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, normalized.get("lat"),
normalized.get("lon"), now, irwin_id),
)
if not gate.broadcast:
# Case-(iii) suppress (no forward change OR inside cooldown) ran the
# stale-fire cleanup in the original; the age-gate suppress did not.
if gate.lifecycle == "cooldown":
_cleanup_stale_fires(conn)
return None
# ---- broadcast: render the wire (back-compat) + stamp data.
prefix = "Update" if gate.lifecycle == "update" else "New"
wire = _render(normalized, prefix=prefix,
last_bcast_acres=gate.data_patch.get("last_bcast_acres"),
last_bcast_contained=gate.data_patch.get("last_bcast_contained"))
# Cutover gate keys on the category THIS broadcast carries: New first-sight
# is wildfire_declared, growth Update stays wildfire_incident.
_cut = (is_cutover("wildfire_incident") if gate.lifecycle == "update"
else is_cutover("wildfire_declared"))
if isinstance(data, dict):
if _cut:
# NEW PATH: formatter re-renders from canonical + data_patch.
data.update(canonical)
data.update(gate.data_patch)
data["_broadcast_audit"] = {"table": "fires", "pk": irwin_id}
_raw_commit = gate.commit
_log_row_id = log_id
def _on_commit(committed_at: float) -> None:
if _raw_commit is not None:
_raw_commit(committed_at)
if _log_row_id is not None:
try:
get_db().execute(
"UPDATE event_log SET handled=1 WHERE id=?",
(int(_log_row_id),))
except Exception:
logger.exception(
"wfigs commit: event_log update failed")
data["_on_broadcast_committed"] = _on_commit
else:
# LEGACY verbatim (byte-for-byte identical live output).
# New first-sight tags wildfire_declared; Update keeps the
# envelope-derived wildfire_incident (no category override).
if gate.lifecycle != "update":
data["category"] = "wildfire_declared"
data["_severity_override"] = "priority"
_attach_commit_handles(data, irwin_id=irwin_id,
acres=acres, contained_pct=contained_pct,
event_log_row_id=log_id)
return wire
# chore/ripout-2dii: the dead Central NATS-envelope entrypoint ``handle_wfigs``
# (envelope-driven New/Update/tombstone dispatch, event_log accounting, the
# `is_cutover` legacy-vs-new branch) has been REMOVED -- it had zero live
# production callers (Central's consumer that drove it is gone). The LIVE
# WFIGS path is `env/fires.py` (native adapter) -> `env/store.py::_emit_event`
# (forced onto `gating.fire.decide` + the shared fire formatter via
# `cutover.NATIVE_ALWAYS_DECIDE`, independent of this dead handler) -- see
# `tests/test_fire_native_growth.py`. `_render` (below) remains LIVE: it is
# called directly by `env.fire_fusion._handle_pass_boundary` on the FIRMS
# `wildfire_growth` path, and by `tests/test_fire_refactor.py` /
# `tests/test_wfigs_handler.py` as a byte-identity oracle for the shared fire
# formatter (`notifications/formatters/fire.py`).
# ---------- commit-callback factory ---------------------------------------
@ -419,47 +175,6 @@ def _attach_commit_handles(data: Optional[dict], *, irwin_id: str,
data["_dedup_suffix"] = f"{acres}|{contained_pct}"
# ---------- helpers -------------------------------------------------------
def _coerce_severity(sev: Any) -> Optional[str]:
if sev is None: return None
if isinstance(sev, str): return sev or None
try: return str(int(sev))
except (TypeError, ValueError): return str(sev)
def _log_event(conn, *, now, source, category, severity_word, irwin_id,
subject, handled, table_name, table_pk) -> None:
"""Insert an event_log row; void return (used for tombstones/perimeters
where the handled flag is fixed at write-time)."""
conn.execute(
"INSERT INTO event_log(received_at, source, category, severity_word, "
"event_id_external, nats_subject, handled, table_name, table_pk) "
"VALUES (?,?,?,?,?,?,?,?,?)",
(now, source, category, severity_word, irwin_id, subject,
int(bool(handled)), table_name, table_pk),
)
def _log_event_returning_id(conn, *, now, source, category, severity_word,
irwin_id, subject, handled, table_name,
table_pk) -> int:
"""Insert an event_log row and return its primary key id.
Used for active-incident logging where the commit callback updates
the same row to handled=1 once a broadcast actually goes out.
"""
cur = conn.execute(
"INSERT INTO event_log(received_at, source, category, severity_word, "
"event_id_external, nats_subject, handled, table_name, table_pk) "
"VALUES (?,?,?,?,?,?,?,?,?)",
(now, source, category, severity_word, irwin_id, subject,
int(bool(handled)), table_name, table_pk),
)
return int(cur.lastrowid)
# ---------- renderer ------------------------------------------------------

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

@ -122,19 +122,19 @@ 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)
# Phase-3c: FIRMS fusion broadcasts. `wildfire_spotting` / `wildfire_halted`
# have their own terse wires -> formatters/firms.py, registered below.
# `wildfire_growth` is NOT registered here (chore/ripout-2dii): its events are
# always fully precomposed (env/firms.py sets `_meshai_precomposed=True` +
# `title=<wire>`), and it is not in `cutover.NATIVE_ALWAYS_DECIDE`, so
# compose_mesh_message's formatter-invocation branch
# (notifications/renderers/composer.py:343-347) can never reach a registered
# formatter for this category live -- the precomposed-title bypass always wins
# first. wildfire_growth's live wire comes from `env.fire_render._render`,
# called directly by `env.fire_fusion._handle_pass_boundary`. A prior
# registration here (reusing the fire formatter) was dead code kept from the
# Phase-3c migration bake and has been removed; see TestRegistration in
# tests/test_firms_refactor.py for the regression guard.
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)

86
work/tests/test_fire_age_gate.py
View file

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

226
work/tests/test_fire_refactor.py
View file

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

150
work/tests/test_fire_tracker_phase1.py
View file

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

80
work/tests/test_fire_tracker_phase2.py
View file

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

105
work/tests/test_fire_tracker_phase3.py
View file

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

347
work/tests/test_firms_handler.py
View file

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

72
work/tests/test_firms_native_fusion.py
View file

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

188
work/tests/test_firms_refactor.py
View file

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

29
work/tests/test_tail_followups.py
View file

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

178
work/tests/test_tombstone_broadcast.py
View file

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

468
work/tests/test_wfigs_handler.py
View file

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