fix(hydro): restore USGS stream-gauge flood alerts (silently dead since the all-native flip) (#156)

* fix(hydro): make native USGS gauge flood alerts renderable

env/usgs.py emits stream_flood_warning / stream_high_water Events for
elevated stream gauges, but neither category had a registered gating
decider or formatter, and event.data was left empty -- so every
detected flood/high-water reading was silently dropped before it ever
reached the mesh (get_decider/get_formatter both returned None, and
compose_mesh_message fell through with nothing to render).

- Register the existing hydro.decide()/hydro.format() (already used by
  the Central-only `stream_flow` category) under stream_flood_warning
  and stream_high_water too -- same shared-decider/formatter pattern
  already used for avalanche_warning/_watch, weather_*, wildfire_*, and
  emergency_*.
- Add both categories to cutover.NATIVE_ALWAYS_DECIDE so the decider and
  formatter actually run unconditionally (mirrors the native WFIGS fire
  categories): Central never emits these two category strings, so there
  is no shadow-bake window to wait out.
- env/usgs.py's to_event() now populates event.data with the canonical
  hydro schema (site_id, gauge_name, stage_ft, flow_cfs, unit,
  threshold_state, reading_time, lat, lon, parameter_code) the shared
  gate/formatter expect, mapping the adapter's flood_status strings onto
  the ranked threshold_state vocabulary.
- gating/hydro.py's decide() now also OWNS an unconditional
  gauge_readings INSERT for the native source (source != "nwis"): the
  table had no writer since Central's nwis_handler stopped running
  (2026-07-05), so every native prior-state lookup returned "normal"
  forever and every elevated reading would have rebroadcast on every
  15-minute tick instead of once per crossing. The Central source keeps
  its own inline handler-owned INSERT unchanged (decide() stays
  read-only for source="nwis").

Left as-is (not this fix): the toggle mapping for these categories
(get_toggle() -> "seismic") is unchanged. It already matches the
sibling `stream_flow` category and is enforced by
test_water_v057.py::test_existing_hydro_entries_unchanged /
test_water_categories_have_required_fields; there is no separate
water/flood toggle family in VALID_TOGGLES, and inventing one is a
config-surface change outside this fix's scope.

Known limitation documented in gating/hydro.py: since to_event() only
ever emits elevated readings (never a "back to normal" reading), a full
recede-to-normal followed by a later re-crossing into the same tier
will not re-broadcast until a higher tier is reached -- degrades toward
silence, not spam.

* test(hydro): cover native gauge flood-alert registration, gating, and wire format

Registration tests prove stream_flood_warning/stream_high_water now
resolve a decider AND a formatter (the thing that was broken), that
neither resolves to the earthquake decider despite sharing the
"seismic" toggle name, and that both are in NATIVE_ALWAYS_DECIDE.

Gate tests drive env/usgs.py's to_event() through the real
gating.hydro.decide(): a first elevated reading broadcasts (graceful
no-prior-data handling), a sustained same-band reading suppresses, an
escalation broadcasts again, and a routine reading never reaches the
gate at all (unchanged pre-fix adapter behavior). A dedicated test
confirms the new native-persistence write in decide() does not leak
into the Central source="nwis" path.

Golden formatter tests pin the wire string for a high-water and two
flood-warning tiers, plus the missing-coords drop case -- all rendered
through the same formatters.hydro.format() the Central `stream_flow`
path uses (test_hydro_refactor.py already proves that formatter is
byte-identical to the old central.nwis_handler._render()). Native
events never carry flow_cfs (to_event() only ever emits stage/height
readings), so that segment's absence is captured explicitly as current
behavior, not ported from Central.

An end-to-end test drives to_event() -> decider -> compose_mesh_message
to prove the NATIVE_ALWAYS_DECIDE gate takes effect for the actual mesh
render path, not just formatter/decider resolution in isolation.

---------

Co-authored-by: Matt Johnson <mj@k7zvx.com>

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')}")

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",
+    }
 )
 
 

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,

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

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

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"