fix(mqtt): address adversarial review of the neighbours feature

- api_endpoints: the broker rebuild is a strict field whitelist, so a client
  that omits `neighbors` (every UI build predating this feature) reset every
  broker to false on any MQTT save, silently disabling the feature. Fall back
  to the stored per-broker value instead. An explicit false still turns it off.
- neighbor_scopes: `_query_one` cleared `_pending` in an `except Exception`,
  which does not catch CancelledError. A shutdown cancel during the injector
  await -- where a query spends most of its wall time -- left a dead pending
  query that kept consuming RESPONSE packets and hiding them from every
  companion bridge. One try/finally around the whole method.
- neighbor_scopes: refresh_config had no caller after construction despite
  promising live updates; it now re-reads before each sweep, so a live
  delays.direct_tx_delay_factor change resizes the response window.
- neighbor_scopes: truncate scopes at the first NUL as the responder's C-string
  writer intends, rather than only rstrip-ing the cipher's zero padding.
- api_endpoints: accept the documented max_sweep_seconds and
  duty_cycle_abort_seconds, reject unknown keys instead of reporting success on
  a typo, and reject non-bool `enabled` / fractional interval_hours rather than
  silently coercing them.
- mqtt_handler: check paho's rc before counting a neighbours publish as
  delivered; the payload is uncapped, so a size/queue rejection is realistic.
- neighbors_publisher: retry after 15 minutes when a cycle publishes nothing,
  instead of burning the full 24h interval.
- neighbors_publisher: track the manual `discover.scopes` cycle so shutdown can
  cancel it, and move the already-running check onto the event loop.

Tests: router-hook integration (a matching response is consumed, an unrelated
one still reaches companions), cancellation, sweep re-entry, malformed and
hostile response payloads, config validation, and the broker save round-trip.
All ten regression tests fail against the pre-fix code.

repeater/data_acquisition/mqtt_handler.py

@@ -1058,11 +1058,22 @@ class MeshCoreToMqttPusher:
                     )
                     continue
                 result = conn.publish("neighbors", message, retain=False, qos=1)
+                # This is by far the largest payload the node emits and it is not
+                # size-capped, so an oversized or queue-full rejection is a real
+                # outcome. Check paho's rc rather than reporting a publish that
+                # never left the client as a success.
+                rc = getattr(result, "rc", None)
+                if result is None or (rc is not None and rc != mqtt.MQTT_ERR_SUCCESS):
+                    logger.warning(
+                        f"Neighbors publish rejected by {conn.broker['name']} "
+                        f"(rc={rc}, bytes={len(message.encode('utf-8'))})"
+                    )
+                    continue
                 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")
+            logger.warning("Neighbors table was not published to any broker")
 
         return results
 

repeater/handler_helpers/mesh_cli.py

@@ -1,3 +1,4 @@
+import concurrent.futures
 import logging
 from pathlib import Path
 from typing import Any, Callable, Dict, Optional
@@ -1324,9 +1325,6 @@ class MeshCLI:
         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
@@ -1340,10 +1338,21 @@ class MeshCLI:
             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"
+            # trigger_cycle owns the "already running" check and tracks the task
+            # so shutdown can cancel it. Doing it here instead would race: this
+            # runs on the CLI thread, the cycle starts on the event loop.
+            started = concurrent.futures.Future()
+
+            def _start():
+                try:
+                    started.set_result(publisher.trigger_cycle())
+                except Exception as exc:  # pragma: no cover - defensive
+                    started.set_exception(exc)
+
+            loop.call_soon_threadsafe(_start)
+            if started.result(timeout=5):
+                return "OK - neighbor scope discovery started"
+            return "Err - neighbors cycle already active"
         except Exception as e:
             logger.error(f"discover.scopes failed: {e}", exc_info=True)
             return f"Error: {e}"

repeater/handler_helpers/neighbor_scopes.py

