Composer asks the gateway: fetch a repeater's neighbour table mid-path

The topology graph gains its fourth evidence source — the only one not
derived from our own reception. A repeater we hold admin rights on
reports its neighbour table (who it hears directly, at what SNR) when
asked over the mesh, revealing links and whole nodes our radio has
never received anything from: exactly the second vantage point the
composer runs dry without, two hops out.

Probed live against YUL-Cartierville (v1.15.0) through the Waymarker
before building: firmware answers the binary NEIGHBOURS request but
silently ignores guests, so the flow is login-first, tx-optimize style
(stored password, one-time prompt, forget on rejection). An empty table
is a normal answer — the gateway's own table was empty during the probe
(confirmed via its text CLI: '-none-') — so the UI says "ask again
later" instead of erroring.

Standing on a fetchable repeater, the composer shows a "fetch
neighbours" row with the suggestions it extends; the fetch runs under
the abortable spinner dialog, persists as a runs-row-owned snapshot
(schema v8, neighbour_reports), and the composer reopens mid-thought
with the same hops and refreshed suggestions tagged N. Refetching
supersedes the old snapshot per (repeater, neighbour) pair rather than
inflating the link's sample count: a table is current state, not an
event stream.

Verified end-to-end on hardware via MeshCoreDevice over BLE (login +
fetch + clean empty-table return) and in the mock (guest refusal,
discovery of a never-heard node, composer render + hand-off). 441
tests pass; the real database migrated v7 -> v8 cleanly.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

meshterm/core/connection.py

@@ -18,7 +18,7 @@ import logging
 import random
 import sys
 from abc import ABC, abstractmethod
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from typing import TYPE_CHECKING, Callable, Optional
 
 from .channels import CHANNEL_SLOT_PROBE_CAP
@@ -30,8 +30,10 @@ from .models import (
     Contact,
     Hop,
     Message,
+    NeighbourInfo,
     Observation,
     TraceResult,
+    utcnow,
 )
 
 if TYPE_CHECKING:
@@ -347,6 +349,29 @@ class Device(ABC):
             DeviceCommandError: If the node rejected the command.
         """
 
+    @abstractmethod
+    async def fetch_neighbours(self, node: Contact) -> list[NeighbourInfo]:
+        """Ask a remote node for its neighbour table: who it hears directly, and how well.
+
+        A second vantage point for the topology graph: every entry is a link
+        ``node ↔ neighbour`` with SNR measured *at the remote node*, including nodes we
+        have never received anything from ourselves. Requires an authenticated session
+        (:meth:`admin_login` first) — firmware silently ignores the request from guests,
+        which surfaces here as a timeout.
+
+        Args:
+            node: The contact to query (must already be logged in).
+
+        Returns:
+            The reported neighbour entries; empty when the node's table is empty (a
+            normal answer — repeaters forget neighbours across reboots and only relearn
+            them as adverts arrive).
+
+        Raises:
+            DeviceCommandError: If the node never answered (not logged in, out of
+                reach, or firmware without neighbour tables).
+        """
+
     @abstractmethod
     async def run_trace(
         self,
@@ -1263,6 +1288,39 @@ class MeshCoreDevice(Device):
         # since some firmware answers tersely or drops the ack under duty-cycle limits.
         await self._send_admin_cmd(node, f"set tx {value}")
 
+    async def fetch_neighbours(self, node: Contact) -> list[NeighbourInfo]:  # noqa: D102
+        mc = self._require()
+        pub = self._node_pubkey(node)
+        # The library pages through the table (one binary request per ~25 entries) and
+        # concatenates; ``min_timeout`` keeps slow multi-hop replies from being cut off
+        # at the companion's optimistic suggested timeout.
+        result = await mc.commands.fetch_all_neighbours(pub, min_timeout=20)
+        if result is None:
+            raise DeviceCommandError(
+                f"{node.name!r} did not answer the neighbour request. Firmware ignores "
+                "it without an admin login (log in first), and firmware older than "
+                "~v1.15 has no neighbour table at all."
+            )
+        now = utcnow()
+        neighbours: list[NeighbourInfo] = []
+        for entry in result.get("neighbours") or []:
+            pubkey = str(entry.get("pubkey") or "").lower()
+            if not pubkey:
+                continue
+            snr = entry.get("snr")
+            secs_ago = entry.get("secs_ago")
+            heard_at = None
+            if isinstance(secs_ago, (int, float)) and secs_ago >= 0:
+                heard_at = now - timedelta(seconds=float(secs_ago))
+            neighbours.append(
+                NeighbourInfo(
+                    node=pubkey,
+                    snr=float(snr) if snr is not None else None,
+                    heard_at=heard_at,
+                )
+            )
+        return neighbours
+
     async def run_trace(  # noqa: D102 - inherited docstring
         self,
         target: str,
@@ -1789,6 +1847,21 @@ class MockDevice(Device):
         self._admin_sessions: set[str] = set()
         self._remote_tx: dict[str, int] = {}
         self._default_remote_tx = 20
+        # Simulated neighbour tables, keyed by the repeater's key prefix: what each
+        # repeater "hears directly" as ``(neighbour_prefix, snr_db, secs_ago)``. The
+        # ``e5f6a7b8`` entry is deliberately absent from the contact list, so the
+        # fetched-evidence flow exercises discovering a node we never received from.
+        self._neighbour_tables: dict[str, list[tuple[str, float, int]]] = {
+            "a1b2c3d4": [
+                ("d4e5f6a7", 6.5, 300),
+                ("b2c3d4e5", -3.25, 1200),
+                ("e5f6a7b8", 2.0, 3600),
+            ],
+            "b2c3d4e5": [
+                ("c3d4e5f6", 8.0, 240),
+                ("a1b2c3d4", -3.25, 900),
+            ],
+        }
         # Mutable configuration state, keyed exactly like the real SELF_INFO payload so the
         # settings registry behaves identically on the simulator and on hardware.
         self._info: dict = {
@@ -1889,6 +1962,28 @@ class MockDevice(Device):
             )
         self._remote_tx[key] = value
 
+    async def fetch_neighbours(self, node: Contact) -> list[NeighbourInfo]:  # noqa: D102
+        await asyncio.sleep(0)
+        key = self._mock_key(node)
+        # Mirrors real firmware (verified on v1.15): without a login the request is
+        # silently dropped, which the caller experiences as a timeout.
+        if key not in self._admin_sessions:
+            raise DeviceCommandError(
+                f"{node.name!r} did not answer the neighbour request. Firmware ignores "
+                "it without an admin login (log in first)."
+            )
+        table = self._neighbour_tables.get(key[:8])
+        if table is None:
+            raise DeviceCommandError(
+                f"{node.name!r} did not answer the neighbour request (no neighbour "
+                "table on this node type)."
+            )
+        now = utcnow()
+        return [
+            NeighbourInfo(node=prefix, snr=snr, heard_at=now - timedelta(seconds=ago))
+            for prefix, snr, ago in table
+        ]
+
     @staticmethod
     def _mock_key(node: Contact) -> str:
         """Return the lookup key for a remote node (its public key, else key prefix)."""

meshterm/core/models.py

@@ -80,6 +80,28 @@ class Contact:
         return self.node_type == NODE_TYPE_REPEATER
 
 
+@dataclass(slots=True)
+class NeighbourInfo:
+    """One entry of a remote repeater's neighbour table, fetched over the mesh.
+
+    Repeater firmware keeps a table of the nodes it has *heard directly* (zero-hop),
+    each with the SNR measured at the repeater — a vantage point ours can't reproduce.
+    :meth:`~meshterm.core.connection.Device.fetch_neighbours` returns these; the values
+    describe the link ``repeater ↔ node`` as observed *at the repeater*.
+
+    Attributes:
+        node: The neighbour's hex public-key prefix, lowercased (the width the firmware
+            replied at — typically 8 hex).
+        snr: SNR (dB) the repeater measured receiving this neighbour, if reported.
+        heard_at: When the repeater last heard the neighbour (derived from the reply's
+            seconds-ago field), or ``None`` when it didn't say.
+    """
+
+    node: str
+    snr: Optional[float] = None
+    heard_at: Optional[datetime] = None
+
+
 @dataclass(slots=True)
 class Observation:
     """A single over-the-air reception heard while passively monitoring the mesh.

