Make passive monitor a non-blocking background toggle with live counts

The monitor was a blocking listen window that monopolized the app. Convert
it to an always-on background subscription that logs overheard adverts/
telemetry while the user keeps using the menu.

- Device: add subscribe_observations() as the non-blocking capture primitive;
  listen() is now a thin foreground wrapper over it. Mock streams a live
  packet burst + cadence instead of a bounded synth loop.
- MonitorService (ctx.monitor): owns the subscription, session/total counters,
  and the on/off preference; logs each packet as it arrives.
- MonitorStore: persist on/off in <config_dir>/monitor.json, default off,
  remembered between sessions and auto-resumed at launch.
- Menu: the monitor entry is now a toggle/heard-nodes control panel; the
  header shows live session/total counts and repaints ~1s via refresh_interval.
- CLI monitor gains --on/--off (preference) and keeps --seconds for a scripted
  foreground capture.
- Add Repository.observation_count(); README + tests (129 pass).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

README.md

@@ -18,7 +18,7 @@ turned into interactive HTML visualizations.
 | **History / DB** | ✅ skeleton | Every tool execution and measurement is persisted and queryable. |
 | **TX-power optimization** | ✅ working | Sweep transmit power (coarse + refine), converge on the best signal, optionally apply, render an interactive chart. |
 | **Link budget** | ✅ working | Predict time-on-air, receiver sensitivity, range, and duty-cycle/dwell headroom for a radio config — offline, no radio needed. |
-| **Passive monitor** | ✅ working | Listen on the air and log every overheard advert/telemetry (SNR, RSSI, location) to a longitudinal history. |
+| **Passive monitor** | ✅ working | Toggle a non-blocking background logger that records every overheard advert/telemetry (SNR, RSSI, location) to a longitudinal history while you keep using the app; on/off is remembered between sessions and live packet counts show in the menu. |
 | **Link health** | ✅ working | Compare a target's recent traces to its rolling baseline and flag SNR/reliability regressions, end-to-end and per hop. |
 | **Route map** | ✅ working | Aggregate trace history into an interactive route-stability graph — which links are stable, which the mesh flaps between, with churn metrics. |
 | **Path optimization** | 🚧 planned | Discover and score mesh paths from neighbors outward. |

meshtools/context.py

@@ -8,7 +8,7 @@ never open a serial port.
 from __future__ import annotations
 
 from dataclasses import dataclass, field
-from typing import Optional
+from typing import TYPE_CHECKING, Optional
 
 from rich.console import Console
 
@@ -21,6 +21,9 @@ from .core.selection import resolve_device
 from .persistence.logging import get_logger
 from .persistence.repository import Repository
 
+if TYPE_CHECKING:
+    from .services.monitor_service import MonitorService
+
 
 @dataclass(slots=True)
 class AppContext:
@@ -55,12 +58,30 @@ class AppContext:
     selected_device: Optional[DiscoveredDevice] = None
     explicit_selection: bool = False
     _device: Optional[Device] = field(default=None, init=False, repr=False)
+    _monitor: "Optional[MonitorService]" = field(default=None, init=False, repr=False)
 
     @property
     def profile_name(self) -> Optional[str]:
         """Name of the active profile, if any."""
         return self.profile.name if self.profile else None
 
+    @property
+    def monitor(self) -> "MonitorService":
+        """Return the session's passive-monitor service, creating it on first use.
+
+        The service is built lazily so the (cheap) database read for the "total"
+        observation count and the on/off preference load happen only once, when
+        monitoring is first referenced.
+        """
+        if self._monitor is None:
+            from .core.monitor_store import MonitorStore
+            from .services.monitor_service import MonitorService
+
+            self._monitor = MonitorService(
+                self, MonitorStore(self.settings.config_dir / "monitor.json")
+            )
+        return self._monitor
+
     @property
     def log(self):  # type: ignore[no-untyped-def]
         """The application logger."""
@@ -105,7 +126,9 @@ class AppContext:
         return self._device
 
     async def aclose(self) -> None:
-        """Disconnect the device (if connected) and close the repository."""
+        """Stop monitoring, disconnect the device (if connected), and close the repo."""
+        if self._monitor is not None:
+            await self._monitor.aclose()
         if self._device is not None:
             await self._device.disconnect()
             self._device = None

meshtools/core/connection.py

@@ -23,6 +23,14 @@ from .models import Contact, Hop, Observation, TraceResult
 #: Callback invoked with each :class:`Observation` as passive monitoring captures it.
 ObservationCallback = Callable[[Observation], None]
 
+#: Zero-argument callable returned by :meth:`Device.subscribe_observations` that stops the
+#: subscription and releases its resources when invoked.
+Unsubscribe = Callable[[], None]
+
+#: How often the :class:`MockDevice` simulator emits a fresh burst of synthetic packets
+#: while a passive-monitor subscription is open (seconds).
+_MOCK_MONITOR_INTERVAL_S = 0.05
+
 TX_POWER_MIN = 1
 TX_POWER_MAX = 22
 
@@ -148,18 +156,38 @@ class Device(ABC):
     # -- passive monitoring ------------------------------------------------------
 
     @abstractmethod
+    async def subscribe_observations(
+        self, on_observation: "ObservationCallback"
+    ) -> "Unsubscribe":
+        """Begin passively capturing overheard packets, without blocking the caller.
+
+        Subscribes to the adverts (and, where the firmware surfaces them, telemetry
+        frames) the companion overhears and invokes ``on_observation`` with each
+        :class:`~meshtools.core.models.Observation` as it arrives. Capture continues in
+        the background until the returned callable is invoked to stop it; the radio is
+        never asked to transmit, it only listens. This is the primitive behind both the
+        always-on background monitor and the bounded :meth:`listen` helper.
+
+        Args:
+            on_observation: Callback invoked with each observation as it is heard.
+
+        Returns:
+            A zero-argument callable that stops capture and releases the subscription.
+        """
+
     async def listen(
         self,
         duration_s: float,
         *,
         on_observation: Optional["ObservationCallback"] = None,
     ) -> list[Observation]:
