feat(mqtt): periodic neighbours publication with serialized scope discovery

Python port of the firmware's WITH_MQTT_NEIGHBORS feature (MeshCore PR #35 plus
the aba571ed pacing rework). Each cycle refreshes the zero-hop neighbour table
with a node-discovery broadcast, asks every neighbour for its region scopes, and
publishes the assembled table to the MQTT neighbors topic.

- NeighborScopeHelper: client side of the anon-regions request (the server side
  already existed in openhop_core). Queries are strictly serialized -- firing
  them as a burst makes the responses collide, which is what the firmware fix
  addressed. router.inject_packet resolves at the firmware's logTx/logTxFail
  boundary, so the response deadline is armed only once the request is on air
  and the firmware's QUEUED/PENDING state machine collapses into an await.
  The response window is sized from radio parameters, mirroring
  neighborDiscoverQueryTimeoutMs().
- NeighborsPublisher: owns the schedule, the immutable per-cycle snapshot, and
  the payload. The snapshot merges live discovery responses over the stored
  table because get_neighbors() serves a 60s cache that store_advert() does not
  invalidate.
- packet_router: offer PAYLOAD_TYPE_RESPONSE to the scope helper before the
  companion fan-out; it consumes only on an authenticated tag match.
- Per-broker `neighbors: true` opt-in (default off) plus a
  mqtt_brokers.neighbors.enabled master switch; interval is 12-336h as in
  firmware, rejected rather than clamped.
- `discover.scopes` mesh CLI command triggers one cycle immediately.

The whole zero-hop table is published including neighbours that did not answer
(timeout / send_failed), matching the firmware payload. No 10 KB cap: that
existed for a fixed PSRAM buffer.

config.yaml.example

@@ -545,6 +545,26 @@ mqtt_brokers:
   email: ""
   brokers: []
 
+  # Periodic neighbours publication (the "neighbors" topic).
+  #
+  # Each cycle runs a zero-hop node-discovery broadcast to refresh the neighbour
+  # table, then asks each zero-hop repeater for its region scopes with an
+  # anonymous regions request, then publishes the assembled table. Scope queries
+  # are issued ONE AT A TIME - firing them as a burst makes the responses collide.
+  #
+  # Off unless a broker opts in below with `neighbors: true`; `enabled` here is a
+  # master kill switch for the whole feature.
+  # neighbors:
+  #   enabled: true                     # master on/off (default true)
+  #   interval_hours: 24                # 12-336, how often a cycle runs
+  #   discovery_timeout_seconds: 60     # node-discovery collection window
+  #   scope_response_timeout_seconds: 0 # 0 = size the window from radio settings
+  #   max_neighbors: 32                 # cap on neighbours queried per cycle
+  #   max_neighbor_age_seconds: 86400   # ignore zero-hop rows older than this
+  #   max_sweep_seconds: 900            # give up on a cycle after this long
+  #
+  # `discover.scopes` over the mesh CLI runs one cycle immediately.
+
   # Below is the broker object schema:
   # enabled: true|false # Enable this specific mqtt broker
   # name: "" # Internal name for this broker
@@ -560,6 +580,7 @@ mqtt_brokers:
   #   letsmesh, waev - MC2MQTT family flavors (same topic structure, network-specific identity)
   #   mqtt           - legacy openhop-repeater local-broker convention (custom topic, singular 'packet')
   # retain_status: true|false # Sets MQTT "retain" on status messages so they remain on the broker when disconnected. Also enforces a QOS of 1 (guaranteed delivery)
+  # neighbors: true|false # Publish the periodic neighbours table to this broker's "neighbors" topic (default false). Only enable for brokers that expect the topic - some reject unknown topics and close the connection.
   # tls:
   #   enabled: true|false # Enable TLS. If the endpoint's certificate is self-signed, the Root CA should be added to the OS's certificate store. 
   #   insecure: true|false # Validate TLS certificates

repeater/data_acquisition/mqtt_handler.py

@@ -239,6 +239,12 @@ class _BrokerConnection:
 
         self.enabled = broker.get("enabled", False)
         self.retain_status = broker.get("retain_status", False)
+        # Opt-in per broker, default off. The neighbors topic is not part of every
+        # MC2MQTT deployment's contract, and a broker that rejects an unexpected
+        # topic closes the connection (rc=16) - see the format auto-correction
+        # below for the same hazard. This is the openhop equivalent of the
+        # firmware's packets-only MeshRank exclusion.
+        self.neighbors_enabled = bool(broker.get("neighbors", False))
 
         self._tls_verified = False
 
@@ -1016,6 +1022,50 @@ class MeshCoreToMqttPusher:
 
         return results
 
