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

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

Browse source 
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>
This commit is contained in:
Matt Johnson 2026-07-18 04:03:47 +00:00
commit 2476b74908
1 changed files with 28 additions and 313 deletions
Download patch file Download diff file Expand all files Collapse all files

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