-        """Passively capture packets the companion overhears for ``duration_s`` seconds.
+        """Capture observations for a bounded window, then stop and return them.
 
-        Adverts (and, where the firmware surfaces them, telemetry frames) heard on the
-        air are returned as :class:`~meshtools.core.models.Observation` records with
-        whatever SNR/RSSI/location the radio reported. The radio is not asked to transmit;
-        it simply listens.
+        A thin, foreground convenience wrapper over :meth:`subscribe_observations` for
+        one-shot captures (e.g. the scripted ``monitor --seconds`` command): it
+        subscribes, waits ``duration_s`` seconds, unsubscribes, and returns everything
+        heard. For continuous, non-blocking capture, use :meth:`subscribe_observations`
+        directly.
 
         Args:
             duration_s: How long to listen, in seconds.
@@ -169,6 +197,19 @@ class Device(ABC):
         Returns:
             Every observation captured during the window, in arrival order.
         """
+        captured: list[Observation] = []
+
+        def collect(obs: Observation) -> None:
+            captured.append(obs)
+            if on_observation is not None:
+                on_observation(obs)
+
+        unsubscribe = await self.subscribe_observations(collect)
+        try:
+            await asyncio.sleep(duration_s)
+        finally:
+            unsubscribe()
+        return captured
 
     # -- configuration: extra reads ---------------------------------------------
 
@@ -615,12 +656,9 @@ class MeshCoreDevice(Device):
             return b"".join(hops), path_hash_flags(trace_size) or 0
         return None
 
-    async def listen(  # noqa: D102 - inherited docstring
-        self,
-        duration_s: float,
-        *,
-        on_observation: Optional[ObservationCallback] = None,
-    ) -> list[Observation]:
+    async def subscribe_observations(  # noqa: D102 - inherited docstring
+        self, on_observation: ObservationCallback
+    ) -> Unsubscribe:
         from meshcore import EventType
 
         mc = self._require()
@@ -631,15 +669,10 @@ class MeshCoreDevice(Device):
                 "monitoring isn't available. Upgrade the 'meshcore' library."
             )
 
-        observations: list[Observation] = []
-
         def make_handler(kind: str):  # type: ignore[no-untyped-def]
             def handler(event) -> None:  # noqa: ANN001
                 obs = observation_from_event(event, kind)
-                if obs is None:
-                    return
-                observations.append(obs)
-                if on_observation is not None:
+                if obs is not None:
                     on_observation(obs)
 
             return handler
@@ -657,9 +690,8 @@ class MeshCoreDevice(Device):
             etype = getattr(EventType, attr, None)
             if etype is not None:
                 subs.append(subscribe(etype, make_handler(kind)))
-        try:
-            await asyncio.sleep(duration_s)
-        finally:
+
+        def unsubscribe() -> None:
             for sub in subs:
                 unsub = getattr(sub, "unsubscribe", None)
                 if unsub is not None:
@@ -667,7 +699,8 @@ class MeshCoreDevice(Device):
                         unsub()
                     except Exception:  # noqa: BLE001 - best-effort cleanup
                         pass
-        return observations
+
+        return unsubscribe
 
     @staticmethod
     def _ok(event):  # type: ignore[no-untyped-def]
@@ -848,6 +881,9 @@ class MockDevice(Device):
         self._channels: dict[int, dict] = {}
         self._device_pin = 0
         self._private_key = "11" * 32
+        # Background emitter tasks spawned by ``subscribe_observations``; tracked so they
+        # can be cancelled on disconnect and are never garbage-collected while pending.
+        self._bg_tasks: set[asyncio.Task] = set()
 
     async def connect(self) -> None:  # noqa: D102 - inherited docstring
         await asyncio.sleep(0)
@@ -855,6 +891,9 @@ class MockDevice(Device):
 
     async def disconnect(self) -> None:  # noqa: D102 - inherited docstring
         self._connected = False
+        for task in list(self._bg_tasks):
+            task.cancel()
+        self._bg_tasks.clear()
 
     async def get_self_info(self) -> dict:  # noqa: D102 - inherited docstring
         return {**self._info, "tx_power": self._tx_power}
@@ -987,41 +1026,71 @@ class MockDevice(Device):
         self._custom_vars.clear()
         self._channels.clear()
 
-    async def listen(  # noqa: D102 - inherited docstring
-        self,
-        duration_s: float,
-        *,
-        on_observation: Optional[ObservationCallback] = None,
-    ) -> list[Observation]:
-        # Synthesize plausible adverts from the known contacts, cycling through them.
-        # Two of them ("…Repeater") advertise a fixed location so location-aware features
-        # (e.g. the coverage map) have data. The wall-clock duration is honored only
-        # loosely — the simulator emits a count proportional to it and paces minimally so
-        # tests stay fast.
-        repeats = max(1, round(duration_s / 5.0))
-        locations = {
-            "Yagi-Repeater": (45.5019, -73.5674),
-            "Local-Repeater": (45.4768, -73.5990),
-        }
-        observations: list[Observation] = []
-        for r in range(repeats):
-            for c in self._contacts:
-                kind = "telemetry" if (r + len(observations)) % 4 == 3 else "advert"
-                lat_lon = locations.get(c.name)
-                obs = Observation(
-                    node=c.key_prefix or c.public_key[:12],
-                    name=c.name,
-                    kind=kind,
-                    snr=round(self._rng.gauss(6.0, 3.0), 1),
-                    rssi=round(self._rng.gauss(-95.0, 8.0), 1),
-                    lat=lat_lon[0] if lat_lon else None,
-                    lon=lat_lon[1] if lat_lon else None,
-                )
-                observations.append(obs)
-                if on_observation is not None:
-                    on_observation(obs)
-                await asyncio.sleep(0)
-        return observations
+    async def subscribe_observations(  # noqa: D102 - inherited docstring
+        self, on_observation: ObservationCallback
+    ) -> Unsubscribe:
+        # Simulate a live packet stream. Emit a burst of synthetic adverts/telemetry from
+        # the known contacts *synchronously* here, then keep emitting at a steady cadence
+        # from a background task until unsubscribed. The immediate first burst means even
+        # a zero-length capture window always sees every contact (two of which carry a
+        # location), keeping bounded ``listen`` runs and their tests deterministic.
+        stop = asyncio.Event()
+        seq = 0
+
+        def emit_burst() -> None:
+            nonlocal seq
+            for contact in self._contacts:
+                on_observation(self._synth_observation(contact, seq))
+                seq += 1
+
+        emit_burst()
+
+        async def emit_loop() -> None:
+            while not stop.is_set():
+                try:
+                    await asyncio.wait_for(stop.wait(), _MOCK_MONITOR_INTERVAL_S)
+                except asyncio.TimeoutError:
+                    pass  # cadence tick elapsed; emit the next burst
+                if not stop.is_set():
+                    emit_burst()
+
+        task = asyncio.create_task(emit_loop())
+        self._bg_tasks.add(task)
+        task.add_done_callback(self._bg_tasks.discard)
+
+        def unsubscribe() -> None:
+            stop.set()  # wakes the loop's wait immediately; it exits on the next check
+
+        return unsubscribe
+
+    #: Fixed locations advertised by the two simulated repeaters, so location-aware
+    #: features (e.g. the coverage map) always have coordinates to work with.
+    _MOCK_LOCATIONS = {
+        "Yagi-Repeater": (45.5019, -73.5674),
+        "Local-Repeater": (45.4768, -73.5990),
+    }
+
+    def _synth_observation(self, contact: Contact, seq: int) -> Observation:
+        """Build one plausible synthetic observation for ``contact`` (simulator only).
+
+        Args:
+            contact: The contact to synthesize a reception from.
+            seq: Monotonic emission counter, used to vary the packet kind.
+
+        Returns:
+            A noisy :class:`Observation` tagged ``telemetry`` on every fourth packet and
+            ``advert`` otherwise, carrying a location for the simulated repeaters.
+        """
+        lat_lon = self._MOCK_LOCATIONS.get(contact.name)
+        return Observation(
+            node=contact.key_prefix or contact.public_key[:12],
+            name=contact.name,
+            kind="telemetry" if seq % 4 == 3 else "advert",
+            snr=round(self._rng.gauss(6.0, 3.0), 1),
+            rssi=round(self._rng.gauss(-95.0, 8.0), 1),
+            lat=lat_lon[0] if lat_lon else None,
+            lon=lat_lon[1] if lat_lon else None,
+        )
 
     def _expected_snr(self, hop_index: int) -> float:
         """Model SNR for a hop as an inverted-U in TX power plus distance falloff.