@@ -127,6 +127,9 @@ class NeighborScopeHelper:
         self.local_identity = local_identity
         self.packet_injector = packet_injector
         self.airtime_manager = airtime_manager
+        # Held by reference so a live config update is visible; re-read at the
+        # start of every sweep rather than cached at construction.
+        self.config = config if config is not None else {}
 
         self._pending: Optional[_PendingQuery] = None
         self._sweep_lock = asyncio.Lock()
@@ -135,13 +138,17 @@ class NeighborScopeHelper:
         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 {})
+        self.refresh_config(self.config)
 
     # ------------------------------------------------------------------
     # Configuration
     # ------------------------------------------------------------------
-    def refresh_config(self, config: dict) -> None:
-        """Re-read the tunables so a live config update takes effect next sweep."""
+    def refresh_config(self, config: Optional[dict] = None) -> None:
+        """Re-read the tunables. Called at construction and before each sweep."""
+        if config is None:
+            config = self.config
+        else:
+            self.config = config
         neighbors_cfg = (config.get("mqtt_brokers", {}) or {}).get("neighbors", {}) or {}
 
         self._response_timeout_override = _as_float(
@@ -216,6 +223,9 @@ class NeighborScopeHelper:
             raise RuntimeError("neighbor scope sweep already active")
 
         async with self._sweep_lock:
+            # Pick up live config edits (the mesh CLI can change
+            # delays.direct_tx_delay_factor, which sizes the response window).
+            self.refresh_config(self.config)
             deadline = time.monotonic() + self._max_sweep_seconds
             timeout = self.response_timeout()
             logger.info(
@@ -276,30 +286,35 @@ class NeighborScopeHelper:
         )
         self._pending = pending
 
+        # One finally for the whole method: the injector await spends most of its
+        # wall time in the engine TX path, so a shutdown cancel lands there. A
+        # CancelledError escaping with _pending still set would leave a dead query
+        # matching (and hiding from the companion bridges) every later RESPONSE.
         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)