meshterm/persistence/db.py

@@ -10,7 +10,7 @@ from __future__ import annotations
 import sqlite3
 from pathlib import Path
 
-SCHEMA_VERSION = 7
+SCHEMA_VERSION = 8
 
 _SCHEMA = """
 CREATE TABLE IF NOT EXISTS schema_meta (
@@ -83,6 +83,19 @@ CREATE TABLE IF NOT EXISTS path_candidates (
     created_at    TEXT    NOT NULL
 );
 
+-- One entry of a remote repeater's neighbour table, fetched over the mesh (v8). Each row
+-- is a link the repeater reported hearing directly, with SNR measured at the repeater;
+-- refetching appends a new snapshot and readers take the latest row per pair.
+CREATE TABLE IF NOT EXISTS neighbour_reports (
+    id         INTEGER PRIMARY KEY AUTOINCREMENT,
+    run_id     INTEGER NOT NULL REFERENCES runs(id) ON DELETE CASCADE,
+    repeater   TEXT    NOT NULL,  -- canonical id of the repeater that was asked
+    neighbour  TEXT    NOT NULL,  -- reported neighbour's key-prefix hex, as replied
+    snr        REAL,              -- dB, measured at the repeater
+    heard_at   TEXT,              -- when the repeater last heard the neighbour
+    fetched_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,
@@ -120,6 +133,7 @@ CREATE TABLE IF NOT EXISTS messages (
 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_neighbour_reports_pair ON neighbour_reports(repeater, neighbour);
 CREATE INDEX IF NOT EXISTS idx_observations_run ON observations(run_id);
 CREATE INDEX IF NOT EXISTS idx_observations_node ON observations(node);
 CREATE INDEX IF NOT EXISTS idx_messages_peer ON messages(peer);

meshterm/persistence/repository.py

@@ -17,6 +17,7 @@ from ..core.models import (
     ChatMessage,
     HeardNode,
     Hop,
+    NeighbourInfo,
     Observation,
     TraceResult,
     TxLevelResult,
@@ -93,6 +94,29 @@ class PacketPath:
     hops: list[str]
 
 
+@dataclass(slots=True)
+class NeighbourLink:
+    """One repeater-reported direct link, as evidence for the topology graph.
+
+    The fetched counterpart of :class:`TracedPath`/:class:`PacketPath`: instead of being
+    inferred from what *we* received, this link was asserted by a remote repeater about
+    its own reception (see :meth:`Repository.neighbour_links`).
+
+    Attributes:
+        when: The link's recency — when the repeater last heard the neighbour, falling
+            back to when we fetched the table.
+        repeater: Canonical id of the repeater that reported the link.
+        neighbour: The neighbour's hex hash as the repeater replied it (any width; the
+            topology layer canonicalizes).
+        snr: SNR (dB) measured at the repeater, if reported.
+    """
+
+    when: datetime
+    repeater: str
+    neighbour: str
+    snr: Optional[float]
+
+
 @dataclass(slots=True)
 class RunRecord:
     """A summary row from the ``runs`` table.