meshtools/core/monitor_store.py

@@ -0,0 +1,54 @@
+"""Persistence for the passive-monitor on/off preference.
+
+Whether background monitoring is enabled is remembered between sessions in a small JSON
+file in the config directory (``<config_dir>/monitor.json``), following the same pattern
+as :class:`~meshtools.core.device_store.DeviceStore`: it is machine/user state managed by
+the app rather than user-authored config, and it must survive a per-invocation ``--db``
+override, so it lives beside the config rather than in the swappable database.
+
+The preference defaults to *off*: a fresh install never captures until the user opts in.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+
+class MonitorStore:
+    """Reads and writes the remembered passive-monitor on/off preference."""
+
+    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
+
+    def load_enabled(self) -> bool:
+        """Return whether monitoring is enabled, defaulting to ``False``.
+
+        A missing or corrupt file is treated as "disabled" rather than an error, so a
+        stray edit never blocks startup and never silently starts capturing.
+
+        Returns:
+            ``True`` if monitoring was left enabled, otherwise ``False``.
+        """
+        try:
+            data = json.loads(self._path.read_text(encoding="utf-8"))
+        except (OSError, ValueError):
+            return False
+        return bool(data.get("enabled", False)) if isinstance(data, dict) else False
+
+    def save_enabled(self, enabled: bool) -> None:
+        """Persist the monitoring on/off preference.
+
+        Args:
+            enabled: Whether background monitoring should be on for future sessions.
+        """
+        self._path.parent.mkdir(parents=True, exist_ok=True)
+        # Atomic-ish replace so a crash mid-write can't leave a truncated file.
+        tmp = self._path.with_suffix(self._path.suffix + ".tmp")
+        tmp.write_text(json.dumps({"enabled": bool(enabled)}, indent=2), encoding="utf-8")
+        tmp.replace(self._path)

meshtools/persistence/repository.py

@@ -335,6 +335,18 @@ class Repository:
         )
         self._conn.commit()
 
+    def observation_count(self) -> int:
+        """Return the total number of overheard packets stored across every run.
+
+        Backs the monitor's "total" counter, so it counts all observations ever logged,
+        not just the current session's.
+
+        Returns:
+            The row count of the ``observations`` table.
+        """
+        row = self._conn.execute("SELECT COUNT(*) AS n FROM observations").fetchone()
+        return int(row["n"]) if row else 0
+
     def heard_nodes(self, *, since: Optional[datetime] = None) -> list[HeardNode]:
         """Aggregate stored observations into per-node reception statistics.
 

meshtools/services/monitor_service.py

