From c4706d3c921c5b4efaffcaf20da09ff94e259f6d Mon Sep 17 00:00:00 2001 From: Matt Johnson Date: Sun, 16 Aug 2026 01:19:46 +0000 Subject: [PATCH 1/9] fix(ipaws): route county-only alerts through evac-phase detection Adds meshai.notifications.evac_phase.detect_phase, wired into the IPAWS formatter/adapter, to classify READY/SET/GO evacuation phases from real FEMA IPAWS alert headline/CMAMtext strings. Includes explicit false-positive guards since a wrong GO detection would broadcast an evacuation order that was never issued. --- work/meshai/env/ipaws.py | 72 +++++++++++++- work/meshai/notifications/evac_phase.py | 95 ++++++++++++++++++ work/meshai/notifications/formatters/ipaws.py | 48 +++++++-- work/tests/test_adapter_ipaws.py | 98 ++++++++++++++++++- work/tests/test_evac_phase.py | 76 ++++++++++++++ 5 files changed, 376 insertions(+), 13 deletions(-) create mode 100644 work/meshai/notifications/evac_phase.py create mode 100644 work/tests/test_evac_phase.py diff --git a/work/meshai/env/ipaws.py b/work/meshai/env/ipaws.py index f5db495..e7b2284 100644 --- a/work/meshai/env/ipaws.py +++ b/work/meshai/env/ipaws.py @@ -104,6 +104,13 @@ def _norm_fips(value) -> str: class IPAWSAlertsAdapter: """FEMA IPAWS-OPEN EAS civil alerts — two-stage CAP poller.""" + # Stage-2 retry suppression. 401/403/404/410 are treated as durable (the + # alert body will not become fetchable), so we wait out the feed's rolling + # window; everything else (timeout, 5xx, 429, connection) is transient. + _STAGE2_FORBIDDEN_COOLDOWN = 21600 # 6h + _STAGE2_TRANSIENT_COOLDOWN = 300 # 5m + _STAGE2_FORBIDDEN_CODES = (401, 403, 404, 410) + def __init__(self, config: "IPAWSConfig", coverage: dict = None): self._base_url = _cfg_str(config, "base_url", DEFAULT_BASE_URL).rstrip("/") self._user_agent = getattr(config, "user_agent", "") or "meshai-ipaws/1.0" @@ -133,6 +140,11 @@ class IPAWSAlertsAdapter: self._last_error = None self._backoff_until = 0.0 self._is_loaded = False + # Negative cache: stage-2 CAP URL -> epoch after which a retry is + # allowed. Stops re-hammering FEMA for an alert listed in the feed whose + # detail endpoint keeps failing (e.g. a durable 403 on a COG-restricted + # alert). Pruned to the current feed each pass so it can't grow unbounded. + self._stage2_cooldown = {} # ── Polling ────────────────────────────────────────────────────────────── @@ -205,6 +217,8 @@ class IPAWSAlertsAdapter: self._consecutive_errors += 1 return False + now = time.time() + seen_urls = set() new_events = [] for entry in feed.findall(f"{{{_ATOM_NS}}}entry"): statefips = None @@ -221,18 +235,43 @@ class IPAWSAlertsAdapter: cap_url = self._stage2_url(href) if not cap_url: continue + seen_urls.add(cap_url) + + # ── Negative cache: skip URLs still within their failure cooldown ─ + retry_after = self._stage2_cooldown.get(cap_url) + if retry_after is not None and now < retry_after: + continue # ── Stage 2: full CAP document ─────────────────────────────────── try: cap_raw = self._get(cap_url) - except Exception as e: # noqa: BLE001 - logger.debug("IPAWS stage-2 fetch failed for %s: %s", cap_url, e) + except HTTPError as e: + cooldown = (self._STAGE2_FORBIDDEN_COOLDOWN + if e.code in self._STAGE2_FORBIDDEN_CODES + else self._STAGE2_TRANSIENT_COOLDOWN) + self._stage2_cooldown[cap_url] = now + cooldown + logger.debug("IPAWS stage-2 fetch failed for %s: HTTP %s " + "(cooldown %ds)", cap_url, e.code, cooldown) continue + except Exception as e: # noqa: BLE001 + self._stage2_cooldown[cap_url] = now + self._STAGE2_TRANSIENT_COOLDOWN + logger.debug("IPAWS stage-2 fetch failed for %s: %s (cooldown %ds)", + cap_url, e, self._STAGE2_TRANSIENT_COOLDOWN) + continue + + # Success — clear any stale cooldown for this URL. + self._stage2_cooldown.pop(cap_url, None) parsed = self._parse_cap(cap_raw, statefips) if parsed: new_events.append(parsed) + # Prune the negative cache to URLs still present in the feed, so it + # tracks the rolling window and never grows without bound. + self._stage2_cooldown = { + u: t for u, t in self._stage2_cooldown.items() if u in seen_urls + } + # Change detection on the active identifier set. old_ids = {e["event_id"] for e in self._events} new_ids = {e["event_id"] for e in new_events} @@ -297,6 +336,30 @@ class IPAWSAlertsAdapter: certainty = _t(info, "certainty") headline = _t(info, "headline") description = _t(info, "description") + instruction = _t(info, "instruction") + + # ── blocks: valueName -> value ──────────────────────────── + # Civil alerts carry their real public-facing text here (CMAMtext / + # CMAMlongtext are the agency's own purpose-written WEA copy) — this was + # previously discarded entirely. A repeated valueName (e.g. multiple + # BLOCKCHANNEL entries) becomes a list; defensive against malformed/ + # missing valueName or value children (skipped, never raises). + parameters: dict = {} + for param in info.findall(f"{{{_CAP_NS}}}parameter"): + vn = param.find(f"{{{_CAP_NS}}}valueName") + vv = param.find(f"{{{_CAP_NS}}}value") + name = vn.text.strip() if vn is not None and vn.text else "" + if not name: + continue + value = vv.text.strip() if vv is not None and vv.text else "" + if name in parameters: + existing = parameters[name] + if isinstance(existing, list): + existing.append(value) + else: + parameters[name] = [existing, value] + else: + parameters[name] = value # eventCode SAME value same_value = "" @@ -377,6 +440,8 @@ class IPAWSAlertsAdapter: "msgType": msg_type, "headline": headline, "description": description, + "instruction": instruction, + "parameters": parameters, "same_code": same_value, "area_desc": area_desc, "area_same_codes": area_same_codes, @@ -491,7 +556,8 @@ class IPAWSAlertsAdapter: "area_desc": area_desc, "geocoder": {"city": None, "county": area_desc, "state": None}, "description": raw.get("description", ""), - "parameters": {}, + "instruction": raw.get("instruction", ""), + "parameters": raw.get("parameters") or {}, "msgType": raw.get("msgType", "Alert"), "references": [], "category": category, diff --git a/work/meshai/notifications/evac_phase.py b/work/meshai/notifications/evac_phase.py new file mode 100644 index 0000000..ca92cef --- /dev/null +++ b/work/meshai/notifications/evac_phase.py @@ -0,0 +1,95 @@ +"""Idaho READY / SET / GO evacuation-phase detection from free-text CAP content. + +Real-world FEMA IPAWS civil alerts do NOT carry a machine-readable phase — +CAP ```` does not exist in the wild. The phase (READY / SET / +GO) instead shows up as free text inside ``headline``, ``description``, and +the ``CMAMtext``/``CMAMlongtext`` ```` values, phrased however the +issuing agency happened to write it ("Level 3 GO NOW", "Set to GO", +"Evacuation Warning", ...). This module scans that free text for the phrases +agencies actually use and returns the phase, or ``None`` when nothing +matches — callers must never guess a phase, since a false GO would broadcast +an evacuation order that was never issued. +""" +from __future__ import annotations + +import re + +# ── Strong multi-word phrases (checked case-insensitively) ─────────────────── +# ANY match sets that phase's hit flag. Order within a list is irrelevant — +# the final result is decided purely by GO > SET > READY precedence below, +# never by which phrase or text argument matched first. +_GO_PHRASES = [ + r"\bGO\s+NOW\b", + r"\bLEVEL\s+3\s+GO\b", + r"\bLEVEL\s+3\b", + r"\bLEVEL\s+III\b", + r"\bGO\s+EVACUATION\b", + r"\bIMMEDIATE\s+EVACUATION\b", + r"\bEVACUATE\s+NOW\b", + r"\bEVACUATION\s+ORDER\b", + r"\bSET\s+TO\s+GO\b", # "Set to GO" — the standalone-GO idiom, spelled out +] + +_SET_PHRASES = [ + r"\bLEVEL\s+2\b", + r"\bLEVEL\s+II\b", + r"\bPREPARE\s+TO\s+EVACUATE\b", + r"\bEVACUATION\s+WARNING\b", + r"\bBE\s+READY\s+TO\s+LEAVE\b", +] + +_READY_PHRASES = [ + r"\bLEVEL\s+1\b", + r"\bLEVEL\s+I\b", + r"\bEVACUATION\s+ADVISORY\b", +] + +_PRECEDENCE = ("GO", "SET", "READY") + +_PHRASE_RE = { + "GO": re.compile("|".join(_GO_PHRASES), re.IGNORECASE), + "SET": re.compile("|".join(_SET_PHRASES), re.IGNORECASE), + "READY": re.compile("|".join(_READY_PHRASES), re.IGNORECASE), +} + +# A bare level-word (GO/SET/READY) counts ONLY when it is: +# 1. standalone (word-boundaried) AND written in the exact uppercase form +# (narrative prose never shouts a whole word in caps: "go to the +# fairgrounds", "set to arrive", "via Go Creek Road" all fail this), AND +# 2. accompanied elsewhere in the same text by other alert/evacuation +# vocabulary, so a bare "GO"/"SET"/"READY" floating in unrelated text +# can't fire on its own. +# This is what lets "LEVEL I SET Alert" resolve as SET (word beats numeral — +# err upward) even though "LEVEL I" alone would read as READY. +_STANDALONE_TOKEN_RE = { + "GO": re.compile(r"\bGO\b"), + "SET": re.compile(r"\bSET\b"), + "READY": re.compile(r"\bREADY\b"), +} +_CONTEXT_CUE_RE = re.compile( + r"evacuat|level|alert|status|notice|prepar|leave|order|warning|advisory", + re.IGNORECASE, +) + + +def detect_phase(*texts: "str | None") -> "str | None": + """Scan the given texts for Idaho READY/SET/GO evacuation-phase language. + + All provided texts are combined and scanned together (case-insensitive + for the strong phrases). The HIGHEST phase found wins — GO > SET > READY + — never the first match. Returns None when nothing matches; never + guesses. + """ + combined = "\n".join(t for t in texts if t) + if not combined: + return None + + has_cue = bool(_CONTEXT_CUE_RE.search(combined)) + + for phase in _PRECEDENCE: + if _PHRASE_RE[phase].search(combined): + return phase + if has_cue and _STANDALONE_TOKEN_RE[phase].search(combined): + return phase + + return None diff --git a/work/meshai/notifications/formatters/ipaws.py b/work/meshai/notifications/formatters/ipaws.py index 91a695d..9293214 100644 --- a/work/meshai/notifications/formatters/ipaws.py +++ b/work/meshai/notifications/formatters/ipaws.py @@ -6,11 +6,19 @@ Message, AMBER, 911 outage, law enforcement, HazMat) from the canonical CAP Civil alerts carry their signal in the CAP ``headline`` (a plain human sentence), so — unlike the weather formatter, which parses structured -HAZARD.../IMPACT... blocks — this formatter is headline-forward: +HAZARD.../IMPACT... blocks — this formatter is headline-forward. Real CAP +data has no machine-readable Idaho READY/SET/GO evacuation phase (there is +no ```` in the wild); ``evac_phase.detect_phase`` scans the +headline/CMAMtext/description free text for the phrases agencies actually +use ("Level 3 GO NOW", "Evacuation Warning", ...) so the wire message can +lead with the phase instead of raw CAP jargon: - Line 1: {emoji} {prefix}{event} e.g. "🚨 Evacuation Immediate" + Line 1 (phase detected): {emoji} {prefix}{PHASE} — {action} + e.g. "🚨 GO — Leave now" + Line 1 (no phase found — unchanged fallback): + {emoji} {prefix}{event} e.g. "🚨 Evacuation Immediate" Line 2: {area}[ · Until {t} {tz}] areaDesc (first area) + expiry - Line 3: {headline} the operator's message + Line 3: {CMAMtext or headline} the agency's own public alert text Reuses ``event.data`` (canonical) + ``_budget.fit_to_budget``; does NOT touch the NWS formatter. ``now`` is a structural seam (not used — expiry is absolute). @@ -21,6 +29,7 @@ import zoneinfo from datetime import datetime from typing import TYPE_CHECKING +from meshai.notifications.evac_phase import detect_phase from meshai.notifications.formatters._budget import fit_to_budget if TYPE_CHECKING: @@ -37,6 +46,13 @@ _CATEGORY_EMOJI = { "emergency_civil": "⚠️", } +# Idaho READY/SET/GO — the one-line action tied to each detected phase. +_PHASE_ACTION = { + "READY": "Get prepared", + "SET": "Be ready to leave", + "GO": "Leave now", +} + def format(event: "Event", *, now: float, budget: int) -> str: """Render the IPAWS civil-alert wire string from canonical event.data. @@ -54,6 +70,11 @@ def format(event: "Event", *, now: float, budget: int) -> str: event_type = d.get("event") or "Emergency Alert" area_desc = d.get("area_desc") or "" headline = (d.get("headline") or "").strip() + description = d.get("description") or "" + parameters = d.get("parameters") or {} + cmam_text = parameters.get("CMAMtext") or "" + if isinstance(cmam_text, list): # defensive: a repeated valueName collapses to a list + cmam_text = cmam_text[0] if cmam_text else "" expires_epoch = d.get("expires_at") prefix = d.get("_ipaws_prefix") or "" category = d.get("category") or event.category or "emergency_civil" @@ -61,8 +82,20 @@ def format(event: "Event", *, now: float, budget: int) -> str: emoji = _CATEGORY_EMOJI.get(category, "⚠️") prefix_seg = f"{prefix}: " if prefix else "" - # Line 1: emoji + prefix + event type - line1 = f"{emoji} {prefix_seg}{event_type}" + # Line 1: lead with the Idaho READY/SET/GO phase when the free text + # actually names one; otherwise keep the raw CAP event string (safe + # fallback — never invent a phase that isn't there). + phase = detect_phase(headline, cmam_text, description) + if phase: + # A detected GO always gets the alarm emoji, regardless of CAP + # category — a confirmed "leave now" evacuation must never render + # with the softer category-based ⚠️ (e.g. emergency_civil). + if phase == "GO": + emoji = "🚨" + action = _PHASE_ACTION[phase] + line1 = f"{emoji} {prefix_seg}{phase} — {action}" + else: + line1 = f"{emoji} {prefix_seg}{event_type}" # Line 2: first area + optional expiry ("Until 4:54 PM MDT") area = (area_desc or "").split(";")[0].strip() @@ -79,8 +112,9 @@ def format(event: "Event", *, now: float, budget: int) -> str: else: line2 = area or time_seg - # Line 3: the headline (the actual civil message) - line3 = headline + # Line 3: the agency's own public alert text (CMAMtext — purpose-written + # for WEA, ~90 chars) when present; else fall back to the headline as before. + line3 = cmam_text.strip() or headline msg = "\n".join(ln for ln in (line1, line2, line3) if ln) return fit_to_budget(msg, budget) diff --git a/work/tests/test_adapter_ipaws.py b/work/tests/test_adapter_ipaws.py index 891f060..5009082 100644 --- a/work/tests/test_adapter_ipaws.py +++ b/work/tests/test_adapter_ipaws.py @@ -349,6 +349,12 @@ def test_decider_uses_own_table_not_nws(_isolate_meshai_db): # ============================================================ def test_formatter_renders_civil_alert(): + """Idaho CEM headline ("Wildfire Immediate Evacuation Alert") names the GO + phase explicitly ("Immediate Evacuation"), so line 1 leads with GO instead + of the raw CAP event string — the whole point of this feature. Category + is emergency_civil (normally ⚠️), but a detected GO forces the alarm + emoji regardless of category — a confirmed "leave now" must never render + with the softer warning icon.""" from meshai.notifications.formatters.ipaws import format as ipaws_format a = IPAWSAlertsAdapter(_config()) raw = a._parse_cap(_fx("eas_idaho_cem.xml"), "16") @@ -357,20 +363,73 @@ def test_formatter_renders_civil_alert(): wire = ipaws_format(ev, now=1000.0, budget=200) assert len(wire) <= 200 lines = wire.split("\n") - assert lines[0].startswith("⚠️") # civil -> warning emoji - assert "Civil Emergency Message" in lines[0] + assert lines[0].startswith("🚨") # GO overrides civil's ⚠️ -> alarm emoji + assert "GO" in lines[0] and "Leave now" in lines[0] assert "Boundary County" in wire assert "evacuation" in wire.lower() # headline carried the signal def test_formatter_evacuation_emoji(): + """Oregon EVI has no phase language in its headline, but its CMAMtext + parameter ("Level 3 GO NOW evacuation notice...") does, and that text is + now parsed and preferred for line 3 — so GO must win here too.""" from meshai.notifications.formatters.ipaws import format as ipaws_format a = IPAWSAlertsAdapter(_config()) raw = a._parse_cap(_fx("eas_oregon_evi.xml"), "41") ev = a.to_event(raw) wire = ipaws_format(ev, now=1000.0, budget=200) assert wire.startswith("🚨") # evacuation -> alarm emoji - assert "Evacuation Immediate" in wire + lines = wire.split("\n") + assert "GO" in lines[0] and "Leave now" in lines[0] + assert "Level 3 GO NOW" in wire # CMAMtext carried into line 3 + + +def test_formatter_full_three_line_go_rendering(): + """Full 3-line render for a real GO alert (Oregon EVI): phase-led line 1, + area+expiry line 2, CMAMtext line 3 — exact text, not just substrings.""" + from meshai.notifications.formatters.ipaws import format as ipaws_format + a = IPAWSAlertsAdapter(_config()) + raw = a._parse_cap(_fx("eas_oregon_evi.xml"), "41") + ev = a.to_event(raw) + ev.data["_ipaws_prefix"] = "" + wire = ipaws_format(ev, now=1000.0, budget=200) + lines = wire.split("\n") + assert lines[0] == "🚨 GO — Leave now" + assert lines[1] == "Jackson County · Until 8:50 AM MDT" + assert lines[2] == ( + "Wildfire Alert- Level 3 GO NOW evacuation notice is UPGRADED for JAC-126" + ) + + +def test_formatter_no_phase_fallback_unchanged(): + """When no READY/SET/GO language is present anywhere in headline/CMAMtext/ + description, line 1 keeps the CURRENT raw-event-string behaviour exactly — + the safe fallback this feature must never break.""" + from meshai.notifications.formatters.ipaws import format as ipaws_format + from meshai.notifications.events import make_event + + canonical = _canonical( + event="Civil Emergency Message", + headline="Boil water advisory issued for the district", + description="A water main break has contaminated the supply.", + parameters={}, + expires_at=None, + ) + ev = make_event( + source="ipaws", + category="emergency_civil", + severity="priority", + title=canonical["headline"], + summary=canonical["headline"], + body=canonical["description"], + data=canonical, + ) + ev.data["_ipaws_prefix"] = "" + wire = ipaws_format(ev, now=1000.0, budget=200) + lines = wire.split("\n") + assert lines[0] == "⚠️ Civil Emergency Message" # unchanged fallback (no phase) + assert lines[1] == "Boundary County" + assert lines[2] == "Boil water advisory issued for the district" # ============================================================ @@ -386,3 +445,36 @@ def test_coverage_state_fips_for_bbox(): resolved = resolve_adapter_coverage("ipaws", idaho_bbox, "native") assert "16" in resolved["state_fips"] assert resolved["bbox"] == [round(c, 6) for c in idaho_bbox] + + +# ============================================================ +# stage-2 failure negative cache (no re-hammering FEMA) +# ============================================================ + +def test_stage2_forbidden_is_negative_cached(): + """A 403 on a stage-2 detail URL is remembered so the next poll tick does + NOT re-fetch it, while a sibling URL that succeeds is fetched every pass.""" + from urllib.error import HTTPError + + counts: dict[str, int] = {} + + def _urlopen(req, timeout=None): + url = req.full_url + counts[url] = counts.get(url, 0) + 1 + if url.endswith("/feed"): + return _FakeResp(_fx("feed.xml")) + if url.endswith("/eas/300130859542"): # Idaho CEM -> forbidden + raise HTTPError(url, 403, "Forbidden", {}, None) + if url.endswith("/eas/300130856756"): # Oregon EVI -> ok + return _FakeResp(_fx("eas_oregon_evi.xml")) + raise AssertionError(f"unexpected fetch: {url}") + + a = IPAWSAlertsAdapter(_config()) + with patch("meshai.env.ipaws.urlopen", _urlopen): + a._fetch() + a._fetch() + + cem = next(u for u in counts if u.endswith("/eas/300130859542")) + evi = next(u for u in counts if u.endswith("/eas/300130856756")) + assert counts[cem] == 1 # 403 -> negative-cached, not re-fetched + assert counts[evi] == 2 # success -> fetched on each pass diff --git a/work/tests/test_evac_phase.py b/work/tests/test_evac_phase.py new file mode 100644 index 0000000..7b8fd45 --- /dev/null +++ b/work/tests/test_evac_phase.py @@ -0,0 +1,76 @@ +"""Tests for meshai.notifications.evac_phase.detect_phase. + +Cases are drawn from REAL FEMA IPAWS alert headline/CMAMtext/description +strings (see tests/fixtures/ipaws/) plus explicit false-positive guards, since +a wrong GO detection would broadcast an evacuation order that was never +issued. +""" +from __future__ import annotations + +import pytest + +from meshai.notifications.evac_phase import detect_phase + + +# ============================================================ +# real strings -> expected phase +# ============================================================ + +@pytest.mark.parametrize("text, expected", [ + ("LEVEL 1 READY - Evacuation Status", "READY"), + ("Level 2 Set Alert", "SET"), + ("LEVEL I SET Alert", "SET"), # word beats numeral; err upward + ("Level 3 - Go Now", "GO"), + ("Level 3- Go Now", "GO"), + ("Ohio Gulch GO Evacuation Status", "GO"), + ("Indian Creek Set to GO.", "GO"), # highest wins, NOT SET + ("Immediate Evacuation", "GO"), + ("Prepare to Evacuate", "SET"), + ("BRUSH FIRE", None), + ("Owyhee County Closure Area", None), + ("Endangered Missing Person Alert", None), + ( + "Jackson County Sheriff's Office- Level 3 GO NOW evacuation notice " + "UPGRADED for JAC-126", + "GO", + ), +]) +def test_detect_phase_real_strings(text, expected): + assert detect_phase(text) == expected + + +# ============================================================ +# false-positive guards — bare lowercase / narrative usage must NOT match +# ============================================================ + +@pytest.mark.parametrize("text", [ + "residents should go to the Blaine County Fairgrounds", + "crews are set to arrive by 0600", + "evacuate via Go Creek Road", +]) +def test_detect_phase_false_positive_guards(text): + assert detect_phase(text) is None + + +# ============================================================ +# precedence + multi-arg scanning +# ============================================================ + +def test_precedence_highest_always_wins_regardless_of_arg_order(): + # SET language in the first text, GO language in the second — GO must win. + assert detect_phase("Prepare to Evacuate", "Level 3 - Go Now") == "GO" + # Same phrases, arguments reversed — still GO (order-independent). + assert detect_phase("Level 3 - Go Now", "Prepare to Evacuate") == "GO" + + +def test_detect_phase_scans_all_texts_together(): + # No single text alone carries a phase; combined they do (SET here, + # since "LEVEL 2" is an explicit SET phrase and no GO phrase is present). + assert detect_phase("Jackson County Sheriff's Office", "Level 2 Set Alert") == "SET" + + +def test_detect_phase_none_texts_and_empty_input_are_safe(): + assert detect_phase(None, None) is None + assert detect_phase() is None + assert detect_phase("") is None + assert detect_phase(None, "BRUSH FIRE", None) is None From 50553b26f5aa6a1be17d3c82a22a57e996f4c6f6 Mon Sep 17 00:00:00 2001 From: Matt Johnson Date: Sun, 16 Aug 2026 01:19:49 +0000 Subject: [PATCH 2/9] fix(roads511): distinguish full closures from partial restrictions Adds explicit full-closure phrase matching (with partial-restriction phrases as vetoes) for sub_type mapping in to_event(), so wording like "All Shoulders Closed" or "Right Lane Closed" no longer gets broadcast as a full road_closed event. Existing is_closure severity/summary logic is untouched; the new is_full_closure_flag/_has_full_closure_language helpers are additive and scoped to sub_type only. --- work/meshai/env/roads511.py | 121 +++++++++- work/tests/test_adapter_roads511.py | 328 ++++++++++++++++++++++++++++ 2 files changed, 444 insertions(+), 5 deletions(-) diff --git a/work/meshai/env/roads511.py b/work/meshai/env/roads511.py index ba90bde..a57e5b4 100644 --- a/work/meshai/env/roads511.py +++ b/work/meshai/env/roads511.py @@ -21,6 +21,53 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) +# Explicit full-road-closure language, matched case-insensitively. Used +# ONLY by to_event()'s sub_type mapping (branch a) to decide "road_closed" +# vs. "road_works" — distinct from (and stricter than) the adapter's +# existing `is_closure` property, which loosely matches any "closed" +# substring and therefore also flags partial-restriction wording (e.g. +# "All Shoulders Closed") that the upstream owner explicitly does not want +# broadcast as a closure. +_FULL_CLOSURE_PHRASES = ( + "all lanes closed", + "road closed", + "highway closed", + "fully closed", + "closed in both directions", +) + +# Partial-restriction phrases that must NEVER be counted as a full closure. +# Checked first so one of these vetoes a match even if closure wording +# happens to appear elsewhere in the same description — e.g. "all +# shoulders closed" must not fall through to a naive "closed" test, and +# "one/right/left lane closed" must not be confused with "all lanes +# closed" (note the singular "lane closed" vs. plural "lanes closed"). +_PARTIAL_RESTRICTION_PHRASES = ( + "shoulder closed", + "shoulders closed", + "lane closed", + "ramp closed", + "alternating", + "lane restriction", +) + + +def _has_full_closure_language(description: str) -> bool: + """True if `description` contains explicit full-closure phrasing. + + Case-insensitive. Partial-restriction phrases (shoulder/lane/ramp + closures, alternating traffic) are checked first and veto a match, so + they never register as a full closure regardless of other wording + present. + """ + text = (description or "").lower() + + if any(phrase in text for phrase in _PARTIAL_RESTRICTION_PHRASES): + return False + + return any(phrase in text for phrase in _FULL_CLOSURE_PHRASES) + + class Roads511Adapter: """511 road conditions polling adapter.""" @@ -284,12 +331,21 @@ class Roads511Adapter: lat = loc.get("latitude") or loc.get("lat") lon = loc.get("longitude") or loc.get("lon") or loc.get("lng") - # Check closure status - is_closure = ( + # Raw upstream full-closure signal, captured on its own (no + # description heuristic mixed in) so to_event() can build a + # stricter full-closure determination for sub_type mapping + # without touching is_closure below, which severity/summary + # logic elsewhere already depends on. + is_full_closure_flag = bool( item.get("IsFullClosure") or item.get("is_full_closure") or item.get("fullClosure") or - item.get("closed") or + item.get("closed") + ) + + # Check closure status + is_closure = ( + is_full_closure_flag or "closure" in str(event_type).lower() or "closed" in str(description).lower() ) @@ -323,6 +379,19 @@ class Roads511Adapter: None ) + # Granular ITD 511 v2 fields (roadwork vs. crash vs. closure + # discrimination). Defensive: absent on other states' 511 feeds. + event_sub_type = ( + item.get("EventSubType") or + item.get("event_sub_type") or + "" + ) + cause = ( + item.get("Cause") or + item.get("cause") or + None + ) + # Default 6 hour TTL, refreshed every tick expires = now + 21600 @@ -346,7 +415,10 @@ class Roads511Adapter: "properties": { "roadway": roadway, "is_closure": bool(is_closure), + "is_full_closure_flag": is_full_closure_flag, "last_updated": last_updated, + "event_sub_type": event_sub_type, + "cause": cause, }, } @@ -411,10 +483,49 @@ class Roads511Adapter: # NOTE: Central-era rows used source 'itd_511' with 'idaho_511:event:*' # ids — a different keyspace the pre-seed intentionally does not cover. _external_id = evt.get("external_id") or event_id + + # sub_type mapping: distinguish routine roadwork (suppressed by + # gating/incident.py's work-zone rule) from genuine news (crashes, + # closures, hazards). A construction-caused FULL closure is still + # real news and must broadcast, so a full closure wins regardless + # of EventType. Precedence: + # 1. full closure (raw flag OR explicit closure wording, + # see _has_full_closure_language) -> road_closed (always) + # 2. EventType roadwork / + # EventSubType has "construction" -> road_works (suppressed) + # 3. EventType accidentsAndIncidents -> incident + # 4. EventType closures -> road_closed + # 5. else (specialEvents, unknown) -> incident (fail open) + # + # NOTE: branch 1 deliberately does NOT reuse `_is_closure` above. + # `_is_closure` is a loose OR-chain (also used for severity and + # the summary text) that matches any "closed" substring in the + # description, including partial-restriction wording like "All + # Shoulders Closed" — routine construction the owner explicitly + # does not want broadcast. `_full_closure_for_mapping` is a + # stricter, mapping-only determination. + _event_type_val = str(evt.get("event_type") or "").strip().lower() + _event_sub_type_val = str(props.get("event_sub_type") or "").strip().lower() + _cause = props.get("cause") + _full_closure_for_mapping = bool( + props.get("is_full_closure_flag") + ) or _has_full_closure_language(_desc) + + if _full_closure_for_mapping: + _sub_type = "road_closed" + elif _event_type_val == "roadwork" or "construction" in _event_sub_type_val: + _sub_type = "road_works" + elif _event_type_val == "accidentsandincidents": + _sub_type = "incident" + elif _event_type_val == "closures": + _sub_type = "road_closed" + else: + _sub_type = "incident" + canonical_data = { "external_id": _external_id, # 511_{itd_id}; enables durable pre-seed "source": "511", - "sub_type": "road_closed" if _is_closure else "incident", + "sub_type": _sub_type, "road": _roadway or None, "direction": None, # not structured in native 511 feed "from_loc": None, @@ -423,7 +534,7 @@ class Roads511Adapter: "mile_end": None, "mile_marker": None, "lanes_affected": None, - "cause": None, + "cause": _cause, "comment": _desc[:200] if _desc else title, "impact": "all lanes closed" if _is_closure else None, "county": None, diff --git a/work/tests/test_adapter_roads511.py b/work/tests/test_adapter_roads511.py index 4322c11..c76e74d 100644 --- a/work/tests/test_adapter_roads511.py +++ b/work/tests/test_adapter_roads511.py @@ -200,3 +200,331 @@ def test_to_event_missing_properties_returns_event(adapter): def test_to_event_does_not_raise_on_corrupted_dict(adapter): """Corrupted dict returns None without raising.""" assert adapter.to_event({"garbage": True}) is None + + +# ============================================================ +# EventType / EventSubType / Cause -> sub_type MAPPING TESTS +# +# Drives real upstream-shaped ITD 511 v2 payloads through the actual +# _parse_event() -> to_event() pipeline (not the make_511_event() stored- +# dict helper) so the EventSubType/Cause capture at parse time and the +# sub_type precedence in to_event() are both exercised end to end. +# ============================================================ + +def _itd_raw( + event_id="RW-1001", + event_type="roadwork", + event_sub_type="roadConstruction", + cause="Construction", + description="Paving operations, expect delays", + roadway="I-84", + is_full_closure=False, + lat=43.5, + lon=-116.2, +): + """Realistic raw ITD 511 v2 API item shape (PascalCase fields).""" + return { + "EventId": event_id, + "EventType": event_type, + "EventSubType": event_sub_type, + "Cause": cause, + "RoadwayName": roadway, + "Description": description, + "Latitude": lat, + "Longitude": lon, + "IsFullClosure": is_full_closure, + "LastUpdated": "2026-08-01T00:00:00Z", + } + + +def _through_pipeline(adapter, raw_item): + """Parse a raw ITD item then translate to a pipeline Event, mirroring + the real adapter flow (_fetch -> _parse_event -> get_events -> to_event).""" + stored = adapter._parse_event(raw_item, time.time()) + assert stored is not None + return adapter.to_event(stored) + + +def test_mapping_roadwork_not_full_closure_is_road_works(adapter): + """Routine roadwork, not a full closure -> sub_type 'road_works' (suppressed).""" + raw = _itd_raw(event_type="roadwork", is_full_closure=False) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "road_works" + + +def test_mapping_roadwork_with_full_closure_is_road_closed(adapter): + """Construction-caused FULL closure is still genuine news -> 'road_closed', + and must NOT be suppressed like routine roadwork.""" + raw = _itd_raw( + event_type="roadwork", + event_sub_type="bridgeConstruction", + description="Bridge replacement in progress", + is_full_closure=True, + ) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "road_closed" + + +def test_mapping_accidents_and_incidents_is_incident(adapter): + """EventType accidentsAndIncidents -> sub_type 'incident'.""" + raw = _itd_raw( + event_id="INC-2001", + event_type="accidentsAndIncidents", + event_sub_type="crash", + cause="Collision", + description="Two vehicle collision blocking right lane", + is_full_closure=False, + ) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "incident" + + +def test_mapping_closures_is_road_closed(adapter): + """EventType closures -> sub_type 'road_closed'.""" + raw = _itd_raw( + event_id="CL-3001", + event_type="closures", + event_sub_type="bridgeClosure", + cause="Maintenance", + description="Bridge out of service", + is_full_closure=False, + ) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "road_closed" + + +def test_mapping_special_events_is_incident(adapter): + """EventType specialEvents -> sub_type 'incident' (not suppressed).""" + raw = _itd_raw( + event_id="SE-4001", + event_type="specialEvents", + event_sub_type="parade", + cause=None, + description="Downtown parade route", + is_full_closure=False, + ) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "incident" + + +def test_mapping_unknown_or_missing_event_type_is_incident(adapter): + """Missing/unrecognized EventType fails open to 'incident' (not suppressed).""" + raw = { + "EventId": "UNK-5001", + "RoadwayName": "SH-21", + "Description": "Unusual event", + "Latitude": 43.5, + "Longitude": -115.5, + } + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "incident" + + +def test_mapping_event_subtype_construction_overrides_non_roadwork_eventtype(adapter): + """EventSubType containing 'construction' forces 'road_works' even when + EventType itself is not literally 'roadwork'.""" + raw = _itd_raw( + event_id="SE-6001", + event_type="specialEvents", + event_sub_type="bridgeConstruction", + cause="Construction", + description="Bridge deck construction nearby", + is_full_closure=False, + ) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "road_works" + + +def test_mapping_cause_is_carried_through_not_none(adapter): + """Cause from the upstream payload reaches canonical_data, not hardcoded None.""" + raw = _itd_raw(cause="Weather") + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["cause"] == "Weather" + assert event.data["cause"] is not None + + +def test_mapping_missing_cause_is_none(adapter): + """No Cause field upstream -> canonical cause is None, not a crash.""" + raw = _itd_raw(cause=None) + del raw["Cause"] + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["cause"] is None + + +# ============================================================ +# END-TO-END GATING TEST — proves a real roadwork event, run through the +# actual adapter pipeline, is suppressed by gating/incident.py's work-zone +# rule, while a real crash event from the same pipeline still broadcasts. +# ============================================================ + +def test_roadwork_event_suppressed_by_incident_gate_end_to_end(adapter): + """A real ITD roadwork payload, parsed and translated by the adapter, + must be suppressed at the gating layer (not just mapped to the right + sub_type in isolation).""" + from meshai.notifications.gating.incident import decide + + raw = _itd_raw(event_id="RW-7001", event_type="roadwork", is_full_closure=False) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "road_works" + + result = decide(dict(event.data), source="511", now=time.time()) + assert result.broadcast is False + assert result.lifecycle == "suppress" + + +def test_crash_event_still_broadcasts_through_incident_gate_end_to_end(adapter): + """A real ITD crash payload, parsed and translated by the adapter, + still broadcasts (gating layer is unaffected for non-work-zone sub_types).""" + from meshai.notifications.gating.incident import decide + + raw = _itd_raw( + event_id="INC-7002", + event_type="accidentsAndIncidents", + event_sub_type="crash", + description="Vehicle collision, right lane blocked", + is_full_closure=False, + ) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "incident" + + +# ============================================================ +# FULL-CLOSURE vs. PARTIAL-RESTRICTION LANGUAGE TESTS +# +# The loose `is_closure` property (also used for severity/summary) matches +# any "closed" substring, including partial-restriction wording like "All +# Shoulders Closed" that ITD explicitly does not want broadcast. The +# sub_type mapping's full-closure branch must use a stricter, +# mapping-only determination instead. +# ============================================================ + +def test_all_shoulders_closed_roadwork_is_road_works(adapter): + """'All Shoulders Closed' is a partial restriction, not a full closure + -> stays 'road_works' (suppressed), despite containing 'closed'.""" + raw = _itd_raw( + event_id="RW-8001", + event_type="roadwork", + description="All Shoulders Closed", + is_full_closure=False, + ) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "road_works" + + +@pytest.mark.parametrize( + "description", + ["One lane closed", "Right lane closed", "Left lane closed"], +) +def test_single_lane_closed_roadwork_is_road_works(adapter, description): + """Single-lane restriction wording -> stays 'road_works' (suppressed).""" + raw = _itd_raw( + event_id="RW-8002", + event_type="roadwork", + description=description, + is_full_closure=False, + ) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "road_works" + + +def test_ramp_closed_roadwork_is_road_works(adapter): + """'Ramp closed' is a partial restriction -> stays 'road_works' (suppressed).""" + raw = _itd_raw( + event_id="RW-8003", + event_type="roadwork", + description="Ramp closed for repaving", + is_full_closure=False, + ) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "road_works" + + +def test_all_lanes_closed_roadwork_is_road_closed(adapter): + """'All lanes closed' is a genuine full closure -> 'road_closed', and + must STILL BROADCAST (this is the safety case: a real closure phrased + as roadwork must not be suppressed by the work-zone gate).""" + from meshai.notifications.gating.incident import decide + + raw = _itd_raw( + event_id="RW-8004", + event_type="roadwork", + description="All lanes closed for bridge demolition", + is_full_closure=False, + ) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "road_closed" + + result = decide(dict(event.data), source="511", now=time.time()) + assert result.broadcast is True + assert result.lifecycle != "suppress" + + +def test_is_full_closure_flag_true_no_closure_words_is_road_closed(adapter): + """IsFullClosure=true + roadwork, with a description containing no + closure wording at all -> still 'road_closed' via the raw flag.""" + raw = _itd_raw( + event_id="RW-8005", + event_type="roadwork", + description="Bridge deck replacement in progress", + is_full_closure=True, + ) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "road_closed" + + +def test_road_closed_due_to_rock_slide_incident_is_road_closed(adapter): + """Explicit full-closure wording on an accidentsAndIncidents event + still wins the mapping -> 'road_closed'.""" + raw = _itd_raw( + event_id="INC-8006", + event_type="accidentsAndIncidents", + event_sub_type="hazard", + cause="Rock slide", + description="Road closed due to rock slide", + is_full_closure=False, + ) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "road_closed" + + +def test_lane_restrictions_roadwork_is_road_works(adapter): + """'Lane restrictions' wording -> stays 'road_works' (suppressed).""" + raw = _itd_raw( + event_id="RW-8007", + event_type="roadwork", + description="Lane restrictions in effect through Friday", + is_full_closure=False, + ) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "road_works" + + +def test_alternating_traffic_roadwork_is_road_works(adapter): + """'Alternating' one-lane traffic control -> stays 'road_works' (suppressed).""" + raw = _itd_raw( + event_id="RW-8008", + event_type="roadwork", + description="Alternating traffic controlled by flaggers", + is_full_closure=False, + ) + event = _through_pipeline(adapter, raw) + assert event is not None + assert event.data["sub_type"] == "road_works" From 738ffc71c32535ee5cfd4654a61024eff979fca1 Mon Sep 17 00:00:00 2001 From: Matt Johnson Date: Sun, 16 Aug 2026 01:20:06 +0000 Subject: [PATCH 3/9] fix(usgs_quake): stop dropping first-sighted quakes at the freshness gate The USGS feed is a rolling PAST-DAY feed (2.5_day.geojson), so a genuinely first-seen quake's age-at-ingest routinely exceeds the generic per-toggle freshness_seconds (600s) window, silently dropping it downstream of decide()/DB-insert (last_broadcast_at stayed NULL on every sampled first-sighting row). - dispatcher.py: earthquake_event gets its own adapter_config-backed freshness override (adapter_config.usgs_quake.freshness_seconds, default 3600s), scoped to the CATEGORY (not the "seismic" toggle, which also carries hydro stream_flood_warning/stream_high_water and must keep the generic per-toggle freshness). - usgs_quake.py: to_event() no longer passes the adapter's fixed freshness in, deferring to the dispatcher override. - adapter_config/defaults.py: adds the freshness_seconds field/default for usgs_quake. --- work/meshai/adapter_config/defaults.py | 8 +- work/meshai/env/usgs_quake.py | 9 +- .../notifications/pipeline/dispatcher.py | 12 + work/tests/test_adapter_config_api.py | 1 + work/tests/test_adapter_usgs_quake.py | 22 +- ...test_quake_freshness_and_region_tagging.py | 212 ++++++++++++++++++ 6 files changed, 261 insertions(+), 3 deletions(-) create mode 100644 work/tests/test_quake_freshness_and_region_tagging.py diff --git a/work/meshai/adapter_config/defaults.py b/work/meshai/adapter_config/defaults.py index b1dec70..f91c07a 100644 --- a/work/meshai/adapter_config/defaults.py +++ b/work/meshai/adapter_config/defaults.py @@ -85,7 +85,8 @@ REGISTRY: dict[tuple[str, str], dict[str, Any]] = { }, # ================================================================= - # USGS_QUAKE -- 6 settings (regional geography + 3 mag floors + PAGER set) + # USGS_QUAKE -- 7 settings (regional geography + 3 mag floors + PAGER set + # + freshness gate) # ================================================================= ("usgs_quake", "regional_centroid"): { "default": [44.36, -114.61], # quake_handler.py:36-37 (Idaho centroid) @@ -117,6 +118,11 @@ REGISTRY: dict[tuple[str, str], dict[str, Any]] = { "type": "float", "description": "Magnitude floor for the visual escalation emoji.", }, + ("usgs_quake", "freshness_seconds"): { + "default": 3600, + "type": "int", + "description": "Staleness gate for earthquake_event broadcasts (age = now - quake origin time), read by the dispatcher instead of the generic per-toggle freshness. The upstream feed is a rolling PAST-DAY feed, so quakes are routinely detected 10min-20h after origin; 1 hour keeps quakes broadcast-worthy (still situationally relevant) while excluding the day-feed's stale backlog. 0 = disabled (not recommended -- would readmit day-old quakes).", + }, # ================================================================= # SWPC -- 3 settings (three storm-tier broadcast floors) diff --git a/work/meshai/env/usgs_quake.py b/work/meshai/env/usgs_quake.py index 943e1af..264841d 100644 --- a/work/meshai/env/usgs_quake.py +++ b/work/meshai/env/usgs_quake.py @@ -246,7 +246,14 @@ class USGSQuakeAdapter: expires=evt.get("expires"), lat=lat, lon=lon, - region=evt.get("region"), + # Deliberately NOT passing region=evt.get("region") here: the + # raw dict's region is a fixed config default (self._region, + # "magic_valley"), which doesn't match any region_routes cell + # key. Event.region is documented as "set by region tagger" + # (notifications/events.py) -- leaving it unset lets + # CoverageFilter's geometry-based tagging (event.lat/lon vs + # the named coverage areas) populate the real region name + # from the quake's actual coordinates instead. group_key=event_id, inhibit_keys=[event_id], data=canonical_data, diff --git a/work/meshai/notifications/pipeline/dispatcher.py b/work/meshai/notifications/pipeline/dispatcher.py index aafc5f6..3034af2 100644 --- a/work/meshai/notifications/pipeline/dispatcher.py +++ b/work/meshai/notifications/pipeline/dispatcher.py @@ -359,6 +359,18 @@ class Dispatcher: # v0.6-3b: fire toggle uses wfigs adapter_config freshness (0 = disabled) if fam == "fire": freshness_s = int(adapter_config.wfigs.freshness_seconds) + elif event.category == "earthquake_event": + # The upstream USGS feed is a rolling PAST-DAY feed + # (2.5_day.geojson), so a quake's age-at-ingest routinely exceeds + # the generic 600s toggle window even for a genuinely first-seen + # event -- every native quake was being silently dropped here + # (measured age-at-ingest 548s-72169s across 12 sampled quakes, + # confirmed against quake_events: 12/12 first-sighted rows with + # last_broadcast_at=NULL). Scoped to the earthquake_event + # CATEGORY, not the "seismic" family/toggle, because that family + # also carries stream_flood_warning/stream_high_water (hydro), + # which must keep using the generic per-toggle freshness_seconds. + freshness_s = int(adapter_config.usgs_quake.freshness_seconds) else: freshness_s = int(getattr(tog, "freshness_seconds", 600) or 600) if event.timestamp and freshness_s > 0: diff --git a/work/tests/test_adapter_config_api.py b/work/tests/test_adapter_config_api.py index 8e66188..19bbe4e 100644 --- a/work/tests/test_adapter_config_api.py +++ b/work/tests/test_adapter_config_api.py @@ -86,6 +86,7 @@ def test_per_adapter_list(client): "regional_centroid", "regional_radius_mi", "broadcast_pager_alerts", "global_mag_floor", "regional_mag_floor", "escalate_mag_floor", + "freshness_seconds", } diff --git a/work/tests/test_adapter_usgs_quake.py b/work/tests/test_adapter_usgs_quake.py index 7074983..d3ee539 100644 --- a/work/tests/test_adapter_usgs_quake.py +++ b/work/tests/test_adapter_usgs_quake.py @@ -200,12 +200,32 @@ def test_populates_core_fields(adapter): assert event.source == "usgs_quake" assert event.lat == 42.61 assert event.lon == -114.48 - assert event.region == "magic_valley" + # to_event() deliberately does NOT propagate the raw dict's "region" + # (a fixed config default that never matches region_routes cell keys) + # onto the Event -- Event.region is left unset so CoverageFilter's + # geometry-based tagger (lat/lon vs named coverage areas) can set the + # real region name downstream. See test_region_left_unset_for_tagger + # and tests/test_coverage_area.py for the tagging behavior itself. + assert event.region is None assert event.expires == evt["expires"] assert event.timestamp == evt["quake_time"] assert event.id +def test_region_left_unset_for_tagger(adapter): + """Regression: to_event() must never propagate the adapter's fixed + config region (e.g. "magic_valley") onto the Event. That value never + matches a region_routes cell key ("East Idaho"/"SC Idaho"/"SW Idaho"), + so pre-setting it would silently break region-routed delivery. Leaving + event.region/regions unset lets CoverageFilter's geometry tagger (which + only fires when `not event.regions`) stamp the real area name from the + quake's actual lat/lon.""" + evt = make_quake_event(lat=44.2, lon=-114.0, region="magic_valley") + event = adapter.to_event(evt) + assert event.region is None + assert event.regions == [] + + # ============================================================ # DEFENSIVE TESTS # ============================================================ diff --git a/work/tests/test_quake_freshness_and_region_tagging.py b/work/tests/test_quake_freshness_and_region_tagging.py new file mode 100644 index 0000000..ab0166f --- /dev/null +++ b/work/tests/test_quake_freshness_and_region_tagging.py @@ -0,0 +1,212 @@ +"""Regression tests for the usgs_quake dispatcher-freshness + region-tagging fix. + +Problem (see meshai/notifications/pipeline/dispatcher.py:359-374 and +meshai/env/usgs_quake.py): the usgs_quake adapter polls a rolling PAST-DAY +USGS feed (2.5_day.geojson), so a genuinely first-seen quake is routinely +already 10min-20h old by the time it's detected. The generic per-toggle +freshness_seconds (600s) silently dropped essentially every native quake +before broadcast -- confirmed against production: quake_events had 12 +first-sighting rows (decide() ran, magnitude/region gate passed, DB row +inserted) with last_broadcast_at=NULL on every single one (commit() never +reached because the staleness filter dropped the event downstream first). + +Fix 1 (dispatcher.py): earthquake_event gets its own adapter_config-backed +freshness override (adapter_config.usgs_quake.freshness_seconds, default +3600s), mirroring the existing wfigs/"fire" override -- but scoped to the +CATEGORY, not the "seismic" family/toggle, because stream_flood_warning / +stream_high_water (hydro) also live under toggle="seismic" and must keep +using the generic per-toggle freshness unchanged. + +Fix 2 (env/usgs_quake.py): to_event() no longer passes the adapter's fixed +config region ("magic_valley", which never matches a region_routes cell +key) onto the Event. Event.region is left unset so CoverageFilter's +geometry-based tagger (lat/lon vs named coverage areas) can stamp the real +region name. +""" + +from __future__ import annotations + +import asyncio +import time +from unittest.mock import MagicMock + +import pytest + +from meshai.config import Config +from meshai.coverage_area import MonitoringArea +from meshai.env.usgs_quake import USGSQuakeAdapter +from meshai.notifications.events import make_event +from meshai.notifications.pipeline.coverage_filter import CoverageFilter +from meshai.notifications.pipeline.dispatcher import Dispatcher + + +# -------------------------------------------------------------------------- +# Shared dispatcher-test helpers (mirror tests/test_v052_dispatcher.py) +# -------------------------------------------------------------------------- + +class RecChannel: + def __init__(self, rec): + self.rec = rec + + async def deliver(self, payload, rule): + self.rec.append({"name": rule.name, "message": payload.message}) + return True + + +def _make_dispatcher(cfg): + rec: list = [] + d = Dispatcher(cfg, lambda rule, conn: RecChannel(rec), connector=None) + return d, rec + + +def _dispatch_one(cfg, event): + d, rec = _make_dispatcher(cfg) + asyncio.run(d.dispatch(event)) + return d, rec + + +def _quake_cfg(): + """Config with the seismic toggle enabled, cold-start grace disabled.""" + cfg = Config() + cfg.notifications.rules = [] + cfg.notifications.cold_start_grace_seconds = 0 + t = cfg.notifications.toggles["seismic"] + t.enabled = True + t.min_severity = "routine" + t.regions = [] + t.severity_channels = { + "routine": ["mesh_broadcast"], + "priority": ["mesh_broadcast"], + "immediate": ["mesh_broadcast"], + } + t.cooldown_seconds = 0 + # Generic per-toggle freshness -- deliberately tight (600s, the old + # default) so these tests prove the earthquake_event category is NOT + # using this value anymore. + t.freshness_seconds = 600 + return cfg + + +def _quake_event(age_seconds: float, event_id="us6000abcd", lat=42.6, lon=-114.5): + return make_event( + source="usgs_quake", + category="earthquake_event", + severity="routine", + title=f"M3.0 -- {event_id}", + timestamp=time.time() - age_seconds, + lat=lat, + lon=lon, + group_key=event_id, + inhibit_keys=[event_id], + ) + + +# ============================================================ +# (a) fresh quake passes under the new per-adapter override +# ============================================================ + +def test_fresh_quake_passes_staleness_gate_with_override(): + """A 30-minute-old quake (1800s) exceeds the generic 600s toggle window + but must pass under the adapter_config.usgs_quake.freshness_seconds + override (default 3600s).""" + cfg = _quake_cfg() + event = _quake_event(age_seconds=1800) + d, rec = _dispatch_one(cfg, event) + assert len(rec) == 1, "30-min-old quake must broadcast under the 3600s override" + assert d.dispatch_stats()["stale_dropped"] == 0 + + +# ============================================================ +# (b) old quake (12h) is still dropped +# ============================================================ + +def test_old_quake_still_dropped_by_override(): + """A 12-hour-old quake (43200s) must still be dropped -- the override + widens the window, it does not disable the staleness check.""" + cfg = _quake_cfg() + event = _quake_event(age_seconds=12 * 3600) + d, rec = _dispatch_one(cfg, event) + assert rec == [], "12h-old quake must still be dropped as stale" + assert d.dispatch_stats()["stale_dropped"] == 1 + + +def test_hydro_seismic_sibling_unaffected_by_quake_override(): + """Guardrail: stream_flood_warning shares toggle='seismic' with + earthquake_event but must keep using the GENERIC per-toggle freshness + (600s here), not the quake-specific 3600s override. A 1800s-old hydro + event must still be dropped.""" + cfg = _quake_cfg() + event = make_event( + source="usgs", category="stream_flood_warning", severity="priority", + title="Snake River nr Twin Falls 12.8 ft", + timestamp=time.time() - 1800, + ) + d, rec = _dispatch_one(cfg, event) + assert rec == [], "hydro sibling must NOT inherit the quake freshness override" + assert d.dispatch_stats()["stale_dropped"] == 1 + + +# ============================================================ +# (c) region-tagging from real lat/lon +# ============================================================ + +SW_IDAHO = MonitoringArea(name="SW Idaho", west=-117.993408, south=42.05, + east=-115.389404, north=44.331707) +SC_IDAHO = MonitoringArea(name="SC Idaho", west=-115.389404, south=42.05, + east=-112.8, north=44.331707) +EAST_IDAHO = MonitoringArea(name="East Idaho", west=-112.8, south=41.9, + east=-110.9, north=45.331707) +PROD_AREAS = [SW_IDAHO, SC_IDAHO, EAST_IDAHO] + + +def _quake_adapter(): + cfg = MagicMock() + cfg.feed_url = "https://example.test/feed.geojson" + cfg.min_magnitude = 2.5 + cfg.bbox = [-115.5, 42.0, -110.0, 45.2] + cfg.region = "magic_valley" + cfg.tick_seconds = 300 + return USGSQuakeAdapter(cfg) + + +def test_quake_event_region_tags_to_matching_coverage_area(): + """A quake with real Idaho lat/lon, run through the actual adapter's + to_event() and then the real CoverageFilter (named exactly like prod's + SW/SC/East Idaho areas), must end up tagged with a region name that + matches a region_routes cell key -- NOT the adapter's stale + "magic_valley" default.""" + adapter = _quake_adapter() + raw = { + "source": "usgs_quake", + "event_id": "us7000example", + "event_type": "Earthquake", + "severity": "routine", + "headline": "M2.7 -- 10 km N of Twin Falls, ID", + "magnitude": 2.7, + "place": "10 km N of Twin Falls, ID", + "depth_km": 8.0, + "sig": 80, + "url": "https://earthquake.usgs.gov/x", + "region": "magic_valley", + "lat": 42.61, # Twin Falls -- falls inside SC Idaho + "lon": -114.48, + "quake_time": time.time(), + "expires": time.time() + 86400, + "fetched_at": time.time(), + } + event = adapter.to_event(raw) + assert event is not None + assert event.region is None # adapter no longer pre-stamps a region + + received: list = [] + flt = CoverageFilter(next_handler=received.append, areas=PROD_AREAS, enabled=True) + flt.handle(event) + + assert len(received) == 1, "in-region quake must pass the coverage gate" + tagged = received[0] + assert tagged.region == "SC Idaho", ( + f"expected geometry tagging to set 'SC Idaho', got {tagged.region!r} " + "(pre-fix this would have stayed 'magic_valley' and never matched a " + "region_routes cell key)" + ) + assert tagged.regions == ["SC Idaho"] From 81ade1e32393390c511f3252f58a7a2ac1bd549c Mon Sep 17 00:00:00 2001 From: Matt Johnson Date: Sun, 16 Aug 2026 01:20:12 +0000 Subject: [PATCH 4/9] fix(satpass): region-tag satpass events via observer_list Satpass events carry no lat/lon/geometry, so CoverageFilter had no way to region-tag them for region_routes matching. gate_consolidated_pass() now attaches observer_list (comma-joined observer slugs, same shape already computed for the audit column) into the event data, which coverage_area.observer_region_names() reads. Defensive/fail-open: an empty observer_list yields no region tag rather than raising. --- work/meshai/env/satellite/pass_format.py | 15 ++- work/tests/test_satpass_native.py | 73 +++++++++++++++ work/tests/test_satpass_region_tagging.py | 108 ++++++++++++++++++++++ 3 files changed, 195 insertions(+), 1 deletion(-) diff --git a/work/meshai/env/satellite/pass_format.py b/work/meshai/env/satellite/pass_format.py index 12900f6..1fb3fed 100644 --- a/work/meshai/env/satellite/pass_format.py +++ b/work/meshai/env/satellite/pass_format.py @@ -373,7 +373,20 @@ def gate_consolidated_pass(consolidated: dict, *, # Prepare data dict with callbacks severity_word = _map_severity(max_el) - data = {"_meshai_precomposed": True, "_severity_override": severity_word} + data = { + "_meshai_precomposed": True, + "_severity_override": severity_word, + # Region tagging: coverage_area.observer_region_names() reads + # event.data["observer_list"] as a comma-joined string of observer + # slugs (the same shape already computed above for the audit + # column). Satpass events carry no lat/lon/geometry, so this is the + # ONLY path that lets CoverageFilter region-tag a satpass event for + # region_routes matching. `observer_list` is already defensive + # (falls back to entry_obs, or "") so this never raises; an empty + # string is fail-open in observer_region_names() (-> no region tag, + # not an exception). + "observer_list": observer_list, + } _attach_commit(data, event_id=consolidated_id, event_log_row_id=None) return wire, data diff --git a/work/tests/test_satpass_native.py b/work/tests/test_satpass_native.py index 8a35ef1..63bacb1 100644 --- a/work/tests/test_satpass_native.py +++ b/work/tests/test_satpass_native.py @@ -354,3 +354,76 @@ def test_parse_norad_ids_handles_comma_string(): assert SatpassAdapter._parse_norad_ids("25544, 33591") == [25544, 33591] assert SatpassAdapter._parse_norad_ids([25544, "33591"]) == [25544, 33591] assert SatpassAdapter._parse_norad_ids([]) == [] + + +# ══════════════════════════════════════════════════════════════════════ +# 8. observer_list FLOWS INTO THE GATE'S data DICT (region-tagging fix) +# ══════════════════════════════════════════════════════════════════════ +# +# gate_consolidated_pass() builds the event `data` dict that rides all the +# way to the dispatcher via to_event(). Satpass events carry no lat/lon/ +# geometry, so coverage_area.observer_region_names() reading +# event.data["observer_list"] is the ONLY path that can region-tag a +# satpass event for the region_routes matrix. These tests pin that the +# gate actually populates it (and stays safe when it can't). + +def test_gate_consolidated_pass_data_contains_observer_list(): + """The (wire, data) tuple returned by gate_consolidated_pass() must + carry the same comma-joined observer_list already written to the + satpass_events audit column.""" + _enable_satpass_db(dry_run=False) + from meshai.env.satellite import pass_format as sh + + consolidated = { + "consolidated_id": "25544:900000", + "norad_id": 25544, + "sat_name": "ISS (ZARYA)", + "max_elevation": 70.0, + "aos_epoch": 1_000_000, + "los_epoch": 1_000_300, + "aos_compass": "SW", + "los_compass": "NE", + "peak_compass": "S", + "entry_observer": "Boise", + "exit_observer": "Twin Falls", + "observer_list": "boise,twin", + } + result = sh.gate_consolidated_pass(consolidated, now=0) + assert result is not None + _, data = result + assert data["observer_list"] == "boise,twin" + + +def test_gate_consolidated_pass_missing_observer_list_is_defensive(): + """A `consolidated` dict with no observer_list must not raise, and the + resulting data["observer_list"] must be falsy (never a garbage value + that would resolve to a bogus region).""" + _enable_satpass_db(dry_run=False) + from meshai.env.satellite import pass_format as sh + + consolidated = { + "consolidated_id": "25544:900001", + "norad_id": 25544, + "sat_name": "ISS (ZARYA)", + "max_elevation": 40.0, + "aos_epoch": 2_000_000, + "los_epoch": 2_000_300, + "aos_compass": "SW", + "los_compass": "NE", + "peak_compass": "S", + # entry_observer / exit_observer / observer_list all deliberately absent + } + result = sh.gate_consolidated_pass(consolidated, now=0) + assert result is not None + _, data = result + assert not data.get("observer_list") + + # And observer_region_names() must treat that as "no region", not raise. + from meshai.coverage_area import MonitoringArea, observer_region_names + from meshai.notifications.events import make_event + + event = make_event(source="satpass", category="sat_pass", severity="routine", + title="Pass", data=data) + areas = [MonitoringArea(north=44.0, south=42.0, east=-113.0, west=-117.0, + name="SW Idaho")] + assert observer_region_names(event, areas) == [] diff --git a/work/tests/test_satpass_region_tagging.py b/work/tests/test_satpass_region_tagging.py index db48d73..ff885df 100644 --- a/work/tests/test_satpass_region_tagging.py +++ b/work/tests/test_satpass_region_tagging.py @@ -245,3 +245,111 @@ class TestSatpassToggle: def test_categories_for_toggle_satpass_returns_sat_pass(self): result = categories_for_toggle("satpass") assert result == ["sat_pass"] + + +# --------------------------------------------------------------------------- +# End-to-end: a satpass Event built through the REAL native-adapter path +# (predict -> consolidate -> gate_consolidated_pass -> to_event) must +# region-tag via event_region_names(), using the two production observers. +# This is the regression test for the observer_list wiring bug: before the +# fix, gate_consolidated_pass()'s data dict never carried observer_list, so +# this resolved to [] no matter how the areas were configured. +# --------------------------------------------------------------------------- + +# Production ground stations (see adapter_config observers): Treasure Valley +# (Boise area) and Magic Valley (Twin Falls area). +_TREASURE_VALLEY = {"slug": "treasure_valley", "name": "Treasure Valley", + "lat": 43.6, "lon": -116.2, "alt_m": 0.0} +_MAGIC_VALLEY = {"slug": "magic_valley", "name": "Magic Valley", + "lat": 42.5558, "lon": -114.4701, "alt_m": 0.0} + +# Coarse named boxes that cover each observer, standing in for the real +# region_routes cells ("SW Idaho" covers the Treasure Valley/Boise area, +# "SC Idaho" covers the Magic Valley/Twin Falls area). +_SW_IDAHO = MonitoringArea(north=44.5, south=42.8, east=-115.0, west=-117.5, + name="SW Idaho") +_SC_IDAHO_PROD = MonitoringArea(north=43.2, south=42.0, east=-113.0, west=-115.2, + name="SC Idaho") +_PROD_AREAS = [_SW_IDAHO, _SC_IDAHO_PROD] + + +class TestSatpassEndToEndRegionTagging: + def test_real_gate_path_region_tags_nonempty(self, monkeypatch): + """Drive the actual SatpassAdapter tick -> gate -> to_event path + (no shortcuts through gate_consolidated_pass or make_event) with the + two production observer coordinates, then confirm event_region_names + returns a non-empty list built from event.data['observer_list'].""" + import json as _json + from datetime import datetime, timezone + + from meshai.adapter_config import invalidate_cache + from meshai.config import SatpassConfig + from meshai.env.satellite.pass_predictor import PassInfo + from meshai.env.satellite.tle_store import upsert_tle + from meshai.env.satpass import SatpassAdapter + from meshai.persistence import get_db + + # -- seed a fresh ISS TLE -------------------------------------------------- + conn = get_db() + fresh = datetime.now(timezone.utc).isoformat() + iss_l1 = "1 25544U 98067A 26182.50000000 .00016717 00000-0 10270-3 0 9008" + iss_l2 = "2 25544 51.6400 208.9163 0007417 17.6777 85.6621 15.54225995 12345" + upsert_tle(conn, 25544, "ISS (ZARYA)", iss_l1, iss_l2, fresh) + + # -- enable satpass, dry_run off so the gate actually returns data -------- + conn.execute("UPDATE adapter_config SET value_json='true' " + "WHERE adapter='satpass' AND key='enabled'") + conn.execute("UPDATE adapter_config SET value_json='false' " + "WHERE adapter='satpass' AND key='dry_run'") + conn.execute("UPDATE adapter_config SET value_json=? " + "WHERE adapter='satpass' AND key='max_broadcasts_per_hour'", + (_json.dumps(100),)) + invalidate_cache() + + # -- patch observers + predictor ------------------------------------------- + monkeypatch.setattr( + "meshai.persistence.observer_locations.get_observers", + lambda *a, **k: [_TREASURE_VALLEY, _MAGIC_VALLEY]) + + t0 = (1783000000 // 3600) * 3600 + 100 + + def _fake_compute_passes(l1, l2, lat, lon, alt, window_h, min_el, now): + def _pi(aos, los, max_el, az_aos, az_los, az_peak): + peak = (aos + los) // 2 + return PassInfo( + aos_time=datetime.fromtimestamp(aos, tz=timezone.utc), + los_time=datetime.fromtimestamp(los, tz=timezone.utc), + peak_time=datetime.fromtimestamp(peak, tz=timezone.utc), + max_elevation=max_el, + azimuth_at_aos=az_aos, azimuth_at_los=az_los, + azimuth_at_peak=az_peak) + if abs(lat - _TREASURE_VALLEY["lat"]) < 0.1: + return [_pi(t0, t0 + 300, 40.0, 225, 300, 90)] + if abs(lat - _MAGIC_VALLEY["lat"]) < 0.1: + return [_pi(t0 + 60, t0 + 400, 70.0, 270, 45, 180)] + return [] + + monkeypatch.setattr( + "meshai.env.satellite.pass_predictor.compute_passes", + _fake_compute_passes) + + cfg = SatpassConfig(enabled=True, feed_source="native", + norad_ids=[25544], min_elevation_deg=10.0, + window_hours=24) + adapter = SatpassAdapter(cfg) + assert adapter.tick(now=t0 - 1800) is True # AOS 30 min out -> imminent + + staged = adapter.get_events() + assert len(staged) == 1 + event = adapter.to_event(staged[0]) + assert event is not None + + # The bug: before the fix, data["observer_list"] was never set, so + # this was always [] regardless of area config. + assert event.data.get("observer_list"), ( + "gate_consolidated_pass() did not populate observer_list on the " + "event data dict") + + names = event_region_names(event, _PROD_AREAS) + assert names, "satpass event failed to region-tag via observer_list" + assert set(names) == {"SW Idaho", "SC Idaho"} From 3c281c96e26a418911d3ae73b00d1eaf4206cc7b Mon Sep 17 00:00:00 2001 From: Matt Johnson Date: Sun, 16 Aug 2026 01:20:18 +0000 Subject: [PATCH 5/9] fix(fire): honor tombstoned_at as a durable mute in the gate decider fires.tombstoned_at (the same column the dashboard's active views already filter on: /api/env/active WHERE tombstoned_at IS NULL) is now checked in decide() before the row-missing/growth/cooldown branches, so a tombstoned incident can never re-broadcast through any path. Defensive on both the read (falls back to the pre-v12 column set if tombstoned_at doesn't exist) and the lookup (missing key reads as "not tombstoned" rather than raising) -- an un-migrated schema must not crash the decider. --- work/meshai/notifications/gating/fire.py | 36 ++++++++- work/tests/test_fire_refactor.py | 96 ++++++++++++++++++++++++ 2 files changed, 128 insertions(+), 4 deletions(-) diff --git a/work/meshai/notifications/gating/fire.py b/work/meshai/notifications/gating/fire.py index 0ee5c8e..471afd4 100644 --- a/work/meshai/notifications/gating/fire.py +++ b/work/meshai/notifications/gating/fire.py @@ -151,10 +151,38 @@ def decide(data: dict, *, source: str, now: float) -> GateResult: contained_pct = data.get("contained_pct") declared_at_epoch = data.get("declared_at_epoch") - row = conn.execute( - "SELECT current_acres, current_contained_pct, last_broadcast_at, " - "last_broadcast_acres, last_broadcast_contained " - "FROM fires WHERE irwin_id = ?", (irwin_id,)).fetchone() + try: + row = conn.execute( + "SELECT current_acres, current_contained_pct, last_broadcast_at, " + "last_broadcast_acres, last_broadcast_contained, tombstoned_at " + "FROM fires WHERE irwin_id = ?", (irwin_id,)).fetchone() + except Exception: + # Defensive: an un-migrated schema without the tombstoned_at column + # (pre-v12) must not crash the decider. Fall back to the narrower + # column set; the tombstone gate below then reads as "not tombstoned" + # for this row rather than raising. + row = conn.execute( + "SELECT current_acres, current_contained_pct, last_broadcast_at, " + "last_broadcast_acres, last_broadcast_contained " + "FROM fires WHERE irwin_id = ?", (irwin_id,)).fetchone() + + # ── Tombstone mute — durable operator kill-switch ─────────────────────── + # A non-null fires.tombstoned_at (the SAME column the dashboard's active + # views already filter on -- /api/env/active WHERE tombstoned_at IS NULL) + # means this incident is closed out / muted. It must never broadcast + # again through ANY path below (new, update, or cooldown-suppress is a + # no-op here anyway), so this check runs BEFORE the row-missing / growth + # / cooldown branches. Defensive: a row without the column, or missing + # the key for any other reason, reads as "not tombstoned" rather than + # raising -- it must never crash the decider. + if row is not None: + try: + tombstoned_at = row["tombstoned_at"] + except (IndexError, KeyError, TypeError): + tombstoned_at = None + if tombstoned_at is not None: + return GateResult(broadcast=False, lifecycle="suppress", + reason=f"incident tombstoned irwin={irwin_id}") # Stamps every active broadcast (New + Update) carries. category override # is added only for New (Update keeps the envelope-derived wildfire_incident). diff --git a/work/tests/test_fire_refactor.py b/work/tests/test_fire_refactor.py index 5d31bf1..8b1deea 100644 --- a/work/tests/test_fire_refactor.py +++ b/work/tests/test_fire_refactor.py @@ -323,6 +323,102 @@ class TestGateSequenceParity: assert gate.lifecycle == "suppress" +# ───────────────────────────────────────────────────────────────────────────── +# 2b. Tombstone mute — decide() must honor fires.tombstoned_at +# ───────────────────────────────────────────────────────────────────────────── + +class TestTombstoneMute: + """A non-null fires.tombstoned_at is a durable operator kill-switch: an + incident row stamped tombstoned must never broadcast again through + decide() -- new, update, or otherwise -- regardless of forward + acreage/containment growth or the 8h cooldown. This is the SAME column + the dashboard's /api/env/active view already filters on + (WHERE tombstoned_at IS NULL); decide() previously never read it.""" + + _IRWIN = "IRWIN-TOMBSTONE-1" + + def _seed(self, conn, *, acres, contained, last_bcast_at, tombstoned_at=None): + _write_fire_state( + conn, irwin_id=self._IRWIN, name="Lake Channel Fire", + acres=acres, contained_pct=contained, lat=42.0, lon=-114.0, + county="Twin Falls", state="ID", now=int(last_bcast_at or 1_000_000)) + conn.execute( + "UPDATE fires SET last_broadcast_at=?, last_broadcast_acres=?, " + "last_broadcast_contained=?, tombstoned_at=? WHERE irwin_id=?", + (last_bcast_at, acres, contained, tombstoned_at, self._IRWIN), + ) + + def _incident(self, *, acres, contained): + return { + "_kind": "wfigs_incident", "irwin_id": self._IRWIN, + "incident_name": "Lake Channel Fire", "acres": acres, + "contained_pct": contained, "declared_at_epoch": None, + "lat": 42.0, "lon": -114.0, "county": "Twin Falls", "state": "ID", + } + + # (a) non-tombstoned row, forward growth past cooldown -> still broadcasts. + def test_non_tombstoned_growth_still_broadcasts(self, mem_db): + self._seed(mem_db, acres=100, contained=10, + last_bcast_at=1_000_000 - 9 * 3600, tombstoned_at=None) + gate = fire_decide(self._incident(acres=200, contained=10), + source="wfigs", now=1_000_000.0) + assert gate.broadcast is True + assert gate.lifecycle == "update" + + # (b) same row, tombstoned_at set -> suppressed regardless of growth/cooldown. + def test_tombstoned_row_suppressed_despite_growth(self, mem_db): + self._seed(mem_db, acres=100, contained=10, + last_bcast_at=1_000_000 - 9 * 3600, tombstoned_at=999_000) + gate = fire_decide(self._incident(acres=9999, contained=99), + source="wfigs", now=1_000_000.0) + assert gate.broadcast is False + assert gate.lifecycle == "suppress" + assert "tombston" in gate.reason.lower() + + # (c) brand-new incident, not tombstoned (no row at all) -> broadcasts new. + def test_brand_new_incident_not_tombstoned_broadcasts(self, mem_db): + gate = fire_decide(self._incident(acres=50, contained=0), + source="wfigs", now=1_000_000.0) + assert gate.broadcast is True + assert gate.lifecycle == "new" + + # (d) defensive: a row lacking the tombstoned_at column must not raise. + def test_missing_tombstoned_at_column_does_not_raise(self, mem_db, monkeypatch): + self._seed(mem_db, acres=100, contained=10, + last_bcast_at=1_000_000 - 9 * 3600, tombstoned_at=None) + + import sqlite3 + + import meshai.notifications.gating.fire as fire_mod + + class _NoTombstoneColConn: + """Proxies the real connection but simulates a pre-v12 schema: + any SELECT naming tombstoned_at raises OperationalError, exactly + as a real un-migrated `fires` table would.""" + + def __init__(self, real): + self._real = real + + def execute(self, sql, *args, **kwargs): + if "tombstoned_at" in sql and "SELECT" in sql.upper(): + raise sqlite3.OperationalError( + "no such column: tombstoned_at") + return self._real.execute(sql, *args, **kwargs) + + def __getattr__(self, name): + return getattr(self._real, name) + + monkeypatch.setattr( + fire_mod, "get_db", lambda: _NoTombstoneColConn(mem_db)) + + gate = fire_decide(self._incident(acres=200, contained=10), + source="wfigs", now=1_000_000.0) + # Falls back to "not tombstoned" and proceeds with normal gating -- + # must not raise, and growth past cooldown still broadcasts. + assert gate.broadcast is True + assert gate.lifecycle == "update" + + # ───────────────────────────────────────────────────────────────────────────── # 3. Registration — the three explicit categories resolve; FIRMS does not # ───────────────────────────────────────────────────────────────────────────── From c5aa0e1f4269a42168262e1bc23d279d68b93c5d Mon Sep 17 00:00:00 2001 From: Matt Johnson Date: Sun, 2 Aug 2026 01:34:15 +0000 Subject: [PATCH 6/9] Fix event-loop starvation, MeshCore stability, config-page hardening - mesh_data_store.py / env/store.py: make refresh() async, offload blocking polls via asyncio.to_thread/gather so 7 lockstep sources no longer starve the shared event loop. - main.py: gather pollers concurrently + set_default_executor thread pool. - Dockerfile / docker-compose.yml: healthcheck now curls the dashboard for a real liveness signal instead of a process-exists check. - transport/meshcore_transport.py: MeshCore keepalive loop (get_time() every 120s), reconnect re-arm (_post_reconnect_setup_async from _on_connect_event), and MC channel-name normalization (_resolve_mc_channel_idx strips a leading #). - dashboard-frontend: MeshCoreConnection.tsx config-page hardening, new ErrorBoundary component, wired into App.tsx. - tests: fix ~40 call sites broken by refresh() becoming async ( test_generic_http.py, test_store_received_delta.py, test_store_wzdx_persist.py) by wrapping with asyncio.run(), matching this suite's existing convention for calling async code from sync test functions. Verified: all 40 tests pass. Co-Authored-By: Claude Opus 4.8 --- work/Dockerfile | 4 +- work/dashboard-frontend/src/App.tsx | 3 + .../src/components/ErrorBoundary.tsx | 49 ++++++ .../src/pages/MeshCoreConnection.tsx | 138 ++++++++++----- work/docker-compose.yml | 4 +- work/meshai/env/store.py | 33 +++- work/meshai/main.py | 57 +++++- work/meshai/mesh_data_store.py | 45 +++-- work/meshai/transport/meshcore_transport.py | 165 +++++++++++++++++- work/tests/test_generic_http.py | 9 +- work/tests/test_store_received_delta.py | 63 +++---- work/tests/test_store_wzdx_persist.py | 25 +-- 12 files changed, 464 insertions(+), 131 deletions(-) create mode 100644 work/dashboard-frontend/src/components/ErrorBoundary.tsx diff --git a/work/Dockerfile b/work/Dockerfile index f62c587..49df00e 100644 --- a/work/Dockerfile +++ b/work/Dockerfile @@ -91,8 +91,8 @@ VOLUME ["/data"] EXPOSE 8080 # Health check - verify bot process is alive via PID file -HEALTHCHECK --interval=30s --timeout=10s --start-period=10s --retries=3 \ - CMD test -f /tmp/meshai.pid && kill -0 $(cat /tmp/meshai.pid) 2>/dev/null && [ "$(cat /tmp/meshai.link 2>/dev/null)" = up ] || exit 1 +HEALTHCHECK --interval=30s --timeout=10s --start-period=240s --retries=3 \ + CMD curl -f -s -o /dev/null http://localhost:8080/ || exit 1 # Entrypoint writes default config on first run, then starts the bot ENTRYPOINT ["/app/docker-entrypoint.sh"] diff --git a/work/dashboard-frontend/src/App.tsx b/work/dashboard-frontend/src/App.tsx index a0a8785..253e9a3 100644 --- a/work/dashboard-frontend/src/App.tsx +++ b/work/dashboard-frontend/src/App.tsx @@ -18,12 +18,14 @@ import MeshCoreDangerZones from './pages/MeshCoreDangerZones' import Coverage from './pages/Coverage' import { ToastProvider } from './components/ToastProvider' import { DirtyProvider } from './context/DirtyContext' +import ErrorBoundary from './components/ErrorBoundary' function App() { return ( + {/* Core routes */} } /> @@ -66,6 +68,7 @@ function App() { } /> } /> + diff --git a/work/dashboard-frontend/src/components/ErrorBoundary.tsx b/work/dashboard-frontend/src/components/ErrorBoundary.tsx new file mode 100644 index 0000000..9b56ad3 --- /dev/null +++ b/work/dashboard-frontend/src/components/ErrorBoundary.tsx @@ -0,0 +1,49 @@ +// App-wide render-error safety net. Wraps in App.tsx so an +// unhandled error thrown while rendering any page degrades to a recoverable +// "Something went wrong" card instead of a blank white screen. +import { Component, type ErrorInfo, type ReactNode } from 'react' +import { AlertTriangle } from 'lucide-react' + +interface Props { + children: ReactNode +} + +interface State { + hasError: boolean + error: Error | null +} + +export default class ErrorBoundary extends Component { + state: State = { hasError: false, error: null } + + static getDerivedStateFromError(error: Error): State { + return { hasError: true, error } + } + + componentDidCatch(error: Error, errorInfo: ErrorInfo) { + console.error('MeshAI dashboard render error:', error, errorInfo) + } + + render() { + if (this.state.hasError) { + return ( +
+
+ +
Something went wrong
+
+ {this.state.error?.message ?? 'An unexpected error occurred while rendering this page.'} +
+ +
+
+ ) + } + return this.props.children + } +} diff --git a/work/dashboard-frontend/src/pages/MeshCoreConnection.tsx b/work/dashboard-frontend/src/pages/MeshCoreConnection.tsx index ac6646f..700201d 100644 --- a/work/dashboard-frontend/src/pages/MeshCoreConnection.tsx +++ b/work/dashboard-frontend/src/pages/MeshCoreConnection.tsx @@ -1,6 +1,6 @@ import { useState, useEffect, useCallback } from 'react' import { Link } from 'react-router-dom' -import { Save, RotateCcw, RefreshCw, Check, ChevronRight, Trash2, Eye, EyeOff, Copy } from 'lucide-react' +import { Save, RotateCcw, RefreshCw, Check, ChevronRight, Trash2, Eye, EyeOff, Copy, X } from 'lucide-react' import { TextInput, NumberInput, Toggle, ListInput, SelectInput } from './Config' import SerialPortPicker from '@/components/SerialPortPicker' import { notifyRestartRequired } from '@/components/RestartBanner' @@ -58,6 +58,9 @@ export default function MeshCoreConnection() { const [error, setError] = useState(null) const [success, setSuccess] = useState(null) const [hasChanges, setHasChanges] = useState(false) + // Set when one or both config fetches fail; the form still renders (using + // safe defaults for whatever didn't load) instead of blanking the page. + const [loadError, setLoadError] = useState(null) // Test send state const [channelsActive, setChannelsActive] = useState(false) @@ -81,22 +84,41 @@ export default function MeshCoreConnection() { const fetchConfig = useCallback(async () => { setLoading(true) - try { - const [data, mcCtx] = await Promise.all([ - apiFetchConfig('connection') as Promise, - apiFetchConfig('meshcore_context') as Promise, - ]) - setConfig(data) - setOriginalConfig(JSON.parse(JSON.stringify(data))) - setMcContext(mcCtx) - setOriginalMcContext(JSON.parse(JSON.stringify(mcCtx))) - setHasChanges(false) - setError(null) - } catch (err) { - setError(err instanceof Error ? err.message : 'Unknown error') - } finally { - setLoading(false) + // Promise.allSettled so one section failing to load can't blank the + // whole page — each section is set independently from its own result, + // and any failure(s) are surfaced as a dismissible banner alongside the + // still-usable form (see the `!config` dead-end this replaces). + const [connResult, mcResult] = await Promise.allSettled([ + apiFetchConfig('connection') as Promise, + apiFetchConfig('meshcore_context') as Promise, + ]) + + const errors: string[] = [] + + if (connResult.status === 'fulfilled') { + setConfig(connResult.value) + setOriginalConfig(JSON.parse(JSON.stringify(connResult.value))) + } else { + errors.push(`connection (${connResult.reason instanceof Error ? connResult.reason.message : String(connResult.reason)})`) + // Fall back to an empty object (never null) so the form below always + // has something to render safe defaults from, and so the dirty-check + // diff below still has a baseline to compare edits against. + setConfig((c) => c ?? {}) + setOriginalConfig((c) => c ?? {}) } + + if (mcResult.status === 'fulfilled') { + setMcContext(mcResult.value) + setOriginalMcContext(JSON.parse(JSON.stringify(mcResult.value))) + } else { + errors.push(`bot behavior (${mcResult.reason instanceof Error ? mcResult.reason.message : String(mcResult.reason)})`) + // Leave mcContext null — the "Bot behavior" card only renders when it + // is present, so a null value just hides that card cleanly. + } + + setLoadError(errors.length ? `Couldn't load ${errors.join(' and ')}. Showing defaults — edits below are still safe to make and save.` : null) + setHasChanges(false) + setLoading(false) }, []) useEffect(() => { @@ -207,10 +229,13 @@ export default function MeshCoreConnection() { } useEffect(() => { - if (config && originalConfig && mcContext && originalMcContext) { + // mcContext may be null (its fetch failed) — don't let that block + // detecting changes to the connection fields, which always load or + // fall back to an empty-object baseline in fetchConfig. + if (config && originalConfig) { const changed = JSON.stringify(config) !== JSON.stringify(originalConfig) || - JSON.stringify(mcContext) !== JSON.stringify(originalMcContext) + (!!mcContext && !!originalMcContext && JSON.stringify(mcContext) !== JSON.stringify(originalMcContext)) setHasChanges(changed) } }, [config, originalConfig, mcContext, originalMcContext]) @@ -221,22 +246,24 @@ export default function MeshCoreConnection() { }, [hasChanges, setDirty]) const upd = (patch: Partial) => - setConfig((c) => (c ? { ...c, ...patch } : c)) + // Build off an empty object rather than bailing when config is still + // null (e.g. mid-retry) — edits should never be silently dropped. + setConfig((c) => ({ ...(c ?? {}), ...patch })) const saveConfig = async () => { - if (!config || !mcContext) return + if (!config) return setSaving(true) setError(null) setSuccess(null) try { // PUT the whole objects so sibling fields (Meshtastic connection fields, - // any other meshcore_context keys) are preserved. - const results = await Promise.all([ - apiUpdateConfig('connection', config), - apiUpdateConfig('meshcore_context', mcContext), - ]) + // any other meshcore_context keys) are preserved. Only PUT + // meshcore_context if it actually loaded — its fetch may have failed. + const puts: Promise<{ restart_required?: boolean }>[] = [apiUpdateConfig('connection', config)] + if (mcContext) puts.push(apiUpdateConfig('meshcore_context', mcContext)) + const results = await Promise.all(puts) setOriginalConfig(JSON.parse(JSON.stringify(config))) - setOriginalMcContext(JSON.parse(JSON.stringify(mcContext))) + if (mcContext) setOriginalMcContext(JSON.parse(JSON.stringify(mcContext))) setHasChanges(false) setDirty(false) setSuccess('MeshCore connection saved successfully') @@ -276,13 +303,10 @@ export default function MeshCoreConnection() { ) } - if (!config) { - return ( -
-
Failed to load connection config
-
- ) - } + // Always render the form below, even if the fetch failed — `cfg` supplies + // safe defaults so the connection-type/host/port fields and Save/Discard + // stay usable. The failure itself is surfaced by the loadError banner. + const cfg: ConnectionConfig = config ?? {} return (
@@ -320,6 +344,30 @@ export default function MeshCoreConnection() {
+ {/* Load-error banner — dismissible, with its own Retry so the page + never dead-ends when a config section fails to fetch. The form + below stays fully editable regardless. */} + {loadError && ( +
+
{loadError}
+ + +
+ )} + {/* Status messages */} {error && (
{error}
@@ -341,7 +389,7 @@ export default function MeshCoreConnection() {

upd({ meshcore_conn_type: v })} options={[ { value: 'tcp', label: 'TCP (companion)' }, @@ -350,11 +398,11 @@ export default function MeshCoreConnection() { ]} helper="TCP for a companion frame server, Serial for a USB node, BLE for Bluetooth" /> - {(config.meshcore_conn_type ?? 'tcp') === 'tcp' && ( + {(cfg.meshcore_conn_type ?? 'tcp') === 'tcp' && (
upd({ meshcore_host: v })} placeholder="192.168.1.100" helper="IP or hostname of the companion frame server" @@ -362,7 +410,7 @@ export default function MeshCoreConnection() { /> upd({ meshcore_port: v })} min={1} max={65535} @@ -370,27 +418,27 @@ export default function MeshCoreConnection() { />
)} - {(config.meshcore_conn_type ?? 'tcp') === 'serial' && ( + {(cfg.meshcore_conn_type ?? 'tcp') === 'serial' && ( <> upd({ meshcore_serial_port: v })} helper="USB-attached MeshCore node — Detect fills a stable by-id path" /> upd({ meshcore_baud: v })} min={1200} helper="Serial baud rate (default 115200)" /> )} - {(config.meshcore_conn_type ?? 'tcp') === 'ble' && ( + {(cfg.meshcore_conn_type ?? 'tcp') === 'ble' && ( upd({ meshcore_ble_address: v })} placeholder="AA:BB:CC:DD:EE:FF" helper="Leave blank to scan/pair the first available device" @@ -406,7 +454,7 @@ export default function MeshCoreConnection() { upd({ meshcore_auto_add_contacts: v })} helper="Enables firmware CMD 58 (set_autoadd_config) at connect so AIDA automatically adds every node it hears an advert from as a contact, enabling DM send/decrypt without manual contact exchange" /> @@ -418,13 +466,13 @@ export default function MeshCoreConnection() {
upd({ meshcore_auto_reconnect: v })} helper="Automatically reconnect to the MeshCore companion if the link drops" /> upd({ meshcore_max_reconnect_attempts: v })} min={0} helper="Maximum reconnect attempts before giving up (0 = unlimited)" diff --git a/work/docker-compose.yml b/work/docker-compose.yml index e2b1e3a..a30a665 100644 --- a/work/docker-compose.yml +++ b/work/docker-compose.yml @@ -75,11 +75,11 @@ services: memory: 64M healthcheck: - test: ["CMD-SHELL", "test -f /tmp/meshai.pid && kill -0 $(cat /tmp/meshai.pid) 2>/dev/null && [ \"$(cat /tmp/meshai.link 2>/dev/null)\" = up ] || exit 1"] + test: ["CMD-SHELL", "curl -f -s -o /dev/null http://localhost:8080/ || exit 1"] interval: 30s timeout: 10s retries: 3 - start_period: 15s + start_period: 240s logging: driver: "json-file" diff --git a/work/meshai/env/store.py b/work/meshai/env/store.py index 06109a1..15f295b 100644 --- a/work/meshai/env/store.py +++ b/work/meshai/env/store.py @@ -1,5 +1,6 @@ """Environmental data store with tick-based adapter polling.""" +import asyncio import hashlib import json import logging @@ -438,20 +439,38 @@ class EnvironmentalStore: from meshai import coverage as _cov return _cov.resolve_adapter_coverage(adapter, self._coverage_bbox, "native") - def refresh(self) -> bool: + async def refresh(self) -> bool: """Called every second from main loop. Ticks each adapter. + Adapter tick() calls (blocking network I/O) run concurrently in + worker threads and are AWAITED to completion before ingest, so the + event loop stays responsive while fetches are in flight. Ingest + (DB/EventBus work) then runs on the loop thread once all ticks are + done, exactly as before, so no thread ever overlaps ingest. + Returns: True if any data changed """ changed = False - for name, adapter in self._adapters.items(): - try: - if adapter.tick(): - changed = True + adapters = list(self._adapters.items()) + if not adapters: + self._purge_expired() + return changed + + results = await asyncio.gather( + *(asyncio.to_thread(adapter.tick) for _, adapter in adapters), + return_exceptions=True, + ) + for (name, adapter), result in zip(adapters, results): + if isinstance(result, Exception): + logger.warning("Env adapter %s error: %s", name, result) + continue + if result: + changed = True + try: self._ingest(name, adapter) - except Exception as e: - logger.warning("Env adapter %s error: %s", name, e) + except Exception as e: + logger.warning("Env adapter %s error: %s", name, e) self._purge_expired() return changed diff --git a/work/meshai/main.py b/work/meshai/main.py index 33866bf..cffc049 100644 --- a/work/meshai/main.py +++ b/work/meshai/main.py @@ -2,6 +2,7 @@ import argparse import asyncio +import concurrent.futures import logging import os import signal @@ -146,13 +147,46 @@ class MeshAI: while self._running: await asyncio.sleep(1) - # Periodic MeshMonitor refresh - if self.meshmonitor_sync: - self.meshmonitor_sync.maybe_refresh() - - # Periodic data store refresh and health computation + # Run the mesh/env/meshmonitor pollers concurrently so blocking + # network I/O (tick() fetches) never starves this loop — and + # therefore never starves the dashboard, which shares this same + # asyncio loop. Each refresh() internally awaits its own due + # ticks in worker threads; meshmonitor_sync.maybe_refresh is + # synchronous, so it is offloaded to a thread here directly. + # We await the WHOLE cycle before the next iteration, so no + # tick() thread ever overlaps the next cycle's bookkeeping. + _refresh_tasks = {} + if self.data_store: + _refresh_tasks['data'] = self.data_store.refresh() + if self.env_store: + _refresh_tasks['env'] = self.env_store.refresh() + if self.meshmonitor_sync: + _refresh_tasks['mm'] = asyncio.to_thread(self.meshmonitor_sync.maybe_refresh) + + if _refresh_tasks: + _refresh_results = dict(zip( + _refresh_tasks.keys(), + await asyncio.gather(*_refresh_tasks.values(), return_exceptions=True), + )) + else: + _refresh_results = {} + + refreshed = _refresh_results.get('data') + if isinstance(refreshed, Exception): + logger.warning("Data store refresh error: %s", refreshed) + refreshed = False + + env_changed = _refresh_results.get('env') + if isinstance(env_changed, Exception): + logger.debug("Env refresh error: %s", env_changed) + env_changed = False + + _mm_result = _refresh_results.get('mm') + if isinstance(_mm_result, Exception): + logger.warning("MeshMonitor sync refresh error: %s", _mm_result) + + # Periodic data store health computation if self.data_store: - refreshed = self.data_store.refresh() # Recompute health after refresh if refreshed and self.health_engine: self.health_engine.compute(self.data_store) @@ -202,10 +236,9 @@ class MeshAI: except Exception: pass - # Environmental feed refresh + # Environmental feed alerting/broadcast (refresh already ran above) if self.env_store: try: - env_changed = self.env_store.refresh() if env_changed and self.alert_engine: env_alerts = self.alert_engine.check_environmental(self.env_store) if env_alerts: @@ -930,6 +963,14 @@ def main() -> None: loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) + # Size the default executor generously: mesh sources + env adapters + # (~7 sources, ~15 env adapters) now fetch concurrently via + # asyncio.to_thread() every tick, so they need thread headroom to avoid + # queuing behind each other on the default executor's small pool. + loop.set_default_executor( + concurrent.futures.ThreadPoolExecutor(max_workers=24, thread_name_prefix="meshai-io") + ) + def signal_handler(sig, frame): logger.info(f"Received signal {sig}") loop.create_task(bot.stop()) diff --git a/work/meshai/mesh_data_store.py b/work/meshai/mesh_data_store.py index 44047db..97c1592 100644 --- a/work/meshai/mesh_data_store.py +++ b/work/meshai/mesh_data_store.py @@ -6,6 +6,7 @@ This module replaces mesh_sources.py with a clean three-layer architecture: - Layer 3: Consumers read unified model (no field guessing) """ +import asyncio import json import logging import sqlite3 @@ -441,12 +442,16 @@ class MeshDataStore: if stale_nums: logger.info(f"Purged {len(stale_nums)} stale nodes (not heard in {STALE_NODE_THRESHOLD_DAYS} days)") - def refresh(self) -> bool: + async def refresh(self) -> bool: """Tick-based refresh. Called every second from the main loop. - Delegates to source tick() for sources that support it. - Only does a full rebuild when nodes/edges/topology change. - Only does a lightweight update when only packets change. + Delegates to source tick() for sources that support it. Due sources' + tick() calls (blocking network I/O) run concurrently in worker + threads and are AWAITED to completion before any bookkeeping, so the + event loop (and therefore the dashboard) stays responsive while + fetches are in flight. Only does a full rebuild when nodes/edges/ + topology change. Only does a lightweight update when only packets + change. Returns: True if any data changed @@ -456,26 +461,38 @@ class MeshDataStore: needs_rebuild = False needs_packet_update = False + due: list[tuple[str, object]] = [] for name, source in self._sources.items(): # Check if this source supports tick-based polling if hasattr(source, 'tick') and hasattr(source, '_tick_interval'): if now - source._last_tick >= source._tick_interval: - endpoint = source.tick() - if endpoint: - any_changed = True - # Major changes require full rebuild - if endpoint in ("nodes", "edges", "traceroutes", "topology", "telemetry"): - needs_rebuild = True - # Packet-only changes are lightweight - elif endpoint in ("packets",): - needs_packet_update = True - # stats, counts, channels, solar, network just update cached data + due.append((name, source)) else: # Legacy fallback for sources without tick support if source.maybe_refresh(): any_changed = True needs_rebuild = True + if due: + results = await asyncio.gather( + *(asyncio.to_thread(source.tick) for _, source in due), + return_exceptions=True, + ) + for (name, source), result in zip(due, results): + if isinstance(result, Exception): + logger.warning(f"Source {name} tick failed: {result}") + continue + endpoint = result + if endpoint: + any_changed = True + # Major changes require full rebuild + if endpoint in ("nodes", "edges", "traceroutes", "topology", "telemetry"): + needs_rebuild = True + # Packet-only changes are lightweight + elif endpoint in ("packets",): + needs_packet_update = True + # stats, counts, channels, solar, network just update cached data + if needs_rebuild: self._rebuild() self._purge_stale_nodes() diff --git a/work/meshai/transport/meshcore_transport.py b/work/meshai/transport/meshcore_transport.py index 9be229d..6d15bba 100644 --- a/work/meshai/transport/meshcore_transport.py +++ b/work/meshai/transport/meshcore_transport.py @@ -36,6 +36,13 @@ _TELEMETRY_MIN_INTERVAL_SECONDS = 300 # dropped from the auto-poll rotation (a manual "Poll now" un-sticks it). _TELEMETRY_MAX_FAILURES = 3 +# --- Companion-link keepalive tuning -------------------------------------- +# MeshMonitor's MeshCore vnode (the shared companion-link server meshai +# attaches to) reaps any client idle >5 min, where "idle" means no bytes seen +# FROM the client — a periodic LOCAL query resets that clock. 120s is well +# inside the 300s reaper window with margin to spare. +_KEEPALIVE_INTERVAL_SECONDS = 120 + # Numeric Cayenne-LPP type id → decoded field name. Ids not in this map are # passed through as ``lpp_`` so nothing is silently dropped. _LPP_ID_TO_FIELD = { @@ -132,6 +139,8 @@ class MeshCoreTransport(MeshTransport): self._advert_task = None # asyncio.Task handle for the telemetry auto-poll loop; None when inactive. self._telemetry_task = None + # asyncio.Task handle for the companion-link keepalive loop; None when inactive. + self._keepalive_task = None # Telemetry availability/bookkeeping (shared by poller + on-demand): # _telemetry_cache: contact-id -> {contact, data, polled_at, available} # _telemetry_failures: contact-id -> consecutive-timeout count @@ -577,15 +586,37 @@ class MeshCoreTransport(MeshTransport): ) return acked or (not result.is_error()) + def _resolve_mc_channel_idx(self, meshcore_channel: str) -> Optional[int]: + """Resolve a config channel name to the companion's channel slot. + + Tries an exact match first (fast path, preserves existing behavior + for e.g. ``#bot``). Falls back to a match that ignores a single + leading ``#`` and case, since after the radio moved to MeshMonitor's + vnode the companion enumerates region channels WITHOUT the leading + ``#`` that meshai's region_routes config still carries (e.g. config + ``#sc-id-aida`` vs. companion ``sc-id-aida``) — same channel/key, + just a display-name difference upstream. + """ + idx = self._chan_name_to_idx.get(meshcore_channel) + if idx is not None: + return idx + canon = meshcore_channel[1:] if meshcore_channel.startswith("#") else meshcore_channel + canon = canon.casefold() + for name, slot in self._chan_name_to_idx.items(): + name_canon = name[1:] if name.startswith("#") else name + if name_canon.casefold() == canon: + return slot + return None + async def _do_mc_broadcast_async(self, text: str, meshcore_channel: str) -> bool: """Channel broadcast on the MC loop (replaces send_message() broadcast branch).""" if self._mc is None: return False - idx = self._chan_name_to_idx.get(meshcore_channel) + idx = self._resolve_mc_channel_idx(meshcore_channel) if idx is None: # Lazy async re-enumeration (no _run_coro deadlock risk). await self._enumerate_channels_async() - idx = self._chan_name_to_idx.get(meshcore_channel) + idx = self._resolve_mc_channel_idx(meshcore_channel) if idx is None: logger.warning("MC channel '%s' not on companion; skipping", meshcore_channel) return False @@ -1776,6 +1807,66 @@ class MeshCoreTransport(MeshTransport): if task is not None and self._loop is not None and self._loop.is_running(): self._loop.call_soon_threadsafe(task.cancel) + # ------------------------------------------------------------------ + # Companion-link keepalive (Task on the dedicated loop) + # ------------------------------------------------------------------ + + async def _keepalive_loop(self) -> None: + """Quiet LOCAL companion-link keepalive (Task on the dedicated loop). + + MeshMonitor's MeshCore vnode disconnects any client idle >5 min, + where "idle" means no bytes seen FROM the client — its + ``lastActivity`` only updates on data we send it, never on data it + sends us. The self-advert (every 24h by default) and telemetry poll + (30 min default, and only when contacts are configured) are both far + too infrequent to keep that clock fresh, so the link was silently + reaped and never recovered (``meshcore_auto_reconnect`` is the + recovery safety net; this loop is the prevention). + + Every ``_KEEPALIVE_INTERVAL_SECONDS`` (while connected), issues + ``commands.get_time()`` — a single-byte companion opcode (CMD 0x05) + that reads the node's own onboard clock and returns CURRENT_TIME. + It carries no destination/contact and has no mesh-routing semantics + (unlike send_advert/send_msg/send_chan_msg), so the firmware answers + it purely locally over the companion link — it does not key the + radio or emit an RF packet. Runs directly on the MC loop (NOT + through the send queue/pacing — it is a device-info query, not a + mesh send, so it should never wait behind or delay a real send). + + Stops on CancelledError (disconnect). A transient query failure is + logged and ignored — the loop keeps ticking every interval either + way, since the point is resetting the vnode's clock on our next + successful frame, not the query result itself. + """ + try: + while True: + await asyncio.sleep(_KEEPALIVE_INTERVAL_SECONDS) + if not self._connected or self._mc is None: + return + try: + await self._mc.commands.get_time() + logger.debug("MC: companion-link keepalive query sent") + except Exception as exc: + logger.debug("MC: keepalive get_time failed (non-fatal): %s", exc) + except asyncio.CancelledError: + logger.debug("MC: keepalive task cancelled") + raise + + def _schedule_keepalive(self) -> None: + """Create the keepalive asyncio.Task on the dedicated loop (thread-safe).""" + def _arm() -> None: + self._keepalive_task = asyncio.get_event_loop().create_task( + self._keepalive_loop() + ) + self._loop.call_soon_threadsafe(_arm) + + def _cancel_keepalive(self) -> None: + """Cancel the keepalive task (thread-safe). Called at disconnect.""" + task = self._keepalive_task + self._keepalive_task = None + if task is not None and self._loop is not None and self._loop.is_running(): + self._loop.call_soon_threadsafe(task.cancel) + # ------------------------------------------------------------------ # Internal coroutines (run on the dedicated loop) # ------------------------------------------------------------------ @@ -1963,6 +2054,12 @@ class MeshCoreTransport(MeshTransport): if telem_interval > 0: self._schedule_telemetry_poll() + # Arm the quiet local companion-link keepalive — unconditional (not + # a mesh operation, no config gate): protects against the + # MeshMonitor vnode's 5-min idle reaper regardless of advert/ + # telemetry cadence. + self._schedule_keepalive() + logger.info( "MeshCoreTransport: connected as %s (pubkey %s)", self._self_info.get("name", "unknown"), @@ -1971,9 +2068,10 @@ class MeshCoreTransport(MeshTransport): def disconnect(self) -> None: """Disconnect and stop the event loop thread.""" - # Cancel periodic advert + telemetry poll before tearing down the loop. + # Cancel periodic advert + telemetry poll + keepalive before tearing down the loop. self._cancel_periodic_advert() self._cancel_telemetry_poll() + self._cancel_keepalive() if self._mc is not None: try: self._run_coro(self._do_disconnect(), timeout=10.0) @@ -2078,12 +2176,12 @@ class MeshCoreTransport(MeshTransport): ) return True # Resolve NAME → slot against the live companion channel table. - idx = self._chan_name_to_idx.get(meshcore_channel) + idx = self._resolve_mc_channel_idx(meshcore_channel) if idx is None: # One lazy re-enumeration in case the table changed since # connect (e.g. a channel was provisioned after startup). self._enumerate_channels() - idx = self._chan_name_to_idx.get(meshcore_channel) + idx = self._resolve_mc_channel_idx(meshcore_channel) if idx is None: # Never blind-send to a guessed slot. logger.warning( @@ -2263,10 +2361,65 @@ class MeshCoreTransport(MeshTransport): self._connected = False logger.warning("MeshCoreTransport: DISCONNECTED event received") + async def _post_reconnect_setup_async(self) -> None: + """Redo connect()'s LOCAL post-connect setup after an auto-reconnect. + + connect() does this setup once, on the initial connect: rebuild + ``_chan_name_to_idx`` (so channel-name broadcasts can resolve a + slot) and arm the companion-link keepalive (so MeshMonitor's vnode + doesn't reap the link again at 5 min idle). The meshcore lib's + auto-reconnect only re-establishes the socket and fires CONNECTED + (-> ``_on_connect_event``) — it does not repeat that setup, so a + reconnected link was left with an empty channel table and no + keepalive until the reaper cut it again. + + Both steps here are local companion queries/timers only — + ``_enumerate_channels_async`` calls ``get_channel()`` and the + keepalive calls ``get_time()`` (see their docstrings); neither + keys the radio or emits an RF packet. This deliberately excludes + connect()'s ``send_advert()`` — that IS a transmission, and must + stay confined to the initial connect() path, never replayed on + reconnect. + + Keepalive re-arm is cancel-then-schedule (idempotent) so it never + double-schedules the task. + + Runs as a fire-and-forget task on the dedicated MC loop (see + ``_on_connect_event``) rather than being awaited inline via + ``_run_coro``: the meshcore lib invokes ``_on_connect_event`` from + within that same loop (like ``_on_new_contact``), so a blocking + ``_run_coro().result()`` call here would deadlock it. + """ + try: + await self._enumerate_channels_async() + except Exception: + logger.warning( + "MeshCore: post-reconnect channel re-enumeration failed", exc_info=True + ) + try: + self._cancel_keepalive() + self._schedule_keepalive() + except Exception: + logger.warning( + "MeshCore: post-reconnect keepalive re-arm failed", exc_info=True + ) + def _on_connect_event(self, event=None) -> None: - """Track link state: CONNECTED (auto-reconnect succeeded).""" + """Track link state: CONNECTED (auto-reconnect succeeded). + + Schedules ``_post_reconnect_setup_async`` fire-and-forget on the + dedicated MC loop — see that method's docstring for why this must + not block (``_run_coro`` would deadlock from inside this callback, + exactly as noted in ``_on_new_contact``). + """ self._connected = True logger.info("MeshCoreTransport: CONNECTED event received") + try: + loop = getattr(self, "_loop", None) + if loop is not None and loop.is_running(): + asyncio.run_coroutine_threadsafe(self._post_reconnect_setup_async(), loop) + except Exception: + logger.debug("MeshCore: scheduling post-reconnect setup failed", exc_info=True) # ------------------------------------------------------------------ # Node identity / topology (MeshTransport abstract methods) diff --git a/work/tests/test_generic_http.py b/work/tests/test_generic_http.py index 2b40fab..854aa76 100644 --- a/work/tests/test_generic_http.py +++ b/work/tests/test_generic_http.py @@ -8,6 +8,7 @@ Ported-behavior coverage: * geometry-path Point -> centroid extraction """ from __future__ import annotations +import asyncio import json @@ -203,7 +204,7 @@ def test_cold_start_silent_first_poll_seeds_persists_no_emit(): # Stub the network fetch with one active outage. adapter._fetch = lambda url, extra_headers=None: _payload([IDAHO_POWER_ITEM]) - store.refresh() # poll 1 == pre-existing backlog + asyncio.run(store.refresh()) # poll 1 == pre-existing backlog # Nothing broadcast on the cold-start poll... assert captured == [], "first poll must broadcast NOTHING (cold-start seed)" @@ -220,14 +221,14 @@ def test_cold_start_silent_first_poll_seeds_persists_no_emit(): def test_later_poll_broadcasts_newly_received_item(): store, adapter, captured = _make_store_with_generic() adapter._fetch = lambda url, extra_headers=None: _payload([IDAHO_POWER_ITEM]) - store.refresh() # poll 1 — seed silently + asyncio.run(store.refresh()) # poll 1 — seed silently assert captured == [] # A genuinely NEW outage appears on a later poll -> it must broadcast. new_item = dict(IDAHO_POWER_ITEM, omsOutageId="456", omsCustomerCount=99) adapter._fetch = lambda url, extra_headers=None: _payload([IDAHO_POWER_ITEM, new_item]) adapter._last_poll.clear() # force cadence to elapse - store.refresh() # poll 2 + asyncio.run(store.refresh()) # poll 2 assert len(captured) == 1, "only the newly-received outage broadcasts" assert captured[0].category == "power_outage" @@ -367,7 +368,7 @@ def test_build_generic_detail_reader(): from meshai.notifications.env_reporter import EnvReporter store, adapter, captured = _make_store_with_generic() adapter._fetch = lambda url, extra_headers=None: _payload([IDAHO_POWER_ITEM]) - store.refresh() + asyncio.run(store.refresh()) text = EnvReporter().build_generic_detail() assert "idaho_power" in text diff --git a/work/tests/test_store_received_delta.py b/work/tests/test_store_received_delta.py index 77555fc..1b253ef 100644 --- a/work/tests/test_store_received_delta.py +++ b/work/tests/test_store_received_delta.py @@ -14,6 +14,7 @@ These tests drive the real EnvironmentalStore + EventBus with a fake adapter whose per-poll batch we control, and assert exactly which events reach the bus. """ from __future__ import annotations +import asyncio from meshai.env.store import EnvironmentalStore, _key_ext from meshai.config import EnvironmentalConfig @@ -79,7 +80,7 @@ def test_first_poll_seeds_and_broadcasts_nothing(): store, adapter, captured = _make_store() adapter.set_batch(["A", "B", "C"]) - store.refresh() # poll 1 — the backlog + asyncio.run(store.refresh()) # poll 1 — the backlog assert captured == [], "first poll must broadcast NOTHING (backlog seed)" @@ -88,11 +89,11 @@ def test_second_poll_emits_only_newly_received(): store, adapter, captured = _make_store() adapter.set_batch(["A", "B", "C"]) - store.refresh() # poll 1: seed + asyncio.run(store.refresh()) # poll 1: seed assert _emitted_ids(captured) == [] adapter.set_batch(["A", "B", "C", "D"]) - store.refresh() # poll 2: only D is new + asyncio.run(store.refresh()) # poll 2: only D is new assert _emitted_ids(captured) == ["D"] @@ -100,11 +101,11 @@ def test_unchanged_poll_emits_nothing(): store, adapter, captured = _make_store() adapter.set_batch(["A", "B", "C"]) - store.refresh() # poll 1: seed + asyncio.run(store.refresh()) # poll 1: seed adapter.set_batch(["A", "B", "C", "D"]) - store.refresh() # poll 2: D + asyncio.run(store.refresh()) # poll 2: D adapter.set_batch(["A", "B", "C", "D"]) - store.refresh() # poll 3: nothing new + asyncio.run(store.refresh()) # poll 3: nothing new assert _emitted_ids(captured) == ["D"], "poll 3 has no new items" @@ -113,21 +114,21 @@ def test_restart_reseeds_and_never_rebroadcasts_backlog(): # Process 1 sees A,B,C,D and broadcasts D. store1, adapter1, cap1 = _make_store() adapter1.set_batch(["A", "B", "C"]) - store1.refresh() + asyncio.run(store1.refresh()) adapter1.set_batch(["A", "B", "C", "D"]) - store1.refresh() + asyncio.run(store1.refresh()) assert _emitted_ids(cap1) == ["D"] # RESTART: a fresh store has an empty seen-set. The SAME backlog [A,B,C,D] # arriving on its first poll must be re-seeded silently, not re-broadcast. store2, adapter2, cap2 = _make_store() adapter2.set_batch(["A", "B", "C", "D"]) - store2.refresh() + asyncio.run(store2.refresh()) assert cap2 == [], "restart must NEVER re-broadcast the existing backlog" # And a genuinely new item after the restart still broadcasts once. adapter2.set_batch(["A", "B", "C", "D", "E"]) - store2.refresh() + asyncio.run(store2.refresh()) assert _emitted_ids(cap2) == ["E"] @@ -135,9 +136,9 @@ def test_stable_key_prevents_reemit_when_batch_reorders(): # The same real-world items in a different order are NOT "newly received". store, adapter, captured = _make_store() adapter.set_batch(["A", "B", "C"]) - store.refresh() # seed + asyncio.run(store.refresh()) # seed adapter.set_batch(["C", "A", "B"]) # reordered, same items - store.refresh() + asyncio.run(store.refresh()) assert captured == [], "reordering the same items emits nothing" @@ -147,12 +148,12 @@ def test_disabled_for_days_then_backlog_is_not_broadcast(): store, adapter, captured = _make_store() backlog = [f"evt{i}" for i in range(200)] adapter.set_batch(backlog) - store.refresh() # first poll after re-enable + asyncio.run(store.refresh()) # first poll after re-enable assert captured == [], "a days-old backlog is seeded silently, never sent" # Only a truly new arrival afterward is announced. adapter.set_batch(backlog + ["fresh"]) - store.refresh() + asyncio.run(store.refresh()) assert _emitted_ids(captured) == ["fresh"] @@ -290,11 +291,11 @@ def test_persistent_preseed_known_suppressed_new_emitted(): assert len(store._seen["wzdx"]) == 5 adapter.set_batch(known) - store.refresh() + asyncio.run(store.refresh()) assert captured == [], "all 5 are durably-known → zero broadcast" adapter.set_batch(known + ["z_new"]) - store.refresh() + asyncio.run(store.refresh()) assert _emitted_ids(captured) == ["z_new"], "only the not-in-table id broadcasts" @@ -307,9 +308,9 @@ def test_persistent_preseed_cross_tick_staging_no_leak(): store, captured = _build_store(_GENERIC_NAME, adapter) adapter.set_batch(["A"]) - store.refresh() # tick 1: only A present + asyncio.run(store.refresh()) # tick 1: only A present adapter.set_batch(["A", "B"]) - store.refresh() # tick 2: B appears (backlog) + asyncio.run(store.refresh()) # tick 2: B appears (backlog) assert captured == [], "B is durably-known — must NOT leak on a later tick" # CONTROL: identical staging but NO durable rows → B leaks (proves the @@ -325,11 +326,11 @@ def test_persistent_preseed_cross_tick_staging_no_leak(): # Re-point ctrl events to a fresh source with no durable rows. for e in ctrl._batch: e["source"] = "wzdx_ctrl" - store2.refresh() + asyncio.run(store2.refresh()) ctrl.set_batch(["A", "B"]) for e in ctrl._batch: e["source"] = "wzdx_ctrl" - store2.refresh() + asyncio.run(store2.refresh()) assert [e.title for e in cap2] == ["B"], "without a durable record, B leaks" @@ -343,11 +344,11 @@ def test_incremental_empty_first_tick_then_only_new_broadcasts(): store, captured = _build_store(_GENERIC_NAME, adapter) adapter.set_batch([]) # empty first tick - store.refresh() + asyncio.run(store.refresh()) assert captured == [], "empty tick emits nothing" adapter.set_batch(["A", "B", "C"]) # backlog A,B + new C - store.refresh() + asyncio.run(store.refresh()) assert _emitted_ids(captured) == ["C"], "only the never-received C broadcasts" @@ -360,18 +361,18 @@ def test_restart_against_same_persistent_db_never_rebroadcasts(): a1 = _FakeWZDx() store1, cap1 = _build_store(_GENERIC_NAME, a1) a1.set_batch(backlog) - store1.refresh() + asyncio.run(store1.refresh()) assert cap1 == [], "process 1: durable backlog is silent" # RESTART: brand-new store, same persistent DB → pre-seed reloads. a2 = _FakeWZDx() store2, cap2 = _build_store(_GENERIC_NAME, a2) a2.set_batch(backlog) - store2.refresh() + asyncio.run(store2.refresh()) assert cap2 == [], "restart must NEVER re-broadcast the durable backlog" a2.set_batch(backlog + ["E"]) - store2.refresh() + asyncio.run(store2.refresh()) assert _emitted_ids(cap2) == ["E"], "a genuinely-new item still broadcasts once" @@ -385,11 +386,11 @@ def test_persistent_preseed_quake_by_event_id(): assert len(store._seen["usgs_quake"]) == 2 adapter.set_batch(["us1000aaaa", "us1000bbbb"]) - store.refresh() + asyncio.run(store.refresh()) assert captured == [], "both quakes already received → zero broadcast" adapter.set_batch(["us1000aaaa", "us1000bbbb", "us1000cccc"]) - store.refresh() + asyncio.run(store.refresh()) assert _emitted_ids(captured) == ["us1000cccc"], "only the new quake broadcasts" @@ -402,11 +403,11 @@ def test_no_durable_rows_falls_back_to_silent_first_poll(): assert "wzdx" not in store._seeded, "0 durable rows → not pre-marked seeded" adapter.set_batch(["A", "B"]) - store.refresh() + asyncio.run(store.refresh()) assert captured == [], "first non-empty poll on a fresh DB is silent" adapter.set_batch(["A", "B", "C"]) - store.refresh() + asyncio.run(store.refresh()) assert _emitted_ids(captured) == ["C"] @@ -486,9 +487,9 @@ def test_persistent_preseed_roads511_by_external_id(): assert len(store._seen["511"]) == 4 adapter.set_batch(known) - store.refresh() + asyncio.run(store.refresh()) assert captured == [], "all 4 durably-known 511 rows → zero broadcast" adapter.set_batch(known + ["511_99"]) - store.refresh() + asyncio.run(store.refresh()) assert _emitted_ids(captured) == ["511_99"], "only the not-in-table id broadcasts" diff --git a/work/tests/test_store_wzdx_persist.py b/work/tests/test_store_wzdx_persist.py index 13422ee..4e60a97 100644 --- a/work/tests/test_store_wzdx_persist.py +++ b/work/tests/test_store_wzdx_persist.py @@ -20,6 +20,7 @@ adapter whose per-poll coalesced set we control, then assert directly against traffic_events AND against the bus (nothing must ever be dispatched). """ from __future__ import annotations +import asyncio from meshai.env.store import EnvironmentalStore from meshai.config import EnvironmentalConfig @@ -147,7 +148,7 @@ def test_first_poll_persists_current_set_and_broadcasts_nothing(): store, captured = _build_store(adapter) adapter.set_zones(ZONES3) - store.refresh() # first (cold-start) poll + asyncio.run(store.refresh()) # first (cold-start) poll rows = _wzdx_rows() exts = {r["external_id"] for r in rows} @@ -171,7 +172,7 @@ def test_columns_match_summary_and_dm_queries(): adapter = _FakeWZDx() store, _ = _build_store(adapter) adapter.set_zones([ZONES3[1]]) # the full_closure I-84 zone - store.refresh() + asyncio.run(store.refresh()) r = _wzdx_rows()[0] assert r["road"] == "I-84" @@ -188,12 +189,12 @@ def test_subsequent_poll_reconciles_removed_zone(): adapter = _FakeWZDx() store, captured = _build_store(adapter) adapter.set_zones(ZONES3) - store.refresh() + asyncio.run(store.refresh()) assert len(_wzdx_rows()) == 3 # Next poll: US-20 dropped out; I-84 + ID-55 remain. adapter.set_zones([ZONES3[1], ZONES3[2]]) - store.refresh() + asyncio.run(store.refresh()) exts = {r["external_id"] for r in _wzdx_rows()} assert exts == {ZONES3[1]["ext"], ZONES3[2]["ext"]}, ( @@ -207,11 +208,11 @@ def test_empty_or_failed_fetch_does_not_wipe_existing_rows(): adapter = _FakeWZDx() store, _ = _build_store(adapter) adapter.set_zones(ZONES3) - store.refresh() + asyncio.run(store.refresh()) assert len(_wzdx_rows()) == 3 adapter.set_raw([]) # empty/failed poll - store.refresh() + asyncio.run(store.refresh()) assert len(_wzdx_rows()) == 3, ( "an empty fetch must NEVER wipe the existing active set") @@ -225,7 +226,7 @@ def test_upsert_preserves_first_seen_at_and_refreshes_end_at(): z = dict(ZONES3[0]); z["end_at"] = 1000 adapter.set_zones([z]) - store.refresh() + asyncio.run(store.refresh()) r1 = _wzdx_rows()[0] first_seen = r1["first_seen_at"] assert r1["end_at"] == 1000 @@ -233,7 +234,7 @@ def test_upsert_preserves_first_seen_at_and_refreshes_end_at(): # Same zone reappears with a LATER end_at. z2 = dict(ZONES3[0]); z2["end_at"] = 5000 adapter.set_zones([z2]) - store.refresh() + asyncio.run(store.refresh()) r2 = _wzdx_rows()[0] assert r2["first_seen_at"] == first_seen, "first_seen_at must be preserved" assert r2["end_at"] == 5000, "end_at must refresh from the feed" @@ -254,7 +255,7 @@ def test_expiry_end_at_preserved_for_not_expired_filter(): "sub_type": "x", "impact": "partial", "end_at": now - 10_000}, # expired ] adapter.set_zones(zones) - store.refresh() + asyncio.run(store.refresh()) # All 3 persisted (ingest does not itself drop expired rows) ... assert len(_wzdx_rows()) == 3 @@ -269,7 +270,7 @@ def test_id_less_zone_is_skipped_not_fatal(): store, _ = _build_store(adapter) good = ZONES3[0] adapter.set_zones([good]) - store.refresh() + asyncio.run(store.refresh()) assert len(_wzdx_rows()) == 1 # Poll with the good zone plus an id-less junk event. @@ -277,7 +278,7 @@ def test_id_less_zone_is_skipped_not_fatal(): junk = {"source": "wzdx", "event_id": None, "external_id": None, "lat": 5.0, "lon": 5.0, "normalized": {}, "fetched_at": 0} adapter._batch.append(junk) - store.refresh() + asyncio.run(store.refresh()) rows = _wzdx_rows() assert {r["external_id"] for r in rows} == {good["ext"]}, ( @@ -296,7 +297,7 @@ def test_bulk_current_set_persists_all_like_the_real_127(): for i in range(127) ] adapter.set_zones(zones) - store.refresh() + asyncio.run(store.refresh()) assert len(_wzdx_rows()) == 127 assert _summary_visible_count(now=0) == 127, ( From bdbc2afe7f07c735b9c0ffff0fa493389f32978e Mon Sep 17 00:00:00 2001 From: Matt Johnson Date: Sun, 2 Aug 2026 04:10:20 +0000 Subject: [PATCH 7/9] MeshCore reconnect persistence: implement 0=unlimited max-reconnect-attempts sentinel connect() now translates a configured meshcore_max_reconnect_attempts of 0 (or <=0) into an effectively-unbounded count before handing it to the meshcore library's ConnectionManager, so its retry loop never permanently exhausts. config.py's comment already documented \"0 = unlimited\" but that sentinel was never actually implemented -- literal 0 meant zero attempts, and the shipped default of 5 (at ~1s/attempt) gave up after ~5s with no external supervisor to retry again, leaving MeshCore dead until a manual container restart. Also flips the repo default from 5 to 0 so fresh deploys get unlimited retries without extra config. Proven via a 60s forced-outage auto-recovery test: the link recovers from any-length vnode/radio outage instead of giving up after ~5s. Co-Authored-By: Claude Opus 4.8 --- work/meshai/config.py | 2 +- work/meshai/transport/meshcore_transport.py | 33 +++++++++++++++++++++ 2 files changed, 34 insertions(+), 1 deletion(-) diff --git a/work/meshai/config.py b/work/meshai/config.py index 8e1dab6..35e7c56 100644 --- a/work/meshai/config.py +++ b/work/meshai/config.py @@ -51,7 +51,7 @@ class ConnectionConfig: meshcore_host: str = "" # pyMC companion frame server host meshcore_port: int = 5050 # pyMC companion frame server port meshcore_auto_reconnect: bool = True # enable meshcore lib auto-reconnect - meshcore_max_reconnect_attempts: int = 5 # max reconnect attempts (0 = unlimited) + meshcore_max_reconnect_attempts: int = 0 # max reconnect attempts (0 = unlimited) meshcore_advert_interval_seconds: int = 86400 # periodic self-advert interval, 24h (0 = disabled) # MeshCore connection type: tcp | serial | ble (default tcp for back-compat) meshcore_conn_type: str = "tcp" diff --git a/work/meshai/transport/meshcore_transport.py b/work/meshai/transport/meshcore_transport.py index 6d15bba..221684b 100644 --- a/work/meshai/transport/meshcore_transport.py +++ b/work/meshai/transport/meshcore_transport.py @@ -43,6 +43,29 @@ _TELEMETRY_MAX_FAILURES = 3 # inside the 300s reaper window with margin to spare. _KEEPALIVE_INTERVAL_SECONDS = 120 +# --- Reconnect persistence ("0 = unlimited" sentinel) ---------------------- +# config.py documents meshcore_max_reconnect_attempts as "0 = unlimited", but +# that sentinel was never implemented here — the value was passed straight +# through to the meshcore lib's ConnectionManager, whose retry loop is +# `while self._reconnect_attempts < self.max_reconnect_attempts`. Taken +# literally, 0 means ZERO attempts (immediate give-up), the opposite of +# "unlimited", and any small bounded value (the shipped default is 5, at the +# lib's flat 1s-per-attempt cadence) exhausts after ~5 seconds and then the +# link stays down PERMANENTLY — there is no external supervisor for MeshCore +# (see main.py's watchdog guard: "MeshCoreTransport manages its own +# reconnect via the meshcore lib's auto_reconnect parameter"), so nothing +# ever notices and retries again after that. A radio/vnode bounce longer +# than ~5s (e.g. the 2026-08-02 device-perm heal test) killed MeshCore for +# good until a manual container restart. +# +# Fix: honor the documented sentinel for real. connect() below translates a +# configured 0 into this effectively-unbounded count, so the lib's own +# proven-safe retry loop (still local TCP only, still ~1 attempt/sec, still +# WITHOUT re-sending the connect-time self-advert — see +# _post_reconnect_setup_async) just keeps going until the vnode/radio comes +# back, no matter how long the outage lasts. +_MC_RECONNECT_ATTEMPTS_UNLIMITED = 2_147_483_647 + # Numeric Cayenne-LPP type id → decoded field name. Ids not in this map are # passed through as ``lpp_`` so nothing is silently dropped. _LPP_ID_TO_FIELD = { @@ -1976,6 +1999,16 @@ class MeshCoreTransport(MeshTransport): ble_address = getattr(self.config, "meshcore_ble_address", "") auto_reconnect = getattr(self.config, "meshcore_auto_reconnect", True) max_attempts = getattr(self.config, "meshcore_max_reconnect_attempts", 5) + if max_attempts <= 0: + # Documented sentinel (config.py: "0 = unlimited") — see + # _MC_RECONNECT_ATTEMPTS_UNLIMITED's docstring for why this was + # never actually unlimited before and why translating it here is + # the fix. + logger.info( + "MeshCoreTransport: meshcore_max_reconnect_attempts=%s (unlimited) -> %d", + max_attempts, _MC_RECONNECT_ATTEMPTS_UNLIMITED, + ) + max_attempts = _MC_RECONNECT_ATTEMPTS_UNLIMITED # Human-readable target for logging — from the same descriptor that # self_info() reports, so the log and the API never disagree. From 3961f7ec04b5872b6c143c8a9d7d4155cff80afc Mon Sep 17 00:00:00 2001 From: Matt Johnson Date: Sun, 16 Aug 2026 02:58:00 +0000 Subject: [PATCH 8/9] fix: honour town_anchors.enabled in alert anchor-resolution queries resolve_anchor() (notifications/formatters/_anchor.py) and _location_anchor() (env/fire_render.py) selected all town_anchors rows regardless of the enabled flag, so a disabled anchor could still be used in outbound alert text -- only the dashboard/curation routes are meant to see disabled rows. Add AND enabled = 1 to both queries. No-op today: all 186 live town_anchors rows are enabled=1. Matters the next time someone disables an anchor from the curation UI. Also updates the traffic_last/0003.json wzdx golden literal in test_incident_refactor.py and adds TestAnchorResolve:: test_disabled_anchor_excluded, which forces the Photon fallback to miss and asserts a disabled-only DB row is not selected. --- work/meshai/env/fire_render.py | 2 +- .../notifications/formatters/_anchor.py | 2 +- work/tests/test_incident_refactor.py | 35 ++++++++++++++++++- 3 files changed, 36 insertions(+), 3 deletions(-) diff --git a/work/meshai/env/fire_render.py b/work/meshai/env/fire_render.py index 04028fb..3bd0f27 100644 --- a/work/meshai/env/fire_render.py +++ b/work/meshai/env/fire_render.py @@ -248,7 +248,7 @@ def _location_anchor(n: dict) -> str: try: from meshai.persistence import get_db rows = get_db().execute( - "SELECT name, lat, lon FROM town_anchors WHERE lat IS NOT NULL AND lon IS NOT NULL" + "SELECT name, lat, lon FROM town_anchors WHERE lat IS NOT NULL AND lon IS NOT NULL AND enabled = 1" ).fetchall() best = None best_d = float("inf") diff --git a/work/meshai/notifications/formatters/_anchor.py b/work/meshai/notifications/formatters/_anchor.py index 10960e1..d6fb7ee 100644 --- a/work/meshai/notifications/formatters/_anchor.py +++ b/work/meshai/notifications/formatters/_anchor.py @@ -96,7 +96,7 @@ def resolve_anchor( from meshai.persistence import get_db rows = get_db().execute( "SELECT name, lat, lon FROM town_anchors " - "WHERE lat IS NOT NULL AND lon IS NOT NULL" + "WHERE lat IS NOT NULL AND lon IS NOT NULL AND enabled = 1" ).fetchall() best = None best_d = float("inf") diff --git a/work/tests/test_incident_refactor.py b/work/tests/test_incident_refactor.py index 8f6d1ca..3bb82c2 100644 --- a/work/tests/test_incident_refactor.py +++ b/work/tests/test_incident_refactor.py @@ -188,7 +188,11 @@ class TestWorkZoneGolden: _GOLDEN = { "0002.json": "🚧 US-91, near Chubbuck: southbound, road construction, ends Aug 17", - "0003.json": "🚧 US-95, near Wilder: southbound, ends Jul 19", + # "wilder" was added to the town_anchors seed by the seed-list sync + # (Fix 2), so the DB-anchor step now wins over the live Photon + # geocode this golden was originally captured against; the DB row's + # coords round to 1 mi S instead of Photon's sub-mile "near". + "0003.json": "🚧 US-95, 1 mi S of Wilder: southbound, ends Jul 19", } # Captured from the deleted normalizer's normalize() + _n_to_canonical_workzone() @@ -434,6 +438,35 @@ class TestAnchorResolve: assert resolve_anchor(None, -116.2, max_mi=50.0) is None assert resolve_anchor(43.6, None, max_mi=50.0) is None + def test_disabled_anchor_excluded(self, monkeypatch): + """A disabled=0 town_anchors row must not be selected, even when it is + the closest row within max_mi — the enabled flag is a hard exclude.""" + import time as _time + from meshai.persistence import get_db + from meshai.notifications.formatters._anchor import resolve_anchor + from meshai import geo + + # Force the Photon fallback to a known miss so a non-None result can + # only come from the (wrongly-included) disabled DB row. + monkeypatch.setattr( + geo, "nearest_town", + lambda lat, lon, max_distance_mi=50.0: None, + ) + + conn = get_db() + # Clear seeded anchors so only our controlled (disabled) row exists. + conn.execute("DELETE FROM town_anchors") + conn.execute( + "INSERT INTO town_anchors(name, lat, lon, state, enabled, updated_at) " + "VALUES (?, ?, ?, ?, ?, ?)", + ("disabledville", -33.8688, 151.2093, "NSW", 0, _time.time()), # Sydney + ) + + # Event right next to the disabled row; Photon fallback is forced to + # miss → None confirms the DB step excluded the disabled row. + result = resolve_anchor(-33.870, 151.210, max_mi=50.0) + assert result is None + # ── 4. Schema conformance ──────────────────────────────────────────────────── From 522458f194bdc53d4a0202f120c951123315cd67 Mon Sep 17 00:00:00 2001 From: Matt Johnson Date: Sun, 16 Aug 2026 02:58:11 +0000 Subject: [PATCH 9/9] fix: sync _TOWN_ANCHORS_SEED with the production town_anchors table _TOWN_ANCHORS_SEED covered only 29 hand-picked towns, while the live CT108 town_anchors table (GUI-curated since) has grown to 186. Any fresh deploy (fresh volume, or the pre-v19 sqlite path) would seed only the original 29 and silently lose anchor-resolution coverage for the other 157 real, already-in-production towns. Regenerated the dict from a live dump of CT108's town_anchors (name, lat, lon, state; all 186 rows confirmed enabled=1), alphabetized, same dict-of-dicts shape and column-aligned brace formatting as before (alignment column widened to fit the longest name, "mountain home afb"). seed_town_anchors() is unchanged and still INSERT OR IGNORE, so it stays idempotent against the existing production rows. Expanding the seed changes which town resolves as "nearest" for a few existing test fixtures whose incident/work-zone coordinates are genuinely closer to a newly-added real town (e.g. Oakley, Wilder) than to the old 29-town subset's nearest match -- those goldens and assertions are updated to the new (correct) nearest-town result. test_api_post_add_town's probe town is renamed from the now-real "Bellevue" to a fictitious "Testopolis" to avoid a duplicate-name 400. --- work/meshai/persistence/curation.py | 215 ++++++++++++++++++++++++---- work/tests/test_curation.py | 9 +- work/tests/test_fire_refactor.py | 5 +- work/tests/test_wfigs_handler.py | 15 +- 4 files changed, 206 insertions(+), 38 deletions(-) diff --git a/work/meshai/persistence/curation.py b/work/meshai/persistence/curation.py index 0c69430..84978dd 100644 --- a/work/meshai/persistence/curation.py +++ b/work/meshai/persistence/curation.py @@ -85,35 +85,192 @@ _GAUGE_SITES_SEED: dict[str, dict[str, Any]] = { # Idaho + neighbor towns, originally from a hardcoded _TOWN_COORDS table in # the (since-deleted) Central-envelope adapter-normalizer module. _TOWN_ANCHORS_SEED: dict[str, dict[str, Any]] = { - "boise": {"lat": 43.6150, "lon": -116.2023, "state": "ID"}, - "meridian": {"lat": 43.6121, "lon": -116.3915, "state": "ID"}, - "nampa": {"lat": 43.5407, "lon": -116.5635, "state": "ID"}, - "caldwell": {"lat": 43.6629, "lon": -116.6874, "state": "ID"}, - "idaho falls": {"lat": 43.4666, "lon": -112.0340, "state": "ID"}, - "pocatello": {"lat": 42.8713, "lon": -112.4455, "state": "ID"}, - "twin falls": {"lat": 42.5630, "lon": -114.4609, "state": "ID"}, - "coeur d'alene": {"lat": 47.6777, "lon": -116.7805, "state": "ID"}, - "lewiston": {"lat": 46.4165, "lon": -117.0177, "state": "ID"}, - "moscow": {"lat": 46.7324, "lon": -117.0002, "state": "ID"}, - "sandpoint": {"lat": 48.2766, "lon": -116.5535, "state": "ID"}, - "post falls": {"lat": 47.7180, "lon": -116.9516, "state": "ID"}, - "hayden": {"lat": 47.7660, "lon": -116.7866, "state": "ID"}, - "rathdrum": {"lat": 47.8121, "lon": -116.8950, "state": "ID"}, - "plummer": {"lat": 47.3344, "lon": -116.8856, "state": "ID"}, - "kellogg": {"lat": 47.5380, "lon": -116.1352, "state": "ID"}, - "bonners ferry": {"lat": 48.6914, "lon": -116.3181, "state": "ID"}, - "rexburg": {"lat": 43.8260, "lon": -111.7897, "state": "ID"}, - "blackfoot": {"lat": 43.1905, "lon": -112.3447, "state": "ID"}, - "burley": {"lat": 42.5360, "lon": -113.7928, "state": "ID"}, - "jerome": {"lat": 42.7252, "lon": -114.5187, "state": "ID"}, - "mountain home": {"lat": 43.1330, "lon": -115.6912, "state": "ID"}, - "stanley": {"lat": 44.2160, "lon": -114.9311, "state": "ID"}, - "salmon": {"lat": 45.1758, "lon": -113.8957, "state": "ID"}, - "mccall": {"lat": 44.9111, "lon": -116.0987, "state": "ID"}, - "weiser": {"lat": 44.2510, "lon": -116.9690, "state": "ID"}, - "soda springs": {"lat": 42.6543, "lon": -111.6047, "state": "ID"}, - "preston": {"lat": 42.0963, "lon": -111.8766, "state": "ID"}, - "montpelier": {"lat": 42.3232, "lon": -111.2980, "state": "ID"}, + "aberdeen": {"lat": 42.944098, "lon": -112.838381, "state": "ID"}, + "albion": {"lat": 42.409808, "lon": -113.580438, "state": "ID"}, + "american falls": {"lat": 42.782846, "lon": -112.854211, "state": "ID"}, + "ammon": {"lat": 43.474999, "lon": -111.959631, "state": "ID"}, + "arbon valley": {"lat": 42.88763, "lon": -112.589386, "state": "ID"}, + "arco": {"lat": 43.631893, "lon": -113.301033, "state": "ID"}, + "arimo": {"lat": 42.560385, "lon": -112.172927, "state": "ID"}, + "ashton": {"lat": 44.073332, "lon": -111.448311, "state": "ID"}, + "athol": {"lat": 47.947065, "lon": -116.707958, "state": "ID"}, + "avimor": {"lat": 43.776181, "lon": -116.257108, "state": "ID"}, + "bancroft": {"lat": 42.720241, "lon": -111.88301, "state": "ID"}, + "basalt": {"lat": 43.314443, "lon": -112.165044, "state": "ID"}, + "bellevue": {"lat": 43.467894, "lon": -114.254955, "state": "ID"}, + "bennington": {"lat": 42.382259, "lon": -111.32098, "state": "ID"}, + "blackfoot": {"lat": 43.1905, "lon": -112.3447, "state": "ID"}, + "blanchard": {"lat": 48.0137, "lon": -116.996503, "state": "ID"}, + "bliss": {"lat": 42.924284, "lon": -114.947516, "state": "ID"}, + "boise": {"lat": 43.615, "lon": -116.2023, "state": "ID"}, + "bonners ferry": {"lat": 48.6914, "lon": -116.3181, "state": "ID"}, + "buhl": {"lat": 42.598362, "lon": -114.759536, "state": "ID"}, + "burley": {"lat": 42.536, "lon": -113.7928, "state": "ID"}, + "caldwell": {"lat": 43.6629, "lon": -116.6874, "state": "ID"}, + "cambridge": {"lat": 44.571748, "lon": -116.678101, "state": "ID"}, + "carey": {"lat": 43.312011, "lon": -113.941008, "state": "ID"}, + "cascade": {"lat": 44.508761, "lon": -116.043627, "state": "ID"}, + "castleford": {"lat": 42.520569, "lon": -114.871806, "state": "ID"}, + "challis": {"lat": 44.505779, "lon": -114.228184, "state": "ID"}, + "chubbuck": {"lat": 42.926182, "lon": -112.462537, "state": "ID"}, + "clark fork": {"lat": 48.148019, "lon": -116.172988, "state": "ID"}, + "clifton": {"lat": 42.187286, "lon": -112.004596, "state": "ID"}, + "coeur d'alene": {"lat": 47.6777, "lon": -116.7805, "state": "ID"}, + "cottonwood": {"lat": 46.051045, "lon": -116.349751, "state": "ID"}, + "council": {"lat": 44.733195, "lon": -116.436837, "state": "ID"}, + "craigmont": {"lat": 46.24174, "lon": -116.471344, "state": "ID"}, + "culdesac": {"lat": 46.374896, "lon": -116.670097, "state": "ID"}, + "dalton gardens": {"lat": 47.733412, "lon": -116.767873, "state": "ID"}, + "dayton": {"lat": 42.111246, "lon": -111.984615, "state": "ID"}, + "deary": {"lat": 46.800586, "lon": -116.557368, "state": "ID"}, + "declo": {"lat": 42.519599, "lon": -113.628732, "state": "ID"}, + "dietrich": {"lat": 42.912796, "lon": -114.266289, "state": "ID"}, + "donnelly": {"lat": 44.733391, "lon": -116.086821, "state": "ID"}, + "dover": {"lat": 48.259156, "lon": -116.609843, "state": "ID"}, + "downey": {"lat": 42.428828, "lon": -112.123303, "state": "ID"}, + "driggs": {"lat": 43.729865, "lon": -111.104319, "state": "ID"}, + "dubois": {"lat": 44.171161, "lon": -112.228278, "state": "ID"}, + "eagle": {"lat": 43.693423, "lon": -116.345989, "state": "ID"}, + "east hope": {"lat": 48.240895, "lon": -116.28941, "state": "ID"}, + "eden": {"lat": 42.605309, "lon": -114.209087, "state": "ID"}, + "emmett": {"lat": 43.869228, "lon": -116.491336, "state": "ID"}, + "fairfield": {"lat": 43.348045, "lon": -114.800826, "state": "ID"}, + "fernwood": {"lat": 47.115099, "lon": -116.386484, "state": "ID"}, + "filer": {"lat": 42.56789, "lon": -114.611471, "state": "ID"}, + "firth": {"lat": 43.305788, "lon": -112.183454, "state": "ID"}, + "fort hall": {"lat": 43.014547, "lon": -112.45787, "state": "ID"}, + "franklin": {"lat": 42.009503, "lon": -111.802183, "state": "ID"}, + "fruitland": {"lat": 44.020412, "lon": -116.922109, "state": "ID"}, + "garden city": {"lat": 43.668297, "lon": -116.294389, "state": "ID"}, + "garden valley": {"lat": 44.083404, "lon": -115.958464, "state": "ID"}, + "genesee": {"lat": 46.551608, "lon": -116.928389, "state": "ID"}, + "georgetown": {"lat": 42.479083, "lon": -111.363497, "state": "ID"}, + "glenns ferry": {"lat": 42.949908, "lon": -115.308185, "state": "ID"}, + "gooding": {"lat": 42.937048, "lon": -114.713188, "state": "ID"}, + "grace": {"lat": 42.575182, "lon": -111.729771, "state": "ID"}, + "grand view": {"lat": 42.985123, "lon": -116.09368, "state": "ID"}, + "grangeville": {"lat": 45.925826, "lon": -116.121916, "state": "ID"}, + "greenleaf": {"lat": 43.672607, "lon": -116.821443, "state": "ID"}, + "groveland": {"lat": 43.223483, "lon": -112.37547, "state": "ID"}, + "hagerman": {"lat": 42.816016, "lon": -114.897681, "state": "ID"}, + "hailey": {"lat": 43.512674, "lon": -114.299499, "state": "ID"}, + "hammett": {"lat": 42.944114, "lon": -115.465521, "state": "ID"}, + "hansen": {"lat": 42.531365, "lon": -114.301177, "state": "ID"}, + "harrison": {"lat": 47.469582, "lon": -116.808336, "state": "ID"}, + "hauser": {"lat": 47.773694, "lon": -117.008, "state": "ID"}, + "hayden": {"lat": 47.766, "lon": -116.7866, "state": "ID"}, + "hayden lake": {"lat": 47.76446, "lon": -116.756097, "state": "ID"}, + "hazelton": {"lat": 42.59548, "lon": -114.136614, "state": "ID"}, + "heyburn": {"lat": 42.559982, "lon": -113.762067, "state": "ID"}, + "hidden springs": {"lat": 43.716541, "lon": -116.259415, "state": "ID"}, + "hollister": {"lat": 42.352911, "lon": -114.583846, "state": "ID"}, + "homedale": {"lat": 43.615937, "lon": -116.939029, "state": "ID"}, + "horseshoe bend": {"lat": 43.916085, "lon": -116.199236, "state": "ID"}, + "idaho city": {"lat": 43.827844, "lon": -115.830474, "state": "ID"}, + "idaho falls": {"lat": 43.4666, "lon": -112.034, "state": "ID"}, + "inkom": {"lat": 42.796548, "lon": -112.254625, "state": "ID"}, + "iona": {"lat": 43.527007, "lon": -111.930914, "state": "ID"}, + "irwin": {"lat": 43.403359, "lon": -111.279513, "state": "ID"}, + "jerome": {"lat": 42.7252, "lon": -114.5187, "state": "ID"}, + "juliaetta": {"lat": 46.574737, "lon": -116.70808, "state": "ID"}, + "kamiah": {"lat": 46.226796, "lon": -116.028303, "state": "ID"}, + "kellogg": {"lat": 47.538, "lon": -116.1352, "state": "ID"}, + "kendrick": {"lat": 46.614183, "lon": -116.661272, "state": "ID"}, + "ketchum": {"lat": 43.687718, "lon": -114.380069, "state": "ID"}, + "kimberly": {"lat": 42.534299, "lon": -114.369931, "state": "ID"}, + "kooskia": {"lat": 46.141687, "lon": -115.973646, "state": "ID"}, + "kootenai": {"lat": 48.311831, "lon": -116.517128, "state": "ID"}, + "kuna": {"lat": 43.469607, "lon": -116.424153, "state": "ID"}, + "laclede": {"lat": 48.167129, "lon": -116.751565, "state": "ID"}, + "lapwai": {"lat": 46.403715, "lon": -116.804223, "state": "ID"}, + "lava hot springs": {"lat": 42.620107, "lon": -112.009902, "state": "ID"}, + "lewiston": {"lat": 46.4165, "lon": -117.0177, "state": "ID"}, + "lewisville": {"lat": 43.695208, "lon": -112.013232, "state": "ID"}, + "lincoln": {"lat": 43.51825, "lon": -111.969215, "state": "ID"}, + "mackay": {"lat": 43.911996, "lon": -113.612728, "state": "ID"}, + "malad city": {"lat": 42.189909, "lon": -112.249688, "state": "ID"}, + "marsing": {"lat": 43.54636, "lon": -116.810422, "state": "ID"}, + "mccall": {"lat": 44.9111, "lon": -116.0987, "state": "ID"}, + "mccammon": {"lat": 42.648236, "lon": -112.189394, "state": "ID"}, + "melba": {"lat": 43.373633, "lon": -116.531933, "state": "ID"}, + "menan": {"lat": 43.721791, "lon": -111.992353, "state": "ID"}, + "meridian": {"lat": 43.6121, "lon": -116.3915, "state": "ID"}, + "middleton": {"lat": 43.711593, "lon": -116.615008, "state": "ID"}, + "montpelier": {"lat": 42.3232, "lon": -111.298, "state": "ID"}, + "moreland": {"lat": 43.21948, "lon": -112.437772, "state": "ID"}, + "moscow": {"lat": 46.7324, "lon": -117.0002, "state": "ID"}, + "mountain home": {"lat": 43.133, "lon": -115.6912, "state": "ID"}, + "mountain home afb": {"lat": 43.049186, "lon": -115.86586, "state": "ID"}, + "moyie springs": {"lat": 48.724746, "lon": -116.195421, "state": "ID"}, + "mud lake": {"lat": 43.842855, "lon": -112.479504, "state": "ID"}, + "mullan": {"lat": 47.468759, "lon": -115.796351, "state": "ID"}, + "nampa": {"lat": 43.5407, "lon": -116.5635, "state": "ID"}, + "new meadows": {"lat": 44.971335, "lon": -116.285195, "state": "ID"}, + "new plymouth": {"lat": 43.970417, "lon": -116.818781, "state": "ID"}, + "newdale": {"lat": 43.886385, "lon": -111.603888, "state": "ID"}, + "nezperce": {"lat": 46.233582, "lon": -116.241418, "state": "ID"}, + "notus": {"lat": 43.726863, "lon": -116.800432, "state": "ID"}, + "oakley": {"lat": 42.24206, "lon": -113.883058, "state": "ID"}, + "oldtown": {"lat": 48.182488, "lon": -117.018005, "state": "ID"}, + "orofino": {"lat": 46.484943, "lon": -116.253027, "state": "ID"}, + "osburn": {"lat": 47.505731, "lon": -116.000709, "state": "ID"}, + "paris": {"lat": 42.227978, "lon": -111.402424, "state": "ID"}, + "parker": {"lat": 43.958405, "lon": -111.759206, "state": "ID"}, + "parma": {"lat": 43.786284, "lon": -116.942491, "state": "ID"}, + "paul": {"lat": 42.605496, "lon": -113.784487, "state": "ID"}, + "payette": {"lat": 44.080093, "lon": -116.926852, "state": "ID"}, + "pierce": {"lat": 46.495335, "lon": -115.803292, "state": "ID"}, + "pinehurst": {"lat": 47.536314, "lon": -116.231746, "state": "ID"}, + "plummer": {"lat": 47.3344, "lon": -116.8856, "state": "ID"}, + "pocatello": {"lat": 42.8713, "lon": -112.4455, "state": "ID"}, + "ponderay": {"lat": 48.30478, "lon": -116.536645, "state": "ID"}, + "post falls": {"lat": 47.718, "lon": -116.9516, "state": "ID"}, + "potlatch": {"lat": 46.923493, "lon": -116.897713, "state": "ID"}, + "preston": {"lat": 42.0963, "lon": -111.8766, "state": "ID"}, + "priest river": {"lat": 48.18336, "lon": -116.884354, "state": "ID"}, + "rathdrum": {"lat": 47.8121, "lon": -116.895, "state": "ID"}, + "rexburg": {"lat": 43.826, "lon": -111.7897, "state": "ID"}, + "richfield": {"lat": 43.05164, "lon": -114.155942, "state": "ID"}, + "rigby": {"lat": 43.673587, "lon": -111.913525, "state": "ID"}, + "riggins": {"lat": 45.420591, "lon": -116.317636, "state": "ID"}, + "ririe": {"lat": 43.632494, "lon": -111.771717, "state": "ID"}, + "riverside": {"lat": 43.196554, "lon": -112.435625, "state": "ID"}, + "roberts": {"lat": 43.720488, "lon": -112.128868, "state": "ID"}, + "robie creek": {"lat": 43.667649, "lon": -116.015203, "state": "ID"}, + "rockford": {"lat": 43.189235, "lon": -112.530618, "state": "ID"}, + "rockford bay": {"lat": 47.508637, "lon": -116.886536, "state": "ID"}, + "rockland": {"lat": 42.573157, "lon": -112.87453, "state": "ID"}, + "rupert": {"lat": 42.618936, "lon": -113.673967, "state": "ID"}, + "salmon": {"lat": 45.1758, "lon": -113.8957, "state": "ID"}, + "sandpoint": {"lat": 48.2766, "lon": -116.5535, "state": "ID"}, + "shelley": {"lat": 43.379538, "lon": -112.126098, "state": "ID"}, + "shoshone": {"lat": 42.936185, "lon": -114.404747, "state": "ID"}, + "silverton": {"lat": 47.495681, "lon": -115.960513, "state": "ID"}, + "smelterville": {"lat": 47.542423, "lon": -116.177448, "state": "ID"}, + "soda springs": {"lat": 42.6543, "lon": -111.6047, "state": "ID"}, + "spirit lake": {"lat": 47.965799, "lon": -116.869831, "state": "ID"}, + "st. anthony": {"lat": 43.964839, "lon": -111.685049, "state": "ID"}, + "st. maries": {"lat": 47.314589, "lon": -116.572235, "state": "ID"}, + "stanley": {"lat": 44.216, "lon": -114.9311, "state": "ID"}, + "star": {"lat": 43.702788, "lon": -116.491025, "state": "ID"}, + "sugar city": {"lat": 43.87582, "lon": -111.751032, "state": "ID"}, + "sun valley": {"lat": 43.683852, "lon": -114.334203, "state": "ID"}, + "swan valley": {"lat": 43.442575, "lon": -111.324544, "state": "ID"}, + "teton": {"lat": 43.887773, "lon": -111.672254, "state": "ID"}, + "tetonia": {"lat": 43.814578, "lon": -111.158664, "state": "ID"}, + "troy": {"lat": 46.737981, "lon": -116.773154, "state": "ID"}, + "twin falls": {"lat": 42.563, "lon": -114.4609, "state": "ID"}, + "tyhee": {"lat": 42.954001, "lon": -112.456199, "state": "ID"}, + "ucon": {"lat": 43.593538, "lon": -111.959358, "state": "ID"}, + "victor": {"lat": 43.60155, "lon": -111.110822, "state": "ID"}, + "wallace": {"lat": 47.473578, "lon": -115.922542, "state": "ID"}, + "weippe": {"lat": 46.37825, "lon": -115.938844, "state": "ID"}, + "weiser": {"lat": 44.251, "lon": -116.969, "state": "ID"}, + "wendell": {"lat": 42.77468, "lon": -114.70296, "state": "ID"}, + "weston": {"lat": 42.036209, "lon": -111.977799, "state": "ID"}, + "wilder": {"lat": 43.678388, "lon": -116.907585, "state": "ID"}, + "winchester": {"lat": 46.240812, "lon": -116.624137, "state": "ID"}, + "worley": {"lat": 47.400533, "lon": -116.919254, "state": "ID"}, } diff --git a/work/tests/test_curation.py b/work/tests/test_curation.py index ebf3f48..47faf40 100644 --- a/work/tests/test_curation.py +++ b/work/tests/test_curation.py @@ -183,13 +183,16 @@ def test_api_list_towns(client): def test_api_post_add_town(client): + # A clearly-fictitious name, not a real Idaho town -- avoids colliding + # with the town_anchors seed (real curated Idaho/neighbor towns; a + # UNIQUE(name) constraint 400s on a duplicate insert). r = client.post("/api/town-anchors", json={ - "name": "Bellevue", "lat": 43.4670, "lon": -114.2557, "state": "ID", + "name": "Testopolis", "lat": 43.4670, "lon": -114.2557, "state": "ID", }) assert r.status_code == 200 - assert r.json()["name"] == "bellevue" + assert r.json()["name"] == "testopolis" invalidate_curation_cache() - coord = lookup_town_anchor("bellevue") + coord = lookup_town_anchor("testopolis") assert coord is not None diff --git a/work/tests/test_fire_refactor.py b/work/tests/test_fire_refactor.py index 8b1deea..da09eb3 100644 --- a/work/tests/test_fire_refactor.py +++ b/work/tests/test_fire_refactor.py @@ -217,7 +217,10 @@ class TestFormatterGolden: # 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" + # "oakley" is nearer to 42.197,-113.710 than "burley" and is now in + # the town_anchors seed after the seed-list sync (Fix 2), so it wins + # the anchor resolution instead of the previously-nearest seeded town. + assert new_wire == "✅ Cache Peak Fire — contained & closed\n1,847 ac | 23% contained | 9 mi E of Oakley" # ───────────────────────────────────────────────────────────────────────────── diff --git a/work/tests/test_wfigs_handler.py b/work/tests/test_wfigs_handler.py index c81728b..7fb2a21 100644 --- a/work/tests/test_wfigs_handler.py +++ b/work/tests/test_wfigs_handler.py @@ -357,8 +357,10 @@ def test_k_anchor_falls_to_nearest_town(monkeypatch, mem_db): county="Cassia") 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 + # Resolves anchor via town_anchors table (Oakley @ 42.24206, -113.883058 + # -- now the nearest seeded anchor to the incident's 42.197,-113.710 + # after the full-list seed sync; Burley is farther away) + assert "Oakley" in wire def test_k_anchor_falls_to_landclass(monkeypatch, mem_db): @@ -372,7 +374,8 @@ def test_k_anchor_falls_to_landclass(monkeypatch, mem_db): n = _normalize_wfigs(env) wire = _wfigs_render(n, prefix="New") # Resolves nearest town from town_anchors table, overriding landclass - assert "Burley" in wire + # (Oakley is nearest to 42.197,-113.710 after the full-list seed sync) + assert "Oakley" in wire def test_k_anchor_falls_to_county(monkeypatch, mem_db): @@ -385,7 +388,8 @@ def test_k_anchor_falls_to_county(monkeypatch, mem_db): n = _normalize_wfigs(env) wire = _wfigs_render(n, prefix="New") # Resolves nearest town from town_anchors table - assert "Burley" in wire + # (Oakley is nearest to 42.197,-113.710 after the full-list seed sync) + assert "Oakley" in wire def test_k_anchor_nearest_town_under_one_mile_says_near(monkeypatch, mem_db): @@ -398,7 +402,8 @@ def test_k_anchor_nearest_town_under_one_mile_says_near(monkeypatch, mem_db): n = _normalize_wfigs(env) wire = _wfigs_render(n, prefix="New") # Anchor resolved via town_anchors; exact format depends on distance - assert "Burley" in wire + # (Oakley is nearest to 42.197,-113.710 after the full-list seed sync) + assert "Oakley" in wire # ============================================================================