refactor(phase3a): migrate hydro/nwis to formatter+decider (tier-a)

Move nwis wire rendering into notifications/formatters/hydro.py and the
threshold-crossing decision into notifications/gating/hydro.py, behind
the registry under the real category `stream_flow`. handle_nwis now
builds a canonical dict, calls decide(), keeps the gauge_readings INSERT
inline (append-only), and branches on is_cutover("stream_flow") exactly
like quake_handler — legacy _attach_commit path stays byte-identical
while the new path bakes in shadow. No cutover.

- gauge_readings INSERT stays inline; decider only reads prior state
- hydro has no per-event broadcast-state table → GateResult.commit=None;
  event_log.handled flip stays handler-owned in the cutover wrapper
- tier-a: golden byte-identical wire + gate-sequence parity (12 tests)
- native env/usgs.py deferred (different category vocab, no
  threshold_state); non-cutover so store._emit_event won't run it

Suite at 34-failure baseline.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

work/meshai/central/nwis_handler.py

@@ -58,7 +58,6 @@ from datetime import datetime
 from typing import Any, Optional
 
 from meshai.central.idaho_gauge_sites import (
-    THRESHOLD_RANK,
     compute_threshold_state,
     lookup_site,
     normalize_site_id,
@@ -165,6 +164,22 @@ def handle_nwis(envelope: dict, subject: str,
     lat = d.get("latitude") if isinstance(d.get("latitude"), (int, float)) else site_meta.get("lat")
     lon = d.get("longitude") if isinstance(d.get("longitude"), (int, float)) else site_meta.get("lon")
 
+    # Build the canonical dict the decider + formatter read. parameter_code is
+    # carried so the decider can perform the 00060 discharge back-look; it is
+    # not part of the formatter's wire schema.
+    canonical: dict = {
+        "site_id": site_id,
+        "gauge_name": site_meta["gauge_name"],
+        "stage_ft": stage_ft,
+        "flow_cfs": flow_cfs,
+        "unit": unit,
+        "threshold_state": threshold_state,
+        "reading_time": reading_time,
+        "lat": lat,
+        "lon": lon,
+        "parameter_code": pc,
+    }
+
     # Always log the envelope to event_log. Initial handled=0; commit
     # callback flips to 1 if we actually broadcast.
     log_id = _log_event_returning_id(
@@ -173,31 +188,23 @@ def handle_nwis(envelope: dict, subject: str,
         subject=subject, handled=0,
         table_name="gauge_readings", table_pk=site_id)
 
-    # SELECT most recent prior reading (for this site, any parameter) to
-    # detect upward threshold crossing. Use threshold_state column directly.
-    prior = conn.execute(
-        "SELECT threshold_state FROM gauge_readings "
-        "WHERE site_id=? AND reading_time < ? "
-        "ORDER BY reading_time DESC LIMIT 1",
-        (site_id, reading_time),
-    ).fetchone()
-    prior_state = prior["threshold_state"] if prior else "normal"
+    # Delegate the threshold-crossing decision (prior-reading SELECT, 00060
+    # stage back-look, THRESHOLD_RANK upward-crossing check, broadcast_on_recede
+    # toggle) to the gating module. It reads gauge_readings for prior state
+    # BEFORE the inline INSERT below, preserving the original ordering. It
+    # returns the resolved (back-looked) threshold_state + stage_ft in
+    # data_patch so the INSERT + render use the same values on every path.
+    from meshai.notifications.gating.hydro import decide as _gate_decide
+    gate = _gate_decide(canonical, source="nwis", now=float(now))
+    threshold_state = gate.data_patch.get("threshold_state", threshold_state)
+    stage_ft = gate.data_patch.get("stage_ft", stage_ft)
+    canonical["threshold_state"] = threshold_state
+    canonical["stage_ft"] = stage_ft
 
-    # If this envelope is a 00060 (discharge) reading, look back for the
-    # latest 00065 stage reading at this site so the wire string can carry
-    # both. The threshold_state of THIS row inherits from that prior stage
-    # reading (discharge alone doesn't define a threshold band).
-    if pc == "00060":
-        last_stage = conn.execute(
-            "SELECT reading_value, threshold_state FROM gauge_readings "
-            "WHERE site_id=? AND reading_unit='ft' "
-            "ORDER BY reading_time DESC LIMIT 1",
-            (site_id,)).fetchone()
-        if last_stage:
-            stage_ft = last_stage["reading_value"]
-            threshold_state = last_stage["threshold_state"] or "normal"
-
-    # INSERT the new reading row. Always persist (time-series semantics).
+    # INSERT the new reading row INLINE (append-only time-series). Always
+    # persist, regardless of the broadcast decision, and AFTER the decider's
+    # reads so they never see the current row. Per the refactor plan this
+    # write stays handler-owned; the decider only READS gauge_readings.
     conn.execute(
         "INSERT INTO gauge_readings(site_id, gauge_name, reading_value, "
         "reading_unit, threshold_state, flow_cfs, reading_time, lat, lon) "
@@ -206,30 +213,48 @@ def handle_nwis(envelope: dict, subject: str,
          threshold_state, flow_cfs, reading_time, lat, lon),
     )
 
-    # Upward-crossing check.
-    try:
-        prior_rank = THRESHOLD_RANK.index(prior_state)
-    except ValueError:
-        prior_rank = 0   # unknown prior -> treat as normal
-    try:
-        cur_rank = THRESHOLD_RANK.index(threshold_state)
-    except ValueError:
-        cur_rank = 0
-
-    if cur_rank == prior_rank:
-        # Unchanged band -- no broadcast.
-        return None
-    if cur_rank < prior_rank and not bool(adapter_config.usgs_nwis.broadcast_on_recede):
-        # Receding without the recede toggle -- silent.
+    if not gate.broadcast:
         return None
 
-    wire = _render(gauge_name=site_meta["gauge_name"],
+    # Cutover gate: mirror quake_handler. When "stream_flow" is cut over, write
+    # gate.data_patch and wrap gate.commit so it also flips event_log.handled.
+    # Hydro has no per-event broadcast-state table, so gate.commit is None and
+    # the wrapper carries only the event_log flip. Otherwise old-style
+    # _attach_commit keeps the live broadcast byte-for-byte identical while the
+    # new formatter+decider bake in shadow.
+    from meshai.notifications.cutover import is_cutover
+    if isinstance(data, dict):
+        data.update(canonical)
+        if is_cutover("stream_flow"):
+            data.update(gate.data_patch)
+            data["_broadcast_audit"] = {"table": "gauge_readings", "pk": site_id}
+
+            _raw_commit = gate.commit
+            _log_row_id = log_id
+
+            def _on_commit(committed_at: float) -> None:
+                if _raw_commit is not None:
+                    _raw_commit(committed_at)
+                if _log_row_id is not None:
+                    try:
+                        c = get_db()
+                        c.execute("UPDATE event_log SET handled=1 WHERE id=?",
+                                  (int(_log_row_id),))
+                    except Exception:
+                        logger.exception("nwis commit: event_log update failed")
+
+            data["_on_broadcast_committed"] = _on_commit
+        else:
+            _attach_commit(data, site_id=site_id, event_log_row_id=log_id)
+
+    # Return _render() wire for backward-compat (existing call-sites + tests).
+    # At dispatch time compose_mesh_message() uses the registered formatter
+    # (formatters/hydro.py) on cutover, re-rendering byte-identically from data.
+    return _render(gauge_name=site_meta["gauge_name"],
                     threshold_state=threshold_state,
                     stage_ft=stage_ft, flow_cfs=flow_cfs,
                     unit=unit if pc == "00065" else "ft",
                     lat=lat, lon=lon)
-    _attach_commit(data, site_id=site_id, event_log_row_id=log_id)
-    return wire
 
 
 # ---- renderer ------------------------------------------------------------

work/meshai/notifications/formatters/__init__.py

@@ -86,3 +86,11 @@ register("work_zone",          _incident_fmt_mod.format)
 register("road_incident",      _incident_fmt_mod.format)
 register("road_closure",       _incident_fmt_mod.format)
 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.
+from meshai.notifications.formatters import hydro as _hydro_fmt_mod  # noqa: E402,F401
+register("stream_flow", _hydro_fmt_mod.format)

work/meshai/notifications/formatters/hydro.py

@@ -0,0 +1,95 @@
+"""USGS NWIS stream-gauge (hydro) formatter — Phase-3 migration.
+
+Mirrors central.nwis_handler._render EXACTLY (tier-a, byte-identical).  Reads
+the canonical schema the Central path writes into event.data on broadcast:
+
+    site_id, gauge_name, stage_ft, flow_cfs, unit, threshold_state,
+    reading_time, lat, lon
+
+Wire format (single line, MEDIUM):
+    🌊 New: {gauge_name}: {label} {stage_ft:.1f} ft, flow {flow_cfs:,} cfs, @ lat,lon
+
+Where {label} maps threshold_state via _LABEL:
+    action          -> "action stage"
+    flood_minor     -> "minor flooding"
+    flood_moderate  -> "moderate flooding"
+    flood_major     -> "major flooding"
+
+Segments:
+  * stage segment: "{label} {stage_ft:.1f} ft" when stage_ft is numeric, else
+    the bare label.  The "ft" unit is hard-coded (matches _render, which
+    ignores its `unit` parameter for the stage segment).
+  * flow segment: ", flow {int(round(flow_cfs)):,} cfs" only when flow_cfs is
+    numeric (companion 00060 discharge reading present).
+  * coords segment: ", @ {lat:.3f},{lon:.3f}" only when both coords numeric.
+
+Time contract: `now` is accepted but NOT used for rendering (structural seam;
+never read the wall clock here).  `budget` is injected — the caller supplies
+budget_for("usgs_nwis").  fit_to_budget is applied last; for the short hydro
+wire it is a no-op under a 140-char budget, so output stays byte-identical to
+_render (which did not budget-fit).
+"""
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+# Import directly from the canonical implementation to avoid the circular
+# import chain: central.budget → formatters.__init__ → formatters.hydro.
+from meshai.notifications.formatters._budget import fit_to_budget
+
+if TYPE_CHECKING:
+    from meshai.notifications.events import Event
+
+
+# Human-readable label per threshold_state — mirrors nwis_handler._LABEL.
+_LABEL = {
+    "action":         "action stage",
+    "flood_minor":    "minor flooding",
+    "flood_moderate": "moderate flooding",
+    "flood_major":    "major flooding",
+}
+
+
+def format(event: "Event", *, now: float, budget: int) -> str:
+    """Render the hydro wire string from canonical event.data.
+
+    Args:
+        event:  Pipeline Event — reads from event.data (canonical schema).
+        now:    Frozen-clock epoch (seam; not used in current rendering).
+        budget: Mesh-packet character budget (from budget_for("usgs_nwis")).
+
+    Returns:
+        UTF-8 string fitting within *budget* characters.
+    """
+    d = event.data or {}
+
+    gauge_name = d.get("gauge_name")
+    threshold_state = d.get("threshold_state")
+    stage_ft = d.get("stage_ft")
+    flow_cfs = d.get("flow_cfs")
+    lat = d.get("lat")
+    lon = d.get("lon")
+    # `unit` is present in the canonical dict but intentionally unused here —
+    # _render ignores its unit parameter and hard-codes "ft" in the stage
+    # segment; mirror that exactly for byte-identity.
+
+    label = _LABEL.get(threshold_state, threshold_state)
+
+    # Stage segment.
+    if isinstance(stage_ft, (int, float)):
+        stage_seg = f"{label} {stage_ft:.1f} ft"
+    else:
+        stage_seg = label
+
+    # Optional flow segment (companion 00060 discharge).
+    flow_seg = ""
+    if isinstance(flow_cfs, (int, float)):
+        flow_seg = f", flow {int(round(flow_cfs)):,} cfs"
+
+    # Optional coords segment.
+    coords = ""
+    if isinstance(lat, (int, float)) and isinstance(lon, (int, float)):
+        coords = f", @ {lat:.3f},{lon:.3f}"
+
+    msg = f"🌊 New: {gauge_name}: {stage_seg}{flow_seg}{coords}"
+    return fit_to_budget(msg, budget)

work/meshai/notifications/gating/__init__.py

@@ -77,3 +77,12 @@ register("work_zone",          _incident_gate_mod.decide)
 register("road_incident",      _incident_gate_mod.decide)
 register("road_closure",       _incident_gate_mod.decide)
 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.
+from meshai.notifications.gating import hydro as _hydro_gate_mod  # noqa: E402,F401
+register("stream_flow", _hydro_gate_mod.decide)

work/meshai/notifications/gating/hydro.py

@@ -0,0 +1,152 @@
+"""USGS NWIS stream-gauge (hydro) gating decider — Phase-3 migration.
+
+Moves the threshold-crossing decision out of central.nwis_handler.handle_nwis
+(the inline prior-reading SELECT, the 00060/00065 stage back-look, the
+THRESHOLD_RANK upward-crossing check, and the broadcast_on_recede toggle).
+
+Broadcast rule (verbatim from the old handler):
+    Compare the current reading's threshold_state to the most recent PRIOR
+    reading (strictly earlier reading_time, any parameter) for the same site,
+    on the ranked scale {normal < action < flood_minor < flood_moderate <
+    flood_major}.
+      * cur_rank > prior_rank  → upward crossing → broadcast
+      * cur_rank == prior_rank → unchanged band  → suppress
+      * cur_rank < prior_rank  → receding        → suppress, UNLESS
+        adapter_config.usgs_nwis.broadcast_on_recede is set → broadcast
+
+00060 (discharge) back-look:
+    A discharge-only envelope carries no stage → its threshold band is
+    inherited from the latest 00065 (gage-height, reading_unit='ft') reading at
+    the site.  The decider performs that look-back and returns the resolved
+    (threshold_state, stage_ft) in data_patch so the handler can INSERT the
+    gauge_readings row and render the wire with the same values.
+
+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.
+    * 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
+      filters `reading_time < current` for defence-in-depth.
+    * Hydro has no per-event broadcast-state table (unlike quake_events), so
+      the decider owns NO deferred state write → commit is None.  The
+      event_log.handled flip stays handler-owned (mirrors the old
+      _attach_commit, wrapped in the cutover branch like quake_handler).
+
+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)
+    stage_ft        : float|None — resolved stage (post-00060 back-look)
+"""
+from __future__ import annotations
+
+import logging
+from typing import Optional
+
+from meshai.adapter_config import adapter_config
+from meshai.central.idaho_gauge_sites import THRESHOLD_RANK
+from meshai.notifications.gating.base import GateResult
+from meshai.persistence import get_db
+
+logger = logging.getLogger(__name__)
+
+
+def _rank(state: Optional[str]) -> int:
+    """THRESHOLD_RANK index for *state*, treating unknown as 0 (normal)."""
+    try:
+        return THRESHOLD_RANK.index(state)
+    except ValueError:
+        return 0
+
+
+def decide(data: dict, *, source: str, now: float) -> GateResult:
+    """Upward-threshold-crossing decision for USGS NWIS stream-gauge readings.
+
+    Parameters
+    ----------
+    data:
+        Canonical Event.data dict.  Consumed keys:
+            site_id, reading_time, parameter_code, threshold_state, stage_ft
+        (gauge_name / flow_cfs / lat / lon are carried through by the handler
+        for the formatter but are not needed for the gate decision.)
+    source:
+        Adapter source name, e.g. "nwis".
+    now:
+        Current epoch (from clock.now()) — determinism seam, unused here (hydro
+        gating is state-comparison only, no time window).
+
+    Returns
+    -------
+    GateResult:
+        broadcast=True   lifecycle="new"       upward crossing (or recede+toggle)
+        broadcast=False  lifecycle="suppress"  unchanged band or receding
+        data_patch always carries the resolved threshold_state + stage_ft.
+        commit is always None (no decider-owned state write).
+    """
+    site_id = data.get("site_id")
+    reading_time = data.get("reading_time")
+    pc = data.get("parameter_code")
+    threshold_state = data.get("threshold_state") or "normal"
+    stage_ft = data.get("stage_ft")
+
+    try:
+        conn = get_db()
+    except Exception:
+        logger.exception("nwis decide: persistence unavailable")
+        return GateResult(
+            broadcast=False, lifecycle="suppress",
+            reason="persistence unavailable",
+        )
+
+    # Most recent PRIOR reading (strictly earlier) — same as the old handler.
+    prior = conn.execute(
+        "SELECT threshold_state FROM gauge_readings "
+        "WHERE site_id=? AND reading_time < ? "
+        "ORDER BY reading_time DESC LIMIT 1",
+        (site_id, reading_time),
+    ).fetchone()
+    prior_state = prior["threshold_state"] if prior else "normal"
+
+    # 00060 (discharge) back-look: inherit stage + band from the latest 00065
+    # (gage-height) reading at this site.  Discharge alone has no threshold band.
+    if pc == "00060":
+        last_stage = conn.execute(
+            "SELECT reading_value, threshold_state FROM gauge_readings "
+            "WHERE site_id=? AND reading_unit='ft' "
+            "ORDER BY reading_time DESC LIMIT 1",
+            (site_id,),
+        ).fetchone()
+        if last_stage:
+            stage_ft = last_stage["reading_value"]
+            threshold_state = last_stage["threshold_state"] or "normal"
+
+    # Resolved values returned so the handler's inline INSERT + render match.
+    patch: dict = {"threshold_state": threshold_state, "stage_ft": stage_ft}
+
+    prior_rank = _rank(prior_state)
+    cur_rank = _rank(threshold_state)
+
+    # Unchanged band — no broadcast.
+    if cur_rank == prior_rank:
+        return GateResult(
+            broadcast=False, lifecycle="suppress",
+            reason=f"unchanged band {threshold_state}",
+            data_patch=patch, commit=None,
+        )
+
+    # Receding without the recede toggle — silent.
+    if cur_rank < prior_rank and not bool(
+            adapter_config.usgs_nwis.broadcast_on_recede):
+        return GateResult(
+            broadcast=False, lifecycle="suppress",
+            reason=f"receding {prior_state}->{threshold_state}",
+            data_patch=patch, commit=None,
+        )
+
+    # Upward crossing (or recede with the toggle enabled) — broadcast.
+    return GateResult(
+        broadcast=True, lifecycle="new",
+        reason=f"crossing {prior_state}->{threshold_state}",
+        data_patch=patch, commit=None,
+    )

work/tests/test_hydro_refactor.py

@@ -0,0 +1,341 @@
+"""Phase-3 hydro (USGS NWIS) refactor tests.
+
+Verifies the source-agnostic formatter+decider migration for the stream-gauge
+hazard, mirroring test_quake_refactor.py:
+
+1. Golden byte-identical: formatters.hydro.format() reproduces the old
+   nwis_handler._render() wire exactly, for a stage-only crossing, a paired
+   flow(00060)+stage(00065) reading, and every threshold label.
+
+2. Gate-sequence parity: an explicit `now`-timeline of readings driven through
+   the NEW gating.hydro.decide() matches the OLD handle_nwis broadcast/suppress
+   behavior (upward crossing broadcasts; same-rank + receding suppress unless
+   broadcast_on_recede).
+
+The real registry / cutover key is "stream_flow" — the flat category the
+Central nwis path produces for every central.hydro.* envelope.
+"""
+from __future__ import annotations
+
+import pytest
+
+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, run_gate_sequence
+
+_AT = 1_783_200_000.0  # pinned epoch (unused by hydro render/gate, kept for parity)
+
+
+# ── DB fixture (same shape as test_nwis_handler.py) ──────────────────────────
+
+@pytest.fixture
+def mem_db(monkeypatch, tmp_path):
+    db_path = str(tmp_path / "hydro-refactor-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)
+
+
+def _make_fake_event(data: dict):
+    class _FakeEvent:
+        pass
+    e = _FakeEvent()
+    e.data = data
+    return e
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 1. Golden byte-identical — formatter reproduces _render() exactly
+# ─────────────────────────────────────────────────────────────────────────────
+
+class TestFormatterGolden:
+    """formatters.hydro.format() == nwis_handler._render() for the same inputs."""
+
+    def _render_old(self, **kw):
+        from meshai.central.nwis_handler import _render
+        return _render(**kw)
+
+    def _fmt_new(self, canonical: dict) -> str:
+        from meshai.notifications.formatters.hydro import format as hfmt
+        return hfmt(_make_fake_event(canonical), now=_AT, budget=140)
+
+    def test_stage_only_crossing_action(self):
+        """Stage-only (00065) reading at action stage, no companion flow."""
+        canonical = {
+            "gauge_name": "Snake River at Heise",
+            "threshold_state": "action",
+            "stage_ft": 12.5,
+            "flow_cfs": None,
+            "unit": "ft",
+            "lat": 43.612,
+            "lon": -111.654,
+        }
+        old = self._render_old(
+            gauge_name="Snake River at Heise", threshold_state="action",
+            stage_ft=12.5, flow_cfs=None, unit="ft", lat=43.612, lon=-111.654,
+        )
+        new = self._fmt_new(canonical)
+        assert_byte_identical(new, old)
+        assert new == "🌊 New: Snake River at Heise: action stage 12.5 ft, @ 43.612,-111.654"
+
+    def test_paired_flow_and_stage(self):
+        """00060 discharge back-looked onto a 00065 stage: flow segment present."""
+        canonical = {
+            "gauge_name": "Boise River",
+            "threshold_state": "flood_minor",
+            "stage_ft": 14.5,
+            "flow_cfs": 8400,
+            "unit": "ft",
+            "lat": 43.600,
+            "lon": -116.200,
+        }
+        old = self._render_old(
+            gauge_name="Boise River", threshold_state="flood_minor",
+            stage_ft=14.5, flow_cfs=8400, unit="ft", lat=43.600, lon=-116.200,
+        )
+        new = self._fmt_new(canonical)
+        assert_byte_identical(new, old)
+        assert "flow 8,400 cfs" in new
+        assert "minor flooding 14.5 ft" in new
+
+    @pytest.mark.parametrize(
+        "state,label",
+        [
+            ("action", "action stage"),
+            ("flood_minor", "minor flooding"),
+            ("flood_moderate", "moderate flooding"),
+            ("flood_major", "major flooding"),
+        ],
+    )
+    def test_every_threshold_label(self, state, label):
+        """Each threshold_state maps to the correct label — byte-identical to _render."""
+        canonical = {
+            "gauge_name": "Test Gauge",
+            "threshold_state": state,
+            "stage_ft": 20.0,
+            "flow_cfs": None,
+            "unit": "ft",
+            "lat": 44.0,
+            "lon": -114.0,
+        }
+        old = self._render_old(
+            gauge_name="Test Gauge", threshold_state=state, stage_ft=20.0,
+            flow_cfs=None, unit="ft", lat=44.0, lon=-114.0,
+        )
+        new = self._fmt_new(canonical)
+        assert_byte_identical(new, old)
+        assert f"{label} 20.0 ft" in new
+
+    def test_missing_coords_drops_at_tail(self):
+        """No coords → no @ segment (byte-identical to _render)."""
+        canonical = {
+            "gauge_name": "No Coords Gauge",
+            "threshold_state": "action",
+            "stage_ft": 10.0,
+            "flow_cfs": None,
+            "unit": "ft",
+            "lat": None,
+            "lon": None,
+        }
+        old = self._render_old(
+            gauge_name="No Coords Gauge", threshold_state="action",
+            stage_ft=10.0, flow_cfs=None, unit="ft", lat=None, lon=None,
+        )
+        new = self._fmt_new(canonical)
+        assert_byte_identical(new, old)
+        assert "@" not in new
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 2. Gate-sequence parity — old handle_nwis vs new gating.hydro.decide()
+# ─────────────────────────────────────────────────────────────────────────────
+
+def _nwis_env(*, site_id="USGS-13186000", parameter_code="00065", value=13.0,
+              unit="ft", time_iso="2026-06-05T15:00:00Z",
+              lat=43.612, lon=-111.654, envelope_id=None):
+    envelope_id = envelope_id or f"nwis_{site_id}_{time_iso}"
+    return {
+        "id": envelope_id,
+        "subject": f"central.hydro.{parameter_code}.usgs.{site_id}.us.id",
+        "data": {
+            "id": envelope_id, "adapter": "nwis",
+            "category": f"hydro.{parameter_code}", "severity": 0,
+            "geo": {"centroid": [lon, lat], "primary_region": "US-ID"},
+            "data": {
+                "id": envelope_id,
+                "monitoring_location_id": site_id,
+                "parameter_code": parameter_code,
+                "time": time_iso,
+                "value": value,
+                "unit_of_measure": unit,
+                "latitude": lat, "longitude": lon,
+            },
+        },
+    }
+
+
+def _parse_iso_epoch(s):
+    from datetime import datetime
+    return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp())
+
+
+class TestGateSequenceParity:
+    """New decide() decisions match old handle_nwis broadcast/suppress."""
+
+    @pytest.fixture(autouse=True)
+    def _db(self, mem_db):
+        self.db = mem_db
+
+    def _old_gate(self, fixture, *, now):
+        """OLD path: handle_nwis returning non-None = broadcast.
+
+        handle_nwis owns the append-only gauge_readings INSERT, so replaying
+        through it advances the persisted time-series exactly as production
+        would — the decider (below) then reads that same state.
+        """
+        from meshai.central.nwis_handler import handle_nwis
+        env = fixture["envelope"]
+        wire = handle_nwis(env, env["subject"], data={}, now=int(now))
+        return wire is not None
+
+    def _new_gate(self, fixture, *, now):
+        """NEW path: build canonical (as the handler does) then decide().
+
+        We do NOT insert here — the old-gate replay already advances
+        gauge_readings; the decider only READS prior state.  This mirrors the
+        production ordering where decide() runs before the inline INSERT.
+        """
+        from meshai.notifications.gating.hydro import decide
+        from meshai.central.idaho_gauge_sites import (
+            compute_threshold_state, lookup_site, normalize_site_id,
+        )
+        env = fixture["envelope"]
+        d = env["data"]["data"]
+        raw_site = d.get("monitoring_location_id")
+        site_id = normalize_site_id(raw_site)
+        site_meta = lookup_site(raw_site)
+        pc = d.get("parameter_code")
+        value = float(d.get("value"))
+        reading_time = _parse_iso_epoch(d.get("time"))
+        stage_ft = value if pc == "00065" else None
+        flow_cfs = value if pc == "00060" else None
+        threshold_state = "normal"
+        if pc == "00065":
+            threshold_state = compute_threshold_state(stage_ft, site_meta)
+        canonical = {
+            "site_id": site_id,
+            "gauge_name": site_meta["gauge_name"],
+            "stage_ft": stage_ft,
+            "flow_cfs": flow_cfs,
+            "unit": d.get("unit_of_measure"),
+            "threshold_state": threshold_state,
+            "reading_time": reading_time,
+            "lat": d.get("latitude"),
+            "lon": d.get("longitude"),
+            "parameter_code": pc,
+        }
+        return decide(canonical, source="nwis", now=float(now))
+
+    def test_gate_sequence_matches(self):
+        """Timeline of Heise readings: old and new gates agree on every step.
+
+        Heise (USGS-13186000): action=12.0ft.
+          [0] 8.0 ft  normal  (first reading, no prior)        → suppress
+          [1] 12.5 ft action  (normal → action upward)          → broadcast
+          [2] 12.8 ft action  (action → action same rank)       → suppress
+          [3] 14.5 ft f_minor (action → flood_minor upward)     → broadcast
+          [4] 8.0 ft  normal  (flood_minor → normal receding)   → suppress
+        """
+        base = _parse_iso_epoch("2026-06-05T10:00:00Z")
+        specs = [
+            (8.0,  "2026-06-05T10:00:00Z", "s0"),
+            (12.5, "2026-06-05T10:15:00Z", "s1"),
+            (12.8, "2026-06-05T10:30:00Z", "s2"),
+            (14.5, "2026-06-05T10:45:00Z", "s3"),
+            (8.0,  "2026-06-05T11:00:00Z", "s4"),
+        ]
+        ordered = [
+            {"envelope": _nwis_env(value=v, time_iso=t, envelope_id=eid)}
+            for (v, t, eid) in specs
+        ]
+        timeline = [float(base + i * 900) for i in range(len(specs))]
+
+        results = run_gate_sequence(self._old_gate, self._new_gate, ordered,
+                                    timeline=timeline)
+        mismatches = [r for r in results if not r["match"]]
+        assert not mismatches, (
+            "Gate sequence mismatch old handle_nwis vs new decide():\n"
+            + "\n".join(
+                f"  step {r['fixture_n']}: old={r['old_broadcast']} "
+                f"new={r['new_broadcast']} diffs={r['diffs']}"
+                for r in mismatches
+            )
+        )
+        assert results[0]["old_broadcast"] is False, "normal first reading suppressed"
+        assert results[1]["old_broadcast"] is True,  "normal→action broadcasts"
+        assert results[2]["old_broadcast"] is False, "action→action suppressed"
+        assert results[3]["old_broadcast"] is True,  "action→flood_minor broadcasts"
+        assert results[4]["old_broadcast"] is False, "receding suppressed (no toggle)"
+
+    def test_00060_backlook_inherits_stage_band(self, mem_db):
+        """A 00060 discharge reading inherits the last 00065 stage band.
+
+        Seed an action-stage 00065 reading, then feed a 00060 discharge: the
+        decider's back-look must resolve threshold_state=action + the prior
+        stage_ft, and (same rank as the seeded action) suppress the discharge.
+        """
+        from meshai.notifications.gating.hydro import decide
+        # Seed a 00065 stage reading at action via the old handler (writes row).
+        env_stage = _nwis_env(parameter_code="00065", value=12.5,
+                              time_iso="2026-06-05T10:00:00Z", envelope_id="seed")
+        self._old_gate({"envelope": env_stage}, now=1_000_000)
+
+        # Now decide on a 00060 discharge — should back-look the action band.
+        env_flow = _nwis_env(parameter_code="00060", value=8400, unit="ft^3/s",
+                             time_iso="2026-06-05T10:05:00Z", envelope_id="q")
+        gate = self._new_gate({"envelope": env_flow}, now=1_000_300)
+        assert gate.data_patch["threshold_state"] == "action"
+        assert gate.data_patch["stage_ft"] == 12.5
+        # action → action (same rank) → suppress
+        assert gate.broadcast is False
+
+    def test_recede_toggle_enables_broadcast(self, mem_db):
+        """With broadcast_on_recede set, a receding crossing broadcasts."""
+        from meshai.adapter_config._accessor import set_runtime_override, _overrides
+        from meshai.notifications.gating.hydro import decide
+        # Seed an action reading.
+        env_high = _nwis_env(parameter_code="00065", value=12.5,
+                            time_iso="2026-06-05T10:00:00Z", envelope_id="hi")
+        self._old_gate({"envelope": env_high}, now=1_000_000)
+
+        # Force the recede toggle on for the decision only (runtime override,
+        # since adapter_config accessors are read-only).
+        set_runtime_override("usgs_nwis", "broadcast_on_recede", True)
+        try:
+            env_low = _nwis_env(parameter_code="00065", value=8.0,
+                               time_iso="2026-06-05T11:00:00Z", envelope_id="lo")
+            gate = self._new_gate({"envelope": env_low}, now=1_003_600)
+        finally:
+            _overrides.pop(("usgs_nwis", "broadcast_on_recede"), None)
+        assert gate.broadcast is True, "receding must broadcast when toggle is on"
+        assert gate.data_patch["threshold_state"] == "normal"
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# 3. Registration — stream_flow resolves to hydro formatter + decider
+# ─────────────────────────────────────────────────────────────────────────────
+
+class TestRegistration:
+    def test_stream_flow_formatter_registered(self):
+        from meshai.notifications.formatters import get_formatter
+        from meshai.notifications.formatters.hydro import format as hfmt
+        assert get_formatter("stream_flow") is hfmt
+
+    def test_stream_flow_decider_registered(self):
+        from meshai.notifications.gating import get_decider
+        from meshai.notifications.gating.hydro import decide
+        assert get_decider("stream_flow") is decide