Add passive mesh monitor (longitudinal advert/telemetry logger)

Adds a Device.listen capability (real + mock), an Observation/HeardNode
domain model, an observations table with Repository.record_observation /
heard_nodes aggregation, and a `monitor` tool that listens for a window and
logs every overheard advert/telemetry (SNR, RSSI, shared location) to the
database. This is the longitudinal substrate later features read back.

The real-hardware event-payload mapping is best-effort and, like the trace
mapping, flagged for validation against the firmware's event schema; the mock
device synthesizes plausible adverts so the tool and tests run without a radio.

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

README.md

@@ -18,6 +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. |
 | **Path optimization** | 🚧 planned | Discover and score mesh paths from neighbors outward. |
 | **Visualizations** | 🟡 partial | Plotly TX-sweep chart (HTML) done; pyvis mesh graph + time-series planned. |
 

meshtools/core/connection.py

@@ -16,9 +16,12 @@ from __future__ import annotations
 import asyncio
 import random
 from abc import ABC, abstractmethod
-from typing import Optional
+from typing import Callable, Optional
 
-from .models import Contact, Hop, TraceResult
+from .models import Contact, Hop, Observation, TraceResult
+
+#: Callback invoked with each :class:`Observation` as passive monitoring captures it.
+ObservationCallback = Callable[[Observation], None]
 
 TX_POWER_MIN = 1
 TX_POWER_MAX = 22
@@ -142,6 +145,31 @@ class Device(ABC):
             A :class:`TraceResult`; ``success`` is ``False`` on timeout.
         """
 
+    # -- passive monitoring ------------------------------------------------------
+
+    @abstractmethod
+    async def listen(
+        self,
+        duration_s: float,
+        *,
+        on_observation: Optional["ObservationCallback"] = None,
+    ) -> list[Observation]:
+        """Passively capture packets the companion overhears for ``duration_s`` seconds.
+
+        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.
+
+        Args:
+            duration_s: How long to listen, in seconds.
+            on_observation: Optional callback invoked with each observation as it arrives
+                (e.g. to advance a live progress display).
+
+        Returns:
+            Every observation captured during the window, in arrival order.
+        """
+
     # -- configuration: extra reads ---------------------------------------------
 
     @abstractmethod
@@ -587,6 +615,60 @@ 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]:
+        from meshcore import EventType
+
+        mc = self._require()
+        subscribe = getattr(mc, "subscribe", None)
+        if subscribe is None:  # pragma: no cover - depends on installed meshcore build
+            raise DeviceCommandError(
+                "this meshcore build doesn't expose event subscription, so passive "
+                "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:
+                    on_observation(obs)
+
+            return handler
+
+        # Subscribe to whichever advert/telemetry event types this firmware/library build
+        # exposes. The event payload field names below are best-effort and, like the trace
+        # mapping, should be validated against your firmware's event schema.
+        subs = []
+        for attr, kind in (
+            ("ADVERTISEMENT", "advert"),
+            ("ADVERT", "advert"),
+            ("NEW_CONTACT", "advert"),
+            ("TELEMETRY_RESPONSE", "telemetry"),
+        ):
+            etype = getattr(EventType, attr, None)
+            if etype is not None:
+                subs.append(subscribe(etype, make_handler(kind)))
+        try:
+            await asyncio.sleep(duration_s)
+        finally:
+            for sub in subs:
+                unsub = getattr(sub, "unsubscribe", None)
+                if unsub is not None:
+                    try:
+                        unsub()
+                    except Exception:  # noqa: BLE001 - best-effort cleanup
+                        pass
+        return observations
+
     @staticmethod
     def _ok(event):  # type: ignore[no-untyped-def]
         """Return ``event`` if it succeeded, else raise its error payload.