@@ -0,0 +1,189 @@
+"""The passive-monitor background service.
+
+Passive monitoring records every advert/telemetry frame the companion overhears to the
+database, building the longitudinal history the coverage map and link-quality alerting
+read back. Unlike a one-shot capture, it runs as a non-blocking background subscription:
+once started it logs observations as they arrive while the user keeps using the app, and
+keeps going across menu actions until stopped or the session ends.
+
+The service is session-scoped state on the :class:`~meshtools.context.AppContext`
+(``ctx.monitor``). It owns the on/off preference (persisted via
+:class:`~meshtools.core.monitor_store.MonitorStore`), the live subscription, and the
+counters shown live in the menu.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Optional
+
+from ..core.connection import Unsubscribe
+from ..core.models import Observation
+
+if TYPE_CHECKING:
+    from ..context import AppContext
+    from ..core.monitor_store import MonitorStore
+
+
+class MonitorService:
+    """Manages the always-on, non-blocking passive-monitor subscription.
+
+    Attributes are private; interact through the properties and the async lifecycle
+    methods (:meth:`enable`, :meth:`disable`, :meth:`toggle`, :meth:`start`, :meth:`stop`,
+    :meth:`aclose`).
+    """
+
+    def __init__(self, ctx: "AppContext", store: "MonitorStore") -> None:
+        """Initialize the service and load the persisted on/off preference.
+
+        Args:
+            ctx: The shared application context (device, repository, logger).
+            store: Persistence for the on/off preference.
+        """
+        self._ctx = ctx
+        self._store = store
+        self._enabled = store.load_enabled()
+        self._unsubscribe: Optional[Unsubscribe] = None
+        self._run_id: Optional[int] = None
+        self._session_count = 0
+        self._run_start_count = 0
+        # Total observations already in the database when the session began; the live
+        # "total" is this plus what we capture this session (this process is the only
+        # writer during an interactive session), avoiding a DB count on every repaint.
+        self._start_total = ctx.repo.observation_count()
+        self._error: Optional[str] = None
+
+    @property
+    def enabled(self) -> bool:
+        """Whether monitoring is the desired state (persisted across sessions)."""
+        return self._enabled
+
+    @property
+    def active(self) -> bool:
+        """Whether a live subscription is currently capturing."""
+        return self._unsubscribe is not None
+
+    @property
+    def session_count(self) -> int:
+        """Observations captured since this process started."""
+        return self._session_count
+
+    @property
+    def last_error(self) -> Optional[str]:
+        """The reason the last :meth:`start` attempt failed, if any."""
+        return self._error
+
+    def total_count(self) -> int:
+        """Return the total observations logged, all time (including this session)."""
+        return self._start_total + self._session_count
+
+    def status_text(self) -> str:
+        """Return a compact one-line status for the live menu header.
+
+        Returns:
+            A glyph-prefixed summary: capture counts when active, a waiting note when
+            enabled but not yet capturing, or an off marker.
+        """
+        if self.active:
+            return (
+                f"● monitor ON · {self._session_count} this session "
+                f"· {self.total_count()} total"
+            )
+        if self._enabled:
+            return "● monitor ON (waiting for a device)"
+        return "○ monitor OFF"
+
+    async def start(self) -> None:
+        """Open a background subscription and begin logging observations.
+
+        Idempotent: a no-op if already active. Opens the device connection (which may
+        raise if no device can be selected) and records a ``monitor`` run that the
+        captured observations are linked to.
+
+        Raises:
+            Exception: Propagates any device/subscription error after recording it; the
+                on/off preference is left unchanged so the caller can surface the problem.
+        """
+        if self.active:
+            return
+        try:
+            device = await self._ctx.device()
+        except Exception as exc:  # noqa: BLE001 - remember why, then re-raise
+            self._error = str(exc)
+            raise
+        run_id = self._ctx.repo.start_run(
+            "monitor", {"mode": "background"}, self._ctx.profile_name
+        )
+        self._run_start_count = self._session_count
+
+        def on_observation(obs: Observation) -> None:
+            # Runs on the event loop as packets arrive; keep it cheap and defensive so a
+            # single bad write can never take down the subscription.
+            self._session_count += 1
+            try:
+                self._ctx.repo.record_observation(run_id, obs)
+            except Exception as exc:  # noqa: BLE001 - never let logging break capture
+                self._ctx.log.debug("monitor: failed to record observation: %s", exc)
+
+        try:
+            self._unsubscribe = await device.subscribe_observations(on_observation)
+        except Exception as exc:  # noqa: BLE001 - record, remember, and re-raise
+            self._ctx.repo.finish_run(run_id, "error", {"error": str(exc)})
+            self._error = str(exc)
+            raise
+        self._run_id = run_id
+        self._error = None
+        self._ctx.log.info("passive monitor started (run %s)", run_id)
+
+    async def stop(self) -> None:
+        """Stop the background subscription and close its run record.
+
+        Idempotent: a no-op if not active.
+        """
+        if not self.active:
+            return
+        try:
+            assert self._unsubscribe is not None
+            self._unsubscribe()
+        finally:
+            self._unsubscribe = None
+        if self._run_id is not None:
+            captured = self._session_count - self._run_start_count
+            self._ctx.repo.finish_run(self._run_id, "ok", {"observations": captured})
+            self._ctx.log.info("passive monitor stopped (run %s, %s pkts)", self._run_id, captured)
+            self._run_id = None
+
+    async def enable(self) -> None:
+        """Turn monitoring on, persist the preference, and start capturing.
+
+        Raises:
+            Exception: If capture could not start (e.g. no device); the preference is
+                still persisted as *on* so it resumes once a device is available.
+        """
+        self._enabled = True
+        self._store.save_enabled(True)
+        await self.start()
+
+    async def disable(self) -> None:
+        """Turn monitoring off, persist the preference, and stop capturing."""
+        self._enabled = False
+        self._store.save_enabled(False)
+        await self.stop()
+
+    async def toggle(self) -> bool:
+        """Flip the on/off state, starting or stopping capture accordingly.
+
+        Returns:
+            The new enabled state (``True`` if now on).
+
+        Raises:
+            Exception: Propagates a failure to start capture when turning on.
+        """
+        if self._enabled:
+            await self.disable()
+        else:
+            await self.enable()
+        return self._enabled
+
+    async def aclose(self) -> None:
+        """Stop capture at session end without changing the persisted preference."""
+        await self.stop()

meshtools/tools/monitor.py

@@ -1,9 +1,19 @@
-"""The ``monitor`` tool: passively log adverts/telemetry the mesh emits over time.
+"""The ``monitor`` tool: control the passive background logger and review its history.
 
-Standard MeshCore tooling is point-in-time. This tool sits on the companion radio and
-records every advert and telemetry frame it overhears — with SNR, RSSI, and any shared
-location — to the database, building the longitudinal history that the coverage map and
-link-quality alerting read back. It transmits nothing; it only listens.
+Passive monitoring records every advert and telemetry frame the companion overhears —
+with SNR, RSSI, and any shared location — to the database, building the longitudinal
+history that the coverage map and link-quality alerting read back. It transmits nothing;
+it only listens.
+
+Capture itself runs as a non-blocking background subscription owned by
+:class:`~meshtools.services.monitor_service.MonitorService` (``ctx.monitor``), so it never
+blocks the menu. This tool is the control panel for it: toggling capture on/off (a
+preference remembered between sessions) and reviewing the heard-node history. The live
+packet counters are shown in the main-menu header, not here.
+
+The CLI subcommand additionally offers a one-shot foreground capture (``--seconds``) for
+scripting, since a short-lived CLI process has no long-running menu to host a background
+subscription.
 """
 
 from __future__ import annotations