+    def has_neighbors_brokers(self) -> bool:
+        """True when at least one enabled broker opted into the neighbors topic."""
+        return any(conn.enabled and conn.neighbors_enabled for conn in self.connections)
+
+    def has_connected_neighbors_brokers(self) -> bool:
+        """True when an opted-in broker is connected and could receive a publish."""
+        return any(
+            conn.enabled and conn.neighbors_enabled and conn.is_connected()
+            for conn in self.connections
+        )
+
+    def publish_neighbors(self, payload: dict):
+        """Publish the neighbours table to every broker that opted in.
+
+        QoS 1 and non-retained, matching the firmware's ``publishNeighbors()``:
+        the table is a periodic snapshot, so a stale retained copy would outlive
+        its usefulness, but it is infrequent enough to be worth delivering once.
+        """
+        message = json.dumps(payload)
+        neighbor_count = len(payload.get("neighbors", []))
+        logger.debug(
+            f"Publishing topic='neighbors', neighbors={neighbor_count}, "
+            f"bytes={len(message.encode('utf-8'))}"
+        )
+
+        results = []
+        with self._lock:
+            for conn in self.connections:
+                if not (conn.enabled and conn.neighbors_enabled):
+                    continue
+                if not conn.is_connected():
+                    logger.warning(
+                        f"Cannot publish neighbors to {conn.broker['name']} - not connected"
+                    )
+                    continue
+                result = conn.publish("neighbors", message, retain=False, qos=1)
+                results.append((conn.broker["name"], result))
+                _trace(f"Published to {conn.broker['name']} -- neighbors")
+
+        if not results:
+            logger.warning("No connected broker opted into the neighbors topic")
+
+        return results
+
     def publish_mqtt(self, payload: dict, subtopic: str, retain: bool = False, qos: int = 0):
         """Publish message to brokers using the legacy custom-MQTT format only.
 

repeater/handler_helpers/__init__.py

@@ -3,6 +3,7 @@
 from .advert import AdvertHelper
 from .discovery import DiscoveryHelper
 from .login import LoginHelper
+from .neighbor_scopes import NeighborScopeHelper
 from .path import PathHelper
 from .protocol_request import ProtocolRequestHelper
 from .text import TextHelper
@@ -13,6 +14,7 @@ __all__ = [
     "DiscoveryHelper",
     "AdvertHelper",
     "LoginHelper",
+    "NeighborScopeHelper",
     "TextHelper",
     "PathHelper",
     "ProtocolRequestHelper",

repeater/handler_helpers/mesh_cli.py

@@ -247,6 +247,8 @@ class MeshCLI:
             return self._cmd_neighbors()
         elif command.startswith("neighbor.remove "):
             return self._cmd_neighbor_remove(command)
+        elif command.startswith("discover.scopes"):
+            return self._cmd_discover_scopes(command)
         elif command.startswith("discover.neighbors"):
             return self._cmd_discover_neighbors(command)
 
@@ -326,6 +328,7 @@ class MeshCLI:
             "  neighbors           List neighbors",
             "  neighbor.remove <key>  Remove neighbor by pubkey",
             "  discover.neighbors  Send zero-hop neighbor discovery",
+            "  discover.scopes     Discover neighbor scopes, publish to MQTT",
             "  tempradio <freq> <bw> <sf> <cr> <timeout_mins>",
             "  setperm <pubkey> <perm>  Set ACL permissions",
             "  log start|stop|erase    Logging control",
@@ -380,6 +383,12 @@ class MeshCLI:
             ),
             "neighbors": "List known neighbor nodes from the routing table.",
             "discover.neighbors": "Send a neighbor discovery request.",
+            "discover.scopes": (
+                "discover.scopes\n"
+                "  Refresh the zero-hop neighbor table, query each neighbor for its\n"
+                "  region scopes, then publish the table to the MQTT neighbors topic.\n"
+                "  Requires a broker configured with neighbors: true."
+            ),
             "setperm": "setperm <pubkey_hex> <permission_int> \u2014 Set ACL permissions for a node.",
             "log": "log start|stop|erase \u2014 Control logging.",
         }
@@ -1295,6 +1304,50 @@ class MeshCLI:
             logger.error(f"discover.neighbors failed: {e}", exc_info=True)
             return f"Error: {e}"
 
+    def _cmd_discover_scopes(self, command: str) -> str:
+        """Refresh the neighbour table, query each neighbour's scopes, publish once.
+
+        Manual trigger for the periodic MQTT neighbors cycle (firmware
+        ``discover.scopes``). The cycle takes minutes -- a node-discovery window
+        plus one serialized scope query per neighbour -- so this schedules it and
+        returns immediately rather than holding the CLI reply open.
+        """
+        sub = command[15:]
+        if sub.strip():
+            return "Err - discover.scopes has no options"
+
+        daemon_instance = getattr(self.config_manager, "daemon", None)
+        publisher = getattr(daemon_instance, "neighbors_publisher", None)
+        if not publisher:
+            return "Error: Neighbors publisher not available"
+
+        if not publisher.enabled():
+            return "Err - neighbors publishing is disabled (no broker opted in)"
+
+        if publisher.status().get("phase") == "active":
+            return "Err - neighbors cycle already active"
+
+        import asyncio
+
+        loop = self._event_loop
+        if loop is None:
+            try:
+                loop = asyncio.get_running_loop()
+            except RuntimeError:
+                loop = None
+
+        if loop is None or not loop.is_running():
+            return "Error: Event loop not available"
+
+        try:
+            loop.call_soon_threadsafe(
+                lambda: asyncio.create_task(publisher.run_cycle(trigger="manual"))
+            )
+            return "OK - neighbor scope discovery started"
+        except Exception as e:
+            logger.error(f"discover.scopes failed: {e}", exc_info=True)
+            return f"Error: {e}"
+
     # ==================== Temporary Radio Commands ====================
 
     def _cmd_tempradio(self, command: str) -> str:

repeater/handler_helpers/neighbor_scopes.py

@@ -0,0 +1,403 @@
+"""Neighbour scope-query helper for openHop Repeater.
+
+Client side of the anonymous *regions* request. The server side already lives in
+:class:`~openhop_core.node.handlers.anon_request.AnonRequestHandler` (wired up by
+:mod:`repeater.handler_helpers.login`); this module asks the question instead of
+answering it, so the MQTT ``neighbors`` topic can report which region scopes each
+zero-hop neighbour serves.
+
+Wire format mirrors firmware ``MyMesh::sendAnonRegionsReq``: a PAYLOAD_TYPE_ANON_REQ
+whose plaintext is ``tag(4) + ANON_REQ_TYPE_REGIONS + 0x00``, where the trailing
+``0x00`` asks for a zero-hop reply path. The reply is a PAYLOAD_TYPE_RESPONSE
+datagram (``dest_hash(1) + src_hash(1) + cipher``) whose plaintext is
+``tag(4) + clock(4) + comma_separated_scope_names``.
+
+Two firmware behaviours are load-bearing and are reproduced here:
+
+* The request must be **route-direct** — ``AnonRequestHandler`` (and the firmware
+  regions handler it mirrors) ignores flooded discovery sub-types outright.
+* Queries are issued **strictly one at a time**. Firing them as a burst makes every
+  responder answer into the same window and the replies collide; the firmware fixed
+  this by walking the snapshot one entry at a time and only arming the response
+  deadline once the request has actually transmitted (MeshCore ``aba571ed``).
+  ``router.inject_packet`` already resolves at that exact point — it awaits
+  ``dispatcher.send_packet`` under the engine TX lock — so the firmware's
+  QUEUED/PENDING state machine collapses into a sequential ``await`` here.
+"""
+
+import asyncio
+import logging
+import time
+from dataclasses import dataclass, field
+from typing import Callable, Dict, List, Optional
+
+from openhop_core.protocol import CryptoUtils, Identity
+from openhop_core.protocol.constants import (
+    ANON_REQ_TYPE_REGIONS,
+    MAX_PACKET_PAYLOAD,
+    PAYLOAD_TYPE_RESPONSE,
+)
+
+logger = logging.getLogger("NeighborScopes")
+
+# Per-neighbour outcome, published verbatim in the MQTT neighbors payload.
+STATUS_RESPONDED = "responded"
+STATUS_TIMEOUT = "timeout"
+STATUS_SEND_FAILED = "send_failed"
+
+# Responder-side reply delay to budget for. Firmware uses SERVER_RESPONSE_DELAY
+# (500 ms); openhop_core's AnonRequestHandler uses 300 ms. Budget the larger so a
+# firmware neighbour is not written off early.
+SERVER_RESPONSE_DELAY_MS = 500.0
+
+# Slack for scheduler jitter, matching the firmware's own 360 ms constant.
+RESPONSE_TIMEOUT_SLACK_MS = 360.0
+
+# Bounds for the auto-sized response timeout. The floor keeps a fast radio config
+# from producing an unreachably tight window; the ceiling keeps a slow one (SF12,
+# narrow bandwidth) from stalling a whole sweep on one dead neighbour.
+MIN_RESPONSE_TIMEOUT_SEC = 5.0
+MAX_RESPONSE_TIMEOUT_SEC = 120.0
+
+# Used when no AirtimeManager is available to size the window from radio params.
+FALLBACK_RESPONSE_TIMEOUT_SEC = 30.0
+
+# Whole-sweep budget, and how long a duty-cycle backlog may be before the sweep
+# gives up rather than queueing behind forwarding traffic for minutes.
+DEFAULT_MAX_SWEEP_SECONDS = 900.0
+DEFAULT_DUTY_CYCLE_ABORT_SECONDS = 30.0
+
+
+@dataclass(frozen=True)
+class NeighborSnapshot:
+    """One neighbour, frozen at sweep start.
+
+    Firmware originally indexed into the live ``neighbours[]`` table and had to
+    stop doing that (``aba571ed``) because an advert arriving mid-pass could
+    reshuffle it. Same reasoning here: the sqlite neighbour table keeps changing
+    while a multi-minute sweep runs, so what gets published is decided up front.
+    """
+
+    pubkey: str
+    last_seen: float = 0.0
+    snr: float = 0.0
+
+    @property
+    def pubkey_bytes(self) -> bytes:
+        return bytes.fromhex(self.pubkey)
+
+
+@dataclass
+class ScopeResult:
+    status: str
+    scopes: str = ""
+
+
+@dataclass
+class _PendingQuery:
+    pubkey: str
+    tag: int
+    src_hash: int
+    future: "asyncio.Future[str]" = field(repr=False, default=None)
+
+
+class _ScopeTarget:
+    """Minimal contact stand-in for ``PacketBuilder.create_anon_request``.
+
+    ``out_path_len = 0`` (rather than -1) selects direct routing with an empty
+    path, i.e. the zero-hop request the responder's route-direct gate requires.
+    """
+
+    def __init__(self, pubkey_hex: str):
+        self.public_key = pubkey_hex
+        self.out_path_len = 0
+        self.out_path = b""
+
+
+class NeighborScopeHelper:
+    """Issues anon-regions queries and matches their responses."""
+
+    def __init__(
+        self,
+        local_identity,
+        packet_injector: Optional[Callable] = None,
+        airtime_manager=None,
+        config: Optional[dict] = None,
+    ):
+        self.local_identity = local_identity
+        self.packet_injector = packet_injector
+        self.airtime_manager = airtime_manager
+
+        self._pending: Optional[_PendingQuery] = None
+        self._sweep_lock = asyncio.Lock()
+
+        self._response_timeout_override = 0.0
+        self._max_sweep_seconds = DEFAULT_MAX_SWEEP_SECONDS
+        self._duty_cycle_abort_seconds = DEFAULT_DUTY_CYCLE_ABORT_SECONDS
+        self._direct_tx_delay_factor = 0.5
+        self.refresh_config(config or {})
+
+    # ------------------------------------------------------------------
+    # Configuration
+    # ------------------------------------------------------------------
+    def refresh_config(self, config: dict) -> None:
+        """Re-read the tunables so a live config update takes effect next sweep."""
+        neighbors_cfg = (config.get("mqtt_brokers", {}) or {}).get("neighbors", {}) or {}
+
+        self._response_timeout_override = _as_float(
+            neighbors_cfg.get("scope_response_timeout_seconds", 0), 0.0
+        )
+        self._max_sweep_seconds = max(
+            1.0, _as_float(neighbors_cfg.get("max_sweep_seconds"), DEFAULT_MAX_SWEEP_SECONDS)
+        )
+        self._duty_cycle_abort_seconds = max(
+            0.0,
+            _as_float(
+                neighbors_cfg.get("duty_cycle_abort_seconds"), DEFAULT_DUTY_CYCLE_ABORT_SECONDS
+            ),
+        )
+        self._direct_tx_delay_factor = _as_float(
+            (config.get("delays", {}) or {}).get("direct_tx_delay_factor"), 0.5
+        )
+
+    def response_timeout(self) -> float:
+        """How long to wait for a reply once the request is on air.
+
+        Mirrors firmware ``neighborDiscoverQueryTimeoutMs()``: the responder's
+        fixed reply delay, plus however long it may defer the transmission, plus
+        airtime for one packet ahead of the response and the response itself.
+        The firmware term there is ``getCADFailMaxDuration()``; openhop's
+        equivalent deferral is the engine's random direct-TX window,
+        ``[0, 5 * airtime * direct_tx_delay_factor]``.
+        """
+        if self._response_timeout_override > 0:
+            return self._response_timeout_override
+
+        airtime_ms = self._estimate_response_airtime_ms()
+        if airtime_ms <= 0:
+            return FALLBACK_RESPONSE_TIMEOUT_SEC
+
+        deferral_ms = 5.0 * airtime_ms * max(0.0, self._direct_tx_delay_factor)
+        total_ms = (
+            SERVER_RESPONSE_DELAY_MS + deferral_ms + (2.0 * airtime_ms) + RESPONSE_TIMEOUT_SLACK_MS
+        )
+        return min(MAX_RESPONSE_TIMEOUT_SEC, max(MIN_RESPONSE_TIMEOUT_SEC, total_ms / 1000.0))
+
+    def _estimate_response_airtime_ms(self) -> float:
+        if not self.airtime_manager:
+            return 0.0
+        try:
+            # Worst-case response size, matching the firmware's
+            # getEstAirtimeFor(MAX_PACKET_PAYLOAD + 2).
+            return float(self.airtime_manager.calculate_airtime(MAX_PACKET_PAYLOAD + 2))
+        except Exception as e:
+            logger.debug(f"Could not estimate response airtime: {e}")
+            return 0.0
+
+    # ------------------------------------------------------------------
+    # Sweep
+    # ------------------------------------------------------------------
+    @property
+    def active(self) -> bool:
+        return self._sweep_lock.locked()
+
+    async def sweep(self, targets: List[NeighborSnapshot]) -> Dict[str, ScopeResult]:
+        """Query every target's scopes, one request in flight at a time.
+
+        Returns a result per target keyed by lowercase pubkey hex. Targets that
+        are never reached (sweep budget or duty-cycle backlog) come back as
+        ``timeout``, matching the firmware's treatment of unsent entries.
+        """
+        results: Dict[str, ScopeResult] = {}
+        if not targets:
+            return results
+
+        if self._sweep_lock.locked():
+            raise RuntimeError("neighbor scope sweep already active")
+
+        async with self._sweep_lock:
+            deadline = time.monotonic() + self._max_sweep_seconds
+            timeout = self.response_timeout()
+            logger.info(
+                "Scope sweep starting: %d neighbour(s), %.1fs response window, %.0fs sweep budget",
+                len(targets),
+                timeout,
+                self._max_sweep_seconds,
+            )
+
+            abandoned = False
+            for target in targets:
+                if abandoned or time.monotonic() >= deadline:
+                    if not abandoned:
+                        logger.warning(
+                            "Scope sweep budget exhausted; %s and later marked timeout",
+                            target.pubkey[:8],
+                        )
+                        abandoned = True
+                    results[target.pubkey] = ScopeResult(STATUS_TIMEOUT)
+                    continue
+
+                result = await self._query_one(target, timeout)
+                results[target.pubkey] = result
+
+                if result.status == STATUS_SEND_FAILED and self._duty_cycle_backlogged():
+                    # The airtime budget is gone; the rest of the sweep would sit
+                    # behind forwarding traffic. Publish what resolved instead.
+                    logger.warning("Scope sweep abandoned: duty-cycle budget exhausted")
+                    abandoned = True
+
+            responded = sum(1 for r in results.values() if r.status == STATUS_RESPONDED)
+            logger.info("Scope sweep complete: %d/%d responded", responded, len(results))
+            return results
+
+    async def _query_one(self, target: NeighborSnapshot, timeout: float) -> ScopeResult:
+        if not self.packet_injector:
+            logger.warning("No packet injector available - cannot query neighbour scopes")
+            return ScopeResult(STATUS_SEND_FAILED)
+
+        try:
+            packet, tag = self._build_request(target)
+        except Exception as e:
+            logger.warning(f"Could not build scope request for {target.pubkey[:8]}: {e}")
+            return ScopeResult(STATUS_SEND_FAILED)
+
+        if not self._duty_cycle_allows(packet):
+            logger.warning(
+                "Skipping scope query for %s: duty-cycle budget exhausted", target.pubkey[:8]
+            )
+            return ScopeResult(STATUS_SEND_FAILED)
+
+        loop = asyncio.get_running_loop()
+        pending = _PendingQuery(
+            pubkey=target.pubkey,
+            tag=tag,
+            src_hash=target.pubkey_bytes[0],
+            future=loop.create_future(),
+        )
+        self._pending = pending
+
+        try:
+            # Resolves only once the packet is actually on air (or has failed):
+            # the engine awaits dispatcher.send_packet under its TX lock and
+            # defers local TX until the duty cycle allows. This is the firmware's
+            # logTx / logTxFail boundary, which is where the response deadline is
+            # armed -- hence the wait_for below, and not a moment earlier.
+            sent = await self.packet_injector(packet, wait_for_ack=False)
+        except Exception as e:
+            self._pending = None
+            logger.warning(f"Scope request send failed for {target.pubkey[:8]}: {e}")
+            return ScopeResult(STATUS_SEND_FAILED)
+
+        if not sent:
+            self._pending = None
+            logger.debug(f"Scope request not transmitted for {target.pubkey[:8]}")
+            return ScopeResult(STATUS_SEND_FAILED)
+
+        try:
+            scopes = await asyncio.wait_for(pending.future, timeout)
+            logger.debug(f"Scope response from {target.pubkey[:8]}: '{scopes}'")
+            return ScopeResult(STATUS_RESPONDED, scopes)
+        except asyncio.TimeoutError:
+            logger.debug(f"Scope query timed out for {target.pubkey[:8]}")
+            return ScopeResult(STATUS_TIMEOUT)
+        finally:
+            self._pending = None
+
+    def _build_request(self, target: NeighborSnapshot):
+        from openhop_core.protocol.packet_builder import PacketBuilder
+
+        # tag(4) is prepended by create_anon_request as the request timestamp;
+        # ANON_REQ_TYPE_REGIONS selects the scopes reply and 0x00 asks for a
+        # zero-hop reply path (firmware inner[4], inner[5]).
+        return PacketBuilder.create_anon_request(
+            _ScopeTarget(target.pubkey),
+            self.local_identity,
+            req_data=bytes([ANON_REQ_TYPE_REGIONS, 0x00]),
+        )
+
+    # ------------------------------------------------------------------
+    # Duty cycle
+    # ------------------------------------------------------------------
+    def _duty_cycle_allows(self, packet) -> bool:
+        """Pre-flight the airtime budget.
+
+        Stands in for the firmware's ``getFreeCount() >= 5`` packet-pool guard:
+        different scarce resource, same job of refusing to enqueue a request the
+        transmit path cannot honour. A short backlog is fine — the engine defers
+        local TX on its own — so only a long one is treated as a failure.
+        """
+        if not self.airtime_manager:
+            return True
+        try:
+            airtime_ms = self.airtime_manager.calculate_airtime(packet.get_raw_length())
+            can_tx, wait_time = self.airtime_manager.can_transmit(airtime_ms)
+        except Exception as e:
+            logger.debug(f"Duty-cycle pre-flight failed, allowing send: {e}")
+            return True
+        return can_tx or wait_time <= self._duty_cycle_abort_seconds
+
+    def _duty_cycle_backlogged(self) -> bool:
+        if not self.airtime_manager:
+            return False
+        try:
+            airtime_ms = self.airtime_manager.calculate_airtime(MAX_PACKET_PAYLOAD)
+            can_tx, wait_time = self.airtime_manager.can_transmit(airtime_ms)
+        except Exception:
+            return False
+        return not can_tx and wait_time > self._duty_cycle_abort_seconds
+
+    # ------------------------------------------------------------------
+    # Response matching
+    # ------------------------------------------------------------------
+    async def process_response_packet(self, packet) -> bool:
+        """Consume a PAYLOAD_TYPE_RESPONSE that answers the in-flight query.
+
+        Returns True only when the packet decrypts under the pending neighbour's
+        shared secret *and* echoes its tag, so an unrelated response still falls
+        through to the companion bridges. Like firmware
+        ``handleNeighborDiscoverResponse``, a reply that arrives after its entry
+        has timed out is not accepted.
+        """
+        pending = self._pending
+        if pending is None or pending.future is None or pending.future.done():
+            return False
+
+        try:
+            if packet.get_payload_type() != PAYLOAD_TYPE_RESPONSE:
+                return False
+
+            payload = bytes(getattr(packet, "payload", b"") or b"")
+            if len(payload) < 3:
+                return False
+
+            our_hash = self.local_identity.get_public_key()[0]
+            if payload[0] != our_hash or payload[1] != pending.src_hash:
+                return False
+
+            peer = Identity(bytes.fromhex(pending.pubkey))
+            shared_secret = peer.calc_shared_secret(self.local_identity.get_private_key())
+            plaintext = CryptoUtils.mac_then_decrypt(shared_secret[:16], shared_secret, payload[2:])
+            if not plaintext or len(plaintext) < 8:
+                return False
+
+            # plaintext: tag(4) + responder clock(4) + comma-separated scope names
+            if int.from_bytes(plaintext[:4], "little") != pending.tag:
+                return False
+
+            scopes = bytes(plaintext[8:]).decode("utf-8", errors="replace").rstrip("\x00").strip()
+            if not pending.future.done():
+                pending.future.set_result(scopes)
+            return True
+
+        except Exception as e:
+            logger.debug(f"Error matching scope response: {e}")
+            return False
+
+
+def _as_float(value, default: float) -> float:
+    try:
+        if value is None:
+            return default
+        return float(value)
+    except (TypeError, ValueError):
+        return default

repeater/main.py

@@ -31,6 +31,7 @@ from repeater.handler_helpers import (
     AdvertHelper,
     DiscoveryHelper,
     LoginHelper,
+    NeighborScopeHelper,
     PathHelper,
     ProtocolRequestHelper,
     TextHelper,
@@ -38,6 +39,7 @@ from repeater.handler_helpers import (
 )
 from repeater.identity_manager import IdentityConfigurationError, IdentityManager, IdentitySpec
 from repeater.logging_utils import normalize_log_level
+from repeater.neighbors_publisher import NeighborsPublisher
 from repeater.packet_router import PacketRouter
 from repeater.region_map_builder import build_region_map
 from repeater.sensors import SensorManager
@@ -99,6 +101,8 @@ class RepeaterDaemon:
         self.trace_helper = None
         self.advert_helper = None
         self.discovery_helper = None
+        self.neighbor_scope_helper = None
+        self.neighbors_publisher = None
         self.login_helper = None
         self.text_helper = None
         self.path_helper = None
@@ -654,6 +658,35 @@ class RepeaterDaemon:
             # When trace reaches final node, push PUSH_CODE_TRACE_DATA (0x89) to companion clients (firmware onTraceRecv)
             self.trace_helper.on_trace_complete = self._on_trace_complete_for_companions
 
+            # Neighbour scope discovery + periodic MQTT neighbors publication.
+            # Created unconditionally and self-gating: the publisher idles unless
+            # the master switch is on and some broker opted in with neighbors:true.
+            self.neighbor_scope_helper = NeighborScopeHelper(
+                local_identity=self.local_identity,
+                packet_injector=self.router.inject_packet,
+                airtime_manager=(
+                    getattr(self.repeater_handler, "airtime_mgr", None)
+                    if self.repeater_handler
+                    else None
+                ),
+                config=self.config,
+            )
+            self.neighbors_publisher = NeighborsPublisher(
+                config=self.config,
+                local_identity=self.local_identity,
+                discovery_helper=self.discovery_helper,
+                scope_helper=self.neighbor_scope_helper,
+                mqtt_handler_provider=lambda: getattr(
+                    getattr(self.repeater_handler, "storage", None), "mqtt_handler", None
+                ),
+                storage_provider=lambda: getattr(self.repeater_handler, "storage", None),
+                self_scopes_fn=(
+                    self.login_helper._format_region_names if self.login_helper else None
+                ),
+            )
+            self.neighbors_publisher.start()
+            logger.info("Neighbors publisher initialized")
+
             # Optional pyMC_Glass integration loop (inform/control plane)
             self.glass_handler = GlassHandler(
                 config=self.config,
@@ -1632,6 +1665,13 @@ class RepeaterDaemon:
                 "http server", asyncio.to_thread(self.http_server.stop), timeout=3
             )
 
+        # Stop the neighbours publication loop.
+        if self.neighbors_publisher:
+            try:
+                await self.neighbors_publisher.stop()
+            except Exception as e:
+                logger.warning(f"Error stopping neighbors publisher: {e}")
+
         # Stop Glass inform loop
         if self.glass_handler:
             await self._shutdown_step("glass handler", self.glass_handler.stop())

repeater/neighbors_publisher.py

@@ -0,0 +1,559 @@
+"""Periodic neighbours publication for the MQTT ``neighbors`` topic.
+
+Python port of the firmware's two-stage neighbours cycle
+(MeshCore ``simple_repeater/MyMesh.cpp``, ``WITH_MQTT_NEIGHBORS``):
+
+* **Stage 1** — a zero-hop node-discovery broadcast refreshes the neighbour table.
+  Firmware calls ``sendNodeDiscoverReq()`` and waits out its collection window;
+  here that is a :class:`~repeater.handler_helpers.discovery.DiscoveryHelper`
+  session, whose results are persisted by the same enricher the ``discover.neighbors``
+  CLI command uses.
+* **Stage 2** — one anon-regions scope query per neighbour, issued serially by
+  :class:`~repeater.handler_helpers.neighbor_scopes.NeighborScopeHelper`.
+* **Publish** — the assembled table goes to every enabled broker that opted in
+  with ``neighbors: true``.
+
+The published table is the whole zero-hop neighbour snapshot, including entries
+that did not answer the scope query (``timeout`` / ``send_failed``), matching the
+firmware payload. Unlike the firmware there is no 10 KB cap: the ESP32 needed a
+fixed PSRAM buffer, a Linux host does not.
+"""
+
+import asyncio
+import logging
+import time
+from datetime import datetime, timezone
+from typing import Any, Callable, Dict, List, Optional
+
+from repeater.handler_helpers.neighbor_scopes import (
+    STATUS_RESPONDED,
+    STATUS_TIMEOUT,
+    NeighborSnapshot,
+    ScopeResult,
+)
+
+logger = logging.getLogger("NeighborsPublisher")
+
+# Firmware parity: mqtt.neighbors.interval accepts 12-336 hours, default 24, and
+# rejects out-of-range values rather than clamping them.
+MIN_INTERVAL_HOURS = 12
+MAX_INTERVAL_HOURS = 336
+DEFAULT_INTERVAL_HOURS = 24
+
+# Firmware waits out the 60 s node-discover collection window before querying scopes.
+DEFAULT_DISCOVERY_TIMEOUT_SECONDS = 60.0
+
+# Firmware bounds the table by MAX_NEIGHBOURS; this is the openhop equivalent so a
+# large advert history cannot turn into an unbounded sweep.
+DEFAULT_MAX_NEIGHBORS = 32
+
+# Zero-hop rows older than this are treated as gone and left out of the pass.
+DEFAULT_MAX_NEIGHBOR_AGE_SECONDS = 86400.0
+
+# How often the loop wakes to re-evaluate its schedule.
+_TICK_SECONDS = 30.0
+
+# Phases reported by status(), mirroring the firmware's NeighborsPhase.
+PHASE_DISABLED = "disabled"
+PHASE_SCHEDULED = "scheduled"
+PHASE_ACTIVE = "active"
+PHASE_DUE = "due"
+
+
+def build_neighbors_payload(
+    *,
+    origin: str,
+    origin_id: str,
+    self_scopes: str,
+    entries: List[dict],
+    timestamp: Optional[str] = None,
+) -> dict:
+    """Assemble the ``neighbors`` topic payload.
+
+    ``entries`` are ordered most- to least-useful (most recently heard first, then
+    stronger SNR, then pubkey) exactly as the firmware orders them. The firmware
+    needs that order so it can drop the tail when its fixed buffer fills; we keep
+    it because it is the documented shape of the topic and it puts the useful rows
+    first for consumers.
+    """
+    ordered = sorted(
+        entries,
+        key=lambda e: (e.get("heard_secs_ago", 0), -float(e.get("snr", 0.0)), e.get("pubkey", "")),
+    )
+    return {
+        "timestamp": timestamp or datetime.now(timezone.utc).isoformat(),
+        "origin": origin,
+        "origin_id": origin_id,
+        "self": {"scopes": self_scopes or ""},
+        "neighbors": ordered,
+    }
+
+
+class NeighborsPublisher:
+    """Owns the periodic neighbours cycle and its manual trigger."""
+
+    def __init__(
+        self,
+        config: dict,
+        *,
+        local_identity=None,
+        discovery_helper=None,
+        scope_helper=None,
+        mqtt_handler_provider: Optional[Callable[[], Any]] = None,
+        storage_provider: Optional[Callable[[], Any]] = None,
+        self_scopes_fn: Optional[Callable[[], str]] = None,
+    ):
+        self.config = config
+        self.local_identity = local_identity
+        self.discovery_helper = discovery_helper
+        self.scope_helper = scope_helper
+        self._mqtt_handler_provider = mqtt_handler_provider
+        self._storage_provider = storage_provider
+        self._self_scopes_fn = self_scopes_fn
+
+        self._task: Optional[asyncio.Task] = None
+        self._running = False
+        self._active = False
+        # Discovery responses collected during the current cycle, keyed by pubkey.
+        self._discovery_seen: Dict[str, dict] = {}
+        self._next_publish_at: Optional[float] = None
+        self._last_result: Optional[str] = None
+        self._last_publish_at: Optional[float] = None
+
+    # ------------------------------------------------------------------
+    # Config accessors (re-read every cycle so live edits take effect)
+    # ------------------------------------------------------------------
+    @property
+    def _neighbors_config(self) -> dict:
+        return (self.config.get("mqtt_brokers", {}) or {}).get("neighbors", {}) or {}
+
+    @property
+    def master_enabled(self) -> bool:
+        """Feature kill switch. Defaults on; the per-broker flags are the control."""
+        return bool(self._neighbors_config.get("enabled", True))
+
+    @property
+    def interval_seconds(self) -> float:
+        hours = self._neighbors_config.get("interval_hours", DEFAULT_INTERVAL_HOURS)
+        return normalize_interval_hours(hours) * 3600.0
+
+    @property
+    def discovery_timeout(self) -> float:
+        try:
+            value = float(
+                self._neighbors_config.get(
+                    "discovery_timeout_seconds", DEFAULT_DISCOVERY_TIMEOUT_SECONDS
+                )
+            )
+        except (TypeError, ValueError):
+            return DEFAULT_DISCOVERY_TIMEOUT_SECONDS
+        return max(1.0, value)
+
+    @property
+    def max_neighbors(self) -> int:
+        try:
+            value = int(self._neighbors_config.get("max_neighbors", DEFAULT_MAX_NEIGHBORS))
+        except (TypeError, ValueError):
+            return DEFAULT_MAX_NEIGHBORS
+        return max(1, value)
+
+    @property
+    def max_neighbor_age(self) -> float:
+        try:
+            value = float(
+                self._neighbors_config.get(
+                    "max_neighbor_age_seconds", DEFAULT_MAX_NEIGHBOR_AGE_SECONDS
+                )
+            )
+        except (TypeError, ValueError):
+            return DEFAULT_MAX_NEIGHBOR_AGE_SECONDS
+        return max(1.0, value)
+
+    # ------------------------------------------------------------------
+    # Lifecycle
+    # ------------------------------------------------------------------
+    def start(self) -> None:
+        if self._task is not None and not self._task.done():
+            return
+        self._running = True
+        self._task = asyncio.create_task(self._run_loop(), name="neighbors-publisher")
+        logger.info("Neighbors publisher started (interval %.1fh)", self.interval_seconds / 3600.0)
+
+    async def stop(self) -> None:
+        self._running = False
+        task = self._task
+        self._task = None
+        if task and not task.done():
+            task.cancel()
+            try:
+                await task
+            except asyncio.CancelledError:
+                pass
+            except Exception as e:
+                logger.debug(f"Neighbors publisher shutdown error ignored: {e}")
+        logger.info("Neighbors publisher stopped")
+
+    def status(self) -> dict:
+        if not self.enabled():
+            phase = PHASE_DISABLED
+        elif self._active:
+            phase = PHASE_ACTIVE
+        elif self._next_publish_at is None or time.monotonic() >= self._next_publish_at:
+            phase = PHASE_DUE
+        else:
+            phase = PHASE_SCHEDULED
+
+        secs_until_next = None
+        if phase == PHASE_SCHEDULED and self._next_publish_at is not None:
+            secs_until_next = max(0, int(self._next_publish_at - time.monotonic()))
+
+        return {
+            "phase": phase,
+            "secs_until_next": secs_until_next,
+            "last_result": self._last_result,
+            "last_publish_at": self._last_publish_at,
+            "interval_hours": self.interval_seconds / 3600.0,
+        }
+
+    def enabled(self) -> bool:
+        """True when the master switch is on and some broker opted in."""
+        if not self.master_enabled:
+            return False
+        handler = self._mqtt_handler()
+        return bool(handler and handler.has_neighbors_brokers())
+
+    def _mqtt_handler(self):
+        if not self._mqtt_handler_provider:
+            return None
+        try:
+            return self._mqtt_handler_provider()
+        except Exception as e:
+            logger.debug(f"MQTT handler unavailable: {e}")
+            return None
+
+    def _storage(self):
+        if not self._storage_provider:
+            return None
+        try:
+            return self._storage_provider()
+        except Exception as e:
+            logger.debug(f"Storage unavailable: {e}")
+            return None
+
+    def _local_pubkey_hex(self) -> str:
+        if not self.local_identity:
+            return ""
+        try:
+            return self.local_identity.get_public_key().hex().lower()
+        except Exception:
+            return ""
+
+    # ------------------------------------------------------------------
+    # Loop
+    # ------------------------------------------------------------------
+    async def _run_loop(self) -> None:
+        try:
+            while self._running:
+                try:
+                    await self._tick()
+                except asyncio.CancelledError:
+                    raise
+                except Exception as e:
+                    logger.error(f"Neighbors publisher cycle failed: {e}", exc_info=True)
+                    self._last_result = f"error: {e}"
+                    self._reschedule()
+                await asyncio.sleep(_TICK_SECONDS)
+        except asyncio.CancelledError:
+            logger.debug("Neighbors publisher loop cancelled")
+            raise
+
+    async def _tick(self) -> None:
+        if not self.enabled():
+            # Drop the schedule so re-enabling runs a pass promptly, matching the
+            # firmware's next_neighbors_publish = 0 reset.
+            self._next_publish_at = None
+            return
+
+        handler = self._mqtt_handler()
+        if not handler or not handler.has_connected_neighbors_brokers():
+            logger.debug("Neighbors publish deferred: no connected opted-in broker")
+            return
+
+        now = time.monotonic()
+        if self._next_publish_at is not None and now < self._next_publish_at:
+            return
+
+        await self.run_cycle(trigger="periodic")
+
+    def _reschedule(self) -> None:
+        self._next_publish_at = time.monotonic() + self.interval_seconds
+
+    # ------------------------------------------------------------------
+    # Cycle
+    # ------------------------------------------------------------------
+    async def run_cycle(self, trigger: str = "manual") -> dict:
+        """Run one full discovery + scopes + publish cycle."""
+        if self._active:
+            return {"success": False, "error": "neighbors cycle already active"}
+
+        self._active = True
+        started = time.monotonic()
+        self._discovery_seen = {}
+        try:
+            await self._refresh_neighbor_table()
+            targets = self._snapshot_neighbors()
+
+            scope_results: Dict[str, ScopeResult] = {}
+            if targets and self.scope_helper:
+                scope_results = await self.scope_helper.sweep(targets)
+            elif targets:
+                logger.warning("No scope helper available; publishing without scopes")
+
+            payload = self._build_payload(targets, scope_results)
+            published = self._publish(payload)
+
+            responded = sum(1 for r in scope_results.values() if r.status == STATUS_RESPONDED)
+            self._last_result = (
+                f"ok ({len(targets)} neighbours, {responded} with scopes)"
+                if published
+                else "publish failed (no connected broker)"
+            )
+            self._last_publish_at = time.time()
+            logger.info(
+                "Neighbors %s cycle finished in %.1fs: %d neighbour(s), %d with scopes",
+                trigger,
+                time.monotonic() - started,
+                len(targets),
+                responded,
+            )
+            return {
+                "success": True,
+                "neighbors": len(targets),
+                "responded": responded,
+                "published": published,
+            }
+        finally:
+            self._active = False
+            self._reschedule()
+
+    async def _refresh_neighbor_table(self) -> None:
+        """Stage 1: zero-hop node discovery, awaited to completion.
+
+        ``prefix_only=False`` is required, not cosmetic: the scope query needs the
+        neighbour's full 32-byte public key to derive a shared secret.
+        """
+        if not self.discovery_helper:
+            logger.debug("No discovery helper; using the stored neighbour table as-is")
+            return
+
+        try:
+            self.discovery_helper.cleanup_sessions()
+            session = self.discovery_helper.create_session(
+                timeout=self.discovery_timeout,
+                filter_mask=(1 << 2),  # repeaters
+                since=0,
+                prefix_only=False,
+                result_enricher=self._enrich_discovery_result,
+            )
+            await self.discovery_helper.execute_session(session["session_id"])
+        except Exception as e:
+            logger.warning(f"Neighbour table refresh failed, using stored table: {e}")
+
+    def _enrich_discovery_result(self, result: dict) -> dict:
+        """Record each discovery response for this cycle, and persist it.
+
+        This is what makes stage 1 matter: without it the snapshot would report
+        whatever ``last_seen``/``snr`` the advert history happened to hold, rather
+        than what answered just now. Mirrors firmware ``putNeighbour()`` on a
+        discovery response.
+
+        The in-memory copy is not redundant with the write. ``get_neighbors()``
+        serves a 60 s cache that ``store_advert()`` does not invalidate, so a
+        neighbour that answered seconds ago can be missing from, or stale in, the
+        table read that follows. The snapshot merges this dict over that read.
+        """
+        pubkey = str(result.get("pub_key") or "").strip().lower()
+        if not pubkey or len(pubkey) != 64:
+            return result
+
+        local_pubkey = self._local_pubkey_hex()
+        if local_pubkey and pubkey == local_pubkey:
+            return result
+
+        node_type_raw = int(result.get("node_type", 0) or 0)
+        snr_raw = result.get("response_snr", result.get("snr"))
+        if node_type_raw == 2:
+            self._discovery_seen[pubkey] = {
+                "last_seen": time.time(),
+                "snr": float(snr_raw) if snr_raw is not None else 0.0,
+            }
+
+        storage = self._storage()
+        record_advert = getattr(storage, "record_advert", None) if storage else None
+        if not callable(record_advert):
+            return result
+
+        node_type = int(result.get("node_type", 0) or 0)
+        rssi = result.get("rssi")
+        snr = result.get("response_snr", result.get("snr"))
+        try:
+            record_advert(
+                {
+                    "timestamp": time.time(),
+                    "pubkey": pubkey,
+                    "node_name": result.get("node_name"),
+                    "is_repeater": node_type == 2,
+                    "route_type": 2,
+                    "contact_type": {1: "Chat Node", 2: "Repeater", 3: "Room Server"}.get(
+                        node_type, "Unknown"
+                    ),
+                    "latitude": None,
+                    "longitude": None,
+                    "rssi": int(rssi) if rssi is not None else None,
+                    "snr": float(snr) if snr is not None else None,
+                    "is_new_neighbor": True,
+                    "zero_hop": True,
+                }
+            )
+        except Exception as e:
+            logger.debug(f"Could not persist discovery result for {pubkey[:8]}: {e}")
+
+        return result
+
+    def _snapshot_neighbors(self) -> List[NeighborSnapshot]:
+        """Freeze the zero-hop repeater table for this pass.
+
+        Taken once, up front: the sweep can run for minutes and the table keeps
+        changing underneath it (firmware hit the same problem and stopped indexing
+        into its live ``neighbours[]`` in ``aba571ed``).
+
+        Sources are the stored zero-hop repeater table (which also carries
+        neighbours heard via advert but silent during discovery, as the firmware's
+        table does) merged with this cycle's discovery responses, which win on
+        ``last_seen``/``snr`` because they are first-hand and the table read is
+        served from a cache the advert write does not invalidate.
+        """
+        storage = self._storage()
+        neighbors = {}
+        if storage:
+            try:
+                neighbors = storage.get_neighbors() or {}
+            except Exception as e:
+                logger.warning(f"Could not read neighbour table: {e}")
+
+        local_pubkey = self._local_pubkey_hex()
+        now = time.time()
+        max_age = self.max_neighbor_age
+
+        merged: Dict[str, dict] = {}
+        for pubkey, info in neighbors.items():
+            if not info.get("is_repeater") or not info.get("zero_hop"):
+                continue
+            key = str(pubkey or "").lower()
+            if len(key) != 64:
+                # Scope queries need the full key for ECDH; a prefix cannot be used.
+                continue
+            last_seen = float(info.get("last_seen") or 0.0)
+            if last_seen <= 0 or (now - last_seen) > max_age:
+                continue
+            merged[key] = {"last_seen": last_seen, "snr": float(info.get("snr") or 0.0)}
+
+        for key, seen in (self._discovery_seen or {}).items():
+            merged[key] = dict(seen)
+
+        snapshots = [
+            NeighborSnapshot(pubkey=key, last_seen=info["last_seen"], snr=info["snr"])
+            for key, info in merged.items()
+            if not (local_pubkey and key == local_pubkey)
+        ]
+
+        # Freshest first, then strongest, then pubkey for a deterministic order --
+        # the same ordering the firmware applies before it starts querying.
+        snapshots.sort(key=lambda s: (-s.last_seen, -s.snr, s.pubkey))
+        if len(snapshots) > self.max_neighbors:
+            logger.info(
+                "Neighbour table has %d entries; querying the freshest %d",
+                len(snapshots),
+                self.max_neighbors,
+            )
+            snapshots = snapshots[: self.max_neighbors]
+        return snapshots
+
+    def _build_payload(
+        self, targets: List[NeighborSnapshot], scope_results: Dict[str, ScopeResult]
+    ) -> dict:
+        now = time.time()
+        entries = []
+        for target in targets:
+            result = scope_results.get(target.pubkey) or ScopeResult(STATUS_TIMEOUT)
+            heard_secs_ago = int(max(0.0, now - target.last_seen)) if target.last_seen else 0
+            entries.append(
+                {
+                    "pubkey": target.pubkey,
+                    "snr": round(target.snr, 2),
+                    "heard_secs_ago": heard_secs_ago,
+                    "scopes": result.scopes or "",
+                    "status": result.status,
+                }
+            )
+
+        handler = self._mqtt_handler()
+        origin = getattr(handler, "node_name", "") if handler else ""
+        origin_id = getattr(handler, "public_key", "") if handler else ""
+        if not origin:
+            origin = self.config.get("repeater", {}).get("node_name", "openHop-Repeater")
+        if not origin_id and self.local_identity:
+            try:
+                origin_id = self.local_identity.get_public_key().hex().upper()
+            except Exception:
+                origin_id = ""
+
+        return build_neighbors_payload(
+            origin=origin,
+            origin_id=origin_id,
+            self_scopes=self._self_scopes(),
+            entries=entries,
+        )
+
+    def _self_scopes(self) -> str:
+        if not self._self_scopes_fn:
+            return ""
+        try:
+            return self._self_scopes_fn() or ""
+        except Exception as e:
+            logger.debug(f"Could not read local scopes: {e}")
+            return ""
+
+    def _publish(self, payload: dict) -> bool:
+        handler = self._mqtt_handler()
+        if not handler:
+            return False
+        try:
+            results = handler.publish_neighbors(payload)
+        except Exception as e:
+            logger.error(f"Neighbors publish failed: {e}")
+            return False
+        return bool(results)
+
+
+def normalize_interval_hours(value) -> float:
+    """Validate an interval in hours, falling back to the default when invalid.
+
+    Firmware rejects out-of-range values instead of clamping them; the API layer
+    surfaces that rejection to the user, and this fallback keeps a hand-edited
+    config.yaml from producing a nonsense schedule.
+    """
+    try:
+        hours = float(value)
+    except (TypeError, ValueError):
+        return float(DEFAULT_INTERVAL_HOURS)
+    if hours < MIN_INTERVAL_HOURS or hours > MAX_INTERVAL_HOURS:
+        logger.warning(
+            "mqtt_brokers.neighbors.interval_hours=%s outside %d-%d; using %d",
+            value,
+            MIN_INTERVAL_HOURS,
+            MAX_INTERVAL_HOURS,
+            DEFAULT_INTERVAL_HOURS,
+        )
+        return float(DEFAULT_INTERVAL_HOURS)
+    return hours

repeater/packet_router.py

@@ -713,7 +713,28 @@ class PacketRouter:
             # to first hop instead of original requester).
             consumed = False
             dest_hash = packet.payload[0] if packet.payload and len(packet.payload) >= 1 else None
-            companion_bridges = self._companion_bridges_for_packet(packet, metadata)
+
+            # A neighbour's answer to our own anon-regions scope query is addressed
+            # to this repeater's identity, and no companion bridge owns it. Offer it
+            # to the scope helper first; it consumes the packet only when the
+            # ciphertext authenticates under the pending neighbour's shared secret
+            # AND echoes that query's tag, so an unrelated RESPONSE still falls
+            # through to the companion paths below.
+            scope_helper = getattr(self.daemon, "neighbor_scope_helper", None)
+            scope_consumed = False
+            if scope_helper is not None:
+                try:
+                    scope_consumed = await scope_helper.process_response_packet(packet)
+                except Exception as e:
+                    logger.debug(f"Neighbor scope response matching failed: {e}")
+
+            if scope_consumed:
+                packet.mark_do_not_retransmit()
+                processed_by_injection = True
+                self._record_for_ui(packet, metadata)
+                companion_bridges = {}
+            else:
+                companion_bridges = self._companion_bridges_for_packet(packet, metadata)
             local_hash = getattr(self.daemon, "local_hash", None)
             if dest_hash is not None and dest_hash in companion_bridges:
                 _, consumed = await self._fan_out_to_bridges(

repeater/web/api_endpoints.py

@@ -2258,13 +2258,25 @@ class APIEndpoints:
                                 "reconnecting": conn.has_pending_reconnect(),
                             },
                             "format": conn.format,
+                            "neighbors": getattr(conn, "neighbors_enabled", False),
                         }
                     )
 
+            # Schedule summary for the neighbors topic, the openhop equivalent of
+            # the firmware's trailing "nbr: <next>/<last>" in `get mqtt.status`.
+            neighbors_status = None
+            publisher = getattr(self.daemon_instance, "neighbors_publisher", None)
+            if publisher:
+                try:
+                    neighbors_status = publisher.status()
+                except Exception as exc:
+                    logger.debug(f"mqtt_status could not read neighbors publisher: {exc}")
+
             return self._success(
                 {
                     "handler_active": handler is not None,
                     "brokers": connected_brokers,
+                    "neighbors": neighbors_status,
                 }
             )
         except Exception as e:
@@ -2325,6 +2337,79 @@ class APIEndpoints:
             logger.error(f"Error listing broker presets: {e}")
             return self._error(str(e))
 
+    @staticmethod
+    def _validate_neighbors_settings(raw):
+        """Validate the ``mqtt_brokers.neighbors`` block.
+
+        Returns ``(settings, error)``; ``error`` is None on success. The interval
+        is rejected rather than clamped when out of range, matching the firmware's
+        ``set mqtt.neighbors.interval`` behavior.
+        """
+        from repeater.neighbors_publisher import (
+            DEFAULT_INTERVAL_HOURS,
+            MAX_INTERVAL_HOURS,
+            MIN_INTERVAL_HOURS,
+        )
+
+        if not isinstance(raw, dict):
+            return None, "neighbors must be an object"
+
+        settings = {}
+
+        if "enabled" in raw:
+            settings["enabled"] = bool(raw["enabled"])
+
+        if "interval_hours" in raw:
+            try:
+                interval = int(raw["interval_hours"])
+            except (TypeError, ValueError):
+                return None, "neighbors.interval_hours must be a number"
+            if interval < MIN_INTERVAL_HOURS or interval > MAX_INTERVAL_HOURS:
+                return (
+                    None,
+                    f"neighbors.interval_hours must be between {MIN_INTERVAL_HOURS} "
+                    f"and {MAX_INTERVAL_HOURS} (default {DEFAULT_INTERVAL_HOURS})",
+                )
+            settings["interval_hours"] = interval
+
+        if "discovery_timeout_seconds" in raw:
+            try:
+                timeout = float(raw["discovery_timeout_seconds"])
+            except (TypeError, ValueError):
+                return None, "neighbors.discovery_timeout_seconds must be a number"
+            if timeout < 5 or timeout > 300:
+                return None, "neighbors.discovery_timeout_seconds must be between 5 and 300"
+            settings["discovery_timeout_seconds"] = timeout
+
+        if "scope_response_timeout_seconds" in raw:
+            try:
+                scope_timeout = float(raw["scope_response_timeout_seconds"])
+            except (TypeError, ValueError):
+                return None, "neighbors.scope_response_timeout_seconds must be a number"
+            if scope_timeout < 0 or scope_timeout > 300:
+                return None, "neighbors.scope_response_timeout_seconds must be between 0 and 300"
+            settings["scope_response_timeout_seconds"] = scope_timeout
+
+        if "max_neighbors" in raw:
+            try:
+                max_neighbors = int(raw["max_neighbors"])
+            except (TypeError, ValueError):
+                return None, "neighbors.max_neighbors must be a number"
+            if max_neighbors < 1 or max_neighbors > 255:
+                return None, "neighbors.max_neighbors must be between 1 and 255"
+            settings["max_neighbors"] = max_neighbors
+
+        if "max_neighbor_age_seconds" in raw:
+            try:
+                max_age = float(raw["max_neighbor_age_seconds"])
+            except (TypeError, ValueError):
+                return None, "neighbors.max_neighbor_age_seconds must be a number"
+            if max_age < 60:
+                return None, "neighbors.max_neighbor_age_seconds must be at least 60"
+            settings["max_neighbor_age_seconds"] = max_age
+
+        return settings, None
+
     @cherrypy.expose
     @cherrypy.tools.json_out()
     @cherrypy.tools.json_in()
@@ -2365,6 +2450,17 @@ class APIEndpoints:
                 mqtt_updates["owner"] = str(data["owner"]).strip()
             if "email" in data:
                 mqtt_updates["email"] = str(data["email"]).strip()
+            if "neighbors" in data:
+                neighbors_settings, error = self._validate_neighbors_settings(data["neighbors"])
+                if error:
+                    return self._error(error)
+                # update_and_save replaces a section's key outright, so merge onto
+                # the stored block instead of letting a partial POST drop the
+                # settings it did not mention.
+                existing_neighbors = (self.config.get("mqtt_brokers", {}) or {}).get(
+                    "neighbors", {}
+                ) or {}
+                mqtt_updates["neighbors"] = {**existing_neighbors, **neighbors_settings}
             # if "disallowed_packet_types" in data:
             #     mqtt_updates["disallowed_packet_types"] = list(data["disallowed_packet_types"])
             if "brokers" in data:
@@ -2402,6 +2498,9 @@ class APIEndpoints:
                         "format": str(b["format"]).strip(),
                         "disallowed_packet_types": list(b.get("disallowed_packet_types", [])),
                         "retain_status": bool(b.get("retain_status", False)),
+                        # Opt-in per broker; brokers that do not expect the
+                        # neighbors topic can reject it and drop the connection.
+                        "neighbors": bool(b.get("neighbors", False)),
                         "tls": {
                             "enabled": bool(
                                 b.get("tls", {}).get("enabled", True if port == 443 else False)

tests/test_mqtt_neighbors.py

@@ -0,0 +1,626 @@
+"""Tests for the MQTT neighbours feature.
+
+Covers the three pieces that carry real risk:
+
+* the serialized scope sweep (one query in flight, deadline armed only after the
+  request transmits) — the pacing the firmware had to be fixed to get right;
+* response matching, which must consume a reply only when it authenticates AND
+  echoes the pending tag, so unrelated RESPONSE traffic still reaches companions;
+* the publish gate, which must reach opted-in brokers only.
+"""
+
+import asyncio
+import json
+import time
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest
+
+from openhop_core.protocol import CryptoUtils, Identity, LocalIdentity
+from openhop_core.protocol.constants import PAYLOAD_TYPE_RESPONSE
+from repeater.data_acquisition.mqtt_handler import MeshCoreToMqttPusher
+from repeater.handler_helpers.neighbor_scopes import (
+    STATUS_RESPONDED,
+    STATUS_SEND_FAILED,
+    STATUS_TIMEOUT,
+    NeighborScopeHelper,
+    NeighborSnapshot,
+)
+from repeater.neighbors_publisher import (
+    MAX_INTERVAL_HOURS,
+    MIN_INTERVAL_HOURS,
+    NeighborsPublisher,
+    build_neighbors_payload,
+    normalize_interval_hours,
+)
+
+
+# ====================================================================
+# Scaffolding
+# ====================================================================
+class _FakePacket:
+    """Minimal Packet stand-in for the response-matching path."""
+
+    def __init__(self, payload: bytes):
+        self.payload = bytearray(payload)
+
+    def get_payload_type(self):
+        return PAYLOAD_TYPE_RESPONSE
+
+    def get_raw_length(self):
+        return len(self.payload) + 2
+
+
+def _make_response_packet(
+    responder: LocalIdentity, requester: LocalIdentity, tag: int, scopes: str
+) -> _FakePacket:
+    """Build the reply a neighbour sends: dest_hash + src_hash + encrypted body."""
+    requester_identity = Identity(requester.get_public_key())
+    shared_secret = requester_identity.calc_shared_secret(responder.get_private_key())
+    plaintext = tag.to_bytes(4, "little") + int(time.time()).to_bytes(4, "little") + scopes.encode()
+    cipher = CryptoUtils.encrypt_then_mac(shared_secret[:16], shared_secret, plaintext)
+    payload = bytes([requester.get_public_key()[0], responder.get_public_key()[0]]) + cipher
+    return _FakePacket(payload)
+
+
+def _helper_with_injector(local_identity, injector, config=None):
+    return NeighborScopeHelper(
+        local_identity=local_identity,
+        packet_injector=injector,
+        airtime_manager=None,
+        config=config or {},
+    )
+
+
+# ====================================================================
+# Payload
+# ====================================================================
+def test_payload_orders_most_useful_first():
+    payload = build_neighbors_payload(
+        origin="node",
+        origin_id="AA" * 32,
+        self_scopes="DEN,APRS",
+        entries=[
+            {"pubkey": "cc", "snr": 3.0, "heard_secs_ago": 900, "scopes": "", "status": "timeout"},
+            {
+                "pubkey": "aa",
+                "snr": 1.0,
+                "heard_secs_ago": 10,
+                "scopes": "DEN",
+                "status": "responded",
+            },
+            {"pubkey": "bb", "snr": 9.0, "heard_secs_ago": 10, "scopes": "", "status": "timeout"},
+        ],
+    )
+
+    assert [e["pubkey"] for e in payload["neighbors"]] == ["bb", "aa", "cc"]
+    assert payload["self"] == {"scopes": "DEN,APRS"}
+    assert payload["origin_id"] == "AA" * 32
+    assert payload["timestamp"]
+
+
+def test_payload_keeps_every_entry_regardless_of_size():
+    """Unlike the firmware there is no fixed publish buffer, so nothing is dropped."""
+    entries = [
+        {
+            "pubkey": f"{i:064x}",
+            "snr": 1.0,
+            "heard_secs_ago": i,
+            "scopes": "SCOPE" * 20,
+            "status": "responded",
+        }
+        for i in range(200)
+    ]
+    payload = build_neighbors_payload(
+        origin="node", origin_id="AA" * 32, self_scopes="", entries=entries
+    )
+
+    assert len(payload["neighbors"]) == 200
+    assert len(json.dumps(payload)) > 10240
+
+
+@pytest.mark.parametrize(
+    "value,expected",
+    [
+        (24, 24.0),
+        (MIN_INTERVAL_HOURS, float(MIN_INTERVAL_HOURS)),
+        (MAX_INTERVAL_HOURS, float(MAX_INTERVAL_HOURS)),
+        (MIN_INTERVAL_HOURS - 1, 24.0),  # out of range -> default, never clamped
+        (MAX_INTERVAL_HOURS + 1, 24.0),
+        ("nonsense", 24.0),
+        (None, 24.0),
+    ],
+)
+def test_interval_validation(value, expected):
+    assert normalize_interval_hours(value) == expected
+
+
+# ====================================================================
+# Scope sweep pacing
+# ====================================================================
+@pytest.mark.asyncio
+async def test_sweep_keeps_exactly_one_query_in_flight():
+    """The pacing guarantee: query N+1 is not sent until N has resolved."""
+    local = LocalIdentity()
+    peers = [LocalIdentity() for _ in range(3)]
+    targets = [
+        NeighborSnapshot(pubkey=p.get_public_key().hex(), last_seen=time.time(), snr=5.0)
+        for p in peers
+    ]
+
+    peers_by_hex = {p.get_public_key().hex(): p for p in peers}
+    in_flight = 0
+    max_in_flight = 0
+    sent_to = []
+    helper = None
+
+    async def injector(packet, wait_for_ack=False):
+        nonlocal in_flight, max_in_flight
+        in_flight += 1
+        max_in_flight = max(max_in_flight, in_flight)
+        pending = helper._pending
+        sent_to.append(pending.pubkey)
+        await asyncio.sleep(0)  # yield, so an overlapping send would be observable
+        # The neighbour answers once the request has "transmitted".
+        response = _make_response_packet(peers_by_hex[pending.pubkey], local, pending.tag, "DEN")
+        asyncio.get_running_loop().call_soon(
+            lambda: asyncio.ensure_future(helper.process_response_packet(response))
+        )
+        in_flight -= 1
+        return True
+
+    helper = _helper_with_injector(local, injector)
+    results = await helper.sweep(targets)
+
+    assert max_in_flight == 1
+    assert sent_to == [t.pubkey for t in targets]
+    assert all(r.status == STATUS_RESPONDED for r in results.values())
+    assert all(r.scopes == "DEN" for r in results.values())
+
+
+@pytest.mark.asyncio
+async def test_sweep_marks_timeout_without_a_response():
+    local = LocalIdentity()
+    peer = LocalIdentity()
+    target = NeighborSnapshot(pubkey=peer.get_public_key().hex(), last_seen=time.time())
+
+    async def injector(packet, wait_for_ack=False):
+        return True
+
+    helper = _helper_with_injector(
+        local,
+        injector,
+        config={"mqtt_brokers": {"neighbors": {"scope_response_timeout_seconds": 0.05}}},
+    )
+    results = await helper.sweep([target])
+
+    assert results[target.pubkey].status == STATUS_TIMEOUT
+    assert helper._pending is None
+
+
+@pytest.mark.asyncio
+async def test_sweep_marks_send_failed_when_transmit_fails():
+    local = LocalIdentity()
+    peer = LocalIdentity()
+    target = NeighborSnapshot(pubkey=peer.get_public_key().hex(), last_seen=time.time())
+
+    async def injector(packet, wait_for_ack=False):
+        return False
+
+    helper = _helper_with_injector(local, injector)
+    started = time.monotonic()
+    results = await helper.sweep([target])
+
+    assert results[target.pubkey].status == STATUS_SEND_FAILED
+    # No response window is opened for a packet that never transmitted.
+    assert time.monotonic() - started < 1.0
+    assert helper._pending is None
+
+
+@pytest.mark.asyncio
+async def test_sweep_budget_marks_remaining_targets_timeout():
+    local = LocalIdentity()
+    peers = [LocalIdentity() for _ in range(3)]
+    targets = [
+        NeighborSnapshot(pubkey=p.get_public_key().hex(), last_seen=time.time()) for p in peers
+    ]
+
+    async def injector(packet, wait_for_ack=False):
+        await asyncio.sleep(0.05)
+        return True
+
+    helper = _helper_with_injector(
+        local,
+        injector,
+        config={
+            "mqtt_brokers": {
+                "neighbors": {
+                    "scope_response_timeout_seconds": 0.05,
+                    "max_sweep_seconds": 0.12,
+                }
+            }
+        },
+    )
+    results = await helper.sweep(targets)
+
+    assert len(results) == 3
+    assert results[targets[-1].pubkey].status == STATUS_TIMEOUT
+
+
+@pytest.mark.asyncio
+async def test_response_timeout_scales_with_radio_settings():
+    local = LocalIdentity()
+
+    class _Airtime:
+        def __init__(self, airtime_ms):
+            self._airtime_ms = airtime_ms
+
+        def calculate_airtime(self, payload_len):
+            return self._airtime_ms
+
+    fast = NeighborScopeHelper(local, airtime_manager=_Airtime(50), config={})
+    slow = NeighborScopeHelper(local, airtime_manager=_Airtime(2000), config={})
+
+    assert slow.response_timeout() > fast.response_timeout()
+
+    override = NeighborScopeHelper(
+        local,
+        airtime_manager=_Airtime(2000),
+        config={"mqtt_brokers": {"neighbors": {"scope_response_timeout_seconds": 7}}},
+    )
+    assert override.response_timeout() == 7
+
+
+# ====================================================================
+# Response matching
+# ====================================================================
+@pytest.mark.asyncio
+async def test_response_with_matching_tag_is_consumed():
+    local = LocalIdentity()
+    peer = LocalIdentity()
+    target = NeighborSnapshot(pubkey=peer.get_public_key().hex(), last_seen=time.time())
+
+    captured = {}
+
+    async def injector(packet, wait_for_ack=False):
+        captured["tag"] = helper._pending.tag
+        response = _make_response_packet(peer, local, captured["tag"], "DEN,APRS")
+        asyncio.get_running_loop().call_soon(
+            lambda: asyncio.ensure_future(helper.process_response_packet(response))
+        )
+        return True
+
+    helper = _helper_with_injector(local, injector)
+    results = await helper.sweep([target])
+
+    assert results[target.pubkey].status == STATUS_RESPONDED
+    assert results[target.pubkey].scopes == "DEN,APRS"
+
+
+@pytest.mark.asyncio
+async def test_response_with_wrong_tag_is_not_consumed():
+    local = LocalIdentity()
+    peer = LocalIdentity()
+    target = NeighborSnapshot(pubkey=peer.get_public_key().hex(), last_seen=time.time())
+
+    async def injector(packet, wait_for_ack=False):
+        bogus = _make_response_packet(peer, local, helper._pending.tag ^ 0xFFFF, "DEN")
+        assert await helper.process_response_packet(bogus) is False
+        return True
+
+    helper = _helper_with_injector(
+        local,
+        injector,
+        config={"mqtt_brokers": {"neighbors": {"scope_response_timeout_seconds": 0.05}}},
+    )
+    results = await helper.sweep([target])
+
+    assert results[target.pubkey].status == STATUS_TIMEOUT
+
+
+@pytest.mark.asyncio
+async def test_response_from_unrelated_node_is_not_consumed():
+    local = LocalIdentity()
+    peer = LocalIdentity()
+    stranger = LocalIdentity()
+    target = NeighborSnapshot(pubkey=peer.get_public_key().hex(), last_seen=time.time())
+
+    async def injector(packet, wait_for_ack=False):
+        # Correct tag, wrong sender: must not resolve the pending query.
+        foreign = _make_response_packet(stranger, local, helper._pending.tag, "DEN")
+        assert await helper.process_response_packet(foreign) is False
+        return True
+
+    helper = _helper_with_injector(
+        local,
+        injector,
+        config={"mqtt_brokers": {"neighbors": {"scope_response_timeout_seconds": 0.05}}},
+    )
+    results = await helper.sweep([target])
+
+    assert results[target.pubkey].status == STATUS_TIMEOUT
+
+
+@pytest.mark.asyncio
+async def test_response_with_no_query_pending_is_ignored():
+    local = LocalIdentity()
+    peer = LocalIdentity()
+    helper = _helper_with_injector(local, None)
+
+    packet = _make_response_packet(peer, local, 1234, "DEN")
+    assert await helper.process_response_packet(packet) is False
+
+
+# ====================================================================
+# Publish gating
+# ====================================================================
+def _pusher_with_brokers(brokers):
+    config = {
+        "repeater": {"node_name": "test-node"},
+        "radio": {
+            "spreading_factor": 8,
+            "bandwidth": 62500,
+            "coding_rate": 8,
+            "preamble_length": 17,
+            "frequency": 869618000,
+        },
+        "mqtt_brokers": {
+            "iata_code": "LAX",
+            "status_interval": 0,
+            "owner": "",
+            "email": "",
+            "brokers": brokers,
+        },
+    }
+    identity = SimpleNamespace(get_public_key=lambda: bytes.fromhex("AB" * 32))
+    return MeshCoreToMqttPusher(local_identity=identity, config=config)
+
+
+def _broker(name, *, neighbors=False, enabled=True, fmt="letsmesh"):
+    return {
+        "name": name,
+        "enabled": enabled,
+        "host": f"{name}.example",
+        "port": 1883,
+        "transport": "tcp",
+        "format": fmt,
+        "use_jwt_auth": False,
+        "neighbors": neighbors,
+        "tls": {"enabled": False, "insecure": False},
+    }
+
+
+def _capture(conn):
+    captured = []
+
+    def _fake_publish(topic, payload, retain=False, qos=0):
+        captured.append({"topic": topic, "payload": payload, "retain": retain, "qos": qos})
+        return None
+
+    conn.client = MagicMock()
+    conn.client.publish = _fake_publish
+    conn._running = True
+    return captured
+
+
+def test_publish_neighbors_reaches_only_opted_in_brokers():
+    pusher = _pusher_with_brokers([_broker("opted-in", neighbors=True), _broker("plain")])
+    opted_in, plain = pusher.connections
+    opted_capture = _capture(opted_in)
+    plain_capture = _capture(plain)
+
+    pusher.publish_neighbors({"neighbors": [], "self": {"scopes": ""}})
+
+    assert len(opted_capture) == 1
+    assert plain_capture == []
+    assert opted_capture[0]["topic"] == "meshcore/LAX/" + "AB" * 32 + "/neighbors"
+    assert opted_capture[0]["qos"] == 1
+    assert opted_capture[0]["retain"] is False
+
+
+def test_publish_neighbors_skips_disabled_broker():
+    pusher = _pusher_with_brokers([_broker("off", neighbors=True, enabled=False)])
+    captured = _capture(pusher.connections[0])
+
+    results = pusher.publish_neighbors({"neighbors": []})
+
+    assert captured == []
+    assert results == []
+
+
+def test_publish_neighbors_uses_custom_base_topic_for_legacy_format():
+    pusher = _pusher_with_brokers([_broker("lan", neighbors=True, fmt="mqtt")])
+    captured = _capture(pusher.connections[0])
+
+    pusher.publish_neighbors({"neighbors": []})
+
+    assert captured[0]["topic"] == "meshcore/repeater/test-node/neighbors"
+
+
+def test_broker_opt_in_flags():
+    pusher = _pusher_with_brokers([_broker("plain")])
+    assert pusher.has_neighbors_brokers() is False
+
+    pusher = _pusher_with_brokers([_broker("opted-in", neighbors=True)])
+    assert pusher.has_neighbors_brokers() is True
+    assert pusher.has_connected_neighbors_brokers() is False
+
+    pusher.connections[0]._running = True
+    assert pusher.has_connected_neighbors_brokers() is True
+
+
+# ====================================================================
+# Publisher gating and snapshotting
+# ====================================================================
+def _publisher(config, *, handler=None, storage=None, **kwargs):
+    return NeighborsPublisher(
+        config=config,
+        mqtt_handler_provider=lambda: handler,
+        storage_provider=lambda: storage,
+        **kwargs,
+    )
+
+
+def test_publisher_disabled_without_opted_in_broker():
+    handler = SimpleNamespace(has_neighbors_brokers=lambda: False)
+    publisher = _publisher({"mqtt_brokers": {}}, handler=handler)
+
+    assert publisher.enabled() is False
+    assert publisher.status()["phase"] == "disabled"
+
+
+def test_master_switch_overrides_broker_opt_in():
+    handler = SimpleNamespace(has_neighbors_brokers=lambda: True)
+    config = {"mqtt_brokers": {"neighbors": {"enabled": False}}}
+    publisher = _publisher(config, handler=handler)
+
+    assert publisher.master_enabled is False
+    assert publisher.enabled() is False
+
+    config["mqtt_brokers"]["neighbors"]["enabled"] = True
+    assert publisher.enabled() is True
+
+
+def test_snapshot_filters_to_fresh_zero_hop_repeaters():
+    now = time.time()
+    local = LocalIdentity()
+    local_key = local.get_public_key().hex()
+    storage = SimpleNamespace(
+        get_neighbors=lambda: {
+            "aa" * 32: {"is_repeater": True, "zero_hop": True, "last_seen": now, "snr": 4.0},
+            "bb" * 32: {"is_repeater": True, "zero_hop": False, "last_seen": now, "snr": 9.0},
+            "cc" * 32: {"is_repeater": False, "zero_hop": True, "last_seen": now, "snr": 9.0},
+            "dd" * 32: {
+                "is_repeater": True,
+                "zero_hop": True,
+                "last_seen": now - 100000,
+                "snr": 9.0,
+            },
+            "ee": {"is_repeater": True, "zero_hop": True, "last_seen": now, "snr": 9.0},
+            local_key: {"is_repeater": True, "zero_hop": True, "last_seen": now, "snr": 9.0},
+        }
+    )
+    publisher = _publisher({"mqtt_brokers": {}}, storage=storage, local_identity=local)
+
+    snapshot = publisher._snapshot_neighbors()
+
+    # Multi-hop, non-repeater, stale, short-key and self rows are all excluded.
+    assert [s.pubkey for s in snapshot] == ["aa" * 32]
+
+
+def test_snapshot_merges_this_cycle_discovery_over_the_cached_table():
+    """get_neighbors() serves a 60s cache that store_advert() does not invalidate.
+
+    A neighbour that answered discovery seconds ago must still be queried, with
+    its live SNR, even when the cached table has not caught up.
+    """
+    now = time.time()
+    storage = SimpleNamespace(
+        get_neighbors=lambda: {
+            "aa" * 32: {
+                "is_repeater": True,
+                "zero_hop": True,
+                "last_seen": now - 500,
+                "snr": -3.0,
+            }
+        }
+    )
+    publisher = _publisher({"mqtt_brokers": {}}, storage=storage)
+
+    # Responses seen during this cycle: one refresh, one brand-new neighbour.
+    publisher._discovery_seen = {
+        "aa" * 32: {"last_seen": now, "snr": 8.0},
+        "bb" * 32: {"last_seen": now, "snr": 6.0},
+    }
+
+    snapshot = {s.pubkey: s for s in publisher._snapshot_neighbors()}
+
+    assert set(snapshot) == {"aa" * 32, "bb" * 32}
+    assert snapshot["aa" * 32].snr == 8.0  # live value wins over the cached row
+    assert snapshot["aa" * 32].last_seen == now
+
+
+def test_snapshot_is_capped_and_ordered_freshest_first():
+    now = time.time()
+    storage = SimpleNamespace(
+        get_neighbors=lambda: {
+            f"{i:064x}": {
+                "is_repeater": True,
+                "zero_hop": True,
+                "last_seen": now - i,
+                "snr": 1.0,
+            }
+            for i in range(10)
+        }
+    )
+    publisher = _publisher({"mqtt_brokers": {"neighbors": {"max_neighbors": 3}}}, storage=storage)
+
+    snapshot = publisher._snapshot_neighbors()
+
+    assert [s.pubkey for s in snapshot] == [f"{i:064x}" for i in range(3)]
+
+
+@pytest.mark.asyncio
+async def test_cycle_publishes_table_with_unanswered_neighbors():
+    """Firmware parity: the whole zero-hop table is published, timeouts included."""
+    now = time.time()
+    published = []
+    handler = SimpleNamespace(
+        has_neighbors_brokers=lambda: True,
+        has_connected_neighbors_brokers=lambda: True,
+        publish_neighbors=lambda payload: published.append(payload) or [("b", None)],
+        node_name="test-node",
+        public_key="AB" * 32,
+    )
+    storage = SimpleNamespace(
+        get_neighbors=lambda: {
+            "aa" * 32: {"is_repeater": True, "zero_hop": True, "last_seen": now, "snr": 4.0},
+            "bb" * 32: {
+                "is_repeater": True,
+                "zero_hop": True,
+                "last_seen": now - 30,
+                "snr": 2.0,
+            },
+        }
+    )
+
+    class _Sweeper:
+        async def sweep(self, targets):
+            from repeater.handler_helpers.neighbor_scopes import ScopeResult
+
+            return {
+                targets[0].pubkey: ScopeResult(STATUS_RESPONDED, "DEN"),
+                targets[1].pubkey: ScopeResult(STATUS_TIMEOUT),
+            }
+
+    publisher = _publisher(
+        {"mqtt_brokers": {}},
+        handler=handler,
+        storage=storage,
+        scope_helper=_Sweeper(),
+        self_scopes_fn=lambda: "DEN,APRS",
+    )
+
+    result = await publisher.run_cycle(trigger="test")
+
+    assert result["success"] is True
+    assert result["neighbors"] == 2
+    assert result["responded"] == 1
+
+    payload = published[0]
+    assert payload["self"] == {"scopes": "DEN,APRS"}
+    statuses = {e["pubkey"]: e["status"] for e in payload["neighbors"]}
+    assert statuses == {"aa" * 32: STATUS_RESPONDED, "bb" * 32: STATUS_TIMEOUT}
+    # A cycle always arms the next one, so a failure cannot wedge the schedule.
+    assert publisher.status()["phase"] == "scheduled"
+
+
+@pytest.mark.asyncio
+async def test_cycle_rejects_reentry_while_active():
+    publisher = _publisher({"mqtt_brokers": {}})
+    publisher._active = True
+
+    result = await publisher.run_cycle()
+
+    assert result["success"] is False