Keep the console responsive during MQTT shutdown

docs/cli_commands.md

@@ -4794,6 +4794,7 @@ MQTT preferences and retains explicit settings across reboots.
 - `set mqtt.enabled on`
 - `set mqtt.enabled off`
 - `get mqtt.running`
+- `get mqtt.stopping`
 - `get mqtt.status`
 
 These commands are the same on MQTT-capable Companion, Repeater, Room Server,
@@ -4803,6 +4804,14 @@ brokers to reconnect. `get mqtt.running` checks the service's runtime state;
 `get mqtt.status` reports individual broker connections. Enabling MQTT does
 not imply that a broker is connected.
 
+On ESP32 Repeater and Room Server images, `set mqtt.enabled off` acknowledges
+the stop request while the MQTT task disconnects and destroys its clients.
+`get mqtt.stopping` stays `on` until that cleanup completes; `mqtt.running`
+also stays `on` while those resources are still owned. Wait for
+`mqtt.stopping` to become `off` before enabling MQTT again. UART, ESP-NOW,
+and the radio continue running during this cleanup. OTA still waits for a
+completed MQTT shutdown before it can write firmware.
+
 USB logging remains independent. On images with both outputs,
 `set logging.output off|usb|wifi|both` selects USB and MQTT together.
 

examples/simple_repeater/MyMesh.cpp

@@ -12860,8 +12860,12 @@ void __attribute__((noinline)) MyMesh::servicePostMeshLoop() {
   expireRecentRepeatersIfDue();
 #endif
 
+#if defined(WITH_MQTT_BRIDGE)
+  // MQTT owns TLS on Core 0; the radio loop only reaps an acknowledged stop.
+  if (mqtt_bridge && mqtt_bridge->isRunning()) mqtt_bridge->loop();
+#endif
 #if defined(WITH_ESPNOW_BRIDGE)
-  // MQTT runs on Core 0. ESP-NOW remains cooperative, including in the
+  // ESP-NOW remains cooperative, including in the
   // combined Full image where both transports share the WiFi station radio.
   #if defined(WITH_MQTT_BRIDGE) || defined(WITH_RS232_BRIDGE)
   if (espnow_bridge.isRunning()) espnow_bridge.loop();

examples/simple_repeater/MyMesh.h

@@ -1348,6 +1348,7 @@ public:
       _alerter.setBridge(nullptr);
       return !mqtt_bridge || !mqtt_bridge->isRunning();
     }
+    if (mqtt_bridge && mqtt_bridge->isStopping()) return false;
     // Give the WiFi station to MQTT while it associates. ESP-NOW is restarted
     // after the AP channel is known, which avoids a raw ESP-NOW-only session
     // pinning the station to a different channel.
