Watchtower: a passive sentinel with alerts for the nodes you care about

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 <noreply@anthropic.com>

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:
             ``<config_dir>/adverts.json`` when not injected).
         remote_store: Store for remote-node admin state — cached settings and CLI
             history (defaults to ``<config_dir>/remote.json`` when not injected).
+        watch_store: Store for the Watchtower — watched nodes, rules, and the alert
+            log (defaults to ``<config_dir>/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:

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 (``<config_dir>/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

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()

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).
+        """

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:

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))

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):

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)