@@ -403,6 +427,83 @@ class Repository:
             )
         return paths
 
+    def record_neighbours(
+        self, run_id: int, repeater: str, neighbours: list[NeighbourInfo]
+    ) -> None:
+        """Persist one fetched neighbour-table snapshot from a remote repeater.
+
+        Every entry becomes a row; refetching the same repeater later appends a fresh
+        snapshot rather than overwriting, and :meth:`neighbour_links` reads back only
+        the latest row per ``(repeater, neighbour)`` pair — a neighbour table is the
+        repeater's *current* state, so a new snapshot supersedes the old one instead
+        of stacking as extra evidence.
+
+        Args:
+            run_id: The owning run.
+            repeater: Canonical id of the repeater the table came from.
+            neighbours: The fetched entries (may be empty — recorded as no rows).
+        """
+        fetched_at = utcnow().isoformat()
+        self._conn.executemany(
+            "INSERT INTO neighbour_reports "
+            "(run_id, repeater, neighbour, snr, heard_at, fetched_at) "
+            "VALUES (?, ?, ?, ?, ?, ?)",
+            [
+                (
+                    run_id,
+                    repeater,
+                    n.node,
+                    n.snr,
+                    n.heard_at.isoformat() if n.heard_at is not None else None,
+                    fetched_at,
+                )
+                for n in neighbours
+            ],
+        )
+        self._conn.commit()
+
+    def neighbour_links(self, *, limit: int = 2000) -> list[NeighbourLink]:
+        """Return the current repeater-reported links, newest report first.
+
+        The fetched side of the topology evidence. Only the most recent row per
+        ``(repeater, neighbour)`` pair is returned (see :meth:`record_neighbours`), so
+        repeatedly refreshing a table never inflates a link's sample count.
+
+        Args:
+            limit: Maximum number of links to load.
+
+        Returns:
+            One :class:`NeighbourLink` per currently-reported link.
+        """
+        rows = self._conn.execute(
+            "SELECT r.repeater, r.neighbour, r.snr, r.heard_at, r.fetched_at "
+            "FROM neighbour_reports r "
+            "JOIN (SELECT MAX(id) AS id FROM neighbour_reports "
+            "      GROUP BY repeater, neighbour) latest ON latest.id = r.id "
+            "ORDER BY r.id DESC LIMIT ?",
+            (limit,),
+        ).fetchall()
+        links: list[NeighbourLink] = []
+        for row in rows:
+            when: Optional[datetime] = None
+            for stamp in (row["heard_at"], row["fetched_at"]):
+                try:
+                    when = datetime.fromisoformat(stamp)
+                    break
+                except (TypeError, ValueError):
+                    continue
+            if when is None:
+                continue  # a malformed stray contributes no evidence
+            links.append(
+                NeighbourLink(
+                    when=when,
+                    repeater=row["repeater"],
+                    neighbour=row["neighbour"],
+                    snr=row["snr"],
+                )
+            )
+        return links
+
     def record_tx_sample(self, run_id: int, level: TxLevelResult) -> None:
         """Persist one robust TX-power level from an optimization sweep.
 

meshterm/services/topology.py

@@ -4,9 +4,12 @@ MeshCore never tells us the network's shape outright, but almost everything we *
 carries a fragment of it: a successful trace demonstrably crossed every link in its path
 (out and back, with an SNR reading at each hop), the firmware's per-contact ``out_path``
 is a route distilled from received floods, and — when the companion's packet logging is
-on — every overheard frame reports the relay chain it rode in on. This module folds those
-fragments into one undirected evidence graph and answers the questions the trace path
-composer asks of it:
+on — every overheard frame reports the relay chain it rode in on. A fourth fragment can
+be *asked for*: a repeater we hold admin rights on reports its own neighbour table (who
+it hears directly, at what SNR) when queried over the mesh — a second vantage point that
+reveals links, and whole nodes, our radio has never received anything from. This module
+folds those fragments into one undirected evidence graph and answers the questions the
+trace path composer asks of it:
 
 * *who neighbours whom* — :meth:`MeshTopology.next_hops`, the suggestion list while
   composing a path hop by hop, sorted by the strongest observed link first;
@@ -34,7 +37,7 @@ from datetime import datetime
 from typing import Optional
 
 from ..core.models import Contact, utcnow
-from ..persistence.repository import PacketPath, TracedPath
+from ..persistence.repository import NeighbourLink, PacketPath, TracedPath
 
 #: Canonical id for our own device in the graph (its 12-hex key prefix when known).
 #: Kept distinct from hop hashes by construction: it is derived from the full public key.
@@ -96,7 +99,8 @@ class Link:
         samples: How many independent readings crossed this link, any source.
         snrs: The SNR readings (dB) attributed to the link, either direction.
         last_seen: When the link was most recently observed.
-        sources: Which evidence classes saw it (``trace`` / ``route`` / ``packet``).
+        sources: Which evidence classes saw it (``trace`` / ``route`` / ``packet`` /
+            ``neighbour``).
     """
 
     a: str