@@ -1430,6 +1431,9 @@ public:
 #endif
 
   bool setBridgeState(bool enable) override {
+#ifdef WITH_MQTT_BRIDGE
+    if (enable && mqtt_bridge && mqtt_bridge->isStopping()) return false;
+#endif
 #if defined(WITH_MQTT_BRIDGE) && defined(WITH_ESPNOW_BRIDGE)
     if (!enable) {
       const bool espnow_stopped = setEspNowBridgeState(false);
@@ -1665,6 +1669,24 @@ public:
 #endif
   }
 
+  bool requestMqttBridgeStop() override {
+#ifdef WITH_MQTT_BRIDGE
+    _alerter.setBridge(nullptr);
+    if (mqtt_bridge) mqtt_bridge->requestStop();
+    return true;
+#else
+    return false;
+#endif
+  }
+
+  bool isMqttBridgeStopping() override {
+#ifdef WITH_MQTT_BRIDGE
+    return mqtt_bridge && mqtt_bridge->isStopping();
+#else
+    return false;
+#endif
+  }
+
   bool syncMqttNtp() override {
 #ifdef WITH_MQTT_BRIDGE
     if (!mqtt_bridge || !mqtt_bridge->isRunning()) return false;

examples/simple_room_server/MyMesh.cpp

@@ -2604,7 +2604,8 @@ void MyMesh::loop() {
   sampleTelemetryHistory();
 #endif
 #ifdef WITH_MQTT_BRIDGE
-  // bridge.loop() is now handled by FreeRTOS task on Core 0 - no need to call it here
+  // TLS runs on Core 0; the radio loop reaps only an acknowledged CLI stop.
+  if (bridge && bridge->isRunning()) bridge->loop();
 #endif
 #ifdef WITH_ESPNOW_BRIDGE
   if (espnow_bridge.isRunning()) espnow_bridge.loop();

examples/simple_room_server/MyMesh.h

@@ -610,6 +610,7 @@ public:
       _alerter.setBridge(nullptr);
       return !bridge || !bridge->isRunning();
     }
+    if (bridge && bridge->isStopping()) return false;
     // Give the WiFi station to MQTT while it associates. ESP-NOW is restarted
     // after the AP channel is known, which avoids a raw ESP-NOW-only session
     // pinning the station to a different channel.
@@ -702,6 +703,9 @@ public:
   }
 
   bool setBridgeState(bool enable) override {
+#ifdef WITH_MQTT_BRIDGE
+    if (enable && bridge && bridge->isStopping()) return false;
+#endif
 #if defined(WITH_MQTT_BRIDGE) && defined(WITH_ESPNOW_BRIDGE)
     if (!enable) return setEspNowBridgeState(false) && setMqttBridgeState(false);
     const bool mqtt_ok = _prefs.bridge_enabled
@@ -846,6 +850,16 @@ public:
     return bridge && bridge->isRunning();
   }
 
+  bool requestMqttBridgeStop() override {
+    _alerter.setBridge(nullptr);
+    if (bridge) bridge->requestStop();
+    return true;
+  }
+
+  bool isMqttBridgeStopping() override {
+    return bridge && bridge->isStopping();
+  }
+
   bool syncMqttNtp() override {
     if (!bridge || !bridge->isRunning()) return false;
     // Queue the sync onto the MQTT task (Core 0) without blocking: this runs on

src/helpers/CommonCLI.cpp

@@ -3786,6 +3786,10 @@ void CommonCLI::handleSetCmd(uint32_t sender_timestamp, char* command, char* rep
       strcpy(reply, "Error: usage set logging.output off|usb|wifi|both");
       return;
     }
+    if (wifi_enabled && _callbacks->isMqttBridgeStopping()) {
+      strcpy(reply, "Error: MQTT is stopping; retry when mqtt.stopping is off");
+      return;
+    }
     const uint8_t previous_usb = _prefs->usb_logging_enabled;
     const uint8_t previous_bridge = _prefs->bridge_enabled;
     _prefs->usb_logging_enabled = usb_enabled ? 1 : 0;
@@ -3797,7 +3801,9 @@ void CommonCLI::handleSetCmd(uint32_t sender_timestamp, char* command, char* rep
       return;
     }
     mesh::setUsbLoggingEnabled(usb_enabled);
-    if (!_callbacks->setMqttBridgeState(wifi_enabled)) {
+    const bool applied = wifi_enabled ? _callbacks->setMqttBridgeState(true)
+                                      : _callbacks->requestMqttBridgeStop();
+    if (!applied) {
       strcpy(reply, "Error: logging output saved, but MQTT runtime change failed");
       return;
     }
@@ -3856,9 +3862,21 @@ void CommonCLI::handleSetCmd(uint32_t sender_timestamp, char* command, char* rep
       strcpy(reply, "Error: use set mqtt.enabled on|off");
     } else {
       const bool enable = strcmp(value, "on") == 0;
+      if (enable && _callbacks->isMqttBridgeStopping()) {
+        strcpy(reply, "Error: MQTT is stopping; retry when mqtt.stopping is off");
+        return;
+      }
+      const uint8_t previous = _prefs->bridge_enabled;
       _prefs->bridge_enabled = enable ? 1 : 0;
-      const bool applied = _callbacks->setMqttBridgeState(enable);
-      savePrefs();
+      // A cooperative stop cannot be cancelled safely once signalled. Save
+      // intent first so a failed write leaves the running service unchanged.
+      if (!trySavePrefs()) {
+        _prefs->bridge_enabled = previous;
+        strcpy(reply, "Error: MQTT setting not saved; unchanged");
+        return;
+      }
+      const bool applied = enable ? _callbacks->setMqttBridgeState(true)
+                                  : _callbacks->requestMqttBridgeStop();
       strcpy(reply, applied ? "OK"
                             : "Error: MQTT runtime change failed; setting saved");
     }

src/helpers/CommonCLI.h

@@ -603,6 +603,10 @@ public:
   // legacy bridge behavior by default. Combined bridge roles override
   // them to let each transport be started, stopped, and queried independently.
   virtual bool setMqttBridgeState(bool enable) { return setBridgeState(enable); }
+  // CLI disable may acknowledge before worker-owned TLS teardown finishes.
+  // Ownership barriers continue to use the synchronous state setter.
+  virtual bool requestMqttBridgeStop() { return setMqttBridgeState(false); }
+  virtual bool isMqttBridgeStopping() { return false; }
   virtual bool setEspNowBridgeState(bool enable) { return setBridgeState(enable); }
   virtual bool setRs232BridgeState(bool enable) { return setBridgeState(enable); }
   virtual bool restartMqttBridge() { return restartBridge(); }

src/helpers/CommonCLI_Observer.cpp

@@ -880,6 +880,10 @@ bool CommonCLI::handleObserverGetCmd(uint32_t sender_timestamp, const char* conf
     snprintf(reply, 160, "> %s", _callbacks->isMqttBridgeRunning() ? "on" : "off");
     return true;
   }
+  if (strcmp(config, "mqtt.stopping") == 0) {
+    snprintf(reply, 160, "> %s", _callbacks->isMqttBridgeStopping() ? "on" : "off");
+    return true;
+  }
   if (memcmp(config, "snmp.community", 14) == 0) {
     sprintf(reply, "> %s", _mqtt_prefs.snmp_community);
   } else if (memcmp(config, "snmp", 4) == 0 && (config[4] == '\0' || config[4] == '\n' || config[4] == '\r')) {

src/helpers/bridges/MQTTBridge.cpp

@@ -683,7 +683,7 @@ void MQTTBridge::formatSlotDiagReply(char* buf, size_t bufsize, int slot_index)
 // end() returns as soon as the task acks (it checks _stop_acked before ticking
 // the timeout), so a larger bound does NOT slow a healthy stop -- it only length-
 // ens the wait before force-killing a genuinely wedged task. The timeout is set
-// per stop in end() via computeStopTimeoutMs() based on the enabled-slot count.
+// per stop in requestStop() based on the enabled-slot count.
 static const uint32_t MQTT_STOP_TIMEOUT_BASE_MS     = 5000;   // fixed teardown overhead
 static const uint32_t MQTT_STOP_TIMEOUT_PER_SLOT_MS = 8000;   // ~5-6 s measured + headroom
 
@@ -899,8 +899,8 @@ void MQTTBridge::begin() {
 
   // Idempotent start: a second begin() would re-run allocation and re-create
   // the task, leaking the previous queue and task.
-  if (_initialized) {
-    MQTT_DEBUG_PRINTLN("MQTT Bridge already running - begin() ignored");
+  if (_initialized || _lifecycle.isStopInProgress()) {
+    MQTT_DEBUG_PRINTLN("MQTT Bridge running or stopping - begin() ignored");
     return;
   }
 
@@ -1078,6 +1078,10 @@ void MQTTBridge::begin() {
     }
   }
 
+  // Reset the handshake before starting either the worker or cooperative loop.
+  _stop_requested.store(false, std::memory_order_release);
+  _stop_acked.store(false, std::memory_order_release);
+
   #ifdef ESP_PLATFORM
   // Create FreeRTOS queue; use PSRAM storage when available
   #ifdef BOARD_HAS_PSRAM
@@ -1121,11 +1125,6 @@ void MQTTBridge::begin() {
   // Task stack: dynamic allocation (internal RAM). A PSRAM-backed stack was tried and
   // reverted -- it resets some boards (e.g. Heltec V4) when the task runs from PSRAM.
   _mqtt_task_handle = nullptr;
-  // Clear the cooperative-stop handshake before the new task starts reading it.
-  // deliverStop() leaves _stop_requested latched true after a stop cycle, so a
-  // restart must reset it or the fresh task would self-terminate immediately.
-  _stop_requested = false;
-  _stop_acked = false;
   BaseType_t create_result = xTaskCreatePinnedToCore(
     mqttTask,
     "MQTTBridge",
@@ -1179,16 +1178,11 @@ void MQTTBridge::begin() {
 }
 
 // ---------------------------------------------------------------------------
-// end()
+// requestStop() - signal only; worker retains ownership until acknowledgment
 // ---------------------------------------------------------------------------
-void MQTTBridge::end() {
-  MQTT_DEBUG_PRINTLN("Stopping MQTT Bridge...");
-
-  // Idempotent stop: nothing to tear down if we never started (or already stopped).
-  if (!_initialized) {
-    MQTT_DEBUG_PRINTLN("MQTT Bridge already stopped - end() ignored");
-    return;
-  }
+void MQTTBridge::requestStop() {
+  // A repeated request must not clear an acknowledgment or restart its budget.
+  if (!_initialized || _lifecycle.isStopInProgress()) return;
 
   // Stop new diagnostic reads through the singleton before teardown begins.
   s_mqtt_bridge_instance = nullptr;
@@ -1216,12 +1210,28 @@ void MQTTBridge::end() {
   // task mid-mbedTLS and then free client buffers on a corrupted heap.
   _lifecycle.requestStop();  // Running -> StopRequested; deliverStop() sets _stop_requested
 
+#ifndef ESP_PLATFORM
+  // The cooperative implementation has no worker to acknowledge separately.
+  _stop_acked.store(true, std::memory_order_release);
+  _lifecycle.onTaskStopped();
+  finishStopped();
+#endif
+}
+
+// ---------------------------------------------------------------------------
+// end() - synchronous barrier for OTA and other ownership transfers
+// ---------------------------------------------------------------------------
+void MQTTBridge::end() {
+  MQTT_DEBUG_PRINTLN("Stopping MQTT Bridge...");
+  if (!_initialized) return;
+  requestStop();  // Joins an existing request without re-delivering its signal.
+
 #ifdef ESP_PLATFORM
   // Wait (bounded) for the task to acknowledge. tick() synthesizes the timeout
   // fallback if the task never acks. Checking the ack first each iteration means
   // a stop that completes right as the timeout expires is still treated as clean.
   while (_lifecycle.isStopInProgress()) {
-    if (_stop_acked) {
+    if (_stop_acked.load(std::memory_order_acquire)) {
       _lifecycle.onTaskStopped();   // StopRequested -> Stopped (clean): releaseResources()
       break;
     }
@@ -1229,14 +1239,12 @@ void MQTTBridge::end() {
     if (!_lifecycle.isStopInProgress()) break;
     vTaskDelay(pdMS_TO_TICKS(20));
   }
-#else
-  // Non-ESP32: the bridge runs cooperatively in loop(); there is no separate
-  // task to signal. Drive straight to a clean Stopped and let releaseResources()
-  // perform the (unchanged) synchronous teardown.
-  _stop_acked = true;
-  _lifecycle.onTaskStopped();
 #endif
+  finishStopped();
+}
 
+void MQTTBridge::finishStopped() {
+  if (!_initialized || _lifecycle.isStopInProgress()) return;
   // Timezone is inline class storage (_timezone_storage) - nothing to delete.
   // The shared JSON document's pools were freed by releaseRuntimeBuffers() above.
   _initialized = false;
@@ -1249,6 +1257,7 @@ void MQTTBridge::end() {
   _ntp_estimate_done = false;
   _ntp_estimate_ok = false;
   _ntp_estimate_epoch = 0;
+  _staged_raw_valid = false;
   MQTT_DEBUG_PRINTLN("MQTT Bridge stopped (%s)",
                      _lifecycle.stopTimedOut() ? "forced/timeout - OTA blocked" : "clean");
 }
@@ -1256,7 +1265,7 @@ void MQTTBridge::end() {
 // ---------------------------------------------------------------------------
 // LifecycleOps - binds MQTTLifecycle::Ops (the pure, host-tested spec) to the
 // FreeRTOS / PsychicMqttClient runtime. Every method runs on the loop task
-// (Core 1): the Coordinator that calls them is driven only from begin()/end().
+// (Core 1): begin(), requestStop(), end(), and loop() own the Coordinator.
 // ---------------------------------------------------------------------------
 uint32_t MQTTBridge::LifecycleOps::nowMs() {
   return (uint32_t)millis();
@@ -1271,8 +1280,8 @@ void MQTTBridge::LifecycleOps::deliverStop() {
   // Clear any stale ack before raising the request (same ordering as the NTP
   // handshake: clear the done-flag, then set the request). The MQTT task polls
   // _stop_requested at the top of mqttTaskLoop().
-  _b->_stop_acked = false;
-  _b->_stop_requested = true;
+  _b->_stop_acked.store(false, std::memory_order_release);
+  _b->_stop_requested.store(true, std::memory_order_release);
 }
 
 void MQTTBridge::LifecycleOps::releaseResources() {
@@ -1426,6 +1435,23 @@ void MQTTBridge::initializeWiFiInTask() {
   MQTT_DEBUG_PRINTLN("WiFi initialization started in task");
 }
 
+// ---------------------------------------------------------------------------
+// Preserve startup/TLS spacing while allowing the owning task to stop promptly.
+// Unsigned tick subtraction remains valid across the FreeRTOS tick wrap.
+bool MQTTBridge::waitUnlessStopping(uint32_t delay_ms) {
+  const TickType_t duration = pdMS_TO_TICKS(delay_ms);
+  const TickType_t started = xTaskGetTickCount();
+  TickType_t quantum = pdMS_TO_TICKS(20);
+  if (quantum == 0) quantum = 1;
+  while (!_stop_requested) {
+    const TickType_t elapsed = xTaskGetTickCount() - started;
+    if (elapsed >= duration) return !_stop_requested;
+    const TickType_t remaining = duration - elapsed;
+    vTaskDelay(remaining < quantum ? remaining : quantum);
+  }
+  return false;
+}
+
 // ---------------------------------------------------------------------------
 // mqttTaskLoop() - main loop running on Core 0
 // ---------------------------------------------------------------------------
@@ -1434,7 +1460,7 @@ void MQTTBridge::mqttTaskLoop() {
   initializeWiFiInTask();
 
   // Wait a bit for WiFi to start connecting
-  vTaskDelay(pdMS_TO_TICKS(1000));
+  waitUnlessStopping(1000);
 
   // Main task loop
   #ifdef MQTT_MEMORY_DEBUG
@@ -1454,7 +1480,7 @@ void MQTTBridge::mqttTaskLoop() {
         teardownSlot(i);
       }
       destroySlotClients();
-      _stop_acked = true;   // release semantics: set only after teardown is done
+      _stop_acked.store(true, std::memory_order_release);  // teardown precedes Core 1 release
       return;
     }
 
@@ -1556,6 +1582,10 @@ void MQTTBridge::mqttTaskLoop() {
     }
 #endif
 
+    // Network/time probes may have yielded while Core 1 requested a stop.
+    // Return to the owner-core teardown before allocating another client.
+    if (_stop_requested) continue;
+
     // Deferred slot setup: wait until NTP is synced so JWT tokens get valid timestamps.
     // This avoids wasted TLS handshakes that get rejected due to bad token times.
     if (_ntp_synced.load(std::memory_order_acquire) && !_slots_setup_done) {
@@ -1570,6 +1600,7 @@ void MQTTBridge::mqttTaskLoop() {
 
       MQTT_DEBUG_PRINTLN("NTP synced, setting up MQTT slots (max %d active)...", _max_active_slots);
       for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) {
+        if (_stop_requested) break;
         if (_slots[i].enabled) {
           if (!canActivateSlot(i)) {
             MQTT_DEBUG_PRINTLN("MQTT%d skipped: max active slots (%d) reached", i + 1, _max_active_slots);
@@ -1588,20 +1619,26 @@ void MQTTBridge::mqttTaskLoop() {
           // Stagger connections: 5s between slots to avoid simultaneous TLS handshakes
           // which compete for ~40KB internal heap each
           if (i < RUNTIME_MQTT_SLOTS - 1) {
-            vTaskDelay(pdMS_TO_TICKS(5000));
+            if (!waitUnlessStopping(5000)) break;
           }
         }
       }
     }
 
+    // A stop during a stagger must not fall through into reconfiguration,
+    // publication, or connection maintenance before the next loop iteration.
+    if (_stop_requested) continue;
+
     // Process pending slot reconfigures (queued from CLI on Core 1)
     for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) {
+      if (_stop_requested) break;
       if (_slot_reconfigure_pending[i]) {
         _slot_reconfigure_pending[i] = false;
         MQTT_DEBUG_PRINTLN("Applying deferred reconfigure for MQTT%d (preset: %s)", i + 1, _obs->mqtt_slot_preset[i]);
         applySlotPreset(i, _obs->mqtt_slot_preset[i]);
       }
     }
+    if (_stop_requested) continue;
 
     // Publish on-connect status for slots whose onConnect callback fired since
     // the last loop. Raised on the esp-mqtt event task, consumed here on the
@@ -1610,17 +1647,21 @@ void MQTTBridge::mqttTaskLoop() {
     // reconnect during the publish re-arms for the next loop rather than being
     // lost; publishStatusToSlot() re-checks slot.connected and no-ops if dropped.
     for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) {
+      if (_stop_requested) break;
       if (_status_publish_pending[i]) {
         _status_publish_pending[i] = false;
         publishStatusToSlot(i);
       }
     }
+    if (_stop_requested) continue;
 
     // Maintain slot connections (token renewal, reconnect with backoff)
     maintainSlotConnections();
+    if (_stop_requested) continue;
 
     // Process packet queue
     processPacketQueue();
+    if (_stop_requested) continue;
 
 #if defined(WITH_MQTT_NEIGHBORS)
     // Consume a pending neighbors snapshot handed over by the mesh (Core 1).
@@ -1939,6 +1980,7 @@ bool MQTTBridge::canActivateSlot(int index) const {
 // maintainSlotConnections() will retry it -- the allocation failures below are transient
 // memory conditions, not permanent misconfiguration.
 bool MQTTBridge::setupSlot(int index) {
+  if (_stop_requested.load(std::memory_order_acquire)) return false;
   if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return false;
   MQTTSlot& slot = _slots[index];
 
@@ -2209,6 +2251,7 @@ void MQTTBridge::teardownSlot(int index, bool force) {
 
 esp_err_t MQTTBridge::reconnectSlotClient(int index) {
   if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return ESP_ERR_INVALID_ARG;
+  if (_stop_requested.load(std::memory_order_acquire)) return ESP_ERR_INVALID_STATE;
   MQTTSlot& slot = _slots[index];
   if (slot.client == nullptr) return ESP_ERR_INVALID_STATE;
 
@@ -2234,6 +2277,7 @@ esp_err_t MQTTBridge::reconnectSlotClient(int index) {
 }
 
 void MQTTBridge::maintainSlotConnections() {
+  if (_stop_requested.load(std::memory_order_acquire)) return;
   if (!_identity) return;
 
   // Check WiFi status first
@@ -2271,6 +2315,7 @@ void MQTTBridge::maintainSlotConnections() {
   bool setup_retry_this_cycle = false;
 
   for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) {
+    if (_stop_requested) return;
     if (!_slots[i].enabled) continue;
 
     // JWT slots need time sync before we can manage tokens
@@ -2617,6 +2662,7 @@ bool MQTTBridge::createSlotAuthToken(int index) {
 }
 
 bool MQTTBridge::publishToSlot(int index, const char* topic, const char* payload, size_t payload_len, bool retained, uint8_t qos) {
+  if (_stop_requested.load(std::memory_order_acquire)) return false;
   if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return false;
   MQTTSlot& slot = _slots[index];
   if (!slot.client || !slot.connected) {
@@ -2854,6 +2900,7 @@ bool MQTTBridge::isAnySlotConnected() {
 }
 
 void MQTTBridge::setSlotPreset(int slot_index, const char* preset_name) {
+  if (_stop_requested.load(std::memory_order_acquire)) return;
   if (slot_index < 0 || slot_index >= RUNTIME_MQTT_SLOTS) return;
 
   // On ESP32, teardown/setup involves TLS and must run on the MQTT task (Core 0).
@@ -2871,6 +2918,7 @@ void MQTTBridge::setSlotPreset(int slot_index, const char* preset_name) {
 }
 
 void MQTTBridge::applySlotPreset(int slot_index, const char* preset_name) {
+  if (_stop_requested.load(std::memory_order_acquire)) return;
   if (slot_index < 0 || slot_index >= RUNTIME_MQTT_SLOTS) return;
   MQTTSlot& slot = _slots[slot_index];
 
@@ -2943,6 +2991,7 @@ void MQTTBridge::applySlotPreset(int slot_index, const char* preset_name) {
 
 void MQTTBridge::setSlotCustomBroker(int slot_index, const char* host, uint16_t port,
                                       const char* username, const char* password) {
+  if (_stop_requested.load(std::memory_order_acquire)) return;
   if (slot_index < 0 || slot_index >= RUNTIME_MQTT_SLOTS) return;
   MQTTSlot& slot = _slots[slot_index];
 
@@ -3085,7 +3134,8 @@ bool MQTTBridge::handleWiFiConnection(unsigned long now) {
 }
 
 bool MQTTBridge::isReady() const {
-  return _initialized && isWiFiConfigValid(_obs);
+  return _initialized && !_stop_requested.load(std::memory_order_acquire)
+      && isWiFiConfigValid(_obs);
 }
 
 bool MQTTBridge::isIATAValid() const {
@@ -3135,13 +3185,20 @@ bool MQTTBridge::isSlotReady(int index, char* reason_buf, size_t reason_size) co
 }
 
 // ---------------------------------------------------------------------------
-// loop() - non-ESP32 main loop (ESP32 uses mqttTaskLoop via FreeRTOS task)
+// loop() - Core 1 stop reaping on ESP32; cooperative processing elsewhere
 // ---------------------------------------------------------------------------
 void MQTTBridge::loop() {
   if (!_initialized) return;
 
   #ifdef ESP_PLATFORM
-  // On ESP32, loop() is a no-op - all processing happens in the FreeRTOS task
+  // Poll only: a slow TLS teardown must not stall radio/UART forwarding or
+  // force-delete the worker. Only end(), the synchronous OTA barrier, owns
+  // the existing bounded-timeout fallback.
+  if (_lifecycle.isStopInProgress()
+      && _stop_acked.load(std::memory_order_acquire)) {
+    _lifecycle.onTaskStopped();
+    finishStopped();
+  }
   return;
   #else
   unsigned long now = millis();
@@ -3271,7 +3328,8 @@ void MQTTBridge::loop() {
 // ---------------------------------------------------------------------------
 
 void MQTTBridge::onPacketReceived(mesh::Packet *packet) {
-  if (!_initialized || !_obs->mqtt_packets_enabled || !_obs->mqtt_rx_enabled) return;
+  if (!_initialized || _stop_requested.load(std::memory_order_acquire)
+      || !_obs->mqtt_packets_enabled || !_obs->mqtt_rx_enabled) return;
 
   // Drop before the queue copy when no configured slot allows this payload
   // type. A QueuedPacket carries the packet plus up to 256 bytes of raw radio
@@ -3295,7 +3353,8 @@ void MQTTBridge::onPacketReceived(mesh::Packet *packet) {
 
 void MQTTBridge::sendPacket(mesh::Packet *packet) {
   uint8_t tx_mode = _obs->mqtt_tx_enabled;  // Read live from prefs (no restart needed)
-  if (!_initialized || !_obs->mqtt_packets_enabled || tx_mode == 0) return;
+  if (!_initialized || _stop_requested.load(std::memory_order_acquire)
+      || !_obs->mqtt_packets_enabled || tx_mode == 0) return;
 
   // Advert mode: only queue self-originated advert packets
   if (tx_mode == 2) {
@@ -4000,7 +4059,7 @@ void MQTTBridge::requestPublishNeighbors(const char* json, size_t len) {
   if (_neighbors_publish_pending.load(std::memory_order_acquire)) return;
   // A discovery can finish after bridge shutdown. Do not allocate a buffer
   // which has no task left to consume or release it.
-  if (!isRunning()) return;
+  if (!isRunning() || _stop_requested.load(std::memory_order_acquire)) return;
   _neighbors_json_buffer = static_cast<char*>(
       MQTTRuntimeBufferLifecycle::allocateIfMissing(
           _neighbors_json_buffer, NEIGHBORS_JSON_BUFFER_SIZE, psram_malloc));
@@ -4047,6 +4106,7 @@ bool MQTTBridge::publishNeighbors() {
 // ---------------------------------------------------------------------------
 
 void MQTTBridge::queuePacket(mesh::Packet* packet, bool is_tx) {
+  if (!_initialized || _stop_requested.load(std::memory_order_acquire)) return;
   #ifdef ESP_PLATFORM
   // Use FreeRTOS queue for thread-safe operation
   if (_packet_queue_handle == nullptr) {
@@ -4174,6 +4234,7 @@ void MQTTBridge::dequeuePacket() {
 // ---------------------------------------------------------------------------
 
 void MQTTBridge::storeRawRadioData(const uint8_t* raw_data, int len, float snr, float rssi) {
+  if (!_initialized || _stop_requested.load(std::memory_order_acquire)) return;
   // Writes into the Core 1-only staging area. No mutex needed: this function and
   // queuePacket() are both called from Core 1 in guaranteed sequence for each packet.
   if (len > 0 && len <= (int)LAST_RAW_DATA_SIZE) {
@@ -4549,7 +4610,7 @@ bool MQTTBridge::syncTimeWithNTP(bool force, bool primary_only) {
 }
 
 bool MQTTBridge::requestForcedNtpSync(uint32_t timeout_ms) {
-  if (!isRunning()) return false;
+  if (!isRunning() || _stop_requested.load(std::memory_order_acquire)) return false;
 
   // Publish the request to the MQTT task. Clear the completion flags before
   // raising _ntp_force_requested so the task can't observe a stale result.
@@ -4618,7 +4679,8 @@ void MQTTBridge::runNtpEstimateProbe() {
 }
 
 bool MQTTBridge::requestNtpTimeEstimate() {
-  if (!isRunning() || WiFi.status() != WL_CONNECTED || _ntp_estimate_requested) return false;
+  if (!isRunning() || _stop_requested.load(std::memory_order_acquire)
+      || WiFi.status() != WL_CONNECTED || _ntp_estimate_requested) return false;
   _ntp_estimate_done = false;
   _ntp_estimate_ok = false;
   _ntp_estimate_epoch = 0;
@@ -4636,7 +4698,8 @@ bool MQTTBridge::takeNtpTimeEstimate(uint32_t& epoch, bool& finished) {
 }
 
 bool MQTTBridge::ntpDiag(char* reply, size_t reply_size, bool verbose) {
-  if (!isRunning() || reply == nullptr || reply_size == 0) return false;
+  if (!isRunning() || _stop_requested.load(std::memory_order_acquire)
+      || reply == nullptr || reply_size == 0) return false;
 
   // Marshal the probe onto the MQTT task (Core 0); clear the completion flag first.
   _ntp_diag_done = false;

src/helpers/bridges/MQTTBridge.h

@@ -300,12 +300,11 @@ private:
   // Cooperative-shutdown handshake (Phase 5). The loop task (Core 1) raises
   // _stop_requested through the lifecycle Coordinator; the MQTT task (Core 0)
   // sees it, tears down its own clients on Core 0 (where the mbedTLS contexts
-  // live), sets _stop_acked LAST, and self-terminates. end() waits for the ack
-  // before freeing the queue/buffers. Plain volatile matches the existing
-  // NTP/reconfigure handshake idiom above; replacing all of these with a command
-  // channel / task notifications is explicitly deferred (see MQTT_OWNERSHIP.md).
-  volatile bool _stop_requested = false;
-  volatile bool _stop_acked = false;
+  // live), sets _stop_acked LAST, and self-terminates. end() waits for the ack;
+  // loop() can instead reap it without blocking. Atomic release/acquire ordering
+  // makes client teardown visible before Core 1 frees the queue/buffers.
+  std::atomic<bool> _stop_requested{false};
+  std::atomic<bool> _stop_acked{false};
 
   // Timezone handling.
   // _timezone_storage is inline class storage (zero heap) that is reconfigured
@@ -515,6 +514,7 @@ private:
   #ifdef ESP_PLATFORM
   static void mqttTask(void* parameter);
   void mqttTaskLoop();  // Main loop for MQTT task
+  bool waitUnlessStopping(uint32_t delay_ms);
   void initializeWiFiInTask();  // WiFi initialization moved to task
   #endif
   bool publishPacket(mesh::Packet* packet, bool is_tx, bool& has_eligible_target,
@@ -551,13 +551,14 @@ private:
   // of making the bridge unusable.
   void allocateRuntimeBuffers();
   void releaseRuntimeBuffers();
+  void finishStopped();  // Core 1 only, after lifecycle resource release
 
   // --- Cooperative lifecycle (Phase 5) ---------------------------------------
   // The pure state machine, bounded stop timeout, and OTA barrier live in
   // src/helpers/MQTTLifecycle.h and are host-tested by test/test_mqtt_lifecycle/.
   // This nested Ops binds that spec to FreeRTOS/PsychicMqttClient. The
   // Coordinator is owned and driven ONLY by the loop task (Core 1) from
-  // begin()/end(); the MQTT task (Core 0) communicates solely through the
+  // begin()/requestStop()/end()/loop(); the MQTT task (Core 0) communicates solely through the
   // _stop_requested/_stop_acked flags above. Methods are defined in the .cpp.
   class LifecycleOps : public MQTTLifecycle::Ops {
    public:
@@ -587,6 +588,10 @@ public:
 
   void begin() override;
   void end() override;
+  // Core 1: request cooperative teardown without waiting or freeing resources.
+  // Call loop() to reap the acknowledgment; end() remains the OTA stop barrier.
+  void requestStop();
+  bool isStopping() const { return _lifecycle.isStopInProgress(); }
   bool isRunning() const override { return _initialized; }
   void loop() override;
   void onPacketReceived(mesh::Packet *packet) override;
@@ -681,8 +686,8 @@ public:
   int getConnectedBrokers() const;
   int getQueueSize() const;
   bool isReady() const;
-  /** True only after a CLEAN cooperative stop -- end() received the MQTT task's
-   *  acknowledgment within the timeout. A timed-out/forced stop returns false so
+  /** True only after a CLEAN cooperative stop -- the MQTT task acknowledged
+   *  its teardown before resources were released. A timed-out/forced stop returns false so
    *  OTA flashing is withheld until a clean start/stop cycle. Mirrors
    *  MQTTLifecycle::mayBeginFlash(); read on the loop task (Core 1). */
   bool canFlashAfterStop() const { return _lifecycle.mayBeginFlash(); }

test/fixtures/rs232_mqtt_runtime.cpp

@@ -112,6 +112,7 @@ struct MQTTNodeInfo {
 };
 class MQTTBridge : public AbstractBridge {
   bool running = false;
+  bool stopping = false;
 public:
   inline static unsigned live = 0, fail_starts = 0;
   unsigned begins = 0, ends = 0, loops = 0, sent = 0, received = 0;
@@ -127,9 +128,11 @@ public:
     if (fail_starts) { --fail_starts; return; }
     running = true;
   }
-  void end() override { ++ends; running = false; }
+  void end() override { ++ends; running = stopping = false; }
+  void requestStop() { stopping = running; }
+  bool isStopping() const { return stopping; }
   bool isRunning() const override { return running; }
-  void loop() override { ++loops; }
+  void loop() override { ++loops; if (stopping) { ++ends; running = stopping = false; } }
   void sendPacket(mesh::Packet* p) override { if (allow(p)) ++sent; }
   void onPacketReceived(mesh::Packet* p) { if (allow(p)) ++received; }
 };
@@ -171,6 +174,8 @@ struct Callbacks {
   virtual bool restartEspNowBridge() = 0;
   virtual bool isMqttBridgeRunning() = 0;
   virtual bool setMqttBridgeState(bool) = 0;
+  virtual bool requestMqttBridgeStop() = 0;
+  virtual bool isMqttBridgeStopping() = 0;
   virtual bool restartMqttBridge() = 0;
 };
 class MyMesh : public Callbacks {
@@ -277,7 +282,7 @@ int main() {
   assert(mesh.bridge->sent == 1 && mesh.bridge->loops == 1);
   assert(mesh.espnow_bridge.sent == 1 && mesh.espnow_bridge.loops == 1);
   assert(mesh.mqtt_bridge->received == 1 && mesh.mqtt_bridge->sent == 1);
-  assert(mesh.mqtt_bridge->loops == 0); // Its actual task services MQTT independently.
+  assert(mesh.mqtt_bridge->loops == 1); // Core 1 services stop acknowledgement only.
   assert(mesh.filter_checks == 4);
   packet.allowed = false;
   mesh.routeRx(&packet); mesh.routeTx(&packet);
@@ -342,12 +347,34 @@ int main() {
   cli.set("bridge.enabled off", reply);
   assert(strcmp(reply, "OK") == 0 && !mesh.isEspNowBridgeRunning());
   assert(mesh.isMqttBridgeRunning() && mesh.isRs232BridgeRunning());
+  // A failed intent save must not signal an uncancellable worker stop.
+  cli.fail_save = true;
   cli.set("mqtt.enabled off", reply);
-  assert(strcmp(reply, "OK") == 0 && !mesh.isMqttBridgeRunning());
+  assert(strstr(reply, "not saved; unchanged") && mesh._prefs.bridge_enabled);
+  assert(mesh.isMqttBridgeRunning() && !mesh.isMqttBridgeStopping());
+  cli.fail_save = false;
+  cli.set("mqtt.enabled off", reply);
+  assert(strcmp(reply, "OK") == 0 && mesh.isMqttBridgeRunning());
+  assert(mesh.isMqttBridgeStopping() && !mesh._prefs.bridge_enabled);
+  expect_get(cli, "mqtt.running", "> on");
+  expect_get(cli, "mqtt.stopping", "> on");
+  const unsigned stopping_saves = cli.saves;
+  cli.set("mqtt.enabled on", reply);
+  assert(strstr(reply, "MQTT is stopping") && !mesh._prefs.bridge_enabled);
+  assert(cli.saves == stopping_saves && !mesh.setMqttBridgeState(true));
+  cli.set("mqtt.enabled off", reply); // a repeated request is accepted
+  assert(strcmp(reply, "OK") == 0 && mesh.isMqttBridgeStopping());
   assert(mesh.isRs232BridgeRunning() && mesh.bridgesPreventSleep());
   const unsigned uart_tx_before = mesh.bridge->sent;
   mesh.routeRx(&packet); mesh.serviceBridges();
   assert(mesh.bridge->sent == uart_tx_before + 1);
+  assert(!mesh.isMqttBridgeRunning() && !mesh.isMqttBridgeStopping());
+  expect_get(cli, "mqtt.stopping", "> off");
+  cli.fail_save = true;
+  cli.set("mqtt.enabled on", reply);
+  assert(strstr(reply, "not saved; unchanged") && !mesh._prefs.bridge_enabled);
+  assert(!mesh.isMqttBridgeRunning());
+  cli.fail_save = false;
   cli.set("espnow.enabled on", reply); assert(strcmp(reply, "OK") == 0);
   cli.set("mqtt.enabled on", reply); assert(strcmp(reply, "OK") == 0);
   mesh.serviceBridges(); assert(mesh.isEspNowBridgeRunning());
@@ -379,6 +406,8 @@ int main() {
     assert(mesh.isRs232BridgeRunning() && RS232Bridge::live == 1);
   }
   cli.set("rs232.enabled off", reply); cli.set("espnow.enabled off", reply); cli.set("mqtt.enabled off", reply);
+  assert(mesh.isMqttBridgeStopping() && mesh.bridgesPreventSleep());
+  mesh.serviceBridges();
   assert(!mesh.bridgesPreventSleep() && RS232Bridge::live == 0);
   cli.run("start ota", reply); cli.run("stop ota", reply); mesh.serviceBridges();
   assert(!mesh.isRs232BridgeRunning() && !mesh.isMqttBridgeRunning() && !mesh.isEspNowBridgeRunning());

test/test_cli_settings_contract.py

@@ -68,6 +68,7 @@ OBSERVER_QUERY_EXCEPTIONS = {
     'mqtt.ntp.diag': 'NTP diagnostics',
     'mqtt.presets': 'List of available presets',
     'mqtt.running': 'Live state; set mqtt.enabled controls intent',
+    'mqtt.stopping': 'Pending worker-owned shutdown; set mqtt.enabled off requests it',
     'mqtt.stats': 'Runtime MQTT statistics',
     'wifi.status': 'Live connection status',
 }

test/test_common_cli_logging_transaction.py

@@ -22,13 +22,15 @@ namespace mesh {
 void setUsbLoggingEnabled(bool value) { live_usb = value; ++usb_calls; }
 }
 struct Callbacks {
-  bool live_wifi = false, apply_ok = true;
+  bool live_wifi = false, apply_ok = true, stopping = false;
   unsigned calls = 0;
   bool setMqttBridgeState(bool value) {
     ++calls;
     if (apply_ok) live_wifi = value;
     return apply_ok;
   }
+  bool requestMqttBridgeStop() { return setMqttBridgeState(false); }
+  bool isMqttBridgeStopping() { return stopping; }
 };
 struct CLI {
   Prefs prefs, persisted;
@@ -88,6 +90,15 @@ int main() {
   assert(cli.persisted.usb_logging_enabled && cli.persisted.bridge_enabled);
   assert(live_usb && !cli.callbacks.live_wifi);
   assert(strstr(reply, "saved, but MQTT runtime change failed"));
+  // A restart during cleanup must not save ON or change either live output.
+  cli.callbacks.stopping = true;
+  cli.prefs = cli.persisted = {0, 0};
+  live_usb = false;
+  const unsigned before_saves = cli.saves, before_calls = cli.callbacks.calls;
+  cli.set("logging.output both", reply);
+  assert(strstr(reply, "MQTT is stopping"));
+  assert(cli.saves == before_saves && cli.callbacks.calls == before_calls);
+  assert(!cli.prefs.bridge_enabled && !cli.persisted.bridge_enabled && !live_usb);
 #endif
 }
 '''

test/test_mqtt_transport_results.py

@@ -26,6 +26,7 @@ def method(path, signature):
 
 
 PREAMBLE = r'''
+#include <atomic>
 #include <cstdlib>
 #include <cstdint>
 #include <cassert>
@@ -90,6 +91,7 @@ struct MQTTBridge {
   };
   MQTTSlot _slots[RUNTIME_MQTT_SLOTS];
   volatile bool _slot_attempt_pending[RUNTIME_MQTT_SLOTS] = {};
+  std::atomic<bool> _stop_requested{false};
   esp_err_t reconnectSlotClient(int index);
 };
 '''
@@ -152,12 +154,391 @@ int main() {
   sdk_callback=nullptr;
   reconnect_result=ESP_OK;
   assert(b.reconnectSlotClient(0)==ESP_OK && b._slot_attempt_pending[0]);
+  const int stopped_starts = start_calls, stopped_reconnects = reconnect_calls;
+  b._stop_requested = true;
+  assert(b.reconnectSlotClient(0)==ESP_ERR_INVALID_STATE);
+  assert(start_calls==stopped_starts && reconnect_calls==stopped_reconnects);
   return 0;
 }
 '''
 
 
+TASK_PREAMBLE = r'''
+#include <algorithm>
+#include <atomic>
+#include <cassert>
+#include <climits>
+#include <cstdint>
+#include <cstdlib>
+#include <string>
+#include <vector>
+#define PORTABLE_MQTT_OBSERVER 1
+#define MQTT_DEBUG_PRINTLN(...) ((void)0)
+using TickType_t = uint32_t;
+constexpr int RUNTIME_MQTT_SLOTS = 3, WL_CONNECTED = 1;
+constexpr uint32_t STATUS_RETRY_INTERVAL = 30000;
+static uint32_t tick_ms = 1, first_tick = 0, first_millis = 0;
+static uint64_t elapsed_ms = 0, stop_at_ms = UINT64_MAX;
+static int scenario = 0;
+static std::vector<uint32_t> task_sleeps;
+struct MQTTBridge;
+static MQTTBridge* active_bridge = nullptr;
+#define pdMS_TO_TICKS(ms) static_cast<TickType_t>((ms) / tick_ms)
+TickType_t xTaskGetTickCount() {
+  return first_tick + static_cast<TickType_t>(elapsed_ms / tick_ms);
+}
+uint32_t millis() { return first_millis + static_cast<uint32_t>(elapsed_ms); }
+void vTaskDelay(TickType_t ticks);
+struct { int status() { return WL_CONNECTED; } } WiFi;
+bool mqttNtpRefreshDue(unsigned long, unsigned long, unsigned long) { return false; }
+struct MQTTPrefs {
+  char mqtt_slot_preset[RUNTIME_MQTT_SLOTS][16] = {};
+  bool mqtt_status_enabled = true;
+};
+struct MQTTBridge {
+  struct Slot {
+    bool enabled = true, initial_connect_done = false, connected = false;
+    unsigned long last_reconnect_attempt = 0;
+  } _slots[RUNTIME_MQTT_SLOTS];
+  struct Latch {
+    void noteGotIp() {}
+    bool consumeIfConnected(bool) { return false; }
+  } _ntp_reconnect_latch;
+  MQTTPrefs prefs;
+  MQTTPrefs* _obs = &prefs;
+  std::atomic<bool> _ntp_synced{true};
+  bool _ntp_sync_pending = false, _slots_setup_done = false;
+  bool _ntp_force_requested = false, _ntp_force_result = false, _ntp_force_done = false;
+  bool _ntp_refresh_pending = false, _cached_has_connected_slots = true;
+  std::atomic<bool> _stop_requested{false}, _stop_acked{false};
+  bool _slot_reconfigure_pending[RUNTIME_MQTT_SLOTS] = {true, true, true};
+  bool _status_publish_pending[RUNTIME_MQTT_SLOTS] = {true, true, true};
+  unsigned long _last_ntp_sync = 0, _ntp_refresh_retry_at = 0;
+  unsigned long _last_status_retry = 0, _last_status_publish = 0, _status_interval = 300000;
+  int _max_active_slots = RUNTIME_MQTT_SLOTS, _queue_count = 0;
+  int late_work = 0, reconfigures = 0, publishes = 0, queue_calls = 0;
+  std::vector<int> setups;
+  std::vector<uint64_t> setup_times;
+  std::vector<std::string> teardown;
+  void work() {
+    if (_stop_requested) ++late_work;
+    assert(!_stop_acked);
+  }
+  void initializeWiFiInTask() {}
+  bool handleWiFiConnection(unsigned long) { work(); return false; }
+  bool syncTimeWithNTP(bool = false, bool = false) {
+    work();
+    if (scenario == 6 || scenario == 7) {
+      _ntp_synced = true;
+      _stop_requested = true;
+    }
+    return true;
+  }
+  bool canActivateSlot(int) { work(); return true; }
+  bool isSlotReady(int, char*, size_t) { work(); return true; }
+  bool setupSlot(int index) {
+    work(); setups.push_back(index); setup_times.push_back(elapsed_ms);
+    if (scenario == 5) _stop_requested = true;
+    return true;
+  }
+  void applySlotPreset(int, const char*) {
+    work(); ++reconfigures;
+    if (scenario == 8) _stop_requested = true;
+  }
+  void publishStatusToSlot(int) {
+    work(); ++publishes;
+    if (scenario == 9) _stop_requested = true;
+  }
+  void maintainSlotConnections() {
+    work();
+    if (scenario == 10) _stop_requested = true;
+  }
+  void processPacketQueue() {
+    work(); ++queue_calls;
+    if (scenario == 11) _stop_requested = true;
+  }
+  void checkConfigurationMismatch() { work(); }
+  void pollNtpRefresh(unsigned long) { work(); }
+  void refreshNTP() { work(); }
+  bool publishStatus() { work(); ++publishes; return true; }
+  void updateCachedConnectionStatus() { work(); }
+  void cancelNtpRefresh() {
+    assert(_stop_requested && !_stop_acked);
+    teardown.push_back("ntp");
+  }
+  void teardownSlot(int index) {
+    assert(_stop_requested && !_stop_acked);
+    teardown.push_back("slot" + std::to_string(index));
+  }
+  void destroySlotClients() {
+    assert(!_stop_acked);
+    assert(teardown == std::vector<std::string>({"ntp", "slot0", "slot1", "slot2"}));
+    teardown.push_back("destroy");
+  }
+  bool waitUnlessStopping(uint32_t delay_ms);
+  void mqttTaskLoop();
+};
+void vTaskDelay(TickType_t ticks) {
+  assert(ticks > 0);
+  const uint32_t duration = ticks * tick_ms;
+  task_sleeps.push_back(duration);
+  elapsed_ms += duration;
+  if (elapsed_ms >= stop_at_ms) active_bridge->_stop_requested = true;
+  // The no-stop scenario exits after one complete worker iteration. Stop is
+  // raised during its ordinary idle delay, after all setup and publications.
+  if (scenario == 3 && active_bridge->setups.size() == RUNTIME_MQTT_SLOTS)
+    active_bridge->_stop_requested = true;
+  assert(elapsed_ms < 30000); // bound a missing cooperative exit deterministically
+}
+'''
+
+TASK_MAIN = r'''
+int main(int argc, char** argv) {
+  assert(argc == 4);
+  scenario = std::atoi(argv[1]);
+  tick_ms = static_cast<uint32_t>(std::atoi(argv[2]));
+  if (std::atoi(argv[3])) {
+    first_tick = UINT32_MAX - 511;
+    first_millis = UINT32_MAX - 511;
+  }
+  MQTTBridge bridge;
+  active_bridge = &bridge;
+  switch (scenario) {
+    case 0: stop_at_ms = 187; break;          // stop while WiFi settles
+    case 1: stop_at_ms = 1173; break;         // stop during first TLS stagger
+    case 2: stop_at_ms = 6191; break;         // stop during second TLS stagger
+    case 3: break;                           // preserve every normal delay
+    case 4: bridge._stop_requested = true; break;
+    case 5: break;                           // setup callback requests stop
+    case 6: bridge._ntp_force_requested = true; break;
+    case 7: bridge._ntp_sync_pending = true; bridge._ntp_synced = false; break;
+    case 8: case 9: case 10: case 11: break;  // stop raised during worker callbacks
+    default: assert(false);
+  }
+  bridge.mqttTaskLoop();
+  assert(bridge._stop_acked && bridge.late_work == 0);
+  assert(bridge.teardown == std::vector<std::string>({
+    "ntp", "slot0", "slot1", "slot2", "destroy"}));
+  if (scenario == 3) {
+    assert(bridge.setups == std::vector<int>({0, 1, 2}));
+    assert(bridge.setup_times == std::vector<uint64_t>({1000, 6000, 11000}));
+    assert(bridge.reconfigures == RUNTIME_MQTT_SLOTS);
+    assert(bridge.publishes >= RUNTIME_MQTT_SLOTS && bridge.queue_calls == 1);
+  } else {
+    const size_t wanted = scenario >= 8 ? RUNTIME_MQTT_SLOTS
+                          : scenario == 0 || scenario == 4 || scenario >= 6 ? 0
+                          : scenario == 2 ? 2 : 1;
+    assert(bridge.setups.size() == wanted);
+    if (scenario >= 8) {
+      assert(bridge.reconfigures == (scenario == 8 ? 1 : RUNTIME_MQTT_SLOTS));
+      assert(bridge.publishes == (scenario == 8 ? 0 : scenario == 9 ? 1 : RUNTIME_MQTT_SLOTS));
+      assert(bridge.queue_calls == (scenario == 11 ? 1 : 0));
+    } else {
+      assert(bridge.reconfigures == 0 && bridge.publishes == 0 && bridge.queue_calls == 0);
+    }
+    const uint32_t poll_bound = std::max(uint32_t{20}, tick_ms);
+    assert(std::all_of(task_sleeps.begin(), task_sleeps.end(),
+                       [poll_bound](uint32_t duration) { return duration <= poll_bound; }));
+    if (scenario == 4) assert(elapsed_ms == 0 && task_sleeps.empty());
+    else if (scenario >= 8) assert(elapsed_ms == 11000);
+    else if (scenario >= 5) assert(elapsed_ms == 1000);
+    else assert(elapsed_ms >= stop_at_ms && elapsed_ms - stop_at_ms < poll_bound);
+  }
+  // A completed zero-duration wait preserves its caller's stop decision.
+  const auto sleeps_before = task_sleeps.size();
+  bridge._stop_requested = false;
+  assert(bridge.waitUnlessStopping(0));
+  bridge._stop_requested = true;
+  assert(!bridge.waitUnlessStopping(0));
+  assert(task_sleeps.size() == sleeps_before);
+}
+'''
+
+ASYNC_STOP_PREAMBLE = r'''
+#include <atomic>
+#include <cassert>
+#include <cstdint>
+#include <limits>
+#include "helpers/MQTTLifecycle.h"
+#define ESP_PLATFORM 1
+#define MQTT_DEBUG_PRINTLN(...) ((void)0)
+#define pdMS_TO_TICKS(ms) (ms)
+constexpr int RUNTIME_MQTT_SLOTS = 3;
+static constexpr uint32_t MQTT_STOP_TIMEOUT_BASE_MS = 5000;
+static constexpr uint32_t MQTT_STOP_TIMEOUT_PER_SLOT_MS = 8000;
+static uint32_t clock_ms = 0, sleeps = 0;
+static uint64_t elapsed_ms = 0, ack_after_ms = UINT64_MAX;
+uint32_t millis() { return clock_ms; }
+struct MQTTBridge;
+static MQTTBridge* s_mqtt_bridge_instance = nullptr;
+static MQTTBridge* bridge_waiting = nullptr;
+void vTaskDelay(uint32_t duration);
+struct MQTTBridge {
+  struct Slot { bool enabled = true; } _slots[RUNTIME_MQTT_SLOTS];
+  struct LifecycleOps : MQTTLifecycle::Ops {
+    MQTTBridge* _b;
+    explicit LifecycleOps(MQTTBridge* bridge) : _b(bridge) {}
+    uint32_t nowMs() override;
+    void startTask() override;
+    void deliverStop() override;
+    void releaseResources() override;
+    void onStopComplete(bool clean) override;
+  };
+  struct RecordingOps : LifecycleOps {
+    int signals = 0;
+    explicit RecordingOps(MQTTBridge* bridge) : LifecycleOps(bridge) {}
+    void deliverStop() override { ++signals; LifecycleOps::deliverStop(); }
+  } ops{this};
+  MQTTLifecycle::Coordinator _lifecycle{ops, 1};
+  std::atomic<bool> _stop_requested{false}, _stop_acked{false};
+  bool _initialized = true, resources_owned = true;
+  bool _slots_setup_done = true, _staged_raw_valid = true;
+  bool _ntp_estimate_requested = true, _ntp_estimate_done = true, _ntp_estimate_ok = true;
+  uint32_t _ntp_estimate_epoch = 123;
+  int releases = 0, cancels = 0, completions = 0;
+  bool clean_completion = false;
+  MQTTBridge() {
+    assert(_lifecycle.requestStart());
+    assert(_lifecycle.onTaskStarted());
+    s_mqtt_bridge_instance = this;
+  }
+  void cancelNtpRefresh() { ++cancels; }
+  void requestStop();
+  void end();
+  void finishStopped();
+  void loop();
+};
+void MQTTBridge::LifecycleOps::releaseResources() {
+  assert(_b->_stop_acked.load() || _b->_lifecycle.stopTimedOut());
+  assert(_b->resources_owned && _b->_initialized);
+  _b->resources_owned = false;
+  ++_b->releases;
+}
+void MQTTBridge::LifecycleOps::onStopComplete(bool clean) {
+  ++_b->completions;
+  _b->clean_completion = clean;
+}
+void vTaskDelay(uint32_t duration) {
+  assert(duration > 0);
+  ++sleeps;
+  clock_ms += duration;
+  elapsed_ms += duration;
+  if (elapsed_ms >= ack_after_ms)
+    bridge_waiting->_stop_acked.store(true, std::memory_order_release);
+  assert(elapsed_ms < 60000);
+}
+static void assert_finished(MQTTBridge& bridge, bool clean) {
+  assert(!bridge._initialized && !bridge.resources_owned);
+  assert(bridge.releases == 1 && bridge.completions == 1);
+  assert(bridge.clean_completion == clean);
+  assert(!bridge._lifecycle.isStopInProgress() && bridge._lifecycle.mayRestart());
+  assert(bridge._lifecycle.mayBeginFlash() == clean);
+  assert(!bridge._slots_setup_done && !bridge._staged_raw_valid);
+  assert(!bridge._ntp_estimate_requested && !bridge._ntp_estimate_done);
+  assert(!bridge._ntp_estimate_ok && bridge._ntp_estimate_epoch == 0);
+  assert(bridge.cancels == 1);
+}
+'''
+
+ASYNC_STOP_MAIN = r'''
+int main() {
+  {
+    MQTTBridge bridge;
+    bridge._stop_acked = true; // stale prior-epoch ACK must be cleared exactly once
+    bridge.requestStop();
+    assert(sleeps == 0 && elapsed_ms == 0 && bridge.ops.signals == 1);
+    assert(bridge._stop_requested && !bridge._stop_acked);
+    assert(bridge._initialized && bridge.resources_owned && bridge.releases == 0);
+    assert(s_mqtt_bridge_instance == nullptr);
+    const auto timeout = bridge._lifecycle.stopTimeoutMs();
+    assert(timeout > 5000);
+    for (int pass = 0; pass < 10; ++pass) {
+      clock_ms += timeout; // ordinary loop must never force-free a slow worker
+      bridge.requestStop();
+      bridge.loop();
+      assert(bridge.ops.signals == 1 && sleeps == 0);
+      assert(bridge._initialized && bridge.resources_owned && bridge.releases == 0);
+      assert(bridge._lifecycle.isStopInProgress() && !bridge._lifecycle.stopTimedOut());
+    }
+    bridge._stop_acked.store(true, std::memory_order_release);
+    bridge.requestStop();
+    assert(bridge._stop_acked && bridge.ops.signals == 1); // duplicate cannot revoke ACK
+    bridge.loop();
+    assert_finished(bridge, true);
+    bridge.loop();
+    bridge.requestStop();
+    bridge.end();
+    assert(bridge.releases == 1 && bridge.ops.signals == 1 && bridge.cancels == 1);
+  }
+  for (int at_deadline = 0; at_deadline < 2; ++at_deadline) {
+    MQTTBridge bridge;
+    clock_ms = UINT32_MAX - 500; elapsed_ms = 0; sleeps = 0;
+    bridge_waiting = &bridge;
+    bridge.requestStop();
+    ack_after_ms = at_deadline ? bridge._lifecycle.stopTimeoutMs() : 8000;
+    bridge.end(); // synchronous OTA barrier joins the existing request
+    assert(elapsed_ms == ack_after_ms && sleeps > 0 && bridge.ops.signals == 1);
+    assert(!bridge._lifecycle.stopTimedOut());
+    assert_finished(bridge, true);
+  }
+  {
+    MQTTBridge bridge;
+    clock_ms = 0; elapsed_ms = 0; sleeps = 0;
+    bridge_waiting = &bridge; ack_after_ms = UINT64_MAX;
+    bridge.end(); // reviewed timeout remains confined to the explicit barrier
+    assert(bridge.ops.signals == 1 && bridge._lifecycle.stopTimedOut());
+    assert(elapsed_ms == bridge._lifecycle.stopTimeoutMs());
+    assert_finished(bridge, false);
+  }
+}
+'''
+
+
 class MqttTransportResultsTests(unittest.TestCase):
+    def test_actual_async_stop_preserves_resources_until_ack_and_ota_joins(self):
+        compiler = shutil.which("g++") or shutil.which("c++")
+        self.assertIsNotNone(compiler, "a host C++ compiler is required")
+        bridge = "src/helpers/bridges/MQTTBridge.cpp"
+        source = ASYNC_STOP_PREAMBLE + "\n".join(method(bridge, signature) for signature in (
+            "static inline uint32_t mqttStopTimeoutForSlots(int slots)",
+            "void MQTTBridge::requestStop()", "void MQTTBridge::end()",
+            "void MQTTBridge::finishStopped()", "void MQTTBridge::loop()",
+            "uint32_t MQTTBridge::LifecycleOps::nowMs()",
+            "void MQTTBridge::LifecycleOps::startTask()",
+            "void MQTTBridge::LifecycleOps::deliverStop()")) + ASYNC_STOP_MAIN
+        with tempfile.TemporaryDirectory(prefix="meshcore-mqtt-async-stop-") as temp:
+            path = Path(temp) / "stop.cpp"
+            path.write_text(source, encoding="utf-8")
+            exe = Path(temp) / ("stop.exe" if os.name == "nt" else "stop")
+            result = subprocess.run([compiler, "-std=c++17", "-Wall", "-Wextra",
+                                     "-I", str(ROOT / "src"), str(path), "-o", str(exe)],
+                                    text=True, capture_output=True)
+            self.assertEqual(result.returncode, 0, result.stderr)
+            result = subprocess.run([str(exe)], text=True, capture_output=True, timeout=5)
+            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+
+    def test_actual_worker_interrupts_startup_and_stagger_without_late_work(self):
+        compiler = shutil.which("g++") or shutil.which("c++")
+        self.assertIsNotNone(compiler, "a host C++ compiler is required")
+        bridge = "src/helpers/bridges/MQTTBridge.cpp"
+        source = TASK_PREAMBLE + method(bridge,
+            "bool MQTTBridge::waitUnlessStopping(uint32_t delay_ms)")
+        source += method(bridge, "void MQTTBridge::mqttTaskLoop()") + TASK_MAIN
+        with tempfile.TemporaryDirectory(prefix="meshcore-mqtt-stop-") as temp:
+            path = Path(temp) / "worker.cpp"
+            path.write_text(source, encoding="utf-8")
+            exe = Path(temp) / ("worker.exe" if os.name == "nt" else "worker")
+            result = subprocess.run([compiler, "-std=c++17", "-Wall", "-Wextra",
+                                     str(path), "-o", str(exe)], text=True, capture_output=True)
+            self.assertEqual(result.returncode, 0, result.stderr)
+            for scenario in range(12):
+                for tick_ms in (1, 10, 50):
+                    for wrap in (0, 1):
+                        with self.subTest(scenario=scenario, tick_ms=tick_ms, wrap=wrap):
+                            result = subprocess.run([str(exe), str(scenario), str(tick_ms), str(wrap)],
+                                                    text=True, capture_output=True, timeout=5)
+                            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+
     def test_real_adapter_and_bridge_methods_for_both_idf_layouts(self):
         compiler = shutil.which("g++") or shutil.which("c++")
         self.assertIsNotNone(compiler, "a host C++ compiler is required")

test/test_rs232_mqtt_runtime.py

@@ -35,7 +35,8 @@ class RS232MQTTRuntimeTests(unittest.TestCase):
             "bool setRs232BridgeState(bool enable)", "bool restartRs232Bridge()",
             "bool startSharedEspNowBridgeIfReady()", "bool isEspNowBridgeRunning()",
             "bool setEspNowBridgeState(bool enable)", "bool restartEspNowBridge()",
-            "bool isMqttBridgeRunning()", "bool setMqttBridgeState(bool enable)",
+            "bool isMqttBridgeRunning()", "bool requestMqttBridgeStop()", "bool isMqttBridgeStopping()",
+            "bool setMqttBridgeState(bool enable)",
             "bool restartMqttBridge()", "bool setBridgeState(bool enable)", "bool restartBridge()",
         ))
         defaults_start = implementation.index("// bridge defaults")
@@ -49,7 +50,7 @@ class RS232MQTTRuntimeTests(unittest.TestCase):
             start = implementation.index("#ifdef WITH_MQTT_BRIDGE\n", start)
             end = implementation.index("  if (_logging)", start)
             routing[name] = implementation[start:end]
-        service_start = implementation.index("#if defined(WITH_ESPNOW_BRIDGE)",
+        service_start = implementation.index("#if defined(WITH_MQTT_BRIDGE)\n  // MQTT owns TLS",
                                              implementation.index("MyMesh::servicePostMeshLoop()"))
         service_end = implementation.index("  if (next_flood_advert", service_start)
         pending_start = implementation.index("#if defined(WITH_BRIDGE)",
@@ -80,7 +81,8 @@ class RS232MQTTRuntimeTests(unittest.TestCase):
             'if (configKeyEquals(config, "bridge.uart"))',
         ))
         gets += "\nelse " + "\nelse ".join(extract_braced(observer, signature) for signature in (
-            'if (strcmp(config, "mqtt.enabled") == 0)', 'if (strcmp(config, "mqtt.running") == 0)'))
+            'if (strcmp(config, "mqtt.enabled") == 0)', 'if (strcmp(config, "mqtt.running") == 0)',
+            'if (strcmp(config, "mqtt.stopping") == 0)'))
         ota_start = cli.index('    } else if (memcmp(command, "start ota", 9)')
         ota_end = cli.index('    } else if (memcmp(command, "clock", 5)', ota_start)
         code = (ROOT / "test/fixtures/rs232_mqtt_runtime.cpp").read_text(encoding="ascii")
@@ -128,7 +130,8 @@ class RS232MQTTRuntimeTests(unittest.TestCase):
             "RS232Bridge* createRS232Bridge()", "bool beginRS232Bridge()", "bool endRS232Bridge()",
             "bool rs232BridgeEnabled() const", "bool isBridgeRunning() const override",
             "bool isRs232BridgeRunning()", "bool setRs232BridgeState(bool enable)",
-            "bool restartRs232Bridge()", "bool isMqttBridgeRunning()", "bool setBridgeState(bool enable)",
+            "bool restartRs232Bridge()", "bool isMqttBridgeRunning()", "bool requestMqttBridgeStop()",
+            "bool isMqttBridgeStopping()", "bool setBridgeState(bool enable)",
             "bool restartBridge()"))
         code = (ROOT / "test/fixtures/rs232_mqtt_runtime.cpp").read_text(encoding="ascii")
         code = code[:code.index("int main() {")]
@@ -144,7 +147,7 @@ class RS232MQTTRuntimeTests(unittest.TestCase):
         boot_start = implementation.index('mesh::hilStartupTrace("mesh_bridge_begin");')
         boot_start = implementation.index("#if defined(WITH_BRIDGE)", boot_start)
         boot_end = implementation.index('mesh::hilStartupTrace("mesh_bridge_ready");', boot_start)
-        service_start = implementation.index("#if defined(WITH_ESPNOW_BRIDGE)", implementation.index("MyMesh::servicePostMeshLoop()"))
+        service_start = implementation.index("#if defined(WITH_MQTT_BRIDGE)\n  // MQTT owns TLS", implementation.index("MyMesh::servicePostMeshLoop()"))
         service_end = implementation.index("  if (next_flood_advert", service_start)
         pending_start = implementation.index("#if defined(WITH_BRIDGE)", implementation.index("bool MyMesh::hasPendingWork()"))
         pending_end = implementation.index("  if (radio_driver.isWatchdogObserving()", pending_start)
@@ -199,6 +202,8 @@ class RS232MQTTRuntimeTests(unittest.TestCase):
   cli.run("stop ota", reply);
   assert(!mesh._cli.board.ota && mesh.isMqttBridgeRunning() && Serial2.begins == uart_starts);
   cli.set("rs232.enabled off", reply); cli.set("mqtt.enabled off", reply);
+  assert(!mesh.bridge && mesh.isMqttBridgeStopping() && mesh.bridgesPreventSleep());
+  mesh.serviceBridges();
   assert(!mesh.bridge && !mesh.isMqttBridgeRunning() && !mesh.bridgesPreventSleep());
   cli.set("rs232.enabled on", reply); mesh.mqtt_bridge->fail_starts = 1;
   cli.set("mqtt.enabled on", reply);