feat(mqtt): add neighbors publish path to MQTTBridge

Port the periodic-neighbors publication from mqtt-bridge-implementation-flex,
adapted to this branch's structure:

- Add MQTT_PUBLICATION_NEIGHBORS ("neighbors") to the pure MQTTTopicRouter
  instead of flex's messageTypeSuffix() helper (this branch already routes
  every publication type through mqttBuildPublicationTopic()). Neighbors is a
  MeshCore/custom publication type, so it resolves to
  meshcore/{iata}/{device}/neighbors and honors custom templates.
- Deliberately do NOT port flex's "all message types to MeshRank" change:
  this branch documents and host-tests a packets-only MeshRank contract
  (MQTTPresets.h, MQTT_IMPLEMENTATION.md, MeshRankContractIsPacketsOnly). So
  neighbors follows status/raw and is rejected on MeshRank slots.
- WITH_MQTT_NEIGHBORS guard (PSRAM + MAX_NEIGHBOURS) gates all new surface.
- MSG_NEIGHBORS message type + enum-drift static_assert.
- Persistent ~10KB PSRAM neighbors buffer allocated/freed via the existing
  MQTTRuntimeBufferLifecycle path (allocate/release), not the ctor as flex did.
- Core1->Core0 handoff: requestPublishNeighbors() (mesh) fills the buffer with
  a release store; the MQTT task consumes it with an acquire load, publishes
  via publishNeighbors() (QoS1, retain = preset->allow_retain, custom=false),
  and clears the pending flag. A second snapshot is dropped while one is
  in flight.
- setNeighborsSchedule()/NeighborsPhase let the mesh report the timer summary;
  formatMqttStatusReply() gains a "nbr: <when>/<last>" field via formatDuration.

Also fix the on-connect status publish (publishStatusToSlot) to honor
preset->allow_retain instead of hardcoding retain=true, matching the periodic
publishStatus() path. Brokers with allow_retain=false (e.g. the waev MeshCore
preset) reject retained publishes, so the on-connect status was being dropped
there. This is flex followup 028a5dca, reconciled to this branch's custom-slot
default of non-retained.

Extends the host topic-router test to cover the neighbors type across all
routes and freezes the new enum value. Bridge itself is on-target only.

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/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;

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