+            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 asyncio.CancelledError:
+                raise
+            except Exception as e:
+                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)
+            if not sent:
+                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)
+            try:
+                scopes = await asyncio.wait_for(pending.future, timeout)
+            except asyncio.TimeoutError:
+                logger.debug(f"Scope query timed out for {target.pubkey[:8]}")
+                return ScopeResult(STATUS_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
 
@@ -384,7 +399,11 @@ class NeighborScopeHelper:
             if int.from_bytes(plaintext[:4], "little") != pending.tag:
                 return False
 
-            scopes = bytes(plaintext[8:]).decode("utf-8", errors="replace").rstrip("\x00").strip()
+            # The responder builds this field as a C string and the block cipher
+            # zero-pads the tail, so stop at the first NUL exactly as the firmware
+            # reader does -- rstrip alone would let "DEN\x00junk" through.
+            raw_scopes = bytes(plaintext[8:]).split(b"\x00", 1)[0]
+            scopes = raw_scopes.decode("utf-8", errors="replace").strip()
             if not pending.future.done():
                 pending.future.set_result(scopes)
             return True

repeater/neighbors_publisher.py

@@ -53,6 +53,10 @@ DEFAULT_MAX_NEIGHBOR_AGE_SECONDS = 86400.0
 # How often the loop wakes to re-evaluate its schedule.
 _TICK_SECONDS = 30.0
 
+# Delay before retrying a cycle that failed or published nothing, instead of
+# waiting out the full interval.
+RETRY_DELAY_SECONDS = 900.0
+
 # Phases reported by status(), mirroring the firmware's NeighborsPhase.
 PHASE_DISABLED = "disabled"
 PHASE_SCHEDULED = "scheduled"
@@ -112,6 +116,7 @@ class NeighborsPublisher:
         self._self_scopes_fn = self_scopes_fn
 
         self._task: Optional[asyncio.Task] = None
+        self._manual_task: Optional[asyncio.Task] = None
         self._running = False
         self._active = False
         # Discovery responses collected during the current cycle, keyed by pubkey.
@@ -179,12 +184,39 @@ class NeighborsPublisher:
         self._task = asyncio.create_task(self._run_loop(), name="neighbors-publisher")
         logger.info("Neighbors publisher started (interval %.1fh)", self.interval_seconds / 3600.0)
 
+    def trigger_cycle(self) -> bool:
+        """Start a manual cycle, tracked so shutdown can cancel it.
+
+        A cycle runs for minutes (a discovery window plus one serialized scope
+        query per neighbour), so an untracked task would keep transmitting while
+        the daemon tears down. Returns False when a cycle is already running.
+        """
+        if self._active or (self._manual_task is not None and not self._manual_task.done()):
+            return False
+        self._manual_task = asyncio.create_task(
+            self.run_cycle(trigger="manual"), name="neighbors-manual-cycle"
+        )
+        self._manual_task.add_done_callback(self._on_manual_task_done)
+        return True
+
+    def _on_manual_task_done(self, task: asyncio.Task) -> None:
+        if self._manual_task is task:
+            self._manual_task = None
+        try:
+            task.result()
+        except asyncio.CancelledError:
+            pass
+        except Exception as e:
+            logger.error(f"Manual neighbors cycle failed: {e}", exc_info=True)
+
     async def stop(self) -> None:
         self._running = False
-        task = self._task
+        tasks = [t for t in (self._task, self._manual_task) if t and not t.done()]
         self._task = None
-        if task and not task.done():
+        self._manual_task = None
+        for task in tasks:
             task.cancel()
+        for task in tasks:
             try:
                 await task
             except asyncio.CancelledError:
@@ -261,7 +293,7 @@ class NeighborsPublisher:
                 except Exception as e:
                     logger.error(f"Neighbors publisher cycle failed: {e}", exc_info=True)
                     self._last_result = f"error: {e}"
-                    self._reschedule()
+                    self._reschedule(retry=True)
                 await asyncio.sleep(_TICK_SECONDS)
         except asyncio.CancelledError:
             logger.debug("Neighbors publisher loop cancelled")
@@ -285,8 +317,16 @@ class NeighborsPublisher:
 
         await self.run_cycle(trigger="periodic")
 
-    def _reschedule(self) -> None:
-        self._next_publish_at = time.monotonic() + self.interval_seconds
+    def _reschedule(self, *, retry: bool = False) -> None:
+        """Arm the next cycle.
+
+        A cycle that produced no publish retries on a short delay instead of
+        burning the whole interval: the usual cause is a broker that was briefly
+        unreachable or rejected the payload, and waiting a day to find out
+        otherwise is not useful.
+        """
+        delay = min(RETRY_DELAY_SECONDS, self.interval_seconds) if retry else self.interval_seconds
+        self._next_publish_at = time.monotonic() + delay
 
     # ------------------------------------------------------------------
     # Cycle
@@ -299,6 +339,7 @@ class NeighborsPublisher:
         self._active = True
         started = time.monotonic()
         self._discovery_seen = {}
+        published = False
         try:
             await self._refresh_neighbor_table()
             targets = self._snapshot_neighbors()
@@ -316,15 +357,17 @@ class NeighborsPublisher:
             self._last_result = (
                 f"ok ({len(targets)} neighbours, {responded} with scopes)"
                 if published
-                else "publish failed (no connected broker)"
+                else "publish failed (broker unreachable or rejected the payload)"
             )
             self._last_publish_at = time.time()
             logger.info(
-                "Neighbors %s cycle finished in %.1fs: %d neighbour(s), %d with scopes",
+                "Neighbors %s cycle finished in %.1fs: %d neighbour(s), %d with scopes, "
+                "published=%s",
                 trigger,
                 time.monotonic() - started,
                 len(targets),
                 responded,
+                published,
             )
             return {
                 "success": True,
@@ -334,7 +377,7 @@ class NeighborsPublisher:
             }
         finally:
             self._active = False
-            self._reschedule()
+            self._reschedule(retry=not published)
 
     async def _refresh_neighbor_table(self) -> None:
         """Stage 1: zero-hop node discovery, awaited to completion.

repeater/web/api_endpoints.py

@@ -2354,16 +2354,36 @@ class APIEndpoints:
         if not isinstance(raw, dict):
             return None, "neighbors must be an object"
 
+        # Reject unknown keys rather than accepting them silently: a typo would
+        # otherwise return success while changing nothing the runtime reads.
+        known_keys = {
+            "enabled",
+            "interval_hours",
+            "discovery_timeout_seconds",
+            "scope_response_timeout_seconds",
+            "max_sweep_seconds",
+            "duty_cycle_abort_seconds",
+            "max_neighbors",
+            "max_neighbor_age_seconds",
+        }
+        unknown = sorted(set(raw) - known_keys)
+        if unknown:
+            return None, f"Unknown neighbors settings: {', '.join(unknown)}"
+
         settings = {}
 
         if "enabled" in raw:
-            settings["enabled"] = bool(raw["enabled"])
+            if not isinstance(raw["enabled"], bool):
+                return None, "neighbors.enabled must be true or false"
+            settings["enabled"] = raw["enabled"]
 
         if "interval_hours" in raw:
-            try:
-                interval = int(raw["interval_hours"])
-            except (TypeError, ValueError):
+            interval = raw["interval_hours"]
+            if isinstance(interval, bool) or not isinstance(interval, (int, float)):
                 return None, "neighbors.interval_hours must be a number"
+            if float(interval) != int(interval):
+                return None, "neighbors.interval_hours must be a whole number of hours"
+            interval = int(interval)
             if interval < MIN_INTERVAL_HOURS or interval > MAX_INTERVAL_HOURS:
                 return (
                     None,
@@ -2408,6 +2428,24 @@ class APIEndpoints:
                 return None, "neighbors.max_neighbor_age_seconds must be at least 60"
             settings["max_neighbor_age_seconds"] = max_age
 
+        if "max_sweep_seconds" in raw:
+            try:
+                max_sweep = float(raw["max_sweep_seconds"])
+            except (TypeError, ValueError):
+                return None, "neighbors.max_sweep_seconds must be a number"
+            if max_sweep < 30 or max_sweep > 7200:
+                return None, "neighbors.max_sweep_seconds must be between 30 and 7200"
+            settings["max_sweep_seconds"] = max_sweep
+
+        if "duty_cycle_abort_seconds" in raw:
+            try:
+                abort_after = float(raw["duty_cycle_abort_seconds"])
+            except (TypeError, ValueError):
+                return None, "neighbors.duty_cycle_abort_seconds must be a number"
+            if abort_after < 0 or abort_after > 600:
+                return None, "neighbors.duty_cycle_abort_seconds must be between 0 and 600"
+            settings["duty_cycle_abort_seconds"] = abort_after
+
         return settings, None
 
     @cherrypy.expose
@@ -2467,6 +2505,18 @@ class APIEndpoints:
                 brokers = data["brokers"]
                 if not isinstance(brokers, list):
                     return self._error("brokers must be a list")
+
+                # The rebuild below is a strict field whitelist, so any key a
+                # client omits is reset to its default. For neighbors that would
+                # silently switch the feature off whenever a UI that predates it
+                # saves an unrelated MQTT setting, so fall back to the stored
+                # value per broker name instead of to False.
+                stored_neighbors_by_name = {
+                    str(existing.get("name")): bool(existing.get("neighbors", False))
+                    for existing in (self.config.get("mqtt_brokers", {}) or {}).get("brokers", [])
+                    if isinstance(existing, dict) and existing.get("name")
+                }
+
                 validated = []
                 for i, b in enumerate(brokers):
                     if not isinstance(b, dict):
@@ -2500,7 +2550,11 @@ class APIEndpoints:
                         "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)),
+                        "neighbors": bool(
+                            b["neighbors"]
+                            if "neighbors" in b
+                            else stored_neighbors_by_name.get(str(b["name"]).strip(), False)
+                        ),
                         "tls": {
                             "enabled": bool(
                                 b.get("tls", {}).get("enabled", True if port == 443 else False)

tests/test_mqtt_neighbors.py

@@ -42,6 +42,11 @@ from repeater.neighbors_publisher import (
 class _FakePacket:
     """Minimal Packet stand-in for the response-matching path."""
 
+    do_not_retransmit = False
+
+    def mark_do_not_retransmit(self):
+        self.do_not_retransmit = True
+
     def __init__(self, payload: bytes):
         self.payload = bytearray(payload)
 
@@ -352,6 +357,215 @@ async def test_response_with_no_query_pending_is_ignored():
     assert await helper.process_response_packet(packet) is False
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "payload",
+    [
+        b"",
+        b"\x01",
+        b"\x01\x02",  # header only, no ciphertext
+        b"\x01\x02\x03\x04\x05",  # too short to carry a MAC + block
+        bytes(64),  # right shape, garbage contents
+    ],
+)
+async def test_malformed_response_never_raises_or_matches(payload):
+    local = LocalIdentity()
+    peer = LocalIdentity()
+    target = NeighborSnapshot(pubkey=peer.get_public_key().hex(), last_seen=time.time())
+
+    async def injector(packet, wait_for_ack=False):
+        # Truncated/garbage payloads must be rejected quietly, not blow up the
+        # router thread that offers every RESPONSE to this matcher.
+        malformed = _FakePacket(payload)
+        assert await helper.process_response_packet(malformed) 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_scopes_are_truncated_at_the_first_nul():
+    """The responder builds a C string; the cipher zero-pads the tail."""
+    local = LocalIdentity()
+    peer = LocalIdentity()
+    target = NeighborSnapshot(pubkey=peer.get_public_key().hex(), last_seen=time.time())
+
+    async def injector(packet, wait_for_ack=False):
+        response = _make_response_packet(peer, local, helper._pending.tag, "DEN\x00junk")
+        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].scopes == "DEN"
+
+
+@pytest.mark.asyncio
+async def test_cancelling_a_sweep_mid_send_clears_the_pending_query():
+    """A leaked pending query would keep hiding RESPONSE packets from companions.
+
+    The injector await is where a shutdown cancel lands: it covers the engine's
+    TX-delay and duty-cycle deferral, which is most of a query's wall time.
+    """
+    local = LocalIdentity()
+    peer = LocalIdentity()
+    target = NeighborSnapshot(pubkey=peer.get_public_key().hex(), last_seen=time.time())
+
+    async def injector(packet, wait_for_ack=False):
+        await asyncio.sleep(30)  # still "transmitting" when the cancel arrives
+        return True
+
+    helper = _helper_with_injector(local, injector)
+    task = asyncio.create_task(helper.sweep([target]))
+    await asyncio.sleep(0.05)
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+
+    assert helper._pending is None
+    assert helper.active is False
+
+    # A response arriving afterwards must not be consumed by the dead query.
+    stale = _make_response_packet(peer, local, 1, "DEN")
+    assert await helper.process_response_packet(stale) is False
+
+
+@pytest.mark.asyncio
+async def test_concurrent_sweep_is_rejected():
+    local = LocalIdentity()
+    peer = LocalIdentity()
+    target = NeighborSnapshot(pubkey=peer.get_public_key().hex(), last_seen=time.time())
+
+    async def injector(packet, wait_for_ack=False):
+        await asyncio.sleep(0.2)
+        return True
+
+    helper = _helper_with_injector(local, injector)
+    first = asyncio.create_task(helper.sweep([target]))
+    await asyncio.sleep(0.05)
+
+    with pytest.raises(RuntimeError):
+        await helper.sweep([target])
+
+    first.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await first
+
+
+@pytest.mark.asyncio
+async def test_sweep_rereads_live_config():
+    """delays.direct_tx_delay_factor is live-updatable and sizes the window."""
+    local = LocalIdentity()
+    config = {"mqtt_brokers": {"neighbors": {"scope_response_timeout_seconds": 5}}}
+    helper = _helper_with_injector(local, None, config=config)
+    assert helper.response_timeout() == 5
+
+    config["mqtt_brokers"]["neighbors"]["scope_response_timeout_seconds"] = 11
+    await helper.sweep([NeighborSnapshot(pubkey="aa" * 32, last_seen=time.time())])
+
+    assert helper.response_timeout() == 11
+
+
+# ====================================================================
+# Router integration
+# ====================================================================
+class _StubRouter:
+    """Exercises the real PacketRouter RESPONSE branch against a stub daemon."""
+
+    def __init__(self, scope_helper):
+        self.fanned_out = []
+        self.recorded = []
+        self.daemon = SimpleNamespace(
+            neighbor_scope_helper=scope_helper,
+            local_hash=0x11,
+            repeater_handler=None,
+        )
+
+    async def _fan_out_to_bridges(self, packet, bridges, context=""):
+        self.fanned_out.append((packet, dict(bridges), context))
+        return (bool(bridges), False)
+
+    def _companion_bridges_for_packet(self, packet, metadata):
+        return {0x11: object()}
+
+    def _record_for_ui(self, packet, metadata):
+        self.recorded.append(packet)
+
+
+async def _route_response(router_stub, packet):
+    from repeater.packet_router import PacketRouter
+
+    return await PacketRouter._route_packet(router_stub, packet)
+
+
+@pytest.mark.asyncio
+async def test_router_consumes_only_a_matching_scope_response():
+    local = LocalIdentity()
+    peer = LocalIdentity()
+    target = NeighborSnapshot(pubkey=peer.get_public_key().hex(), last_seen=time.time())
+    routed = {}
+
+    async def injector(packet, wait_for_ack=False):
+        response = _make_response_packet(peer, local, helper._pending.tag, "DEN")
+        stub = _StubRouter(helper)
+        await _route_response(stub, response)
+        routed["consumed"] = stub
+        return True
+
+    helper = _helper_with_injector(local, injector)
+    results = await helper.sweep([target])
+
+    stub = routed["consumed"]
+    assert results[target.pubkey].status == STATUS_RESPONDED
+    # Consumed: not retransmitted, recorded, and never offered to a bridge.
+    assert stub.fanned_out == []
+    assert stub.recorded
+
+
+@pytest.mark.asyncio
+async def test_router_still_delivers_unrelated_responses_to_companions():
+    """A companion's login reply must not be swallowed by an active sweep.
+
+    Same 1-byte dest hash, same instant, different sender: the matcher must
+    decline it so the companion bridge still sees it.
+    """
+    local = LocalIdentity()
+    peer = LocalIdentity()
+    stranger = LocalIdentity()
+    target = NeighborSnapshot(pubkey=peer.get_public_key().hex(), last_seen=time.time())
+    routed = {}
+
+    async def injector(packet, wait_for_ack=False):
+        foreign = _make_response_packet(stranger, local, helper._pending.tag, "NOPE")
+        foreign.payload[0] = 0x11  # collide with the companion's dest hash
+        stub = _StubRouter(helper)
+        await _route_response(stub, foreign)
+        routed["stub"] = stub
+        return True
+
+    helper = _helper_with_injector(
+        local,
+        injector,
+        config={"mqtt_brokers": {"neighbors": {"scope_response_timeout_seconds": 0.05}}},
+    )
+    results = await helper.sweep([target])
+
+    stub = routed["stub"]
+    assert results[target.pubkey].status == STATUS_TIMEOUT
+    assert len(stub.fanned_out) == 1
+    assert stub.fanned_out[0][2] == "RESPONSE"
+
+
 # ====================================================================
 # Publish gating
 # ====================================================================
@@ -624,3 +838,180 @@ async def test_cycle_rejects_reentry_while_active():
     result = await publisher.run_cycle()
 
     assert result["success"] is False
+
+
+@pytest.mark.asyncio
+async def test_failed_publish_retries_sooner_than_a_full_interval():
+    """A rejected payload must not cost a whole 24h interval."""
+    from repeater.neighbors_publisher import RETRY_DELAY_SECONDS
+
+    handler = SimpleNamespace(
+        has_neighbors_brokers=lambda: True,
+        has_connected_neighbors_brokers=lambda: True,
+        publish_neighbors=lambda payload: [],  # nothing reached a broker
+        node_name="n",
+        public_key="AB" * 32,
+    )
+    publisher = _publisher({"mqtt_brokers": {}}, handler=handler, storage=None)
+
+    result = await publisher.run_cycle(trigger="test")
+
+    assert result["published"] is False
+    assert "publish failed" in publisher._last_result
+    assert publisher.status()["secs_until_next"] <= RETRY_DELAY_SECONDS
+
+
+def test_enricher_records_only_full_key_repeaters_and_never_self():
+    local = LocalIdentity()
+    publisher = _publisher({"mqtt_brokers": {}}, local_identity=local)
+    publisher._discovery_seen = {}
+
+    publisher._enrich_discovery_result({"pub_key": "aa" * 32, "node_type": 2, "response_snr": 7.5})
+    publisher._enrich_discovery_result({"pub_key": "bb" * 32, "node_type": 1})  # chat node
+    publisher._enrich_discovery_result({"pub_key": "cc", "node_type": 2})  # prefix only
+    publisher._enrich_discovery_result(
+        {"pub_key": local.get_public_key().hex(), "node_type": 2}
+    )  # self
+
+    assert set(publisher._discovery_seen) == {"aa" * 32}
+    assert publisher._discovery_seen["aa" * 32]["snr"] == 7.5
+
+
+def test_enricher_persists_discovery_results_to_storage():
+    storage = SimpleNamespace(record_advert=MagicMock(), get_neighbors=lambda: {})
+    publisher = _publisher({"mqtt_brokers": {}}, storage=storage)
+    publisher._discovery_seen = {}
+
+    publisher._enrich_discovery_result(
+        {"pub_key": "aa" * 32, "node_type": 2, "response_snr": 3.0, "rssi": -90}
+    )
+
+    record = storage.record_advert.call_args.args[0]
+    assert record["pubkey"] == "aa" * 32
+    assert record["is_repeater"] is True
+    assert record["zero_hop"] is True
+    assert record["snr"] == 3.0
+
+
+# ====================================================================
+# Config validation
+# ====================================================================
+def _validate(raw):
+    from repeater.web.api_endpoints import APIEndpoints
+
+    return APIEndpoints._validate_neighbors_settings(raw)
+
+
+@pytest.mark.parametrize(
+    "raw,expect_error",
+    [
+        ({"interval_hours": 24}, False),
+        ({"interval_hours": MIN_INTERVAL_HOURS}, False),
+        ({"interval_hours": MAX_INTERVAL_HOURS}, False),
+        ({"interval_hours": MIN_INTERVAL_HOURS - 1}, True),
+        ({"interval_hours": MAX_INTERVAL_HOURS + 1}, True),
+        ({"interval_hours": 12.5}, True),  # silently truncating would be worse
+        ({"interval_hours": "24"}, True),
+        ({"enabled": True}, False),
+        ({"enabled": "false"}, True),  # would coerce to True
+        ({"discovery_timeout_seconds": 60}, False),
+        ({"discovery_timeout_seconds": 1}, True),
+        ({"scope_response_timeout_seconds": 0}, False),
+        ({"max_neighbors": 0}, True),
+        ({"max_neighbor_age_seconds": 10}, True),
+        ({"max_sweep_seconds": 900}, False),
+        ({"duty_cycle_abort_seconds": 30}, False),
+        ({"max_sweep_secondz": 5}, True),  # typo must not report success
+        ("not a dict", True),
+    ],
+)
+def test_neighbors_settings_validation(raw, expect_error):
+    settings, error = _validate(raw)
+    assert (error is not None) is expect_error
+    if not expect_error:
+        assert settings
+
+
+def _api_with_stored_brokers(monkeypatch, brokers, neighbors_block=None):
+    import cherrypy
+
+    from repeater.web.api_endpoints import APIEndpoints
+
+    request = SimpleNamespace(method="POST", params={}, json={})
+    response = SimpleNamespace(headers={}, status=200)
+    monkeypatch.setattr(cherrypy, "request", request, raising=False)
+    monkeypatch.setattr(cherrypy, "response", response, raising=False)
+
+    api = APIEndpoints.__new__(APIEndpoints)
+    api.config = {"mqtt_brokers": {"brokers": brokers, "neighbors": neighbors_block or {}}}
+    api.daemon_instance = None
+    api.send_advert_func = None
+    api.event_loop = None
+    api.stats_getter = None
+    api._config_path = "/tmp/test-config.yaml"
+    api.config_manager = MagicMock()
+    api.config_manager.update_and_save.return_value = {"success": True, "saved": True}
+    return api, request
+
+
+def test_broker_neighbors_flag_survives_a_save_that_omits_it(monkeypatch):
+    """A UI that predates the feature must not silently disable it.
+
+    The broker rebuild is a strict field whitelist, so a client that never learned
+    about `neighbors` would otherwise reset every broker to False on any unrelated
+    MQTT save — turning the feature off with no error.
+    """
+    api, request = _api_with_stored_brokers(
+        monkeypatch,
+        [{"name": "keeper", "neighbors": True}, {"name": "plain", "neighbors": False}],
+    )
+
+    request.json = {
+        "email": "someone@example.com",
+        "brokers": [
+            {"name": "keeper", "host": "h", "port": 1883, "format": "letsmesh"},
+            {"name": "plain", "host": "h", "port": 1883, "format": "letsmesh"},
+        ],
+    }
+    assert api.update_mqtt_config()["success"] is True
+
+    saved = api.config_manager.update_and_save.call_args.kwargs["updates"]["mqtt_brokers"]
+    by_name = {b["name"]: b["neighbors"] for b in saved["brokers"]}
+    assert by_name == {"keeper": True, "plain": False}
+
+
+def test_broker_neighbors_flag_can_be_turned_off_explicitly(monkeypatch):
+    api, request = _api_with_stored_brokers(monkeypatch, [{"name": "keeper", "neighbors": True}])
+
+    request.json = {
+        "brokers": [
+            {"name": "keeper", "host": "h", "port": 1883, "format": "letsmesh", "neighbors": False}
+        ]
+    }
+    assert api.update_mqtt_config()["success"] is True
+
+    saved = api.config_manager.update_and_save.call_args.kwargs["updates"]["mqtt_brokers"]
+    assert saved["brokers"][0]["neighbors"] is False
+
+
+def test_partial_neighbors_post_keeps_unmentioned_settings(monkeypatch):
+    api, request = _api_with_stored_brokers(
+        monkeypatch, [], neighbors_block={"interval_hours": 48, "max_neighbors": 8}
+    )
+
+    request.json = {"neighbors": {"enabled": False}}
+    assert api.update_mqtt_config()["success"] is True
+
+    saved = api.config_manager.update_and_save.call_args.kwargs["updates"]["mqtt_brokers"]
+    assert saved["neighbors"] == {"interval_hours": 48, "max_neighbors": 8, "enabled": False}
+
+
+def test_invalid_neighbors_interval_is_rejected_by_the_endpoint(monkeypatch):
+    api, request = _api_with_stored_brokers(monkeypatch, [])
+
+    request.json = {"neighbors": {"interval_hours": 1}}
+    out = api.update_mqtt_config()
+
+    assert out["success"] is False
+    assert "between 12 and 336" in out["error"]
+    api.config_manager.update_and_save.assert_not_called()