From b6fcda0f917ed182f295e2a953821323e67901e2 Mon Sep 17 00:00:00 2001 From: Jean-Pierre Martineau Date: Sat, 11 Jul 2026 05:35:36 -0400 Subject: [PATCH] Watchtower: a passive sentinel with alerts for the nodes you care about MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Star the nodes that matter (the roof repeater, the far gateway) and session-long rules watch what the always-on hub already hears: a silence alarm when a starred node goes quiet past its threshold (latched per quiet spell, re-armed — with a recovery note — when it returns), an SNR-sag warning when its receptions' median drops well below the run before it, and a mesh-wide heads-up the first time a never-before-seen node id appears (baselined against the whole stored observation history, announced at most once ever). Nothing transmits. Alerts persist in watchtower.json alongside the watchlist and rules, show as a red ▲ badge in the header, and are read, acknowledged, and configured in a new Mesh-section Watchtower screen. A remote-battery rule was considered and dropped: observations carry no parsed battery telemetry, so there is no honest data source for it yet. Co-Authored-By: Claude Fable 5 --- meshterm/context.py | 24 ++ meshterm/core/watch_store.py | 408 +++++++++++++++++++++++++++++++ meshterm/services/watchtower.py | 232 ++++++++++++++++++ meshterm/tools/watchtower.py | 71 ++++++ meshterm/ui/menu.py | 8 + meshterm/ui/watchtower_screen.py | 275 +++++++++++++++++++++ tests/test_header.py | 9 +- tests/test_watchtower.py | 187 ++++++++++++++ 8 files changed, 1213 insertions(+), 1 deletion(-) create mode 100644 meshterm/core/watch_store.py create mode 100644 meshterm/services/watchtower.py create mode 100644 meshterm/tools/watchtower.py create mode 100644 meshterm/ui/watchtower_screen.py create mode 100644 tests/test_watchtower.py diff --git a/meshterm/context.py b/meshterm/context.py index 85a42ee..68c3339 100644 --- a/meshterm/context.py +++ b/meshterm/context.py @@ -16,6 +16,7 @@ from .core.admin_store import AdminStore from .core.advert_store import AdvertStore from .core.config import DeviceProfile, Settings from .core.remote_store import RemoteStore +from .core.watch_store import WatchStore from .core.connection import Device, make_device from .core.device_store import DeviceStore from .core.discovery import DiscoveredDevice, discover_devices @@ -28,6 +29,7 @@ if TYPE_CHECKING: from .services.chat_service import ChatService from .services.event_hub import EventHub from .services.monitor_service import MonitorService + from .services.watchtower import WatchtowerService from .ui.surface import Ui @@ -46,6 +48,8 @@ class AppContext: ``/adverts.json`` when not injected). remote_store: Store for remote-node admin state — cached settings and CLI history (defaults to ``/remote.json`` when not injected). + watch_store: Store for the Watchtower — watched nodes, rules, and the alert + log (defaults to ``/watchtower.json`` when not injected). mock: Whether the simulator device is in use. port_override: Explicit serial port (from ``--port`` or the interactive picker), overriding the profile. @@ -66,6 +70,7 @@ class AppContext: admin_store: AdminStore advert_store: Optional[AdvertStore] = None remote_store: Optional[RemoteStore] = None + watch_store: Optional[WatchStore] = None profile: Optional[DeviceProfile] = None mock: bool = False port_override: Optional[str] = None @@ -91,6 +96,7 @@ class AppContext: _monitor: "Optional[MonitorService]" = field(default=None, init=False, repr=False) _chat: "Optional[ChatService]" = field(default=None, init=False, repr=False) _adverts: "Optional[AdvertScheduler]" = field(default=None, init=False, repr=False) + _watchtower: "Optional[WatchtowerService]" = field(default=None, init=False, repr=False) _ui: "Optional[Ui]" = field(default=None, init=False, repr=False) def __post_init__(self) -> None: @@ -99,6 +105,8 @@ class AppContext: self.advert_store = AdvertStore(self.settings.config_dir / "adverts.json") if self.remote_store is None: self.remote_store = RemoteStore(self.settings.config_dir / "remote.json") + if self.watch_store is None: + self.watch_store = WatchStore(self.settings.config_dir / "watchtower.json") @property def profile_name(self) -> Optional[str]: @@ -241,6 +249,20 @@ class AppContext: self._adverts = AdvertScheduler(self) return self._adverts + @property + def watchtower(self) -> "WatchtowerService": + """Return the session's Watchtower sentinel, creating it on first use. + + Created idle here; the interactive session starts it alongside the other + always-on services. It only ever listens (rules over hub events), so scripted + CLI runs simply never start it. + """ + if self._watchtower is None: + from .services.watchtower import WatchtowerService + + self._watchtower = WatchtowerService(self) + return self._watchtower + @property def log(self): # type: ignore[no-untyped-def] """The application logger.""" @@ -454,6 +476,8 @@ class AppContext: async def aclose(self) -> None: """Stop monitoring and chat, stop the event hub, disconnect, and close the repo.""" + if self._watchtower is not None: + await self._watchtower.aclose() if self._adverts is not None: await self._adverts.aclose() if self._monitor is not None: diff --git a/meshterm/core/watch_store.py b/meshterm/core/watch_store.py new file mode 100644 index 0000000..4dfec43 --- /dev/null +++ b/meshterm/core/watch_store.py @@ -0,0 +1,408 @@ +"""Persistence for the Watchtower: watched nodes, their rules, and the alert log. + +The Watchtower (see :mod:`meshterm.services.watchtower`) is a passive sentinel: the user +stars the nodes they care about — the repeater on the roof, the gateway across town — +and rules watch what the always-on event hub already hears. This store remembers, per +watched node (keyed by the 12-hex canonical id observations use), the rule settings and +the last-heard mark the silence rule counts from, plus the rolling alert log and its +acknowledged state, so an alarm raised overnight is still waiting in the morning even +across a restart. + +Like the remembered devices (:mod:`meshterm.core.device_store`), this is global machine +state in a small JSON file (``/watchtower.json``) rather than the +per-invocation SQLite database. Reads are served from memory after the first load — the +header badge reads the unacked count on every repaint — and the packet-driven +``note_heard`` writes are throttled so a busy mesh doesn't grind the disk. +""" + +from __future__ import annotations + +import json +import time +from dataclasses import dataclass, field +from datetime import datetime +from pathlib import Path +from typing import Optional + +from .models import utcnow + +#: Silence-rule choices offered in the editor, in hours (:data:`OFF` disables it). +SILENCE_CHOICES_H = (1, 3, 6, 12, 24, 48) + +#: Default silence threshold: half a day covers nodes that only advertise daily-ish +#: without crying wolf over a quiet afternoon. +DEFAULT_SILENCE_HOURS = 12 + +#: The stored value meaning "this rule is off". +OFF = 0 + +#: How many alerts the log retains (oldest dropped first). +ALERT_CAP = 200 + +#: Minimum seconds between disk flushes for the packet-driven last-heard updates. +_FLUSH_EVERY_S = 60.0 + + +@dataclass(slots=True) +class WatchedNode: + """One starred node: its rules and the marks the rules count from. + + Attributes: + key: The node's canonical 12-hex id (what observations carry). + name: Display label, refreshed whenever the node is heard with a name. + silence_hours: Hours of silence before the alarm (:data:`OFF` disables it). + snr_watch: Whether the SNR-sag rule watches this node's receptions. + last_heard: When the node was last heard (seeded at watch time so the + silence countdown is armed immediately). + silent_since: Set while a silence alarm is active, so it fires once and + re-arms only after the node is heard again. + """ + + key: str + name: str + silence_hours: int = DEFAULT_SILENCE_HOURS + snr_watch: bool = True + last_heard: Optional[datetime] = None + silent_since: Optional[datetime] = None + + +@dataclass(slots=True) +class Alert: + """One raised alert, kept until it scrolls off the capped log. + + Attributes: + ident: Monotonic id (stable across restarts) used to acknowledge it. + when: When the rule tripped. + kind: ``silence`` / ``recovered`` / ``snr`` / ``new-node``. + label: The node's display label at the time. + message: The human-readable one-liner. + acked: Whether the user has acknowledged it (feeds the header badge). + """ + + ident: int + when: datetime + kind: str + label: str + message: str + acked: bool = False + + +@dataclass(slots=True) +class _State: + """The store's in-memory shape (mirrors the JSON file).""" + + watched: dict[str, WatchedNode] = field(default_factory=dict) + alerts: list[Alert] = field(default_factory=list) + known: set[str] = field(default_factory=set) + new_node_alerts: bool = True + next_id: int = 1 + + +class WatchStore: + """Reads and writes the Watchtower's state, memory-first.""" + + def __init__(self, path: Path) -> None: + """Open the store against a JSON file location. + + Args: + path: Path to the JSON state file (created lazily on first write). + """ + self._path = path + self._state: Optional[_State] = None + self._dirty = False + self._last_flush = 0.0 + + # --- state --------------------------------------------------------------------- + + @property + def state(self) -> _State: + """The in-memory state, loaded from disk on first access.""" + if self._state is None: + self._state = self._load() + return self._state + + def _load(self) -> _State: + """Parse the file into a :class:`_State`, or a fresh default on any trouble.""" + try: + data = json.loads(self._path.read_text(encoding="utf-8")) + except (OSError, ValueError): + data = {} + if not isinstance(data, dict): + data = {} + state = _State( + new_node_alerts=bool(data.get("new_node_alerts", True)), + next_id=max(1, _as_int(data.get("next_id"), 1)), + known={str(k) for k in data.get("known", []) if isinstance(k, str)}, + ) + watched = data.get("watched") + if isinstance(watched, dict): + for key, record in watched.items(): + if not isinstance(record, dict): + continue + state.watched[str(key)] = WatchedNode( + key=str(key), + name=str(record.get("name") or key), + silence_hours=_as_int(record.get("silence_hours"), DEFAULT_SILENCE_HOURS), + snr_watch=bool(record.get("snr_watch", True)), + last_heard=_as_time(record.get("last_heard")), + silent_since=_as_time(record.get("silent_since")), + ) + for record in data.get("alerts", []): + if not isinstance(record, dict): + continue + when = _as_time(record.get("when")) + if when is None: + continue + state.alerts.append( + Alert( + ident=_as_int(record.get("ident"), 0), + when=when, + kind=str(record.get("kind") or "alert"), + label=str(record.get("label") or "?"), + message=str(record.get("message") or ""), + acked=bool(record.get("acked", False)), + ) + ) + return state + + # --- watched nodes --------------------------------------------------------------- + + def watched(self) -> dict[str, WatchedNode]: + """The watched nodes by canonical id (the live mapping — treat as read-mostly).""" + return self.state.watched + + def is_watched(self, key: str) -> bool: + """Whether ``key`` is on the watchlist.""" + return key in self.state.watched + + def watch(self, key: str, name: str, *, last_seen: Optional[datetime] = None) -> None: + """Star a node with default rules. + + The silence countdown arms immediately: it counts from the contact's known + ``last_seen`` when there is one (so a node that has already been quiet for a + day alarms on the next sweep, which is exactly why it was starred), else from + now. + + Args: + key: The node's canonical 12-hex id. + name: Display label. + last_seen: When the node was last heard, if known. + """ + self.state.watched[key] = WatchedNode( + key=key, name=name, last_heard=last_seen or utcnow() + ) + self._save() + + def unwatch(self, key: str) -> None: + """Drop a node from the watchlist (its past alerts stay in the log).""" + if self.state.watched.pop(key, None) is not None: + self._save() + + def set_silence(self, key: str, hours: int) -> None: + """Set a watched node's silence threshold (:data:`OFF` disables the rule).""" + entry = self.state.watched.get(key) + if entry is not None: + entry.silence_hours = int(hours) + self._save() + + def set_snr_watch(self, key: str, on: bool) -> None: + """Enable/disable the SNR-sag rule for a watched node.""" + entry = self.state.watched.get(key) + if entry is not None: + entry.snr_watch = bool(on) + self._save() + + def note_heard( + self, key: str, *, when: Optional[datetime] = None, name: Optional[str] = None + ) -> None: + """Record that a watched node was heard (packet-driven; writes throttled). + + Args: + key: The node's canonical id. + when: Reception time (defaults to now); older-than-known stamps are kept. + name: A freshly advertised name, adopted as the display label when given. + """ + entry = self.state.watched.get(key) + if entry is None: + return + stamp = when or utcnow() + if entry.last_heard is None or stamp > entry.last_heard: + entry.last_heard = stamp + if name: + entry.name = name + self._dirty = True + self.flush(only_if_due=True) + + def mark_silent(self, key: str, when: Optional[datetime] = None) -> None: + """Latch a node's active silence alarm so it fires once per quiet spell.""" + entry = self.state.watched.get(key) + if entry is not None: + entry.silent_since = when or utcnow() + self._save() + + def clear_silent(self, key: str) -> None: + """Re-arm a node's silence alarm after it has been heard again.""" + entry = self.state.watched.get(key) + if entry is not None and entry.silent_since is not None: + entry.silent_since = None + self._save() + + # --- the new-node baseline --------------------------------------------------------- + + @property + def new_node_alerts(self) -> bool: + """Whether hearing a never-before-seen node raises an alert.""" + return self.state.new_node_alerts + + def set_new_node_alerts(self, on: bool) -> None: + """Toggle the new-node rule.""" + self.state.new_node_alerts = bool(on) + self._save() + + def known_contains(self, node: str) -> bool: + """Whether ``node`` has already been recorded (never re-announced).""" + return node in self.state.known + + def remember_known(self, node: str) -> None: + """Record ``node`` as seen so it is announced as new at most once, ever.""" + if node not in self.state.known: + self.state.known.add(node) + self._dirty = True + self.flush(only_if_due=True) + + # --- alerts --------------------------------------------------------------------- + + def add_alert( + self, kind: str, label: str, message: str, *, when: Optional[datetime] = None + ) -> Alert: + """Append an alert to the log (capped) and persist immediately. + + Args: + kind: The rule that tripped (``silence``/``recovered``/``snr``/``new-node``). + label: The node's display label. + message: The human-readable one-liner. + when: When it tripped (defaults to now). + + Returns: + The stored :class:`Alert`. + """ + state = self.state + alert = Alert( + ident=state.next_id, when=when or utcnow(), kind=kind, + label=label, message=message, + ) + state.next_id += 1 + state.alerts.append(alert) + if len(state.alerts) > ALERT_CAP: + del state.alerts[: len(state.alerts) - ALERT_CAP] + self._save() + return alert + + def alerts(self) -> list[Alert]: + """The alert log, newest first (a copy, safe to slice).""" + return list(reversed(self.state.alerts)) + + def ack(self, ident: int) -> None: + """Acknowledge one alert by id.""" + for alert in self.state.alerts: + if alert.ident == ident and not alert.acked: + alert.acked = True + self._save() + return + + def ack_all(self) -> None: + """Acknowledge every alert.""" + changed = False + for alert in self.state.alerts: + if not alert.acked: + alert.acked = True + changed = True + if changed: + self._save() + + def clear_acked(self) -> None: + """Drop acknowledged alerts from the log, keeping the live ones.""" + state = self.state + kept = [a for a in state.alerts if not a.acked] + if len(kept) != len(state.alerts): + state.alerts = kept + self._save() + + def unacked_count(self) -> int: + """How many alerts are waiting — the header badge's number.""" + return sum(1 for a in self.state.alerts if not a.acked) + + # --- persistence ----------------------------------------------------------------- + + def flush(self, *, only_if_due: bool = False) -> None: + """Write pending throttled changes to disk. + + Args: + only_if_due: Skip when the last flush was recent (the packet-driven + path); a plain ``flush()`` writes any pending change immediately. + """ + if not self._dirty: + return + if only_if_due and time.monotonic() - self._last_flush < _FLUSH_EVERY_S: + return + self._save() + + def _save(self) -> None: + """Persist the whole state atomically (crash mid-write keeps the old file).""" + state = self.state + data = { + "new_node_alerts": state.new_node_alerts, + "next_id": state.next_id, + "known": sorted(state.known), + "watched": { + key: { + "name": e.name, + "silence_hours": e.silence_hours, + "snr_watch": e.snr_watch, + "last_heard": e.last_heard.isoformat() if e.last_heard else None, + "silent_since": e.silent_since.isoformat() if e.silent_since else None, + } + for key, e in state.watched.items() + }, + "alerts": [ + { + "ident": a.ident, + "when": a.when.isoformat(), + "kind": a.kind, + "label": a.label, + "message": a.message, + "acked": a.acked, + } + for a in state.alerts + ], + } + self._path.parent.mkdir(parents=True, exist_ok=True) + tmp = self._path.with_suffix(self._path.suffix + ".tmp") + tmp.write_text(json.dumps(data, indent=2), encoding="utf-8") + tmp.replace(self._path) + self._dirty = False + self._last_flush = time.monotonic() + + +def _as_int(value: object, default: int) -> int: + """Coerce a stored number to a non-negative int, falling back to ``default``.""" + try: + number = int(value) # type: ignore[arg-type] + except (TypeError, ValueError): + return default + return number if number >= 0 else default + + +def _as_time(value: object) -> Optional[datetime]: + """Parse a stored ISO-8601 timestamp, or ``None`` if absent/corrupt. + + A timestamp without a zone is treated as corrupt too: the store only ever writes + aware UTC stamps, and a naive one would poison the silence arithmetic. + """ + if not isinstance(value, str): + return None + try: + parsed = datetime.fromisoformat(value) + except ValueError: + return None + return parsed if parsed.tzinfo is not None else None diff --git a/meshterm/services/watchtower.py b/meshterm/services/watchtower.py new file mode 100644 index 0000000..6b154a2 --- /dev/null +++ b/meshterm/services/watchtower.py @@ -0,0 +1,232 @@ +"""The Watchtower: passive rules over what the hub already hears, raising alerts. + +A sentinel, not a prober — it transmits nothing and asks the radio for nothing. It rides +the always-on :class:`~meshterm.services.event_hub.EventHub` like the monitor does, and +watches for the things an operator actually loses sleep over: + +* **silence** — a starred node hasn't been heard for its configured threshold. Fires + once per quiet spell (latched in the store) and re-arms when the node returns, which + also raises a friendly **recovered** note. +* **SNR sag** — a starred node's receptions are degrading: the median of its last few + readings sits well below the median of the readings before them. A cooldown keeps a + slowly dying link from paging every minute. +* **new node** — something never before seen (not in the observation history at start, + not in the store's memory) just appeared on the mesh. Announced at most once per id, + ever. + +Alerts land in the :class:`~meshterm.core.watch_store.WatchStore` (which persists them), +feed the header's alert badge, and are read and acknowledged in the Watchtower screen +(:mod:`meshterm.ui.watchtower_screen`). The service is session-scoped state on the +:class:`~meshterm.context.AppContext` (``ctx.watchtower``), started alongside the other +always-on services; scripted CLI runs never start it. +""" + +from __future__ import annotations + +import asyncio +import statistics +from collections import deque +from datetime import datetime +from typing import TYPE_CHECKING, Optional + +from ..core.connection import Unsubscribe +from ..core.events import EventKind, MeshEvent +from ..core.models import Observation, utcnow +from ..core.watch_store import OFF + +if TYPE_CHECKING: + from ..context import AppContext + +#: Seconds between rule sweeps (the silence rule is time-driven, not packet-driven). +SWEEP_S = 30.0 + +#: How many recent SNR readings form the "now" side of the sag comparison. +SNR_WINDOW = 4 + +#: How far (dB) the recent median must sit below the prior median to alert. +SNR_SAG_DB = 6.0 + +#: Seconds before the sag rule may fire again for the same node. +SNR_COOLDOWN_S = 6 * 3600.0 + +#: How many SNR readings are kept per node (the comparison uses at most this many). +_SNR_KEEP = 12 + + +class WatchtowerService: + """Watches starred nodes and the mesh's cast of characters; raises alerts. + + Attributes are private; interact through the properties and the async lifecycle + methods (:meth:`start`, :meth:`stop`, :meth:`aclose`). Rule evaluation lives in + the synchronous :meth:`note` / :meth:`evaluate` so tests can drive it directly. + """ + + def __init__(self, ctx: "AppContext") -> None: + """Initialize the (idle) service. + + Args: + ctx: The shared application context (store, repository, event hub). + """ + self._ctx = ctx + self._unsubscribe: Optional[Unsubscribe] = None + self._task: Optional[asyncio.Task] = None + #: Every node id ever seen (DB history + store memory + this session), the + #: baseline the new-node rule compares against. ``None`` until started. + self._known: Optional[set[str]] = None + #: Rolling SNR readings per watched node, session-scoped. + self._snr: dict[str, deque[float]] = {} + #: When the sag rule last fired per node (the cooldown clock). + self._snr_fired: dict[str, datetime] = {} + + @property + def active(self) -> bool: + """Whether the sentinel is currently watching.""" + return self._unsubscribe is not None + + def unacked_count(self) -> int: + """Alerts awaiting acknowledgement — the header badge's number.""" + return self._ctx.watch_store.unacked_count() + + # --- lifecycle --------------------------------------------------------------------- + + async def start(self) -> None: + """Begin watching. Idempotent, device-free, and safe before any connection. + + Seeds the new-node baseline from the observation history (every id the DB has + ever heard) plus the store's memory of past announcements, then subscribes to + the hub and starts the sweep. Like the monitor, packets flow in whenever the + hub is pumping — including a hub that opens lazily later. + """ + if self.active: + return + store = self._ctx.watch_store + baseline = {h.node for h in self._ctx.repo.heard_nodes() if h.node} + baseline |= set(store.state.known) + baseline |= set(store.watched()) + self._known = baseline + + def on_event(event: MeshEvent) -> None: + obs = event.observation + if obs is not None: + try: + self.note(obs) + except Exception as exc: # noqa: BLE001 - a rule bug must not kill the hub + self._ctx.log.debug("watchtower: note failed: %s", exc) + + self._unsubscribe = self._ctx.events.subscribe(on_event, EventKind.OBSERVATION) + self._task = asyncio.ensure_future(self._sweep()) + self._ctx.log.info("watchtower watching (%d nodes starred)", len(store.watched())) + + async def stop(self) -> None: + """Stop watching and flush pending store writes. Idempotent.""" + if self._unsubscribe is not None: + self._unsubscribe() + self._unsubscribe = None + if self._task is not None: + self._task.cancel() + try: + await self._task + except (asyncio.CancelledError, Exception): # noqa: BLE001 - teardown + pass + self._task = None + self._ctx.watch_store.flush() + + async def aclose(self) -> None: + """Stop watching at session end.""" + await self.stop() + + async def _sweep(self) -> None: + """Run the time-driven rules on a slow heartbeat.""" + while True: + await asyncio.sleep(SWEEP_S) + try: + self.evaluate() + except Exception as exc: # noqa: BLE001 - keep the sentinel alive + self._ctx.log.debug("watchtower: sweep failed: %s", exc) + + # --- packet-driven rules ------------------------------------------------------------- + + def note(self, obs: Observation) -> None: + """Fold one observation into the rules (called per hub packet; keep cheap). + + Args: + obs: The overheard observation. + """ + node = obs.node + if not node: + return + store = self._ctx.watch_store + entry = store.watched().get(node) + + # New node: never in the DB, the store's memory, or this session before now. + # Starring a node is itself an introduction, so watched nodes are never "new". + if self._known is not None and node not in self._known: + self._known.add(node) + store.remember_known(node) + if entry is None and store.new_node_alerts: + label = obs.name or node + store.add_alert( + "new-node", label, + "first appearance — never heard on this mesh before", + when=obs.observed_at, + ) + + if entry is None: + return + was_silent = entry.silent_since is not None + store.note_heard(node, when=obs.observed_at, name=obs.name) + if was_silent: + store.clear_silent(node) + store.add_alert( + "recovered", entry.name, + "back on the air — heard again after a silence alarm", + when=obs.observed_at, + ) + # Packet-kind rows measure our link to the last relay, not to the node, so + # they prove liveness (above) but say nothing about the node's own signal. + if entry.snr_watch and obs.snr is not None and obs.kind != "packet": + self._track_snr(node, entry.name, obs) + + def _track_snr(self, node: str, label: str, obs: Observation) -> None: + """Fold one SNR reading in and fire the sag rule when the trend warrants.""" + readings = self._snr.setdefault(node, deque(maxlen=_SNR_KEEP)) + readings.append(float(obs.snr)) # type: ignore[arg-type] + if len(readings) < 2 * SNR_WINDOW: + return + recent = statistics.median(list(readings)[-SNR_WINDOW:]) + prior = statistics.median(list(readings)[:-SNR_WINDOW]) + if prior - recent < SNR_SAG_DB: + return + fired = self._snr_fired.get(node) + now = obs.observed_at or utcnow() + if fired is not None and (now - fired).total_seconds() < SNR_COOLDOWN_S: + return + self._snr_fired[node] = now + self._ctx.watch_store.add_alert( + "snr", label, + f"reception sagging — median {prior:+.1f} dB → {recent:+.1f} dB", + when=now, + ) + + # --- time-driven rules --------------------------------------------------------------- + + def evaluate(self, now: Optional[datetime] = None) -> None: + """Run the silence rule over every watched node and flush the store. + + Args: + now: The evaluation time (defaults to the current time; tests inject). + """ + now = now or utcnow() + store = self._ctx.watch_store + for key, entry in store.watched().items(): + hours = entry.silence_hours + if hours == OFF or entry.last_heard is None or entry.silent_since is not None: + continue + quiet_s = (now - entry.last_heard).total_seconds() + if quiet_s >= hours * 3600: + store.mark_silent(key, now) + store.add_alert( + "silence", entry.name, + f"nothing heard for {hours} h", when=now, + ) + store.flush() diff --git a/meshterm/tools/watchtower.py b/meshterm/tools/watchtower.py new file mode 100644 index 0000000..08847a7 --- /dev/null +++ b/meshterm/tools/watchtower.py @@ -0,0 +1,71 @@ +"""The ``watchtower`` tool: a passive sentinel over the nodes you care about. + +Opens the Watchtower screen (see :mod:`meshterm.ui.watchtower_screen`): the alert log +with acknowledge/clear, the watchlist, and per-node rules. The rules themselves run in +the background for the whole session (see :mod:`meshterm.services.watchtower`) — a +silence alarm when a starred node goes quiet, an SNR-sag warning when its receptions +degrade, a recovery note when it returns, and a mesh-wide heads-up when a never-before- +seen node appears. Unacknowledged alerts show as the ``▲ n`` badge in the header, so +nothing here needs to be open to be on duty. + +Menu-only: the sentinel is inherently a live, session-long service; scripted runs can +read the same history through ``monitor``/``nodes``. +""" + +from __future__ import annotations + +from typing import Any, Optional + +import typer + +from ..context import AppContext +from .base import Tool, ToolResult, register + + +@register +class WatchtowerTool(Tool): + """Watch starred nodes: silence alarms, SNR sag, new-node sightings.""" + + name = "watchtower" + title = "Watchtower" + icon = "🚨" + help = "Alerts for watched nodes — silence, SNR sag, new arrivals" + category = "Mesh" + order = 7 # right under the Dashboard: the section's other always-on eye + + async def prompt_params(self, ctx: AppContext) -> Optional[dict[str, Any]]: + """Run the Watchtower screen; there are never parameters to collect. + + The screen presents everything itself and returns when dismissed, so + returning ``None`` tells the menu the invocation is complete (the same + pattern as the ``dashboard`` tool). + + Args: + ctx: Shared application context. + + Returns: + Always ``None``. + """ + from ..ui.watchtower_screen import open_watchtower + + await open_watchtower(ctx) + return None + + async def run(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult: + """Nothing runs here: the Watchtower lives inside :meth:`prompt_params`. + + Args: + ctx: Shared application context. + params: Unused. + + Returns: + An empty :class:`ToolResult` (only reachable from a scripted call). + """ + return ToolResult(summary={}) + + def register_cli(self, app: typer.Typer) -> None: + """Register no CLI command — the Watchtower is a live, menu-only feature. + + Args: + app: The Typer application (untouched). + """ diff --git a/meshterm/ui/menu.py b/meshterm/ui/menu.py index f72b5e3..5cd56a8 100644 --- a/meshterm/ui/menu.py +++ b/meshterm/ui/menu.py @@ -160,6 +160,12 @@ def _header(ctx: AppContext, cache: dict, width: int) -> Text: header.append(" · ") header.append("●", style="err") header.append(f" {unread}", style="warn") + alerts = ctx.watchtower.unacked_count() + if alerts: + # The Watchtower's badge: a triangle so it never reads as unread mail. + header.append(" · ") + header.append("▲", style="err") + header.append(f" {alerts}", style="warn") header.append(" · ") header.append("●" if ctx.events.active else "○", style="ok" if ctx.events.active else "muted") header.append(" ") @@ -502,6 +508,8 @@ async def _resume_monitor(ctx: AppContext) -> None: # each pass checks for a connected device and skips quietly without one, so starting # it here never opens the radio (and it simply waits out a deferred connect). await ctx.adverts.start() + # The Watchtower only listens (rules over hub events), so it too is safe from launch. + await ctx.watchtower.start() if not ctx.settings.connect_on_start: return try: diff --git a/meshterm/ui/watchtower_screen.py b/meshterm/ui/watchtower_screen.py new file mode 100644 index 0000000..f43f2c2 --- /dev/null +++ b/meshterm/ui/watchtower_screen.py @@ -0,0 +1,275 @@ +"""The Watchtower screens: the alert log, the watchlist, and per-node rules. + +The interactive face of the ``watchtower`` tool. One select-list screen carries the +whole feature (the persistent-backdrop pattern the config editors use): the alert log +newest-first — unacknowledged alerts lead with the header badge's red marker, Enter +acknowledges one — followed by the watchlist (Enter opens a node's rule popover) and +the actions: star another node, toggle the mesh-wide new-node rule, acknowledge or +clear in bulk. + +Everything here reads and writes the :class:`~meshterm.core.watch_store.WatchStore`; +the rules themselves run in :mod:`meshterm.services.watchtower`, which the menu starts +with the other always-on services (and this screen nudges, idempotently, in case it +is opened before that ever happened). Nothing transmits. +""" + +from __future__ import annotations + +import asyncio +from typing import TYPE_CHECKING, Any, Optional + +from rich.text import Text + +from ..core.models import NODE_TYPE_REPEATER, Contact +from ..core.watch_store import OFF, SILENCE_CHOICES_H, Alert, WatchedNode +from .tui import Choice, Separator +from .widgets import _age_seconds, _format_age + +if TYPE_CHECKING: + from ..context import AppContext + +#: Alert-kind display styles: alarms red, warnings amber, notes calm. +_KIND_STYLES = { + "silence": "err", + "snr": "warn", + "new-node": "brand", + "recovered": "ok", +} + +#: How many alerts the screen lists (the store keeps more; the tail rarely matters). +_SHOWN_ALERTS = 40 + +# Menu action sentinels (tuples so they never collide with alert ids or node keys). +_WATCH = ("watch",) +_TOGGLE_NEW = ("toggle-new",) +_ACK_ALL = ("ack-all",) +_CLEAR = ("clear",) + + +def contact_watch_key(contact: Contact) -> Optional[str]: + """The watch-store key for a contact: the 12-hex id observations carry.""" + ident = (contact.public_key or contact.key_prefix or "").lower().removeprefix("0x") + return ident[:12] or None + + +async def open_watchtower(ctx: "AppContext") -> Optional[dict[str, Any]]: + """Run the Watchtower screen until dismissed. + + Args: + ctx: The shared application context (must be running the interactive TUI). + + Returns: + A summary of what happened (for the tool's log), or ``None`` on plain exit. + + Raises: + RuntimeError: If called outside the interactive menu (no full-screen session). + """ + from .surface import TuiUi + from .tui import CANCEL, SelectScreen + + if not isinstance(ctx.ui, TuiUi): # pragma: no cover - guarded by the menu-only caller + raise RuntimeError("the watchtower is only available in the menu") + session = ctx.ui.session + await ctx.watchtower.start() # idempotent; normally already running + + contacts: list[Contact] = [] + try: + if ctx.is_connected or ctx.settings.connect_on_start: + device = await ctx.device() + contacts = await device.get_contacts() + except Exception: # noqa: BLE001 - the log and rules render fine without contacts + contacts = [] + + store = ctx.watch_store + loop = asyncio.get_running_loop() + cursor: Any = None + while True: + items = _menu_items(store.alerts(), store.watched(), store.new_node_alerts) + menu = SelectScreen( + "🚨 Watchtower — alerts & watched nodes", + items, + default=cursor, + wrap=False, + footer_hint="↑↓ move · Enter select/acknowledge · Esc back", + ) + menu.future = loop.create_future() + session.push(menu) + try: + choice = await menu.future + if choice in (None, CANCEL): + return {"watched": len(store.watched()), "unacked": store.unacked_count()} + cursor = choice + if choice == _WATCH: + await _pick_node(ctx, contacts) + elif choice == _TOGGLE_NEW: + store.set_new_node_alerts(not store.new_node_alerts) + elif choice == _ACK_ALL: + store.ack_all() + elif choice == _CLEAR: + store.clear_acked() + cursor = None # the row itself disappears + elif isinstance(choice, tuple) and choice[0] == "ack": + store.ack(int(choice[1])) + elif isinstance(choice, tuple) and choice[0] == "node": + await _node_rules(ctx, str(choice[1])) + finally: + session.pop(menu) + + +# --- the menu --------------------------------------------------------------------------- + + +def _menu_items( + alerts: list[Alert], watched: dict[str, WatchedNode], new_node_alerts: bool +) -> list: + """Build the screen's rows: alerts, then the watchlist, then the actions.""" + items: list = [Separator("Alerts", style="accent")] + if not alerts: + items.append(Separator(" nothing yet — tripped rules land here")) + for alert in alerts[:_SHOWN_ALERTS]: + items.append(Choice(_alert_row(alert), ("ack", alert.ident))) + unacked = sum(1 for a in alerts if not a.acked) + acked = len(alerts) - unacked + if unacked: + items.append(Choice(f"✔ Acknowledge all ({unacked})", _ACK_ALL)) + if acked: + items.append(Choice("🗑 Clear acknowledged", _CLEAR)) + + items.append(Separator("")) + items.append(Separator("Watched nodes", style="accent")) + if not watched: + items.append(Separator(" none starred yet — silence and SNR rules need one")) + for key in sorted(watched, key=lambda k: watched[k].name.casefold()): + items.append(Choice(_watched_row(watched[key]), ("node", key))) + items.append(Choice("⭐ Watch a node…", _WATCH)) + + items.append(Separator("")) + state = "[ok]on[/ok]" if new_node_alerts else "[muted]off[/muted]" + items.append( + Choice( + Text.from_markup(f"🔔 New-node alerts: {state} [muted]— announce first-ever appearances[/muted]"), + _TOGGLE_NEW, + ) + ) + return items + + +def _alert_row(alert: Alert) -> Text: + """One alert as a row: marker, age, kind, node, and the message.""" + row = Text() + if alert.acked: + row.append("○ ", style="muted") + else: + row.append("● ", style="err") + age = _format_age(_age_seconds(alert.when)) + row.append(f"{age:>5} ", style="muted") + row.append(alert.kind.ljust(10), style=_KIND_STYLES.get(alert.kind, "brand")) + row.append(alert.label, style="muted" if alert.acked else None) + row.append(f" — {alert.message}", style="muted") + return row + + +def _watched_row(entry: WatchedNode) -> Text: + """One watched node as a row: name, its rules, and when it was last heard.""" + row = Text() + row.append("▲ ", style="err" if entry.silent_since is not None else "brand") + row.append(entry.name) + silence = "off" if entry.silence_hours == OFF else f"{entry.silence_hours} h" + row.append(f" silence {silence}", style="muted") + row.append(" · SNR watch " + ("on" if entry.snr_watch else "off"), style="muted") + if entry.silent_since is not None: + row.append(" · ⚠ silent", style="err") + elif entry.last_heard is not None: + row.append(f" · heard {_format_age(_age_seconds(entry.last_heard))}", style="muted") + return row + + +# --- the flows -------------------------------------------------------------------------- + + +async def _pick_node(ctx: "AppContext", contacts: list[Contact]) -> None: + """Float the star-a-node picker: unwatched contacts, most recently heard first.""" + session = ctx.ui.session + store = ctx.watch_store + candidates = [ + (key, c) + for c in contacts + if (key := contact_watch_key(c)) is not None and not store.is_watched(key) + ] + if not candidates: + message = ( + "Every known contact is already watched." + if contacts + else "No contacts available — connect a device that knows some nodes first." + ) + await session.message_dialog(Text(message, style="muted"), title="watch a node") + return + + def recency(pair: tuple[str, Contact]) -> float: + seen = _age_seconds(pair[1].last_seen) + return seen if seen is not None else float("inf") + + items: list = [] + for key, contact in sorted(candidates, key=recency): + row = Text() + row.append("▲ " if contact.node_type == NODE_TYPE_REPEATER else "● ", style="brand") + row.append(contact.name) + row.append(f" heard {_format_age(_age_seconds(contact.last_seen))}", style="muted") + items.append(Choice(row, (key, contact))) + picked = await session.select( + "⭐ Watch a node", + items, + prompt="Silence and SNR rules will watch it from now on:", + ) + if picked is None: + return + key, contact = picked + store.watch(key, contact.name, last_seen=contact.last_seen) + + +async def _node_rules(ctx: "AppContext", key: str) -> None: + """Float one watched node's rule editor until dismissed (or the node is unstarred).""" + session = ctx.ui.session + store = ctx.watch_store + while True: + entry = store.watched().get(key) + if entry is None: + return + silence = "off" if entry.silence_hours == OFF else f"after {entry.silence_hours} h" + items = [ + Choice(f"🕐 Silence alarm {silence}", "silence"), + Choice("📶 SNR watch " + ("on" if entry.snr_watch else "off"), "snr"), + Separator(""), + Choice("✖ Stop watching this node", "unwatch"), + ] + picked = await session.select( + f"Rules — {entry.name}", items, filterable=False, + footer_hint="↑↓ move · Enter change · Esc back", + ) + if picked is None: + return + if picked == "silence": + await _pick_silence(ctx, key, entry) + elif picked == "snr": + store.set_snr_watch(key, not entry.snr_watch) + elif picked == "unwatch": + store.unwatch(key) + return + + +async def _pick_silence(ctx: "AppContext", key: str, entry: WatchedNode) -> None: + """Float the silence-threshold picker for one watched node.""" + session = ctx.ui.session + items = [Choice("Off — never alarm on silence", OFF)] + for hours in SILENCE_CHOICES_H: + label = f"After {hours} h of silence" + if hours == entry.silence_hours: + label += " (current)" + items.append(Choice(label, hours)) + picked = await session.select( + f"Silence alarm — {entry.name}", items, + default=entry.silence_hours, filterable=False, + footer_hint="↑↓ move · Enter set · Esc keep", + ) + if picked is not None: + ctx.watch_store.set_silence(key, int(picked)) diff --git a/tests/test_header.py b/tests/test_header.py index 7536c46..b72f5ab 100644 --- a/tests/test_header.py +++ b/tests/test_header.py @@ -7,16 +7,23 @@ from types import SimpleNamespace from meshterm.ui.menu import _HEADER_ACTIVITY_LEVELS, _header -def _ctx(histogram=(), unread=0, active=True) -> SimpleNamespace: +def _ctx(histogram=(), unread=0, active=True, alerts=0) -> SimpleNamespace: """A minimal stand-in exposing only what the header reads (simulator path).""" return SimpleNamespace( mock=True, chat=SimpleNamespace(unread_total=lambda: unread), events=SimpleNamespace(active=active), monitor=SimpleNamespace(activity_histogram=lambda: tuple(histogram)), + watchtower=SimpleNamespace(unacked_count=lambda: alerts), ) +def test_header_shows_the_watchtower_badge() -> None: + """Unacknowledged alerts appear as the triangle badge; none means no badge.""" + assert "▲ 2" in _header(_ctx(alerts=2), {}, 80).plain + assert "▲" not in _header(_ctx(alerts=0), {}, 80).plain + + def test_header_sparkline_fills_the_row_exactly() -> None: """Every cell the fixed segments leave is sparkline — no short rows, no overflow.""" for width in (60, 100): diff --git a/tests/test_watchtower.py b/tests/test_watchtower.py new file mode 100644 index 0000000..c7d4d9f --- /dev/null +++ b/tests/test_watchtower.py @@ -0,0 +1,187 @@ +"""Watchtower tests: the store's persistence and the sentinel's rules. + +The service is driven synchronously through :meth:`note`/:meth:`evaluate` against a +stub context (the monitor tests' approach), so no timers or hardware are involved. +""" + +from __future__ import annotations + +import logging +from datetime import timedelta +from pathlib import Path + +from meshterm.core.models import Observation, utcnow +from meshterm.core.watch_store import ALERT_CAP, OFF, WatchStore +from meshterm.persistence.repository import Repository +from meshterm.services.watchtower import SNR_WINDOW, WatchtowerService + +NODE = "3d" * 6 + + +class _StubContext: + """Minimal stand-in for :class:`~meshterm.context.AppContext` for rule tests.""" + + def __init__(self, tmp_path: Path) -> None: + self.repo = Repository(tmp_path / "wt.db") + self.watch_store = WatchStore(tmp_path / "watchtower.json") + self.log = logging.getLogger("test.watchtower") + + +def _service(tmp_path: Path) -> WatchtowerService: + ctx = _StubContext(tmp_path) + service = WatchtowerService(ctx) + service._known = set() # what start() seeds from the DB; empty history here + return service + + +def _obs(node=NODE, name="YUL", snr=None, age_s=0, kind="advert") -> Observation: + return Observation( + node=node, name=name, kind=kind, snr=snr, + observed_at=utcnow() - timedelta(seconds=age_s), + ) + + +# --- the store -------------------------------------------------------------------------- + + +def test_store_watch_round_trips_and_persists(tmp_path: Path) -> None: + """Watched nodes and their rules survive a fresh store instance.""" + path = tmp_path / "watchtower.json" + store = WatchStore(path) + seen = utcnow() - timedelta(hours=2) + store.watch(NODE, "YUL", last_seen=seen) + store.set_silence(NODE, 24) + store.set_snr_watch(NODE, False) + + again = WatchStore(path) + entry = again.watched()[NODE] + assert entry.name == "YUL" and entry.silence_hours == 24 and not entry.snr_watch + assert entry.last_heard == seen + again.unwatch(NODE) + assert not WatchStore(path).watched() + + +def test_store_alert_log_caps_acks_and_clears(tmp_path: Path) -> None: + """Alerts append newest-first, cap, acknowledge, and clear.""" + store = WatchStore(tmp_path / "watchtower.json") + first = store.add_alert("silence", "YUL", "quiet") + store.add_alert("snr", "YUL", "sagging") + assert [a.kind for a in store.alerts()] == ["snr", "silence"] + assert store.unacked_count() == 2 + + store.ack(first.ident) + assert store.unacked_count() == 1 + store.ack_all() + assert store.unacked_count() == 0 + store.clear_acked() + assert store.alerts() == [] + + for i in range(ALERT_CAP + 10): + store.add_alert("new-node", f"n{i}", "hi") + assert len(store.alerts()) == ALERT_CAP + assert store.alerts()[0].label == f"n{ALERT_CAP + 9}" # newest survives the cap + + +def test_store_note_heard_updates_and_refreshes_name(tmp_path: Path) -> None: + """Hearing a watched node advances its mark and adopts a fresh name.""" + store = WatchStore(tmp_path / "watchtower.json") + store.watch(NODE, NODE) + later = utcnow() + timedelta(minutes=5) + store.note_heard(NODE, when=later, name="YUL-Cartierville") + entry = store.watched()[NODE] + assert entry.last_heard == later and entry.name == "YUL-Cartierville" + + +# --- the silence rule --------------------------------------------------------------------- + + +def test_silence_fires_once_then_rearms_on_recovery(tmp_path: Path) -> None: + """Quiet past the threshold alarms once; hearing again notes the recovery.""" + service = _service(tmp_path) + store = service._ctx.watch_store + store.watch(NODE, "YUL", last_seen=utcnow() - timedelta(hours=13)) + + service.evaluate() # default threshold is 12 h — already past it + service.evaluate() # latched: no duplicate + alerts = store.alerts() + assert [a.kind for a in alerts] == ["silence"] + assert "12 h" in alerts[0].message + + service.note(_obs()) # the node returns + alerts = store.alerts() + assert [a.kind for a in alerts] == ["recovered", "silence"] + assert store.watched()[NODE].silent_since is None # re-armed + + service.evaluate() # freshly heard: quiet again only after another 12 h + assert len(store.alerts()) == 2 + + +def test_silence_respects_off_and_unheard(tmp_path: Path) -> None: + """An OFF rule never alarms, however stale the mark.""" + service = _service(tmp_path) + store = service._ctx.watch_store + store.watch(NODE, "YUL", last_seen=utcnow() - timedelta(days=30)) + store.set_silence(NODE, OFF) + service.evaluate() + assert store.alerts() == [] + + +# --- the SNR rule --------------------------------------------------------------------- + + +def test_snr_sag_alerts_with_cooldown(tmp_path: Path) -> None: + """A clear drop in median SNR alerts once, not on every subsequent packet.""" + service = _service(tmp_path) + store = service._ctx.watch_store + store.watch(NODE, "YUL") + + for _ in range(SNR_WINDOW): + service.note(_obs(snr=8.0)) + for _ in range(SNR_WINDOW): + service.note(_obs(snr=-2.0)) + sags = [a for a in store.alerts() if a.kind == "snr"] + assert len(sags) == 1 + assert "+8.0" in sags[0].message and "-2.0" in sags[0].message + + service.note(_obs(snr=-2.0)) # still sagging, but inside the cooldown + assert len([a for a in store.alerts() if a.kind == "snr"]) == 1 + + +def test_snr_ignores_packet_rows_and_disabled_watch(tmp_path: Path) -> None: + """Relay-measured packet SNR and an off switch both keep the rule quiet.""" + service = _service(tmp_path) + store = service._ctx.watch_store + store.watch(NODE, "YUL") + store.set_snr_watch(NODE, False) + for snr in (9.0,) * SNR_WINDOW + (-9.0,) * SNR_WINDOW: + service.note(_obs(snr=snr)) + assert [a for a in store.alerts() if a.kind == "snr"] == [] + + store.set_snr_watch(NODE, True) + for snr in (9.0,) * SNR_WINDOW + (-9.0,) * SNR_WINDOW: + service.note(_obs(snr=snr, kind="packet")) + assert [a for a in store.alerts() if a.kind == "snr"] == [] + + +# --- the new-node rule ------------------------------------------------------------------ + + +def test_new_node_announced_once_ever(tmp_path: Path) -> None: + """A first-ever id alerts once; repeats and restarts stay quiet.""" + service = _service(tmp_path) + store = service._ctx.watch_store + service.note(_obs(node="f7" * 6, name="Newcomer")) + service.note(_obs(node="f7" * 6, name="Newcomer")) + news = [a for a in store.alerts() if a.kind == "new-node"] + assert len(news) == 1 and news[0].label == "Newcomer" + assert store.known_contains("f7" * 6) # persisted: restarts stay quiet too + + +def test_new_node_rule_can_be_switched_off(tmp_path: Path) -> None: + """With the toggle off, first sightings are remembered but never announced.""" + service = _service(tmp_path) + store = service._ctx.watch_store + store.set_new_node_alerts(False) + service.note(_obs(node="c9" * 6)) + assert store.alerts() == [] + assert store.known_contains("c9" * 6)