Merge pull request #35 from agessaman/feat/mqtt-neighbors

feat(mqtt): MQTT neighbors feature (flex port onto webconfig)

MQTT_IMPLEMENTATION.md

@@ -382,6 +382,8 @@ These settings apply across all MQTT slots:
 - `get mqtt.rx` - Get RX packet uplinking setting (on/off)
 - `get mqtt.tx` - Get TX packet uplinking setting (on/off/advert)
 - `get mqtt.interval` - Get status publish interval
+- `get mqtt.neighbors` - Get periodic neighbors publishing setting (on/off; PSRAM only)
+- `get mqtt.neighbors.interval` - Get neighbors publish interval in hours (PSRAM only)
 - `get mqtt.ntp` - Get effective NTP server hostname
 - `get mqtt.ntp.diag` - Probe every configured NTP server for connectivity (does not change the clock; serial console shows each server's reported time, LoRa shows a compact `<server> ok|fail` list)
 - `get mqtt.owner` - Get owner public key (serial console only)
@@ -399,6 +401,8 @@ These settings apply across all MQTT slots:
   - `advert` - Uplink only this node's own advert packets (self-originated)
   - `off` - Disable TX packet uplinking
 - `set mqtt.interval <minutes>` - Set status publish interval (1-60 minutes)
+- `set mqtt.neighbors on|off` - Enable/disable periodic neighbors publishing (PSRAM only; read live, no restart)
+- `set mqtt.neighbors.interval <hours>` - Set neighbors publish interval (12-336 hours, default 24; PSRAM only)
 - `set mqtt.ntp <hostname>` - Set custom NTP server (validated with immediate sync); `none` reverts to default
 - `set mqtt.owner <64-hex-char-public-key>` - Set owner public key
 - `set mqtt.email <email>` - Set owner email address
@@ -584,6 +588,12 @@ Full packet data with RF characteristics and metadata.
 ### Raw Topic: `meshcore/{IATA}/{DEVICE_PUBLIC_KEY}/raw`
 Minimal raw packet data for map integration.
 
+### Neighbors Topic: `meshcore/{IATA}/{DEVICE_PUBLIC_KEY}/neighbors`
+Periodic snapshot of this node's zero-hop neighbor table plus each neighbor's
+region scopes (PSRAM boards only; disabled by default). Published non-retained at
+QoS 1 on the interval set by `mqtt.neighbors.interval` (12–336 h, default 24 h).
+Like status/raw, this topic is **not** sent to MeshRank slots (packets-only contract).
+
 **Note**: `{DEVICE_PUBLIC_KEY}` is the device's public key in hexadecimal format (64 characters).
 
 ## JSON Message Formats
@@ -662,6 +672,28 @@ Minimal raw packet data for map integration.
 }
 ```
 
+### Neighbors Message
+```json
+{
+  "timestamp": "2024-01-01T12:00:00.000000+00:00",
+  "origin": "MeshCore-HOWL",
+  "origin_id": "A1B2C3D4E5F67890...",
+  "self": { "scopes": "DEN,APRS" },
+  "neighbors": [
+    {
+      "pubkey": "0011223344556677...",
+      "snr": 9.75,
+      "heard_secs_ago": 42,
+      "scopes": "DEN,APRS",
+      "status": "responded"
+    }
+  ]
+}
+```
+Entries are ordered most- to least-useful (most recently heard, then stronger
+SNR); the tail is dropped if the payload would exceed the 10 KB publish buffer.
+`status` is `responded`, `timeout`, or `send_failed` per neighbor.
+
 ## Key Features
 
 ### Slot-Based Preset System

MQTT_INTERNALS.md

@@ -57,6 +57,37 @@ scheduled time are expired at dequeue, so under throttle the queue holds only fr
 traffic and admin responses reach the trickle of TX budget. Non-observer builds keep
 the upstream pool behavior.
 
+### Neighbors publication path (PSRAM only)
+
+Periodic neighbors publishing is gated on `WITH_MQTT_NEIGHBORS`
+(`defined(BOARD_HAS_PSRAM) && defined(MAX_NEIGHBOURS) && MAX_NEIGHBOURS > 0`,
+defined in `MQTTBridge.h`). It spans two subsystems and two cores:
+
+- **Mesh side (Core 1), `MyMesh`**: the `loop()` runs a two-stage refresh driven by
+  `mqtt_neighbors_interval`. Stage 1 sends a zero-hop `sendNodeDiscoverReq()` and waits
+  out its 60 s collection window to refresh `neighbours[]`. Stage 2
+  (`startNeighborDiscover`) fires one anon-regions scope query per heard neighbor,
+  overlaying them onto the peer-index space at `NEIGHBOR_DISCOVER_PEER_BASE` so their
+  `PAYLOAD_TYPE_RESPONSE` packets decrypt via `searchPeersByHash`/`getPeerSharedSecret`/
+  `onPeerDataRecv` even when the neighbor is not an ACL client. After all responses land
+  or a 30 s window expires, `finishNeighborDiscover()` builds the JSON with
+  `MQTTMessageBuilder::buildNeighborsMessage` into a transient PSRAM buffer and hands it
+  to the bridge.
+- **Bridge side, handoff**: `requestPublishNeighbors(json, len)` (Core 1) memcpys into a
+  persistent ~10 KB PSRAM buffer (`NEIGHBORS_JSON_BUFFER_SIZE`) and sets
+  `_neighbors_publish_pending` with a release store; the MQTT task (`mqttTaskLoop`, Core 0)
+  consumes it with an acquire load, calls `publishNeighbors()`, and clears the flag. A
+  second snapshot is dropped while one is in flight. `publishNeighbors()` sends QoS 1,
+  retain = `preset->allow_retain` (custom slots non-retained). MeshRank slots are skipped
+  (the topic router rejects non-packets for MeshRank).
+- **Status reporting**: `MyMesh` reports the schedule each loop via
+  `setNeighborsSchedule(phase, secs)`; `formatMqttStatusReply` renders it as the trailing
+  `nbr: <when>/<last>` field in `get mqtt.status` while the feature is enabled.
+
+The JSON builder lives in the pure, host-tested `MQTTPayloadBuilder`
+(`test/test_mqtt_payload_builder`); the topic type in `MQTTTopicRouter`
+(`test/test_mqtt_topic_router`). The mesh↔bridge orchestration above is on-target only.
+
 ### `/mqtt_prefs` file format
 
 `/mqtt_prefs` is written with an 8-byte `MQTTPrefsHeader` (`magic`, `version`,

docs/cli_commands.md

@@ -113,6 +113,20 @@ This document provides an overview of CLI commands that can be sent to MeshCore
 
 ---
 
+### Discover neighbor scopes (MQTT observer, PSRAM only)
+
+Refreshes the zero-hop neighbor table, then queries each neighbor for its region
+scopes and publishes the assembled table to the MQTT `neighbors` topic once.
+
+**Usage:**
+- `discover.scopes`
+
+**Note:** Requires a PSRAM board with the MQTT bridge running. On non-PSRAM MQTT
+builds it replies `Err - not supported (requires PSRAM)`. If a `discover.neighbors`
+refresh is already in flight, the scope pass is queued behind it.
+
+---
+
 ## Statistics
 
 ### Clear Stats
@@ -1076,6 +1090,39 @@ region save
 
 ---
 
+#### View or change periodic neighbors publishing (MQTT observer, PSRAM only)
+**Usage:**
+- `get mqtt.neighbors`
+- `set mqtt.neighbors <on|off>`
+
+**Parameters:**
+- `on`: periodically discover neighbor scopes and publish the neighbor table to the `neighbors` topic
+- `off`: disable periodic neighbors publishing
+
+**Default:** `off`
+
+> **Note:** Requires a PSRAM board. On non-PSRAM MQTT builds this replies
+> `Err - not supported (requires PSRAM)`. The setting is read live by the mesh
+> loop — no restart required; enabling it triggers a discovery on the next pass.
+> While enabled, `get mqtt.status` gains a trailing `nbr: <next>/<last>` field
+> (time to next publish, and how the last publish went).
+
+---
+
+#### View or change the neighbors publish interval (MQTT observer, PSRAM only)
+**Usage:**
+- `get mqtt.neighbors.interval`
+- `set mqtt.neighbors.interval <hours>`
+
+**Parameters:**
+- `hours`: how often to publish the neighbor table (12–336, default 24)
+
+**Default:** `24` (hours)
+
+> **Note:** Out-of-range values are rejected (not clamped). Requires a PSRAM board.
+
+---
+
 #### View or change the NTP server (MQTT observer only)
 **Usage:**
 - `get mqtt.ntp`

examples/simple_repeater/MyMesh.cpp

@@ -666,7 +666,22 @@ void MyMesh::onAnonDataRecv(mesh::Packet *packet, const uint8_t *secret, const m
 
 int MyMesh::searchPeersByHash(const uint8_t *hash) {
   int n = 0;
-  for (int i = 0; i < acl.getNumClients(); i++) {
+#if defined(WITH_MQTT_NEIGHBORS)
+  // While a neighbor-scope discovery is active, overlay the heard neighbours
+  // that are NOT already ACL clients so their RESPONSE packets can be decoded.
+  // Overlay indices are offset by NEIGHBOR_DISCOVER_PEER_BASE to keep them
+  // distinct from real ACL indices.
+  if (neighbor_discover_active) {
+    for (int i = 0; i < neighbor_discover_count && n < MAX_CLIENTS; i++) {
+      auto& nb = neighbours[neighbor_discover[i].neighbour_idx];
+      if (acl.getClient(nb.id.pub_key, PUB_KEY_SIZE) != nullptr) continue;
+      if (nb.heard_timestamp > 0 && nb.id.isHashMatch(hash)) {
+        matching_peer_indexes[n++] = NEIGHBOR_DISCOVER_PEER_BASE + i;
+      }
+    }
+  }
+#endif
+  for (int i = 0; i < acl.getNumClients() && n < MAX_CLIENTS; i++) {
     if (acl.getClientByIdx(i)->id.isHashMatch(hash)) {
       matching_peer_indexes[n++] = i; // store the INDEXES of matching contacts (for subsequent 'peer' methods)
     }
@@ -676,6 +691,16 @@ int MyMesh::searchPeersByHash(const uint8_t *hash) {
 
 void MyMesh::getPeerSharedSecret(uint8_t *dest_secret, int peer_idx) {
   int i = matching_peer_indexes[peer_idx];
+#if defined(WITH_MQTT_NEIGHBORS)
+  // Overlay entries have no precomputed shared secret; derive it on the fly.
+  if (neighbor_discover_active && i >= NEIGHBOR_DISCOVER_PEER_BASE) {
+    int oi = i - NEIGHBOR_DISCOVER_PEER_BASE;
+    if (oi >= 0 && oi < neighbor_discover_count) {
+      self_id.calcSharedSecret(dest_secret, neighbours[neighbor_discover[oi].neighbour_idx].id);
+      return;
+    }
+  }
+#endif
   if (i >= 0 && i < acl.getNumClients()) {
     // lookup pre-calculated shared_secret
     memcpy(dest_secret, acl.getClientByIdx(i)->shared_secret, PUB_KEY_SIZE);
@@ -707,11 +732,34 @@ void MyMesh::onAdvertRecv(mesh::Packet *packet, const mesh::Identity &id, uint32
 void MyMesh::onPeerDataRecv(mesh::Packet *packet, uint8_t type, int sender_idx, const uint8_t *secret,
                             uint8_t *data, size_t len) {
   int i = matching_peer_indexes[sender_idx];
+#if defined(WITH_MQTT_NEIGHBORS)
+  // Overlay response: a heard neighbour (not an ACL client) answering our
+  // anon-regions scope query. Consume it and stop — it is not a client packet.
+  if (neighbor_discover_active && i >= NEIGHBOR_DISCOVER_PEER_BASE) {
+    int oi = i - NEIGHBOR_DISCOVER_PEER_BASE;
+    if (type == PAYLOAD_TYPE_RESPONSE && oi >= 0 && oi < neighbor_discover_count) {
+      handleNeighborDiscoverResponse(oi, data, len);
+    }
+    return;
+  }
+#endif
   if (i < 0 || i >= acl.getNumClients()) { // get from our known_clients table (sender SHOULD already be known in this context)
     MESH_DEBUG_PRINTLN("onPeerDataRecv: invalid peer idx: %d", i);
     return;
   }
   ClientInfo* client = acl.getClientByIdx(i);
+#if defined(WITH_MQTT_NEIGHBORS)
+  // A neighbour that IS an ACL client resolves to a normal index above, so a
+  // scope-query response from it lands here — match it against the overlay.
+  if (neighbor_discover_active && type == PAYLOAD_TYPE_RESPONSE) {
+    for (int oi = 0; oi < neighbor_discover_count; oi++) {
+      auto& nb = neighbours[neighbor_discover[oi].neighbour_idx];
+      if (client->id.matches(nb.id) && handleNeighborDiscoverResponse(oi, data, len)) {
+        return;
+      }
+    }
+  }
+#endif
 
   if (type == PAYLOAD_TYPE_REQ) { // request (from a Known admin client!)
     uint32_t timestamp;
@@ -986,6 +1034,16 @@ MyMesh::MyMesh(mesh::MainBoard &board, mesh::Radio &radio, mesh::MillisecondCloc
   pending_discover_tag = 0;
   pending_discover_until = 0;
 
+#if defined(WITH_MQTT_NEIGHBORS)
+  neighbor_discover_count = 0;
+  neighbor_discover_active = false;
+  neighbor_table_refresh_active = false;
+  neighbor_table_refresh_periodic = false;
+  neighbor_discover_until = 0;
+  next_neighbors_publish = 0;
+  self_scopes_buf[0] = 0;
+#endif
+
   memset(default_scope.key, 0, sizeof(default_scope.key));
 }
 
@@ -1519,6 +1577,36 @@ void MyMesh::handleCommand(uint32_t sender_timestamp, char *command, char *reply
       sendNodeDiscoverReq();
       strcpy(reply, "OK - Discover sent");
     }
+#if defined(WITH_MQTT_NEIGHBORS)
+  } else if (memcmp(command, "discover.scopes", 15) == 0) {
+    const char* sub = command + 15;
+    while (*sub == ' ') sub++;
+    if (*sub != 0) {
+      strcpy(reply, "Err - discover.scopes has no options");
+    } else if (pending_discover_tag != 0 &&
+               !millisHasNowPassed(pending_discover_until) &&
+               !neighbor_discover_active) {
+      // A zero-hop table refresh is already collecting; queue the scope pass
+      // behind it (as a manual, non-periodic request) rather than starting a
+      // second refresh.
+      if (!neighborDiscoverReady(reply)) {
+        // reply already set by neighborDiscoverReady
+      } else {
+        neighbor_table_refresh_active = true;
+        neighbor_table_refresh_periodic = false;
+        long remaining_ms = (long)(pending_discover_until - futureMillis(0));
+        unsigned remaining_secs = remaining_ms > 0
+          ? (unsigned)(((unsigned long)remaining_ms + 999UL) / 1000UL) : 0;
+        sprintf(reply, "OK - scopes queued (%us discovery remaining)", remaining_secs);
+        MESH_DEBUG_PRINTLN("Neighbor scopes queued behind active discovery (%us remaining)", remaining_secs);
+      }
+    } else if (!startNeighborDiscover(reply)) {
+      // reply already set by startNeighborDiscover
+    }
+#elif defined(WITH_MQTT_BRIDGE)
+  } else if (memcmp(command, "discover.scopes", 15) == 0) {
+    strcpy(reply, "Err - not supported (requires PSRAM)");
+#endif
   } else{
     _cli.handleCommand(sender_timestamp, command, reply);  // common CLI commands
   }
@@ -1615,6 +1703,66 @@ void MyMesh::loop() {
   _alerter.onLoop(now);
 #endif
 
+#if defined(WITH_MQTT_NEIGHBORS)
+  // Two-stage periodic neighbors publication:
+  //   stage 1 - zero-hop node-discover refreshes the neighbour table (60s window)
+  //   stage 2 - anon-regions scope query per neighbour (startNeighborDiscover)
+  // then the table JSON is published and the next cycle is rescheduled.
+  bool periodic_neighbors_enabled = _cli.getObserverPrefs()->mqtt_neighbors_enabled;
+  if (neighbor_discover_active) {
+    loopNeighborDiscover();
+  } else if (neighbor_table_refresh_active) {
+    if (neighbor_table_refresh_periodic && !periodic_neighbors_enabled) {
+      // periodic switched off mid-refresh -> cancel (leave pending_discover_tag alone)
+      neighbor_table_refresh_active = false;
+      neighbor_table_refresh_periodic = false;
+      next_neighbors_publish = 0;
+    } else if (pending_discover_tag == 0 || millisHasNowPassed(pending_discover_until)) {
+      // 60s zero-hop window done -> begin the per-neighbour scope queries
+      bool was_periodic = neighbor_table_refresh_periodic;
+      pending_discover_tag = 0;
+      neighbor_table_refresh_active = false;
+      neighbor_table_refresh_periodic = false;
+      char tmp_reply[80];
+      const char* origin_str = was_periodic ? "periodic" : "manual";
+      if (startNeighborDiscover(tmp_reply)) {
+        MESH_DEBUG_PRINTLN("MQTT %s %s", origin_str, tmp_reply);
+      } else {
+        if (periodic_neighbors_enabled) {
+          next_neighbors_publish = futureMillis(_cli.getObserverPrefs()->mqtt_neighbors_interval);
+        }
+        MESH_DEBUG_PRINTLN("MQTT %s neighbor scope discovery failed: %s", origin_str, tmp_reply);
+      }
+    }
+  } else if (periodic_neighbors_enabled && bridge && bridge->isRunning()) {
+    if (next_neighbors_publish == 0 ||
+        (next_neighbors_publish != 0 && millisHasNowPassed(next_neighbors_publish))) {
+      if (pending_discover_tag == 0 || millisHasNowPassed(pending_discover_until)) {
+        pending_discover_tag = 0;
+        sendNodeDiscoverReq();
+        MESH_DEBUG_PRINTLN("MQTT periodic neighbor table refresh started");
+      } else {
+        MESH_DEBUG_PRINTLN("MQTT periodic refresh joined active neighbor discovery");
+      }
+      neighbor_table_refresh_active = true;
+      neighbor_table_refresh_periodic = true;
+    }
+  }
+
+  // Report the schedule state back to the bridge for `get mqtt.status`.
+  if (bridge) {
+    if (neighbor_discover_active || neighbor_table_refresh_active) {
+      bridge->setNeighborsSchedule(MQTTBridge::NBR_ACTIVE, 0);
+    } else if (next_neighbors_publish == 0 || millisHasNowPassed(next_neighbors_publish)) {
+      bridge->setNeighborsSchedule(MQTTBridge::NBR_DUE, 0);
+    } else {
+      long remaining_ms = (long)(next_neighbors_publish - futureMillis(0));
+      uint32_t remaining_secs = remaining_ms > 0 ? (uint32_t)(remaining_ms / 1000) : 0;
+      bridge->setNeighborsSchedule(MQTTBridge::NBR_SCHEDULED, remaining_secs);
+    }
+  }
+#endif
+
 #ifdef WITH_SNMP
   // Push radio stats to SNMP agent every 2 seconds
   if (_snmp_agent.isRunning()) {
@@ -1635,6 +1783,222 @@ void MyMesh::loop() {
 #endif
 }
 
+#if defined(WITH_MQTT_NEIGHBORS)
+#include "helpers/MQTTMessageBuilder.h"
+#if defined(ESP_PLATFORM)
+#include <esp_heap_caps.h>
+#endif
+
+// This node's own non-flood scope names, same source the anon-regions server
+// reply uses. Empty string when the node has no scoped regions.
+void MyMesh::getLocalScopes(char* buf, size_t len) {
+  if (!buf || len == 0) return;
+  buf[0] = 0;
+  region_map.exportNamesTo(buf, (int)len, REGION_DENY_FLOOD);
+}
+
+// Client side of the anon-regions request (the server side is handleAnonRegionsReq).
+// Inner payload: {tag(4)}{ANON_REQ_TYPE_REGIONS}{0x00 = zero-hop reply path}.
+bool MyMesh::sendAnonRegionsReq(const mesh::Identity& target, uint32_t& tag) {
+  uint8_t secret[PUB_KEY_SIZE];
+  self_id.calcSharedSecret(secret, target);
+
+  tag = getRTCClock()->getCurrentTimeUnique();
+  uint8_t inner[6];
+  memcpy(inner, &tag, 4);
+  inner[4] = ANON_REQ_TYPE_REGIONS;
+  inner[5] = 0x00; // request a zero-hop reply path
+
+  mesh::Packet* pkt = createAnonDatagram(PAYLOAD_TYPE_ANON_REQ, self_id, target, secret, inner, sizeof(inner));
+  if (!pkt) return false;
+  sendDirect(pkt, NULL, 0, 0);
+  return true;
+}
+
+// Match a RESPONSE against the pending overlay entry by tag; copy its scope
+// string (payload after the 8-byte {tag}{clock} header) into the entry.
+bool MyMesh::handleNeighborDiscoverResponse(int overlay_idx, const uint8_t* data, size_t len) {
+  if (overlay_idx < 0 || overlay_idx >= neighbor_discover_count) return false;
+  NeighborDiscoverEntry& entry = neighbor_discover[overlay_idx];
+  if (entry.status != ND_PENDING || len < 8) return false;
+
+  uint32_t tag;
+  memcpy(&tag, data, 4);
+  if (tag != entry.tag) return false;
+
+  size_t scope_len = len - 8;
+  if (scope_len >= sizeof(entry.scopes)) {
+    scope_len = sizeof(entry.scopes) - 1;
+  }
+  memcpy(entry.scopes, &data[8], scope_len);
+  entry.scopes[scope_len] = 0;
+  entry.status = ND_RESPONDED;
+  return true;
+}
+
+// Publish-ordering: most recently heard first, then stronger SNR, then pubkey.
+// The JSON builder drops the tail if the buffer fills, so the head must be the
+// most useful entries.
+static bool neighborPublishEntryComesBefore(
+    const MQTTMessageBuilder::NeighborsMessageEntry& lhs,
+    const MQTTMessageBuilder::NeighborsMessageEntry& rhs) {
+  if (lhs.heard_secs_ago != rhs.heard_secs_ago) {
+    return lhs.heard_secs_ago < rhs.heard_secs_ago;  // newer first
+  }
+  if (lhs.snr != rhs.snr) {
+    return lhs.snr > rhs.snr;  // stronger first when equally recent
+  }
+  return strcmp(lhs.pubkey_hex, rhs.pubkey_hex) < 0;
+}
+
+// Build the neighbors-table JSON and hand it to the bridge, then reschedule.
+void MyMesh::finishNeighborDiscover() {
+  getLocalScopes(self_scopes_buf, sizeof(self_scopes_buf));
+
+  char self_pubkey_hex[65];
+  mesh::Utils::toHex(self_pubkey_hex, self_id.pub_key, PUB_KEY_SIZE);
+
+  char origin[32];
+  MQTTBridge::getEffectiveMqttOrigin(&_prefs, _cli.getObserverPrefs(), origin, sizeof(origin));
+
+  char timestamp[40];
+  MQTTMessageBuilder::formatIsoTimestampForMqtt(getRTCClock()->getCurrentTime(), 0, nullptr, timestamp, sizeof(timestamp));
+
+  char pubkey_hex[MAX_NEIGHBOURS][65];
+  MQTTMessageBuilder::NeighborsMessageEntry entries[MAX_NEIGHBOURS];
+  uint32_t now_secs = getRTCClock()->getCurrentTime();
+
+  for (int i = 0; i < neighbor_discover_count; i++) {
+    auto& nb = neighbours[neighbor_discover[i].neighbour_idx];
+    mesh::Utils::toHex(pubkey_hex[i], nb.id.pub_key, PUB_KEY_SIZE);
+    entries[i].pubkey_hex = pubkey_hex[i];
+    entries[i].snr = nb.snr / 4.0f;
+    entries[i].heard_secs_ago = (nb.heard_timestamp > 0 && now_secs >= nb.heard_timestamp)
+      ? (now_secs - nb.heard_timestamp) : 0;
+    entries[i].scopes = neighbor_discover[i].scopes;
+    switch (neighbor_discover[i].status) {
+      case ND_RESPONDED:   entries[i].status = "responded"; break;
+      case ND_SEND_FAILED: entries[i].status = "send_failed"; break;
+      default:             entries[i].status = "timeout"; break;
+    }
+  }
+
+  // insertion sort: most useful first (JSON builder drops the tail on overflow)
+  for (int i = 1; i < neighbor_discover_count; i++) {
+    MQTTMessageBuilder::NeighborsMessageEntry entry = entries[i];
+    int j = i;
+    while (j > 0 && neighborPublishEntryComesBefore(entry, entries[j - 1])) {
+      entries[j] = entries[j - 1];
+      j--;
+    }
+    entries[j] = entry;
+  }
+
+#if defined(ESP_PLATFORM)
+  char* json_buf = (char*)heap_caps_malloc(MQTTBridge::NEIGHBORS_JSON_BUFFER_SIZE, MALLOC_CAP_SPIRAM);
+#else
+  char* json_buf = (char*)malloc(MQTTBridge::NEIGHBORS_JSON_BUFFER_SIZE);
+#endif
+  if (!json_buf) {
+    neighbor_discover_active = false;
+    neighbor_discover_count = 0;
+    if (_cli.getObserverPrefs()->mqtt_neighbors_enabled) {
+      next_neighbors_publish = futureMillis(_cli.getObserverPrefs()->mqtt_neighbors_interval);
+    }
+    return;
+  }
+
+  JsonDocument doc;
+  int json_len = MQTTMessageBuilder::buildNeighborsMessage(
+    doc, origin, self_pubkey_hex, timestamp, self_scopes_buf,
+    entries, neighbor_discover_count,
+    json_buf, MQTTBridge::NEIGHBORS_JSON_BUFFER_SIZE);
+
+  if (json_len > 0 && bridge) {
+    bridge->requestPublishNeighbors(json_buf, (size_t)json_len);
+  }
+
+#if defined(ESP_PLATFORM)
+  heap_caps_free(json_buf);
+#else
+  free(json_buf);
+#endif
+
+  neighbor_discover_active = false;
+  neighbor_discover_count = 0;
+  if (_cli.getObserverPrefs()->mqtt_neighbors_enabled) {
+    next_neighbors_publish = futureMillis(_cli.getObserverPrefs()->mqtt_neighbors_interval);
+  }
+}
+
+// Advance the scope-query phase; publish once all entries resolve or the window
+// times out (stragglers marked ND_TIMEOUT).
+void MyMesh::loopNeighborDiscover() {
+  if (!neighbor_discover_active) return;
+
+  bool all_done = true;
+  for (int i = 0; i < neighbor_discover_count; i++) {
+    if (neighbor_discover[i].status == ND_PENDING) { all_done = false; break; }
+  }
+  if (!all_done && !millisHasNowPassed(neighbor_discover_until)) return;
+  if (!all_done) {
+    for (int i = 0; i < neighbor_discover_count; i++) {
+      if (neighbor_discover[i].status == ND_PENDING) neighbor_discover[i].status = ND_TIMEOUT;
+    }
+  }
+  finishNeighborDiscover();
+}
+
+// Shared precondition for starting a discovery: PSRAM present + bridge running.
+bool MyMesh::neighborDiscoverReady(char* reply) {
+#if defined(ESP_PLATFORM)
+  if (!psramFound()) { strcpy(reply, "Err - PSRAM not available"); return false; }
+#endif
+  if (!bridge || !bridge->isRunning()) { strcpy(reply, "Err - MQTT bridge not running"); return false; }
+  return true;
+}
+
+// Snapshot the neighbor table into the overlay and fire one anon-regions query
+// per heard neighbour; arm the 30s scope-query window.
+bool MyMesh::startNeighborDiscover(char* reply) {
+  if (neighbor_discover_active) {
+    strcpy(reply, "Err - neighbor discover already active");
+    return false;
+  }
+  if (!neighborDiscoverReady(reply)) {
+    return false;  // reply already set
+  }
+
+  getLocalScopes(self_scopes_buf, sizeof(self_scopes_buf));
+  neighbor_discover_count = 0;
+  for (int i = 0; i < MAX_NEIGHBOURS; i++) {
+    if (neighbours[i].heard_timestamp > 0) {
+      neighbor_discover[neighbor_discover_count].neighbour_idx = (uint8_t)i;
+      neighbor_discover[neighbor_discover_count].scopes[0] = 0;
+      neighbor_discover[neighbor_discover_count].status = ND_PENDING;
+      uint32_t tag;
+      if (sendAnonRegionsReq(neighbours[i].id, tag)) {
+        neighbor_discover[neighbor_discover_count].tag = tag;
+      } else {
+        neighbor_discover[neighbor_discover_count].status = ND_SEND_FAILED;
+      }
+      neighbor_discover_count++;
+    }
+  }
+
+  neighbor_discover_active = true;
+  neighbor_discover_until = futureMillis(NEIGHBOR_DISCOVER_TIMEOUT_MS);
+
+  if (neighbor_discover_count == 0) {
+    finishNeighborDiscover();
+    strcpy(reply, "OK - neighbor discover started (0 neighbors, self only)");
+  } else {
+    sprintf(reply, "OK - neighbor discover started (%u neighbors)", (unsigned)neighbor_discover_count);
+  }
+  return true;
+}
+#endif // WITH_MQTT_NEIGHBORS
+
 // To check if there is pending work
 bool MyMesh::hasPendingWork() const {
 #if defined(WITH_BRIDGE)

examples/simple_repeater/MyMesh.h

@@ -147,6 +147,44 @@ class MyMesh : public mesh::Mesh, public CommonCLICallbacks
   uint8_t _wc_slot_restart_mask = 0;
 #endif
 
+#if defined(WITH_MQTT_NEIGHBORS)
+  // Neighbor-scope discovery: a snapshot of the neighbor table overlaid with an
+  // in-flight anon-regions query per neighbor, published to the MQTT neighbors
+  // topic once every neighbor has responded or the window times out.
+  enum NeighborDiscoverStatus : uint8_t {
+    ND_PENDING = 1,
+    ND_RESPONDED = 2,
+    ND_TIMEOUT = 3,
+    ND_SEND_FAILED = 4,
+  };
+  struct NeighborDiscoverEntry {
+    uint8_t neighbour_idx;   // index into neighbours[]
+    uint32_t tag;            // anon-regions request tag we're waiting on
+    char scopes[96];         // scope names from the response
+    uint8_t status;          // NeighborDiscoverStatus
+  };
+  NeighborDiscoverEntry neighbor_discover[MAX_NEIGHBOURS];
+  uint8_t neighbor_discover_count;
+  bool neighbor_discover_active;          // scope-query phase in flight
+  bool neighbor_table_refresh_active;     // zero-hop table refresh (stage 1) in flight
+  bool neighbor_table_refresh_periodic;   // that refresh was kicked by the periodic timer
+  unsigned long neighbor_discover_until;  // scope-query timeout deadline
+  unsigned long next_neighbors_publish;   // periodic publish deadline (0 = fire ASAP)
+  char self_scopes_buf[96];
+
+  bool sendAnonRegionsReq(const mesh::Identity& target, uint32_t& tag);
+  bool neighborDiscoverReady(char* reply);
+  bool startNeighborDiscover(char* reply);
+  void loopNeighborDiscover();
+  void finishNeighborDiscover();
+  bool handleNeighborDiscoverResponse(int overlay_idx, const uint8_t* data, size_t len);
+  void getLocalScopes(char* buf, size_t len);
+  // Overlay peer indices are offset by this base so onPeerDataRecv can tell a
+  // discovery response apart from a normal ACL-client index.
+  static const int NEIGHBOR_DISCOVER_PEER_BASE = 1000;
+  static const unsigned long NEIGHBOR_DISCOVER_TIMEOUT_MS = 30000;
+#endif
+
   void putNeighbour(const mesh::Identity& id, uint32_t timestamp, float snr);
   uint8_t handleLoginReq(const mesh::Identity& sender, const uint8_t* secret, uint32_t sender_timestamp, const uint8_t* data, bool is_flood);
   uint8_t handleAnonRegionsReq(const mesh::Identity& sender, uint32_t sender_timestamp, const uint8_t* data);

src/helpers/CommonCLI.cpp

@@ -807,6 +807,16 @@ void CommonCLI::loadMQTTPrefs(
     mqtt_rewrite_pending = true;
     MESH_DEBUG_PRINTLN("MQTT: Migrated observer settings from legacy /com_prefs trailing block");
   }
+
+  // Keep persisted values inside the signed-delta millis() scheduling window.
+  // This also repairs any manually-written or experimental value from firmware
+  // that briefly accepted intervals longer than the supported two-week cap.
+  if (_mqtt_prefs.mqtt_neighbors_interval < MQTT_NEIGHBORS_MIN_INTERVAL_MS ||
+      _mqtt_prefs.mqtt_neighbors_interval > MQTT_NEIGHBORS_MAX_INTERVAL_MS) {
+    _mqtt_prefs.mqtt_neighbors_interval = MQTT_NEIGHBORS_DEFAULT_INTERVAL_MS;
+    MESH_DEBUG_PRINTLN("MQTT: invalid neighbors interval reset to %u hours",
+                       (unsigned)MQTT_NEIGHBORS_DEFAULT_INTERVAL_HOURS);
+  }
   _legacy_tail.valid = false;
 
   if (mqtt_rewrite_pending) {

src/helpers/CommonCLI_Observer.cpp

@@ -263,6 +263,30 @@ bool CommonCLI::handleObserverSetCmd(uint32_t sender_timestamp, const char* conf
     } else {
       strcpy(reply, "Error: interval must be between 1-60 minutes");
     }
+#if defined(WITH_MQTT_NEIGHBORS)
+  } else if (memcmp(config, "mqtt.neighbors.interval ", 24) == 0) {
+    // Hours in, milliseconds stored. The 12-336h band keeps the interval under
+    // INT32_MAX so the mesh's wrap-safe signed-delta millis math stays valid.
+    uint32_t hours = _atoi(&config[24]);
+    if (hours >= MQTT_NEIGHBORS_MIN_INTERVAL_HOURS && hours <= MQTT_NEIGHBORS_MAX_INTERVAL_HOURS) {
+      _mqtt_prefs.mqtt_neighbors_interval = hours * 3600000UL;
+      savePrefs();
+      sprintf(reply, "OK - neighbors interval set to %u hours (%lu ms)", (unsigned)hours,
+              (unsigned long)_mqtt_prefs.mqtt_neighbors_interval);
+    } else {
+      strcpy(reply, "Error: neighbors interval must be between 12-336 hours");
+    }
+  } else if (memcmp(config, "mqtt.neighbors ", 15) == 0) {
+    // The mesh loop reads this live, so no bridge restart is needed; enabling it
+    // triggers a discovery on the next eligible loop pass.
+    _mqtt_prefs.mqtt_neighbors_enabled = memcmp(&config[15], "on", 2) == 0;
+    savePrefs();
+    strcpy(reply, "OK");
+#elif defined(WITH_MQTT_BRIDGE)
+  } else if (memcmp(config, "mqtt.neighbors.interval ", 24) == 0 ||
+             memcmp(config, "mqtt.neighbors ", 15) == 0) {
+    strcpy(reply, "Err - not supported (requires PSRAM)");
+#endif
   } else if (memcmp(config, "mqtt.ntp ", 9) == 0) {
     const char* host = &config[9];
     while (*host == ' ') host++;
@@ -738,6 +762,19 @@ bool CommonCLI::handleObserverGetCmd(uint32_t sender_timestamp, const char* conf
   } else if (memcmp(config, "mqtt.interval", 13) == 0) {
     uint32_t minutes = (_mqtt_prefs.mqtt_status_interval + 29999) / 60000;
     sprintf(reply, "> %u minutes (%lu ms)", minutes, (unsigned long)_mqtt_prefs.mqtt_status_interval);
+#if defined(WITH_MQTT_NEIGHBORS)
+  // Longer token first: a bare "mqtt.neighbors" (14) would otherwise swallow
+  // "mqtt.neighbors.interval" since the GET tokens carry no trailing space.
+  } else if (memcmp(config, "mqtt.neighbors.interval", 23) == 0) {
+    uint32_t hours = (_mqtt_prefs.mqtt_neighbors_interval + 3599999) / 3600000;
+    sprintf(reply, "> %u hours (%lu ms)", (unsigned)hours, (unsigned long)_mqtt_prefs.mqtt_neighbors_interval);
+  } else if (memcmp(config, "mqtt.neighbors", 14) == 0) {
+    sprintf(reply, "> %s", _mqtt_prefs.mqtt_neighbors_enabled ? "on" : "off");
+#elif defined(WITH_MQTT_BRIDGE)
+  } else if (memcmp(config, "mqtt.neighbors.interval", 23) == 0 ||
+             memcmp(config, "mqtt.neighbors", 14) == 0) {
+    strcpy(reply, "Err - not supported (requires PSRAM)");
+#endif
   } else if (memcmp(config, "mqtt.ntp.diag", 13) == 0 && (config[13] == '\0' || config[13] == ' ')) {
 #ifdef ESP_PLATFORM
     // Connectivity probe across all configured NTP servers; never updates the clock.

src/helpers/MQTTDefaults.h

@@ -100,6 +100,11 @@ static inline void applyMQTTDefaults(MQTTPrefs* prefs) {
   prefs->alert_wifi_minutes = 30;
   prefs->alert_mqtt_minutes = 240;
   prefs->alert_min_interval_min = 60;
+
+  // Neighbors publishing defaults off; a defaulted tail is a valid 24h interval
+  // (not 0) so an in-lineage upgrade from a pre-neighbors payload is sane.
+  prefs->mqtt_neighbors_enabled = 0;
+  prefs->mqtt_neighbors_interval = MQTT_NEIGHBORS_DEFAULT_INTERVAL_MS;
 }
 
 #endif // WITH_MQTT_BRIDGE

src/helpers/MQTTMessageBuilder.cpp

@@ -102,6 +102,22 @@ int MQTTMessageBuilder::buildRawMessage(
       origin, origin_id, timestamp, raw, buffer, buffer_size);
 }
 
+int MQTTMessageBuilder::buildNeighborsMessage(
+  JsonDocument& doc,
+  const char* origin,
+  const char* origin_id,
+  const char* timestamp,
+  const char* self_scopes,
+  const NeighborsMessageEntry* neighbors,
+  int neighbor_count,
+  char* buffer,
+  size_t buffer_size
+) {
+  return MQTTPayloadBuilder::buildNeighborsMessage(
+      doc, origin, origin_id, timestamp, self_scopes, neighbors, neighbor_count,
+      buffer, buffer_size);
+}
+
 int MQTTMessageBuilder::buildPacketJSON(
   JsonDocument& doc,
   mesh::Packet* packet,

src/helpers/MQTTMessageBuilder.h

@@ -2,6 +2,7 @@
 
 #include "MeshCore.h"
 #include <ArduinoJson.h>
+#include "MQTTPayloadBuilder.h"
 #include <Mesh.h>
 #include <Timezone.h>
 
@@ -153,6 +154,22 @@ public:
     size_t buffer_size
   );
 
+  // Neighbors table entry + JSON builder. The layout logic lives in the pure,
+  // host-tested MQTTPayloadBuilder; this is the firmware-facing alias/delegate,
+  // matching the status/packet/raw builders.
+  using NeighborsMessageEntry = MQTTPayloadBuilder::NeighborsMessageEntry;
+  static int buildNeighborsMessage(
+    JsonDocument& doc,
+    const char* origin,
+    const char* origin_id,
+    const char* timestamp,
+    const char* self_scopes,
+    const NeighborsMessageEntry* neighbors,
+    int neighbor_count,
+    char* buffer,
+    size_t buffer_size
+  );
+
   /**
    * Convert packet to JSON message
    *

src/helpers/MQTTPayloadBuilder.cpp

@@ -184,3 +184,48 @@ int MQTTPayloadBuilder::buildRawMessage(
 
   return serializeComplete(root, buffer, buffer_size);
 }
+
+int MQTTPayloadBuilder::buildNeighborsMessage(
+  JsonDocument& doc,
+  const char* origin,
+  const char* origin_id,
+  const char* timestamp,
+  const char* self_scopes,
+  const NeighborsMessageEntry* neighbors,
+  int neighbor_count,
+  char* buffer,
+  size_t buffer_size
+) {
+  if (!buffer || buffer_size == 0) return 0;
+
+  doc.clear();
+  JsonObject root = doc.to<JsonObject>();
+  root["timestamp"] = timestamp;
+  root["origin"] = origin;
+  root["origin_id"] = origin_id;
+
+  JsonObject self = root["self"].to<JsonObject>();
+  self["scopes"] = self_scopes ? self_scopes : "";
+
+  JsonArray arr = root["neighbors"].to<JsonArray>();
+  if (measureJson(root) >= buffer_size) return 0;
+
+  for (int i = 0; i < neighbor_count; i++) {
+    JsonObject nb = arr.add<JsonObject>();
+    nb["pubkey"] = neighbors[i].pubkey_hex;
+    nb["snr"] = neighbors[i].snr;
+    nb["heard_secs_ago"] = neighbors[i].heard_secs_ago;
+    nb["scopes"] = neighbors[i].scopes ? neighbors[i].scopes : "";
+    nb["status"] = neighbors[i].status;
+
+    // Entries arrive ordered most- to least-useful. Stop as soon as the next
+    // one would fill the fixed publish buffer, dropping the remaining tail so
+    // document growth stays bounded.
+    if (measureJson(root) >= buffer_size) {
+      arr.remove(arr.size() - 1);
+      break;
+    }
+  }
+
+  return serializeComplete(root, buffer, buffer_size);
+}

src/helpers/MQTTPayloadBuilder.h

@@ -68,5 +68,28 @@ public:
     char* buffer,
     size_t buffer_size
   );
+
+  struct NeighborsMessageEntry {
+    const char* pubkey_hex;
+    float snr;
+    uint32_t heard_secs_ago;
+    const char* scopes;
+    const char* status;
+  };
+
+  // Build neighbors-table JSON for the meshcore/{iata}/{device}/neighbors topic.
+  // Callers order entries most- to least-useful; document growth is bounded to
+  // buffer_size and the remaining tail is dropped once the next entry won't fit.
+  static int buildNeighborsMessage(
+    JsonDocument& doc,
+    const char* origin,
+    const char* origin_id,
+    const char* timestamp,
+    const char* self_scopes,
+    const NeighborsMessageEntry* neighbors,
+    int neighbor_count,
+    char* buffer,
+    size_t buffer_size
+  );
 };
 

src/helpers/MQTTPrefsCodec.h

@@ -36,6 +36,7 @@ struct DecodePlan {
 };
 
 static const size_t kV1PreObserverPayloadSize = MQTT_PREFS_V1_PRE_OBSERVER_PAYLOAD_SIZE;
+static const size_t kV1PreNeighborsPayloadSize = MQTT_PREFS_V1_PRE_NEIGHBORS_PAYLOAD_SIZE;
 static const size_t kV1BaselinePayloadSize = MQTT_PREFS_V1_FULL_PAYLOAD_SIZE;
 static const size_t kEncodedSize = sizeof(MQTTPrefsHeader) + kV1BaselinePayloadSize;
 
@@ -99,6 +100,12 @@ inline DecodePlan classify(const uint8_t* prefix, size_t prefix_read, size_t fil
       if (header.payload_len == kV1BaselinePayloadSize) {
         return {Source::Current, false, false, true, kV1BaselinePayloadSize};
       }
+      if (header.payload_len == kV1PreNeighborsPayloadSize) {
+        // Written by observer/webconfig firmware before the neighbors tail
+        // existed. The observer fields ARE present; only the neighbors tail is
+        // missing, so it loads and keeps its defaults (off / 24h).
+        return {Source::Current, false, false, true, kV1PreNeighborsPayloadSize};
+      }
       if (header.payload_len == kV1PreObserverPayloadSize) {
         return {Source::Current, false, false, false, kV1PreObserverPayloadSize};
       }

src/helpers/MQTTPrefsStorage.h

@@ -108,12 +108,35 @@ struct MQTTPrefs {
   uint16_t alert_min_interval_min;
   char alert_hashtag[24];
   char alert_region[31];
+
+  // Neighbors publishing (PSRAM boards only). Appended at the end of the
+  // observer tail so a shorter (pre-neighbors) /mqtt_prefs payload from earlier
+  // firmware still loads with these defaulting off/24h; keeps the format at
+  // VERSION 1. Field order and sizes are kept byte-identical to the flex
+  // neighbors build so a /mqtt_prefs written by either firmware is
+  // interchangeable (see the offsetof static_asserts below).
+  uint8_t mqtt_neighbors_enabled;
+  uint32_t mqtt_neighbors_interval;
 };
 
-// Version-1 has exactly two layouts this firmware can decode. Never infer a
+// Neighbor discovery is scheduled with the wrap-safe millis() helpers, whose
+// signed-delta comparison requires intervals below INT32_MAX ms. The 336h
+// (two-week) cap stays comfortably inside that range.
+static const uint32_t MQTT_NEIGHBORS_MIN_INTERVAL_HOURS = 12;
+static const uint32_t MQTT_NEIGHBORS_MAX_INTERVAL_HOURS = 336;
+static const uint32_t MQTT_NEIGHBORS_DEFAULT_INTERVAL_HOURS = 24;
+static const uint32_t MQTT_NEIGHBORS_MIN_INTERVAL_MS = MQTT_NEIGHBORS_MIN_INTERVAL_HOURS * 3600000UL;
+static const uint32_t MQTT_NEIGHBORS_MAX_INTERVAL_MS = MQTT_NEIGHBORS_MAX_INTERVAL_HOURS * 3600000UL;
+static const uint32_t MQTT_NEIGHBORS_DEFAULT_INTERVAL_MS = MQTT_NEIGHBORS_DEFAULT_INTERVAL_HOURS * 3600000UL;
+
+// Version-1 has three payload layouts this firmware can decode. Never infer a
 // compatible payload from an arbitrary shorter size: raw prefs have no checksum.
+//   - PRE_OBSERVER  (2736): stops before the observer tail (snmp_*/alert_*).
+//   - PRE_NEIGHBORS (2860): full observer tail, no neighbors fields yet.
+//   - FULL          (2864): current baseline, with the neighbors tail.
 static const size_t MQTT_PREFS_V1_PRE_OBSERVER_PAYLOAD_SIZE = 2736;
-static const size_t MQTT_PREFS_V1_FULL_PAYLOAD_SIZE = 2860;
+static const size_t MQTT_PREFS_V1_PRE_NEIGHBORS_PAYLOAD_SIZE = 2860;
+static const size_t MQTT_PREFS_V1_FULL_PAYLOAD_SIZE = 2864;
 
 // /mqtt_prefs starts with a self-describing 8-byte header. Headerless files
 // are deployed legacy layouts and continue to be distinguished by size.
@@ -234,6 +257,15 @@ static_assert(offsetof(MQTTPrefs, snmp_enabled) == MQTT_PREFS_V1_PRE_OBSERVER_PA
               "v1 pre-observer /mqtt_prefs boundary changed");
 static_assert(sizeof(MQTTPrefs) == MQTT_PREFS_V1_FULL_PAYLOAD_SIZE,
               "v1 /mqtt_prefs payload layout changed");
+// Lock the neighbors tail to the flex neighbors build's layout so a /mqtt_prefs
+// written by either firmware is byte-for-byte interchangeable. The enable flag
+// lands in the old struct's zeroed trailing padding (offset 2857), and the
+// interval begins exactly at the pre-neighbors payload size (2860) so a
+// pre-neighbors read stops right before it and the interval keeps its default.
+static_assert(offsetof(MQTTPrefs, mqtt_neighbors_enabled) == 2857,
+              "neighbors enable flag must sit at the flex-compatible offset");
+static_assert(offsetof(MQTTPrefs, mqtt_neighbors_interval) == MQTT_PREFS_V1_PRE_NEIGHBORS_PAYLOAD_SIZE,
+              "neighbors interval offset must equal the pre-neighbors payload size");
 static_assert(sizeof(OldMQTTPrefs) == 472, "frozen pre-slot /mqtt_prefs layout changed");
 static_assert(sizeof(PreWifiPowerOldMQTTPrefs) == 472, "frozen pre-WiFi-power /mqtt_prefs layout changed");
 static_assert(offsetof(OldMQTTPrefs, wifi_power_save) == 144,

src/helpers/MQTTTopicRouter.h

@@ -14,6 +14,7 @@ enum MQTTPublicationType {
   MQTT_PUBLICATION_STATUS = 0,
   MQTT_PUBLICATION_PACKETS = 1,
   MQTT_PUBLICATION_RAW = 2,
+  MQTT_PUBLICATION_NEIGHBORS = 3,
 };
 
 enum MQTTTopicRouteStyle {
@@ -31,6 +32,7 @@ static inline const char* mqttPublicationTypeName(int type) {
     case MQTT_PUBLICATION_STATUS: return "status";
     case MQTT_PUBLICATION_PACKETS: return "packets";
     case MQTT_PUBLICATION_RAW: return "raw";
+    case MQTT_PUBLICATION_NEIGHBORS: return "neighbors";
     default: return NULL;
   }
 }
@@ -45,7 +47,8 @@ static inline bool mqttWriteTopic(char* buf, size_t buf_size, const char* format
 }
 
 // Build the complete topic for one publication. MeshRank is deliberately
-// packets-only; status and raw are unsupported by the current broker contract.
+// packets-only; status, raw, and neighbors are unsupported by the current
+// broker contract (the type != PACKETS guard below rejects them all).
 // MeshCore routes require a configured IATA and device id. Custom templates may
 // omit either placeholder, so their individual values are allowed to be empty.
 static inline bool mqttBuildPublicationTopic(MQTTTopicRouteStyle style, int type,

src/helpers/WebConfigKeys.h

@@ -23,7 +23,8 @@ static const char* const WC_ALLOWED_SET_KEYS[] = {
   // MQTTPrefs (WiFi / MQTT / misc observer)
   "wifi.ssid", "wifi.pwd", "wifi.powersave",
   "mqtt.origin", "mqtt.iata", "mqtt.status", "mqtt.packets", "mqtt.raw",
-  "mqtt.tx", "mqtt.rx", "mqtt.interval", "mqtt.ntp", "mqtt.owner", "mqtt.email",
+  "mqtt.tx", "mqtt.rx", "mqtt.interval", "mqtt.neighbors", "mqtt.neighbors.interval",
+  "mqtt.ntp", "mqtt.owner", "mqtt.email",
   "timezone", "timezone.offset", "snmp", "snmp.community",
 };
 static const char* const WC_ALLOWED_SLOT_KEYS[] = {

src/helpers/bridges/MQTTBridge.cpp

@@ -205,6 +205,22 @@ unsigned long MQTTBridge::getWifiConnectedAtMillis() {
   return s_wifi_connected_at;
 }
 
+#if defined(WITH_MQTT_NEIGHBORS)
+// Compact "time remaining" for the `get mqtt.status` nbr field: "3h12m" / "12m" / "45s".
+static void formatDuration(char* buf, size_t len, uint32_t secs) {
+  if (!buf || len == 0) return;
+  uint32_t h = secs / 3600;
+  uint32_t m = (secs % 3600) / 60;
+  if (h > 0) {
+    snprintf(buf, len, "%uh%um", (unsigned)h, (unsigned)m);
+  } else if (m > 0) {
+    snprintf(buf, len, "%um", (unsigned)m);
+  } else {
+    snprintf(buf, len, "%us", (unsigned)secs);
+  }
+}
+#endif
+
 void MQTTBridge::formatMqttStatusReply(char* buf, size_t bufsize, const MQTTPrefs* obs) {
   if (buf == nullptr || bufsize == 0) return;
   const char* msgs = (obs && obs->mqtt_status_enabled) ? "on" : "off";
@@ -251,7 +267,33 @@ void MQTTBridge::formatMqttStatusReply(char* buf, size_t bufsize, const MQTTPref
     }
     pos += snprintf(buf + pos, bufsize - pos, ", %d: %s (%s)", i + 1, name, state);
   }
-  snprintf(buf + pos, bufsize - pos, ", q:%d", q);
+  // snprintf returns the would-be length, so a full buffer can push pos past
+  // bufsize; clamp before the remaining appends so bufsize - pos can't underflow.
+  if (pos >= (int)bufsize) pos = (int)bufsize - 1;
+  pos += snprintf(buf + pos, bufsize - pos, ", q:%d", q);
+  if (pos >= (int)bufsize) pos = (int)bufsize - 1;
+
+#if defined(WITH_MQTT_NEIGHBORS)
+  // Periodic neighbors: time to next publish + how the last one went.
+  if (obs && obs->mqtt_neighbors_enabled && pos < (int)bufsize - 1) {
+    char when[16];
+    switch (b->_neighbors_phase.load(std::memory_order_relaxed)) {
+      case NBR_ACTIVE: strcpy(when, "active"); break;
+      case NBR_DUE:    strcpy(when, "due"); break;
+      default:
+        formatDuration(when, sizeof(when),
+                       b->_neighbors_secs_until_next.load(std::memory_order_relaxed));
+        break;
+    }
+    const char* last;
+    switch (b->_neighbors_last_result.load(std::memory_order_relaxed)) {
+      case NBR_RESULT_OK:   last = "ok"; break;
+      case NBR_RESULT_FAIL: last = "failed"; break;
+      default:              last = "none"; break;
+    }
+    snprintf(buf + pos, bufsize - pos, ", nbr: %s/%s", when, last);
+  }
+#endif
 }
 
 // On-demand publish-health + heap snapshot for the `get mqtt.stats` CLI command.
@@ -589,6 +631,17 @@ MQTTBridge::MQTTBridge(NodePrefs *prefs, MQTTPrefs *obs, mesh::PacketManager *mg
   _ntp_diag_done = false;
   _ntp_diag_count = 0;
 
+#if defined(WITH_MQTT_NEIGHBORS)
+  // Neighbors publish handoff (buffer allocated in begin() after PSRAM probe).
+  // std::atomic has no value-initializing default ctor pre-C++20, so set them here.
+  _neighbors_json_buffer = nullptr;
+  _neighbors_publish_len = 0;
+  _neighbors_publish_pending.store(false, std::memory_order_relaxed);
+  _neighbors_last_result.store(NBR_RESULT_NONE, std::memory_order_relaxed);
+  _neighbors_phase.store(NBR_SCHEDULED, std::memory_order_relaxed);
+  _neighbors_secs_until_next.store(0, std::memory_order_relaxed);
+#endif
+
   // Initialize JWT username
   _jwt_username[0] = '\0';
 
@@ -626,6 +679,13 @@ void MQTTBridge::allocateRuntimeBuffers() {
       _publish_json_buffer, PUBLISH_JSON_BUFFER_SIZE, psram_malloc));
   _status_json_buffer = static_cast<char*>(MQTTRuntimeBufferLifecycle::allocateIfMissing(
       _status_json_buffer, STATUS_JSON_BUFFER_SIZE, psram_malloc));
+#if defined(WITH_MQTT_NEIGHBORS)
+  // Persistent neighbors JSON buffer. Unlike status/packet there is no stack
+  // fallback: the feature is PSRAM-gated, so a nullptr simply disables publishing
+  // (requestPublishNeighbors/publishNeighbors both no-op on nullptr).
+  _neighbors_json_buffer = static_cast<char*>(MQTTRuntimeBufferLifecycle::allocateIfMissing(
+      _neighbors_json_buffer, NEIGHBORS_JSON_BUFFER_SIZE, psram_malloc));
+#endif
   MQTT_DEBUG_PRINTLN("Runtime buffers: raw=%s publish=%s status=%s",
       _last_raw_data ? "PSRAM" : "unavailable",
       _publish_json_buffer ? "PSRAM" : "stack fallback",
@@ -641,6 +701,12 @@ void MQTTBridge::releaseRuntimeBuffers() {
       _publish_json_buffer, psram_free));
   _status_json_buffer = static_cast<char*>(MQTTRuntimeBufferLifecycle::release(
       _status_json_buffer, psram_free));
+#if defined(WITH_MQTT_NEIGHBORS)
+  _neighbors_json_buffer = static_cast<char*>(MQTTRuntimeBufferLifecycle::release(
+      _neighbors_json_buffer, psram_free));
+  _neighbors_publish_len = 0;
+  _neighbors_publish_pending.store(false, std::memory_order_release);
+#endif
   #endif
 
   // Never pair a newly allocated raw buffer with metadata from a prior bridge
@@ -1285,6 +1351,25 @@ void MQTTBridge::mqttTaskLoop() {
     // Process packet queue
     processPacketQueue();
 
+#if defined(WITH_MQTT_NEIGHBORS)
+    // Consume a pending neighbors snapshot handed over by the mesh (Core 1).
+    // The pending flag stays raised across the whole publish so a second
+    // request is rejected until this one completes (see requestPublishNeighbors).
+    if (_neighbors_publish_pending.load(std::memory_order_acquire)) {
+      bool ok = publishNeighbors();
+      _neighbors_last_result.store(ok ? NBR_RESULT_OK : NBR_RESULT_FAIL,
+                                   std::memory_order_relaxed);
+      // MQTT_DEBUG_PRINTLN concatenates its format as a string literal, so the
+      // argument must be a literal, not a ternary expression.
+      if (ok) {
+        MQTT_DEBUG_PRINTLN("Neighbors published");
+      } else {
+        MQTT_DEBUG_PRINTLN("Neighbors publish failed");
+      }
+      _neighbors_publish_pending.store(false, std::memory_order_release);
+    }
+#endif
+
 #ifdef WITH_SNMP
     // SNMP agent loop — process incoming UDP requests
     if (_snmp_agent) {
@@ -2072,7 +2157,8 @@ bool MQTTBridge::buildTopicForSlot(int index, MQTTMessageType type, char* topic_
   static_assert(
       static_cast<int>(MSG_STATUS) == MQTT_PUBLICATION_STATUS &&
       static_cast<int>(MSG_PACKETS) == MQTT_PUBLICATION_PACKETS &&
-      static_cast<int>(MSG_RAW) == MQTT_PUBLICATION_RAW,
+      static_cast<int>(MSG_RAW) == MQTT_PUBLICATION_RAW &&
+      static_cast<int>(MSG_NEIGHBORS) == MQTT_PUBLICATION_NEIGHBORS,
       "topic router enum drift");
 
   if (!mqttTopicSlotIndexValid(index, RUNTIME_MQTT_SLOTS)) return false;
@@ -2177,7 +2263,12 @@ void MQTTBridge::publishStatusToSlot(int index) {
   );
 
   if (len > 0) {
-    int result = slot.client->publish(status_topic, 1, true, json_buffer, strlen(json_buffer));
+    // Honor the preset's retain policy, matching publishStatus() — brokers that
+    // set allow_retain=false (e.g. waev) reject retained publishes, so this
+    // on-connect status must not force retain=true. Custom slots default to
+    // non-retained here too, keeping both status paths consistent.
+    bool use_retain = slot.preset ? slot.preset->allow_retain : false;
+    int result = slot.client->publish(status_topic, 1, use_retain, json_buffer, strlen(json_buffer));
     if (result <= 0) {
       MQTT_DEBUG_PRINTLN("MQTT%d status publish failed", index + 1);
     }
@@ -3138,6 +3229,54 @@ bool MQTTBridge::publishRaw(mesh::Packet* packet) {
   return false;
 }
 
+#if defined(WITH_MQTT_NEIGHBORS)
+// ---------------------------------------------------------------------------
+// Periodic neighbors publication
+// ---------------------------------------------------------------------------
+
+void MQTTBridge::setNeighborsSchedule(NeighborsPhase phase, uint32_t secs_until_next) {
+  _neighbors_phase.store((uint8_t)phase, std::memory_order_relaxed);
+  _neighbors_secs_until_next.store(secs_until_next, std::memory_order_relaxed);
+}
+
+void MQTTBridge::requestPublishNeighbors(const char* json, size_t len) {
+  if (!_neighbors_json_buffer || !json || len == 0) return;
+  // Drop a new snapshot while one is still being published (Core 0 clears the
+  // flag when done). Acquire pairs with the task loop's release store.
+  if (_neighbors_publish_pending.load(std::memory_order_acquire)) return;
+  if (len >= NEIGHBORS_JSON_BUFFER_SIZE) {
+    len = NEIGHBORS_JSON_BUFFER_SIZE - 1;
+  }
+  memcpy(_neighbors_json_buffer, json, len);
+  _neighbors_json_buffer[len] = '\0';
+  _neighbors_publish_len = len;
+  _neighbors_publish_pending.store(true, std::memory_order_release);
+}
+
+bool MQTTBridge::publishNeighbors() {
+  if (!_neighbors_json_buffer || _neighbors_publish_len == 0) return false;
+  if (!_cached_has_connected_slots) return false;
+
+  refreshOriginFromPrefs();
+
+  bool published = false;
+  char topic[128];
+  for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) {
+    if (_slots[i].enabled && _slots[i].client && _slots[i].connected) {
+      // MeshRank slots reject non-packets by contract, so buildTopicForSlot
+      // returns false for them here and the slot is skipped.
+      if (buildTopicForSlot(i, MSG_NEIGHBORS, topic, sizeof(topic))) {
+        bool use_retain = _slots[i].preset ? _slots[i].preset->allow_retain : false;
+        if (publishToSlot(i, topic, _neighbors_json_buffer, use_retain, 1)) {
+          published = true;
+        }
+      }
+    }
+  }
+  return published;
+}
+#endif  // WITH_MQTT_NEIGHBORS
+
 // ---------------------------------------------------------------------------
 // Queue management
 // ---------------------------------------------------------------------------

src/helpers/bridges/MQTTBridge.h

@@ -10,6 +10,7 @@
 #include "helpers/JWTHelper.h"
 #include "helpers/MQTTPresets.h"
 #include "helpers/MQTTLifecycle.h"
+#include <atomic>
 
 #ifdef WITH_SNMP
 class MeshSNMPAgent;  // Forward declaration
@@ -36,6 +37,14 @@ class MeshSNMPAgent;  // Forward declaration
 
 #ifdef WITH_MQTT_BRIDGE
 
+// Periodic neighbors publication is PSRAM-only: it needs a persistent ~10 KB JSON
+// buffer plus a second transient one while the mesh builds the table, and it keys
+// off the mesh neighbor cache (sized by MAX_NEIGHBOURS). Every neighbors-specific
+// member, method, and code block in this bridge is gated on WITH_MQTT_NEIGHBORS.
+#if defined(BOARD_HAS_PSRAM) && defined(MAX_NEIGHBOURS) && MAX_NEIGHBOURS > 0
+#define WITH_MQTT_NEIGHBORS 1
+#endif
+
 /**
  * @brief Bridge implementation using MQTT protocol for packet transport
  *
@@ -276,6 +285,26 @@ private:
   char _status_json_buffer[STATUS_JSON_BUFFER_SIZE];
   #endif
 
+#if defined(WITH_MQTT_NEIGHBORS)
+  // Persistent PSRAM copy of the neighbors-table JSON. The mesh (Core 1) builds
+  // the payload into its own transient buffer, hands it here via
+  // requestPublishNeighbors(), and the MQTT task (Core 0) publishes this copy.
+  // Allocated in allocateRuntimeBuffers()/freed in releaseRuntimeBuffers() like
+  // the other PSRAM buffers (nullptr if the allocation failed).
+  char* _neighbors_json_buffer;
+  size_t _neighbors_publish_len;
+  // Release/acquire handoff from the mesh loop (Core 1) to the MQTT task (Core 0).
+  // A second snapshot is dropped while the current one is still publishing.
+  std::atomic<bool> _neighbors_publish_pending;
+  // Written by the MQTT task (Core 0), read by the CLI (Core 1) for `get mqtt.status`.
+  enum NeighborsResult : uint8_t { NBR_RESULT_NONE, NBR_RESULT_OK, NBR_RESULT_FAIL };
+  std::atomic<uint8_t> _neighbors_last_result;
+  // Written by the mesh loop (Core 1), read by the CLI (Core 1). Cached schedule
+  // summary so the wrap-safe millis math stays on the mesh side that owns the timer.
+  std::atomic<uint8_t> _neighbors_phase;
+  std::atomic<uint32_t> _neighbors_secs_until_next;
+#endif
+
   // JSON document scratch space — inline StaticJsonDocument keeps the pool off the MQTT
   // task stack and eliminates two separate heap allocations (fragmentation reduction).
   StaticJsonDocument<PUBLISH_JSON_BUFFER_SIZE> _packet_json_doc;
@@ -328,7 +357,7 @@ private:
   mesh::MillisecondClock* _ms;    // For uptime
 
   // Topic building
-  enum MQTTMessageType { MSG_STATUS, MSG_PACKETS, MSG_RAW };
+  enum MQTTMessageType { MSG_STATUS, MSG_PACKETS, MSG_RAW, MSG_NEIGHBORS };
   bool buildTopicForSlot(int index, MQTTMessageType type, char* topic_buf, size_t buf_size);
   bool substituteTopicTemplate(const char* tmpl, MQTTMessageType type, int slot_index, char* buf, size_t buf_size);
 
@@ -370,6 +399,11 @@ private:
                      const uint8_t* raw_data = nullptr, int raw_len = 0,
                      float snr = 0.0f, float rssi = 0.0f);
   bool publishRaw(mesh::Packet* packet);
+#if defined(WITH_MQTT_NEIGHBORS)
+  // Publishes the pending _neighbors_json_buffer to every connected slot's
+  // neighbors topic. Runs on the MQTT task (Core 0) only.
+  bool publishNeighbors();
+#endif
   void queuePacket(mesh::Packet* packet, bool is_tx);
   void dequeuePacket();
   bool isAnySlotConnected();
@@ -459,6 +493,29 @@ public:
   void setBuildDate(const char* build_date);
   void storeRawRadioData(const uint8_t* raw_data, int len, float snr, float rssi);
   void setMessageTypes(bool status, bool packets, bool raw);
+
+#if defined(WITH_MQTT_NEIGHBORS)
+  // Single source of truth for the neighbors JSON size, used by both the bridge's
+  // persistent buffer and the mesh's transient build buffer.
+  static const size_t NEIGHBORS_JSON_BUFFER_SIZE = 10240;
+
+  // Called by the mesh (Core 1) once a neighbor-discovery pass has built the
+  // table JSON. Copies it into the persistent PSRAM buffer and raises the
+  // publish-pending flag for the MQTT task; a request is dropped if one is
+  // already in flight or the buffer is unavailable.
+  void requestPublishNeighbors(const char* json, size_t len);
+
+  // Periodic-neighbors schedule, reported by the mesh loop for `get mqtt.status`.
+  // The mesh owns the timer; the bridge only caches the summary so the wrap-safe
+  // millis math stays on the side that already has those helpers.
+  enum NeighborsPhase : uint8_t {
+    NBR_SCHEDULED,  // waiting for the next publish; secs_until_next is valid
+    NBR_ACTIVE,     // zero-hop refresh or scope queries in flight
+    NBR_DUE,        // publish is due, waiting on the bridge/WiFi to come up
+  };
+  void setNeighborsSchedule(NeighborsPhase phase, uint32_t secs_until_next);
+#endif
+
   int getConnectedBrokers() const;
   int getQueueSize() const;
   bool isReady() const;

src/helpers/esp32/WebConfigServer.cpp

@@ -591,6 +591,8 @@ void WebConfigServer::handleConfigGet(AsyncWebServerRequest* req) {
                  : _obs->mqtt_tx_enabled == 1 ? "on" : "off";
     mqtt["rx"] = (bool)_obs->mqtt_rx_enabled;
     mqtt["interval"] = _obs->mqtt_status_interval / 60000;  // CLI takes minutes
+    mqtt["neighbors"] = (bool)_obs->mqtt_neighbors_enabled;
+    mqtt["neighbors_interval"] = _obs->mqtt_neighbors_interval / 3600000UL;  // CLI takes hours
     mqtt["timezone"] = (const char*)_obs->timezone_string;
     mqtt["timezone_offset"] = _obs->timezone_offset;
     mqtt["ntp"] = (const char*)_obs->mqtt_ntp_server;

test/test_mqtt_payload_builder/test_mqtt_payload_builder.cpp

@@ -227,6 +227,87 @@ TEST(MQTTPayloadBuilder, MaximumRepresentativePacketAndRawPayloadsRemainValid) {
   EXPECT_EQ(510U, strlen(parsed_raw["data"].as<const char*>()));
 }
 
+TEST(MQTTPayloadBuilder, NeighborsMessageRoundTripsSelfAndEntries) {
+  MQTTPayloadBuilder::NeighborsMessageEntry neighbors[] = {
+      {"0011223344556677", 9.75f, 42, "DEN,APRS", "active"},
+      {"8899AABBCCDDEEFF", -3.5f, 3600, "", "stale"},
+  };
+
+  JsonDocument scratch;
+  char buffer[1024];
+  int len = MQTTPayloadBuilder::buildNeighborsMessage(
+      scratch, "DEN Repeater", "0123456789ABCDEF", kTimestamp, "DEN,APRS",
+      neighbors, 2, buffer, sizeof(buffer));
+
+  ASSERT_GT(len, 0);
+  EXPECT_EQ(static_cast<size_t>(len), strlen(buffer));
+  JsonDocument parsed;
+  ASSERT_FALSE(deserializeJson(parsed, buffer));
+  EXPECT_STREQ("2026-07-18T12:34:56.123456+00:00", parsed["timestamp"].as<const char*>());
+  EXPECT_STREQ("DEN Repeater", parsed["origin"].as<const char*>());
+  EXPECT_STREQ("0123456789ABCDEF", parsed["origin_id"].as<const char*>());
+  EXPECT_STREQ("DEN,APRS", parsed["self"]["scopes"].as<const char*>());
+
+  JsonArray arr = parsed["neighbors"].as<JsonArray>();
+  ASSERT_EQ(2U, arr.size());
+  EXPECT_STREQ("0011223344556677", arr[0]["pubkey"].as<const char*>());
+  EXPECT_FLOAT_EQ(9.75f, arr[0]["snr"].as<float>());
+  EXPECT_EQ(42U, arr[0]["heard_secs_ago"].as<uint32_t>());
+  EXPECT_STREQ("DEN,APRS", arr[0]["scopes"].as<const char*>());
+  EXPECT_STREQ("active", arr[0]["status"].as<const char*>());
+  EXPECT_STREQ("8899AABBCCDDEEFF", arr[1]["pubkey"].as<const char*>());
+  EXPECT_STREQ("", arr[1]["scopes"].as<const char*>());
+  EXPECT_STREQ("stale", arr[1]["status"].as<const char*>());
+}
+
+TEST(MQTTPayloadBuilder, NeighborsMessageHandlesEmptyTableAndNullScopes) {
+  JsonDocument scratch;
+  char buffer[256];
+  int len = MQTTPayloadBuilder::buildNeighborsMessage(
+      scratch, "node", "id", kTimestamp, nullptr, nullptr, 0,
+      buffer, sizeof(buffer));
+
+  ASSERT_GT(len, 0);
+  JsonDocument parsed;
+  ASSERT_FALSE(deserializeJson(parsed, buffer));
+  EXPECT_STREQ("", parsed["self"]["scopes"].as<const char*>());
+  JsonArray arr = parsed["neighbors"].as<JsonArray>();
+  ASSERT_TRUE(arr.isNull() == false);
+  EXPECT_EQ(0U, arr.size());
+}
+
+TEST(MQTTPayloadBuilder, NeighborsMessageDropsTailWhenBufferFills) {
+  // Twenty entries far exceed a tight buffer; the builder must emit a prefix
+  // that still parses as complete JSON rather than truncating mid-document.
+  MQTTPayloadBuilder::NeighborsMessageEntry neighbors[20];
+  static char keys[20][17];
+  for (int i = 0; i < 20; i++) {
+    snprintf(keys[i], sizeof(keys[i]), "%016X", i);
+    neighbors[i].pubkey_hex = keys[i];
+    neighbors[i].snr = static_cast<float>(i);
+    neighbors[i].heard_secs_ago = static_cast<uint32_t>(i) * 10U;
+    neighbors[i].scopes = "DEN";
+    neighbors[i].status = "active";
+  }
+
+  JsonDocument scratch;
+  char buffer[512];
+  int len = MQTTPayloadBuilder::buildNeighborsMessage(
+      scratch, "node", "id", kTimestamp, "DEN", neighbors, 20,
+      buffer, sizeof(buffer));
+
+  ASSERT_GT(len, 0);
+  EXPECT_LT(static_cast<size_t>(len), sizeof(buffer));
+  JsonDocument parsed;
+  ASSERT_FALSE(deserializeJson(parsed, buffer));
+  JsonArray arr = parsed["neighbors"].as<JsonArray>();
+  ASSERT_FALSE(arr.isNull());
+  EXPECT_GT(arr.size(), 0U);
+  EXPECT_LT(arr.size(), 20U);
+  // Kept entries are the head of the input, in order.
+  EXPECT_STREQ(keys[0], arr[0]["pubkey"].as<const char*>());
+}
+
 }  // namespace
 
 int main(int argc, char** argv) {

test/test_mqtt_prefs_codec/test_mqtt_prefs_codec.cpp

@@ -226,6 +226,8 @@ TEST(MQTTPrefsCodec, CurrentVersionedPayloadRoundTripsExactly) {
   strncpy(source.mqtt_origin, "current-node", sizeof(source.mqtt_origin) - 1);
   strncpy(source.mqtt_slot_password[2], "preserve-me", sizeof(source.mqtt_slot_password[2]) - 1);
   strncpy(source.alert_region, "PNW", sizeof(source.alert_region) - 1);
+  source.mqtt_neighbors_enabled = 1;
+  source.mqtt_neighbors_interval = MQTT_NEIGHBORS_MAX_INTERVAL_MS;
   std::vector<uint8_t> bytes(Codec::kEncodedSize);
   ASSERT_EQ(Codec::kEncodedSize, Codec::encode(source, bytes.data(), bytes.size()));
 
@@ -266,6 +268,44 @@ TEST(MQTTPrefsCodec, CompatibleShortV1PayloadPreservesDefaultsBeyondObserverBoun
   EXPECT_EQ(5, loaded.radio_watchdog_minutes);
 }
 
+TEST(MQTTPrefsCodec, PreNeighborsV1PayloadLoadsObserverFieldsAndDefaultsNeighborsTail) {
+  // A /mqtt_prefs written by observer/webconfig firmware before the neighbors
+  // tail existed: full observer fields, 2860-byte v1 payload. It must still load
+  // as Current (observer fields present) with the neighbors tail defaulted.
+  MQTTPrefs source = defaults();
+  strncpy(source.mqtt_origin, "pre-neighbors-node", sizeof(source.mqtt_origin) - 1);
+  strncpy(source.alert_region, "PNW", sizeof(source.alert_region) - 1);
+  source.snmp_enabled = 1;
+  source.alert_enabled = 1;
+  source.mqtt_neighbors_enabled = 0;            // old struct's byte 2857 was zero padding
+  source.mqtt_neighbors_interval = 0x11223344;  // must NOT survive a 2860-byte read
+
+  std::vector<uint8_t> bytes(sizeof(MQTTPrefsHeader) + Codec::kV1PreNeighborsPayloadSize, 0);
+  writeHeader(&bytes, MQTT_PREFS_VERSION,
+              static_cast<uint16_t>(Codec::kV1PreNeighborsPayloadSize));
+  memcpy(bytes.data() + sizeof(MQTTPrefsHeader), &source, Codec::kV1PreNeighborsPayloadSize);
+
+  const Codec::DecodePlan plan = classify(bytes);
+  ASSERT_EQ(Codec::Source::Current, plan.source);
+  ASSERT_EQ(Codec::kV1PreNeighborsPayloadSize, plan.payload_len);
+  ASSERT_FALSE(plan.preserve_file);
+  ASSERT_TRUE(plan.observer_fields_present);
+
+  MQTTPrefs loaded = defaults();
+  loaded.mqtt_neighbors_enabled = 1;                                   // pretend stale
+  loaded.mqtt_neighbors_interval = MQTT_NEIGHBORS_DEFAULT_INTERVAL_MS;  // caller's defaulted tail
+  memcpy(&loaded, bytes.data() + sizeof(MQTTPrefsHeader), plan.payload_len);
+
+  EXPECT_STREQ("pre-neighbors-node", loaded.mqtt_origin);
+  EXPECT_STREQ("PNW", loaded.alert_region);
+  EXPECT_EQ(1, loaded.snmp_enabled);
+  EXPECT_EQ(1, loaded.alert_enabled);
+  // Enable flag sits at offset 2857 (inside the 2860 read) -> takes the file's 0.
+  // Interval begins at 2860 (beyond the read) -> keeps the caller's default.
+  EXPECT_EQ(0u, loaded.mqtt_neighbors_enabled);
+  EXPECT_EQ(MQTT_NEIGHBORS_DEFAULT_INTERVAL_MS, loaded.mqtt_neighbors_interval);
+}
+
 TEST(MQTTPrefsCodec, CorruptOrShortVersionedInputsArePreserved) {
   Codec::DecodePlan plan = Codec::classify(nullptr, 0, 0);
   EXPECT_EQ(Codec::Source::Corrupt, plan.source);

test/test_mqtt_topic_router/test_mqtt_topic_router.cpp

@@ -24,6 +24,7 @@ const TypeCase kTypes[] = {
   {MQTT_PUBLICATION_STATUS, "status"},
   {MQTT_PUBLICATION_PACKETS, "packets"},
   {MQTT_PUBLICATION_RAW, "raw"},
+  {MQTT_PUBLICATION_NEIGHBORS, "neighbors"},
 };
 
 TEST(MQTTTopicRouter, EveryMeshCorePresetSupportsEveryPublicationType) {
@@ -57,6 +58,9 @@ TEST(MQTTTopicRouter, MeshRankContractIsPacketsOnly) {
   EXPECT_FALSE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHRANK, MQTT_PUBLICATION_RAW,
                                          nullptr, IATA, DEVICE, TOKEN, topic, sizeof(topic)));
   EXPECT_STREQ("", topic);
+  EXPECT_FALSE(mqttBuildPublicationTopic(MQTT_ROUTE_MESHRANK, MQTT_PUBLICATION_NEIGHBORS,
+                                         nullptr, IATA, DEVICE, TOKEN, topic, sizeof(topic)));
+  EXPECT_STREQ("", topic);
 }
 
 TEST(MQTTTopicRouter, MeshCoreRequiresUsableIataAndDevice) {
@@ -178,9 +182,11 @@ TEST(MQTTTopicRouter, PublicationTypeEnumValuesAreFrozen) {
   EXPECT_EQ(0, MQTT_PUBLICATION_STATUS);
   EXPECT_EQ(1, MQTT_PUBLICATION_PACKETS);
   EXPECT_EQ(2, MQTT_PUBLICATION_RAW);
+  EXPECT_EQ(3, MQTT_PUBLICATION_NEIGHBORS);
   EXPECT_STREQ("status", mqttPublicationTypeName(MQTT_PUBLICATION_STATUS));
   EXPECT_STREQ("packets", mqttPublicationTypeName(MQTT_PUBLICATION_PACKETS));
   EXPECT_STREQ("raw", mqttPublicationTypeName(MQTT_PUBLICATION_RAW));
+  EXPECT_STREQ("neighbors", mqttPublicationTypeName(MQTT_PUBLICATION_NEIGHBORS));
 }
 
 }  // namespace

test/test_webconfig_keys/test_webconfig_keys.cpp

@@ -12,6 +12,8 @@ TEST(WebConfigKeys, AllowsKnownScalarKeys) {
   EXPECT_TRUE(wcIsAllowedSetKey("repeat"));
   EXPECT_TRUE(wcIsAllowedSetKey("wifi.ssid"));
   EXPECT_TRUE(wcIsAllowedSetKey("mqtt.iata"));
+  EXPECT_TRUE(wcIsAllowedSetKey("mqtt.neighbors"));
+  EXPECT_TRUE(wcIsAllowedSetKey("mqtt.neighbors.interval"));
   EXPECT_TRUE(wcIsAllowedSetKey("snmp.community"));
   EXPECT_TRUE(wcIsAllowedSetKey("timezone.offset"));
 }

webui/index.html

@@ -313,12 +313,18 @@ canvas{width:100%;height:56px;display:block}
         <span class="tgl"><input type="checkbox" data-k="mqtt.raw"><u></u></span></div>
       <div class="sw"><span><b>Received packets</b><i>Report packets heard over the air</i></span>
         <span class="tgl"><input type="checkbox" data-k="mqtt.rx"><u></u></span></div>
+      <div class="sw"><span><b>Publish neighbors</b><i>Periodic neighbor table + scopes (PSRAM boards only)</i></span>
+        <span class="tgl"><input type="checkbox" data-k="mqtt.neighbors"><u></u></span></div>
       <div class="row" style="margin-top:6px">
         <div class="f"><label>Transmitted packets</label>
           <select data-k="mqtt.tx"><option value="off">Off</option><option value="on">All TX</option><option value="advert">Own adverts only</option></select>
           <div class="hint">Report packets this node sends.</div></div>
         <div class="f"><label>Status interval (min)</label><input type="number" data-k="mqtt.interval" min="1" max="60"></div>
       </div>
+      <div class="row" style="margin-top:6px">
+        <div class="f"><label>Neighbors interval (hours)</label><input type="number" data-k="mqtt.neighbors.interval" min="12" max="336">
+          <div class="hint">How often to publish the neighbor table (12-336, default 24).</div></div>
+      </div>
     </div>
     <div class="card">
       <h2>Servers</h2>
@@ -520,7 +526,9 @@ function cfgVal(k){ // map a `set` key to its current value string, from st.cfg
     case"mqtt.origin":return q.origin;case"mqtt.iata":return q.iata;
     case"mqtt.status":return q.status?"on":"off";case"mqtt.packets":return q.packets?"on":"off";
     case"mqtt.raw":return q.raw?"on":"off";case"mqtt.tx":return q.tx;case"mqtt.rx":return q.rx?"on":"off";
-    case"mqtt.interval":return String(q.interval);case"mqtt.ntp":return q.ntp;
+    case"mqtt.interval":return String(q.interval);
+    case"mqtt.neighbors":return q.neighbors?"on":"off";case"mqtt.neighbors.interval":return String(q.neighbors_interval);
+    case"mqtt.ntp":return q.ntp;
     case"mqtt.owner":return q.owner;case"mqtt.email":return q.email;
     case"timezone":return q.timezone;case"timezone.offset":return String(q.timezone_offset);
     case"snmp":return q.snmp?"on":"off";case"snmp.community":return q.snmp_community;