diff --git a/work/meshai/env/usgs.py b/work/meshai/env/usgs.py index c341f3d..641853c 100644 --- a/work/meshai/env/usgs.py +++ b/work/meshai/env/usgs.py @@ -8,6 +8,7 @@ import json import logging import time +from datetime import datetime from typing import TYPE_CHECKING, Optional from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen @@ -36,6 +37,35 @@ def _cfg_str(config, attr: str, default: str) -> str: return value if isinstance(value, str) and value else default +# Maps the flood_status strings _fetch() sets (see severity-classification +# block above) onto the ranked threshold_state vocabulary the shared hydro +# gating/formatter modules speak (meshai.central.idaho_gauge_sites. +# THRESHOLD_RANK / meshai.notifications.gating.hydro / formatters.hydro). +# Kept local rather than importing from central.idaho_gauge_sites so this +# adapter has no dependency on the (Central-only, reference-only) central +# package -- only the string vocabulary is shared, not the code. +_FLOOD_STATUS_TO_THRESHOLD_STATE = { + "Action Stage": "action", + "Minor Flood": "flood_minor", + "Moderate Flood": "flood_moderate", + "Major Flood": "flood_major", +} + + +def _parse_iso_epoch(s: Optional[str]) -> Optional[int]: + """Parse a USGS instantaneous-values dateTime string to an epoch int. + + USGS IV timestamps carry a UTC offset (e.g. "2026-06-04T12:00:00.000-06:00"); + accept a trailing "Z" too for robustness. Returns None on any parse failure. + """ + if not s: + return None + try: + return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp()) + except Exception: + return None + + class USGSStreamsAdapter: """USGS instantaneous values for stream gauge readings with NWS flood stages.""" @@ -526,6 +556,33 @@ class USGSStreamsAdapter: summary_parts.append(str(flood_status)) summary = " | ".join(summary_parts)[:300] + # Canonical hydro schema (event.data) — same shape the Central nwis + # path writes (site_id, gauge_name, stage_ft, flow_cfs, unit, + # threshold_state, reading_time, lat, lon, parameter_code), so the + # shared gating.hydro.decide() / formatters.hydro.format() work + # unmodified for both sources. flood_status is only ever computed + # for gage-height (00065) readings above, so a reading that reaches + # this point is always a stage reading -- parameter_code is fixed + # to "00065" and value maps to stage_ft (never flow_cfs; the + # companion discharge reading, if any, arrives as its own separate + # evt with no flood_status and never reaches to_event()). + reading_time = _parse_iso_epoch(props.get("timestamp")) + if reading_time is None: + fetched_at = evt.get("fetched_at") + reading_time = int(fetched_at) if isinstance(fetched_at, (int, float)) else int(time.time()) + data = { + "site_id": props.get("site_id"), + "gauge_name": props.get("site_name") or title, + "stage_ft": value, + "flow_cfs": None, + "unit": unit, + "threshold_state": _FLOOD_STATUS_TO_THRESHOLD_STATE.get(str(flood_status), "normal"), + "reading_time": reading_time, + "lat": lat, + "lon": lon, + "parameter_code": "00065", + } + # event_id is already the stable "{site_id}_{param}" key. Re-polls of # the same gauge/parameter coalesce on this group_key; using it as the # sole inhibit_key lets the pipeline Inhibitor suppress lower-severity @@ -543,6 +600,7 @@ class USGSStreamsAdapter: lon=lon, group_key=event_id, inhibit_keys=[event_id], + data=data, ) except Exception: logger.exception(f"USGS to_event failed for evt: {evt.get('event_id')}") diff --git a/work/meshai/notifications/cutover.py b/work/meshai/notifications/cutover.py index 7604610..7524053 100644 --- a/work/meshai/notifications/cutover.py +++ b/work/meshai/notifications/cutover.py @@ -40,8 +40,22 @@ import os # Central-render impact. Consumed by store._emit_event (decider gate) and # renderers.composer.compose_mesh_message (formatter gate) so native fires take # the exact same single render path a cut-over category would. +# +# stream_flood_warning / stream_high_water (env/usgs.py) are the same shape of +# problem: they are emitted ONLY by the native USGS adapter (the Central nwis +# path emits the separate `stream_flow` category instead -- see +# notifications/gating/__init__.py and notifications/formatters/__init__.py). +# Before this fix neither category had a registered decider/formatter at all, +# so every elevated gauge reading was silently dropped (no formatter -> no +# renderable wire; get_toggle() resolving to "seismic" made no difference +# since nothing was registered under that name either). Forcing them onto the +# shared hydro decider+formatter here, like fire above, has no Central-render +# impact (Central never emits these two category strings). NATIVE_ALWAYS_DECIDE = frozenset( - {"wildfire_incident", "wildfire_declared", "wildfire_closed"} + { + "wildfire_incident", "wildfire_declared", "wildfire_closed", + "stream_flood_warning", "stream_high_water", + } ) diff --git a/work/meshai/notifications/formatters/__init__.py b/work/meshai/notifications/formatters/__init__.py index a2bfb87..2da821c 100644 --- a/work/meshai/notifications/formatters/__init__.py +++ b/work/meshai/notifications/formatters/__init__.py @@ -93,11 +93,17 @@ register("traffic_congestion", _incident_fmt_mod.format) # Phase-3: USGS NWIS stream-gauge hydro. The Central nwis path maps every # `central.hydro.*` envelope to the flat category `stream_flow` (see -# central.consumer.map_category / category_from_subject), so that is the real -# registry key. Native env/usgs.py categories (stream_flood_warning / -# stream_high_water) are deferred — see gating/__init__.py note. +# central.consumer.map_category / category_from_subject). Native env/usgs.py +# categories (stream_flood_warning / stream_high_water) reuse the same +# formatter — env/usgs.py's to_event() now populates event.data with the same +# canonical schema (gauge_name/threshold_state/stage_ft/flow_cfs/lat/lon) the +# Central path writes, so the wire renders identically for both sources. Forced +# onto the live render path unconditionally via cutover.NATIVE_ALWAYS_DECIDE +# (see gating/__init__.py note) since Central never emits these two strings. from meshai.notifications.formatters import hydro as _hydro_fmt_mod # noqa: E402,F401 register("stream_flow", _hydro_fmt_mod.format) +register("stream_flood_warning", _hydro_fmt_mod.format) +register("stream_high_water", _hydro_fmt_mod.format) # Phase-3b: WFIGS wildfire. The Central wfigs_handler emits THREE explicit # categories, all under toggle "fire": `wildfire_declared` (first-sight New, diff --git a/work/meshai/notifications/gating/__init__.py b/work/meshai/notifications/gating/__init__.py index 550e5af..f6965f1 100644 --- a/work/meshai/notifications/gating/__init__.py +++ b/work/meshai/notifications/gating/__init__.py @@ -84,12 +84,18 @@ register("traffic_congestion", _incident_gate_mod.decide) # Phase-3: USGS NWIS stream-gauge hydro. Registered under `stream_flow` — the # flat category the Central nwis path produces for every `central.hydro.*` -# envelope (map_category "hydro." -> "stream_flow"). Native usgs categories -# (stream_flood_warning / stream_high_water, emitted only by env/usgs.py) are a -# deferred follow-up: env/usgs.py is NOT migrated this phase and, since hydro is -# NOT cut over, store._emit_event's native decider hook won't run it. +# envelope (map_category "hydro." -> "stream_flow") — AND under the two +# threshold-classified categories env/usgs.py actually emits natively: +# stream_flood_warning (at/above flood stage) and stream_high_water (action +# stage). All three share the same hydro.decide() gate; store._emit_event +# forces the native two onto the live decider path unconditionally via +# cutover.NATIVE_ALWAYS_DECIDE (mirrors the native fire categories below), +# independent of the MESHAI_CUTOVER_CATEGORIES shadow-bake env var, since +# Central never emits those two category strings. from meshai.notifications.gating import hydro as _hydro_gate_mod # noqa: E402,F401 register("stream_flow", _hydro_gate_mod.decide) +register("stream_flood_warning", _hydro_gate_mod.decide) +register("stream_high_water", _hydro_gate_mod.decide) # Phase-3b: WFIGS wildfire. Three explicit categories the wfigs_handler emits: # `wildfire_declared` (New), `wildfire_incident` (growth Update), and diff --git a/work/meshai/notifications/gating/hydro.py b/work/meshai/notifications/gating/hydro.py index 6ac578b..ccc38ab 100644 --- a/work/meshai/notifications/gating/hydro.py +++ b/work/meshai/notifications/gating/hydro.py @@ -24,7 +24,8 @@ Broadcast rule (verbatim from the old handler): IMPORTANT — division of labour with the handler: * The append-only gauge_readings INSERT stays INLINE in the Central handler (unconditional time-series write). This decider only READS gauge_readings - to determine prior state; it NEVER writes it. + to determine prior state for source="nwis"; it NEVER writes for that + source. See "Native persistence" below for source="usgs". * The decider MUST run BEFORE the handler's INSERT (the old handler did its prior-SELECT and 00060 back-look before inserting the new row), so the reads never see the current reading. The prior-SELECT additionally @@ -34,6 +35,34 @@ IMPORTANT — division of labour with the handler: event_log.handled flip stays handler-owned (mirrors the old _attach_commit, wrapped in the cutover branch like quake_handler). +Native persistence (fix/native-gauge-flood-alerts): + env/usgs.py (source="usgs") has no handler module of its own with a + persistence connection the way central.nwis_handler does for source="nwis" + -- it is a pure HTTP-fetch-and-translate adapter. Central stopped writing + gauge_readings when nwis_handler stopped running, so without SOME writer + every native prior-state SELECT above would return "normal" (no row) + forever, and every elevated reading would rank as an upward crossing and + rebroadcast on every 15-minute tick instead of once per crossing. This + decider owns that write for source != "nwis" (see _persist_native_reading + below), keeping the "decider reads-only, handler writes" contract intact + for the Central source. + + KNOWN LIMITATION: env/usgs.py's to_event() only ever emits ELEVATED + readings (action stage or above) by design -- a routine/below-action + reading has no flood_status and is intentionally never turned into an + Event, so it never reaches this decider and a "back to normal" row is + never written. If a gauge fully recedes below action stage and later + re-crosses into action stage, the most recent gauge_readings row for that + site is still the last ELEVATED tier it saw (not "normal"), so the + re-crossing will compare equal-rank and be SUPPRESSED rather than + broadcast as a new crossing. This degrades toward silence, not spam (the + safe direction), and only resolves once the gauge reaches a HIGHER tier + than it last alerted at. Fixing it properly would mean also emitting + normal-state readings from env/usgs.py so a recede-to-normal row gets + written, which is a materially bigger behavior change to the native + emission/inhibition pipeline than this fix's scope -- left for the + Central rip-out follow-up. + data_patch keys (populated on EVERY call, broadcast or suppress, so the handler's unconditional INSERT + render use the back-looked values): threshold_state : str — resolved band (post-00060 back-look) @@ -124,6 +153,14 @@ def decide(data: dict, *, source: str, now: float) -> GateResult: # Resolved values returned so the handler's inline INSERT + render match. patch: dict = {"threshold_state": threshold_state, "stage_ft": stage_ft} + # Native persistence: own the write for the native source only (source= + # "nwis" keeps its inline handler-owned INSERT, per the module docstring). + # Must run AFTER the reads above (same ordering rule as the Central + # handler's INSERT) so this reading is never its own "prior". + if source != "nwis": + _persist_native_reading(conn, data, threshold_state=threshold_state, + stage_ft=stage_ft, now=now) + prior_rank = _rank(prior_state) cur_rank = _rank(threshold_state) @@ -150,3 +187,39 @@ def decide(data: dict, *, source: str, now: float) -> GateResult: reason=f"crossing {prior_state}->{threshold_state}", data_patch=patch, commit=None, ) + + +def _persist_native_reading(conn, data: dict, *, threshold_state: str, + stage_ft: Optional[float], now: float) -> None: + """Unconditional INSERT of a native (env/usgs.py) gauge reading. + + Mirrors central.nwis_handler's inline INSERT into the same table with the + same column shape, but owned here because env/usgs.py has no handler + module of its own with a persistence connection. Runs regardless of the + broadcast decision -- exactly like the Central handler's INSERT -- so the + NEXT native reading's prior-state SELECT above sees this one. Never + raises: a persistence failure degrades the next lookup (falls back to + "no prior" / first-crossing) rather than blocking the current decision. + """ + reading_time = data.get("reading_time") + if not isinstance(reading_time, (int, float)): + reading_time = now + try: + conn.execute( + "INSERT INTO gauge_readings(site_id, gauge_name, reading_value, " + "reading_unit, threshold_state, flow_cfs, reading_time, lat, lon) " + "VALUES (?,?,?,?,?,?,?,?,?)", + ( + data.get("site_id"), + data.get("gauge_name"), + stage_ft, + data.get("unit"), + threshold_state, + data.get("flow_cfs"), + int(reading_time), + data.get("lat"), + data.get("lon"), + ), + ) + except Exception: + logger.exception("hydro decide: native gauge_readings persist failed") diff --git a/work/tests/test_native_hydro_gauge_alerts.py b/work/tests/test_native_hydro_gauge_alerts.py new file mode 100644 index 0000000..c47abfe --- /dev/null +++ b/work/tests/test_native_hydro_gauge_alerts.py @@ -0,0 +1,323 @@ +"""fix/native-gauge-flood-alerts: native USGS gauge flood alerts are renderable. + +Before this fix, env/usgs.py emitted `stream_flood_warning` / `stream_high_water` +Events that had NO registered decider and NO registered formatter -- they were +silently dropped by store._emit_event / compose_mesh_message (see +notifications/gating/__init__.py's old "deferred follow-up" comment and +notifications/formatters/__init__.py's old "deferred" comment, both since +updated). This file proves the fix along three axes: + +1. Registration: both native categories now resolve a decider AND a formatter + (mirrors TestRegistration in test_hydro_refactor.py, which covers the + Central-only `stream_flow` category). +2. Gate behavior: a first elevated reading broadcasts, a sustained (same-band) + reading suppresses, an escalation broadcasts again -- reusing the SAME + hydro.decide() gate the Central path uses, now also fed by env/usgs.py's + to_event() (which populates event.data with the canonical schema the gate + and formatter both expect -- previously event.data was left empty for + native events). +3. Golden wire format: formatters.hydro.format() renders a native event the + same way it renders a Central `stream_flow` event (test_hydro_refactor.py + already proves that formatter is byte-identical to the old + central.nwis_handler._render()) -- captured here for the native shape + specifically, where flow_cfs is always None (env/usgs.py's to_event() only + ever emits stage/height readings; a paired discharge reading, if any, + arrives as its own separate Event with no flood_status and never reaches + to_event() -- see the docstring added to to_event()/decide() for detail). +""" +from __future__ import annotations + +import time +from unittest.mock import MagicMock + +import pytest + +from meshai.env.usgs import USGSStreamsAdapter +from meshai.persistence import close_thread_connection, init_db +from meshai.persistence import db as persistence_db +from tests.harness.goldens import assert_byte_identical + + +# ── DB fixture (same shape as test_hydro_refactor.py / test_nwis_handler.py) ─ + +@pytest.fixture +def mem_db(monkeypatch, tmp_path): + db_path = str(tmp_path / "native-hydro-test.sqlite") + monkeypatch.setenv("MESHAI_DB_PATH", db_path) + persistence_db._initialised.clear() + close_thread_connection() + conn = init_db() + yield conn + close_thread_connection() + persistence_db._initialised.discard(db_path) + + +@pytest.fixture +def adapter(): + config = MagicMock() + config.sites = [] + config.tick_seconds = 900 + config.flood_thresholds = {} + return USGSStreamsAdapter(config) + + +def make_reading(*, site_id="13186000", site_name="Snake River at Heise", + value, flood_status, ts, lat=43.612, lon=-111.654): + """Mirrors the internal event dict env/usgs.py's _fetch() stores.""" + now = time.time() + return { + "source": "usgs", + "event_id": f"{site_id}_height", + "event_type": "Stream Gauge", + "headline": f"{site_name}: {value} ft" + (f" — {flood_status}" if flood_status else ""), + "severity": "priority" if flood_status and "Flood" in flood_status else "routine", + "lat": lat, + "lon": lon, + "expires": now + 1800, + "fetched_at": now, + "properties": { + "site_id": site_id, + "site_name": site_name, + "parameter": "Gage height", + "value": value, + "unit": "ft", + "timestamp": ts, + "flood_status": flood_status, + "flood_stages": {"action_stage": 9.0, "flood_stage": 10.5}, + }, + } + + +# ───────────────────────────────────────────────────────────────────────────── +# 1. Registration — both native categories resolve a decider AND a formatter +# ───────────────────────────────────────────────────────────────────────────── + +class TestRegistration: + @pytest.mark.parametrize("category", ["stream_flood_warning", "stream_high_water"]) + def test_decider_registered(self, category): + from meshai.notifications.gating import get_decider + from meshai.notifications.gating.hydro import decide + assert get_decider(category) is decide + + @pytest.mark.parametrize("category", ["stream_flood_warning", "stream_high_water"]) + def test_formatter_registered(self, category): + from meshai.notifications.formatters import get_formatter + from meshai.notifications.formatters.hydro import format as hfmt + assert get_formatter(category) is hfmt + + @pytest.mark.parametrize("category", ["stream_flood_warning", "stream_high_water"]) + def test_decider_is_not_the_earthquake_decider(self, category): + """Regression guard: these categories must never resolve to the + seismic/earthquake gate, even though get_toggle() names the shared + family 'seismic' (see test_water_v057.py — that mapping is + intentional and pre-existing: stream_flow already lives on the same + toggle, and every water/hydro registry entry is guarded by + test_alert_categories_water_complete).""" + from meshai.notifications.gating import get_decider + from meshai.notifications.gating.quake import decide as quake_decide + assert get_decider(category) is not quake_decide + + @pytest.mark.parametrize("category", ["stream_flood_warning", "stream_high_water"]) + def test_native_always_decide(self, category): + """Both categories are forced onto the live decider+formatter path + unconditionally (independent of MESHAI_CUTOVER_CATEGORIES), exactly + like the native WFIGS fire categories -- required because Central + never emits these two category strings, so there is no shadow-bake + window to wait out.""" + from meshai.notifications.cutover import NATIVE_ALWAYS_DECIDE + assert category in NATIVE_ALWAYS_DECIDE + + +# ───────────────────────────────────────────────────────────────────────────── +# 2. Gate behavior — first crossing broadcasts, sustained state suppresses +# ───────────────────────────────────────────────────────────────────────────── + +class TestGateBehavior: + def _decide(self, adapter, evt): + from meshai.notifications.gating import get_decider + from meshai.notifications import clock + event = adapter.to_event(evt) + assert event is not None, "adapter unexpectedly suppressed the reading" + decider = get_decider(event.category) + gate = decider(event.data, source=event.source, now=clock.now()) + event.data.update(gate.data_patch) + return event, gate + + def test_first_elevated_reading_broadcasts(self, adapter, mem_db): + """A fresh site (no prior gauge_readings row) at action stage is a + first-crossing -- 'no prior' degrades gracefully to normal->action, + which broadcasts.""" + evt = make_reading(value=9.2, flood_status="Action Stage", + ts="2026-07-17T10:00:00-06:00") + event, gate = self._decide(adapter, evt) + assert event.category == "stream_high_water" + assert gate.broadcast is True + assert gate.lifecycle == "new" + + def test_routine_reading_never_reaches_the_gate(self, adapter, mem_db): + """A below-action reading has no flood_status; the adapter drops it + before to_event() ever returns an Event (unchanged pre-fix behavior + -- this fix does not change what gets emitted, only what happens to + what was already being emitted).""" + evt = make_reading(value=5.0, flood_status=None, + ts="2026-07-17T10:00:00-06:00") + assert adapter.to_event(evt) is None + + def test_sustained_elevated_reading_suppresses(self, adapter, mem_db): + """Same site, same band, next tick: does NOT re-broadcast.""" + first = make_reading(value=9.2, flood_status="Action Stage", + ts="2026-07-17T10:00:00-06:00") + second = make_reading(value=9.3, flood_status="Action Stage", + ts="2026-07-17T10:15:00-06:00") + _, gate1 = self._decide(adapter, first) + assert gate1.broadcast is True + _, gate2 = self._decide(adapter, second) + assert gate2.broadcast is False + assert "unchanged band" in gate2.reason + + def test_escalation_broadcasts_again(self, adapter, mem_db): + """Action stage -> minor flood on the same site is a second, higher + crossing -- broadcasts again.""" + action = make_reading(value=9.2, flood_status="Action Stage", + ts="2026-07-17T10:00:00-06:00") + flood = make_reading(value=10.8, flood_status="Minor Flood", + ts="2026-07-17T10:15:00-06:00") + _, gate1 = self._decide(adapter, action) + assert gate1.broadcast is True + event2, gate2 = self._decide(adapter, flood) + assert event2.category == "stream_flood_warning" + assert gate2.broadcast is True + assert gate2.lifecycle == "new" + + def test_native_write_does_not_leak_into_central_source(self, adapter, mem_db): + """Sanity: the native persistence write in gating.hydro.decide() is + gated on source != 'nwis'. Calling decide() directly with + source='nwis' (the Central source) must NOT insert a row -- the + Central handler owns its own inline INSERT, and decide() must stay + read-only for that source (see hydro.py module docstring).""" + from meshai.notifications.gating.hydro import decide + canonical = { + "site_id": "USGS-99999999", "gauge_name": "Should Not Persist", + "stage_ft": 12.0, "flow_cfs": None, "unit": "ft", + "threshold_state": "action", "reading_time": 1_700_000_000, + "lat": 44.0, "lon": -115.0, "parameter_code": "00065", + } + decide(canonical, source="nwis", now=1_700_000_000.0) + row = mem_db.execute( + "SELECT COUNT(*) AS n FROM gauge_readings WHERE site_id=?", + ("USGS-99999999",), + ).fetchone() + assert row["n"] == 0 + + +# ───────────────────────────────────────────────────────────────────────────── +# 3. Golden wire format — native event renders via the shared hydro formatter +# ───────────────────────────────────────────────────────────────────────────── + +class TestFormatterGolden: + """Captured-from-current-behavior goldens for the native shape. + + Unlike a Central envelope (which can carry a paired 00060 discharge value + via the 00060 back-look), a native reading NEVER carries flow_cfs -- see + to_event()'s docstring/comment: only stage/height readings reach + to_event() at all, so flow_cfs is always None here. The stage/label/coords + segments are otherwise identical to the Central-path golden in + test_hydro_refactor.py::TestFormatterGolden, since both flow through the + exact same formatters.hydro.format(). + """ + + def _render(self, adapter, evt): + from meshai.notifications.gating import get_decider + from meshai.notifications.formatters.hydro import format as hfmt + from meshai.notifications import clock + event = adapter.to_event(evt) + decider = get_decider(event.category) + gate = decider(event.data, source=event.source, now=clock.now()) + event.data.update(gate.data_patch) + return hfmt(event, now=clock.now(), budget=140) + + def test_high_water_wire(self, adapter, mem_db): + evt = make_reading( + site_name="Snake River at Heise", value=9.2, + flood_status="Action Stage", ts="2026-07-17T10:00:00-06:00", + lat=43.612, lon=-111.654, + ) + wire = self._render(adapter, evt) + assert wire == "🌊 New: Snake River at Heise: action stage 9.2 ft, @ 43.612,-111.654" + + def test_flood_warning_wire(self, adapter, mem_db): + evt = make_reading( + site_name="Boise River", value=14.5, + flood_status="Minor Flood", ts="2026-07-17T10:00:00-06:00", + lat=43.600, lon=-116.200, + ) + wire = self._render(adapter, evt) + assert wire == "🌊 New: Boise River: minor flooding 14.5 ft, @ 43.600,-116.200" + + def test_major_flood_wire(self, adapter, mem_db): + evt = make_reading( + site_name="Test Gauge", value=20.0, + flood_status="Major Flood", ts="2026-07-17T10:00:00-06:00", + lat=44.0, lon=-114.0, + ) + wire = self._render(adapter, evt) + assert wire == "🌊 New: Test Gauge: major flooding 20.0 ft, @ 44.000,-114.000" + + def test_missing_coords_drops_at_tail(self, adapter, mem_db): + """Byte-identical drop behavior to the Central-path golden for the + same case (test_hydro_refactor.py::test_missing_coords_drops_at_tail).""" + evt = make_reading( + site_name="No Coords Gauge", value=10.0, + flood_status="Action Stage", ts="2026-07-17T10:00:00-06:00", + lat=None, lon=None, + ) + assert adapter.to_event(evt) is None, "to_event() requires lat/lon" + + # Exercise the formatter directly with coords stripped after the fact + # to prove the "@ ..." segment is correctly omitted, mirroring the + # Central golden (to_event() itself refuses a coord-less reading, so + # this checks the formatter behavior the same way + # test_hydro_refactor.py does: via a synthetic canonical dict). + from meshai.notifications.formatters.hydro import format as hfmt + + class _FakeEvent: + pass + e = _FakeEvent() + e.data = { + "gauge_name": "No Coords Gauge", "threshold_state": "action", + "stage_ft": 10.0, "flow_cfs": None, "unit": "ft", + "lat": None, "lon": None, + } + wire = hfmt(e, now=1_700_000_000.0, budget=140) + assert_byte_identical( + wire, "🌊 New: No Coords Gauge: action stage 10.0 ft" + ) + assert "@" not in wire + + +# ───────────────────────────────────────────────────────────────────────────── +# 4. End-to-end: compose_mesh_message renders the same wire (full pipeline) +# ───────────────────────────────────────────────────────────────────────────── + +def test_compose_mesh_message_end_to_end(adapter, mem_db): + """Full pipeline: to_event() -> decider -> compose_mesh_message(), the + same call sequence store._emit_event() + the mesh dispatcher make in + production. Proves the NATIVE_ALWAYS_DECIDE gate actually takes effect + for compose_mesh_message's formatter dispatch, not just get_formatter() + resolution.""" + from meshai.notifications.gating import get_decider + from meshai.notifications.renderers.composer import compose_mesh_message + from meshai.notifications import clock + + evt = make_reading( + site_name="Snake River at Heise", value=9.2, + flood_status="Action Stage", ts="2026-07-17T10:00:00-06:00", + lat=43.612, lon=-111.654, + ) + event = adapter.to_event(evt) + decider = get_decider(event.category) + gate = decider(event.data, source=event.source, now=clock.now()) + assert gate.broadcast is True + event.data.update(gate.data_patch) + wire = compose_mesh_message(event) + assert wire == "🌊 New: Snake River at Heise: action stage 9.2 ft, @ 43.612,-111.654"