feat: recalibrate P.77/P.75 thresholds; capacity guard on alerts

The first ntfy cycle announced "Warning level at P.77" at 3.02 m. That
gauge's 2.85 m threshold sat below its own dry-season baseline (2.6-2.7 m
at 8-14 % channel capacity): P.77 had been "above warning" for 761 of the
last 2 146 hours, at 22 % capacity. Across 2018-2024, 75-85 % capacity
reads 3.35-4.57 m and 95-105 % reads 4.27-5.08 m; set 4.30 / 4.90. P.75
moved 2.75/3.50 -> 3.20/3.65 on the same evidence (2024: 3.45 / 3.72).
The predictor already handles changed thresholds (regression-derived
probabilities until the Oct 1 retrain).

Second line of defence in notify.py: a clear->alert transition is only
announced when RID's discharge_percent for the reading is >= 60 %, so a
re-rated or datum-shifted gauge cannot page subscribers again. P.1 is
exempt (its stages come from the inundation map, not capacity); readings
without a capacity figure fall back to level only; the all-clear edge is
never blocked. 3 tests.

.env.example

@@ -84,6 +84,19 @@ SMTP_PORT=587
 SMTP_USERNAME=
 SMTP_PASSWORD=
 
+# Public push notifications via self-hosted ntfy (https://ntfy.sh, single binary).
+# Leave NTFY_SERVER empty to disable. Topics published: <prefix>-<station>-warning,
+# <prefix>-<station>-danger, <prefix>-warning, <prefix>-danger, <prefix>-p1-outlook,
+# <prefix>-status. See docs/NOTIFICATIONS.md.
+NTFY_SERVER=
+# Where the monitor POSTs (defaults to NTFY_SERVER). Use the local ntfy
+# address (loopback or Tailscale IP) so publishing does not depend on
+# DNS / the reverse proxy being up.
+NTFY_PUBLISH_URL=
+NTFY_TOPIC_PREFIX=ping
+NTFY_TOKEN=
+PUBLIC_URL=https://water.buildfor.life/
+
 # Matrix Alerting Configuration
 MATRIX_HOMESERVER=https://matrix.org
 MATRIX_ACCESS_TOKEN=

.gitea/workflows/security.yml

@@ -84,17 +84,22 @@ jobs:
           cache: pip
           cache-dependency-path: requirements.txt
 
-      - name: Install
+      # A fresh venv, not the runner's site-packages: the report must list the
+      # project's runtime deps, not whatever the runner image or a previous
+      # workflow happened to leave installed (semgrep once showed up here).
+      - name: Install into a clean venv
         run: |
-          python -m pip install --upgrade pip --root-user-action=ignore
+          python -m venv .lic && . .lic/bin/activate
+          pip install --upgrade pip --root-user-action=ignore
           pip install --root-user-action=ignore -r requirements.txt pip-licenses
 
       - name: Report
         run: |
+          . .lic/bin/activate
           pip-licenses --format=markdown --with-urls --output-file=licenses.md
           pip-licenses --format=json --output-file=licenses.json
-          echo "Copyleft licenses among runtime deps (informational):"
+          echo "Copyleft licenses among runtime deps (informational; LGPL is fine to link from MIT):"
-          pip-licenses --format=plain | grep -iE 'GPL|AGPL|LGPL' || echo "  none"
+          pip-licenses --format=plain --ignore-packages pip-licenses | grep -iE 'GPL|AGPL|LGPL' || echo "  none"
 
       - uses: actions/upload-artifact@v3
         with:

README.md