@@ -21,57 +31,215 @@ from ..services import trace_runner
 from ..ui.widgets import make_progress
 from .base import Tool, ToolResult, register
 
-#: Cap a single listening window so an interactive run can't block indefinitely; longer
-#: passive logging is still possible by re-running or scripting the subcommand.
+#: Cap a single foreground capture window so a scripted run can't block indefinitely;
+#: longer passive logging is done by enabling the background monitor instead.
 MAX_DURATION_S = 3600
 
 
 @register
 class MonitorTool(Tool):
-    """Listen for a window and log every overheard advert/telemetry to the database."""
+    """Toggle the passive background monitor and review the nodes it has heard."""
 
     name = "monitor"
-    help = "Passively log adverts/telemetry heard on the mesh (longitudinal history)."
+    help = "Toggle passive background monitoring on/off and review heard nodes."
     category = "Diagnostics"
     order = 20
 
     async def prompt_params(self, ctx: AppContext) -> Optional[dict[str, Any]]:
-        """Ask how long to listen.
+        """Show the monitor status and offer to toggle it or review heard nodes.
 
         Args:
             ctx: Shared application context.
 
         Returns:
-            A parameter dict, or ``None`` if the user cancelled.
+            A parameter dict naming the chosen ``action``, or ``None`` if cancelled.
         """
-        duration = await questionary.text(
-            f"Listen for how many seconds? (1-{MAX_DURATION_S})",
-            default="60",
-            validate=_is_valid_duration,
+        monitor = ctx.monitor
+        toggle_label = "Turn monitoring OFF" if monitor.enabled else "Turn monitoring ON"
+        choice = await questionary.select(
+            f"Passive monitor — {monitor.status_text()}",
+            choices=[
+                questionary.Choice(toggle_label, value="toggle"),
+                questionary.Choice("View heard nodes (all time)", value="view"),
+                questionary.Choice("Back", value="__back__"),
+            ],
         ).ask_async()
-        if duration is None:
+        if choice in (None, "__back__"):
             return None
-        return {"duration": int(duration)}
+        return {"action": choice}
+
+    async def execute(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult:
+        """Run the control-panel action directly, without a logged ``runs`` row.
+
+        The monitor entry is a control panel, not a measurement, so it is deliberately
+        not wrapped in run-logging (unlike the base :meth:`Tool.execute`). The background
+        capture session records its own ``monitor`` run instead.
+
+        Args:
+            ctx: Shared application context.
+            params: The chosen ``action``.
+
+        Returns:
+            The :class:`ToolResult` from :meth:`run`.
+        """
+        return await self.run(ctx, params)
 
     async def run(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult:
-        """Capture observations for the window, persist them, and summarize heard nodes.
+        """Dispatch the selected control-panel action.
 
         Args:
             ctx: Shared application context.
-            params: ``duration`` (seconds) and the injected ``_run_id``.
+            params: ``action`` — ``"toggle"`` or ``"view"``.
 
         Returns:
-            A :class:`ToolResult` noting how many nodes/packets were heard.
+            A :class:`ToolResult` describing the outcome.
         """