@@ -293,7 +297,8 @@ class MeshTopology:
             snrs: Optional per-link SNR readings aligned with the links (``snrs[i]``
                 belongs to the ``nodes[i] → nodes[i+1]`` link, measured at arrival).
             when: When the walk was observed (stamps every touched link's recency).
-            source: Evidence class tag (``trace`` / ``route`` / ``packet``).
+            source: Evidence class tag (``trace`` / ``route`` / ``packet`` /
+                ``neighbour``).
         """
         for i in range(len(nodes) - 1):
             a, b = nodes[i], nodes[i + 1]
@@ -485,10 +490,11 @@ def build_topology(
     contacts: list[Contact],
     trace_paths: list[TracedPath],
     packet_paths: list[PacketPath],
+    neighbour_links: tuple[NeighbourLink, ...] | list[NeighbourLink] = (),
 ) -> MeshTopology:
-    """Assemble the evidence graph from everything we have received.
+    """Assemble the evidence graph from everything we have received — or asked for.
 
-    Three sources fold in, each one walked through :meth:`MeshTopology.add_walk`:
+    Four sources fold in, each one walked through :meth:`MeshTopology.add_walk`:
 
     * **traces** — each successful trace's hop sequence, bracketed by us at both ends
       (the boomerang leaves us and returns to us; the final hash-less hop *is* us).
@@ -500,12 +506,18 @@ def build_topology(
     * **packets** — RX-logged frames: the relay chain, prefixed by the originator when
       known and always ending at us (we heard the last relay). The measured SNR belongs
       to that final link only.
+    * **neighbour reports** — links a remote repeater asserted about itself when we
+      fetched its neighbour table: one two-node walk per entry, SNR as measured at the
+      repeater. The only evidence here not derived from our own reception — it can
+      introduce nodes no other source has ever seen.
 
     Args:
         self_id: Our device's key/hash (any width; canonicalized to 12 hex).
         contacts: The device's known contacts (canonicalization + route evidence).
         trace_paths: Stored successful trace walks (see ``Repository.trace_paths``).
         packet_paths: Stored RX-logged packet paths (see ``Repository.packet_paths``).
+        neighbour_links: Current repeater-reported links (see
+            ``Repository.neighbour_links``); empty when none have been fetched.
 
     Returns:
         The populated :class:`MeshTopology`.
@@ -543,4 +555,12 @@ def build_topology(
         snrs = [None] * (len(nodes) - 2) + [packet.snr]
         topo.add_walk(nodes, snrs=snrs, when=packet.when, source="packet")
 
+    for reported in neighbour_links:
+        topo.add_walk(
+            [topo.canonical(reported.repeater), topo.canonical(reported.neighbour)],
+            snrs=[reported.snr],
+            when=reported.when,
+            source="neighbour",
+        )
+
     return topo

meshterm/tools/trace.py

@@ -3,10 +3,12 @@
 In the interactive menu this opens the live trace screen
 (:mod:`meshterm.ui.trace_screen`): pick a target from a recency-ordered list and the
 screen opens armed but idle — Enter runs a burst under an abortable tracing dialog, ``p``
-composes a forced path hop by hop from the topology observed in received traffic, and
-``x`` explores ranked path scenarios (with an optional probe that measures each candidate
-and offers the winner). On the CLI it stays a scriptable one-shot: run N traces, print
-the per-trace table and the aggregate summary, exit.
+composes a forced path hop by hop from the topology observed in received traffic (and,
+standing on a repeater with stored admin credentials, can fetch that repeater's live
+neighbour table for suggestions our own radio never heard), and ``x`` explores ranked
+path scenarios (with an optional probe that measures each candidate and offers the
+winner). On the CLI it stays a scriptable one-shot: run N traces, print the per-trace
+table and the aggregate summary, exit.
 
 Both front ends persist identically: one ``runs`` row per burst with every trace recorded
 under it, so stored history reads the same no matter where it came from.

meshterm/ui/path_composer.py

@@ -15,6 +15,11 @@ Interaction, following the reorder screen's cursor-over-rows-and-actions pattern
   even-length hex, an *add custom hop* row appears, so a node we have never observed
   (or a bare hash from another tool) can be forced into the route.
 * Backspace erases the filter first; with the filter empty it removes the last hop.
+* When the path's tail is a repeater we hold admin credentials for, a *fetch
+  neighbours* row asks that repeater over the mesh for its own neighbour table —
+  fresh second-vantage evidence exactly where composing ran out of it. The dialog
+  resolves :class:`FetchNeighbours` and the owning flow fetches, refreshes the
+  topology, and reopens the composer mid-thought (hops preserved).
 * Enter on **Use this path** commits the composed spec; **Auto** hands routing back to
   the device; Esc cancels with no change.
 
@@ -24,6 +29,7 @@ automatically, so the dialog shows the outbound route only and says so.
 
 from __future__ import annotations
 
+from dataclasses import dataclass
 from typing import Optional
 
 from rich.cells import cell_len
@@ -49,8 +55,24 @@ _AUTO = "auto"
 _CANCEL = "cancel"
 
 #: One-letter tags for the evidence classes backing a link, shown beside each
-#: suggestion: T(race), R(oute — the firmware's learned out_path), P(acket log).
-_SOURCE_TAGS = {"trace": "T", "route": "R", "packet": "P"}
+#: suggestion: T(race), R(oute — the firmware's learned out_path), P(acket log),
+#: N(eighbour table fetched from a repeater).
+_SOURCE_TAGS = {"trace": "T", "route": "R", "packet": "P", "neighbour": "N"}
+
+
+@dataclass(frozen=True, slots=True)
+class FetchNeighbours:
+    """The composer's resolution when the user asks a repeater for its neighbours.
+
+    The dialog itself never touches the radio; it resolves this marker and the owning
+    flow performs the login + fetch, refreshes the topology, and reopens the composer
+    with :attr:`PathComposerScreen.hops` re-seeded.
+
+    Attributes:
+        node: Canonical id of the repeater to query (the path's tail when committed).
+    """
+
+    node: str
 
 
 class PathComposerScreen(Screen):
@@ -74,6 +96,7 @@ class PathComposerScreen(Screen):
         topology: MeshTopology,
         width_bytes: int,
         hops: Optional[list[str]] = None,
+        fetch_nodes: frozenset[str] = frozenset(),
     ) -> None:
         """Build the composer.
 
@@ -88,6 +111,9 @@ class PathComposerScreen(Screen):
             width_bytes: Preferred per-hop path-hash width (bytes) for the emitted spec.
             hops: Canonical ids of already-composed intermediate hops (reopening the
                 dialog resumes where the user left off).
+            fetch_nodes: Canonical ids whose live neighbour table can be fetched
+                (repeater contacts with a public key); when the path's tail is one of
+                them, the *fetch neighbours* row appears.
         """
         super().__init__()
         self.title = f"compose path · {target_label}"
@@ -98,9 +124,15 @@ class PathComposerScreen(Screen):
         self._topology = topology
         self._width_bytes = width_bytes
         self._hops: list[str] = list(hops or [])
+        self._fetch_nodes = fetch_nodes
         self._entry = ""
         self._index = 0
 
+    @property
+    def hops(self) -> list[str]:
+        """The composed intermediate hops so far (for re-seeding after a fetch)."""
+        return list(self._hops)
+
     # --- state -----------------------------------------------------------------
 
     def _tail(self) -> str:
@@ -134,12 +166,17 @@ class PathComposerScreen(Screen):
         return needle if _is_hex(needle) else None
 
     def _rows(self) -> list[tuple[str, object]]:
-        """The cursor-addressable rows: custom hop, suggestions, then the actions."""
+        """The cursor-addressable rows: custom hop, suggestions, fetch, then actions."""
         rows: list[tuple[str, object]] = []
         custom = self._custom_hex()
         if custom:
             rows.append(("custom", custom))
         rows.extend(("hop", s) for s in self._suggestions())
+        # Standing on a repeater we hold credentials for, its live neighbour table is
+        # one keypress away — placed with the suggestions, because that is what it
+        # extends: "don't see the node you need? ask the repeater what it hears."
+        if self._tail() in self._fetch_nodes:
+            rows.append(("fetch", self._tail()))
         rows.append(("action", _USE))
         rows.append(("action", _AUTO))
         rows.append(("action", _CANCEL))
@@ -202,6 +239,12 @@ class PathComposerScreen(Screen):
             return text
         if kind == "hop":
             return self._suggestion_text(payload)
+        if kind == "fetch":
+            name = self._topology.display_name(str(payload)) or str(payload)[:12]
+            text = Text("⇣ Fetch neighbours from ", style="")
+            text.append(name, style="brand")
+            text.append("  (asks the repeater over the mesh)", style="muted")
+            return text
         if payload == _USE:
             label = Text.assemble(("✓ ", "ok"), "Use this path")
             label.append(f"  ({self._spec()})", style="muted")
@@ -255,7 +298,10 @@ class PathComposerScreen(Screen):
                 cursor_at = len(lines)
             lines.append(render_to_ansi(text, width))
         if not any(kind == "hop" for kind, _ in rows):
-            note = "(no observed links from here — type a hex hash to force a hop)"
+            if any(kind == "fetch" for kind, _ in rows):
+                note = "(no observed links from here — fetch the repeater's neighbours, or type a hex hash)"
+            else:
+                note = "(no observed links from here — type a hex hash to force a hop)"
             lines.append(render_to_ansi(Text(note, style="muted"), width))
 
         self._cursor = cursor_at
@@ -282,6 +328,8 @@ class PathComposerScreen(Screen):
             self._hops.append(payload.node)  # type: ignore[union-attr]
             self._entry = ""
             self._index = 0
+        elif kind == "fetch":
+            self.resolve(FetchNeighbours(node=str(payload)))
         elif payload == _USE:
             self.resolve(self._spec())
         elif payload == _AUTO:

meshterm/ui/trace_screen.py

@@ -11,7 +11,10 @@ whatever you make it, and nothing transmits until you say so. Three verbs drive
 * **p** opens the path composer (:mod:`~meshterm.ui.path_composer`): build the outbound
   route hop by hop, each step suggested from the links observed in *received* traffic —
   traces, firmware-learned contact routes, and RX-logged packet paths — strongest first,
-  with raw hex entry for nodes the data has never seen. Only the outbound leg is
+  with raw hex entry for nodes the data has never seen. Standing on a repeater you hold
+  admin credentials for, the composer can also *fetch that repeater's neighbour table*
+  over the mesh (login required; firmware ignores guests): second-vantage evidence,
+  persisted and folded straight back into the suggestions. Only the outbound leg is
   composed: the trace protocol replies back along the reversed path automatically.
 * **x** explores scenarios: ranked candidate routes to the target straight from the
   topology evidence (the device's own learned route, the direct shot, and the strongest
@@ -128,6 +131,9 @@ class TracingDialog(Screen):
         self.status = "transmitting…"
         #: The most recent trace result, echoed beneath the status line.
         self.last: Optional[TraceResult] = None
+        #: Whether to render the last-reply line. Trace/probe owners stream replies
+        #: through it; request-shaped owners (a neighbour fetch) have none to show.
+        self.show_last = True
 
     @property
     def dialog_width(self) -> int:
@@ -159,7 +165,8 @@ class TracingDialog(Screen):
         chip = self._spinner.text()
         chip.append(f"  {self.status}", style="")
         lines = render_lines(chip, width)
-        lines.extend(render_lines(self._last_line(), width))
+        if self.show_last:
+            lines.extend(render_lines(self._last_line(), width))
         lines.append("")
         lines.append(render_to_ansi(Text("❯ [ Abort ]", style="selected"), width))
         return lines
@@ -554,10 +561,11 @@ async def open_trace(ctx: "AppContext", target: str) -> int:
     Raises:
         RuntimeError: If called outside the interactive menu (no full-screen session).
     """
-    from ..core.models import LOCAL_DEVICE_LABEL
+    from ..core.connection import DeviceAuthenticationError
+    from ..core.models import LOCAL_DEVICE_LABEL, Contact, NeighbourInfo
     from ..services.path_probe import ProbeCandidate, ProbeOutcome, probe_paths
     from ..services.topology import MeshTopology, PathScenario, _is_hex, build_topology
-    from .path_composer import PathComposerScreen
+    from .path_composer import FetchNeighbours, PathComposerScreen
     from .surface import TuiUi
     from .tui import CANCEL, Choice, SelectScreen, Separator
 
@@ -609,6 +617,7 @@ async def open_trace(ctx: "AppContext", target: str) -> int:
             contacts=contacts,
             trace_paths=ctx.repo.trace_paths(),
             packet_paths=ctx.repo.packet_paths(),
+            neighbour_links=ctx.repo.neighbour_links(),
         )
 
     async def unaddressable() -> None:
@@ -621,8 +630,122 @@ async def open_trace(ctx: "AppContext", target: str) -> int:
             title="no destination hash",
         )
 