@@ -905,6 +987,42 @@ 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
+
     def _expected_snr(self, hop_index: int) -> float:
         """Model SNR for a hop as an inverted-U in TX power plus distance falloff.
 
@@ -1010,6 +1128,55 @@ def parse_trace_hops(payload: dict) -> list[Hop]:
     return hops
 
 
+def observation_from_event(event, kind: str) -> Optional[Observation]:  # noqa: ANN001
+    """Map a meshcore advert/telemetry event into an :class:`Observation`.
+
+    The companion reports a node identifier, optionally a name and shared location, and
+    the SNR/RSSI of the reception. Field names vary across firmware and library versions,
+    so several common spellings are tried for each value. Like the trace mapping this is
+    best-effort and should be validated against your firmware's event schema.
+
+    Args:
+        event: A meshcore event (anything exposing a ``payload`` mapping).
+        kind: The observation class to tag the record with (e.g. ``advert``).
+
+    Returns:
+        The parsed :class:`Observation`, or ``None`` if the payload carried no node id.
+    """
+    payload = dict(getattr(event, "payload", {}) or {})
+    node = (
+        payload.get("public_key")
+        or payload.get("pubkey")
+        or payload.get("hash")
+        or payload.get("key_prefix")
+    )
+    if not node:
+        return None
+    node = str(node).lower().removeprefix("0x")[:12]
+    lat = payload.get("adv_lat", payload.get("lat"))
+    lon = payload.get("adv_lon", payload.get("lon"))
+    return Observation(
+        node=node,
+        name=payload.get("adv_name") or payload.get("name"),
+        kind=kind,
+        snr=_as_float(payload.get("snr")),
+        rssi=_as_float(payload.get("rssi")),
+        lat=_as_float(lat) if lat else None,
+        lon=_as_float(lon) if lon else None,
+        raw=payload,
+    )
+
+
+def _as_float(value: object) -> Optional[float]:
+    """Best-effort float conversion, returning ``None`` on missing/garbage values."""
+    if value is None:
+        return None
+    try:
+        return float(value)  # type: ignore[arg-type]
+    except (TypeError, ValueError):
+        return None
+
+
 def _mock_pub(prefix: str) -> str:
     """Build a 32-byte mock public key from a short hex prefix (simulator only).
 

meshtools/core/models.py

@@ -42,6 +42,102 @@ class Contact:
     last_seen: Optional[datetime] = None
 
 