-        duration = max(1, min(MAX_DURATION_S, int(params.get("duration", 60))))
-        run_id = params["_run_id"]
+        if params.get("action") == "view":
+            return self._view(ctx)
+        return await self._toggle(ctx)
+
+    @staticmethod
+    async def _toggle(ctx: AppContext) -> ToolResult:
+        """Flip background monitoring on or off, surfacing any start failure.
+
+        Args:
+            ctx: Shared application context.
+
+        Returns:
+            A :class:`ToolResult` reporting the new state.
+        """
+        try:
+            now_on = await ctx.monitor.toggle()
+        except Exception as exc:  # noqa: BLE001 - report cleanly, keep the menu alive
+            return ToolResult(
+                summary={"enabled": ctx.monitor.enabled, "active": ctx.monitor.active,
+                         "error": str(exc)},
+                message=f"[warn]monitoring enabled but capture couldn't start:[/warn] {exc}",
+            )
+        state = "[ok]● ON[/ok]" if now_on else "[muted]○ OFF[/muted]"
+        return ToolResult(
+            summary={"enabled": now_on, "active": ctx.monitor.active},
+            message=f"passive monitor is now {state}",
+        )
+
+    @staticmethod
+    def _view(ctx: AppContext) -> ToolResult:
+        """Render the all-time heard-node summary from stored observations.
+
+        Reads only the database, so it needs no device connection and works whether or
+        not monitoring is currently active.
+
+        Args:
+            ctx: Shared application context.
+
+        Returns:
+            A :class:`ToolResult` summarizing the heard nodes.
+        """
+        heard = ctx.repo.heard_nodes()
+        if heard:
+            # Names come from the stored observations, so no device lookup is needed.
+            ctx.console.print(_heard_table(heard, lambda _node: None))
+        else:
+            ctx.console.print("[muted]no packets heard yet — turn monitoring on[/muted]")
+
+        packets = sum(n.count for n in heard)
+        located = sum(1 for n in heard if n.has_location)
+        return ToolResult(
+            summary={
+                "nodes_heard": len(heard),
+                "observations": packets,
+                "nodes_with_location": located,
+            },
+            message=(
+                f"[ok]✓[/ok] heard [brand]{packets}[/brand] packets from "
+                f"[brand]{len(heard)}[/brand] nodes (all time)"
+            ),
+        )
+
+    def register_cli(self, app: typer.Typer) -> None:
+        """Register the ``monitor`` subcommand.
+
+        Args:
+            app: The Typer application.
+        """
+        from ..core.monitor_store import MonitorStore
+
+        @app.command(name=self.name, help=self.help)
+        def _monitor(
+            on: bool = typer.Option(
+                False, "--on", help="Enable background monitoring for future sessions."
+            ),
+            off: bool = typer.Option(
+                False, "--off", help="Disable background monitoring."
+            ),
+            seconds: Optional[int] = typer.Option(
+                None,
+                "--seconds",
+                "-d",
+                help=f"Capture in the foreground for N seconds now, then exit "
+                f"(1-{MAX_DURATION_S}).",
+            ),
+        ) -> None:
+            from ..cli import _state
+
+            if on and off:
+                raise typer.BadParameter("Pass only one of --on / --off.")
+            assert _state is not None  # set by the callback before any subcommand runs
+            ctx = _state
+            store = MonitorStore(ctx.settings.config_dir / "monitor.json")
+
+            if on or off:
+                store.save_enabled(on)
+                state = "[ok]on[/ok]" if on else "[muted]off[/muted]"
+                ctx.console.print(
+                    f"[ok]✓[/ok] background monitoring set {state} for future "
+                    "interactive sessions"
+                )
+                return
+            if seconds is not None:
+                _run_foreground_capture(ctx, seconds)
+                return
+            # No flags: report the current preference and history size.
+            enabled = "[ok]on[/ok]" if store.load_enabled() else "[muted]off[/muted]"
+            ctx.console.print(
+                f"background monitoring is {enabled} · "
+                f"[brand]{ctx.repo.observation_count()}[/brand] observations logged all-time"
+            )
+
+
+def _run_foreground_capture(ctx: AppContext, seconds: int) -> None:
+    """Drive a one-shot foreground capture on its own event loop.
+
+    Args:
+        ctx: Shared application context.
+        seconds: Requested capture duration (clamped to ``1..MAX_DURATION_S``).
+    """
+    import asyncio
+
+    from ..cli import _drive
+
+    asyncio.run(_drive(_foreground_capture(ctx, seconds), ctx))
+
+
+async def _foreground_capture(ctx: AppContext, seconds: int) -> None:
+    """Capture overheard packets for a bounded window and print a heard-node table.
+
+    This is the scripted, blocking counterpart to the background monitor: it records to
+    its own ``monitor`` run and returns when the window elapses. Used by ``monitor
+    --seconds``.
+
+    Args:
+        ctx: Shared application context.
+        seconds: Requested capture duration (clamped to ``1..MAX_DURATION_S``).
+    """
+    duration = max(1, min(MAX_DURATION_S, int(seconds)))
+    run_id = ctx.repo.start_run(
+        "monitor", {"mode": "foreground", "duration": duration}, ctx.profile_name
+    )
+    observations: list[Observation] = []
+    try:
         device = await ctx.device()
-
-        # Resolve node hashes to friendly names where we know the contact.
-        contacts = await device.get_contacts()
-        resolve = trace_runner.make_node_resolver(contacts)
-
-        observations: list[Observation] = []
+        resolve = trace_runner.make_node_resolver(await device.get_contacts())
         with make_progress(ctx.console) as progress:
             task = progress.add_task(f"listening {duration}s", total=duration)
 
@@ -86,42 +254,31 @@ class MonitorTool(Tool):
 
             await device.listen(duration, on_observation=on_observation)
             progress.update(task, completed=duration)
+    except Exception as exc:  # noqa: BLE001 - record then re-raise for the CLI handler
+        ctx.repo.finish_run(run_id, "error", {"error": str(exc)})
+        raise
 
-        heard = _aggregate(observations)
-        if heard:
-            ctx.console.print(_heard_table(heard, resolve))
-        else:
-            ctx.console.print("[muted]no packets heard during the window[/muted]")
+    heard = _aggregate(observations)
+    if heard:
+        ctx.console.print(_heard_table(heard, resolve))
+    else:
+        ctx.console.print("[muted]no packets heard during the window[/muted]")
 
-        located = sum(1 for n in heard if n.has_location)
-        return ToolResult(
-            summary={
-                "duration_s": duration,
-                "observations": len(observations),
-                "nodes_heard": len(heard),
-                "nodes_with_location": located,
-            },
-            message=(
-                f"[ok]✓[/ok] heard [brand]{len(observations)}[/brand] packets from "
-                f"[brand]{len(heard)}[/brand] nodes in {duration}s"
-            ),
-        )
-
-    def register_cli(self, app: typer.Typer) -> None:
-        """Register the ``monitor`` subcommand.
-
-        Args:
-            app: The Typer application.
-        """
-        from ..cli import run_tool_command
-
-        @app.command(name=self.name, help=self.help)
-        def _monitor(
-            duration: int = typer.Option(
-                60, "--duration", "-d", help=f"Seconds to listen (max {MAX_DURATION_S})."
-            ),
-        ) -> None:
-            run_tool_command(self, {"duration": duration})
+    located = sum(1 for n in heard if n.has_location)
+    ctx.repo.finish_run(
+        run_id,
+        "ok",
+        {
+            "duration_s": duration,
+            "observations": len(observations),
+            "nodes_heard": len(heard),
+            "nodes_with_location": located,
+        },
+    )
+    ctx.console.print(
+        f"[ok]✓[/ok] heard [brand]{len(observations)}[/brand] packets from "
+        f"[brand]{len(heard)}[/brand] nodes in {duration}s"
+    )
 
 
 def _distinct(observations: list[Observation]) -> int:
