diff --git a/work/meshai/central/firms_handler.py b/work/meshai/central/firms_handler.py index b5e673b..e868622 100644 --- a/work/meshai/central/firms_handler.py +++ b/work/meshai/central/firms_handler.py @@ -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. diff --git a/work/tests/test_firms_cluster_f3.py b/work/tests/test_firms_cluster_f3.py index bdd3aa6..770c5a2 100644 --- a/work/tests/test_firms_cluster_f3.py +++ b/work/tests/test_firms_cluster_f3.py @@ -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 # ═════════════════════════════════════════════════════════════════════════════ diff --git a/work/tests/test_firms_native_fusion.py b/work/tests/test_firms_native_fusion.py index b4320b3..344947f 100644 --- a/work/tests/test_firms_native_fusion.py +++ b/work/tests/test_firms_native_fusion.py @@ -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)