feat(collector): forecast and ntfy in ping-collector; ThaiWater keeps alerts current at night

Phase 2b/2c of docs/RUST_REWRITE.md. With FORECAST_IN_COLLECTOR=true (shared
.env) ping-collector, after each cycle that brought newer RID or HII readings
and at least hourly: loads the last 336 h (RID + HII mirror fill), fetches the
rain forecast, runs every exported model (rust/forecast), stores
forecast_history and forecast_current, and sends the ntfy notifications.
The Python API then serves forecast_current as /forecast (computing in-process
only when it is older than FORECAST_MAX_AGE_SECONDS) and neither forecasts nor
notifies.

ntfy (collector/src/notify.rs): a port of src/notify.py with the same topics,
texts, hysteresis, capacity guard and notification_state keys, checked word
for word on 751 messages generated by Python (testdata/gen_notify_cases.py).
The change: for stations whose ThaiWater twin is the same sensor, a reading
newer than RID's drives the level alerts (marked as ThaiWater telemetry), and
the feed counts as stale only when neither source has a reading for 3 h. RID
alone lagging is reported once after RID_STALE_ALERT_HOURS (18). This ends the
nightly "gauge feed stale" / "recovered" pairs caused by RID's late reports.

Commands: `forecast [--json] [--save] [--now] [--rain-file]`,
`notify --dry-run`. scripts/forecast_parity.py compares Python and the
collector on the live database (same moment, same rain); locally 48/48 rows
identical. install_collector.sh: models export + that check on every install,
--cutover-forecast (refuses on a failed check) and --rollback-forecast.

Trees are packed into 24-byte nodes: 70 MB peak with all 15 models (was 140).
Forecast values are written to NUMERIC as shortest round-trip text, so they
round as psycopg2's do. DB test schemas get a counter (parallel tests collided
on Windows' coarse clock).

rust/collector/src/notify.rs