+    async def fetch_neighbours_via(repeater: Contact, repeater_id: str) -> bool:
+        """Log in to a repeater, fetch its neighbour table, and persist the snapshot.
+
+        The composer's fetch action lands here: password from the admin store (or a
+        one-time prompt, remembered on success — the tx-optimize convention), then the
+        login + fetch run under a floating spinner dialog with Abort. Every outcome
+        closes its own ``runs`` row; a rejected login also forgets the stored password
+        so the next attempt asks fresh.
+
+        Args:
+            repeater: The repeater contact to query (carries the public key).
+            repeater_id: Its canonical id, the key the snapshot is stored under.
+
+        Returns:
+            ``True`` when new neighbour links were recorded (the caller should rebuild
+            the topology); ``False`` on cancel, failure, or an empty table.
+        """
+        password = ctx.admin_store.get(repeater)
+        if password is None:
+            password = await session.text(
+                f"Admin password for {repeater.name}",
+                prompt="The repeater ignores neighbour requests without an admin login.",
+                password=True,
+            )
+            if not password:
+                return False
+        run_id = ctx.repo.start_run(
+            "trace",
+            {"mode": "neighbours", "repeater": repeater.name},
+            ctx.profile_name,
+        )
+        spinner = Spinner()
+        dialog = TracingDialog(
+            f"neighbours · {repeater.name}", spinner=spinner, on_abort=lambda: None
+        )
+        dialog.show_last = False  # a fetch has no streaming replies to echo
+        dialog.status = f"logging in to {repeater.name}…"
+
+        async def work() -> list[NeighbourInfo]:
+            if not await device.admin_login(repeater, password):
+                raise DeviceAuthenticationError(
+                    f"{repeater.name!r} rejected the admin login (wrong password?). "
+                    "The saved password was cleared; retry to enter a new one."
+                )
+            ctx.admin_store.remember(repeater, password)
+            dialog.status = "fetching the neighbour table…"
+            session.invalidate()
+            return await device.fetch_neighbours(repeater)
+
+        task = asyncio.ensure_future(work())
+        dialog.on_abort = task.cancel
+
+        async def animate() -> None:
+            while True:
+                await asyncio.sleep(_SPINNER_INTERVAL)
+                spinner.tick()
+                session.invalidate()
+
+        ticker = asyncio.ensure_future(animate())
+        session.push(dialog)
+        error: Optional[BaseException] = None
+        aborted = False
+        entries: list[NeighbourInfo] = []
+        try:
+            entries = await task
+        except asyncio.CancelledError:
+            aborted = True
+        except Exception as exc:  # noqa: BLE001 - surface in a dialog, keep composing
+            error = exc
+        finally:
+            ticker.cancel()
+            try:
+                await ticker
+            except asyncio.CancelledError:
+                pass
+            except Exception:  # noqa: BLE001 - a spinner hiccup must never break a fetch
+                pass
+            session.pop(dialog)
+        if aborted:
+            ctx.repo.finish_run(run_id, "error", {"error": "aborted"})
+            return False
+        if error is not None:
+            if isinstance(error, DeviceAuthenticationError):
+                ctx.admin_store.forget(repeater)  # bad password: don't keep reusing it
+            ctx.repo.finish_run(
+                run_id, "error", {"error": str(error) or type(error).__name__}
+            )
+            await session.message_dialog(
+                Text(str(error), style="err"), title="fetch neighbours"
+            )
+            return False
+        ctx.repo.record_neighbours(run_id, repeater_id, entries)
+        ctx.repo.finish_run(
+            run_id, "ok", {"repeater": repeater.name, "neighbours": len(entries)}
+        )
+        if not entries:
+            # Verified on real firmware: an empty table is a normal answer (repeaters
+            # forget neighbours across reboots and relearn them from adverts).
+            await session.message_dialog(
+                Text(
+                    f"{repeater.name} answered, but its neighbour table is empty — "
+                    "it relearns neighbours from received adverts, so ask again later.",
+                    style="muted",
+                ),
+                title="fetch neighbours",
+            )
+            return False
+        return True
+
     async def compose(current: str) -> Optional[str]:
