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

fix(firms): first-fetch silent-seed suppresses fusion wires too (no cold-start)

Browse source 
Extend the FIRMS cold-start seed to suppress growth/spotting/halt fusion
broadcasts on the first fetch, not just clusters — enabling FIRMS must emit
zero broadcasts on the initial hotspot sweep. Persistence, attribution, and
dedup baselines still run during seed; only later new activity broadcasts.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Matt Johnson 2026-07-06 22:08:21 +00:00
commit 3129dd7250
3 changed files with 209 additions and 18 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -263,9 +263,13 @@ def _ingest_pixel_core(conn, *, lat, lon, acq_epoch, frp, confidence,
``fire_pixels`` / ``fire_passes`` / ``fires`` centroid+cursor and runs
the growth / spotting / halt fusion (honoring the Phase-3c deferred
latches). On an attribution MISS it runs the curated
unattributed-cluster path (``_maybe_emit_cluster``); ``seed=True``
suppresses those cluster broadcasts (cold-start silent-seed). A raw
per-pixel hotspot is NEVER broadcast.
unattributed-cluster path (``_maybe_emit_cluster``). ``seed=True``
(cold-start silent-seed) suppresses ALL broadcast output on the first
fetch -- cluster AND growth/spotting/halt fusion -- while persistence,
attribution, per-pass aggregation, centroid/cursor updates, and the
dedup latches (cluster_broadcast_at, halt_broadcast_at) still run so a
genuinely-new post-seed event broadcasts normally. A raw per-pixel
hotspot is NEVER broadcast.
Determinism: ``now`` is threaded explicitly; there is no hidden clock read.
@ -334,10 +338,14 @@ def ingest_hotspot_pixel(pixel: dict, *, now, seed=False) -> list[tuple[str, dic
raw pixel itself -- only these returned fusion/cluster wires.
``seed`` (cold-start silent-seed): on the FIRST fetch after boot, pass
``seed=True`` so any cluster that forms from the day's pre-existing hotspots
is stamped (never re-fires) but emits NOTHING -- no cold-start dump. Only
clusters formed by pixels arriving on a LATER (``seed=False``) fetch
broadcast. Persistence + attribution run regardless of ``seed``.
``seed=True`` so the day's pre-existing hotspots emit NOTHING -- no cold-start
dump of ANY kind. This covers BOTH the curated cluster path AND the
attributed growth / spotting / halt fusion: a cluster or idle-fire halt that
forms from pre-existing pixels is stamped (never re-fires); a growth/spotting
that a pre-existing pass boundary would have produced is withheld. Only
genuinely-new activity arriving on a LATER (``seed=False``) fetch broadcasts.
Persistence, attribution, per-pass aggregation, and dedup baselines run
regardless of ``seed``.
Determinism: ``now`` (epoch) is required and threaded straight through.
"""
@ -480,8 +488,12 @@ def _attribute_or_cluster(conn, *, pixel_row_id, lat, lon, acq_epoch,
"""Try attribution; on miss, run cluster check. Returns wire str | None.
Attribution ALWAYS runs first (a pixel inside a known WFIGS fire's spread
radius grows that fire and never forms a "new" cluster). ``seed`` only
affects the unattributed-cluster branch: see ``_maybe_emit_cluster``.
radius grows that fire and never forms a "new" cluster). ``seed`` (cold-start
silent-seed) suppresses ALL broadcast output on the first fetch -- the
unattributed-cluster branch (see ``_maybe_emit_cluster``) AND the attributed
fusion path (growth / spotting / halt). Persistence, attribution, per-pass
aggregation, centroid/cursor updates, and any dedup latches that must survive
to gate a LATER broadcast all still run; only the wire emission is withheld.
"""
global_default_mi = float(adapter_config.fires.spread_radius_mi_default)
# Conservative bbox prefilter: take the larger of the global default
@ -544,13 +556,13 @@ def _attribute_or_cluster(conn, *, pixel_row_id, lat, lon, acq_epoch,
wire = _handle_pass_boundary(
conn, irwin_id=chosen_irwin, pass_id=this_pass_id,
lat=lat, lon=lon, acq_epoch=acq_epoch, frp=frp,
data=data, now=now,
data=data, now=now, seed=seed,
)
if wire is not None:
return wire
# No growth broadcast; opportunistically run halt detector for
# OTHER fires that may have gone idle.
return _maybe_emit_halt(conn, data=data, now=now)
return _maybe_emit_halt(conn, data=data, now=now, seed=seed)
# 0 matches -- run cluster detection.
wire = _maybe_emit_cluster(
@ -559,7 +571,7 @@ def _attribute_or_cluster(conn, *, pixel_row_id, lat, lon, acq_epoch,
)
if wire is not None:
return wire
return _maybe_emit_halt(conn, data=data, now=now)
return _maybe_emit_halt(conn, data=data, now=now, seed=seed)
def _recompute_centroid_and_stamp(conn, irwin_id: str, *,
@ -743,8 +755,16 @@ def _pass_id(satellite, acq_epoch) -> str:
def _handle_pass_boundary(conn, *, irwin_id, pass_id, lat, lon,
acq_epoch, frp, data, now):
"""Maintain fire_passes row, detect boundary, fire growth on drift."""
acq_epoch, frp, data, now, seed=False):
"""Maintain fire_passes row, detect boundary, fire growth on drift.
``seed`` (cold-start silent-seed): all state -- the fire_passes upsert, the
``fires`` cursor/centroid update, and the prior-pass perimeter close -- runs
unconditionally so the growth/spotting dedup baselines are correct. Only the
wire EMISSION (growth here, spotting in ``_check_spotting``) is withheld on
the first fetch. Growth has no latch, so a genuinely-new post-seed pass
boundary fires normally on a later fetch.
"""
# (1) Recompute the pass aggregate from fire_pixels.
pass_rows = conn.execute(
"SELECT lat, lon, frp, acq_time FROM fire_pixels "
@ -852,7 +872,7 @@ def _handle_pass_boundary(conn, *, irwin_id, pass_id, lat, lon,
conn, irwin_id=irwin_id, pixel_lat=lat, pixel_lon=lon,
current_pass_id=pass_id,
incident_name=fires_row["incident_name"] or "(unnamed fire)",
data=data, now=now,
data=data, now=now, seed=seed,
)
if spotting_wire is not None:
return spotting_wire
@ -878,6 +898,16 @@ def _handle_pass_boundary(conn, *, irwin_id, pass_id, lat, lon,
if not gate.broadcast:
return None
# Cold-start silent-seed: the pass aggregate + fires cursor/centroid were
# already updated above (the growth baseline is intact), but on the first
# fetch after boot we emit NOTHING and leave `data` untouched. Growth has no
# latch, so a genuinely-new post-seed pass boundary broadcasts normally.
if seed:
logger.info(
"firms cold-start silent-seed: absorbed growth for %s "
"(state updated, no broadcast)", irwin_id)
return None
# Drift exceeded the threshold -- emit wildfire_growth via WFIGS renderer.
fire = conn.execute(
"SELECT incident_name, current_acres, current_contained_pct, "
@ -936,11 +966,21 @@ def _render_growth_wire(*, incident_name, direction, speed_mph,
)
def _maybe_emit_halt(conn, *, data, now):
def _maybe_emit_halt(conn, *, data, now, seed=False):
"""Find one fire matching the halt criteria, latch + broadcast.
Returns the wire string when a halt event fires; otherwise None.
``seed`` (cold-start silent-seed): halt is triggered opportunistically by
ANY pixel arrival (it scans all fires for idleness, independent of the
current pixel), so a fire that was already idle at boot WOULD otherwise
halt-broadcast on the next fetch's unrelated new pixel. To absorb that
pre-existing idle state -- mirroring how the cluster seed stamps
``cluster_broadcast_at`` -- the first fetch STILL latches
``halt_broadcast_at`` but emits NOTHING. Re-eligibility is unchanged: once
real post-seed activity advances ``last_pass_at`` past the stamp, the fire
can halt again legitimately.
Phase-3c: the DECISION (the halt-eligibility SELECT) is delegated to
``gating.firms.decide``; the wire is still rendered inline here for
byte-identity. On the not-cutover live path the eager
@ -959,6 +999,22 @@ def _maybe_emit_halt(conn, *, data, now):
p = gate.data_patch
irwin_id = p["irwin_id"]
# Cold-start silent-seed: stamp the latch (cutover-agnostic, eager) so this
# pre-existing idle fire is absorbed and never halt-broadcasts on a later
# fetch, but emit NOTHING and leave `data` untouched. It becomes halt-
# eligible again once real post-seed activity advances last_pass_at past
# this stamp -- exactly the cluster seed's "stamp-then-stay-silent" model.
if seed:
conn.execute(
"UPDATE fires SET halt_broadcast_at=? WHERE irwin_id=?",
(float(now), irwin_id),
)
logger.info(
"firms cold-start silent-seed: absorbed halt for %s "
"(latched, no broadcast)", irwin_id)
return None
name = p["incident_name"]
hours = p["hours"]
wire = f"🔥 {name} no growth in {hours}h"
@ -1062,8 +1118,18 @@ def _close_prev_perimeter(conn, irwin_id: str, prev_pass_id: str) -> None:
def _check_spotting(conn, *, irwin_id, pixel_lat, pixel_lon,
current_pass_id, incident_name, data, now):
"""Return spotting wire if criteria met, else None."""
current_pass_id, incident_name, data, now, seed=False):
"""Return spotting wire if criteria met, else None.
``seed`` (cold-start silent-seed): on the first fetch after boot, suppress
the wire WITHOUT stamping the ``last_spotting_broadcast_at`` cooldown latch.
Spotting is triggered by a specific NEW pixel lying outside the prior-pass
perimeter; a pre-existing pixel is a dedup no-op on later fetches, so it can
never re-trigger. Leaving the latch clear means a genuinely-new post-seed
spotting pixel still fires. Spotting writes no other state to preserve.
"""
if seed:
return None
threshold_mi = float(adapter_config.fires.spotting_distance_threshold_mi)
# Phase-3c: the cooldown gate (and its latch) moved into gating.firms.decide;
# the cooldown seconds are read there. The geometry below stays inline.

117
work/tests/test_firms_cluster_f3.py
View file

@ -232,6 +232,123 @@ def test_attribution_beats_clustering_for_known_fire():
assert "unattributed_hotspot_cluster" not in cats
# ═════════════════════════════════════════════════════════════════════════════
# 3b. Cold-start silent-seed suppresses the FUSION path too (growth/spotting/
# halt), not just clusters — enabling FIRMS must emit ZERO broadcasts on the
# first fetch even for a pre-existing attributed fire (e.g. MORA).
# ═════════════════════════════════════════════════════════════════════════════
# One VIIRS pass is a 90-min (5400 s) bucket (see _pass_id). Anchor pass A and
# pass B in DIFFERENT buckets so the second batch crosses a real pass boundary.
_PASS_A = _acq_epoch("2026-06-06", "1200") # bucket N
_PASS_B = _PASS_A + 6 * 3600 # +6h -> a later bucket
def _pixel_at(*, lat, lon, acq_epoch, frp=20.0):
return {"lat": lat, "lon": lon, "frp": frp, "confidence": "high",
"brightness": 320.0, "satellite": "N20", "acq_epoch": acq_epoch}
def test_cold_start_seed_suppresses_growth_fusion_but_persists():
"""First (seed) fetch: a pre-existing fire (MORA) with pixels that WOULD
produce a wildfire_growth boundary broadcast -> ZERO wires, yet the pixels
are persisted + attributed and the pass/centroid baseline is built."""
from meshai.persistence import get_db
mora_lat, mora_lon = 44.100, -115.600
_seed_fire(irwin_id="ID-MORA", lat=mora_lat, lon=mora_lon)
produced = []
# Pass A: 5 pixels around the anchor (same 90-min bucket).
for i in range(5):
produced += _feed(
_pixel_at(lat=mora_lat + 0.0001 * i, lon=mora_lon + 0.0001 * (i - 2),
acq_epoch=_PASS_A + i, frp=20.0 + i),
now=1780728000 + i, seed=True)
# Pass B: one pixel ~1 mi north in a LATER bucket -> a growth boundary that,
# on a non-seed fetch, WOULD broadcast wildfire_growth.
produced += _feed(
_pixel_at(lat=mora_lat + 1.0 / 69.0, lon=mora_lon, acq_epoch=_PASS_B),
now=1780728100, seed=True)
assert produced == [], "cold-start seed must suppress the growth fusion wire"
conn = get_db()
# Persistence + attribution still happened.
assert conn.execute("SELECT COUNT(*) FROM firms_pixels").fetchone()[0] == 6
assert conn.execute(
"SELECT COUNT(*) FROM fire_pixels").fetchone()[0] == 6
assert conn.execute(
"SELECT COUNT(*) FROM firms_pixels WHERE attributed_at IS NOT NULL"
).fetchone()[0] == 6
# Pass baseline built: the fires cursor advanced to pass B's bucket so a
# genuinely-new LATER pass can measure drift against it.
cursor = conn.execute(
"SELECT last_pass_id, current_centroid_lat FROM fires "
"WHERE irwin_id=?", ("ID-MORA",)).fetchone()
assert cursor["last_pass_id"] is not None
assert cursor["current_centroid_lat"] is not None
def test_growth_fires_on_later_fetch_after_seed():
"""After the cold-start seed built the pass baseline, a genuinely-new pass
on a LATER (seed=False) fetch DOES broadcast wildfire_growth."""
mora_lat, mora_lon = 44.100, -115.600
_seed_fire(irwin_id="ID-MORA", lat=mora_lat, lon=mora_lon)
# --- Cold-start seed: pass A + pass B, all silent. ---
for i in range(5):
assert _feed(
_pixel_at(lat=mora_lat + 0.0001 * i, lon=mora_lon + 0.0001 * (i - 2),
acq_epoch=_PASS_A + i, frp=20.0 + i),
now=1780728000 + i, seed=True) == []
assert _feed(
_pixel_at(lat=mora_lat + 1.0 / 69.0, lon=mora_lon, acq_epoch=_PASS_B),
now=1780728100, seed=True) == []
# --- Later real fetch (seed=False): a NEW pass C, another ~1 mi north of
# pass B -> a genuinely-new boundary drift -> wildfire_growth broadcasts. ---
pass_c = _PASS_B + 6 * 3600
out = _feed(
_pixel_at(lat=mora_lat + 2.0 / 69.0, lon=mora_lon, acq_epoch=pass_c),
now=1780750000, seed=False)
assert len(out) == 1, f"genuinely-new post-seed growth must fire: {out}"
wire, data = out[0]
assert data["category"] == "wildfire_growth"
assert data["_severity_override"] == "immediate"
assert wire.startswith("🔥 MORA")
def test_cold_start_seed_suppresses_halt_but_latches():
"""First (seed) fetch: an already-idle fire that WOULD halt-broadcast is
suppressed AND its halt latch is stamped, so it stays silent on later
unrelated fetches until real activity resumes."""
from meshai.persistence import get_db
now = 1780768800
idle_at = now - 14 * 3600
conn = get_db()
conn.execute(
"INSERT INTO fires(irwin_id, incident_name, lat, lon, last_event_at, "
"last_pass_id, last_pass_at) VALUES (?,?,?,?,?,?,?)",
("ID-IDLE", "Cold Fire", 42.5, -114.5, int(idle_at),
"N20-329627", float(idle_at)))
# A fresh unattributed pixel FAR from the idle fire on the cold-start fetch.
out = _feed(_pixel_at(lat=45.0, lon=-118.0, acq_epoch=_PASS_A), now=now,
seed=True)
assert out == [], "cold-start seed must suppress the halt wire"
latch = conn.execute(
"SELECT halt_broadcast_at FROM fires WHERE irwin_id=?",
("ID-IDLE",)).fetchone()[0]
assert latch == float(now), "seed still latches halt so it can't re-fire"
# A later unrelated fetch: the idle fire is latched -> still silent.
out2 = _feed(_pixel_at(lat=45.1, lon=-118.1, acq_epoch=_PASS_A + 60),
now=now + 120, seed=False)
assert out2 == [], "latched idle fire must not halt on a later fetch"
# ═════════════════════════════════════════════════════════════════════════════
# 4. Per-pixel silent: a raw hotspot never renders an Event
# ═════════════════════════════════════════════════════════════════════════════

8
work/tests/test_firms_native_fusion.py
View file

@ -308,6 +308,10 @@ class TestAdapterTickFusion:
_patch_fetch(monkeypatch, _csv(self._growth_rows(center_lat, center_lon)))
a = _adapter()
# Steady-state growth routing (not cold start): mark the adapter past its
# first-fetch silent-seed so this tick's growth broadcasts. Cold-start
# fusion suppression is covered in test_firms_cluster_f3.
a._firms_seeded = True
assert a.tick() is True
evts = a.get_events()
@ -352,6 +356,10 @@ class TestAdapterTickFusion:
# ingest would (store._ingest marks every source it touched).
store._seen = {}
store._seeded = {"firms", "firms_fusion"}
# Adapter-level cold-start seed is also a first-fetch silence gate; this
# steady-state test wants the growth wire, so mark the adapter seeded too
# (cold-start fusion suppression is covered in test_firms_cluster_f3).
a._firms_seeded = True
assert a.tick() is True
store._ingest("firms", a)