@@ -0,0 +1,879 @@
+//! Public flood notifications over ntfy: the port of src/notify.py, run by
+//! the collector when FORECAST_IN_COLLECTOR=true.
+//!
+//! Same topics, texts, hysteresis, capacity guard, outlook and persisted
+//! state keys (`notification_state`) as Python, so a cutover neither repeats
+//! nor drops a message. One change: for stations whose HII/ThaiWater twin is
+//! the same sensor, a reading newer than RID's (RID's report often lags 2-10 h
+//! at night) is used for the level alerts and the feed check. The feed is
+//! "stale" only when neither source has a reading for 3 h; RID lagging alone
+//! is reported separately after RID_STALE_ALERT_HOURS.
+
+use crate::db::{self, LatestReading};
+use anyhow::Result;
+use chrono::{Duration, NaiveDateTime};
+use ping_forecast::Forecast;
+use ping_forecast::config::config;
+use ping_forecast::hii::FillInfo;
+use sqlx::PgPool;
+use std::collections::HashMap;
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum Source {
+    Rid,
+    /// HII/ThaiWater telemetry of the same sensor, offset to the RID datum.
+    Hii,
+}
+
+#[derive(Clone, Debug, PartialEq)]
+pub struct Reading {
+    pub station_code: String,
+    pub timestamp: NaiveDateTime,
+    pub water_level: Option<f64>,
+    pub discharge_percent: Option<f64>,
+    pub source: Source,
+    /// The station's newest RID reading with a level (for the RID lag check).
+    pub rid_timestamp: Option<NaiveDateTime>,
+}
+
+#[derive(Clone, Debug, PartialEq)]
+pub struct Notification {
+    pub topic: String,
+    pub title: String,
+    pub message: String,
+    pub priority: u8,
+    pub tags: Vec<String>,
+}
+
+/// A notification and the state to record once it was delivered.
+#[derive(Clone, Debug, PartialEq)]
+pub struct Planned {
+    pub notification: Notification,
+    pub state: Option<(String, String, Option<f64>)>,
+}
+
+/// The readings the notifier judges: the newest RID row per station, replaced
+/// by the newest HII fill when that is newer (or RID's newest row has no level).
+pub fn readings_with_hii(rid: &[LatestReading], fills: &[FillInfo]) -> Vec<Reading> {
+    let by_code: HashMap<&str, &FillInfo> =
+        fills.iter().map(|f| (f.station_code.as_str(), f)).collect();
+    rid.iter()
+        .map(|r| {
+            let fill = by_code.get(r.station_code.as_str()).and_then(|f| f.latest);
+            match fill {
+                Some((t, level))
+                    if t > r.timestamp || (t == r.timestamp && r.water_level.is_none()) =>
+                {
+                    Reading {
+                        station_code: r.station_code.clone(),
+                        timestamp: t,
+                        water_level: Some(level),
+                        // capacity drifts slowly: RID's figure if it is recent
+                        discharge_percent: r
+                            .discharge_percent
+                            .filter(|_| t - r.timestamp <= Duration::hours(6)),
+                        source: Source::Hii,
+                        rid_timestamp: r.water_level.map(|_| r.timestamp),
+                    }
+                }
+                _ => Reading {
+                    station_code: r.station_code.clone(),
+                    timestamp: r.timestamp,
+                    water_level: r.water_level,
+                    discharge_percent: r.discharge_percent,
+                    source: Source::Rid,
+                    rid_timestamp: r.water_level.map(|_| r.timestamp),
+                },
+            }
+        })
+        .collect()
+}
+
+/// `(a - b).total_seconds() / 3600` as Python computes it.
+fn hours_between(a: NaiveDateTime, b: NaiveDateTime) -> f64 {
+    (a - b)
+        .num_microseconds()
+        .map_or(f64::INFINITY, |us| us as f64 / 1e6 / 3600.0)
+}
+
+fn slug(code: &str) -> String {
+    code.to_lowercase().replace('.', "")
+}
+
+/// 'clear' | 'warning' | 'danger', with hysteresis on the way down.
+fn level_state(level: f64, warn: f64, danger: f64, prev: &str) -> &'static str {
+    let margin = config().notify.clear_margin_m;
+    if level >= danger {
+        return "danger";
+    }
+    if level >= warn {
+        if prev == "danger" && level >= danger - margin {
+            return "danger";
+        }
+        return "warning";
+    }
+    if (prev == "warning" || prev == "danger") && level >= warn - margin {
+        return "warning";
+    }
+    "clear"
+}
+
+pub struct Settings {
+    pub prefix: String,
+    /// 0 = off
+    pub rid_stale_alert_hours: f64,
+}
+
+impl Settings {
+    fn topic(&self, parts: &[&str]) -> String {
+        std::iter::once(self.prefix.as_str())
+            .chain(parts.iter().copied())
+            .collect::<Vec<_>>()
+            .join("-")
+    }
+}
+
+fn note(n: Notification) -> Planned {
+    Planned {
+        notification: n,
+        state: None,
+    }
+}
+
+fn n(
+    topic: String,
+    title: impl Into<String>,
+    message: String,
+    priority: u8,
+    tags: &[&str],
+) -> Notification {
+    Notification {
+        topic,
+        title: title.into(),
+        message,
+        priority,
+        tags: tags.iter().map(|t| t.to_string()).collect(),
+    }
+}
+
+/// `notify.evaluate` as a plan: every transition against the stored states.
+pub fn evaluate(
+    readings: &[Reading],
+    forecasts: &[Forecast],
+    states: &HashMap<String, String>,
+    settings: &Settings,
+    now: NaiveDateTime,
+) -> Vec<Planned> {
+    let cfg = config();
+    let nc = &cfg.notify;
+    let state = |k: &str| states.get(k).map(String::as_str);
+    let mut plan: Vec<Planned> = Vec::new();
+    let (mut b_danger, mut b_warning, mut b_clear) = (vec![], vec![], vec![]);
+    let mut latest: Option<NaiveDateTime> = None;
+    let mut latest_rid: Option<NaiveDateTime> = None;
+
+    for r in readings {
+        latest_rid = latest_rid.max(r.rid_timestamp);
+        let Some(level) = r.water_level else {
+            continue;
+        };
+        let ts = r.timestamp;
+        latest = latest.max(Some(ts));
+        let code = r.station_code.as_str();
+        let (warn, danger) = cfg.thresholds(code);
+        let key = format!("level:{code}");
+        let prev = state(&key).unwrap_or("clear");
+        let cur = level_state(level, warn, danger, prev);
+        if cur != "clear"
+            && prev == "clear"
+            && !nc.capacity_guard_exempt.iter().any(|c| c == code)
+            && let Some(pct) = r.discharge_percent
+            && pct < nc.capacity_guard_min_pct
+        {
+            tracing::info!(
+                "{code}: level {level:.2} m >= {warn:.2} but only {pct:.0}% capacity; \
+                 threshold looks stale, not alerting"
+            );
+            continue;
+        }
+        if cur == prev {
+            continue;
+        }
+        let name = nc.station_names.get(code).map_or(code, String::as_str);
+        let slug = slug(code);
+        let mut when = ts.format("%d %b %H:%M").to_string();
+        if r.source == Source::Hii {
+            when.push_str(" (ThaiWater telemetry; the RID report is not in yet)");
+        }
+        let (topic_kind, title, message, priority, tag) = match (cur, prev) {
+            ("danger", _) => {
+                b_danger.push(format!("{code} {level:.2} m"));
+                (
+                    "danger",
+                    format!("DANGER level at {code}"),
+                    format!(
+                        "{name}: {level:.2} m at {when}, above the danger level of {danger:.2} m."
+                    ),
+                    5,
+                    "rotating_light",
+                )
+            }
+            ("warning", "danger") => {
+                b_clear.push(format!("{code} below danger ({level:.2} m)"));
+                (
+                    "danger",
+                    format!("{code} back below danger level"),
+                    format!(
+                        "{name}: {level:.2} m at {when}; still above the warning level of {warn:.2} m."
+                    ),
+                    3,
+                    "arrow_down",
+                )
+            }
+            ("warning", _) => {
+                b_warning.push(format!("{code} {level:.2} m"));
+                (
+                    "warning",
+                    format!("Warning level at {code}"),
+                    format!(
+                        "{name}: {level:.2} m at {when}, above the warning level of {warn:.2} m."
+                    ),
+                    4,
+                    "warning",
+                )
+            }
+            _ => {
+                b_clear.push(format!("{code} normal ({level:.2} m)"));
+                (
+                    "warning",
+                    format!("{code} back to normal"),
+                    format!(
+                        "{name}: {level:.2} m at {when}, below the warning level of {warn:.2} m."
+                    ),
+                    2,
+                    "white_check_mark",
+                )
+            }
+        };
+        plan.push(Planned {
+            notification: n(
+                settings.topic(&[&slug, topic_kind]),
+                title,
+                message,
+                priority,
+                &[tag, code],
+            ),
+            // remembered only once delivered, so ntfy being down means a retry
+            state: Some((key, cur.to_string(), Some(level))),
+        });
+    }
+
+    if !b_danger.is_empty() {
+        plan.push(note(n(
+            settings.topic(&["danger"]),
+            "Ping River: danger level reached",
+            b_danger.join("; "),
+            5,
+            &["rotating_light"],
+        )));
+    }
+    if !b_warning.is_empty() {
+        plan.push(note(n(
+            settings.topic(&["warning"]),
+            "Ping River: warning level reached",
+            b_warning.join("; "),
+            4,
+            &["warning"],
+        )));
+    }
+    if !b_clear.is_empty() {
+        plan.push(note(n(
+            settings.topic(&["warning"]),
+            "Ping River: levels falling",
+            b_clear.join("; "),
+            2,
+            &["white_check_mark"],
+        )));
+    }
+
+    // model outlook for the city gauge (opt-in topic, experimental)
+    if let Some(p1) = forecasts.iter().find(|f| {
+        f.station_code == "P.1" && f.horizon_hours == nc.outlook_horizon && f.source == "model"
+    }) {
+        let p = p1.p_warning;
+        let key = "outlook:P.1";
+        let prev = state(key).unwrap_or("off");
+        let cur = if p >= nc.outlook_on || (prev == "on" && p >= nc.outlook_off) {
+            "on"
+        } else {
+            "off"
+        };
+        if cur != prev {
+            let (warn, _) = cfg.thresholds("P.1");
+            let notification = if cur == "on" {
+                n(
+                    settings.topic(&["p1-outlook"]),
+                    "Early warning: Chiang Mai flood risk rising",
+                    format!(
+                        "The forecast model gives a {:.0}% chance that Nawarat Bridge (P.1) reaches \
+                         {warn:.2} m within 24 h (expected peak {:.2} m). Experimental model output, \
+                         not an official warning; follow ThaiWater/TMD for official alerts.",
+                        p * 100.0,
+                        p1.predicted_max_level
+                    ),
+                    4,
+                    &["crystal_ball"],
+                )
+            } else {
+                n(
+                    settings.topic(&["p1-outlook"]),
+                    "Chiang Mai flood risk easing",
+                    format!(
+                        "The model's 24 h probability of reaching {warn:.2} m at P.1 has dropped to {:.0}%.",
+                        p * 100.0
+                    ),
+                    2,
+                    &["crystal_ball"],
+                )
+            };
+            plan.push(Planned {
+                notification,
+                state: Some((key.into(), cur.into(), Some(p))),
+            });
+        }
+    }
+
+    // feed health: stale only when no source has a recent reading
+    let mut feed_stale = false;
+    if let Some(latest) = latest {
+        let age_h = hours_between(now, latest);
+        let prev = state("feed").unwrap_or("ok");
+        feed_stale = age_h >= nc.feed_stale_after_h;
+        let cur = if feed_stale { "stale" } else { "ok" };
+        if cur != prev {
+            let notification = if feed_stale {
+                n(
+                    settings.topic(&["status"]),
+                    "Ping River monitor: gauge feed stale",
+                    format!(
+                        "No new readings for {age_h:.0} h (last {}). Levels and forecasts on the \
+                         dashboard are not current.",
+                        latest.format("%d %b %H:%M")
+                    ),
+                    3,
+                    &["hourglass"],
+                )
+            } else {
+                n(
+                    settings.topic(&["status"]),
+                    "Ping River monitor: feed recovered",
+                    format!(
+                        "Readings are current again (latest {}).",
+                        latest.format("%d %b %H:%M")
+                    ),
+                    2,
+                    &["white_check_mark"],
+                )
+            };
+            plan.push(Planned {
+                notification,
+                state: Some(("feed".into(), cur.into(), Some(age_h))),
+            });
+        }
+    }
+
+    // RID itself far behind while ThaiWater keeps the levels current
+    if settings.rid_stale_alert_hours > 0.0
+        && let Some(rid_ts) = latest_rid
+    {
+        let age_h = hours_between(now, rid_ts);
+        let prev = state("feed:rid").unwrap_or("ok");
+        let cur = if age_h >= settings.rid_stale_alert_hours {
+            "stale"
+        } else {
+            "ok"
+        };
+        // while the whole feed is stale, that message covers RID too
+        if cur != prev && !(cur == "stale" && feed_stale) {
+            let notification = if cur == "stale" {
+                n(
+                    settings.topic(&["status"]),
+                    "Ping River monitor: RID reports delayed",
+                    format!(
+                        "No RID gauge report for {age_h:.0} h (last {}). Levels and alerts continue \
+                         from ThaiWater telemetry for the stations it mirrors.",
+                        rid_ts.format("%d %b %H:%M")
+                    ),
+                    3,
+                    &["hourglass"],
+                )
+            } else {
+                n(
+                    settings.topic(&["status"]),
+                    "Ping River monitor: RID reports current again",
+                    format!("Latest RID report {}.", rid_ts.format("%d %b %H:%M")),
+                    2,
+                    &["white_check_mark"],
+                )
+            };
+            plan.push(Planned {
+                notification,
+                state: Some(("feed:rid".into(), cur.into(), Some(age_h))),
+            });
+        }
+    }
+    plan
+}
+
+/// Posts to ntfy and records delivered transitions.
+pub struct Notifier {
+    pub http: reqwest::Client,
+    pub publish_url: String,
+    pub token: Option<String>,
+    pub dashboard_url: String,
+    pub settings: Settings,
+}
+
+impl Notifier {
+    pub async fn publish(&self, n: &Notification) -> bool {
+        let mut req = self
+            .http
+            .post(format!(
+                "{}/{}",
+                self.publish_url.trim_end_matches('/'),
+                n.topic
+            ))
+            .timeout(std::time::Duration::from_secs(10))
+            .header("Title", &n.title)
+            .header("Priority", n.priority.to_string())
+            .header("Click", &self.dashboard_url)
+            .header(
+                "Actions",
+                format!("view, Open dashboard, {}", self.dashboard_url),
+            )
+            .body(n.message.clone());
+        if !n.tags.is_empty() {
+            req = req.header("Tags", n.tags.join(","));
+        }
+        if let Some(t) = &self.token {
+            req = req.bearer_auth(t);
+        }
+        match req.send().await {
+            Ok(r) if r.status().as_u16() < 300 => true,
+            Ok(r) => {
+                let status = r.status();
+                let body = r.text().await.unwrap_or_default();
+                tracing::warn!(
+                    "ntfy {}: HTTP {status} {}",
+                    n.topic,
+                    body.chars().take(120).collect::<String>()
+                );
+                false
+            }
+            Err(e) => {
+                tracing::warn!("ntfy {}: {e}", n.topic);
+                false
+            }
+        }
+    }
+
+    /// Evaluate and publish; returns the notifications delivered.
+    pub async fn run(
+        &self,
+        pool: &PgPool,
+        readings: &[Reading],
+        forecasts: &[Forecast],
+        now: NaiveDateTime,
+    ) -> Result<Vec<Notification>> {
+        let states = db::notification_states(pool).await?;
+        let mut sent = vec![];
+        for p in evaluate(readings, forecasts, &states, &self.settings, now) {
+            if !self.publish(&p.notification).await {
+                continue;
+            }
+            if let Some((key, state, value)) = &p.state {
+                db::set_notification_state(pool, key, state, *value, now).await?;
+            }
+            sent.push(p.notification);
+        }
+        Ok(sent)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use chrono::NaiveDate;
+
+    fn now() -> NaiveDateTime {
+        NaiveDate::from_ymd_opt(2026, 10, 1)
+            .unwrap()
+            .and_hms_opt(12, 0, 0)
+            .unwrap()
+    }
+
+    fn settings() -> Settings {
+        Settings {
+            prefix: "ping".into(),
+            rid_stale_alert_hours: 18.0,
+        }
+    }
+
+    fn reading(code: &str, level: f64, hours_ago: i64) -> Reading {
+        Reading {
+            station_code: code.into(),
+            timestamp: now() - Duration::hours(hours_ago),
+            water_level: Some(level),
+            discharge_percent: None,
+            source: Source::Rid,
+            rid_timestamp: Some(now() - Duration::hours(hours_ago)),
+        }
+    }
+
+    /// Run `evaluate` and apply the state of every planned transition, as a
+    /// successful publish would.
+    fn step(
+        rs: &[Reading],
+        fcs: &[Forecast],
+        st: &mut HashMap<String, String>,
+        at: NaiveDateTime,
+    ) -> Vec<Notification> {
+        let plan = evaluate(rs, fcs, st, &settings(), at);
+        for p in &plan {
+            if let Some((k, s, _)) = &p.state {
+                st.insert(k.clone(), s.clone());
+            }
+        }
+        plan.into_iter().map(|p| p.notification).collect()
+    }
+
+    fn titles(ns: &[Notification]) -> Vec<(&str, &str)> {
+        ns.iter()
+            .map(|n| (n.topic.as_str(), n.title.as_str()))
+            .collect()
+    }
+
+    #[test]
+    fn quiet_river_sends_nothing() {
+        let mut st = HashMap::new();
+        assert!(step(&[reading("P.1", 1.2, 1)], &[], &mut st, now()).is_empty());
+    }
+
+    #[test]
+    fn warning_once_then_silence_then_clear_with_hysteresis() {
+        let mut st = HashMap::new();
+        let sent = step(&[reading("P.1", 3.75, 0)], &[], &mut st, now());
+        assert_eq!(
+            titles(&sent),
+            [
+                ("ping-p1-warning", "Warning level at P.1"),
+                ("ping-warning", "Ping River: warning level reached")
+            ]
+        );
+        assert!(
+            sent[0]
+                .message
+                .starts_with("Nawarat Bridge, Chiang Mai city: 3.75 m at 01 Oct 12:00")
+        );
+        assert!(step(&[reading("P.1", 3.8, 0)], &[], &mut st, now()).is_empty());
+        // within the 0.10 m margin: still warning
+        assert!(step(&[reading("P.1", 3.65, 0)], &[], &mut st, now()).is_empty());
+        let sent = step(&[reading("P.1", 3.5, 0)], &[], &mut st, now());
+        assert_eq!(titles(&sent)[0], ("ping-p1-warning", "P.1 back to normal"));
+        assert_eq!(
+            titles(&sent)[1],
+            ("ping-warning", "Ping River: levels falling")
+        );
+    }
+
+    #[test]
+    fn danger_escalation_and_deescalation() {
+        let mut st = HashMap::new();
+        step(&[reading("P.1", 3.8, 0)], &[], &mut st, now());
+        let sent = step(&[reading("P.1", 4.3, 0)], &[], &mut st, now());
+        assert_eq!(titles(&sent)[0], ("ping-p1-danger", "DANGER level at P.1"));
+        assert_eq!(sent[0].priority, 5);
+        assert!(step(&[reading("P.1", 4.15, 0)], &[], &mut st, now()).is_empty());
+        let sent = step(&[reading("P.1", 4.0, 0)], &[], &mut st, now());
+        assert_eq!(
+            titles(&sent)[0],
+            ("ping-p1-danger", "P.1 back below danger level")
+        );
+    }
+
+    #[test]
+    fn capacity_guard_blocks_a_stale_threshold_but_not_p1() {
+        let mut st = HashMap::new();
+        let mut r = reading("P.77", 4.5, 0);
+        r.discharge_percent = Some(22.0);
+        assert!(step(&[r.clone()], &[], &mut st, now()).is_empty());
+        r.discharge_percent = Some(80.0);
+        assert_eq!(step(&[r], &[], &mut st, now()).len(), 2);
+        let mut p1 = reading("P.1", 3.8, 0);
+        p1.discharge_percent = Some(10.0);
+        assert_eq!(step(&[p1], &[], &mut st, now()).len(), 2);
+    }
+
+    fn outlook(p: f64) -> Forecast {
+        Forecast {
+            station_code: "P.1".into(),
+            horizon_hours: 24,
+            p_warning: p,
+            p_danger: 0.0,
+            predicted_max_level: 3.9,
+            current_level: 3.0,
+            as_of: now(),
+            model_version: "hgb-v3+test".into(),
+            trained_at: None,
+            source: "model".into(),
+            threshold_warning: 3.7,
+            threshold_danger: 4.2,
+            stages: None,
+        }
+    }
+
+    #[test]
+    fn outlook_on_off_with_hysteresis_and_heuristic_ignored() {
+        let mut st = HashMap::new();
+        let r = [reading("P.1", 3.0, 0)];
+        let sent = step(&r, &[outlook(0.62)], &mut st, now());
+        assert_eq!(
+            titles(&sent),
+            [(
+                "ping-p1-outlook",
+                "Early warning: Chiang Mai flood risk rising"
+            )]
+        );
+        assert!(sent[0].message.contains("62% chance") && sent[0].message.contains("3.90 m"));
+        assert!(step(&r, &[outlook(0.3)], &mut st, now()).is_empty());
+        assert_eq!(
+            step(&r, &[outlook(0.2)], &mut st, now())[0].title,
+            "Chiang Mai flood risk easing"
+        );
+        let mut h = outlook(0.9);
+        h.source = "heuristic".into();
+        assert!(step(&r, &[h], &mut st, now()).is_empty());
+    }
+
+    fn rid_row(code: &str, hours_ago: i64, level: Option<f64>, pct: Option<f64>) -> LatestReading {
+        LatestReading {
+            timestamp: now() - Duration::hours(hours_ago),
+            station_code: code.into(),
+            water_level: level,
+            discharge_percent: pct,
+        }
+    }
+
+    fn fill(code: &str, hours_ago: i64, level: f64) -> FillInfo {
+        FillInfo {
+            station_code: code.into(),
+            offset: 300.0,
+            filled_hours: 1,
+            latest: Some((now() - Duration::hours(hours_ago), level)),
+        }
+    }
+
+    #[test]
+    fn hii_reading_replaces_a_lagging_rid_reading() {
+        let rid = [
+            rid_row("P.1", 5, Some(3.2), Some(40.0)),
+            rid_row("P.21", 5, Some(1.0), None),
+            rid_row("P.67", 1, None, None),
+        ];
+        let rs = readings_with_hii(
+            &rid,
+            &[
+                fill("P.1", 0, 3.8),
+                fill("P.67", 1, 2.0),
+                fill("P.103", 0, 1.0),
+            ],
+        );
+        assert_eq!(rs[0].source, Source::Hii);
+        assert_eq!(rs[0].water_level, Some(3.8));
+        assert_eq!(rs[0].discharge_percent, Some(40.0)); // RID figure 5 h old
+        assert_eq!(rs[1].source, Source::Rid); // no twin
+        assert_eq!((rs[2].source, rs[2].water_level), (Source::Hii, Some(2.0))); // RID row had no level
+        let mut st = HashMap::new();
+        let sent = step(&rs, &[], &mut st, now());
+        assert!(
+            sent[0]
+                .message
+                .contains("(ThaiWater telemetry; the RID report is not in yet)")
+        );
+        // an older HII value never replaces a newer RID one
+        let rs = readings_with_hii(
+            &[rid_row("P.1", 0, Some(1.0), None)],
+            &[fill("P.1", 2, 3.9)],
+        );
+        assert_eq!(rs[0].source, Source::Rid);
+    }
+
+    #[test]
+    fn feed_is_stale_only_when_both_sources_are() {
+        let mut st = HashMap::new();
+        // RID 5 h behind, ThaiWater current: no stale alert (the 2026-10 nights)
+        let rs = readings_with_hii(
+            &[rid_row("P.1", 5, Some(1.2), None)],
+            &[fill("P.1", 0, 1.25)],
+        );
+        assert!(step(&rs, &[], &mut st, now()).is_empty());
+        // both behind: stale, then recovered
+        let rs = readings_with_hii(
+            &[rid_row("P.1", 5, Some(1.2), None)],
+            &[fill("P.1", 4, 1.25)],
+        );
+        let sent = step(&rs, &[], &mut st, now());
+        assert_eq!(
+            titles(&sent),
+            [("ping-status", "Ping River monitor: gauge feed stale")]
+        );
+        assert!(
+            sent[0]
+                .message
+                .starts_with("No new readings for 4 h (last 01 Oct 08:00)")
+        );
+        let rs = readings_with_hii(
+            &[rid_row("P.1", 5, Some(1.2), None)],
+            &[fill("P.1", 0, 1.25)],
+        );
+        assert_eq!(
+            step(&rs, &[], &mut st, now())[0].title,
+            "Ping River monitor: feed recovered"
+        );
+    }
+
+    #[test]
+    fn rid_lagging_a_day_is_reported_once_while_hii_is_current() {
+        let mut st = HashMap::new();
+        let rs = readings_with_hii(
+            &[rid_row("P.1", 20, Some(1.2), None)],
+            &[fill("P.1", 0, 1.25)],
+        );
+        let sent = step(&rs, &[], &mut st, now());
+        assert_eq!(
+            titles(&sent),
+            [("ping-status", "Ping River monitor: RID reports delayed")]
+        );
+        assert!(step(&rs, &[], &mut st, now()).is_empty());
+        let rs = readings_with_hii(&[rid_row("P.1", 0, Some(1.2), None)], &[]);
+        assert_eq!(
+            step(&rs, &[], &mut st, now())[0].title,
+            "Ping River monitor: RID reports current again"
+        );
+        // whole feed stale: only the general message
+        let mut st = HashMap::new();
+        let sent = step(&[reading("P.1", 1.2, 30)], &[], &mut st, now());
+        assert_eq!(
+            titles(&sent),
+            [("ping-status", "Ping River monitor: gauge feed stale")]
+        );
+    }
+
+    #[test]
+    fn undelivered_transitions_are_retried() {
+        let mut st = HashMap::new();
+        let plan = evaluate(&[reading("P.1", 3.8, 0)], &[], &st, &settings(), now());
+        assert_eq!(plan[0].state.as_ref().unwrap().1, "warning");
+        // nothing recorded (publish failed): the next run plans it again
+        let again = evaluate(&[reading("P.1", 3.8, 0)], &[], &st, &settings(), now());
+        assert_eq!(plan, again);
+        st.insert("level:P.1".into(), "warning".into());
+        assert!(
+            evaluate(&[reading("P.1", 3.8, 0)], &[], &st, &settings(), now())
+                .iter()
+                .all(|p| p.notification.topic == "ping-warning")
+        );
+    }
+
+    /// 652 hourly steps from src/notify.py itself (testdata/gen_notify_cases.py):
+    /// the same notifications, word for word, in the same order.
+    #[test]
+    fn matches_python_notify_on_generated_sequences() {
+        use std::io::Read;
+        let raw = include_bytes!("../testdata/notify_python_cases.json.gz");
+        let mut text = String::new();
+        flate2::read::GzDecoder::new(&raw[..])
+            .read_to_string(&mut text)
+            .unwrap();
+        let fixture: serde_json::Value = serde_json::from_str(&text).unwrap();
+        let ts = |v: &serde_json::Value| {
+            NaiveDateTime::parse_from_str(v.as_str().unwrap(), "%Y-%m-%dT%H:%M:%S").unwrap()
+        };
+        let mut compared = 0;
+        for seq in fixture["sequences"].as_array().unwrap() {
+            let mut states: HashMap<String, String> = HashMap::new();
+            for step in seq.as_array().unwrap() {
+                let readings: Vec<Reading> = step["readings"]
+                    .as_array()
+                    .unwrap()
+                    .iter()
+                    .map(|r| {
+                        let level = r["water_level"].as_f64();
+                        Reading {
+                            station_code: r["station_code"].as_str().unwrap().into(),
+                            timestamp: ts(&r["timestamp"]),
+                            water_level: level,
+                            discharge_percent: r["discharge_percent"].as_f64(),
+                            source: Source::Rid,
+                            rid_timestamp: level.map(|_| ts(&r["timestamp"])),
+                        }
+                    })
+                    .collect();
+                let forecasts: Vec<Forecast> = step["forecasts"]
+                    .as_array()
+                    .unwrap()
+                    .iter()
+                    .map(|f| Forecast {
+                        horizon_hours: f["horizon_hours"].as_u64().unwrap() as u32,
+                        p_warning: f["p_warning"].as_f64().unwrap(),
+                        predicted_max_level: f["predicted_max_level"].as_f64().unwrap(),
+                        source: f["source"].as_str().unwrap().into(),
+                        ..outlook(0.0)
+                    })
+                    .collect();
+                let down: Vec<&str> = step["down"]
+                    .as_array()
+                    .unwrap()
+                    .iter()
+                    .map(|v| v.as_str().unwrap())
+                    .collect();
+                let mut delivered = vec![];
+                for p in evaluate(
+                    &readings,
+                    &forecasts,
+                    &states,
+                    &settings(),
+                    ts(&step["now"]),
+                ) {
+                    if down.contains(&p.notification.topic.as_str()) {
+                        continue;
+                    }
+                    if let Some((k, s, _)) = &p.state {
+                        states.insert(k.clone(), s.clone());
+                    }
+                    delivered.push(p.notification);
+                }
+                let want: Vec<Notification> = step["sent"]
+                    .as_array()
+                    .unwrap()
+                    .iter()
+                    .map(|n| Notification {
+                        topic: n["topic"].as_str().unwrap().into(),
+                        title: n["title"].as_str().unwrap().into(),
+                        message: n["message"].as_str().unwrap().into(),
+                        priority: n["priority"].as_u64().unwrap() as u8,
+                        tags: n["tags"]
+                            .as_array()
+                            .unwrap()
+                            .iter()
+                            .map(|t| t.as_str().unwrap().into())
+                            .collect(),
+                    })
+                    .collect();
+                assert_eq!(delivered, want, "at {}", step["now"]);
+                compared += want.len();
+            }
+        }
+        assert!(compared > 700, "only {compared} notifications compared");
+    }
+}