@@ -36,7 +36,11 @@ background in [Teaching a Model to See the Ping River Rise 13 Hours Early](https
 - **Shows it.** A Leaflet map with the river drawn as OSM geometry and styled by live
   discharge, rain gauges, the Chiang Mai inundation zones, per-station history, the
   forecast card, a replay of the 2024 flood, English/Thai, light/dark.
-- **Alerts** (optional) to a Matrix room when a gauge crosses its thresholds.
+- **Notifies.** Public push alerts over a self-hosted [ntfy](https://ntfy.sh): one
+  message when a gauge crosses its warning or danger level, one all-clear on the
+  way down, an opt-in early-warning topic from the model, nothing in between.
+  Subscribe from the free app, no account. Matrix room alerts for a team are
+  also supported.
 
 ## Quick start
 
@@ -79,6 +83,8 @@ Asia/Bangkok wall-clock without an offset suffix.
 | `GET /api/forecast/history/{code}?hours=N&horizon=24` | Forecasts as issued, for auditing lead time after the fact |
 | `GET /api/hii/rainfall/latest`, `/api/hii/waterlevel/latest` | Latest ThaiWater/HII gauge readings |
 | `GET /api/hii/rainfall/catchment?days=N` | HII gauge catchment-mean rain next to the Open-Meteo series the model uses |
+| `GET /api/forecast/skill?station_code=P.1` | Issued forecasts vs what happened, per deployed model version |
+| `GET /api/notifications` | ntfy server and topic names for the subscribe panel |
 | `GET /api/stats` | Row counts per source, date range, coverage |
 | `GET /health` | DB / upstream / memory checks |
 
@@ -122,7 +128,8 @@ models/              trained bundles + metrics.json (gitignored) and evaluation
 - [docs/DATA_SOURCES.md](docs/DATA_SOURCES.md) — every ingested and candidate source, endpoints, quirks
 - [docs/STATION_MANAGEMENT_GUIDE.md](docs/STATION_MANAGEMENT_GUIDE.md) — adding/editing gauges
 - [docs/DATABASE_DEPLOYMENT_GUIDE.md](docs/DATABASE_DEPLOYMENT_GUIDE.md), [POSTGRESQL_SETUP.md](POSTGRESQL_SETUP.md) — database setup
-- [docs/MATRIX_QUICK_START.md](docs/MATRIX_QUICK_START.md) — alert delivery
+- [docs/NOTIFICATIONS.md](docs/NOTIFICATIONS.md) — public push alerts: topics, semantics, ntfy deployment
+- [docs/MATRIX_QUICK_START.md](docs/MATRIX_QUICK_START.md) — Matrix room alerts for a team
 - [docs/GAP_FILLING_GUIDE.md](docs/GAP_FILLING_GUIDE.md) — data integrity tooling
 - [docs/references/NOTABLE_DOCUMENTS.md](docs/references/NOTABLE_DOCUMENTS.md) — official Thai government resources
 - Public overview: [buildfor.life/docs/tooling/ping-river-monitor](https://buildfor.life/docs/tooling/ping-river-monitor/)

docs/NOTIFICATIONS.md

@@ -0,0 +1,132 @@
+# Flood notifications (ntfy)
+
+Public push notifications for threshold crossings, without accounts, mailing
+lists or app-store review: the monitor publishes to a self-hosted
+[ntfy](https://ntfy.sh) server, and anyone subscribes to the topics they care
+about from the free ntfy app (iOS, Android, F-Droid) or a browser tab.
+
+ntfy is one Go binary with a sqlite cache: ~30 MB RSS idle, negligible CPU. It
+runs on the same VPS as the monitor.
+
+## What subscribers get
+
+Every message is a **transition**, never a state. Crossing up into a level sends
+one message; dropping back below it (with 0.10 m hysteresis) sends one
+all-clear. A river that sits at 3.9 m for three days produces two messages, not
+seventy-two. In a quiet season a subscriber hears nothing.
+
+| Topic | Trigger | Priority |
+|---|---|---|
+| `ping-warning` | any gauge crosses its warning threshold; levels falling back | 4 (high) / 2 |
+| `ping-danger` | any gauge crosses its danger threshold | 5 (max, breaks Do-Not-Disturb) |
+| `ping-<station>-warning` | that gauge crosses warning; back to normal | 4 / 2 |
+| `ping-<station>-danger` | that gauge crosses danger; back below danger | 5 / 3 |
+| `ping-p1-outlook` | model P(warning within 24 h) at P.1 rises through 50 % (clears below 25 %) | 4 / 2 |
+| `ping-status` | gauge feed stale ≥ 3 h; feed recovered | 3 / 2 |
+
+Station slugs are the code lowercased without the dot: `p1`, `p103`, `p67`.
+Thresholds are the ones in `src/ml/features.py` (`THRESHOLDS`): P.1 3.70 /
+4.20 m, P.103 5.95 / 6.75 m, and so on.
+
+The outlook topic is opt-in for a reason: it is model output, and the message
+says so. Observed-crossing topics only ever report a gauge reading.
+
+Each message carries a click-through and an "Open dashboard" action button to
+the public dashboard.
+
+## How it runs
+
+`src/notify.py` is called once per collection cycle inside the API process
+(leader only), right after the forecast precompute, so it sees exactly the
+readings and forecasts the dashboard shows. Per-key last-sent state is stored
+in the `notification_state` table of the monitor's own database, so a restart
+or redeploy never re-sends and never misses a crossing that happened while
+the service was down (the next cycle compares against the persisted state).
+
+If ntfy is unreachable the transition is **not** recorded, so it is retried
+on the next cycle rather than silently lost. Any other failure in the notify
+step is logged and never reaches the collection loop.
+
+The dashboard's "🔔 Get alerts" button appears only when `NTFY_SERVER` is
+set; it reads `GET /api/notifications` and renders subscribe links
+(`ntfy://` deep links for the app, https links for the web UI).
+
+## Deployment
+
+On the monitor VPS, as root:
+
+```bash
+cd /opt/thailand-water-monitor
+NTFY_DOMAIN=ntfy.buildfor.life bash scripts/install_ntfy.sh
+```
+
+This installs the ntfy .deb, writes `/etc/ntfy/server.yml` (listen on the
+host's Tailscale address, port 2586; anonymous read, token-only write, 72 h
+message cache, signup/login/metrics off, tight visitor limits), enables the
+systemd unit,
+creates the `monitor` user with **write-only access to `ping-*`**, mints a
+token, and appends `NTFY_SERVER` (public URL for subscribers),
+`NTFY_PUBLISH_URL` (loopback, what the monitor POSTs to), `NTFY_TOPIC_PREFIX`
+and `NTFY_TOKEN` to `.env` if they are not there yet. Then:
+
+```bash
+systemctl restart water-monitor
+journalctl -u water-monitor -n 20 | grep ntfy      # "ntfy notifications: https://... topics ping-*"
+curl -s 'https://ntfy.buildfor.life/ping-status/json?poll=1'   # anonymous read works
+```
+
+The reverse proxy is a separate VPS on the same tailnet, so ntfy listens on
+the monitor host's Tailscale address and nothing is exposed on a public
+interface. On the Caddy machine:
+
+```caddyfile
+ntfy.buildfor.life {
+    reverse_proxy <monitor tailscale ip>:2586
+}
+```
+
+Caddy proxies websockets and keeps long-poll connections open by default;
+subscribers hold one open. `behind-proxy: true` makes ntfy rate-limit on
+`X-Forwarded-For` rather than treating every subscriber as the proxy.
+
+Publishing does not depend on the domain: `NTFY_PUBLISH_URL` points the
+monitor at the Tailscale address directly, so a DNS or proxy problem never
+holds back an alert. Test the pipeline before the domain is live with
+`curl -s 'http://<tailscale ip>:2586/ping-status/json?poll=1'`.
+
+## Configuration
+
+| Variable | Default | Meaning |
+|---|---|---|
+| `NTFY_SERVER` | *(empty = off)* | public base URL subscribers use; shown on the dashboard |
+| `NTFY_PUBLISH_URL` | = `NTFY_SERVER` | where the monitor POSTs; the local ntfy address (`http://<tailscale ip>:2586`), so publishing never waits on DNS/proxy |
+| `NTFY_TOPIC_PREFIX` | `ping` | first segment of every topic |
+| `NTFY_TOKEN` | *(empty)* | bearer token if the server requires auth to publish (it does, see above) |
+| `PUBLIC_URL` | `https://water.buildfor.life/` | click-through target in messages |
+
+Tunables in `src/notify.py`: `CLEAR_MARGIN_M` (0.10), `OUTLOOK_ON` / `OUTLOOK_OFF`
+(0.50 / 0.25), stale feed threshold (3 h, argument to `evaluate`).
+
+## Testing
+
+`tests/test_notify.py` covers the state machine: quiet river sends nothing;
+crossing once, then silence while above, then all-clear; hysteresis on the way
+down; escalation to danger and back; basin digest grouping; outlook on/off;
+heuristic forecasts ignored; stale feed and recovery; state survives a restart
+through sqlite; a failed publish is retried next cycle.
+
+To exercise the real path against a real ntfy locally: run `ntfy serve` (any
+platform, same binary), set `NTFY_SERVER`/`NTFY_TOKEN`, seed readings, and
+poll the topic JSON. `scripts/e2e_notify.py` does exactly that if you want a
+template.
+
+## Why ntfy and not …
+
+- **Matrix** (`src/alerting.py`, still there): needs a homeserver account per
+  subscriber and a room invite; fine for a team, wrong for the public.
+- **Gotify**: also self-hosted and light, but Android-only client and one
+  account per subscriber.
+- **Email / SMS**: deliverability work, cost per message, no priority
+  semantics; ntfy can forward to email per subscription if someone wants it.
+- **Telegram / LINE bots**: platform lock-in and a bot token in the loop; can be
+  added later as ntfy→webhook fan-out without touching the monitor.

scripts/e2e_notify.py

@@ -0,0 +1,132 @@
+"""Drive the production notify path in-process: startup init -> seeded readings
+-> forecast cache -> _notify_transitions -> sqlite state -> real ntfy."""
+
+import asyncio
+import datetime
+import json
+import os
+import sys
+
+import requests
+
+os.environ.update(
+    DB_TYPE="sqlite",
+    WATER_DB_PATH=os.path.join(os.environ["LOCALAPPDATA"], "Temp", "smoke3.db"),
+    NTFY_SERVER="http://127.0.0.1:2586",
+    NTFY_TOKEN=os.environ.get("NTFY_TOKEN", ""),
+    NTFY_TOPIC_PREFIX="ping",
+)
+for f in ("smoke3.db",):
+    p = os.path.join(os.environ["LOCALAPPDATA"], "Temp", f)
+    if os.path.exists(p):
+        os.remove(p)
+
+from src import web_api  # noqa: E402
+from src.config import Config  # noqa: E402
+
+assert Config.NTFY_SERVER
+
+
+async def main():
+    # what the lifespan does at startup, minus the scheduler
+    from src import notify as notify_mod
+    from src.forecast_history import ForecastHistoryStore
+    from src.water_scraper_v3 import EnhancedWaterMonitorScraper
+
+    db_config = Config.get_database_config()
+    web_api.app_state["scraper"] = EnhancedWaterMonitorScraper(db_config)
+    store = ForecastHistoryStore(db_config["connection_string"], db_config["type"])
+    store.connect()
+    web_api.app_state["forecast_store"] = store
+    state = notify_mod.NotificationState(store.engine, store.db_type)
+    pub = notify_mod.NtfyPublisher(
+        Config.NTFY_SERVER, prefix=Config.NTFY_TOPIC_PREFIX, token=Config.NTFY_TOKEN
+    )
+    web_api.app_state["notify"] = (pub, state)
+
+    scraper = web_api.app_state["scraper"]
+    now = datetime.datetime.now().replace(minute=0, second=0, microsecond=0)
+
+    def seed(level_p1, level_p103, ts):
+        rows = [
+            {
+                "station_code": "P.1",
+                "station_id": 1,
+                "timestamp": ts,
+                "water_level": level_p1,
+                "discharge": 400.0,
+                "station_name_en": "Nawarat Bridge",
+                "station_name_th": "สะพานนวรัฐ",
+                "discharge_percent": 30.0,
+                "status": "active",
+            },
+            {
+                "station_code": "P.103",
+                "station_id": 2,
+                "timestamp": ts,
+                "water_level": level_p103,
+                "discharge": 300.0,
+                "station_name_en": "Ring Road 3",
+                "station_name_th": "วงแหวน 3",
+                "discharge_percent": 20.0,
+                "status": "active",
+            },
+        ]
+        scraper.db_adapter.save_measurements(rows)
+
+    def forecast(p):
+        with web_api.FORECAST_CACHE_LOCK:
+            web_api.FORECAST_CACHE["all"] = (
+                0,
+                [
+                    {
+                        "station_code": "P.1",
+                        "horizon_hours": 24,
+                        "p_warning": p,
+                        "predicted_max_level": 3.9,
+                        "source": "model",
+                    }
+                ],
+            )
+
+    def poll(topic):
+        out = []
+        for line in (
+            requests.get(f"{Config.NTFY_SERVER}/{topic}/json?poll=1", timeout=5)
+            .text.strip()
+            .splitlines()
+        ):
+            m = json.loads(line)
+            if m.get("event") == "message":
+                out.append(m.get("title") or m.get("message", "")[:40])
+        return out
+
+    # cycle 1: quiet
+    seed(1.6, 3.2, now - datetime.timedelta(hours=2))
+    forecast(0.02)
+    await web_api._notify_transitions()
+    # cycle 2: P.1 crosses warning, model outlook on
+    seed(3.75, 3.3, now - datetime.timedelta(hours=1))
+    forecast(0.7)
+    await web_api._notify_transitions()
+    # cycle 3: same state -> silence
+    seed(3.80, 3.3, now)
+    forecast(0.65)
+    await web_api._notify_transitions()
+
+    print("ping-p1-warning:", poll("ping-p1-warning"))
+    print("ping-warning:   ", poll("ping-warning"))
+    print("ping-p1-outlook:", poll("ping-p1-outlook"))
+    print("ping-p103-warning:", poll("ping-p103-warning"))
+    from sqlalchemy import text
+
+    with store.engine.connect() as c:
+        print(
+            "state table:",
+            c.execute(
+                text("SELECT key, state, value FROM notification_state ORDER BY key")
+            ).fetchall(),
+        )
+
+
+asyncio.run(main())

scripts/install_ntfy.sh

@@ -0,0 +1,102 @@
+#!/usr/bin/env bash
+# Install ntfy (https://ntfy.sh) as the public notification server for the
+# Ping River Monitor. Run as root on the monitor VPS. Idempotent.
+#
+#   NTFY_DOMAIN=ntfy.buildfor.life bash scripts/install_ntfy.sh
+#
+# What it does:
+#   - installs the ntfy .deb from the official GitHub release (single Go
+#     binary, ~30 MB RSS, sqlite message cache)
+#   - writes /etc/ntfy/server.yml: listens on the Tailscale address only
+#     (the reverse proxy is another VPS on the tailnet; nothing is exposed
+#     on a public interface), anonymous READ on all topics, WRITE only with
+#     a token. Override with NTFY_LISTEN=host:port.
+#   - creates the `monitor` publishing user + token, writes NTFY_SERVER /
+#     NTFY_TOKEN into /opt/thailand-water-monitor/.env if not present
+#
+# Reverse proxy (on the Caddy VPS, over Tailscale):
+#     ntfy.buildfor.life {
+#         reverse_proxy <this host's tailscale ip>:2586
+#     }
+# Caddy passes websockets and keeps long-poll connections open by default;
+# subscribers hold one open. ntfy runs with behind-proxy: true so rate
+# limits key on X-Forwarded-For, not on the proxy's address.
+set -euo pipefail
+
+NTFY_DOMAIN="${NTFY_DOMAIN:?set NTFY_DOMAIN, e.g. ntfy.buildfor.life}"
+NTFY_VERSION="${NTFY_VERSION:-2.28.0}"
+MONITOR_DIR="${MONITOR_DIR:-/opt/thailand-water-monitor}"
+TS_IP="$(tailscale ip -4 2>/dev/null | head -1 || true)"
+LISTEN="${NTFY_LISTEN:-${TS_IP:-127.0.0.1}:2586}"
+echo "ntfy will listen on ${LISTEN}"
+
+if ! command -v ntfy >/dev/null || [[ "$(ntfy --version 2>/dev/null | awk '{print $3}')" != "$NTFY_VERSION" ]]; then
+    tmp=$(mktemp -d)
+    curl -fsSL -o "$tmp/ntfy.deb" \
+        "https://github.com/binwiederhier/ntfy/releases/download/v${NTFY_VERSION}/ntfy_${NTFY_VERSION}_linux_amd64.deb"
+    dpkg -i "$tmp/ntfy.deb"
+    rm -rf "$tmp"
+fi
+
+install -d -m 755 /var/cache/ntfy /var/lib/ntfy
+cat > /etc/ntfy/server.yml <<EOF
+# Ping River Monitor notification server. Managed by scripts/install_ntfy.sh.
+base-url: "https://${NTFY_DOMAIN}"
+listen-http: "${LISTEN}"
+behind-proxy: true
+
+# Messages are kept so a phone that was offline still gets the crossing.
+cache-file: "/var/cache/ntfy/cache.db"
+cache-duration: "72h"
+
+# Everyone may subscribe; only the monitor (token) may publish.
+auth-file: "/var/lib/ntfy/user.db"
+auth-default-access: "read-only"
+
+# The monitor publishes a handful of messages per flood; be strict with
+# everything else so the box cannot be used as a free relay.
+visitor-request-limit-burst: 30
+visitor-request-limit-replenish: "10s"
+visitor-subscription-limit: 60
+visitor-message-daily-limit: 200
+attachment-cache-dir: ""
+enable-signup: false
+enable-login: false
+enable-metrics: false
+EOF
+
+systemctl enable --now ntfy
+systemctl restart ntfy
+sleep 1
+curl -fsS "http://${LISTEN}/v1/health" >/dev/null && echo "ntfy up on ${LISTEN}"
+
+# Publishing identity for the monitor
+if ! ntfy user list 2>/dev/null | grep -q '^user monitor (role'; then
+    NTFY_PASSWORD="$(openssl rand -base64 24)" ntfy user add --role=user monitor
+fi
+ntfy access monitor 'ping-*' write-only >/dev/null
+# 'ping-*' read stays anonymous via auth-default-access
+
+token=$(ntfy token list monitor 2>/dev/null | awk '/^- tk_/{print $2; exit}')  # '- tk_xxx (label), ...'
+if [[ -z "$token" ]]; then
+    token=$(ntfy token add --label "water-monitor" monitor | grep -oE 'tk_[A-Za-z0-9]+' | head -1)  # 'token tk_xxx created for user monitor'
+fi
+
+env_file="${MONITOR_DIR}/.env"
+if [[ -f "$env_file" ]] && ! grep -q '^NTFY_SERVER=' "$env_file"; then
+    {
+        echo ""
+        echo "# ntfy public notifications (scripts/install_ntfy.sh)"
+        echo "NTFY_SERVER=https://${NTFY_DOMAIN}"
+        echo "NTFY_PUBLISH_URL=http://${LISTEN}"
+        echo "NTFY_TOPIC_PREFIX=ping"
+        echo "NTFY_TOKEN=${token}"
+    } >> "$env_file"
+    echo "wrote NTFY_* to ${env_file}; restart water-monitor to enable"
+else
+    echo "NTFY_TOKEN=${token}"
+fi
+
+echo
+echo "Subscribe test (anonymous read):  curl -s 'http://${LISTEN}/ping-status/json?poll=1'"
+echo "Publish test (needs token):       curl -s -H 'Authorization: Bearer ${token}' -d 'hello' http://${LISTEN}/ping-status"

src/config.py

@@ -38,6 +38,17 @@ class Config:
     TARGET_URL = "https://hyd-app-db.rid.go.th/hydro1h.html"
     API_URL = "https://hyd-app-db.rid.go.th/webservice/getGroupHourlyWaterLevelReportAllHL.ashx"
     THAIWATER_API_KEY = os.getenv("THAIWATER_API_KEY")
+
+    # Public flood notifications (ntfy). Off unless NTFY_SERVER is set.
+    # NTFY_SERVER is what subscribers use (public https URL, shown on the
+    # dashboard). NTFY_PUBLISH_URL is where the monitor POSTs; defaults to
+    # NTFY_SERVER, set it to http://127.0.0.1:2586 when ntfy runs on the same
+    # host so publishing never depends on DNS/proxy/tunnel being up.
+    NTFY_SERVER = os.getenv("NTFY_SERVER", "").strip()
+    NTFY_PUBLISH_URL = os.getenv("NTFY_PUBLISH_URL", "").strip() or NTFY_SERVER
+    NTFY_TOPIC_PREFIX = os.getenv("NTFY_TOPIC_PREFIX", "ping").strip()
+    NTFY_TOKEN = os.getenv("NTFY_TOKEN", "").strip()  # publish token if ACL enabled
+    PUBLIC_URL = os.getenv("PUBLIC_URL", "https://water.buildfor.life/").strip()
     REQUEST_TIMEOUT = int(os.getenv("REQUEST_TIMEOUT", "30"))
     USER_AGENT = (
         "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "

src/ml/features.py

@@ -37,9 +37,17 @@ THRESHOLDS: Dict[str, Tuple[float, float]] = {
     "P.4A": (3.40, 3.90),
     "P.5": (4.55, 4.95),
     "P.67": (2.45, 2.90),
-    "P.75": (2.75, 3.50),
+    # P.75: 2024 (the only year with a full flood record, 191% capacity peak)
+    # puts 75-85% at 3.45 m and 95-105% at 3.72 m; 2018/2022 agree within
+    # 0.15 m. The 2026-08 value (2.75) alerted on 15 quiet-season hours.
+    "P.75": (3.20, 3.65),
     "P.76": (5.35, 5.45),
-    "P.77": (2.85, 3.35),
+    # P.77: recalibrated 2026-09-12. The 2026-08 value (2.85) sat below the
+    # gauge's own dry-season baseline (2.6-2.7 m at 8-14% capacity), so the
+    # first ntfy cycle fired a "warning" at 22% capacity. Across 2018-2024,
+    # 75-85% capacity reads 3.35-4.57 m and 95-105% 4.27-5.08 m; 2024 (the
+    # best-sampled flood year) gives 4.57 / 5.08. Slightly conservative:
+    "P.77": (4.30, 4.90),
     "P.81": (5.15, 6.30),
     # P.82 never reached 100% capacity in the record (max level 3.78, max 96.4%);
     # danger sits just below the observed maximum so the head can actually train.

src/notify.py

@@ -0,0 +1,454 @@
+"""Public flood notifications over ntfy.
+
+Runs once per collection cycle inside the API process (leader only), right
+after the forecast precompute, so it sees the same readings and forecasts the
+dashboard shows. Publishes to a self-hosted ntfy server; anyone subscribes to
+a topic from the free app or a browser, no account needed.
+
+Topics (all under one configurable prefix, default "ping"):
+
+  {prefix}-{station}-warning     observed level crossed the station's warning threshold
+  {prefix}-{station}-danger      observed level crossed the danger threshold
+  {prefix}-warning               any station crossed warning (basin-wide digest)
+  {prefix}-danger                any station crossed danger
+  {prefix}-p1-outlook            model early warning for Chiang Mai city: P.1's 24 h
+                                 warning probability crossed the alert level (opt-in;
+                                 the forecast is experimental and says so)
+  {prefix}-status                feed/monitor health: data stale, recovered
+
+Each notification is a TRANSITION, not a state: crossing UP into a level sends
+one message; dropping back below (with hysteresis) sends an all-clear. While
+the river sits above a threshold nothing is repeated, so a subscriber in a
+flood gets a handful of messages, not one an hour. The per-topic state is
+persisted (notification_state table) so a restart never re-sends.
+
+Everything is fail-safe: ntfy unreachable, table missing, malformed
+reading -> a logged warning, never an exception into the collection loop.
+"""
+
+import datetime
+import logging
+from dataclasses import dataclass
+from typing import Dict, Iterable, List, Optional
+
+import requests
+
+from .ml import features
+
+logger = logging.getLogger(__name__)
+
+# Hysteresis: an all-clear needs the level this far BELOW the threshold, so a
+# river bobbing around 3.70 m does not toggle warning/clear every hour.
+CLEAR_MARGIN_M = 0.10
+# Capacity guard. The level thresholds in features.THRESHOLDS were calibrated
+# from RID's discharge_percent (% of channel capacity); if RID re-rates a
+# gauge or moves its datum, the level crosses while capacity says the channel
+# is nearly empty (P.77, 2026-09: 3.0 m "warning" at 22 %). A crossing is
+# only announced when the reported capacity agrees that the river is high.
+# P.1 is exempt: its stages come from the municipal inundation map, not from
+# capacity. Readings without a capacity figure fall back to level only.
+CAPACITY_GUARD_MIN_PCT = 60.0
+CAPACITY_GUARD_EXEMPT = {"P.1"}
+# Outlook alert fires when p_warning(24h) rises through ON, clears below OFF.
+OUTLOOK_ON = 0.50
+OUTLOOK_OFF = 0.25
+# Below this the outlook is not announced at all (avoid "5 % chance" noise).
+OUTLOOK_HORIZON = 24
+
+STATION_NAMES: Dict[str, str] = {
+    "P.1": "Nawarat Bridge, Chiang Mai city",
+    "P.103": "Ring Road Bridge 3, Chiang Mai",
+    "P.67": "Ban Tae (Mae Taeng)",
+    "P.21": "Ban Rim Tai (Mae Rim)",
+    "P.75": "Ban Chai Lat",
+    "P.92": "Ban Muang Aut",
+    "P.20": "Ban Chiang Dao",
+    "P.4A": "Ban Mae Taeng",
+    "P.5": "Tha Nang Bridge (downstream)",
+    "P.81": "Ban Pong (downstream)",
+    "P.82": "Ban Sob Win",
+    "P.84": "Ban Panton",
+    "P.87": "Ban Pa Sang",
+    "P.77": "Ban Sop Mae Sapuat",
+    "P.85": "Ban Lai Kaew",
+    "P.76": "Ban Mae I Hai",
+}
+
+
+def _slug(code: str) -> str:
+    return code.lower().replace(".", "")
+
+
+@dataclass
+class Notification:
+    topic: str
+    title: str
+    message: str
+    priority: int = 3  # ntfy: 1 min .. 5 max
+    tags: Optional[List[str]] = None
+    click: Optional[str] = None
+
+
+class NtfyPublisher:
+    def __init__(
+        self,
+        server: str,
+        prefix: str = "ping",
+        token: Optional[str] = None,
+        dashboard_url: str = "https://water.buildfor.life/",
+        timeout: int = 10,
+    ):
+        self.server = server.rstrip("/")
+        self.prefix = prefix
+        self.token = token
+        self.dashboard_url = dashboard_url
+        self.timeout = timeout
+
+    def topic(self, *parts: str) -> str:
+        return "-".join([self.prefix, *parts])
+
+    def publish(self, n: Notification) -> bool:
+        headers = {
+            "Title": n.title,
+            "Priority": str(n.priority),
+            "Click": n.click or self.dashboard_url,
+            "Actions": f"view, Open dashboard, {n.click or self.dashboard_url}",
+        }
+        if n.tags:
+            headers["Tags"] = ",".join(n.tags)
+        if self.token:
+            headers["Authorization"] = f"Bearer {self.token}"
+        try:
+            r = requests.post(
+                f"{self.server}/{n.topic}",
+                data=n.message.encode("utf-8"),
+                headers=headers,
+                timeout=self.timeout,
+            )
+            if r.status_code >= 300:
+                logger.warning(f"ntfy {n.topic}: HTTP {r.status_code} {r.text[:120]}")
+                return False
+            return True
+        except Exception as error:
+            logger.warning(f"ntfy {n.topic}: {error}")
+            return False
+
+
+class NotificationState:
+    """Per-key last-sent state, in the monitor's own SQL database."""
+
+    def __init__(self, engine, db_type: str):
+        self.engine = engine
+        self.db_type = db_type
+        self._ensure()
+
+    def _ensure(self) -> None:
+        from sqlalchemy import text
+
+        ddl = (
+            "CREATE TABLE IF NOT EXISTS notification_state ("
+            "key VARCHAR(64) PRIMARY KEY, state VARCHAR(16) NOT NULL, "
+            "value NUMERIC(8,3), updated_at TIMESTAMP NOT NULL)"
+        )
+        try:
+            with self.engine.begin() as conn:
+                conn.execute(text(ddl))
+        except Exception as error:
+            # Postgres: two sessions racing CREATE TABLE IF NOT EXISTS can
+            # both pass the existence check; the loser fails with a unique
+            # violation on pg_type. The table exists either way; verify.
+            with self.engine.connect() as conn:
+                conn.execute(text("SELECT 1 FROM notification_state WHERE 1=0"))
+            logger.debug(f"notification_state DDL raced, table present: {error}")
+
+    def get(self, key: str) -> Optional[str]:
+        from sqlalchemy import text
+
+        with self.engine.connect() as conn:
+            row = conn.execute(
+                text("SELECT state FROM notification_state WHERE key = :k"), {"k": key}
+            ).fetchone()
+        return row[0] if row else None
+
+    def set(self, key: str, state: str, value: Optional[float] = None) -> None:
+        from sqlalchemy import text
+
+        now = datetime.datetime.now()
+        with self.engine.begin() as conn:
+            if self.db_type == "mysql":
+                sql = (
+                    "INSERT INTO notification_state (key, state, value, updated_at) "
+                    "VALUES (:k, :s, :v, :t) ON DUPLICATE KEY UPDATE "
+                    "state = VALUES(state), value = VALUES(value), updated_at = VALUES(updated_at)"
+                )
+            else:
+                sql = (
+                    "INSERT INTO notification_state (key, state, value, updated_at) "
+                    "VALUES (:k, :s, :v, :t) ON CONFLICT (key) DO UPDATE SET "
+                    "state = EXCLUDED.state, value = EXCLUDED.value, updated_at = EXCLUDED.updated_at"
+                )
+            conn.execute(text(sql), {"k": key, "s": state, "v": value, "t": now})
+
+
+class InMemoryState(NotificationState):
+    """For tests and when no SQL engine is available (loses state on restart)."""
+
+    def __init__(self):  # noqa: D107 - intentionally skips the SQL parent
+        self._d: Dict[str, str] = {}
+
+    def get(self, key: str) -> Optional[str]:
+        return self._d.get(key)
+
+    def set(self, key: str, state: str, value: Optional[float] = None) -> None:
+        self._d[key] = state
+
+
+def _level_state(level: float, warn: float, danger: float, prev: Optional[str]) -> str:
+    """'clear' | 'warning' | 'danger', with hysteresis on the way down."""
+    if level >= danger:
+        return "danger"
+    if level >= warn:
+        # from danger: stay 'danger' until below danger - margin
+        if prev == "danger" and level >= danger - CLEAR_MARGIN_M:
+            return "danger"
+        return "warning"
+    if prev in ("warning", "danger") and level >= warn - CLEAR_MARGIN_M:
+        return "warning"
+    return "clear"
+
+
+def evaluate(
+    readings: Iterable[dict],
+    forecasts: Iterable[dict],
+    state: NotificationState,
+    publisher: NtfyPublisher,
+    stale_after_h: float = 3.0,
+    now: Optional[datetime.datetime] = None,
+) -> List[Notification]:
+    """Compare current readings/forecasts with last-sent state; publish transitions.
+
+    readings: rows with station_code, water_level, timestamp (latest per station)
+    forecasts: /forecast rows (station_code, horizon_hours, p_warning, predicted_max_level)
+    Returns the notifications that were published (for logs/tests).
+    """
+    now = now or datetime.datetime.now()
+    sent: List[Notification] = []
+
+    def emit(n: Notification) -> bool:
+        ok = publisher.publish(n)
+        if ok:
+            sent.append(n)
+        return ok
+
+    # ---- observed levels, per station, plus basin-wide fan-out
+    basin_changes: Dict[str, List[str]] = {"warning": [], "danger": [], "clear": []}
+    latest_ts: Optional[datetime.datetime] = None
+    for r in readings:
+        code = r.get("station_code")
+        level = r.get("water_level")
+        if not code or level is None:
+            continue
+        try:
+            level = float(level)
+        except (TypeError, ValueError):
+            continue
+        ts = r.get("timestamp")
+        if isinstance(ts, str):
+            try:
+                ts = datetime.datetime.fromisoformat(ts)
+            except ValueError:
+                ts = None
+        if isinstance(ts, datetime.datetime) and (latest_ts is None or ts > latest_ts):
+            latest_ts = ts
+        warn, danger = features.get_thresholds(code)
+        key = f"level:{code}"
+        prev = state.get(key) or "clear"
+        cur = _level_state(level, warn, danger, prev)
+        pct = r.get("discharge_percent")
+        if (
+            cur != "clear"
+            and prev == "clear"
+            and code not in CAPACITY_GUARD_EXEMPT
+            and pct is not None
+        ):
+            try:
+                if float(pct) < CAPACITY_GUARD_MIN_PCT:
+                    logger.info(
+                        f"{code}: level {level:.2f} m >= {warn:.2f} but only "
+                        f"{float(pct):.0f}% capacity; threshold looks stale, not alerting"
+                    )
+                    continue
+            except (TypeError, ValueError):
+                pass
+        if cur == prev:
+            continue
+        name = STATION_NAMES.get(code, code)
+        slug = _slug(code)
+        when = (
+            ts.strftime("%d %b %H:%M") if isinstance(ts, datetime.datetime) else "now"
+        )
+        if cur == "danger":
+            ok = emit(
+                Notification(
+                    publisher.topic(slug, "danger"),
+                    f"DANGER level at {code}",
+                    f"{name}: {level:.2f} m at {when}, above the danger level of {danger:.2f} m.",
+                    priority=5,
+                    tags=["rotating_light", code],
+                )
+            )
+            basin_changes["danger"].append(f"{code} {level:.2f} m")
+        elif cur == "warning":
+            if prev == "danger":
+                ok = emit(
+                    Notification(
+                        publisher.topic(slug, "danger"),
+                        f"{code} back below danger level",
+                        f"{name}: {level:.2f} m at {when}; still above the warning level of {warn:.2f} m.",
+                        priority=3,
+                        tags=["arrow_down", code],
+                    )
+                )
+                basin_changes["clear"].append(f"{code} below danger ({level:.2f} m)")
+            else:
+                ok = emit(
+                    Notification(
+                        publisher.topic(slug, "warning"),
+                        f"Warning level at {code}",
+                        f"{name}: {level:.2f} m at {when}, above the warning level of {warn:.2f} m.",
+                        priority=4,
+                        tags=["warning", code],
+                    )
+                )
+                basin_changes["warning"].append(f"{code} {level:.2f} m")
+        else:  # clear
+            ok = emit(
+                Notification(
+                    publisher.topic(slug, "warning"),
+                    f"{code} back to normal",
+                    f"{name}: {level:.2f} m at {when}, below the warning level of {warn:.2f} m.",
+                    priority=2,
+                    tags=["white_check_mark", code],
+                )
+            )
+            basin_changes["clear"].append(f"{code} normal ({level:.2f} m)")
+        # Only remember the transition once it was actually delivered: if ntfy
+        # was down, the next cycle retries instead of silently swallowing a
+        # flood crossing.
+        if ok:
+            state.set(key, cur, level)
+
+    if basin_changes["danger"]:
+        emit(
+            Notification(
+                publisher.topic("danger"),
+                "Ping River: danger level reached",
+                "; ".join(basin_changes["danger"]),
+                priority=5,
+                tags=["rotating_light"],
+            )
+        )
+    if basin_changes["warning"]:
+        emit(
+            Notification(
+                publisher.topic("warning"),
+                "Ping River: warning level reached",
+                "; ".join(basin_changes["warning"]),
+                priority=4,
+                tags=["warning"],
+            )
+        )
+    if basin_changes["clear"]:
+        emit(
+            Notification(
+                publisher.topic("warning"),
+                "Ping River: levels falling",
+                "; ".join(basin_changes["clear"]),
+                priority=2,
+                tags=["white_check_mark"],
+            )
+        )
+
+    # ---- model outlook for the city gauge (opt-in topic, experimental)
+    p1 = next(
+        (
+            f
+            for f in forecasts
+            if f.get("station_code") == "P.1"
+            and f.get("horizon_hours") == OUTLOOK_HORIZON
+            and f.get("source") == "model"
+        ),
+        None,
+    )
+    if p1 and p1.get("p_warning") is not None:
+        p = float(p1["p_warning"])
+        key = "outlook:P.1"
+        prev = state.get(key) or "off"
+        cur = (
+            "on" if (p >= OUTLOOK_ON or (prev == "on" and p >= OUTLOOK_OFF)) else "off"
+        )
+        if cur != prev:
+            peak = p1.get("predicted_max_level")
+            warn, _ = features.get_thresholds("P.1")
+            if cur == "on":
+                ok = emit(
+                    Notification(
+                        publisher.topic("p1-outlook"),
+                        "Early warning: Chiang Mai flood risk rising",
+                        f"The forecast model gives a {p * 100:.0f}% chance that Nawarat Bridge (P.1) "
+                        f"reaches {warn:.2f} m within 24 h"
+                        + (
+                            f" (expected peak {float(peak):.2f} m)"
+                            if peak is not None
+                            else ""
+                        )
+                        + ". Experimental model output, not an official warning; "
+                        "follow ThaiWater/TMD for official alerts.",
+                        priority=4,
+                        tags=["crystal_ball"],
+                    )
+                )
+            else:
+                ok = emit(
+                    Notification(
+                        publisher.topic("p1-outlook"),
+                        "Chiang Mai flood risk easing",
+                        f"The model's 24 h probability of reaching {warn:.2f} m at P.1 has dropped to {p * 100:.0f}%.",
+                        priority=2,
+                        tags=["crystal_ball"],
+                    )
+                )
+            if ok:
+                state.set(key, cur, p)
+
+    # ---- feed health
+    if latest_ts is not None:
+        age_h = (now - latest_ts).total_seconds() / 3600.0
+        key = "feed"
+        prev = state.get(key) or "ok"
+        cur = "stale" if age_h >= stale_after_h else "ok"
+        if cur != prev:
+            if cur == "stale":
+                ok = emit(
+                    Notification(
+                        publisher.topic("status"),
+                        "Ping River monitor: gauge feed stale",
+                        f"No new readings for {age_h:.0f} h (last {latest_ts:%d %b %H:%M}). "
+                        "Levels and forecasts on the dashboard are not current.",
+                        priority=3,
+                        tags=["hourglass"],
+                    )
+                )
+            else:
+                ok = emit(
+                    Notification(
+                        publisher.topic("status"),
+                        "Ping River monitor: feed recovered",
+                        f"Readings are current again (latest {latest_ts:%d %b %H:%M}).",
+                        priority=2,
+                        tags=["white_check_mark"],
+                    )
+                )
+            if ok:
+                state.set(key, cur, age_h)
+    return sent

src/static/dashboard.html

@@ -113,6 +113,25 @@
         .lang-toggle { padding: 9px 12px; font-size: .78rem; font-weight: 800; white-space: nowrap; }
         .lang-toggle[data-active-lang="th"] { background: var(--mint); border-color: var(--mint-border); color: var(--mint-ink); }
         .theme-toggle { padding: 9px 11px; font-size: .95rem; line-height: 1; }
+        .alerts-panel { margin-bottom: 14px; padding: 18px 20px; border-radius: 16px; background: var(--card); border: 1px solid var(--border); box-shadow: var(--shadow); }
+        .alerts-head { display: flex; justify-content: space-between; align-items: center; gap: 12px; }
+        .alerts-head h2 { margin: 0; font-size: 1.15rem; }
+        .alerts-close { background: transparent; border: 0; color: var(--muted); font-size: 1.1rem; cursor: pointer; padding: 4px 8px; }
+        .alerts-intro { color: var(--muted); font-size: .92rem; line-height: 1.5; margin: 8px 0 12px; }
+        .alerts-server { display: flex; align-items: center; gap: 10px; flex-wrap: wrap; font-size: .9rem; margin-bottom: 12px; }
+        .alerts-server code { background: var(--surface); border: 1px solid var(--border); padding: 4px 8px; border-radius: 8px; font-size: .9rem; }
+        .alerts-server button { padding: 4px 10px; font-size: .8rem; }
+        .alerts-grid { display: grid; grid-template-columns: repeat(auto-fill, minmax(280px, 1fr)); gap: 10px; }
+        .alerts-topic { border: 1px solid var(--border); border-radius: 12px; padding: 10px 12px; background: var(--surface); display: flex; flex-direction: column; gap: 4px; }
+        .alerts-topic .name { font-weight: 600; font-size: .95rem; }
+        .alerts-topic .desc { color: var(--muted); font-size: .82rem; line-height: 1.4; }
+        .alerts-topic .row { display: flex; align-items: center; gap: 8px; margin-top: 4px; flex-wrap: wrap; }
+        .alerts-topic code { font-size: .82rem; background: var(--card); border: 1px solid var(--border); padding: 2px 6px; border-radius: 6px; }
+        .alerts-topic a { font-size: .82rem; }
+        .alerts-topic.danger { border-color: rgba(220, 38, 38, .45); }
+        .alerts-topic.outlook { border-style: dashed; }
+        .alerts-foot { color: var(--muted); font-size: .82rem; margin: 12px 0 0; line-height: 1.6; }
+        .alerts-disclaimer { display: block; margin-top: 4px; }
         .leaflet-popup-content-wrapper, .leaflet-popup-tip { background: var(--card); color: var(--ink); }
         .leaflet-container a.leaflet-popup-close-button { color: var(--muted); }
         .leaflet-bar a, .leaflet-control-attribution { background: var(--surface); color: var(--ink); border-color: var(--border); }
@@ -348,10 +367,33 @@
             <button id="refresh-button" type="button" data-i18n="action.refresh">↻ Refresh</button>
             <button id="lang-toggle" class="lang-toggle" type="button" data-active-lang="en" aria-label="Switch to Thai">ไทย</button>
             <button id="theme-toggle" class="theme-toggle" type="button" data-i18n-aria="theme.toggle" aria-label="Switch to dark mode" title="Switch to dark mode">🌙</button>
+            <button id="alerts-button" type="button" data-i18n="alerts.button" style="display:none">🔔 Get alerts</button>
             <button id="replay-2024" type="button" data-i18n="replay.start">▶ Replay Oct 2024 flood</button>
         </div>
     </header>
 
+    <section id="alerts-panel" class="alerts-panel" style="display:none" aria-labelledby="alerts-title">
+        <div class="alerts-head">
+            <h2 id="alerts-title" data-i18n="alerts.title">Flood alerts on your phone</h2>
+            <button type="button" class="alerts-close" id="alerts-close" data-i18n-aria="alerts.close" aria-label="Close">✕</button>
+        </div>
+        <p class="alerts-intro" data-i18n="alerts.intro">Free push notifications when a gauge crosses its warning or danger level, and an all-clear when it drops back. No account: install the ntfy app (iOS / Android / any browser), add the server, subscribe to the topics you want. You get a message only when something changes: a few per flood, none in a quiet season.</p>
+        <div class="alerts-server">
+            <span data-i18n="alerts.server">Server</span>
+            <code id="alerts-server-url"></code>
+            <button type="button" id="alerts-copy" data-i18n="alerts.copy">Copy</button>
+        </div>
+        <div class="alerts-grid" id="alerts-topics"></div>
+        <p class="alerts-foot">
+            <span data-i18n="alerts.apps">Apps:</span>
+            <a href="https://apps.apple.com/us/app/ntfy/id1625396347" target="_blank" rel="noopener">iOS</a> ·
+            <a href="https://play.google.com/store/apps/details?id=io.heckel.ntfy" target="_blank" rel="noopener">Android</a> ·
+            <a href="https://f-droid.org/en/packages/io.heckel.ntfy/" target="_blank" rel="noopener">F-Droid</a> ·
+            <a id="alerts-web-link" href="#" target="_blank" rel="noopener" data-i18n="alerts.web">Web (no install)</a>
+            <span class="alerts-disclaimer" data-i18n="alerts.disclaimer">Unofficial community service, best effort. For official warnings follow ThaiWater / TMD / your district office.</span>
+        </p>
+    </section>
+
     <section id="flood-verdict" role="status" aria-live="polite" style="display:none;margin-bottom:14px;padding:15px 18px;border-radius:16px;border:1px solid;display:none">
         <div style="display:flex;gap:12px;align-items:baseline;flex-wrap:wrap">
             <strong id="verdict-icon" style="font-size:1.2rem"></strong>
@@ -585,6 +627,31 @@
             'skill.worse': (v, prev, d) => `Current model ${v} has a higher peak error than ${prev} so far (+${d} cm).`,
             'skill.caveat.quiet': 'All verified hours so far were below 2 m: this measures quiet-river accuracy only. The model is built and judged for flood onset (lead time before 3.70 m), which no quiet week can test — see the backtests in the documentation.',
             'skill.caveat.regime': 'Versions served different weeks; the ≥ 2 m column compares them on the hours that matter.',
+            'alerts.button': '🔔 Get alerts',
+            'alerts.title': 'Flood alerts on your phone',
+            'alerts.intro': 'Free push notifications when a gauge crosses its warning or danger level, and an all-clear when it drops back. No account: install the ntfy app (iOS / Android / any browser), add the server, subscribe to the topics you want. You get a message only when something changes: a few per flood, none in a quiet season.',
+            'alerts.server': 'Server',
+            'alerts.copy': 'Copy',
+            'alerts.copied': 'Copied',
+            'alerts.close': 'Close',
+            'alerts.apps': 'Apps:',
+            'alerts.web': 'Web (no install)',
+            'alerts.disclaimer': 'Unofficial community service, best effort. For official warnings follow ThaiWater / TMD / your district office.',
+            'alerts.subscribe': 'Subscribe in app',
+            'alerts.t.warning': 'Any gauge: warning level',
+            'alerts.t.warning.d': 'One message when any Ping River gauge crosses its warning level, and when levels fall back. The one to pick if unsure.',
+            'alerts.t.danger': 'Any gauge: danger level',
+            'alerts.t.danger.d': 'Only the serious crossings, basin-wide. Highest priority: rings through Do Not Disturb on most phones.',
+            'alerts.t.p1.warning': 'Chiang Mai city (P.1) warning',
+            'alerts.t.p1.warning.d': 'Nawarat Bridge crosses 3.70 m (stage 1: low-lying riverside areas), and the all-clear.',
+            'alerts.t.p1.danger': 'Chiang Mai city (P.1) danger',
+            'alerts.t.p1.danger.d': 'Nawarat Bridge crosses 4.20 m (stage 5: inner city districts).',
+            'alerts.t.p103.warning': 'Ring Road 3 (P.103) warning',
+            'alerts.t.p103.warning.d': 'Downstream city gauge crosses 5.95 m.',
+            'alerts.t.outlook': 'Early warning (model forecast)',
+            'alerts.t.outlook.d': 'Experimental: the forecast model gives a ≥ 50 % chance that P.1 reaches its warning level within 24 h. Up to ~13 h earlier than the gauge, but it can be wrong.',
+            'alerts.t.status': 'Monitor status',
+            'alerts.t.status.d': 'Gauge feed stale / recovered. For people who rely on the dashboard.',
             'skill.single': (v) => `Only ${v} has enough verified hours yet; the next retrain adds a row to compare.`,
             'skill.young': (v, n, min) => `${v} has ${n} verified hours; a comparison needs ${min}.`,
             'skill.none': 'No verified forecasts yet — the first appear 24 h after a model starts serving.',
@@ -781,6 +848,31 @@
             'skill.worse': (v, prev, d) => `โมเดลปัจจุบัน ${v} มีค่าคลาดเคลื่อนสูงกว่า ${prev} (+${d} ซม.)`,
             'skill.caveat.quiet': 'ชั่วโมงที่ตรวจสอบทั้งหมดอยู่ต่ำกว่า 2 ม.: วัดได้เพียงความแม่นยำช่วงน้ำปกติ โมเดลถูกสร้างและประเมินสำหรับช่วงน้ำเริ่มท่วม (เวลาเตือนล่วงหน้าก่อน 3.70 ม.) ซึ่งสัปดาห์ปกติทดสอบไม่ได้ — ดูผลทดสอบย้อนหลังในเอกสาร',
             'skill.caveat.regime': 'แต่ละเวอร์ชันให้บริการคนละช่วงเวลา คอลัมน์ ≥ 2 ม. เปรียบเทียบเฉพาะชั่วโมงที่สำคัญ',
+            'alerts.button': '🔔 รับการแจ้งเตือน',
+            'alerts.title': 'แจ้งเตือนน้ำท่วมบนมือถือของคุณ',
+            'alerts.intro': 'การแจ้งเตือนฟรีเมื่อระดับน้ำที่สถานีใดข้ามระดับเฝ้าระวังหรือระดับอันตราย และแจ้งเมื่อกลับสู่ปกติ ไม่ต้องสมัครสมาชิก: ติดตั้งแอป ntfy (iOS / Android / เบราว์เซอร์) เพิ่มเซิร์ฟเวอร์ แล้วเลือกหัวข้อที่ต้องการ คุณจะได้รับข้อความเฉพาะเมื่อมีการเปลี่ยนแปลง: ไม่กี่ข้อความต่อเหตุการณ์น้ำท่วม และไม่มีเลยในช่วงปกติ',
+            'alerts.server': 'เซิร์ฟเวอร์',
+            'alerts.copy': 'คัดลอก',
+            'alerts.copied': 'คัดลอกแล้ว',
+            'alerts.close': 'ปิด',
+            'alerts.apps': 'แอป:',
+            'alerts.web': 'เว็บ (ไม่ต้องติดตั้ง)',
+            'alerts.disclaimer': 'บริการชุมชนอย่างไม่เป็นทางการ พยายามอย่างดีที่สุด สำหรับคำเตือนอย่างเป็นทางการโปรดติดตาม ThaiWater / กรมอุตุนิยมวิทยา / สำนักงานอำเภอของคุณ',
+            'alerts.subscribe': 'สมัครในแอป',
+            'alerts.t.warning': 'สถานีใดก็ได้: ระดับเฝ้าระวัง',
+            'alerts.t.warning.d': 'หนึ่งข้อความเมื่อสถานีใดในแม่น้ำปิงข้ามระดับเฝ้าระวัง และเมื่อระดับน้ำลดลง หากไม่แน่ใจให้เลือกอันนี้',
+            'alerts.t.danger': 'สถานีใดก็ได้: ระดับอันตราย',
+            'alerts.t.danger.d': 'เฉพาะการข้ามระดับที่ร้ายแรง ทั้งลุ่มน้ำ ความสำคัญสูงสุด: ดังผ่านโหมดห้ามรบกวนในโทรศัพท์ส่วนใหญ่',
+            'alerts.t.p1.warning': 'เมืองเชียงใหม่ (P.1) ระดับเฝ้าระวัง',
+            'alerts.t.p1.warning.d': 'สะพานนวรัฐข้าม 3.70 ม. (ระยะที่ 1: พื้นที่ริมน้ำที่ต่ำ) และแจ้งเมื่อกลับสู่ปกติ',
+            'alerts.t.p1.danger': 'เมืองเชียงใหม่ (P.1) ระดับอันตราย',
+            'alerts.t.p1.danger.d': 'สะพานนวรัฐข้าม 4.20 ม. (ระยะที่ 5: ย่านใจกลางเมือง)',
+            'alerts.t.p103.warning': 'ถนนวงแหวน 3 (P.103) ระดับเฝ้าระวัง',
+            'alerts.t.p103.warning.d': 'สถานีท้ายเมืองข้าม 5.95 ม.',
+            'alerts.t.outlook': 'เตือนล่วงหน้า (แบบจำลองพยากรณ์)',
+            'alerts.t.outlook.d': 'ทดลอง: แบบจำลองพยากรณ์ให้โอกาส ≥ 50% ที่ P.1 จะถึงระดับเฝ้าระวังภายใน 24 ชม. เร็วกว่าสถานีวัดได้ถึง ~13 ชม. แต่อาจผิดพลาดได้',
+            'alerts.t.status': 'สถานะระบบ',
+            'alerts.t.status.d': 'ข้อมูลสถานีล่าช้า / กลับมาปกติ สำหรับผู้ที่พึ่งพาแดชบอร์ด',
             'skill.single': (v) => `มีเพียง ${v} ที่มีข้อมูลตรวจสอบเพียงพอ การฝึกครั้งถัดไปจะเพิ่มแถวให้เปรียบเทียบ`,
             'skill.young': (v, n, min) => `${v} มีข้อมูลตรวจสอบ ${n} ชั่วโมง ต้องการอย่างน้อย ${min} เพื่อเปรียบเทียบ`,
             'skill.none': 'ยังไม่มีพยากรณ์ที่ตรวจสอบได้ — จะเริ่มมี 24 ชม. หลังโมเดลเริ่มทำงาน',
@@ -905,6 +997,60 @@
         return typeof value === 'function' ? value(...args) : value;
     }
 
+    // ---- public push notifications (ntfy) -------------------------------------
+    let ALERTS_CFG = null;
+    const ALERT_TOPICS = [
+        { key: 'warning',     topic: 'warning',     cls: '' },
+        { key: 'danger',      topic: 'danger',      cls: 'danger' },
+        { key: 'p1.warning',  topic: 'p1-warning',  cls: '' },
+        { key: 'p1.danger',   topic: 'p1-danger',   cls: 'danger' },
+        { key: 'p103.warning', topic: 'p103-warning', cls: '' },
+        { key: 'outlook',     topic: 'p1-outlook',  cls: 'outlook' },
+        { key: 'status',      topic: 'status',      cls: '' },
+    ];
+    async function loadAlertsConfig() {
+        try {
+            const r = await fetch('/api/notifications');
+            if (!r.ok) return;
+            const cfg = await r.json();
+            if (!cfg.enabled || !cfg.server) return;
+            ALERTS_CFG = cfg;
+            $('alerts-button').style.display = '';
+            renderAlertsPanel();
+        } catch (e) { /* no notifications configured */ }
+    }
+    function renderAlertsPanel() {
+        if (!ALERTS_CFG) return;
+        const server = ALERTS_CFG.server.replace(/\/$/, '');
+        const host = server.replace(/^https?:\/\//, '');
+        $('alerts-server-url').textContent = host;
+        $('alerts-web-link').href = server + '/' + ALERTS_CFG.prefix + '-warning';
+        $('alerts-topics').innerHTML = ALERT_TOPICS.map(tp => {
+            const full = ALERTS_CFG.prefix + '-' + tp.topic;
+            const url = server + '/' + full;
+            return `<div class="alerts-topic ${tp.cls}">
+                <div class="name">${esc(t('alerts.t.' + tp.key))}</div>
+                <div class="desc">${esc(t('alerts.t.' + tp.key + '.d'))}</div>
+                <div class="row"><code>${esc(full)}</code> <a href="ntfy://${esc(host)}/${esc(full)}">${esc(t('alerts.subscribe'))}</a> · <a href="${esc(url)}" target="_blank" rel="noopener">web</a></div>
+            </div>`;
+        }).join('');
+    }
+    function esc(x) { return String(x).replace(/[&<>"']/g, c => ({'&':'&amp;','<':'&lt;','>':'&gt;','"':'&quot;',"'":'&#39;'}[c])); }
+    $('alerts-button').addEventListener('click', () => {
+        const p = $('alerts-panel');
+        const open = p.style.display === 'none';
+        p.style.display = open ? '' : 'none';
+        if (open) { renderAlertsPanel(); p.scrollIntoView({ behavior: 'smooth', block: 'start' }); }
+    });
+    $('alerts-close').addEventListener('click', () => { $('alerts-panel').style.display = 'none'; });
+    $('alerts-copy').addEventListener('click', async () => {
+        try {
+            await navigator.clipboard.writeText(ALERTS_CFG ? ALERTS_CFG.server : '');
+            $('alerts-copy').textContent = t('alerts.copied');
+            setTimeout(() => { $('alerts-copy').textContent = t('alerts.copy'); }, 1500);
+        } catch (e) { /* clipboard blocked */ }
+    });
+
     function applyTranslations() {
         document.documentElement.lang = state.lang;
         document.querySelectorAll('[data-i18n]').forEach((el) => {
@@ -959,6 +1105,7 @@
     function cssVar(name) { return getComputedStyle(document.documentElement).getPropertyValue(name).trim(); }
 
     function setLang(lang) {
+        setTimeout(renderAlertsPanel, 0);
         state.lang = lang;
         try { localStorage.setItem(LANG_KEY, lang); } catch (e) { /* private mode */ }
         applyTranslations();
@@ -2052,6 +2199,7 @@
                 : t('forecast.expand', stations.length);
             card.style.display = 'block';
             loadSkill(); // non-blocking; panel stays hidden until there is verified data
+            loadAlertsConfig(); // shows the "Get alerts" button only when ntfy is configured
         } catch (error) {
             card.style.display = 'none';
         }

src/web_api.py

@@ -210,7 +210,9 @@ async def lifespan(app: FastAPI):
     app_state["leader_lock"] = _acquire_collection_leadership(
         Config.COLLECTION_LEADER_PORT
     )
+    app_state["notify"] = None
     if app_state["leader_lock"]:
+        app_state["notify"] = _init_notifications()
         app_state["scraping_task"] = asyncio.create_task(background_scraping_task())
         logger.info("This worker is the background-collection leader")
     else:
@@ -296,6 +298,73 @@ async def _persist_rain():
         logger.warning(f"rain persistence failed: {e}")
 
 
+def _init_notifications():
+    """Publisher + persisted state for ntfy, or None if off/unavailable.
+
+    Called only by the collection leader: it is the one process that
+    publishes, so the notification_state DDL runs exactly once per host.
+    """
+    if not Config.NTFY_SERVER:
+        return None
+    try:
+        from . import notify as notify_mod
+
+        store = app_state.get("forecast_store")
+        if store and not store.engine:
+            store.connect()
+        state = (
+            notify_mod.NotificationState(store.engine, store.db_type)
+            if store and store.engine
+            else notify_mod.InMemoryState()
+        )
+        if isinstance(state, notify_mod.InMemoryState):
+            logger.warning(
+                "ntfy: no SQL store; notification state is in-memory "
+                "(a restart may re-send the current level)"
+            )
+        publisher = notify_mod.NtfyPublisher(
+            Config.NTFY_PUBLISH_URL,
+            prefix=Config.NTFY_TOPIC_PREFIX,
+            token=Config.NTFY_TOKEN or None,
+            dashboard_url=Config.PUBLIC_URL,
+        )
+        logger.info(
+            f"ntfy notifications: publish to {Config.NTFY_PUBLISH_URL}, "
+            f"subscribers use {Config.NTFY_SERVER}, topics {Config.NTFY_TOPIC_PREFIX}-*"
+        )
+        return publisher, state
+    except Exception as e:
+        logger.error(f"ntfy init failed (notifications off): {e}")
+        return None
+
+
+async def _notify_transitions():
+    """Publish flood/outlook/feed transitions to ntfy (leader only, fail-safe)."""
+    cfg = app_state.get("notify")
+    if not cfg:
+        return
+    publisher, state = cfg
+    try:
+        from . import notify as notify_mod
+
+        scraper = app_state["scraper"]
+        readings = await asyncio.to_thread(
+            scraper.db_adapter.get_latest_measurements, 200
+        )
+        with FORECAST_CACHE_LOCK:
+            cached = FORECAST_CACHE.get("all")
+        forecasts = cached[1] if cached else []
+        sent = await asyncio.to_thread(
+            notify_mod.evaluate, readings, forecasts, state, publisher
+        )
+        if sent:
+            logger.info(
+                "ntfy: published " + ", ".join(f"{n.topic}: {n.title}" for n in sent)
+            )
+    except Exception as e:
+        logger.warning(f"ntfy notify cycle failed: {e}")
+
+
 async def _precompute_forecasts():
     """Refresh the forecast cache and persist the issued forecasts (leader only)."""
     try:
@@ -402,6 +471,10 @@ async def background_scraping_task():
                     # evaluation.
                     await _precompute_forecasts()
 
+                    # Push notifications for threshold crossings (uses the
+                    # forecasts just computed; no-op unless NTFY_SERVER set).
+                    await _notify_transitions()
+
                 app_state["is_scraping"] = False
 
             # Calculate next run time
@@ -1230,6 +1303,28 @@ async def get_forecast_history(
     return await asyncio.to_thread(store.fetch, station_code, start_dt, end_dt, horizon)
 
 
+@app.get("/api/notifications")
+async def get_notifications_config():
+    """Public ntfy settings so the dashboard can offer subscribe links."""
+    server = Config.NTFY_SERVER
+    if not server:
+        return {"enabled": False}
+    prefix = Config.NTFY_TOPIC_PREFIX
+    return {
+        "enabled": True,
+        "server": server,
+        "prefix": prefix,
+        "topics": {
+            "warning": f"{prefix}-warning",
+            "danger": f"{prefix}-danger",
+            "p1_outlook": f"{prefix}-p1-outlook",
+            "status": f"{prefix}-status",
+            "station_pattern": f"{prefix}-<station>-warning | {prefix}-<station>-danger (station code lowercase, no dot: p1, p103)",
+        },
+        "semantics": "transitions only: one message on crossing up, one all-clear on the way down (0.10 m hysteresis)",
+    }
+
+
 @app.get("/api/forecast/skill")
 async def get_forecast_skill(
     response: Response,

tests/test_notify.py

@@ -0,0 +1,285 @@
+"""ntfy notification state machine: transitions only, hysteresis, restart-safe."""
+
+import datetime
+
+import pytest
+
+from src import notify
+
+
+class FakePublisher(notify.NtfyPublisher):
+    def __init__(self):
+        super().__init__("http://ntfy.test", prefix="ping")
+        self.sent = []
+
+    def publish(self, n):
+        self.sent.append(n)
+        return True
+
+
+@pytest.fixture
+def pub():
+    return FakePublisher()
+
+
+def _reading(code, level, ts="2026-09-24T12:00:00"):
+    return {"station_code": code, "water_level": level, "timestamp": ts}
+
+
+def _fc(p, peak=None):
+    return [
+        {
+            "station_code": "P.1",
+            "horizon_hours": 24,
+            "p_warning": p,
+            "predicted_max_level": peak,
+            "source": "model",
+        }
+    ]
+
+
+NOW = datetime.datetime(2026, 9, 24, 12, 30)
+
+
+def topics(pub):
+    return [n.topic for n in pub.sent]
+
+
+def test_quiet_river_sends_nothing(pub):
+    state = notify.InMemoryState()
+    for h in range(48):
+        notify.evaluate(
+            [_reading("P.1", 1.6), _reading("P.103", 3.2)],
+            _fc(0.01),
+            state,
+            pub,
+            now=NOW,
+        )
+    assert pub.sent == []
+
+
+def test_warning_crossing_once_then_silence_then_clear(pub):
+    state = notify.InMemoryState()
+    # rising through 3.70 (P.1 warning)
+    notify.evaluate([_reading("P.1", 3.65)], [], state, pub, now=NOW)
+    assert pub.sent == []
+    notify.evaluate([_reading("P.1", 3.72)], [], state, pub, now=NOW)
+    assert topics(pub) == ["ping-p1-warning", "ping-warning"]
+    assert pub.sent[0].priority == 4 and "3.72 m" in pub.sent[0].message
+    # stays above: no repeats for many hours
+    for level in (3.80, 3.95, 4.05, 3.90, 3.75):
+        notify.evaluate([_reading("P.1", level)], [], state, pub, now=NOW)
+    assert len(pub.sent) == 2
+    # dips to 3.65: within hysteresis, still no message
+    notify.evaluate([_reading("P.1", 3.65)], [], state, pub, now=NOW)
+    assert len(pub.sent) == 2
+    # 3.55: clear
+    notify.evaluate([_reading("P.1", 3.55)], [], state, pub, now=NOW)
+    assert topics(pub)[2:] == ["ping-p1-warning", "ping-warning"]
+    assert "back to normal" in pub.sent[2].title
+
+
+def test_danger_escalation_and_deescalation(pub):
+    state = notify.InMemoryState()
+    notify.evaluate([_reading("P.1", 3.9)], [], state, pub, now=NOW)  # warning
+    notify.evaluate(
+        [_reading("P.1", 4.25)], [], state, pub, now=NOW
+    )  # danger (>= 4.20)
+    assert topics(pub) == [
+        "ping-p1-warning",
+        "ping-warning",
+        "ping-p1-danger",
+        "ping-danger",
+    ]
+    assert pub.sent[2].priority == 5
+    notify.evaluate(
+        [_reading("P.1", 4.15)], [], state, pub, now=NOW
+    )  # hysteresis: still danger
+    assert len(pub.sent) == 4
+    notify.evaluate([_reading("P.1", 4.05)], [], state, pub, now=NOW)  # back to warning
+    assert topics(pub)[4:] == ["ping-p1-danger", "ping-warning"]
+    assert "below danger" in pub.sent[4].title
+
+
+def test_jump_straight_to_danger(pub):
+    state = notify.InMemoryState()
+    notify.evaluate(
+        [_reading("P.103", 7.0)], [], state, pub, now=NOW
+    )  # P.103 danger 6.75
+    assert topics(pub) == ["ping-p103-danger", "ping-danger"]
+
+
+def test_basin_digest_groups_stations(pub):
+    state = notify.InMemoryState()
+    notify.evaluate(
+        [_reading("P.1", 3.8), _reading("P.103", 6.0), _reading("P.67", 1.0)],
+        [],
+        state,
+        pub,
+        now=NOW,
+    )
+    basin = [n for n in pub.sent if n.topic == "ping-warning"]
+    assert len(basin) == 1 and "P.1" in basin[0].message and "P.103" in basin[0].message
+
+
+def test_outlook_on_off_with_hysteresis(pub):
+    state = notify.InMemoryState()
+    r = [_reading("P.1", 2.9)]
+    notify.evaluate(r, _fc(0.30), state, pub, now=NOW)
+    assert pub.sent == []
+    notify.evaluate(r, _fc(0.55, 3.9), state, pub, now=NOW)
+    assert topics(pub) == ["ping-p1-outlook"]
+    assert "55%" in pub.sent[0].message and "3.90 m" in pub.sent[0].message
+    assert "not an official warning" in pub.sent[0].message
+    notify.evaluate(
+        r, _fc(0.40), state, pub, now=NOW
+    )  # between OFF and ON: stays on, silent
+    assert len(pub.sent) == 1
+    notify.evaluate(r, _fc(0.20), state, pub, now=NOW)
+    assert len(pub.sent) == 2 and "easing" in pub.sent[1].title
+
+
+def test_heuristic_forecast_ignored(pub):
+    state = notify.InMemoryState()
+    fc = [
+        {
+            "station_code": "P.1",
+            "horizon_hours": 24,
+            "p_warning": 0.9,
+            "source": "heuristic",
+        }
+    ]
+    notify.evaluate([_reading("P.1", 2.0)], fc, state, pub, now=NOW)
+    assert pub.sent == []
+
+
+def test_stale_feed_and_recovery(pub):
+    state = notify.InMemoryState()
+    notify.evaluate(
+        [_reading("P.1", 1.6, "2026-09-24T12:00:00")], [], state, pub, now=NOW
+    )
+    assert pub.sent == []
+    later = NOW + datetime.timedelta(hours=4)
+    notify.evaluate(
+        [_reading("P.1", 1.6, "2026-09-24T12:00:00")], [], state, pub, now=later
+    )
+    assert topics(pub) == ["ping-status"] and "stale" in pub.sent[0].title
+    notify.evaluate(
+        [_reading("P.1", 1.6, "2026-09-24T12:00:00")],
+        [],
+        state,
+        pub,
+        now=later + datetime.timedelta(hours=1),
+    )
+    assert len(pub.sent) == 1  # still stale, no repeat
+    notify.evaluate(
+        [_reading("P.1", 1.6, "2026-09-24T17:00:00")],
+        [],
+        state,
+        pub,
+        now=later + datetime.timedelta(hours=1),
+    )
+    assert len(pub.sent) == 2 and "recovered" in pub.sent[1].title
+
+
+def test_capacity_guard_blocks_stale_threshold(pub):
+    """P.77 2026-09: 3.02 m >= 2.85 m 'warning' at 22 % capacity -> not a flood."""
+    state = notify.InMemoryState()
+    r = {
+        "station_code": "P.77",
+        "water_level": 4.40,
+        "timestamp": "2026-09-24T12:00:00",
+        "discharge_percent": 10.3,
+    }
+    notify.evaluate([r], [], state, pub, now=NOW)
+    assert pub.sent == [] and state.get("level:P.77") is None
+    # same level with capacity agreeing -> alert
+    r["discharge_percent"] = 82.0
+    notify.evaluate([r], [], state, pub, now=NOW)
+    assert topics(pub) == ["ping-p77-warning", "ping-warning"]
+
+
+def test_capacity_guard_exempts_p1_and_missing_pct(pub):
+    state = notify.InMemoryState()
+    notify.evaluate(
+        [
+            {
+                "station_code": "P.1",
+                "water_level": 3.75,
+                "timestamp": "2026-09-24T12:00:00",
+                "discharge_percent": 40.0,
+            }
+        ],
+        [],
+        state,
+        pub,
+        now=NOW,
+    )
+    assert topics(pub) == ["ping-p1-warning", "ping-warning"]
+    pub.sent.clear()
+    notify.evaluate(
+        [
+            {
+                "station_code": "P.103",
+                "water_level": 6.0,
+                "timestamp": "2026-09-24T12:00:00",
+            }
+        ],
+        [],
+        state,
+        pub,
+        now=NOW,
+    )
+    assert topics(pub) == ["ping-p103-warning", "ping-warning"]
+
+
+def test_capacity_guard_does_not_block_clearing(pub):
+    """Guard applies only to the clear->alert edge; the all-clear always goes out."""
+    state = notify.InMemoryState()
+    r = {
+        "station_code": "P.67",
+        "water_level": 2.6,
+        "timestamp": "2026-09-24T12:00:00",
+        "discharge_percent": 90.0,
+    }
+    notify.evaluate([r], [], state, pub, now=NOW)
+    assert len(pub.sent) == 2
+    r.update(water_level=2.2, discharge_percent=30.0)
+    notify.evaluate([r], [], state, pub, now=NOW)
+    assert "back to normal" in pub.sent[2].title
+
+
+def test_state_survives_restart_via_sql(tmp_path, pub):
+    from sqlalchemy import create_engine
+
+    eng = create_engine(f"sqlite:///{tmp_path / 'n.db'}")
+    state = notify.NotificationState(eng, "sqlite")
+    notify.evaluate([_reading("P.1", 3.8)], [], state, pub, now=NOW)
+    assert len(pub.sent) == 2
+    # "restart": new state object on the same DB, same reading -> nothing re-sent
+    state2 = notify.NotificationState(eng, "sqlite")
+    notify.evaluate([_reading("P.1", 3.8)], [], state2, pub, now=NOW)
+    assert len(pub.sent) == 2
+
+
+def test_publish_failure_does_not_advance_state():
+    """If ntfy is down the transition must be retried next cycle, not lost."""
+
+    class Down(notify.NtfyPublisher):
+        def __init__(self):
+            super().__init__("http://ntfy.test")
+            self.calls = 0
+
+        def publish(self, n):
+            self.calls += 1
+            return False
+
+    pub = Down()
+    state = notify.InMemoryState()
+    notify.evaluate([_reading("P.1", 3.8)], [], state, pub, now=NOW)
+    assert pub.calls == 2 and state.get("level:P.1") is None
+    # next cycle, ntfy back: the crossing is delivered
+    good = FakePublisher()
+    notify.evaluate([_reading("P.1", 3.8)], [], state, good, now=NOW)
+    assert topics(good) == ["ping-p1-warning", "ping-warning"]
+    assert state.get("level:P.1") == "warning"