Merge pull request 'Update to v 0.4.0' (#6) from dev into main

Reviewed-on: #6

CHANGELOG.md

@@ -5,6 +5,52 @@ All notable changes to SLMM (Sound Level Meter Manager) will be documented in th
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [0.4.0] - 2026-06-22
+
+### Added
+
+#### Live Monitor (fan-out feed)
+- **Per-device fan-out monitor** - one shared, cached live feed per device. Multiple clients (dashboards, portal, charts) subscribe to the same stream instead of each fighting for the NL-43's single TCP connection: one poller reads the device, all subscribers get the same frames.
+- **WebSocket monitor** - `WS /api/nl43/{unit_id}/monitor` delivers an instant first frame from cache, then live updates.
+- **Monitor control** - `POST /api/nl43/{unit_id}/monitor/{start|stop}`, `GET /api/nl43/_monitor/status`. A persistent `monitor_enabled` flag auto-starts the keepalive on boot.
+- **Adaptive polling** - poll rate adapts to demand; unreachable devices back off; a device-offline alert fires when a monitored unit drops.
+- **De-duplication** - the background poller skips units already covered by an active monitor (no double-polling); a heartbeat keeps the feed warm.
+- **Lower latency** - the monitor caches run state, roughly halving live-feed latency; fan-out emits an instant first frame + offline status to new clients.
+
+#### Alert Engine
+- **Threshold rules** - per-device alert rules (metric + threshold + cooldown) with full CRUD: `POST/GET/PUT/DELETE /api/nl43/{unit_id}/alerts/rules[/{rule_id}]`.
+- **Events + state machine** - onset/clear tracking via `GET /api/nl43/{unit_id}/alerts/events`; acknowledge with `POST .../events/{event_id}/ack`. A `cooldown_s` is enforced between onsets.
+- **24/7 evaluation** - enabled rules pin the monitor on, so rules evaluate continuously even with no UI client connected.
+- **Resilience** - editing or deleting a rule resets its state and closes any open event; device-offline events are raised when a monitored unit goes unreachable.
+
+#### Data & History
+- **Live-chart backfill** - a downsampled DOD trail is persisted to a new `nl43_readings` table, exposed via `GET /api/nl43/{unit_id}/history` so charts can backfill recent history on load.
+- **LN1/LN2 percentiles** - L1/L10 (configurable percentiles) surfaced through SLMM in the status and live-feed payloads.
+- **measurement_start_time** included in the cached `/status` response.
+
+#### Device control
+- **Per-device disconnect** - `POST /api/nl43/{unit_id}/disconnect` drops a device's pooled connection.
+- **Deactivate / standby** - `POST /api/nl43/{unit_id}/deactivate` and global `POST /api/nl43/_system/standby` to quiesce polling/monitoring.
+
+### Changed
+- **DRD streaming reuses the pooled connection** rather than opening a separate socket, avoiding contention with the persistent pool on a single-connection device.
+- **Connection pool** - idle-TTL / max-age checks can now be disabled; pool status is logged periodically.
+
+### Fixed
+- **Measurement-start confirmation** - `/start` now recognizes the device's `Start` state. It previously waited for `Measure`, which never matched, so the start cycle ran the full retry loop and Terra-View's proxy timed out with a misleading "Unknown error" even though the device had started.
+- **Garbled reads** - corrupted measurement-state reads that produced phantom STOPPED/STARTED transitions are now ignored.
+- **DOD parsing** - corrected field parsing and stopped spurious measurement-time resets.
+- **Monitor WebSocket** - quieted a send-after-close race on client disconnect.
+
+### Database
+- **New tables** (auto-created on startup via `Base.metadata.create_all`): `alert_rules`, `alert_events`, `nl43_readings`.
+- **Migrations for existing tables** (run once per database): `migrate_add_ln_percentiles.py` (LN1/LN2 on `nl43_status`), `migrate_add_monitor_enabled.py` (`monitor_enabled` on `nl43_config`).
+
+### Notes
+- Pairs with the matching Terra-View `dev` build, which reads SLMM's `/monitor` fan-out feed for live SLM dashboards (L1/L10 lines, live-chart backfill). Ship the two together.
+
+---
+
 ## [0.3.0] - 2026-02-17
 
 ### Added

README.md

@@ -1,6 +1,6 @@
 # SLMM - Sound Level Meter Manager
 
-**Version 0.3.0**
+**Version 0.4.0**
 
 Backend API service for controlling and monitoring Rion NL-43/NL-53 Sound Level Meters via TCP and FTP protocols.
 
@@ -12,6 +12,9 @@ SLMM is a standalone backend module that provides REST API routing and command t
 
 ## Features
 
+- **Live Monitor (fan-out)**: One shared cached live feed per device — many clients subscribe to the same stream instead of fighting over the meter's single TCP connection
+- **Alert Engine**: Per-device threshold rules with onset/clear events, cooldowns, acks, and 24/7 evaluation
+- **History & Percentiles**: Downsampled DOD trail + history endpoint for live-chart backfill; LN1/LN2 (L1/L10) percentiles surfaced through the feed
 - **Persistent TCP Connections**: Cached per-device connections with OS-level keepalive, tuned for cellular modem reliability
 - **Background Polling**: Continuous automatic polling of devices with configurable intervals
 - **Offline Detection**: Automatic device reachability tracking with failure counters
@@ -44,6 +47,30 @@ SLMM is a standalone backend module that provides REST API routing and command t
                                   └──────────────┘
 ```
 
+### Live Monitor — Fan-Out Feed (v0.4.0)
+
+The NL-43 allows only one TCP control connection at a time, so multiple clients
+polling the same device directly would contend for it. The monitor solves this
+with a single shared, cached feed per device:
+
+- **One reader, many subscribers**: a single poller reads the device; every
+  WebSocket subscriber (`WS /api/nl43/{unit_id}/monitor`) receives the same
+  frames — an instant first frame from cache, then live updates.
+- **Persistent + auto-start**: a `monitor_enabled` flag keeps the feed running
+  and auto-starts it on boot. Enabled alert rules pin the monitor on for 24/7
+  evaluation even with no UI connected.
+- **Adaptive & deduplicated**: poll rate adapts to demand, unreachable devices
+  back off, and the background poller skips units already covered by a monitor.
+
+### Alert Engine (v0.4.0)
+
+Per-device threshold alerting evaluated against the live feed:
+
+- **Rules**: metric + threshold + `cooldown_s`, full CRUD per device
+- **Events**: onset/clear state machine, acknowledgement, and a device-offline
+  alert when a monitored unit drops
+- **Robust**: editing/deleting a rule resets its state and closes open events
+
 ### Persistent TCP Connection Pool (v0.3.0)
 
 SLMM maintains persistent TCP connections to devices with OS-level keepalive, designed for reliable operation over cellular modems:
@@ -145,8 +172,32 @@ Logs are written to:
 |--------|----------|-------------|
 | GET | `/api/nl43/{unit_id}/status` | Get cached measurement snapshot (updated by background poller) |
 | GET | `/api/nl43/{unit_id}/live` | Request fresh DOD data from device (bypasses cache) |
+| GET | `/api/nl43/{unit_id}/history` | Downsampled DOD trail for live-chart backfill |
 | WS | `/api/nl43/{unit_id}/stream` | WebSocket stream for real-time DRD data |
 
+### Live Monitor (fan-out feed)
+
+| Method | Endpoint | Description |
+|--------|----------|-------------|
+| WS | `/api/nl43/{unit_id}/monitor` | Subscribe to the shared cached live feed (instant first frame) |
+| POST | `/api/nl43/{unit_id}/monitor/start` | Start the device's monitor feed |
+| POST | `/api/nl43/{unit_id}/monitor/stop` | Stop the device's monitor feed |
+| GET | `/api/nl43/_monitor/status` | Global monitor status across devices |
+| POST | `/api/nl43/{unit_id}/disconnect` | Drop the device's pooled TCP connection |
+| POST | `/api/nl43/{unit_id}/deactivate` | Quiesce polling/monitoring for one device |
+| POST | `/api/nl43/_system/standby` | Global standby — quiesce all polling/monitoring |
+
+### Alerts
+
+| Method | Endpoint | Description |
+|--------|----------|-------------|
+| GET | `/api/nl43/{unit_id}/alerts/rules` | List alert rules for a device |
+| POST | `/api/nl43/{unit_id}/alerts/rules` | Create an alert rule (metric, threshold, cooldown) |
+| PUT | `/api/nl43/{unit_id}/alerts/rules/{rule_id}` | Update a rule (resets its state, closes open events) |
+| DELETE | `/api/nl43/{unit_id}/alerts/rules/{rule_id}` | Delete a rule |
+| GET | `/api/nl43/{unit_id}/alerts/events` | List alert events (onset/clear) |
+| POST | `/api/nl43/{unit_id}/alerts/events/{event_id}/ack` | Acknowledge an event |
+
 ### Background Polling
 
 | Method | Endpoint | Description |
@@ -273,11 +324,35 @@ Caches latest measurement snapshot:
 - `sd_remaining_mb`: Free SD card space (MB)
 - `sd_free_ratio`: SD card free space ratio
 - `raw_payload`: Raw device response data
-- `is_reachable`: Device reachability status (Boolean) ⭐ NEW
+- `is_reachable`: Device reachability status (Boolean)
-- `consecutive_failures`: Count of consecutive poll failures ⭐ NEW
+- `consecutive_failures`: Count of consecutive poll failures
-- `last_poll_attempt`: Last time background poller attempted to poll ⭐ NEW
+- `last_poll_attempt`: Last time background poller attempted to poll
-- `last_success`: Last successful poll timestamp ⭐ NEW
+- `last_success`: Last successful poll timestamp
-- `last_error`: Last error message (truncated to 500 chars) ⭐ NEW
+- `last_error`: Last error message (truncated to 500 chars)
+- `ln1` / `ln2`: LN1/LN2 (L1/L10) percentile levels ⭐ v0.4.0
+
+### NL43Readings Table ⭐ v0.4.0
+Downsampled DOD trail backing the live-chart history endpoint (one row/minute,
+pruned to a retention window — viewing only, not the report source):
+- `id` (PK), `unit_id`, `timestamp`
+- `lp` / `leq` / `lmax` / `ln1` / `ln2`: cached level samples
+
+### AlertRule Table ⭐ v0.4.0
+Per-device threshold alert rules:
+- `id` (PK), `unit_id`, `name`, `enabled`
+- `metric`, `comparison` (above/below), `threshold_db`, `clear_margin_db` (hysteresis)
+- `duration_s` (sustained), `cooldown_s` (min seconds between onsets)
+- `channels` / `recipients`, optional `schedule_start`/`schedule_end`/`schedule_days`
+
+### AlertEvent Table ⭐ v0.4.0
+Alert onset/clear events for history, inbox, and acknowledgement:
+- `id` (PK), `unit_id`, `rule_id`, `rule_name`, `metric`, `threshold_db`
+- `onset_at` / `onset_value`, `peak_value`, `clear_at`, `status` (active/cleared)
+- `acknowledged_at` / `acknowledged_by`, `notes`
+
+> New tables (`alert_rules`, `alert_events`, `nl43_readings`) auto-create on
+> startup. Existing-table columns ship with migrations:
+> `migrate_add_ln_percentiles.py`, `migrate_add_monitor_enabled.py`.
 
 ## Protocol Details
 

app/alerts.py

@@ -40,6 +40,7 @@ class RuleState:
     edge_since: Optional[float] = None  # when the current edge condition began (clock time)
     peak: float = 0.0
     event_id: Optional[int] = None      # the open AlertEvent row (for the clear update)
+    last_onset: Optional[float] = None  # time of the last onset (for cooldown)
 
 
 def _exceeds(value: float, rule) -> bool:
@@ -68,9 +69,17 @@ def _evaluate_step(state: RuleState, value: float, now: float, rule) -> Optional
             if state.edge_since is None:
                 state.edge_since = now
             if now - state.edge_since >= duration:
+                # Cooldown: suppress a new onset within cooldown_s of the last one
+                # (stops a repeatedly-breaching signal from flooding the history).
+                # Hold edge_since so it fires the moment cooldown lapses if still
+                # breaching — don't reset it here.
+                cooldown = getattr(rule, "cooldown_s", 0) or 0
+                if state.last_onset is not None and (now - state.last_onset) < cooldown:
+                    return None
                 state.phase = "active"
                 state.edge_since = None
                 state.peak = value
+                state.last_onset = now
                 return "onset"
         else:
             state.edge_since = None
@@ -112,6 +121,7 @@ class AlertEvaluator:
     def __init__(self):
         self._states: Dict[Tuple[str, int], RuleState] = {}
         self._rule_cache: Dict[str, Tuple[float, list]] = {}  # unit_id -> (fetched_at, rules)
+        self._offline_events: Dict[str, int] = {}  # unit_id -> open connectivity AlertEvent id
         logger.info("[ALERT] rule-based evaluator ready")
 
     async def evaluate(self, unit_id: str, snap) -> None:
@@ -165,6 +175,12 @@ class AlertEvaluator:
         else:
             self._rule_cache.pop(unit_id, None)
 
+    def forget_rule(self, unit_id: str, rule_id: int) -> None:
+        """Drop a rule's per-(unit, rule) state machine after the rule is edited or
+        deleted, so a stale 'active' phase / open event_id from the old config
+        doesn't bleed into the new one (mis-firing a clear or suppressing an onset)."""
+        self._states.pop((unit_id, rule_id), None)
+
     # -- scheduling ----------------------------------------------------------
 
     def _in_schedule(self, rule) -> bool:
@@ -235,9 +251,71 @@ class AlertEvaluator:
             f"recovered to {value:.1f} dB (peak {peak:.1f} dB)",
         )
 
+    # -- connectivity (device offline/online) -------------------------------
+    #
+    # Raised by the live monitor when it loses / regains contact with a device.
+    # Persisted as an AlertEvent (sentinel rule_id=0, metric="connectivity") so it
+    # lands in the same events/inbox/ack pipeline as threshold alerts. The in-memory
+    # map dedupes; the DB query also dedupes across a process restart.
+
+    async def device_offline(self, unit_id: str) -> None:
+        if unit_id in self._offline_events:
+            return  # already flagged offline
+        from app.database import SessionLocal
+        from app.models import AlertEvent
+        db = SessionLocal()
+        try:
+            existing = db.query(AlertEvent).filter_by(
+                unit_id=unit_id, metric="connectivity", status="active").first()
+            if existing:  # already open in the DB (e.g. carried across a restart)
+                self._offline_events[unit_id] = existing.id
+                return
+            evt = AlertEvent(
+                rule_id=0, unit_id=unit_id, rule_name="Device unreachable",
+                metric="connectivity", threshold_db=0.0, status="active",
+            )
+            db.add(evt)
+            db.commit()
+            db.refresh(evt)
+            self._offline_events[unit_id] = evt.id
+        except Exception as e:
+            logger.warning(f"[ALERT] failed to record offline for {unit_id}: {e}")
+        finally:
+            db.close()
+        await self._dispatch_raw("OFFLINE", unit_id, "Device unreachable",
+                                 "live monitor lost contact with the device")
+
+    async def device_online(self, unit_id: str) -> None:
+        self._offline_events.pop(unit_id, None)
+        from app.database import SessionLocal
+        from app.models import AlertEvent
+        db = SessionLocal()
+        cleared = 0
+        try:
+            opened = db.query(AlertEvent).filter_by(
+                unit_id=unit_id, metric="connectivity", status="active").all()
+            for evt in opened:
+                evt.clear_at = datetime.utcnow()
+                evt.status = "cleared"
+                cleared += 1
+            if cleared:
+                db.commit()
+        except Exception as e:
+            logger.warning(f"[ALERT] failed to record online for {unit_id}: {e}")
+        finally:
+            db.close()
+        if cleared:  # only announce recovery if it was actually flagged offline
+            await self._dispatch_raw("ONLINE", unit_id, "Device recovered",
+                                     "live monitor regained contact with the device")
+
+    # -- event persistence + dispatch ---------------------------------------
+
     async def _dispatch(self, kind: str, unit_id: str, rule, detail: str) -> None:
+        await self._dispatch_raw(kind, unit_id, rule.name, detail)
+
+    async def _dispatch_raw(self, kind: str, unit_id: str, name: str, detail: str) -> None:
         """POC dispatch: server log. Swap in a Terra-View webhook (email/SMS) here."""
-        logger.warning(f"[ALERT:{kind}] {unit_id} '{rule.name}': {detail}")
+        logger.warning(f"[ALERT:{kind}] {unit_id} '{name}': {detail}")
 
 
 # Module-level singleton (the monitor calls alert_evaluator.evaluate per snapshot)

app/background_poller.py

@@ -178,10 +178,19 @@ class BackgroundPoller:
             now = datetime.utcnow()
             polled_count = 0
 
+            from app.monitor import monitor_manager
+
             for cfg in configs:
                 if not self._running:
                     break
 
+                # Skip units with an active live monitor: it polls them at ~1Hz and
+                # keeps the status cache fresh, so a redundant background poll would just
+                # add load/lock-contention on the device's single connection.
+                if monitor_manager.is_active(cfg.unit_id):
+                    self._logger.debug(f"Skipping {cfg.unit_id} — live monitor active")
+                    continue
+
                 # Get current status
                 status = db.query(NL43Status).filter_by(unit_id=cfg.unit_id).first()
 

app/main.py

@@ -38,6 +38,25 @@ async def lifespan(app: FastAPI):
     await poller.start()
     logger.info("Background poller started")
 
+    # Auto-start keepalive live monitors for units configured for 24/7 monitoring
+    # (monitor_enabled). This is what keeps alerting running unattended across
+    # restarts — without it a feed only runs while someone has the live view open.
+    try:
+        from app.monitor import monitor_manager
+        from app.database import SessionLocal
+        from app.models import NL43Config
+        db = SessionLocal()
+        try:
+            units = db.query(NL43Config).filter_by(monitor_enabled=True, tcp_enabled=True).all()
+            for cfg in units:
+                m = await monitor_manager.get(cfg.unit_id)
+                await m.set_keepalive(True)
+                logger.info(f"Auto-started keepalive monitor for {cfg.unit_id}")
+        finally:
+            db.close()
+    except Exception as e:
+        logger.error(f"Failed to auto-start monitors: {e}")
+
     yield  # Application runs
 
     # Shutdown
@@ -52,7 +71,7 @@ async def lifespan(app: FastAPI):
 app = FastAPI(
     title="SLMM NL43 Addon",
     description="Standalone module for NL43 configuration and status APIs with background polling",
-    version="0.3.0",
+    version="0.4.0",
     lifespan=lifespan,
 )
 

app/models.py

@@ -23,6 +23,10 @@ class NL43Config(Base):
     poll_interval_seconds = Column(Integer, nullable=True, default=60)  # Polling interval (10-3600 seconds)
     poll_enabled = Column(Boolean, default=True)  # Enable/disable background polling for this device
 
+    # Live monitor (fan-out DOD feed). Keepalive runs it 24/7 even with no viewer,
+    # which is what makes alerting continuous. On by default; toggleable from the UI.
+    monitor_enabled = Column(Boolean, default=True)
+
 
 class NL43Status(Base):
     """
@@ -124,3 +128,24 @@ class AlertEvent(Base):
     acknowledged_at = Column(DateTime, nullable=True)
     acknowledged_by = Column(String, nullable=True)
     notes = Column(Text, nullable=True)
+
+
+class NL43Reading(Base):
+    """Downsampled time-series of live-monitor readings, for the live-chart
+    backfill (so a viewer sees recent trend on open, not a blank chart).
+
+    Viewing only — NOT the report source. Reports use the device's authoritative
+    FTP .rnd intervals. This is a short, capped trail (one row/minute, pruned to
+    a retention window) fed by the monitor's keepalive poll loop.
+    """
+
+    __tablename__ = "nl43_readings"
+
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    unit_id = Column(String, index=True, nullable=False)
+    timestamp = Column(DateTime, default=func.now(), index=True)
+    lp = Column(String, nullable=True)
+    leq = Column(String, nullable=True)
+    lmax = Column(String, nullable=True)
+    ln1 = Column(String, nullable=True)
+    ln2 = Column(String, nullable=True)

app/monitor.py

@@ -27,9 +27,41 @@ from app.alerts import alert_evaluator
 
 logger = logging.getLogger(__name__)
 
-# Sleep between DOD polls. Note the 1s device rate-limit (and DOD?+Measure? per
+# Extra idle between DOD polls WHEN A BROWSER IS WATCHING. The 1s device rate-limit
-# poll) already paces the effective rate to a few seconds; this is the extra idle.
+# already paces consecutive DOD? commands, so this just needs to be small — the
-MONITOR_POLL_INTERVAL = float(os.getenv("MONITOR_POLL_INTERVAL", "1.0"))
+# rate-limit is the real floor (~1.25s/poll effective).
+MONITOR_POLL_INTERVAL = float(os.getenv("MONITOR_POLL_INTERVAL", "0.25"))
+
+# Idle cadence when NO browser is subscribed and the feed is only kept alive for
+# alerting. Same data, ~8x fewer polls -> ~8x less cellular traffic on a metered
+# SIM (~1 GB/device/month at full rate -> ~125 MB). NOTE: this also sets the alert
+# sampling resolution when nobody is watching, so keep it <= the smallest alert
+# duration_s you rely on (default 10s comfortably catches a "sustained 30/60s" rule).
+MONITOR_IDLE_POLL_INTERVAL = float(os.getenv("MONITOR_IDLE_POLL_INTERVAL", "10"))
+
+# Exponential backoff once the device is unreachable, so a powered-off / asleep /
+# out-of-signal device stops churning reconnects every cycle (log spam + a trickle
+# of wasted cellular data on failed SYNs). delay = min(BASE * 2**(fails-1), MAX),
+# reset to full-rate on the first good poll. While a browser is actively watching we
+# cap the backoff lower (WATCHED_MAX) so a recovery surfaces quickly for the viewer.
+MONITOR_BACKOFF_BASE_S = float(os.getenv("MONITOR_BACKOFF_BASE_S", "1"))
+MONITOR_BACKOFF_MAX_S = float(os.getenv("MONITOR_BACKOFF_MAX_S", "60"))
+MONITOR_BACKOFF_WATCHED_MAX_S = float(os.getenv("MONITOR_BACKOFF_WATCHED_MAX_S", "5"))
+
+# How often to refresh the run state (Measure?). It changes rarely, so we cache it
+# and skip that second rate-limited command on most polls — roughly halving the
+# per-update latency (~2.5s -> ~1.3s).
+MONITOR_STATE_REFRESH_S = float(os.getenv("MONITOR_STATE_REFRESH_S", "30"))
+
+# Downsampled trail for the live-chart backfill: store one reading per
+# TRAIL_SAMPLE_S and keep TRAIL_RETENTION_HOURS of it (pruned). Viewing only —
+# reports use the device's FTP .rnd data, not this.
+TRAIL_SAMPLE_S = float(os.getenv("MONITOR_TRAIL_SAMPLE_S", "60"))
+TRAIL_RETENTION_HOURS = float(os.getenv("MONITOR_TRAIL_RETENTION_HOURS", "24"))
+
+# If nothing has been broadcast in this many seconds (e.g. device offline and
+# silent), send a keepalive frame so reverse proxies don't drop the idle WS.
+MONITOR_HEARTBEAT_S = float(os.getenv("MONITOR_HEARTBEAT_S", "25"))
 
 
 def _snapshot_payload(snap, unit_id: str, measurement_start_time) -> dict:
@@ -63,6 +95,12 @@ class DeviceMonitor:
         self._keepalive = False
         self._task: Optional[asyncio.Task] = None
         self._lock = asyncio.Lock()
+        self._last_payload: Optional[dict] = None  # replayed to new subscribers
+        self._consec_fail = 0
+        self._reachable = True  # last broadcast reachability (for transition frames)
+        self._cached_state: Optional[str] = None  # run state, refreshed periodically
+        self._last_state_refresh = 0.0
+        self._last_trail_store = 0.0  # downsample throttle for the backfill trail
 
     @property
     def running(self) -> bool:
@@ -82,6 +120,13 @@ class DeviceMonitor:
         q: asyncio.Queue = asyncio.Queue(maxsize=5)
         async with self._lock:
             self._subscribers.add(q)
+            # Replay the last frame so a client connecting mid-stream sees data
+            # (or the current 'unreachable' state) immediately, not after a poll.
+            if self._last_payload is not None:
+                try:
+                    q.put_nowait(self._last_payload)
+                except asyncio.QueueFull:
+                    pass
             self._ensure_task()
         return q
 
@@ -97,20 +142,75 @@ class DeviceMonitor:
 
     async def _run(self) -> None:
         logger.info(f"[MONITOR] {self.unit_id}: feed started")
+        loop = asyncio.get_running_loop()
+        last_send = loop.time()
         try:
             while self._has_demand():
                 snap, mst = await self._poll_once()
                 if snap is not None:
+                    if not self._reachable:
+                        # Recovered from an outage — clear the connectivity alert.
+                        try:
+                            await alert_evaluator.device_online(self.unit_id)
+                        except Exception as e:
+                            logger.warning(f"[MONITOR] {self.unit_id}: online alert failed: {e}")
+                    self._consec_fail = 0
+                    self._reachable = True
                     payload = _snapshot_payload(snap, self.unit_id, mst)
+                    payload["feed_status"] = "ok"
                     self._broadcast(payload)
+                    last_send = loop.time()
                     try:
                         await alert_evaluator.evaluate(self.unit_id, snap)
                     except Exception as e:
                         logger.warning(f"[MONITOR] {self.unit_id}: alert eval failed: {e}")
-                await asyncio.sleep(MONITOR_POLL_INTERVAL)
+                else:
+                    # Tell clients the device went offline — once, on transition, after a
+                    # few failures so a momentary blip doesn't flap the UI. Same edge
+                    # raises the device-offline alert.
+                    self._consec_fail += 1
+                    if self._reachable and self._consec_fail >= 3:
+                        self._reachable = False
+                        self._broadcast({
+                            "unit_id": self.unit_id,
+                            "timestamp": datetime.utcnow().isoformat(),
+                            "feed_status": "unreachable",
+                        })
+                        last_send = loop.time()
+                        try:
+                            await alert_evaluator.device_offline(self.unit_id)
+                        except Exception as e:
+                            logger.warning(f"[MONITOR] {self.unit_id}: offline alert failed: {e}")
+
+                # Heartbeat: during quiet/offline stretches, send a keepalive so an
+                # idle WS isn't dropped by a reverse proxy. Not cached (new subscribers
+                # should still get the last real frame, not a heartbeat).
+                if loop.time() - last_send >= MONITOR_HEARTBEAT_S:
+                    self._broadcast({
+                        "unit_id": self.unit_id,
+                        "timestamp": datetime.utcnow().isoformat(),
+                        "feed_status": "ok" if self._reachable else "unreachable",
+                        "heartbeat": True,
+                    }, cache=False)
+                    last_send = loop.time()
+
+                await asyncio.sleep(self._next_delay())
         finally:
             logger.info(f"[MONITOR] {self.unit_id}: feed stopped")
 
+    def _next_delay(self) -> float:
+        """Inter-poll delay: exponential backoff while unreachable, full-rate while a
+        browser is watching, relaxed cadence when the feed is keepalive-only."""
+        if self._consec_fail > 0:
+            shift = min(self._consec_fail - 1, 6)  # cap growth at 2**6 = 64x base
+            delay = min(MONITOR_BACKOFF_BASE_S * (2 ** shift), MONITOR_BACKOFF_MAX_S)
+            if self._subscribers:
+                delay = min(delay, MONITOR_BACKOFF_WATCHED_MAX_S)
+            return delay
+        if self._subscribers:
+            return MONITOR_POLL_INTERVAL       # a browser is watching — smooth chart
+        return MONITOR_IDLE_POLL_INTERVAL      # keepalive-only (alerting) — save data
+
     async def _poll_once(self):
         """One DOD poll: read, persist, return (snapshot, measurement_start_iso)."""
         db = SessionLocal()
@@ -123,10 +223,25 @@ class DeviceMonitor:
                 ftp_username=cfg.ftp_username, ftp_password=cfg.ftp_password,
                 ftp_port=cfg.ftp_port or 21,
             )
-            snap = await client.request_dod()
+            # Refresh the run state only every MONITOR_STATE_REFRESH_S; reuse the
+            # cached state otherwise so most polls send just DOD? (one rate-limited
+            # command) instead of DOD? + Measure?.
+            now = asyncio.get_running_loop().time()
+            refresh_state = (self._cached_state is None
+                             or now - self._last_state_refresh >= MONITOR_STATE_REFRESH_S)
+            snap = await client.request_dod(
+                measurement_state=None if refresh_state else self._cached_state
+            )
+            if refresh_state:
+                self._cached_state = snap.measurement_state
+                self._last_state_refresh = now
             snap.unit_id = self.unit_id
             persist_snapshot(snap, db)
             db.commit()
+            # Append to the downsampled backfill trail (~one row per TRAIL_SAMPLE_S).
+            if now - self._last_trail_store >= TRAIL_SAMPLE_S:
+                self._last_trail_store = now
+                self._store_trail(snap, db)
             status = db.query(NL43Status).filter_by(unit_id=self.unit_id).first()
             mst = (status.measurement_start_time.isoformat()
                    if status and status.measurement_start_time else None)
@@ -137,7 +252,27 @@ class DeviceMonitor:
         finally:
             db.close()
 
-    def _broadcast(self, payload: dict) -> None:
+    def _store_trail(self, snap, db) -> None:
+        """Append one downsampled reading to the backfill trail and prune old rows."""
+        from datetime import datetime, timedelta
+        from app.models import NL43Reading
+        try:
+            db.add(NL43Reading(
+                unit_id=self.unit_id, timestamp=datetime.utcnow(),
+                lp=snap.lp, leq=snap.leq, lmax=snap.lmax, ln1=snap.ln1, ln2=snap.ln2,
+            ))
+            cutoff = datetime.utcnow() - timedelta(hours=TRAIL_RETENTION_HOURS)
+            db.query(NL43Reading).filter(
+                NL43Reading.unit_id == self.unit_id,
+                NL43Reading.timestamp < cutoff,
+            ).delete()
+            db.commit()
+        except Exception as e:
+            logger.warning(f"[MONITOR] {self.unit_id}: trail store failed: {e}")
+
+    def _broadcast(self, payload: dict, cache: bool = True) -> None:
+        if cache:
+            self._last_payload = payload  # replayed to new subscribers
         for q in list(self._subscribers):
             try:
                 q.put_nowait(payload)
@@ -161,12 +296,23 @@ class MonitorManager:
                 self._monitors[unit_id] = m
             return m
 
+    def is_active(self, unit_id: str) -> bool:
+        """True if this unit has a running monitor feed (so the background poller
+        can skip it — the monitor already polls it more often)."""
+        m = self._monitors.get(unit_id)
+        return m is not None and m.running
+
     def status(self) -> dict:
         return {
             uid: {
                 "running": m.running,
                 "subscribers": m.subscriber_count(),
                 "keepalive": m._keepalive,
+                "reachable": m._reachable,
+                # what cadence the loop is currently using, for observability
+                "mode": ("backoff" if m._consec_fail > 0
+                         else "watched" if m._subscribers
+                         else "idle"),
             }
             for uid, m in self._monitors.items()
         }

app/routers.py

@@ -11,7 +11,7 @@ import os
 import asyncio
 
 from app.database import get_db
-from app.models import NL43Config, NL43Status, AlertRule, AlertEvent
+from app.models import NL43Config, NL43Status, AlertRule, AlertEvent, NL43Reading
 from app.services import NL43Client, persist_snapshot
 
 logger = logging.getLogger(__name__)
@@ -284,33 +284,53 @@ async def monitor_stream(websocket: WebSocket, unit_id: str):
                 payload = await asyncio.wait_for(queue.get(), timeout=1.0)
             except asyncio.TimeoutError:
                 continue  # re-check gone.done()
+            if gone.done():
+                break  # client disconnected while we waited — don't send into a closing socket
             await websocket.send_json(payload)
     except WebSocketDisconnect:
         logger.info(f"Monitor subscriber disconnected for {unit_id}")
     except Exception as e:
-        logger.warning(f"Monitor stream error for {unit_id}: {e}")
+        # A frame that races the close (client vanished mid-send) surfaces as
+        # "Unexpected ASGI message 'websocket.send' after ... websocket.close".
+        # That's expected on disconnect (the portal closes the socket on every tab
+        # switch), not an error — log it quietly.
+        msg = str(e)
+        if "after sending" in msg or "websocket.close" in msg or "response already completed" in msg:
+            logger.debug(f"Monitor stream for {unit_id} closed mid-send (client gone)")
+        else:
+            logger.warning(f"Monitor stream error for {unit_id}: {e}")
     finally:
         gone.cancel()
         await monitor.unsubscribe(queue)
 
 
 @router.post("/{unit_id}/monitor/start")
-async def monitor_start(unit_id: str):
+async def monitor_start(unit_id: str, db: Session = Depends(get_db)):
-    """Keep the device's feed running even with no browser attached, so alerting
+    """Enable 24/7 keepalive monitoring: persist monitor_enabled and start the feed
-    evaluates continuously. Runtime-only (resets on restart)."""
+    now, so alerting evaluates continuously even with no viewer. Survives restarts
+    (auto-started on boot from the persisted flag)."""
+    cfg = db.query(NL43Config).filter_by(unit_id=unit_id).first()
+    if cfg:
+        cfg.monitor_enabled = True
+        db.commit()
     from app.monitor import monitor_manager
     monitor = await monitor_manager.get(unit_id)
     await monitor.set_keepalive(True)
-    return {"status": "ok", "unit_id": unit_id, "running": monitor.running, "keepalive": True}
+    return {"status": "ok", "unit_id": unit_id, "monitor_enabled": True, "running": monitor.running}
 
 
 @router.post("/{unit_id}/monitor/stop")
-async def monitor_stop(unit_id: str):
+async def monitor_stop(unit_id: str, db: Session = Depends(get_db)):
-    """Drop the keep-alive; the feed stops once no browser subscribers remain."""
+    """Disable keepalive monitoring: persist monitor_enabled=False and drop the
+    keepalive (the feed stops once no browser subscribers remain)."""
+    cfg = db.query(NL43Config).filter_by(unit_id=unit_id).first()
+    if cfg:
+        cfg.monitor_enabled = False
+        db.commit()
     from app.monitor import monitor_manager
     monitor = await monitor_manager.get(unit_id)
     await monitor.set_keepalive(False)
-    return {"status": "ok", "unit_id": unit_id, "keepalive": False}
+    return {"status": "ok", "unit_id": unit_id, "monitor_enabled": False}
 
 
 @router.get("/_monitor/status")
@@ -320,6 +340,31 @@ async def monitor_status():
     return {"status": "ok", "monitors": monitor_manager.status()}
 
 
+@router.get("/{unit_id}/history")
+def get_monitor_history(unit_id: str, hours: float = 2.0, db: Session = Depends(get_db)):
+    """Recent downsampled monitor readings (the DOD trail) for the live-chart
+    backfill. Viewing only — NOT the FTP report data."""
+    from datetime import timedelta
+    hours = max(0.1, min(hours, 48.0))
+    cutoff = datetime.utcnow() - timedelta(hours=hours)
+    rows = (db.query(NL43Reading)
+            .filter(NL43Reading.unit_id == unit_id, NL43Reading.timestamp >= cutoff)
+            .order_by(NL43Reading.timestamp.asc()).all())
+    return {
+        "status": "ok",
+        "unit_id": unit_id,
+        "hours": hours,
+        "count": len(rows),
+        "readings": [
+            {
+                "timestamp": r.timestamp.isoformat() if r.timestamp else None,
+                "lp": r.lp, "leq": r.leq, "lmax": r.lmax, "ln1": r.ln1, "ln2": r.ln2,
+            }
+            for r in rows
+        ],
+    }
+
+
 # ============================================================================
 # ALERTS — threshold rules + fired events
 # ============================================================================
@@ -364,14 +409,47 @@ def _event_dict(e: AlertEvent) -> dict:
     }
 
 
+async def _sync_keepalive_to_rules(unit_id: str, db: Session):
+    """Keep a unit's monitor running while it has enabled alert rules, so the
+    evaluator runs 24/7 even with no browser watching. Turns keepalive ON (and
+    persists monitor_enabled so it survives a restart via the boot auto-start)
+    when enabled rules exist; never turns it OFF — a device may be kept alive for
+    other reasons, so operators control that on /admin/slmm."""
+    has_enabled = (db.query(AlertRule)
+                   .filter_by(unit_id=unit_id, enabled=True).first() is not None)
+    if not has_enabled:
+        return
+    cfg = db.query(NL43Config).filter_by(unit_id=unit_id).first()
+    if cfg and not cfg.monitor_enabled:
+        cfg.monitor_enabled = True
+        db.commit()
+    from app.monitor import monitor_manager
+    m = await monitor_manager.get(unit_id)
+    await m.set_keepalive(True)
+
+
+def _reset_rule_runtime(unit_id: str, rule_id: int, db: Session):
+    """After a rule edit/delete: drop its evaluator state machine and close any open
+    event, so a stale 'active' phase doesn't mis-evaluate against the new config and
+    the client portal doesn't stay 'in alarm' on a rule that changed or is gone."""
+    from app.alerts import alert_evaluator
+    alert_evaluator.forget_rule(unit_id, rule_id)
+    now = datetime.utcnow()
+    for evt in db.query(AlertEvent).filter_by(unit_id=unit_id, rule_id=rule_id, status="active").all():
+        evt.clear_at = now
+        evt.status = "cleared"
+    db.commit()
+
+
 @router.post("/{unit_id}/alerts/rules")
-def create_alert_rule(unit_id: str, payload: AlertRulePayload, db: Session = Depends(get_db)):
+async def create_alert_rule(unit_id: str, payload: AlertRulePayload, db: Session = Depends(get_db)):
     rule = AlertRule(unit_id=unit_id, **payload.model_dump())
     db.add(rule)
     db.commit()
     db.refresh(rule)
     from app.alerts import alert_evaluator
     alert_evaluator.invalidate(unit_id)
+    await _sync_keepalive_to_rules(unit_id, db)
     return {"status": "ok", "rule": _rule_dict(rule)}
 
 
@@ -382,7 +460,7 @@ def list_alert_rules(unit_id: str, db: Session = Depends(get_db)):
 
 
 @router.put("/{unit_id}/alerts/rules/{rule_id}")
-def update_alert_rule(unit_id: str, rule_id: int, payload: AlertRulePayload, db: Session = Depends(get_db)):
+async def update_alert_rule(unit_id: str, rule_id: int, payload: AlertRulePayload, db: Session = Depends(get_db)):
     rule = db.query(AlertRule).filter_by(id=rule_id, unit_id=unit_id).first()
     if not rule:
         raise HTTPException(status_code=404, detail="Alert rule not found")
@@ -392,11 +470,13 @@ def update_alert_rule(unit_id: str, rule_id: int, payload: AlertRulePayload, db:
     db.refresh(rule)
     from app.alerts import alert_evaluator
     alert_evaluator.invalidate(unit_id)
+    _reset_rule_runtime(unit_id, rule_id, db)
+    await _sync_keepalive_to_rules(unit_id, db)
     return {"status": "ok", "rule": _rule_dict(rule)}
 
 
 @router.delete("/{unit_id}/alerts/rules/{rule_id}")
-def delete_alert_rule(unit_id: str, rule_id: int, db: Session = Depends(get_db)):
+async def delete_alert_rule(unit_id: str, rule_id: int, db: Session = Depends(get_db)):
     rule = db.query(AlertRule).filter_by(id=rule_id, unit_id=unit_id).first()
     if not rule:
         raise HTTPException(status_code=404, detail="Alert rule not found")
@@ -404,6 +484,8 @@ def delete_alert_rule(unit_id: str, rule_id: int, db: Session = Depends(get_db))
     db.commit()
     from app.alerts import alert_evaluator
     alert_evaluator.invalidate(unit_id)
+    _reset_rule_runtime(unit_id, rule_id, db)   # close its open event so the portal doesn't stay red
+    await _sync_keepalive_to_rules(unit_id, db)  # no-op if no enabled rules remain
     return {"status": "ok", "deleted": rule_id}
 
 
@@ -501,6 +583,7 @@ def get_roster(db: Session = Depends(get_db)):
             "web_enabled": cfg.web_enabled,
             "poll_enabled": cfg.poll_enabled,
             "poll_interval_seconds": cfg.poll_interval_seconds,
+            "monitor_enabled": cfg.monitor_enabled,
             "status": None
         }
 
@@ -749,6 +832,7 @@ def get_status(unit_id: str, db: Session = Depends(get_db)):
             "unit_id": unit_id,
             "last_seen": status.last_seen.isoformat() if status.last_seen else None,
             "measurement_state": status.measurement_state,
+            "measurement_start_time": status.measurement_start_time.isoformat() if status.measurement_start_time else None,
             "lp": status.lp,
             "leq": status.leq,
             "lmax": status.lmax,
@@ -807,6 +891,7 @@ def upsert_status(unit_id: str, payload: StatusPayload, db: Session = Depends(ge
             "unit_id": unit_id,
             "last_seen": status.last_seen.isoformat(),
             "measurement_state": status.measurement_state,
+            "measurement_start_time": status.measurement_start_time.isoformat() if status.measurement_start_time else None,
             "lp": status.lp,
             "leq": status.leq,
             "lmax": status.lmax,
@@ -854,7 +939,7 @@ async def start_measurement(unit_id: str, db: Session = Depends(get_db)):
             db.expire_all()
             status = db.query(NL43Status).filter_by(unit_id=unit_id).first()
             logger.info(f"State check: measurement_state={status.measurement_state if status else 'None'}, start_time={status.measurement_start_time if status else 'None'}")
-            if status and status.measurement_state == "Measure" and status.measurement_start_time:
+            if status and status.measurement_state in ("Start", "Measure") and status.measurement_start_time:
                 logger.info(f"✓ Measurement state confirmed for {unit_id} with start time {status.measurement_start_time}")
                 break
 

app/services.py

@@ -680,10 +680,12 @@ class NL43Client:
         else:
             raise ValueError(f"Unknown result code: {result_code}")
 
-    async def request_dod(self) -> NL43Snapshot:
+    async def request_dod(self, measurement_state: Optional[str] = None) -> NL43Snapshot:
         """Request DOD (Data Output Display) snapshot from device.
 
-        Returns parsed measurement data from the device display.
+        Returns parsed measurement data from the device display. Pass
+        measurement_state to reuse a cached run state and skip the extra Measure?
+        round-trip (the state changes rarely); leave it None to query it.
         """
         # _send_command now handles result code validation and returns the data line
         resp = await self._send_command("DOD?\r\n")
@@ -706,12 +708,15 @@ class NL43Client:
 
         logger.info(f"Parsed {len(parts)} data points from DOD response")
 
-        # Query actual measurement state (DOD doesn't include this information)
+        # DOD doesn't include the run state. Query it only when not supplied by the
-        try:
+        # caller — the monitor passes a cached state most cycles and refreshes it
-            measurement_state = await self.get_measurement_state()
+        # occasionally, avoiding a second rate-limited command on every poll.
-        except Exception as e:
+        if measurement_state is None:
-            logger.warning(f"Failed to get measurement state, defaulting to 'Measure': {e}")
+            try:
-            measurement_state = "Measure"
+                measurement_state = await self.get_measurement_state()
+            except Exception as e:
+                logger.warning(f"Failed to get measurement state, defaulting to 'Measure': {e}")
+                measurement_state = "Measure"
 
         snap = NL43Snapshot(unit_id="", raw_payload=resp, measurement_state=measurement_state)
 

migrate_add_monitor_enabled.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python3
+"""
+Migration: add monitor_enabled column to nl43_config.
+
+Controls whether the live fan-out DOD monitor is kept alive 24/7 for a unit
+(which is what makes alerting continuous). Defaults to enabled. Run once per DB.
+"""
+
+import sqlite3
+import sys
+from pathlib import Path
+
+DB_PATH = Path(__file__).parent / "data" / "slmm.db"
+
+
+def migrate():
+    if not DB_PATH.exists():
+        print(f"Database not found at {DB_PATH}")
+        print("No migration needed - database will be created with new schema")
+        return
+
+    conn = sqlite3.connect(DB_PATH)
+    cursor = conn.cursor()
+    try:
+        cursor.execute("PRAGMA table_info(nl43_config)")
+        columns = [row[1] for row in cursor.fetchall()]
+
+        if "monitor_enabled" in columns:
+            print("✓ monitor_enabled column already exists, no migration needed")
+            return
+
+        print("Adding monitor_enabled column (default enabled)...")
+        # SQLite stores booleans as 0/1; default 1 = enabled.
+        cursor.execute("ALTER TABLE nl43_config ADD COLUMN monitor_enabled BOOLEAN DEFAULT 1")
+        conn.commit()
+        print("✓ Added monitor_enabled column")
+        print("\n✓ Migration completed successfully!")
+
+    except Exception as e:
+        conn.rollback()
+        print(f"✗ Migration failed: {e}", file=sys.stderr)
+        sys.exit(1)
+    finally:
+        conn.close()
+
+
+if __name__ == "__main__":
+    migrate()