+@dataclass(slots=True)
+class Observation:
+    """A single over-the-air reception heard while passively monitoring the mesh.
+
+    Adverts, telemetry frames, and other packets the companion overhears are captured as
+    observations and logged longitudinally, so the mesh's behavior can be reviewed over
+    time rather than only at the instant a command runs.
+
+    Attributes:
+        node: Key prefix / hash identifying the transmitting node, if known.
+        name: Friendly name advertised by the node, if carried.
+        kind: Packet class, e.g. ``advert`` or ``telemetry``.
+        snr: Signal-to-noise ratio (dB) the companion measured, if reported.
+        rssi: Received signal strength (dBm), if reported.
+        lat: Advertised latitude (decimal degrees), when the node shares location.
+        lon: Advertised longitude (decimal degrees), when the node shares location.
+        observed_at: When the packet was heard.
+        raw: Optional raw event payload for debugging/replay.
+    """
+
+    node: Optional[str]
+    name: Optional[str] = None
+    kind: str = "advert"
+    snr: Optional[float] = None
+    rssi: Optional[float] = None
+    lat: Optional[float] = None
+    lon: Optional[float] = None
+    observed_at: datetime = field(default_factory=utcnow)
+    raw: Optional[dict] = None
+
+
+@dataclass(slots=True)
+class HeardNode:
+    """Aggregated reception statistics for one node across many observations.
+
+    Attributes:
+        node: Key prefix / hash of the node (``None`` only if never identified).
+        name: Most recent friendly name seen for the node, if any.
+        count: Number of observations aggregated.
+        median_snr: Median SNR (dB) across observations that reported one.
+        best_snr: Strongest SNR (dB) seen, if any.
+        last_rssi: Most recent RSSI (dBm), if any.
+        last_seen: Timestamp of the most recent observation.
+        lat: Most recent advertised latitude, if the node shared one.
+        lon: Most recent advertised longitude, if the node shared one.
+    """
+
+    node: Optional[str]
+    name: Optional[str]
+    count: int
+    median_snr: Optional[float]
+    best_snr: Optional[float]
+    last_rssi: Optional[float]
+    last_seen: datetime
+    lat: Optional[float] = None
+    lon: Optional[float] = None
+
+    @property
+    def has_location(self) -> bool:
+        """Whether this node reported a usable latitude/longitude."""
+        return self.lat is not None and self.lon is not None
+
+    @classmethod
+    def from_observations(
+        cls, node: Optional[str], observations: list["Observation"]
+    ) -> "HeardNode":
+        """Aggregate one node's observations into reception statistics.
+
+        Args:
+            node: The node identifier these observations belong to.
+            observations: The observations for ``node`` (must be non-empty).
+
+        Returns:
+            A :class:`HeardNode` summarizing them. The most recent observation supplies
+            the name, RSSI, and location; SNR is summarized robustly (median + best).
+        """
+        ordered = sorted(observations, key=lambda o: o.observed_at)
+        latest = ordered[-1]
+        snrs = [o.snr for o in ordered if o.snr is not None]
+        located = next(
+            (o for o in reversed(ordered) if o.lat is not None and o.lon is not None), None
+        )
+        name = next((o.name for o in reversed(ordered) if o.name), None)
+        return cls(
+            node=node,
+            name=name,
+            count=len(ordered),
+            median_snr=statistics.median(snrs) if snrs else None,
+            best_snr=max(snrs) if snrs else None,
+            last_rssi=latest.rssi,
+            last_seen=latest.observed_at,
+            lat=located.lat if located else None,
+            lon=located.lon if located else None,
+        )
+
+
 @dataclass(slots=True)
 class Hop:
     """A single hop in a path trace.

meshtools/persistence/db.py

@@ -10,7 +10,7 @@ from __future__ import annotations
 import sqlite3
 from pathlib import Path
 
-SCHEMA_VERSION = 1
+SCHEMA_VERSION = 2
 
 _SCHEMA = """
 CREATE TABLE IF NOT EXISTS schema_meta (
@@ -76,9 +76,25 @@ CREATE TABLE IF NOT EXISTS path_candidates (
     created_at    TEXT    NOT NULL
 );
 
+-- One packet overheard while passively monitoring the mesh (advert/telemetry/...).
+CREATE TABLE IF NOT EXISTS observations (
+    id          INTEGER PRIMARY KEY AUTOINCREMENT,
+    run_id      INTEGER NOT NULL REFERENCES runs(id) ON DELETE CASCADE,
+    node        TEXT,
+    name        TEXT,
+    kind        TEXT    NOT NULL DEFAULT 'advert',
+    snr         REAL,
+    rssi        REAL,
+    lat         REAL,
+    lon         REAL,
+    observed_at TEXT    NOT NULL
+);
+
 CREATE INDEX IF NOT EXISTS idx_traces_run ON traces(run_id);
 CREATE INDEX IF NOT EXISTS idx_trace_hops_trace ON trace_hops(trace_id);
 CREATE INDEX IF NOT EXISTS idx_tx_samples_run ON tx_samples(run_id);
+CREATE INDEX IF NOT EXISTS idx_observations_run ON observations(run_id);
+CREATE INDEX IF NOT EXISTS idx_observations_node ON observations(node);
 """
 
 

meshtools/persistence/repository.py

@@ -14,7 +14,14 @@ from typing import Any, Optional
 
 from datetime import datetime
 
-from ..core.models import Hop, TraceResult, TxLevelResult, utcnow
+from ..core.models import (
+    HeardNode,
+    Hop,
+    Observation,
+    TraceResult,
+    TxLevelResult,
+    utcnow,
+)
 from . import db
 
 
@@ -232,3 +239,66 @@ class Repository:
             ),
         )
         self._conn.commit()
+
+    # -- observations (passive monitoring) --------------------------------------
+
+    def record_observation(self, run_id: int, obs: Observation) -> None:
+        """Persist one overheard packet from a monitoring run.
+
+        Args:
+            run_id: The owning run.
+            obs: The observation to store.
+        """
+        self._conn.execute(
+            "INSERT INTO observations "
+            "(run_id, node, name, kind, snr, rssi, lat, lon, observed_at) "
+            "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+            (
+                run_id,
+                obs.node,
+                obs.name,
+                obs.kind,
+                obs.snr,
+                obs.rssi,
+                obs.lat,
+                obs.lon,
+                obs.observed_at.isoformat(),
+            ),
+        )
+        self._conn.commit()
+
+    def heard_nodes(self, *, since: Optional[datetime] = None) -> list[HeardNode]:
+        """Aggregate stored observations into per-node reception statistics.
+
+        Spans every monitoring run (optionally limited to recent history), so the result
+        is a longitudinal view of which nodes have been heard, how strongly, and where —
+        the substrate for the monitor summary and the coverage map.
+
+        Args:
+            since: Only include observations at or after this time, if given.
+
+        Returns:
+            One :class:`HeardNode` per distinct node, ordered by most-recently heard.
+        """
+        sql = "SELECT node, name, snr, rssi, lat, lon, observed_at FROM observations"
+        params: list[Any] = []
+        if since is not None:
+            sql += " WHERE observed_at >= ?"
+            params.append(since.isoformat())
+        rows = self._conn.execute(sql, params).fetchall()
+
+        grouped: dict[Optional[str], list[Observation]] = {}
+        for row in rows:
+            obs = Observation(
+                node=row["node"],
+                name=row["name"],
+                snr=row["snr"],
+                rssi=row["rssi"],
+                lat=row["lat"],
+                lon=row["lon"],
+                observed_at=datetime.fromisoformat(row["observed_at"]),
+            )
+            grouped.setdefault(row["node"], []).append(obs)
+
+        nodes = [HeardNode.from_observations(node, obs) for node, obs in grouped.items()]
+        return sorted(nodes, key=lambda n: n.last_seen, reverse=True)

meshtools/tools/monitor.py

@@ -0,0 +1,202 @@
+"""The ``monitor`` tool: passively log adverts/telemetry the mesh emits over time.
+
+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.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Optional
+
+import questionary
+import typer
+from rich.table import Table
+from rich.text import Text
+
+from ..context import AppContext
+from ..core.models import HeardNode, Observation
+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.
+MAX_DURATION_S = 3600
+
+
+@register
+class MonitorTool(Tool):
+    """Listen for a window and log every overheard advert/telemetry to the database."""
+
+    name = "monitor"
+    help = "Passively log adverts/telemetry heard on the mesh (longitudinal history)."
+    category = "Diagnostics"
+    order = 20
+
+    async def prompt_params(self, ctx: AppContext) -> Optional[dict[str, Any]]:
+        """Ask how long to listen.
+
+        Args:
+            ctx: Shared application context.
+
+        Returns:
+            A parameter dict, or ``None`` if the user cancelled.
+        """
+        duration = await questionary.text(
+            f"Listen for how many seconds? (1-{MAX_DURATION_S})",
+            default="60",
+            validate=_is_valid_duration,
+        ).ask_async()
+        if duration is None:
+            return None
+        return {"duration": int(duration)}
+
+    async def run(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult:
+        """Capture observations for the window, persist them, and summarize heard nodes.
+
+        Args:
+            ctx: Shared application context.
+            params: ``duration`` (seconds) and the injected ``_run_id``.
+
+        Returns:
+            A :class:`ToolResult` noting how many nodes/packets were heard.
+        """
+        duration = max(1, min(MAX_DURATION_S, int(params.get("duration", 60))))
+        run_id = params["_run_id"]
+        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] = []
+        with make_progress(ctx.console) as progress:
+            task = progress.add_task(f"listening {duration}s", total=duration)
+
+            def on_observation(obs: Observation) -> None:
+                observations.append(obs)
+                ctx.repo.record_observation(run_id, obs)
+                progress.update(
+                    task,
+                    description=f"heard {len(observations)} pkts · "
+                    f"{_distinct(observations)} nodes",
+                )
+
+            await device.listen(duration, on_observation=on_observation)
+            progress.update(task, completed=duration)
+
+        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})
+
+
+def _distinct(observations: list[Observation]) -> int:
+    """Return the number of distinct nodes among observations."""
+    return len({o.node for o in observations})
+
+
+def _aggregate(observations: list[Observation]) -> list[HeardNode]:
+    """Group observations by node into :class:`HeardNode` records, freshest first.
+
+    Args:
+        observations: The observations captured this run.
+
+    Returns:
+        One aggregate per node, ordered by most-recently heard.
+    """
+    grouped: dict[Optional[str], list[Observation]] = {}
+    for obs in observations:
+        grouped.setdefault(obs.node, []).append(obs)
+    nodes = [HeardNode.from_observations(node, obs) for node, obs in grouped.items()]
+    return sorted(nodes, key=lambda n: n.last_seen, reverse=True)
+
+
+def _heard_table(heard: list[HeardNode], resolve) -> Table:  # noqa: ANN001
+    """Render the heard-node summary table.
+
+    Args:
+        heard: Aggregated per-node statistics.
+        resolve: Maps a raw node hash to a friendly contact name when known.
+
+    Returns:
+        A Rich :class:`Table` of node, packet count, SNR, RSSI, and location.
+    """
+    from ..ui.theme import snr_style
+
+    table = Table(title=f"Heard nodes ({len(heard)})", border_style="muted", expand=False)
+    table.add_column("Node")
+    table.add_column("Pkts", justify="right")
+    table.add_column("Median SNR", justify="right")
+    table.add_column("Best SNR", justify="right")
+    table.add_column("RSSI", justify="right")
+    table.add_column("Loc", justify="center")
+    for node in heard:
+        label = node.name or resolve(node.node) or node.node or "?"
+        median = node.median_snr
+        best = node.best_snr
+        median_cell = (
+            Text(f"{median:+.1f}", style=snr_style(median)) if median is not None else Text("—")
+        )
+        best_cell = (
+            Text(f"{best:+.1f}", style=snr_style(best)) if best is not None else Text("—")
+        )
+        rssi_cell = f"{node.last_rssi:.0f}" if node.last_rssi is not None else "—"
+        loc_cell = Text("●", style="ok") if node.has_location else Text("·", style="muted")
+        table.add_row(
+            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

tests/test_monitor.py

@@ -0,0 +1,115 @@
+"""Tests for the passive mesh monitor: listening, aggregation, and persistence.
+
+All run against the :class:`MockDevice` simulator and an on-disk SQLite database, no
+hardware required.
+"""
+
+from __future__ import annotations
+
+from datetime import timedelta
+from pathlib import Path
+
+from meshtools.core.connection import MockDevice, observation_from_event
+from meshtools.core.models import HeardNode, Observation, utcnow
+from meshtools.persistence.repository import Repository
+
+
+async def test_mock_listen_emits_observations_and_callbacks() -> None:
+    """Listening yields observations and fires the per-observation callback."""
+    device = MockDevice()
+    await device.connect()
+
+    seen: list[Observation] = []
+    observations = await device.listen(0.2, on_observation=seen.append)
+
+    assert observations  # the simulator advertises its known contacts
+    assert seen == observations  # every observation was reported via the callback
+    assert all(o.node for o in observations)
+    # At least one node advertises a location (for the coverage map).
+    assert any(o.lat is not None and o.lon is not None for o in observations)
+
+
+def test_heard_node_aggregation() -> None:
+    """from_observations summarizes count, robust SNR, latest RSSI, and location."""
+    now = utcnow()
+    obs = [
+        Observation(node="a1", name="Yagi", snr=4.0, rssi=-100.0, observed_at=now),
+        Observation(
+            node="a1", name="Yagi", snr=8.0, rssi=-90.0, lat=45.5, lon=-73.5,
+            observed_at=now + timedelta(seconds=5),
+        ),
+        Observation(
+            node="a1", name="Yagi", snr=6.0, rssi=-95.0, observed_at=now + timedelta(seconds=2)
+        ),
+    ]
+    node = HeardNode.from_observations("a1", obs)
+    assert node.count == 3
+    assert node.median_snr == 6.0  # median of [4, 8, 6]
+    assert node.best_snr == 8.0
+    assert node.last_seen == now + timedelta(seconds=5)  # freshest observation
+    assert node.last_rssi == -90.0  # RSSI from that freshest observation
+    assert node.has_location  # location carried by one observation is retained
+
+
+def test_heard_node_latest_wins_for_rssi_and_location() -> None:
+    """The freshest observation supplies RSSI; the freshest *located* one supplies coords."""
+    now = utcnow()
+    obs = [
+        Observation(node="b2", snr=1.0, rssi=-80.0, lat=10.0, lon=20.0, observed_at=now),
+        Observation(node="b2", snr=2.0, rssi=-70.0, observed_at=now + timedelta(seconds=10)),
+    ]
+    node = HeardNode.from_observations("b2", obs)
+    assert node.last_rssi == -70.0  # latest reading
+    assert (node.lat, node.lon) == (10.0, 20.0)  # latest one that actually had a fix
+
+
+def test_observation_from_event_parses_advert() -> None:
+    """The event parser pulls node id, name, SNR/RSSI, and location from a payload."""
+
+    class _Event:
+        payload = {
+            "public_key": "AABBCC0011",
+            "adv_name": "Repeater",
+            "snr": "7.5",
+            "rssi": -92,
+            "adv_lat": 45.5,
+            "adv_lon": -73.6,
+        }
+
+    obs = observation_from_event(_Event(), "advert")
+    assert obs is not None
+    assert obs.node == "aabbcc0011"
+    assert obs.name == "Repeater"
+    assert obs.snr == 7.5
+    assert obs.rssi == -92.0
+    assert (obs.lat, obs.lon) == (45.5, -73.6)
+
+
+def test_observation_from_event_without_node_is_skipped() -> None:
+    """A payload carrying no node identifier yields no observation."""
+
+    class _Event:
+        payload = {"snr": 3.0}
+
+    assert observation_from_event(_Event(), "advert") is None
+
+
+async def test_observations_round_trip_and_aggregate(tmp_path: Path) -> None:
+    """Observations persist and heard_nodes reaggregates them across runs."""
+    repo = Repository(tmp_path / "obs.db")
+    run_id = repo.start_run("monitor", {"duration": 1})
+
+    device = MockDevice()
+    await device.connect()
+    observations = await device.listen(
+        0.2, on_observation=lambda o: repo.record_observation(run_id, o)
+    )
+    assert observations
+
+    nodes = repo.heard_nodes()
+    assert nodes
+    # Every node aggregated should have been heard at least once.
+    assert sum(n.count for n in nodes) == len(observations)
+    # Nodes are ordered most-recently-heard first.
+    assert nodes == sorted(nodes, key=lambda n: n.last_seen, reverse=True)
+    repo.close()