@@ -180,23 +337,3 @@ def _heard_table(heard: list[HeardNode], resolve) -> Table:  # noqa: ANN001
             str(label), str(node.count), median_cell, best_cell, rssi_cell, loc_cell
         )
     return table
-
-
-def _is_valid_duration(value: str) -> bool | str:
-    """Validate a listening duration in ``1..MAX_DURATION_S`` for a text answer.
-
-    Args:
-        value: Raw input.
-
-    Returns:
-        ``True`` if valid, otherwise an error message string.
-    """
-    try:
-        seconds = int(value)
-    except ValueError:
-        return "Enter a whole number of seconds."
-    if seconds < 1:
-        return "Enter a number greater than zero."
-    if seconds > MAX_DURATION_S:
-        return f"Maximum {MAX_DURATION_S} seconds."
-    return True

meshtools/ui/menu.py

@@ -24,13 +24,39 @@ _MENU_STYLE = questionary.Style(
     ]
 )
 
+#: How often (seconds) prompt_toolkit repaints the menu, so the live monitor counters in
+#: the header tick up while the user sits at the prompt without pressing a key.
+_MENU_REFRESH_S = 1.0
+
+
+class _MenuPrompt:
+    """The menu's question line, re-rendered with live monitor counts on every repaint.
+
+    questionary formats the ``message`` into the prompt on each render, so an object
+    whose ``__str__`` reads the current counters gives a genuinely live header (paired
+    with the select's ``refresh_interval``) at no cost when nothing is happening.
+    """
+
+    def __init__(self, ctx: AppContext) -> None:
+        """Bind the prompt to the application context.
+
+        Args:
+            ctx: The shared application context, read for live monitor state.
+        """
+        self._ctx = ctx
+
+    def __str__(self) -> str:
+        """Return the prompt text plus the current passive-monitor status line."""
+        return f"What would you like to do?   [{self._ctx.monitor.status_text()}]"
+
 
 async def run_menu(ctx: AppContext) -> None:
     """Run the interactive menu loop until the user quits.
 
     Before showing the menu, prompts for a companion device (unless the simulator is in
     use or a port/profile was already chosen explicitly), so the banner reflects the
-    target and tools connect to the right radio.
+    target and tools connect to the right radio. If passive monitoring was left enabled
+    in a previous session, it resumes in the background here.
 
     Args:
         ctx: The shared application context.
@@ -38,33 +64,58 @@ async def run_menu(ctx: AppContext) -> None:
     console = ctx.console
     await _select_device_at_startup(ctx)
     console.print(banner(ctx.profile_name, ctx.mock, ctx.selected_device))
+    await _resume_monitor(ctx)
 
-    while True:
-        choices: list[questionary.Choice | questionary.Separator] = []
-        current_category: str | None = None
-        for tool in all_tools():
-            if tool.category != current_category:
-                current_category = tool.category
-                choices.append(questionary.Separator(f"── {current_category} ──"))
-            choices.append(
-                questionary.Choice(title=f"{tool.name}  —  {tool.help}", value=tool.name)
-            )
-        choices.append(questionary.Separator(" "))
-        choices.append(questionary.Choice(title="quit", value="__quit__"))
+    prompt = _MenuPrompt(ctx)
+    try:
+        while True:
+            choices: list[questionary.Choice | questionary.Separator] = []
+            current_category: str | None = None
+            for tool in all_tools():
+                if tool.category != current_category:
+                    current_category = tool.category
+                    choices.append(questionary.Separator(f"── {current_category} ──"))
+                choices.append(
+                    questionary.Choice(title=f"{tool.name}  —  {tool.help}", value=tool.name)
+                )
+            choices.append(questionary.Separator(" "))
+            choices.append(questionary.Choice(title="quit", value="__quit__"))
 
-        selection = await questionary.select(
-            "What would you like to do?",
-            choices=choices,
-            style=_MENU_STYLE,
-            qmark="◆",
-        ).ask_async()
+            selection = await questionary.select(
+                prompt,
+                choices=choices,
+                style=_MENU_STYLE,
+                qmark="◆",
+                refresh_interval=_MENU_REFRESH_S,
+            ).ask_async()
 
-        if selection in (None, "__quit__"):
-            console.print("[muted]bye 73![/muted]")
-            return
+            if selection in (None, "__quit__"):
+                console.print("[muted]bye 73![/muted]")
+                return
 
-        await _run_selection(ctx, selection)
-        console.print()
+            await _run_selection(ctx, selection)
+            console.print()
+    finally:
+        # Stop background capture and close its run record even on an unexpected exit.
+        await ctx.monitor.aclose()
+
+
+async def _resume_monitor(ctx: AppContext) -> None:
+    """Resume passive monitoring if it was left enabled, warning if it can't start.
+
+    A failure to start (typically no companion device selected) is non-fatal: the
+    preference stays on, so monitoring resumes automatically once a device is available.
+
+    Args:
+        ctx: The shared application context.
+    """
+    if not ctx.monitor.enabled:
+        return
+    try:
+        await ctx.monitor.start()
+        ctx.console.print("[muted]passive monitor resumed in the background[/muted]")
+    except Exception as exc:  # noqa: BLE001 - surface, don't crash the menu
+        ctx.console.print(f"[warn]passive monitor could not start:[/warn] {exc}")
 
 
 async def _select_device_at_startup(ctx: AppContext) -> None:

tests/test_monitor.py

@@ -6,12 +6,38 @@ hardware required.
 
 from __future__ import annotations
 
+import asyncio
+import logging
 from datetime import timedelta
 from pathlib import Path
 
-from meshtools.core.connection import MockDevice, observation_from_event
+import pytest
+
+from meshtools.core.connection import (
+    _MOCK_MONITOR_INTERVAL_S as _MOCK_INTERVAL,
+)
+from meshtools.core.connection import (
+    MockDevice,
+    observation_from_event,
+)
 from meshtools.core.models import HeardNode, Observation, utcnow
+from meshtools.core.monitor_store import MonitorStore
 from meshtools.persistence.repository import Repository
+from meshtools.services.monitor_service import MonitorService
+
+
+class _StubContext:
+    """Minimal stand-in for :class:`~meshtools.context.AppContext` for service tests."""
+
+    def __init__(self, repo: Repository, device: MockDevice) -> None:
+        self.repo = repo
+        self._device = device
+        self.profile_name = None
+        self.log = logging.getLogger("test.monitor")
+
+    async def device(self) -> MockDevice:
+        await self._device.connect()
+        return self._device
 
 
 async def test_mock_listen_emits_observations_and_callbacks() -> None:
@@ -29,6 +55,40 @@ async def test_mock_listen_emits_observations_and_callbacks() -> None:
     assert any(o.lat is not None and o.lon is not None for o in observations)
 
 
+async def test_subscribe_observations_streams_then_stops() -> None:
+    """The subscription emits an immediate burst, keeps streaming, and halts on unsub."""
+    device = MockDevice()
+    await device.connect()
+
+    seen: list[Observation] = []
+    unsubscribe = await device.subscribe_observations(seen.append)
+    try:
+        # The first burst is emitted synchronously, before any await.
+        first_burst = len(seen)
+        assert first_burst >= 1
+        assert any(o.lat is not None for o in seen)  # a located repeater is in the burst
+
+        await asyncio.sleep(_MOCK_INTERVAL * 3)
+        assert len(seen) > first_burst  # more packets arrived while we did nothing
+    finally:
+        unsubscribe()
+
+    frozen = len(seen)
+    await asyncio.sleep(_MOCK_INTERVAL * 3)
+    assert len(seen) == frozen  # nothing more after unsubscribing
+
+
+async def test_disconnect_stops_background_emitter() -> None:
+    """Disconnecting cancels any live subscription task rather than leaking it."""
+    device = MockDevice()
+    await device.connect()
+    await device.subscribe_observations(lambda _obs: None)
+    assert device._bg_tasks  # a background emitter is running
+    await device.disconnect()
+    await asyncio.sleep(0)  # let the cancellation settle
+    assert not device._bg_tasks
+
+
 def test_heard_node_aggregation() -> None:
     """from_observations summarizes count, robust SNR, latest RSSI, and location."""
     now = utcnow()
@@ -113,3 +173,70 @@ async def test_observations_round_trip_and_aggregate(tmp_path: Path) -> None:
     # Nodes are ordered most-recently-heard first.
     assert nodes == sorted(nodes, key=lambda n: n.last_seen, reverse=True)
     repo.close()
+
+
+def test_observation_count_totals_every_run(tmp_path: Path) -> None:
+    """observation_count sums observations across all runs, empty database included."""
+    repo = Repository(tmp_path / "count.db")
+    assert repo.observation_count() == 0
+    run_id = repo.start_run("monitor", {})
+    repo.record_observation(run_id, Observation(node="a1", snr=1.0))
+    repo.record_observation(run_id, Observation(node="b2", snr=2.0))
+    assert repo.observation_count() == 2
+    repo.close()
+
+
+def test_monitor_store_defaults_off_and_persists(tmp_path: Path) -> None:
+    """The preference defaults to off and round-trips through a fresh store instance."""
+    path = tmp_path / "monitor.json"
+    assert MonitorStore(path).load_enabled() is False  # nothing saved yet → off
+    MonitorStore(path).save_enabled(True)
+    assert MonitorStore(path).load_enabled() is True  # remembered across instances
+    MonitorStore(path).save_enabled(False)
+    assert MonitorStore(path).load_enabled() is False
+
+
+async def test_monitor_service_captures_in_background(tmp_path: Path) -> None:
+    """Enabling starts a live subscription that logs observations without blocking."""
+    repo = Repository(tmp_path / "svc.db")
+    ctx = _StubContext(repo, MockDevice())
+    store = MonitorStore(tmp_path / "monitor.json")
+    service = MonitorService(ctx, store)
+
+    assert not service.enabled and not service.active
+    await service.enable()
+    assert service.enabled and service.active
+    assert store.load_enabled() is True  # preference persisted for next session
+
+    await asyncio.sleep(_MOCK_INTERVAL * 3)  # let packets stream in while we "do other work"
+    assert service.session_count > 0
+    # Started from an empty database, so total equals what this session captured.
+    assert service.total_count() == service.session_count
+
+    await service.disable()
+    assert not service.enabled and not service.active
+    assert store.load_enabled() is False
+    assert repo.observation_count() == service.session_count  # everything was logged
+
+    frozen = service.session_count
+    await asyncio.sleep(_MOCK_INTERVAL * 3)
+    assert service.session_count == frozen  # capture really stopped
+    repo.close()
+
+
+async def test_monitor_service_enable_without_device_keeps_preference(tmp_path: Path) -> None:
+    """If capture can't start, the on preference is still remembered so it can resume."""
+
+    class _NoDevice(_StubContext):
+        async def device(self) -> MockDevice:
+            raise ValueError("no companion device selected")
+
+    repo = Repository(tmp_path / "nodev.db")
+    service = MonitorService(_NoDevice(repo, MockDevice()), MonitorStore(tmp_path / "m.json"))
+
+    with pytest.raises(ValueError):
+        await service.enable()
+    assert service.enabled is True  # preference stays on to resume later
+    assert service.active is False
+    assert service.last_error == "no companion device selected"
+    repo.close()