-        """Open the hop-by-hop composer seeded with the current spec's hops."""
+        """Open the hop-by-hop composer seeded with the current spec's hops.
+
+        Runs the composer in a loop: a :class:`FetchNeighbours` resolution performs the
+        fetch, rebuilds the topology with the new evidence, and reopens the composer
+        exactly where the user stood (same hops, refreshed suggestions).
+        """
         if target_hash is None:
             await unaddressable()
             return None
@@ -634,8 +757,17 @@ async def open_trace(ctx: "AppContext", target: str) -> int:
             for h in [p for p in current.split(",") if p.strip()][:-1]
             if (cid := topo.canonical(h.strip())) is not None
         ]
-        result = await session.run_screen(
-            PathComposerScreen(
+        while True:
+            # Nodes whose neighbour table can be asked for: repeater contacts with a
+            # public key to log in against (our own node has nothing new to tell us).
+            fetchable: dict[str, Contact] = {}
+            for c in contacts:
+                if not c.is_repeater or not (c.public_key or "").strip():
+                    continue
+                cid = topo.canonical(c.public_key)
+                if cid is not None and cid != topo.self_id:
+                    fetchable[cid] = c
+            screen = PathComposerScreen(
                 target_id=target_id,
                 target_hash=target_hash,
                 target_label=target_label,
@@ -643,9 +775,20 @@ async def open_trace(ctx: "AppContext", target: str) -> int:
                 topology=topo,
                 width_bytes=width_bytes,
                 hops=seed,
+                fetch_nodes=frozenset(fetchable),
             )
-        )
-        return None if result is CANCEL else result
+            result = await session.run_screen(screen)
+            if result is CANCEL:
+                return None
+            if isinstance(result, FetchNeighbours):
+                seed = screen.hops  # resume mid-thought after the fetch
+                repeater = fetchable.get(result.node)
+                if repeater is not None and await fetch_neighbours_via(
+                    repeater, result.node
+                ):
+                    topo = fresh_topology()  # fold the new reports into suggestions
+                continue
+            return result
 
     def scenario_title(scenario: PathScenario, topo: MeshTopology) -> Text:
         """One scenario as a select row: source, route, and its observed evidence."""

tests/test_topology.py

@@ -10,11 +10,19 @@ from __future__ import annotations
 
 from datetime import timedelta
 
-from meshterm.core.models import Contact, Observation, utcnow
-from meshterm.persistence.repository import PacketPath, Repository, TracedPath
+import pytest
+
+from meshterm.core.connection import DeviceCommandError, MockDevice
+from meshterm.core.models import Contact, NeighbourInfo, Observation, utcnow
+from meshterm.persistence.repository import (
+    NeighbourLink,
+    PacketPath,
+    Repository,
+    TracedPath,
+)
 from meshterm.services.path_probe import ProbeCandidate, ProbeOutcome, probe_paths
 from meshterm.services.topology import build_topology, collapse_width
-from meshterm.ui.path_composer import AUTO_SPEC, PathComposerScreen
+from meshterm.ui.path_composer import AUTO_SPEC, FetchNeighbours, PathComposerScreen
 from meshterm.ui.tui.screen import CANCEL
 
 US = "aaaaaaaaaaaa"
@@ -23,12 +31,13 @@ FAR = Contact(name="Far", public_key="f2c24f54551e" + "0" * 52, key_prefix="f2c2
 LEAF = Contact(name="Leaf", public_key="27d4396a2967" + "0" * 52, key_prefix="27d4396a2967")
 
 
-def _topo(trace_paths=(), packet_paths=(), contacts=None):  # noqa: ANN001
+def _topo(trace_paths=(), packet_paths=(), neighbour_links=(), contacts=None):  # noqa: ANN001
     return build_topology(
         self_id=US + "0" * 52,
         contacts=list(contacts) if contacts is not None else [REPEATER, FAR, LEAF],
         trace_paths=list(trace_paths),
         packet_paths=list(packet_paths),
+        neighbour_links=list(neighbour_links),
     )
 
 
@@ -82,6 +91,23 @@ def test_contact_routes_and_packet_paths_feed_the_graph() -> None:
     assert packet_link.snrs == []
 
 
+def test_neighbour_reports_feed_the_graph_as_fetched_evidence() -> None:
+    """A repeater's fetched table adds links at its vantage point, tagged ``neighbour``."""
+    reported = [
+        NeighbourLink(when=utcnow(), repeater="3d63c6429436", neighbour="f2c2", snr=7.0),
+        NeighbourLink(when=utcnow(), repeater="3d63c6429436", neighbour="beefbeef", snr=None),
+    ]
+    topo = _topo(neighbour_links=reported)
+    link = topo.link("3d63c6429436", "f2c24f54551e")  # 'f2c2' canonicalizes onto Far
+    assert link is not None and link.sources == {"neighbour"}
+    assert link.snrs == [7.0]  # the SNR measured at the repeater rides the link
+    # A neighbour no contact matches still counts — discovery of an unseen node.
+    unknown = topo.link("3d63c6429436", "beefbeef")
+    assert unknown is not None and unknown.snrs == []
+    suggested = [s.node for s in topo.next_hops("3d63c6429436")]
+    assert "f2c24f54551e" in suggested and "beefbeef" in suggested
+
+
 # --- suggestions and scenarios ----------------------------------------------------
 
 
@@ -152,6 +178,54 @@ def test_repository_round_trips_packet_paths_and_candidates(tmp_path) -> None:
     repo.close()
 
 
+def test_repository_neighbour_snapshots_keep_latest_per_pair(tmp_path) -> None:  # noqa: ANN001
+    """A refetched table supersedes the old snapshot instead of stacking as evidence."""
+    repo = Repository(tmp_path / "n.db")
+    run = repo.start_run("trace", {"mode": "neighbours"})
+    heard = utcnow() - timedelta(hours=2)
+    repo.record_neighbours(
+        run,
+        "3d63c6429436",
+        [
+            NeighbourInfo(node="f2c24f54", snr=-5.0, heard_at=heard),
+            NeighbourInfo(node="beefbeef", snr=2.0, heard_at=None),
+        ],
+    )
+    links = {link.neighbour: link for link in repo.neighbour_links()}
+    assert len(links) == 2
+    assert links["f2c24f54"].snr == -5.0
+    assert links["f2c24f54"].when == heard  # the repeater's own recency survives
+    assert links["beefbeef"].when is not None  # no heard_at → falls back to fetched_at
+
+    repo.record_neighbours(run, "3d63c6429436", [NeighbourInfo(node="f2c24f54", snr=9.0)])
+    links = {link.neighbour: link for link in repo.neighbour_links()}
+    assert len(links) == 2  # still one link per pair
+    assert links["f2c24f54"].snr == 9.0  # the fresh snapshot won
+    repo.close()
+
+
+async def test_mock_fetch_neighbours_is_login_gated_like_real_firmware() -> None:
+    """No login → ignored (error); logged in → the table, including an unseen node."""
+    device = MockDevice()
+    await device.connect()
+    contacts = await device.get_contacts()
+    yagi = next(c for c in contacts if c.name == "Yagi-Repeater")
+
+    with pytest.raises(DeviceCommandError):
+        await device.fetch_neighbours(yagi)  # mirrors v1.15 firmware: guests are ignored
+
+    assert await device.admin_login(yagi, "admin")
+    entries = await device.fetch_neighbours(yagi)
+    nodes = {e.node for e in entries}
+    assert "e5f6a7b8" in nodes  # a node absent from the contact list: pure discovery
+    assert all(e.heard_at is not None for e in entries)
+
+    alice = next(c for c in contacts if c.name == "Alice")
+    assert await device.admin_login(alice, "admin")
+    with pytest.raises(DeviceCommandError):
+        await device.fetch_neighbours(alice)  # chat nodes keep no neighbour table
+
+
 # --- path probe ----------------------------------------------------------------------
 
 
@@ -180,7 +254,7 @@ async def test_probe_paths_ranks_reliability_first() -> None:
 # --- path composer ---------------------------------------------------------------------
 
 
-def _composer(topo, hops=None):  # noqa: ANN001
+def _composer(topo, hops=None, fetch_nodes=frozenset()):  # noqa: ANN001
     return PathComposerScreen(
         target_id="f2c24f54551e",
         target_hash="f2c24f54551e" + "0" * 52,
@@ -189,6 +263,7 @@ def _composer(topo, hops=None):  # noqa: ANN001
         topology=topo,
         width_bytes=1,
         hops=list(hops or []),
+        fetch_nodes=fetch_nodes,
     )
 
 
@@ -219,6 +294,28 @@ def test_composer_typed_hex_adds_a_custom_hop_and_backspace_removes() -> None:
     assert screen._hops == []
 
 
+async def test_composer_fetch_row_resolves_a_fetch_request() -> None:
+    """Standing on a fetchable repeater, Enter on the fetch row hands off to the owner."""
+    import asyncio
+
+    screen = _composer(
+        _topo(), hops=["3d63c6429436"], fetch_nodes=frozenset({"3d63c6429436"})
+    )
+    screen.future = asyncio.get_running_loop().create_future()
+    body = _rows_plain(screen)
+    assert "Fetch neighbours from" in body and "Hub" in body
+    screen.handle("enter")  # an empty graph offers no suggestions: fetch is the top row
+    result = screen.future.result()
+    assert isinstance(result, FetchNeighbours) and result.node == "3d63c6429436"
+    assert screen.hops == ["3d63c6429436"]  # preserved, so the owner can reopen mid-path
+
+
+def test_composer_hides_fetch_row_off_fetchable_tails() -> None:
+    """The fetch row tracks the path's tail: at our own node there is nothing to ask."""
+    screen = _composer(_topo(), fetch_nodes=frozenset({"3d63c6429436"}))
+    assert "Fetch neighbours" not in _rows_plain(screen)  # tail is us, not the repeater
+
+
 async def test_composer_commits_spec_auto_and_cancel() -> None:
     """Use resolves the target-terminated spec; Auto resolves the empty spec; Esc cancels."""
     import asyncio