Recover stalled MQTT transports and broker acknowledgments

.github/workflows/run-unit-tests.yml

@@ -246,6 +246,7 @@ jobs:
           python3 -B test/test_mqtt_disconnect.py -v
           python3 -B test/test_mqtt_connection_admission.py -v
           python3 -B test/test_mqtt_handshake_scheduler.py -v
+          python3 -B test/test_mqtt_transport_health.py -v
           python3 -B test/test_mqtt_diagnostics.py -v
           python3 -B test/test_mqtt_publish_ack.py -v
           python3 -B test/test_wifi_network_power_policy.py -v

docs/WiFi.md

@@ -734,6 +734,20 @@ get mqtt1.diag
 get mqtt2.diag
 ```
 
+For a connection that works initially but later stops reporting, inspect each
+configured slot's `pub_ack` count in `get mqttN.diag`. Accepted publishes are
+not proof of delivery: a QoS 1 publish is confirmed only when the broker
+acknowledges it. A quiet radio channel alone is not a failed MQTT connection.
+
+The shared MQTT bridge recovers failed transports on its owning task, including
+errors without a subsequent disconnect callback and connection attempts that
+never finish. It also watches accepted QoS 1 publishes for missing broker
+acknowledgments. Recovery retains the existing broker configuration, memory
+admission and reconnect backoff; it does not reboot the node or cycle WiFi.
+See the [MQTT recovery regression](test-results/mqtt_tls_v4_regression.md#long-running-mqtt-recovery)
+for the deadlines, test coverage and hardware limitations. Older preview images
+do not gain these changes until their firmware is updated.
+
 Common causes are:
 
 - no saved SSID or a changed password;

docs/cli_command_availability.md

@@ -518,6 +518,15 @@ clear, set, or toggle selected bits.
 Prints the runtime diagnostic summary for MQTT slot N, including its connection
 state and failure history.
 
+`pub_ack` counts actual broker acknowledgments, not merely accepted writes.
+Recent firmware also shows `rec:<count>/<reason>` for transport recovery and
+`ack_wait:<seconds>s` while an accepted QoS 1 publication awaits acknowledgment.
+Reasons include `error`, `state`, `handshake`, `write`, and `puback`. `recover`
+means cleanup is pending, not that the broker is healthy. Optional fields are
+omitted if they cannot fit completely in the reply. `rec` counts completed
+client cleanup; a higher count alone does not prove the subsequent connection
+or delivery succeeded.
+
 <a id="mqtt-stats"></a>
 ### `get mqtt.stats`
 

docs/test-results/mqtt_tls_v4_regression.md

@@ -199,3 +199,69 @@ Comparison removes console echo/prompt framing and treats the old blank
 `get wifi.ssid` response as the new `(not configured)` response. MQTT and
 power saving were both returned to their original off state. No release assets
 were uploaded as part of this test.
+
+## Long-running MQTT recovery
+
+The [14 October 2026 follow-up report](https://github.com/mikecarper/meshfirmware/issues/16#issuecomment-6169059046)
+says the preview initially connects, but three reporting sites show it offline
+after roughly three hours while WiFi and the console still show connected.
+The earlier 180-second trial does not qualify this longer-running workload.
+
+Code inspection found recovery gaps, not a verified cause of that hardware
+failure. The adapter clears its connection flag on an MQTT error, but the
+bridge previously left its separate flag set until a disconnect callback.
+Its five-second cache refresh only reread that same flag. A lost terminal
+callback could also leave a connection attempt pending indefinitely, blocking
+the other brokers' handshake and renewal requests. Accepted status enqueues
+were treated as successful publication without using the existing PUBACK
+counter to detect stalled delivery.
+
+Recovery now runs on the bridge's owning task. An errored or timed-out SDK
+client must be proven stopped before its pending-handshake reservation is
+released. The client object and credentials are retained, and the existing
+admission checks and reconnect backoff still apply. No WiFi reset, node reboot,
+extra periodic heartbeat, or broker credential change is required. An idle
+connection with no outstanding QoS 1 publication is not timed out merely for
+being quiet.
+
+Both the pending-handshake deadline and the outstanding-PUBACK deadline are
+120 seconds. This leaves a generous margin over the SDK's 10-second network
+timeout and the configured 15-second QoS 1 retransmission interval. A failed
+SDK stop retains its recovery reservation instead of permitting another
+overlapping TLS handshake.
+
+CLI and WebConfig slot status show `recover` during reserved cleanup.
+`get mqttN.diag` retains `pub_ack` and adds a bounded recovery count/reason
+(`rec`) and pending acknowledgment age (`ack_wait`) when they fit. A delayed
+connected callback cannot cancel cleanup after an SDK stop failed.
+
+The delivery watch starts with the first accepted QoS 1 enqueue and observes
+real broker acknowledgments, including the on-connect status publication.
+Further enqueues cannot postpone that oldest outstanding deadline. Timeouts
+are serviced on the next owning-task pass; they are not a guarantee of
+wall-clock recovery if the worker is blocked inside an SDK call. This change
+does not establish or repair an indefinite SDK mutex deadlock.
+
+All 251 native MQTT test cases and 79 MQTT Python host tests passed. The new
+tests execute the production callbacks, publishing, maintenance, recovery and
+admission methods with controllable SDK calls and clocks. They include three
+brokers running beyond three simulated hours, missing PUBACKs and callbacks,
+silent synchronous write failures, failed SDK stops, late connected events,
+WiFi loss, suspension/foreground wake, backoff, memory admission and timer
+rollover. Removing the new error-callback recovery signal makes its regression
+test fail, establishing that it detects the previous behavior. AP ownership
+fixtures also passed with address/undefined-behavior sanitizers. The six pure
+transport-health tests are included in the 79-test total and in CI.
+
+The exact Full `heltec_v4_repeater_observer_mqtt` image compiled and linked
+against Arduino 2.0.17 / ESP-IDF 4.4.7. This local test image used baseline
+`dce323e37` with the MQTT repair applied; it predates the subsequently pulled
+beacon changes. Its application size was 2,237,677
+bytes, within the 6,553,600-byte application slot. The linked runtime RAM gate
+reported 204,752 internal bytes available, 198,432 required and a 171,984-byte
+largest region. These are build-time capacity/admission checks, not free heap
+measured with three live brokers or qualification of maximum simultaneous load.
+
+No new physical three-hour V4 soak or reporter-side retest has been performed
+as part of this repair. These tests establish the repaired recovery behavior,
+not the cause of the reporter's hardware failure or website-side availability.

src/helpers/MQTTTransportHealth.h

@@ -0,0 +1,79 @@
+#pragma once
+
+#include <stdint.h>
+
+// Owner-task transport recovery, separate from the reconnect/backoff policy.
+// A CONNECTED callback or accepted/enqueued publish is not proof of delivery.
+namespace MQTTTransportHealth {
+
+// Twelve times the normal 10-second socket budget gives DNS/TLS/WebSocket/
+// CONNACK setup ample room, while bounding a lost final callback. PUBACK uses
+// the same generous budget: longer than the 15-second retry and SDK outbox
+// expiry, so an ordinary retransmit cannot trigger a premature reconnect.
+static const uint32_t kAttemptTimeoutMs = 120000UL;
+static const uint32_t kAckTimeoutMs = 120000UL;
+
+enum class RecoveryReason : uint8_t {
+  None, TransportError, AdapterDisconnected, HandshakeTimeout,
+  PublishFailure, AckTimeout
+};
+
+inline const char* reasonLabel(RecoveryReason reason) {
+  switch (reason) {
+    case RecoveryReason::TransportError: return "error";
+    case RecoveryReason::AdapterDisconnected: return "state";
+    case RecoveryReason::HandshakeTimeout: return "handshake";
+    case RecoveryReason::PublishFailure: return "write";
+    case RecoveryReason::AckTimeout: return "puback";
+    default: return "none";
+  }
+}
+
+struct State {
+  bool attempt_active = false;
+  uint32_t attempt_started_ms = 0;
+  bool ack_pending = false;
+  uint32_t ack_started_ms = 0;
+  uint32_t ack_snapshot = 0;
+  RecoveryReason recovery_reason = RecoveryReason::None;
+  RecoveryReason last_recovery_reason = RecoveryReason::None;
+  uint32_t recoveries = 0;
+
+  void beginAttempt(uint32_t now) {
+    attempt_active = true;
+    attempt_started_ms = now;
+  }
+  void completeAttempt() { attempt_active = false; }
+  void observeAck(uint32_t count) {
+    if (ack_pending && count != ack_snapshot) ack_pending = false;
+  }
+  void acceptedQos1(uint32_t now, uint32_t ack_before) {
+    observeAck(ack_before);
+    if (!ack_pending) {
+      ack_pending = true;
+      ack_started_ms = now;
+      ack_snapshot = ack_before;
+    }
+  }
+  bool attemptExpired(uint32_t now) const {
+    return attempt_active && now - attempt_started_ms >= kAttemptTimeoutMs;
+  }
+  bool ackExpired(uint32_t now) const {
+    return ack_pending && now - ack_started_ms >= kAckTimeoutMs;
+  }
+  void requestRecovery(RecoveryReason reason) {
+    if (recovery_reason == RecoveryReason::None) recovery_reason = reason;
+  }
+  void resetSession() {
+    attempt_active = false;
+    ack_pending = false;
+  }
+  void recovered() {
+    if (recoveries != UINT32_MAX) ++recoveries;
+    last_recovery_reason = recovery_reason;
+    recovery_reason = RecoveryReason::None;
+    resetSession();
+  }
+};
+
+} // namespace MQTTTransportHealth

src/helpers/bridges/MQTTBridge.cpp

@@ -426,6 +426,10 @@ void MQTTBridge::formatMqttStatusReply(char* buf, size_t bufsize, const MQTTPref
     } else if (!b->isSlotReady(i)) {
       name = slot.preset ? slot.preset->name : "custom";
       state = "wait";
+    } else if (b->_slot_health_recovering[i].load(std::memory_order_acquire) ||
+               b->_slot_transport_fault_pending[i].load(std::memory_order_acquire)) {
+      name = slot.preset ? slot.preset->name : "custom";
+      state = "recover";
     } else if (slot.connected) {
       name = slot.preset ? slot.preset->name : "custom";
       state = "ok";
@@ -551,6 +555,9 @@ bool MQTTBridge::getSlotStatusSnapshot(int slot_index, SlotStatusSnapshot* out)
     out->name = slot.preset ? slot.preset->name : "custom";
     if (!b->isSlotReady(slot_index)) {
       out->state = "wait";
+    } else if (b->_slot_health_recovering[slot_index].load(std::memory_order_acquire) ||
+               b->_slot_transport_fault_pending[slot_index].load(std::memory_order_acquire)) {
+      out->state = "recover";
     } else if (slot.connected) {
       out->state = "ok";
     } else if (slot.circuit_breaker_tripped) {
@@ -657,6 +664,9 @@ void MQTTBridge::formatSlotDiagReply(char* buf, size_t bufsize, int slot_index)
   } else if (!slot.client) {
     // Ready to connect but the client object could not be allocated.
     state = "no client";
+  } else if (b->_slot_health_recovering[slot_index].load(std::memory_order_acquire) ||
+             b->_slot_transport_fault_pending[slot_index].load(std::memory_order_acquire)) {
+    state = "recover";
   } else if (slot.connected) {
     state = "ok";
   } else if (slot.circuit_breaker_tripped) {
@@ -689,7 +699,7 @@ void MQTTBridge::formatSlotDiagReply(char* buf, size_t bufsize, int slot_index)
   }
 
   // Connected with no errors: nothing more to say about the connection.
-  if (slot.connected && slot.last_error_time == 0) {
+  if (slot.connected && strcmp(state, "ok") == 0 && slot.last_error_time == 0) {
     replyAppendf(buf, bufsize, &pos, ", no errors");
   } else if (slot.last_error_time > 0) {
     // TLS error with human-friendly description
@@ -743,6 +753,28 @@ void MQTTBridge::formatSlotDiagReply(char* buf, size_t bufsize, int slot_index)
     }
   }
 
+  // Recovery/ACK age are owner-task snapshots. Append complete fields only;
+  // error details and the existing broker ACK count retain reply priority.
+  const uint32_t recoveries = b->_slot_health_recoveries[slot_index].load(std::memory_order_relaxed);
+  const auto recovery_reason = static_cast<MQTTTransportHealth::RecoveryReason>(
+      b->_slot_health_reason[slot_index].load(std::memory_order_relaxed));
+  if (recoveries || recovery_reason != MQTTTransportHealth::RecoveryReason::None) {
+    char recovery_text[48];
+    const int len = snprintf(recovery_text, sizeof(recovery_text), ", rec:%lu/%s",
+        static_cast<unsigned long>(recoveries), MQTTTransportHealth::reasonLabel(recovery_reason));
+    if (len > 0 && len < (int)sizeof(recovery_text) && len <= (int)bufsize - 1 - pos) {
+      replyAppendf(buf, bufsize, &pos, "%s", recovery_text);
+    }
+  }
+  const uint32_t ack_wait = b->_slot_health_ack_wait_secs[slot_index].load(std::memory_order_relaxed);
+  if (ack_wait != UINT32_MAX) {
+    char wait_text[32];
+    const int len = snprintf(wait_text, sizeof(wait_text), ", ack_wait:%lus", static_cast<unsigned long>(ack_wait));
+    if (len > 0 && len < (int)sizeof(wait_text) && len <= (int)bufsize - 1 - pos) {
+      replyAppendf(buf, bufsize, &pos, "%s", wait_text);
+    }
+  }
+
   // Appended last so it never displaces connection diagnostics. replyAppendf
   // clamps rather than overflows, but a clipped type list is worse than no
   // list: "...,13,14," parses as a real, different allowlist, and this is the
@@ -875,11 +907,11 @@ MQTTBridge::MQTTBridge(const MQTTNodeInfo& node_info, MQTTPrefs *obs,
 
   // Initialize all slots to empty/disabled state
   for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) {
-    memset(&_slots[i], 0, sizeof(MQTTSlot));
+    _slots[i] = MQTTSlot{};
     _slots[i].enabled = false;
     _slots[i].client = nullptr;
     _slots[i].preset = nullptr;
-    // auth_token == nullptr after memset above - allocated on first token creation
+    // auth_token == nullptr after value initialization - allocated on first token creation
     _slots[i].connected = false;
     _slots[i].initial_connect_done = false;
     _slots[i].token_expires_at = 0;
@@ -893,6 +925,13 @@ MQTTBridge::MQTTBridge(const MQTTNodeInfo& node_info, MQTTPrefs *obs,
     _slot_reconfigure_pending[i] = false;
     _slot_force_jwt_mint[i] = false;
     _slot_attempt_pending[i] = false;
+    _slot_transport_fault_pending[i] = false;
+    _slot_health_reset_pending[i] = false;
+    _slots[i].transport_health = MQTTTransportHealth::State();
+    _slot_health_recoveries[i] = 0;
+    _slot_health_reason[i] = static_cast<uint8_t>(MQTTTransportHealth::RecoveryReason::None);
+    _slot_health_recovering[i] = false;
+    _slot_health_ack_wait_secs[i] = UINT32_MAX;
     _status_publish_pending[i] = false;
   }
 
@@ -1024,6 +1063,11 @@ void MQTTBridge::begin() {
     _slots[i].broker_uri[0] = '\0';
     _slots[i].port = 1883;
     _slot_reconfigure_pending[i] = false;
+    _slot_transport_fault_pending[i] = false;
+    _slot_health_reset_pending[i] = false;
+    _slots[i].transport_health.resetSession();
+    _slots[i].transport_health.recovery_reason = MQTTTransportHealth::RecoveryReason::None;
+    updateSlotTransportDiag(i, static_cast<uint32_t>(millis()));
   }
   _cached_has_connected_slots = false;
   _slots_setup_done = false;
@@ -1617,13 +1661,33 @@ bool MQTTBridge::serviceNetworkPower(uint32_t now) {
       cancelNtpRefresh();
       _ntp_sync_pending = false;
       for (int i = 0; i < RUNTIME_MQTT_SLOTS; ++i) {
+        // A foreground/probe request can resume the network policy after a
+        // failed suspension. Keep a transport reservation until stop is proven.
+        if (_slots[i].client && _slots[i].client->isStarted()) {
+          _slot_transport_fault_pending[i].store(true, std::memory_order_release);
+        }
         _slot_attempt_pending[i] = false;
         if (_slots[i].client && _slots[i].client->isStarted()) {
           _slots[i].client->disconnect();
           // The wrapper only clears isStarted after SDK stop succeeds. Never
           // hand station ownership off while any transport still runs.
-          if (_slots[i].client->isStarted()) return false;
+          if (_slots[i].client->isStarted()) {
+            _slots[i].transport_health.requestRecovery(MQTTTransportHealth::RecoveryReason::AdapterDisconnected);
+            _slots[i].connected = false;
+            _slots[i].connected_at_ms = 0;
+            if (_slots[i].current_outage_started_ms == 0) {
+              _slots[i].current_outage_started_ms = now ? now : 1;
+            }
+            updateSlotTransportDiag(i, static_cast<uint32_t>(millis()));
+            updateCachedConnectionStatus();
+            return false;
+          }
         }
+        _slot_transport_fault_pending[i].store(false, std::memory_order_release);
+        _slot_health_reset_pending[i].store(false, std::memory_order_release);
+        _slots[i].transport_health.resetSession();
+        _slots[i].transport_health.recovery_reason = MQTTTransportHealth::RecoveryReason::None;
+        updateSlotTransportDiag(i, static_cast<uint32_t>(millis()));
         _slots[i].connected = false;
         _slots[i].connected_at_ms = 0;
         _status_publish_pending[i] = false;
@@ -2040,8 +2104,8 @@ void MQTTBridge::mqttTaskLoop() {
       }
     }
 
-    // Update cached connection status periodically (every 5 seconds)
-    // This ensures cache stays accurate even if callbacks miss updates
+    // Refresh the aggregate cache periodically; owner maintenance above
+    // reconciles adapter state and bounds missed terminal callbacks.
     static unsigned long last_slot_status_update = 0;
     if (now - last_slot_status_update > 5000) {
       updateCachedConnectionStatus();
@@ -2092,6 +2156,7 @@ bool MQTTBridge::ensureSlotClient(int index) {
     MQTT_DEBUG_PRINTLN("MQTT%d connected", index + 1);
     _slot_attempt_pending[index] = false;
     _slots[index].connected = true;
+    _slot_health_reset_pending[index].store(true, std::memory_order_release);
     _slot_force_jwt_mint[index] = false;
     // NOTE: reconnect_backoff / max_backoff_failures are NOT reset here.
     // A CONNACK alone doesn't prove the link is healthy -- a broker that
@@ -2142,6 +2207,10 @@ bool MQTTBridge::ensureSlotClient(int index) {
     updateCachedConnectionStatus();
   });
   slot.client->onError([this, index](esp_mqtt_error_codes error) {
+    // The adapter is already disconnected at MQTT_EVENT_ERROR. Do not stop
+    // its SDK task from inside that task's callback, or release the global
+    // handshake reservation before the owner proves the SDK has stopped.
+    _slot_transport_fault_pending[index].store(true, std::memory_order_release);
     _slots[index].last_tls_err = error.esp_tls_last_esp_err;
     _slots[index].last_tls_stack_err = error.esp_tls_stack_err;
     _slots[index].last_sock_errno = error.esp_transport_sock_errno;
@@ -2255,7 +2324,9 @@ bool MQTTBridge::hasPendingSlotConnection() const {
   // Callbacks run on esp-mqtt tasks. Time spacing alone cannot establish that
   // an earlier handshake has finished, particularly on a slow WiFi link.
   for (int i = 0; i < RUNTIME_MQTT_SLOTS; ++i) {
-    if (_slot_attempt_pending[i].load(std::memory_order_acquire)) return true;
+    if (_slot_attempt_pending[i].load(std::memory_order_acquire) ||
+        _slot_transport_fault_pending[i].load(std::memory_order_acquire) ||
+        _slots[i].transport_health.recovery_reason != MQTTTransportHealth::RecoveryReason::None) return true;
   }
   return false;
 }
@@ -2326,6 +2397,9 @@ bool MQTTBridge::setupSlot(int index) {
       slot.client->disconnect();
     }
     if (slot.client->isStarted()) {
+      _slot_transport_fault_pending[index].store(true, std::memory_order_release);
+      slot.transport_health.requestRecovery(MQTTTransportHealth::RecoveryReason::AdapterDisconnected);
+      updateSlotTransportDiag(index, static_cast<uint32_t>(millis()));
       slot.last_reconnect_attempt = millis();
       return false;
     }
@@ -2558,7 +2632,14 @@ void MQTTBridge::teardownSlot(int index, bool force) {
     delay(50);
     #endif
   }
-  if (slot.client && !force && slot.client->isStarted()) return;
+  if (slot.client && !force && slot.client->isStarted()) {
+    // Even a deliberate reconfigure must retain admission ownership if SDK
+    // stop failed. The worker will retry cleanup; no second handshake may run.
+    _slot_transport_fault_pending[index].store(true, std::memory_order_release);
+    slot.transport_health.requestRecovery(MQTTTransportHealth::RecoveryReason::AdapterDisconnected);
+    updateSlotTransportDiag(index, static_cast<uint32_t>(millis()));
+    return;
+  }
 
   // Invalidate the token but keep the buffer: the client survives teardown and still
   // holds this pointer in its config (see MQTTSlot::auth_token).
@@ -2575,6 +2656,11 @@ void MQTTBridge::teardownSlot(int index, bool force) {
   slot.last_log_time = 0;
   slot.last_deferred_log_ms = 0;
   _slot_force_jwt_mint[index] = false;
+  _slot_transport_fault_pending[index] = false;
+  _slot_health_reset_pending[index] = false;
+  slot.transport_health.resetSession();
+  slot.transport_health.recovery_reason = MQTTTransportHealth::RecoveryReason::None;
+  updateSlotTransportDiag(index, static_cast<uint32_t>(millis()));
 }
 
 esp_err_t MQTTBridge::reconnectSlotClient(int index) {
@@ -2586,6 +2672,7 @@ esp_err_t MQTTBridge::reconnectSlotClient(int index) {
 
   const bool was_pending = _slot_attempt_pending[index];
   const bool starting = !slot.client->isStarted();
+  slot.transport_health.beginAttempt(static_cast<uint32_t>(millis()));
   _slot_attempt_pending[index] = true;  // callbacks may run before the call returns
   esp_err_t result;
   if (starting) {
@@ -2597,6 +2684,7 @@ esp_err_t MQTTBridge::reconnectSlotClient(int index) {
   if (result != ESP_OK) {
     if (MQTTConnectionHealth::clearPendingAfterRejectedRequest(starting, was_pending)) {
       _slot_attempt_pending[index] = false;
+      slot.transport_health.completeAttempt();
     }
     if (starting) {
       slot.start_failures = MQTTConnectionHealth::incrementFailures(slot.start_failures);
@@ -2605,8 +2693,112 @@ esp_err_t MQTTBridge::reconnectSlotClient(int index) {
   return result;
 }
 
+void MQTTBridge::updateSlotTransportDiag(int index, uint32_t now) {
+  const MQTTTransportHealth::State& health = _slots[index].transport_health;
+  _slot_health_recoveries[index].store(health.recoveries, std::memory_order_relaxed);
+  const auto reason = health.recovery_reason != MQTTTransportHealth::RecoveryReason::None
+      ? health.recovery_reason : health.last_recovery_reason;
+  _slot_health_reason[index].store(static_cast<uint8_t>(reason), std::memory_order_relaxed);
+  _slot_health_recovering[index].store(
+      health.recovery_reason != MQTTTransportHealth::RecoveryReason::None,
+      std::memory_order_release);
+  _slot_health_ack_wait_secs[index].store(
+      health.ack_pending ? (now - health.ack_started_ms) / 1000UL : UINT32_MAX,
+      std::memory_order_relaxed);
+}
+
+bool MQTTBridge::recoverSlotTransport(int index, uint32_t now) {
+  MQTTSlot& slot = _slots[index];
+  auto& health = slot.transport_health;
+  if (!slot.client || health.recovery_reason == MQTTTransportHealth::RecoveryReason::None) return false;
+
+  MQTT_DEBUG_PRINTLN("MQTT%d transport recovery: %s", index + 1,
+                    MQTTTransportHealth::reasonLabel(health.recovery_reason));
+  const bool failed_attempt = MQTTConnectionHealth::disconnectFailedAttempt(
+      _slot_attempt_pending[index].load(std::memory_order_acquire), slot.connected);
+  const uint32_t failures_before_stop = slot.connect_failures;
+  // Retain the client, SDK buffers and credentials. disconnect() closes the
+  // transport and stops its task, and leaves isStarted() true if SDK stop fails.
+  // Never vTaskDelete a live SDK task or admit a second TLS handshake on failure.
+  slot.client->disconnect();
+  if (slot.client->isStarted()) {
+    updateSlotTransportDiag(index, static_cast<uint32_t>(millis()));
+    return false;
+  }
+
+  // SDK stop can consume its socket budget. Space the next handshake from
+  // completed cleanup, not from the earlier watchdog observation.
+  now = static_cast<uint32_t>(millis());
+
+  // A real DISCONNECTED callback may already have accounted this failure.
+  // ERROR-only/lost-callback attempts need the same accounting, once.
+  if (failed_attempt && slot.connect_failures == failures_before_stop) {
+    slot.connect_failures = MQTTConnectionHealth::incrementFailures(slot.connect_failures);
+  }
+  _slot_attempt_pending[index].store(false, std::memory_order_release);
+  _slot_transport_fault_pending[index].store(false, std::memory_order_release);
+  _slot_health_reset_pending[index].store(false, std::memory_order_release);
+  _status_publish_pending[index] = false;
+  slot.connected = false;
+  slot.connected_at_ms = 0;
+  if (slot.current_outage_started_ms == 0) slot.current_outage_started_ms = now ? now : 1;
+  slot.last_reconnect_attempt = now;
+  _last_slot_reconnect_ms = now;
+  health.recovered();
+  updateSlotTransportDiag(index, now);
+  updateCachedConnectionStatus();
+  return true;
+}
+
+void MQTTBridge::serviceSlotTransportHealth(uint32_t now) {
+  if (_stop_requested.load(std::memory_order_acquire)) return;
+  for (int i = 0; i < RUNTIME_MQTT_SLOTS; ++i) {
+    MQTTSlot& slot = _slots[i];
+    auto& health = slot.transport_health;
+    if (_slot_health_reset_pending[i].exchange(false, std::memory_order_acq_rel)) {
+      // A late CONNACK does not release an already-required SDK stop. Its
+      // session clocks may reset, but only proven cleanup clears recovery.
+      health.resetSession();
+    }
+    const bool pending = _slot_attempt_pending[i].load(std::memory_order_acquire);
+    if (!pending) health.completeAttempt();
+    if (slot.client) {
+      health.observeAck(static_cast<uint32_t>(slot.client->getPublishAck()));
+      if (_slot_transport_fault_pending[i].load(std::memory_order_acquire)) {
+        health.requestRecovery(MQTTTransportHealth::RecoveryReason::TransportError);
+      } else if (slot.connected && !slot.client->connected()) {
+        health.requestRecovery(MQTTTransportHealth::RecoveryReason::AdapterDisconnected);
+      } else if (pending && health.attemptExpired(now)) {
+        health.requestRecovery(MQTTTransportHealth::RecoveryReason::HandshakeTimeout);
+      } else if (slot.connected && health.ackExpired(now)) {
+        health.requestRecovery(MQTTTransportHealth::RecoveryReason::AckTimeout);
+      }
+      if (!slot.connected && !pending) health.resetSession();
+    }
+    if (health.recovery_reason != MQTTTransportHealth::RecoveryReason::None) {
+      // Stop advertising a healthy link immediately, including while SDK
+      // cleanup is still failing. Admission is held by the recovery state,
+      // independently of this observable connection flag.
+      slot.connected = false;
+      slot.connected_at_ms = 0;
+      if (slot.current_outage_started_ms == 0) slot.current_outage_started_ms = now ? now : 1;
+    }
+    updateSlotTransportDiag(i, now);
+  }
+  updateCachedConnectionStatus();
+  // Close no more than one SDK transport per pass. A failed stop keeps its
+  // recovery reservation and blocks all new handshakes on subsequent passes.
+  for (int i = 0; i < RUNTIME_MQTT_SLOTS; ++i) {
+    if (_slots[i].client && _slots[i].transport_health.recovery_reason != MQTTTransportHealth::RecoveryReason::None) {
+      recoverSlotTransport(i, now);
+      break;
+    }
+  }
+}
+
 void MQTTBridge::maintainSlotConnections() {
   if (_stop_requested.load(std::memory_order_acquire)) return;
+  serviceSlotTransportHealth(static_cast<uint32_t>(millis()));
   if (!_identity) return;
 
   // Check WiFi status first
@@ -3003,6 +3195,14 @@ bool MQTTBridge::publishToSlot(int index, const char* topic, const char* payload
   if (_stop_requested.load(std::memory_order_acquire)) return false;
   if (index < 0 || index >= RUNTIME_MQTT_SLOTS) return false;
   MQTTSlot& slot = _slots[index];
+  // On-connect status is processed before maintenance. Consume this new-
+  // session latch BEFORE enqueueing it, or maintenance would erase its ACK
+  // watch immediately afterwards. Only this owner task mutates health state.
+  if (_slot_health_reset_pending[index].exchange(false, std::memory_order_acq_rel)) {
+    slot.transport_health.resetSession();
+  }
+  if (slot.transport_health.recovery_reason != MQTTTransportHealth::RecoveryReason::None ||
+      _slot_transport_fault_pending[index].load(std::memory_order_acquire)) return false;
   if (!slot.client || !slot.connected) {
     unsigned long now = millis();
     if (now - slot.last_log_time > SLOT_LOG_INTERVAL) {
@@ -3020,16 +3220,23 @@ bool MQTTBridge::publishToSlot(int index, const char* topic, const char* payload
   //    write bypasses that drain ceiling entirely and does not store in the outbox. It can
   //    block the (Core-0, prio-1) MQTT task on a stalled socket, but only up to
   //    network_timeout_ms (configured in optimizeMqttClientConfig); mesh RX (Core 1) and the
-  //    WiFi/TCP stack (higher-prio system tasks) are unaffected, and a failed write flips
-  //    the slot to disconnected so subsequent packets skip it.
+  //    WiFi/TCP stack (higher-prio system tasks) are unaffected. A failed write reserves
+  //    owner-task transport recovery, so subsequent packets skip the failed slot.
   //  - QoS 1 (low-rate retained status): async, so it keeps the durable outbox + retransmit.
   //
   // Return convention: QoS 0 sync publish returns msg_id == 0 on success (no PUBACK
   // tracking). Negative values (-1 write/failure) are the only actual failures; the queue
   // retry/drop path below handles them.
   bool async = (qos > 0);
+  // Take the actual broker ACK counter before the SDK call: a fast PUBACK can
+  // arrive before enqueue returns. A post-call snapshot would lose that proof.
+  const uint32_t ack_before = static_cast<uint32_t>(slot.client->getPublishAck());
   int result = slot.client->publish(topic, qos, retained, payload, (int)payload_len, async);
   if (result < 0) {
+    if (!async) {
+      slot.transport_health.requestRecovery(MQTTTransportHealth::RecoveryReason::PublishFailure);
+      updateSlotTransportDiag(index, static_cast<uint32_t>(millis()));
+    }
     // QoS0 packet/raw publishes are best-effort and may be retried from the
     // bridge queue; avoid logging transient first-attempt failures here.
     if (qos > 0) {
@@ -3042,6 +3249,12 @@ bool MQTTBridge::publishToSlot(int index, const char* topic, const char* payload
     }
     return false;
   }
+  if (qos > 0) {
+    slot.transport_health.acceptedQos1(static_cast<uint32_t>(millis()), ack_before);
+    // Also reflect an ACK delivered during enqueue, not just on the next tick.
+    slot.transport_health.observeAck(static_cast<uint32_t>(slot.client->getPublishAck()));
+    updateSlotTransportDiag(index, static_cast<uint32_t>(millis()));
+  }
   return true;
 }
 
@@ -3189,8 +3402,7 @@ void MQTTBridge::publishStatusToSlot(int index) {
     // 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, len);
-    if (result <= 0) {
+    if (!publishToSlot(index, status_topic, json_buffer, static_cast<size_t>(len), use_retain, 1)) {
       MQTT_DEBUG_PRINTLN("MQTT%d status publish failed", index + 1);
     }
   }
@@ -3453,11 +3665,28 @@ bool MQTTBridge::handleWiFiConnection(unsigned long now) {
       s_wifi_connected_at = 0;
       // Disconnect all slot clients when WiFi drops
       for (int i = 0; i < RUNTIME_MQTT_SLOTS; i++) {
-        if (_slots[i].client && _slots[i].connected) {
+        // Include pending handshakes, not just sessions that reached CONNACK.
+        if (_slots[i].client && _slots[i].client->isStarted()) {
+          _slot_transport_fault_pending[i].store(true, std::memory_order_release);
           _slot_attempt_pending[i] = false;
           _slots[i].client->disconnect();
+          _slots[i].connected = false;
+          _slots[i].connected_at_ms = 0;
+          if (_slots[i].client->isStarted()) {
+            _slots[i].transport_health.requestRecovery(MQTTTransportHealth::RecoveryReason::AdapterDisconnected);
+            if (_slots[i].current_outage_started_ms == 0) {
+              _slots[i].current_outage_started_ms = now ? now : 1;
+            }
+          } else {
+            _slot_transport_fault_pending[i].store(false, std::memory_order_release);
+            _slot_health_reset_pending[i].store(false, std::memory_order_release);
+            _slots[i].transport_health.resetSession();
+            _slots[i].transport_health.recovery_reason = MQTTTransportHealth::RecoveryReason::None;
+          }
+          updateSlotTransportDiag(i, static_cast<uint32_t>(millis()));
         }
       }
+      updateCachedConnectionStatus();
     } else if (snapshot.down) {
       // Backoff ladder + wrap-safe timing live in MQTTConnectionPolicy (Phase 6),
       // exercised by host tests. Behavior is unchanged: both the link-down

src/helpers/bridges/MQTTBridge.h

@@ -12,6 +12,7 @@
 #include "helpers/JWTHelper.h"
 #include "helpers/MQTTConnectionPolicy.h"
 #include "helpers/MQTTConnectionHealth.h"
+#include "helpers/MQTTTransportHealth.h"
 #include "helpers/MQTTPayloadBuilder.h"
 #include "helpers/MQTTPacketFilter.h"
 #include "helpers/MQTTPresets.h"
@@ -158,6 +159,10 @@ private:
     // DISCONNECTED (including a failed initial attempt). first_disconnect_time is intentionally separate
     // so the existing 'mqttN.diag' "first_disc" semantics don't change.
     unsigned long current_outage_started_ms;
+
+    // Only the bridge worker mutates this state; SDK callbacks raise atomic
+    // latches instead. Retained across reconnects to expose recovery history.
+    MQTTTransportHealth::State transport_health;
   };
 
   MQTTSlot _slots[RUNTIME_MQTT_SLOTS];
@@ -260,6 +265,13 @@ private:
   // Set before starting/reconnecting, cleared by connect/failed-attempt/stop.
   // Must be set before the SDK call because its event task can callback at once.
   std::atomic<bool> _slot_attempt_pending[RUNTIME_MQTT_SLOTS];
+  std::atomic<bool> _slot_transport_fault_pending[RUNTIME_MQTT_SLOTS];
+  std::atomic<bool> _slot_health_reset_pending[RUNTIME_MQTT_SLOTS];
+  // Read-only CLI snapshots: never read the owner-only State across tasks.
+  std::atomic<uint32_t> _slot_health_recoveries[RUNTIME_MQTT_SLOTS];
+  std::atomic<uint8_t> _slot_health_reason[RUNTIME_MQTT_SLOTS];
+  std::atomic<bool> _slot_health_recovering[RUNTIME_MQTT_SLOTS];
+  std::atomic<uint32_t> _slot_health_ack_wait_secs[RUNTIME_MQTT_SLOTS];
 
   // Pending on-connect status publish: set from the onConnect callback (which
   // runs on the esp-mqtt event task, NOT this bridge task), consumed by the MQTT
@@ -509,6 +521,9 @@ private:
   bool canStartSlotConnection(int index) const;
   void teardownSlot(int index, bool force = false);
   esp_err_t reconnectSlotClient(int index);
+  bool recoverSlotTransport(int index, uint32_t now);
+  void serviceSlotTransportHealth(uint32_t now);
+  void updateSlotTransportDiag(int index, uint32_t now);
   void maintainSlotConnections();      // Maintain all slot connections (token renewal, reconnect)
   void maintainSlotConnection(int index, unsigned long now_millis, unsigned long current_time, bool time_synced, bool& reconnect_attempted, bool& teardown_attempted);
   bool createSlotAuthToken(int index); // Create/renew JWT token for a slot

test/README.md

@@ -167,6 +167,7 @@ does not reflect the GoogleTest count -- run the built binary directly
 | `test_topic_template` | `src/helpers/MQTTTopicTemplate.h` | `{iata}/{device}/{token}/{type}` expansion, overflow/NUL-termination, and a buffer-size fuzz |
 | `test_mqtt_topic_router` | `src/helpers/MQTTTopicRouter.h` | complete preset/custom topic-routing contract; MeshRank all types except raw; required identifiers; invalid inputs/slots; exact buffer boundaries |
 | `test_mqtt_connection_policy` | `src/helpers/MQTTConnectionPolicy.h` | reconnect guard/backoff/stagger and breaker transitions; stable reset; JWT lifetime/renewal policy; exact timing boundaries and 32-bit `millis()` rollover |
+| `test_mqtt_transport_health.py`, `test_mqtt_handshake_scheduler.py` | `src/helpers/MQTTTransportHealth.h`, production `MQTTBridge` methods | oldest outstanding QoS 1 acknowledgment deadline; quiet-session safety; lost/error-only callbacks; serialized stop-before-reconnect recovery; multi-hour scheduling and timer rollover |
 | `test_wifi_reconnect_policy` | `src/helpers/WiFiReconnectPolicy.h` | five-minute forced reconnect cadence, connection resets, duplicate disconnect observations, and 32-bit `millis()` rollover |
 | `test_radio_liveness` | `src/helpers/RadioLivenessTracker.h`, `src/helpers/radiolib/RxBoostedGainDefaults.h` | staged radio recovery, activity reset and rollover handling; target-specific RX boosted-gain defaults and SX126x precedence |
 | `test_lr1110_rx_recovery` | `src/helpers/radiolib/LR1110RxRecovery.h` | LR1110 four-byte RX-buffer shift signature; captured and accumulated shifts; ordinary/scoped packet exclusions |

test/fixtures/mqtt_ap_ownership.cpp

@@ -6,6 +6,7 @@
 #include <vector>
 #include "helpers/AlertFaultPolicy.h"
 #include "helpers/MQTTConnectionPolicy.h"
+#include "helpers/MQTTTransportHealth.h"
 #include "helpers/WiFiPowerSave.h"
 #include "helpers/WiFiNetworkPowerPolicy.h"
 #define MQTT_DEBUG_PRINTLN(...) ((void)0)
@@ -90,8 +91,16 @@ struct MQTTBridge{
  struct{bool canonical_wifi=true;}_node_info;
  struct{uint8_t wifi_power_save=1;}prefs,*_obs=&prefs;
  struct Client{void disconnect(){}bool isStarted()const{return false;}};
- struct Slot{Client* client=nullptr;bool connected=false;}_slots[RUNTIME_MQTT_SLOTS];
+ struct Slot{
+  Client* client=nullptr;bool connected=false;
+  uint32_t connected_at_ms=0,current_outage_started_ms=0;
+  MQTTTransportHealth::State transport_health;
+ }_slots[RUNTIME_MQTT_SLOTS];
  bool _slot_attempt_pending[RUNTIME_MQTT_SLOTS]={};
+ std::atomic<bool> _slot_transport_fault_pending[RUNTIME_MQTT_SLOTS]={};
+ std::atomic<bool> _slot_health_reset_pending[RUNTIME_MQTT_SLOTS]={};
+ void updateSlotTransportDiag(int,uint32_t){}
+ void updateCachedConnectionStatus(){}
  std::vector<std::string> cleanup;
  void cancelNtpRefresh(){cleanup.push_back("ntp");}
  void teardownSlot(int index){cleanup.push_back("slot"+std::to_string(index));}

test/test_mqtt_connection_admission.py

@@ -15,6 +15,7 @@ POLICY_PREAMBLE = r'''
 #include <cstddef>
 #include <cstdint>
 #include "helpers/MQTTConnectionAdmission.h"
+#include "helpers/MQTTTransportHealth.h"
 namespace Admission = MQTTConnectionAdmission;
 using MemoryState = Admission::ClientMemoryState;
 '''
@@ -65,8 +66,10 @@ struct MQTTBridge {
     const Preset* preset = nullptr;
     char broker_uri[128] = {}, host[64] = "broker";
     uint16_t port = 8883;
+    MQTTTransportHealth::State transport_health;
   } _slots[RUNTIME_MQTT_SLOTS];
   std::atomic<bool> _slot_attempt_pending[RUNTIME_MQTT_SLOTS] = {};
+  std::atomic<bool> _slot_transport_fault_pending[RUNTIME_MQTT_SLOTS] = {};
   bool hasPendingSlotConnection() const;
   bool canStartSlotConnection(int index) const;
 };

test/test_mqtt_disconnect.py

@@ -27,6 +27,7 @@ HARNESS = r'''
 #include <functional>
 #include <helpers/WiFiNetworkPowerPolicy.h>
 #include <helpers/MQTTConnectionPolicy.h>
+#include <helpers/MQTTTransportHealth.h>
 #define private public
 #include "PsychicMqttClient.h"
 #undef private
@@ -100,6 +101,7 @@ struct MQTTBridge {
     PsychicMqttClient* client = nullptr;
     bool enabled = true, connected = false, circuit_breaker_tripped = false;
     uint32_t connected_at_ms = 0, last_reconnect_attempt = 0;
+    uint32_t current_outage_started_ms = 0;
     uint8_t reconnect_backoff = 0;
     bool initial_connect_done = true;
     char* auth_token = nullptr;
@@ -107,11 +109,14 @@ struct MQTTBridge {
     uint32_t token_expires_at = 1, last_token_renewal = 1;
     uint32_t last_log_time = 1, last_deferred_log_ms = 1;
     uint8_t max_backoff_failures = 0;
+    MQTTTransportHealth::State transport_health;
   } _slots[RUNTIME_MQTT_SLOTS];
   using MQTTSlot = Slot;
   mesh::wifi::NetworkPowerPolicy _network_power_policy;
   std::atomic<bool> _network_suspended{false}, _fresh_ntp_this_boot{false};
   bool _slot_attempt_pending[RUNTIME_MQTT_SLOTS] = {};
+  std::atomic<bool> _slot_transport_fault_pending[RUNTIME_MQTT_SLOTS] = {};
+  std::atomic<bool> _slot_health_reset_pending[RUNTIME_MQTT_SLOTS] = {};
   bool _status_publish_pending[RUNTIME_MQTT_SLOTS] = {};
   bool _slot_force_jwt_mint[RUNTIME_MQTT_SLOTS] = {};
   uint32_t _network_publish_acks[RUNTIME_MQTT_SLOTS] = {};
@@ -127,6 +132,7 @@ struct MQTTBridge {
   Board* _board = nullptr;
   void cancelNtpRefresh() {}
   void updateCachedConnectionStatus() {}
+  void updateSlotTransportDiag(int, uint32_t) {}
   void initializeWiFiInTask() {}
   void processPacketQueue() { ++queues; }
   void teardownSlot(int index, bool force = false);

test/test_mqtt_handshake_scheduler.py

@@ -21,11 +21,14 @@ PREAMBLE = r'''
 #include <cstdio>
 #include <cstring>
 #include <functional>
+#include <new>
+#include <sys/time.h>
 #include <vector>
 #include "helpers/AlertFaultPolicy.h"
 #include "helpers/MQTTConnectionAdmission.h"
 #include "helpers/MQTTConnectionHealth.h"
 #include "helpers/MQTTConnectionPolicy.h"
+#include "helpers/MQTTTransportHealth.h"
 #include "helpers/WiFiPowerSave.h"
 #define ESP_PLATFORM 1
 #define MQTT_DEBUG_PRINTLN(...) ((void)0)
@@ -43,6 +46,11 @@ bool enforceStationChannel() { return true; }
 bool mqttStationWiFiMutationAllowed() { return true; }
 void esp_wifi_set_ps(wifi_ps_type_t) {}
 constexpr int MQTT_AUTH_JWT = 1;
+constexpr int MQTT_ERROR_TYPE_CONNECTION_REFUSED = 5;
+struct esp_mqtt_error_codes {
+  int esp_tls_last_esp_err = 0, esp_tls_stack_err = 0;
+  int esp_transport_sock_errno = 0, error_type = 0, connect_return_code = 0;
+};
 constexpr uint32_t MALLOC_CAP_INTERNAL = 1, MALLOC_CAP_DMA = 2;
 static uint32_t clock_ms = 1000;
 static unsigned long wall_seconds = 1800000000;
@@ -59,7 +67,8 @@ struct {
 } WiFi;
 struct { size_t size = 2097152; size_t getPsramSize() const { return size; } } ESP;
 size_t heap_caps_get_free_size(uint32_t caps) {
-  assert(caps == (MALLOC_CAP_INTERNAL | MALLOC_CAP_DMA)); return dma_free;
+  assert(caps == MALLOC_CAP_INTERNAL || caps == (MALLOC_CAP_INTERNAL | MALLOC_CAP_DMA));
+  return dma_free;
 }
 size_t heap_caps_get_largest_free_block(uint32_t caps) {
   assert(caps == (MALLOC_CAP_INTERNAL | MALLOC_CAP_DMA)); return dma_largest;
@@ -70,17 +79,61 @@ struct Preset {
   int auth_type = 0;
   unsigned long token_lifetime = 3300;
   bool enforce_exp = false;
+  bool allow_retain = false;
 };
 bool mqttPresetEnforcesTokenExp(const Preset* preset) {
   return preset && preset->enforce_exp;
 }
+namespace MQTTMessageBuilder {
+void formatIsoTimestampForMqtt(long, long, const char*, char* buffer, size_t size) {
+  snprintf(buffer, size, "2026-10-14T00:00:00+00:00");
+}
+template<typename... Args>
+int buildStatusMessage(int, const char*, const char*, const char*, const char*,
+                       const char*, const char*, const char*, const char*,
+                       char* buffer, int size, Args...) {
+  assert(size > 2);
+  strcpy(buffer, "{}");
+  return 2;
+}
+}
+void formatRadioInfo(float, float, int, int, char* buffer, size_t size) {
+  snprintf(buffer, size, "909.5,500,5,5");
+}
 struct PsychicMqttClient {
   bool initialized = false, started = false, live = false, model_task_heap = false;
+  bool stop_succeeds = true;
   int starts = 0, reconnects = 0, bounces = 0, credentials = 0, stops = 0;
+  uint32_t acknowledgments = 0;
+  int publish_result = 0, publishes = 0;
+  bool immediate_ack = false;
+  std::function<void(bool)> connected_callback, disconnected_callback;
+  std::function<void(esp_mqtt_error_codes)> error_callback;
   std::function<void()> disconnected;
   bool isInitialized() const { return initialized; }
   bool isStarted() const { return started; }
   bool connected() const { return live; }
+  uint32_t getPublishAck() const { return acknowledgments; }
+  void setAutoReconnect(bool enabled) { assert(!enabled); }
+  void onConnect(std::function<void(bool)> callback) { connected_callback = callback; }
+  void onDisconnect(std::function<void(bool)> callback) { disconnected_callback = callback; }
+  void onError(std::function<void(esp_mqtt_error_codes)> callback) { error_callback = callback; }
+  void errorWithoutDisconnect() {
+    live = false; // The real adapter does this before calling onError.
+    assert(error_callback);
+    error_callback(esp_mqtt_error_codes{});
+  }
+  void connectEvent() {
+    live = true;
+    assert(connected_callback);
+    connected_callback(false);
+  }
+  int publish(const char*, int qos, bool, const char*, int, bool async) {
+    assert(async == (qos > 0));
+    ++publishes;
+    if (immediate_ack && qos > 0 && publish_result >= 0) ++acknowledgments;
+    return publish_result;
+  }
   int connect() {
     ++starts;
     if (model_task_heap) {
@@ -93,9 +146,11 @@ struct PsychicMqttClient {
   int reconnect() { ++reconnects; return ESP_OK; }
   void disconnect() {
     ++stops;
+    if (!stop_succeeds) return;
     if (model_task_heap && started) dma_free += 6536;
     started = live = false;
     if (disconnected) disconnected();
+    if (disconnected_callback) disconnected_callback(false);
   }
   void softDisconnect() {
     ++bounces; live = false;
@@ -117,12 +172,26 @@ struct MQTTBridge {
     uint8_t reconnect_backoff = 0, max_backoff_failures = 0;
     bool circuit_breaker_tripped = false;
     uint32_t start_failures = 0;
+    uint32_t connect_failures = 0, disconnect_count = 0;
+    unsigned long first_disconnect_time = 0, current_outage_started_ms = 0;
+    unsigned long last_log_time = 0;
+    int32_t last_tls_err = 0, last_tls_stack_err = 0;
+    int last_sock_errno = 0;
+    unsigned long last_error_time = 0;
+    MQTTTransportHealth::State transport_health;
   } _slots[RUNTIME_MQTT_SLOTS];
   PsychicMqttClient clients[RUNTIME_MQTT_SLOTS];
   char tokens[RUNTIME_MQTT_SLOTS][16] = {};
   bool setup_succeeds[RUNTIME_MQTT_SLOTS] = {true, true, true, true, true, true};
   bool ready[RUNTIME_MQTT_SLOTS] = {true, true, true, true, true, true};
   std::atomic<bool> _slot_attempt_pending[RUNTIME_MQTT_SLOTS] = {};
+  std::atomic<bool> _slot_transport_fault_pending[RUNTIME_MQTT_SLOTS] = {};
+  std::atomic<bool> _slot_health_reset_pending[RUNTIME_MQTT_SLOTS] = {};
+  std::atomic<uint32_t> _slot_health_recoveries[RUNTIME_MQTT_SLOTS] = {};
+  std::atomic<uint8_t> _slot_health_reason[RUNTIME_MQTT_SLOTS] = {};
+  std::atomic<bool> _slot_health_recovering[RUNTIME_MQTT_SLOTS] = {};
+  std::atomic<uint32_t> _slot_health_ack_wait_secs[RUNTIME_MQTT_SLOTS] = {};
+  bool _status_publish_pending[RUNTIME_MQTT_SLOTS] = {};
   std::atomic<bool> _stop_requested{false}, _ntp_synced{true};
   bool _slot_force_jwt_mint[RUNTIME_MQTT_SLOTS] = {};
   void* _identity = this;
@@ -131,7 +200,11 @@ struct MQTTBridge {
   wl_status_t _last_wifi_status = WL_CONNECTED;
   unsigned long _last_wifi_check = 0, _last_wifi_reconnect_attempt = 0;
   uint8_t _wifi_reconnect_backoff_attempt = 0, _wifi_power_save = 1;
-  struct { bool canonical_wifi = false; } _node_info;
+  struct {
+    bool canonical_wifi = false;
+    float *freq = nullptr, *bw = nullptr;
+    int *sf = nullptr, *cr = nullptr;
+  } _node_info;
   struct Observer { uint8_t wifi_power_save = 1; } observer;
   Observer* _obs = &observer;
   AlertFaultPolicy::OutageSnapshot _wifi_outage{};
@@ -142,6 +215,39 @@ struct MQTTBridge {
   void beginWiFiStation() {}
   const char* _jwt_username = "jwt";
   static const unsigned long SLOT_SETUP_RETRY_INTERVAL = 60000;
+  static const unsigned long SLOT_LOG_INTERVAL = 60000;
+  static constexpr int STATUS_JSON_BUFFER_SIZE = 1024, MSG_STATUS = 0;
+  const char *_timezone = "UTC", *_origin = "node", *_device_id = "id";
+  const char *_board_model = "V4", *_firmware_version = "test";
+  int _json_scratch_doc = 0, _queue_count = 0;
+  char _json_scratch_buffer[STATUS_JSON_BUFFER_SIZE] = {};
+  struct Board { int getBattMilliVolts() const { return 4000; } };
+  struct Clock { uint32_t getMillis() const { return clock_ms; } };
+  struct Dispatcher {
+    int getErrFlags() const { return 0; }
+    int getTotalAirTime() const { return 0; }
+    int getReceiveAirTime() const { return 0; }
+    int getNumSentFlood() const { return 0; }
+    int getNumSentDirect() const { return 0; }
+    int getNumRecvFlood() const { return 0; }
+    int getNumRecvDirect() const { return 0; }
+  };
+  struct Radio {
+    int getNoiseFloor() const { return -120; }
+    int getPacketsRecvErrors() const { return 0; }
+  };
+  Board* _board = nullptr;
+  Clock* _ms = nullptr;
+  Dispatcher* _dispatcher = nullptr;
+  Radio* _radio = nullptr;
+  void refreshOriginFromPrefs() {}
+  void getClientVersion(char* output, size_t size) { snprintf(output, size, "test"); }
+  int repeatStatus() const { return 0; }
+  int collectConnHealth() const { return 0; }
+  bool buildTopicForSlot(int, int, char* output, size_t size) {
+    snprintf(output, size, "status");
+    return true;
+  }
   std::vector<int> setups;
   bool isSlotReady(int index) { return ready[index]; }
   bool setupSlot(int index) {
@@ -162,7 +268,14 @@ struct MQTTBridge {
   void complete(int index) {
     _slot_attempt_pending[index].store(false, std::memory_order_release);
     _slots[index].connected = clients[index].live = true;
+    if (_slots[index].client) _slots[index].client->live = true;
+    _slots[index].connected_at_ms = clock_ms;
+    _slot_health_reset_pending[index] = true;
   }
+  void updateCachedConnectionStatus() {}
+  bool ensureSlotClient(int index);
+  bool publishToSlot(int, const char*, const char*, size_t, bool, uint8_t);
+  void publishStatusToSlot(int);
   void only(int index) {
     for (int i = 0; i < RUNTIME_MQTT_SLOTS; ++i) _slots[i].enabled = i == index;
   }
@@ -173,6 +286,9 @@ struct MQTTBridge {
   esp_err_t reconnectSlotClient(int index);
   unsigned long slotTokenLifetime(int index) const;
   void maintainSlotConnections();
+  bool recoverSlotTransport(int index, uint32_t now);
+  void serviceSlotTransportHealth(uint32_t now);
+  void updateSlotTransportDiag(int index, uint32_t now);
   bool handleWiFiConnection(unsigned long);
   void maintainSlotConnection(int, unsigned long, unsigned long, bool, bool&, bool&);
 };
@@ -188,25 +304,36 @@ class MqttHandshakeSchedulerTest(unittest.TestCase):
             "bool MQTTBridge::hasPendingSlotConnection() const",
             "bool MQTTBridge::canStartSlotConnection(int index) const",
             "esp_err_t MQTTBridge::reconnectSlotClient(int index)",
+            "bool MQTTBridge::recoverSlotTransport(",
+            "void MQTTBridge::serviceSlotTransportHealth(",
+            "void MQTTBridge::updateSlotTransportDiag(",
+            "bool MQTTBridge::ensureSlotClient(",
+            "bool MQTTBridge::publishToSlot(",
+            "void MQTTBridge::publishStatusToSlot(",
             "unsigned long MQTTBridge::slotTokenLifetime(int index) const",
             "bool MQTTBridge::handleWiFiConnection(unsigned long now)",
             "void MQTTBridge::maintainSlotConnections()",
             "void MQTTBridge::maintainSlotConnection("))
 
-    def compile_run(self, main):
+    def compile_run(self, main, fixture=None, expect_rejection=False):
         compiler = shutil.which("g++") or shutil.which("clang++")
         self.assertIsNotNone(compiler, "a host C++ compiler is required")
         with tempfile.TemporaryDirectory(prefix="meshcore-mqtt-scheduler-") as temporary:
             work = Path(temporary)
             source, binary = work / "scheduler.cpp", work / "scheduler"
-            source.write_text(self.fixture() + main, encoding="utf-8")
+            source.write_text((self.fixture() if fixture is None else fixture) + main,
+                              encoding="utf-8")
             result = subprocess.run(
                 [compiler, "-std=c++17", "-Wall", "-Wextra", "-I", str(ROOT / "src"),
                  str(source), "-o", str(binary)], capture_output=True, text=True, timeout=60)
             self.assertEqual(result.returncode, 0, result.stderr)
             result = subprocess.run([str(binary)], cwd=work, capture_output=True,
                                     text=True, timeout=10)
-            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+            if expect_rejection:
+                self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
+                self.assertIn("Assertion", result.stderr)
+            else:
+                self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
 
     def test_first_start_is_immediate_and_delayed_callback_serializes_following_slots(self):
         self.compile_run(r'''
@@ -220,7 +347,7 @@ int main() {
   // still pending. Ordinary maintenance must not launch another SDK request.
   clock_ms = 16000;
   bridge.maintainSlotConnections();
-  clock_ms = 3600000;
+  clock_ms = 60000;
   bridge.maintainSlotConnections();
   assert(bridge.setups == std::vector<int>{0});
   assert(bridge.clients[0].starts == 1 && bridge.clients[0].reconnects == 0);
@@ -419,7 +546,7 @@ int main() {
   bridge.maintainSlotConnections();
   assert(bridge.clients[0].starts == 1 && dma_free == 35840);
   assert(bridge.hasPendingSlotConnection());
-  clock_ms += 3600000;
+  clock_ms += 60000;
   bridge.maintainSlotConnections();
   assert(bridge.clients[1].starts == 0 && bridge.clients[2].starts == 0);
   bridge.complete(0);
@@ -450,6 +577,371 @@ int main() {
 }
 ''')
 
+    def test_missing_handshake_callback_times_out_only_after_sdk_stop(self):
+        self.compile_run(r'''
+int main() {
+  MQTTBridge bridge;
+  bridge.only(0);
+  clock_ms = 1000;
+  bridge.maintainSlotConnections();
+  assert(bridge.clients[0].starts == 1 && bridge._slot_attempt_pending[0]);
+  clock_ms = 120999;
+  bridge.maintainSlotConnections();
+  assert(bridge.clients[0].stops == 0 && bridge._slot_attempt_pending[0]);
+  clock_ms = 121000;
+  bridge.maintainSlotConnections();
+  auto& slot = bridge._slots[0];
+  assert(bridge.clients[0].stops == 1 && !bridge.clients[0].isStarted());
+  assert(!bridge._slot_attempt_pending[0] && !bridge.hasPendingSlotConnection());
+  assert(slot.transport_health.recoveries == 1 && slot.connect_failures == 1);
+  assert(slot.transport_health.last_recovery_reason ==
+      MQTTTransportHealth::RecoveryReason::HandshakeTimeout);
+  assert(slot.last_reconnect_attempt == clock_ms);
+  // Cleanup doesn't start another SDK task in the same maintenance pass.
+  assert(bridge.clients[0].starts == 1);
+  clock_ms += 14999;
+  bridge.maintainSlotConnections();
+  assert(bridge.clients[0].starts == 1);
+  ++clock_ms;
+  bridge.maintainSlotConnections();
+  assert(bridge.clients[0].starts == 2 && bridge._slot_attempt_pending[0]);
+  assert(slot.reconnect_backoff == 1);
+}
+''')
+
+    def test_error_callback_without_disconnect_defers_cleanup_and_preserves_reservation(self):
+        self.compile_run(r'''
+int main() {
+  for (bool established : {false, true}) {
+    MQTTBridge bridge;
+    bridge.only(0);
+    clock_ms = 1000;
+    assert(bridge.ensureSlotClient(0));
+    auto& slot = bridge._slots[0];
+    slot.initial_connect_done = true;
+    auto* client = slot.client;
+    assert(bridge.reconnectSlotClient(0) == ESP_OK);
+    if (established) client->connectEvent();
+    client->errorWithoutDisconnect();
+    // The actual production callback latches work. It cannot stop its own
+    // esp-mqtt task or release a still-running handshake from that callback.
+    assert(client->stops == 0 && client->isStarted());
+    assert(bridge._slot_transport_fault_pending[0]);
+    assert(bridge.hasPendingSlotConnection() && !bridge.canStartSlotConnection(1));
+    client->stop_succeeds = false;
+    bridge.maintainSlotConnections();
+    assert(client->stops == 1 && client->isStarted());
+    assert(bridge._slot_transport_fault_pending[0]);
+    assert(bridge._slot_attempt_pending[0] == !established);
+    assert(bridge.hasPendingSlotConnection() && !bridge.canStartSlotConnection(1));
+    assert(slot.transport_health.recoveries == 0 && client->starts == 1);
+    // A later successful stop proves ownership ended, then recovery may admit
+    // another handshake. Failure accounting remains exactly once.
+    client->stop_succeeds = true;
+    bridge.maintainSlotConnections();
+    assert(client->stops == 2 && !client->isStarted());
+    assert(!bridge.hasPendingSlotConnection() && !bridge._slot_transport_fault_pending[0]);
+    assert(slot.transport_health.recoveries == 1);
+    assert(slot.connect_failures == (established ? 0u : 1u));
+    assert(slot.transport_health.last_recovery_reason ==
+        MQTTTransportHealth::RecoveryReason::TransportError);
+    assert(client->starts == 1); // completed cleanup starts the 15-second gap
+    delete client;
+    slot.client = nullptr;
+  }
+}
+''')
+
+    def test_failed_stop_after_lost_callback_blocks_every_new_handshake(self):
+        self.compile_run(r'''
+int main() {
+  MQTTBridge bridge;
+  clock_ms = 1000;
+  bridge.maintainSlotConnections();
+  bridge.clients[0].stop_succeeds = false;
+  clock_ms += MQTTTransportHealth::kAttemptTimeoutMs;
+  for (int pass = 0; pass < 3; ++pass) {
+    bridge.maintainSlotConnections();
+    assert(bridge.clients[0].stops == pass + 1 && bridge.clients[0].isStarted());
+    assert(bridge._slot_attempt_pending[0] && bridge.hasPendingSlotConnection());
+    assert(bridge._slots[0].transport_health.recoveries == 0);
+    assert(bridge.setups == std::vector<int>{0});
+    for (int i = 0; i < RUNTIME_MQTT_SLOTS; ++i)
+      assert(!bridge.canStartSlotConnection(i));
+    clock_ms += 15000;
+  }
+  bridge.clients[0].stop_succeeds = true;
+  bridge.maintainSlotConnections();
+  assert(!bridge.clients[0].isStarted() && !bridge.hasPendingSlotConnection());
+  assert(bridge._slots[0].transport_health.recoveries == 1);
+  assert(bridge.setups == std::vector<int>{0});
+}
+''')
+
+    def test_three_brokers_run_beyond_three_hours_and_only_missing_ack_recovers(self):
+        self.compile_run(r'''
+int main() {
+  MQTTBridge bridge;
+  clock_ms = 1000;
+  for (int i = 0; i < RUNTIME_MQTT_SLOTS; ++i) {
+    bridge._slots[i].enabled = i < 3;
+    if (i >= 3) continue;
+    auto& slot = bridge._slots[i];
+    slot.client = &bridge.clients[i];
+    slot.initial_connect_done = true;
+    slot.client->initialized = slot.client->started = true;
+    bridge.complete(i);
+    slot.client->immediate_ack = true;
+  }
+  // Two status publishers get real ACK-counter progress; the third is quiet
+  // (for example, its broker does not accept status topics). Neither condition
+  // is a transport stall, irrespective of connection age or packet activity.
+  for (int minute = 0; minute <= 180; ++minute) {
+    assert(bridge.publishToSlot(0, "status", "{}", 2, true, 1));
+    assert(bridge.publishToSlot(1, "status", "{}", 2, true, 1));
+    bridge.maintainSlotConnections();
+    for (int i = 0; i < 3; ++i) assert(bridge.clients[i].stops == 0);
+    clock_ms += 60000;
+    wall_seconds += 60;
+  }
+  assert(clock_ms > 3 * 60 * 60 * 1000);
+  bridge.clients[0].immediate_ack = false;
+  const uint32_t stalled_at = clock_ms;
+  assert(bridge.publishToSlot(0, "status", "{}", 2, true, 1));
+  for (uint32_t elapsed = 0; elapsed < 120000; elapsed += 30000) {
+    clock_ms = stalled_at + elapsed;
+    assert(bridge.publishToSlot(0, "status", "{}", 2, true, 1));
+    assert(bridge.publishToSlot(1, "status", "{}", 2, true, 1));
+    bridge.maintainSlotConnections();
+    assert(bridge.clients[0].stops == 0);
+    assert(bridge._slots[0].transport_health.ack_started_ms == stalled_at);
+  }
+  clock_ms = stalled_at + 120000;
+  bridge.maintainSlotConnections();
+  assert(bridge.clients[0].stops == 1 && !bridge._slots[0].connected);
+  assert(bridge._slots[0].transport_health.recoveries == 1);
+  assert(bridge._slots[0].transport_health.last_recovery_reason ==
+      MQTTTransportHealth::RecoveryReason::AckTimeout);
+  for (int i = 1; i < 3; ++i) {
+    assert(bridge.clients[i].stops == 0 && bridge._slots[i].connected);
+    assert(bridge._slots[i].transport_health.recoveries == 0);
+  }
+  assert(bridge._slot_health_recoveries[0] == 1);
+  assert(!bridge._slot_health_recovering[0]);
+}
+''')
+
+    def test_on_connect_status_watch_survives_maintenance_and_ack_clears_it(self):
+        self.compile_run(r'''
+int main() {
+  MQTTBridge bridge;
+  bridge.only(0); clock_ms = 1000;
+  assert(bridge.ensureSlotClient(0));
+  auto& slot = bridge._slots[0];
+  slot.initial_connect_done = true;
+  assert(bridge.reconnectSlotClient(0) == ESP_OK);
+  slot.client->connectEvent();
+  assert(bridge._slot_health_reset_pending[0] && bridge._status_publish_pending[0]);
+  // The real worker consumes an on-connect status before it calls maintenance.
+  bridge.publishStatusToSlot(0);
+  assert(slot.client->publishes == 1);
+  assert(!bridge._slot_health_reset_pending[0] && slot.transport_health.ack_pending);
+  bridge.maintainSlotConnections();
+  assert(slot.transport_health.ack_pending);
+  clock_ms += 119999;
+  ++slot.client->acknowledgments;
+  bridge.maintainSlotConnections();
+  assert(!slot.transport_health.ack_pending && slot.client->stops == 0);
+  assert(bridge._slot_health_ack_wait_secs[0] == UINT32_MAX);
+  slot.client->immediate_ack = true;
+  assert(bridge.publishToSlot(0, "status", "{}", 2, true, 1));
+  assert(!slot.transport_health.ack_pending); // ACK arrived inside enqueue
+  clock_ms += 3600000;
+  bridge.maintainSlotConnections();
+  assert(slot.client->stops == 0);
+  delete slot.client;
+  slot.client = nullptr;
+}
+''')
+
+    def test_negative_qos0_write_without_disconnect_requests_owner_recovery(self):
+        self.compile_run(r'''
+int main() {
+  MQTTBridge bridge;
+  bridge.only(0); clock_ms = 1000;
+  auto& slot = bridge._slots[0];
+  slot.client = &bridge.clients[0];
+  slot.client->initialized = slot.client->started = true;
+  slot.initial_connect_done = true;
+  bridge.complete(0);
+  slot.client->publish_result = -1;
+  assert(!bridge.publishToSlot(0, "packets", "{}", 2, false, 0));
+  assert(slot.client->stops == 0 && slot.connected && slot.client->connected());
+  assert(slot.transport_health.recovery_reason ==
+      MQTTTransportHealth::RecoveryReason::PublishFailure);
+  assert(bridge.hasPendingSlotConnection());
+  assert(!bridge.publishToSlot(0, "packets", "{}", 2, false, 0));
+  assert(slot.client->publishes == 1); // don't repeatedly hit the broken socket
+  bridge.maintainSlotConnections();
+  assert(slot.client->stops == 1 && !slot.connected && !slot.client->isStarted());
+  assert(slot.transport_health.recoveries == 1);
+  assert(slot.transport_health.last_recovery_reason ==
+      MQTTTransportHealth::RecoveryReason::PublishFailure);
+}
+''')
+
+    def test_handshake_and_ack_recovery_are_millis_wrap_safe(self):
+        self.compile_run(r'''
+int main() {
+  for (bool awaiting_ack : {false, true}) {
+    MQTTBridge bridge;
+    bridge.only(0); clock_ms = UINT32_MAX - 60000;
+    bridge.maintainSlotConnections();
+    auto& slot = bridge._slots[0];
+    if (awaiting_ack) {
+      bridge.complete(0);
+      assert(bridge.publishToSlot(0, "status", "{}", 2, true, 1));
+    }
+    clock_ms += 119999;
+    bridge.maintainSlotConnections();
+    assert(slot.client->stops == 0);
+    ++clock_ms;
+    bridge.maintainSlotConnections();
+    assert(slot.client->stops == 1 && slot.transport_health.recoveries == 1);
+    assert(slot.transport_health.last_recovery_reason == (awaiting_ack
+        ? MQTTTransportHealth::RecoveryReason::AckTimeout
+        : MQTTTransportHealth::RecoveryReason::HandshakeTimeout));
+  }
+}
+''')
+
+    def test_recovery_preserves_backoff_and_restart_memory_admission(self):
+        self.compile_run(r'''
+int main() {
+  MQTTBridge bridge;
+  bridge.only(0); clock_ms = 70000;
+  auto& slot = bridge._slots[0];
+  slot.client = &bridge.clients[0];
+  slot.client->initialized = slot.client->started = true;
+  slot.initial_connect_done = true;
+  bridge.complete(0);
+  slot.reconnect_backoff = 3;
+  slot.max_backoff_failures = 1;
+  bridge._slot_transport_fault_pending[0] = true;
+  bridge.maintainSlotConnections();
+  assert(slot.client->stops == 1 && slot.reconnect_backoff == 3);
+  assert(slot.max_backoff_failures == 1 && slot.connect_failures == 0);
+  clock_ms += 120000;
+  dma_free = 24575; dma_largest = 20000;
+  bridge.maintainSlotConnections();
+  assert(slot.client->starts == 0 && slot.reconnect_backoff == 3);
+  assert(slot.last_reconnect_attempt == 70000 && slot.start_failures == 0);
+  dma_free = 24576; dma_largest = 8191;
+  bridge.maintainSlotConnections();
+  assert(slot.client->starts == 0 && slot.reconnect_backoff == 3);
+  dma_largest = 8192;
+  bridge.maintainSlotConnections();
+  assert(slot.client->starts == 1 && bridge._slot_attempt_pending[0]);
+  assert(slot.reconnect_backoff == 4 && slot.max_backoff_failures == 1);
+}
+''')
+
+    def test_late_connected_callback_cannot_release_failed_stop_reservation(self):
+        self.compile_run(r'''
+int main() {
+  for (bool publish_before_maintenance : {false, true}) {
+    MQTTBridge bridge;
+    bridge.only(0); clock_ms = 1000;
+    assert(bridge.ensureSlotClient(0));
+    auto& slot = bridge._slots[0];
+    slot.initial_connect_done = true;
+    auto* client = slot.client;
+    assert(bridge.reconnectSlotClient(0) == ESP_OK);
+    client->connectEvent();
+    client->errorWithoutDisconnect();
+    client->stop_succeeds = false;
+    bridge.maintainSlotConnections();
+    assert(client->isStarted() && client->stops == 1);
+    assert(slot.transport_health.recovery_reason != MQTTTransportHealth::RecoveryReason::None);
+    // An in-flight SDK event arrives after stop failed. CONNACK may clear the
+    // attempt bit and set a new-session latch, but it cannot prove task join.
+    client->connectEvent();
+    assert(!bridge._slot_attempt_pending[0] && bridge._slot_health_reset_pending[0]);
+    assert(bridge.hasPendingSlotConnection() && !bridge.canStartSlotConnection(1));
+    if (publish_before_maintenance) {
+      assert(!bridge.publishToSlot(0, "status", "{}", 2, true, 1));
+      assert(client->publishes == 0 && bridge.hasPendingSlotConnection());
+    }
+    bridge.maintainSlotConnections();
+    assert(client->stops == 2 && client->isStarted());
+    assert(bridge.hasPendingSlotConnection() && !bridge.canStartSlotConnection(1));
+    assert(client->starts == 1 && bridge.setups.empty());
+    assert(slot.transport_health.recoveries == 0);
+    client->stop_succeeds = true;
+    bridge.maintainSlotConnections();
+    assert(!client->isStarted() && !bridge.hasPendingSlotConnection());
+    assert(slot.transport_health.recoveries == 1 && client->starts == 1);
+    delete client;
+    slot.client = nullptr;
+  }
+}
+''')
+
+    def test_wifi_loss_failed_stop_and_reconnect_cannot_release_sdk_reservation(self):
+        self.compile_run(r'''
+int main() {
+  for (bool established : {false, true}) {
+    MQTTBridge bridge;
+    bridge.only(0); clock_ms = 70000; WiFi.state = WL_CONNECTED;
+    assert(bridge.ensureSlotClient(0));
+    auto& slot = bridge._slots[0];
+    slot.initial_connect_done = true;
+    auto* client = slot.client;
+    assert(bridge.reconnectSlotClient(0) == ESP_OK);
+    if (established) client->connectEvent();
+    client->stop_succeeds = false;
+    WiFi.state = WL_DISCONNECTED;
+    assert(!bridge.handleWiFiConnection(clock_ms));
+    assert(client->isStarted() && client->stops == 1 && !slot.connected);
+    assert(bridge._slot_transport_fault_pending[0] && bridge.hasPendingSlotConnection());
+    assert(!bridge.canStartSlotConnection(1));
+    clock_ms += 11000;
+    WiFi.state = WL_CONNECTED;
+    assert(bridge.handleWiFiConnection(clock_ms));
+    client->connectEvent(); // late SDK callback after WiFi recovered
+    assert(bridge.hasPendingSlotConnection());
+    bridge.maintainSlotConnections();
+    assert(client->isStarted() && client->stops == 2 && !slot.connected);
+    assert(bridge.hasPendingSlotConnection() && !bridge.canStartSlotConnection(1));
+    assert(client->starts == 1 && bridge.setups.empty());
+    client->stop_succeeds = true;
+    bridge.maintainSlotConnections();
+    assert(!client->isStarted() && !bridge.hasPendingSlotConnection());
+    assert(slot.transport_health.recoveries == 1 && client->starts == 1);
+    delete client;
+    slot.client = nullptr;
+  }
+}
+''')
+
+    def test_pre_fix_error_callback_without_owner_signal_fails_regression(self):
+        source = self.fixture()
+        old_callback = source.replace(
+            "_slot_transport_fault_pending[index].store(true, std::memory_order_release);",
+            "/* pre-fix onError had no owner-task recovery signal */", 1)
+        self.assertNotEqual(source, old_callback)
+        self.compile_run(r'''
+int main() {
+  MQTTBridge bridge;
+  bridge.only(0); clock_ms = 1000;
+  assert(bridge.ensureSlotClient(0));
+  auto* client = bridge._slots[0].client;
+  assert(bridge.reconnectSlotClient(0) == ESP_OK);
+  client->errorWithoutDisconnect();
+  assert(bridge._slot_transport_fault_pending[0]);
+}
+''', fixture=old_callback, expect_rejection=True)
+
 
 if __name__ == "__main__":
     unittest.main()

test/test_mqtt_network_timeout.py

@@ -21,6 +21,7 @@ PREAMBLE = r'''
 #include <cstddef>
 #include "helpers/MQTTConnectionAdmission.h"
 #include "helpers/MQTTConnectionPolicy.h"
+#include "helpers/MQTTTransportHealth.h"
 struct esp_mqtt_client_config_t {
 #if ESP_IDF_VERSION_MAJOR == 5
   struct { int keepalive = 0, message_retransmit_timeout = 0; } session;
@@ -98,6 +99,7 @@ int main() {
 RECONFIGURE_PREAMBLE = r'''
 #include "helpers/MQTTConnectionAdmission.h"
 #include "helpers/MQTTConnectionPolicy.h"
+#include "helpers/MQTTTransportHealth.h"
 #define MQTT_DEBUG_PRINTLN(...) ((void)0)
 #define pdMS_TO_TICKS(value) (value)
 constexpr int RUNTIME_MQTT_SLOTS = 2, MQTT_AUTH_JWT = 1;
@@ -147,8 +149,11 @@ struct MQTTBridge {
     unsigned long last_reconnect_attempt = 0, last_log_time = 0, last_deferred_log_ms = 0;
     uint8_t reconnect_backoff = 0, max_backoff_failures = 0;
     bool circuit_breaker_tripped = false;
+    MQTTTransportHealth::State transport_health;
   } _slots[RUNTIME_MQTT_SLOTS];
   std::atomic<bool> _stop_requested{false}, _slot_attempt_pending[RUNTIME_MQTT_SLOTS] = {};
+  std::atomic<bool> _slot_transport_fault_pending[RUNTIME_MQTT_SLOTS] = {};
+  std::atomic<bool> _slot_health_reset_pending[RUNTIME_MQTT_SLOTS] = {};
   bool _slot_force_jwt_mint[RUNTIME_MQTT_SLOTS] = {};
   bool _slot_reconfigure_pending[RUNTIME_MQTT_SLOTS] = {};
   bool _initialized = true;
@@ -161,6 +166,7 @@ struct MQTTBridge {
   bool canStartSlotConnection(int) const;
   void optimizeMqttClientConfig(PsychicMqttClient*, bool);
   void teardownSlot(int, bool force = false);
+  void updateSlotTransportDiag(int, uint32_t) {}
   void applySlotPreset(int, const char*);
   bool setupSlot(int index) {
     auto& slot = _slots[index];

test/test_mqtt_publish_ack.py

@@ -79,6 +79,7 @@ PREAMBLE = r'''
 #undef private
 #include "helpers/MQTTReplyFormat.h"
 #include "helpers/MQTTConnectionHealth.h"
+#include "helpers/MQTTTransportHealth.h"
 #include "helpers/MQTTPacketFilter.h"
 #include "helpers/bridges/MQTTErrorLabels.h"
 template<typename... Args> void test_log(Args&&...) {}
@@ -161,6 +162,14 @@ struct MQTTBridge {
     int last_sock_errno = 0;
   };
   MQTTSlot _slots[RUNTIME_MQTT_SLOTS];
+  std::atomic<bool> _slot_transport_fault_pending[RUNTIME_MQTT_SLOTS] = {};
+  std::atomic<bool> _slot_health_recovering[RUNTIME_MQTT_SLOTS] = {};
+  std::atomic<uint32_t> _slot_health_recoveries[RUNTIME_MQTT_SLOTS] = {};
+  std::atomic<uint8_t> _slot_health_reason[RUNTIME_MQTT_SLOTS] = {};
+  std::atomic<uint32_t> _slot_health_ack_wait_secs[RUNTIME_MQTT_SLOTS] = {};
+  MQTTBridge() {
+    for (auto& wait : _slot_health_ack_wait_secs) wait.store(UINT32_MAX);
+  }
   bool _initialized = true;
   bool ready = true;
   MQTTPrefs* _obs = nullptr;
@@ -437,6 +446,49 @@ int main() {
 }
 ''')
 
+    def test_production_recovery_diagnostics_never_claim_no_errors(self):
+        self.compile_run(r'''
+int main() {
+  PsychicMqttClient client;
+  MQTTBridge bridge;
+  bridge._slots[0].client = &client;
+  s_mqtt_bridge_instance = &bridge;
+  bridge._slot_health_reason[0] = static_cast<uint8_t>(
+      MQTTTransportHealth::RecoveryReason::AckTimeout);
+  bridge._slot_health_recoveries[0] = 3;
+  bridge._slot_health_ack_wait_secs[0] = 120;
+  for (bool fault_latched : {false, true}) {
+    bridge._slot_health_recovering[0] = !fault_latched;
+    bridge._slot_transport_fault_pending[0] = fault_latched;
+    char reply[160];
+    MQTTBridge::formatSlotDiagReply(reply, sizeof(reply), 0);
+    assert(strcmp(reply, "> mqtt1: recover, pub_ack:0, rec:3/puback, ack_wait:120s") == 0);
+    assert(strstr(reply, "no errors") == nullptr);
+  }
+  bridge._slot_health_recovering[0] = false;
+  bridge._slot_transport_fault_pending[0] = false;
+  char recovered[160];
+  MQTTBridge::formatSlotDiagReply(recovered, sizeof(recovered), 0);
+  assert(strstr(recovered, "> mqtt1: ok, no errors") == recovered);
+  assert(strstr(recovered, "rec:3/puback") != nullptr); // retain history
+
+  bridge._slot_health_recovering[0] = true;
+  bridge._slot_health_recoveries[0] = UINT32_MAX;
+  bridge._slot_health_ack_wait_secs[0] = UINT32_MAX - 1;
+  for (size_t capacity = 1; capacity <= 160; ++capacity) {
+    unsigned char bounded[168];
+    memset(bounded, 0xA5, sizeof(bounded));
+    char* reply = reinterpret_cast<char*>(bounded);
+    MQTTBridge::formatSlotDiagReply(reply, capacity, 0);
+    assert(memchr(reply, '\0', capacity) != nullptr);
+    for (size_t pos = capacity; pos < sizeof(bounded); ++pos) assert(bounded[pos] == 0xA5);
+    // Every appended numeric field is omitted whole when it cannot fit.
+    if (strstr(reply, "rec:")) assert(strstr(reply, "rec:4294967295/puback"));
+    if (strstr(reply, "ack_wait:")) assert(strstr(reply, "ack_wait:4294967294s"));
+  }
+}
+''')
+
 
 if __name__ == "__main__":
     unittest.main()

test/test_mqtt_transport_health.py

@@ -0,0 +1,134 @@
+#!/usr/bin/env python3
+"""Exercise bounded MQTT liveness clocks without an SDK or PlatformIO build."""
+from pathlib import Path
+import shutil
+import subprocess
+import tempfile
+import unittest
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+class MqttTransportHealthTest(unittest.TestCase):
+    def compile_run(self, main):
+        compiler = shutil.which("g++") or shutil.which("clang++")
+        self.assertIsNotNone(compiler, "a host C++ compiler is required")
+        source = ('#include <cassert>\n#include <cstring>\n#include <initializer_list>\n'
+                  '#include "helpers/MQTTTransportHealth.h"\n'
+                  'using namespace MQTTTransportHealth;\n') + main
+        with tempfile.TemporaryDirectory(prefix="meshcore-mqtt-health-") as directory:
+            work = Path(directory)
+            path, binary = work / "health.cpp", work / "health"
+            path.write_text(source, encoding="ascii")
+            result = subprocess.run(
+                [compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror",
+                 "-I", str(ROOT / "src"), str(path), "-o", str(binary)],
+                capture_output=True, text=True, timeout=60)
+            self.assertEqual(result.returncode, 0, result.stderr)
+            result = subprocess.run([str(binary)], capture_output=True, text=True, timeout=10)
+            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+
+    def test_repeated_acceptance_cannot_postpone_oldest_missing_ack(self):
+        self.compile_run(r'''
+int main() {
+  State state;
+  state.acceptedQos1(1000, 7);
+  for (uint32_t now = 11000; now < 121000; now += 10000) {
+    state.acceptedQos1(now, 7);
+    assert(state.ack_started_ms == 1000 && state.ack_snapshot == 7);
+    assert(!state.ackExpired(now));
+  }
+  assert(state.ackExpired(121000));
+}
+''')
+
+    def test_ack_advance_clears_watch_and_new_enqueue_starts_a_new_deadline(self):
+        self.compile_run(r'''
+int main() {
+  State state;
+  state.acceptedQos1(1000, 7);
+  state.observeAck(7);
+  assert(state.ack_pending);
+  state.observeAck(8);
+  assert(!state.ack_pending && !state.ackExpired(121000));
+  state.acceptedQos1(300000, 8);
+  assert(state.ack_started_ms == 300000 && state.ack_snapshot == 8);
+  assert(!state.ackExpired(419999) && state.ackExpired(420000));
+  // An ACK arriving between maintenance passes is recognized on enqueue.
+  state.acceptedQos1(450000, 9);
+  assert(state.ack_started_ms == 450000 && state.ack_snapshot == 9);
+}
+''')
+
+    def test_immediate_ack_race_uses_snapshot_from_before_enqueue(self):
+        self.compile_run(r'''
+int main() {
+  State state;
+  const uint32_t before_enqueue = 11;
+  const uint32_t after_enqueue = 12; // broker event arrived inside SDK enqueue
+  state.acceptedQos1(1000, before_enqueue);
+  state.observeAck(after_enqueue);
+  assert(!state.ack_pending && !state.ackExpired(121000));
+}
+''')
+
+    def test_zero_start_ticks_and_millis_wrap_preserve_bounded_deadlines(self):
+        self.compile_run(r'''
+int main() {
+  for (uint32_t start : {uint32_t(0), UINT32_MAX - 60000}) {
+    State state;
+    state.beginAttempt(start);
+    state.acceptedQos1(start, UINT32_MAX);
+    assert(!state.attemptExpired(start + kAttemptTimeoutMs - 1));
+    assert(state.attemptExpired(start + kAttemptTimeoutMs));
+    assert(!state.ackExpired(start + kAckTimeoutMs - 1));
+    assert(state.ackExpired(start + kAckTimeoutMs));
+    // The acknowledgment counter may wrap too; inequality still proves progress.
+    state.observeAck(0);
+    assert(!state.ack_pending);
+    state.completeAttempt();
+    assert(!state.attemptExpired(start + kAttemptTimeoutMs));
+  }
+}
+''')
+
+    def test_recovery_keeps_first_reason_and_saturates_lifetime_counter(self):
+        self.compile_run(r'''
+int main() {
+  State state;
+  state.beginAttempt(1000);
+  state.acceptedQos1(1000, 3);
+  state.requestRecovery(RecoveryReason::TransportError);
+  state.requestRecovery(RecoveryReason::AckTimeout);
+  assert(state.recovery_reason == RecoveryReason::TransportError);
+  state.recovered();
+  assert(state.recoveries == 1 && !state.attempt_active && !state.ack_pending);
+  assert(state.recovery_reason == RecoveryReason::None);
+  assert(state.last_recovery_reason == RecoveryReason::TransportError);
+  state.recoveries = UINT32_MAX;
+  state.requestRecovery(RecoveryReason::PublishFailure);
+  state.recovered();
+  assert(state.recoveries == UINT32_MAX);
+  assert(state.last_recovery_reason == RecoveryReason::PublishFailure);
+  assert(strcmp(reasonLabel(RecoveryReason::HandshakeTimeout), "handshake") == 0);
+  assert(strcmp(reasonLabel(RecoveryReason::AckTimeout), "puback") == 0);
+}
+''')
+
+    def test_quiet_session_without_accepted_qos1_never_times_out(self):
+        self.compile_run(r'''
+int main() {
+  State state;
+  state.beginAttempt(1000);
+  state.completeAttempt();
+  for (uint32_t now = 0; now <= 7 * 24 * 60 * 60 * 1000; now += 60000) {
+    state.observeAck(0);
+    assert(!state.attemptExpired(now) && !state.ackExpired(now));
+    assert(state.recovery_reason == RecoveryReason::None);
+  }
+}
+''')
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_mqtt_transport_results.py

@@ -32,6 +32,7 @@ PREAMBLE = r'''
 #include <cassert>
 #include <functional>
 #include "helpers/MQTTConnectionHealth.h"
+#include "helpers/MQTTTransportHealth.h"
 using esp_err_t = int;
 using esp_mqtt_client_handle_t = void*;
 constexpr int ESP_OK=0, ESP_FAIL=-1, ESP_ERR_INVALID_STATE=2, ESP_ERR_NO_MEM=3;
@@ -53,6 +54,7 @@ static int register_result=ESP_OK, config_result=ESP_OK;
 static int start_result=ESP_OK, reconnect_result=ESP_OK;
 static int init_calls=0, register_calls=0, destroy_calls=0, config_calls=0;
 static int start_calls=0, reconnect_calls=0;
+unsigned long millis() { return 1000; }
 static std::function<void()> sdk_callback;
 void* test_malloc(size_t size) { return fail_malloc ? nullptr : std::malloc(size); }
 #define malloc test_malloc
@@ -88,9 +90,11 @@ struct MQTTBridge {
   struct MQTTSlot {
     PsychicMqttClient* client=nullptr;
     uint32_t start_failures=0;
+    MQTTTransportHealth::State transport_health;
   };
   MQTTSlot _slots[RUNTIME_MQTT_SLOTS];
   std::atomic<bool> _slot_attempt_pending[RUNTIME_MQTT_SLOTS] = {};
+  std::atomic<bool> _slot_transport_fault_pending[RUNTIME_MQTT_SLOTS] = {};
   std::atomic<bool> _stop_requested{false};
   bool hasPendingSlotConnection() const;
   bool canStartSlotConnection(int index) const;

test/test_mqtt_wifi_network_power.py

@@ -22,6 +22,8 @@ BRIDGE = "src/helpers/bridges/MQTTBridge.cpp"
 PREAMBLE = r'''
 #include <helpers/WiFiNetworkPowerPolicy.h>
 #include <helpers/MQTTConnectionPolicy.h>
+#include <helpers/MQTTConnectionHealth.h>
+#include <helpers/MQTTTransportHealth.h>
 #include <helpers/esp32/SntpOperationCoordinator.h>
 #include <atomic>
 #include <cassert>
@@ -139,17 +141,18 @@ void configTime(int a, int b, const char* primary) {
 }
 
 struct Client {
-  bool started = true, fail_stop = false;
+  bool started = true, fail_stop = false, live = false;
   int stops = 0;
   uint32_t publish_acks = 0;
   uint32_t getPublishAck() const { return publish_acks; }
   bool isStarted() const { return started; }
+  bool connected() const { return live; }
   void disconnect() {
     require(!mesh::wifi::networkPower().backgroundSuspended(),
             "TLS stopped after handing station off");
     ++stops;
     events.push_back("tls-stop");
-    if (!fail_stop) started = false;
+    if (!fail_stop) started = live = false;
     if (on_client_stop) on_client_stop();
   }
 };
@@ -185,9 +188,18 @@ struct MQTTBridge {
     uint32_t connected_at_ms = 0, last_reconnect_attempt = 200U;
     uint8_t reconnect_backoff = 5;
     Client* client = nullptr;
+    uint32_t connect_failures = 0, current_outage_started_ms = 0;
+    MQTTTransportHealth::State transport_health;
   } _slots[RUNTIME_MQTT_SLOTS];
+  using MQTTSlot = Slot;
   Client clients[RUNTIME_MQTT_SLOTS];
   std::atomic<bool> _slot_attempt_pending[RUNTIME_MQTT_SLOTS] = {};
+  std::atomic<bool> _slot_transport_fault_pending[RUNTIME_MQTT_SLOTS] = {};
+  std::atomic<bool> _slot_health_reset_pending[RUNTIME_MQTT_SLOTS] = {};
+  std::atomic<bool> _slot_health_recovering[RUNTIME_MQTT_SLOTS] = {};
+  std::atomic<uint32_t> _slot_health_recoveries[RUNTIME_MQTT_SLOTS] = {};
+  std::atomic<uint8_t> _slot_health_reason[RUNTIME_MQTT_SLOTS] = {};
+  std::atomic<uint32_t> _slot_health_ack_wait_secs[RUNTIME_MQTT_SLOTS] = {};
   std::atomic<bool> _status_publish_pending[RUNTIME_MQTT_SLOTS] = {};
   uint32_t _network_publish_acks[RUNTIME_MQTT_SLOTS] = {};
   bool _network_espnow_restore_pending = false;
@@ -237,6 +249,10 @@ struct MQTTBridge {
   bool isRunning() const { return true; }
   bool isNetworkSuspended() const { return _network_suspended.load(); }
   bool requestNtpTimeEstimate();
+  bool hasPendingSlotConnection() const;
+  void serviceSlotTransportHealth(uint32_t now);
+  bool recoverSlotTransport(int index, uint32_t now);
+  void updateSlotTransportDiag(int index, uint32_t now);
   bool serviceNetworkPower(uint32_t now);
   void serviceEspNowChannelRestore(uint32_t now);
   void serviceBackgroundNtp(uint32_t now);
@@ -336,6 +352,40 @@ static void failedSdkStopWithholdsHandoff() {
           "successful retry leaked a transport or radio");
 }
 
+static void failedSuspendLateConnectCannotAbandonTransportOwnership() {
+  resetEnvironment();
+  MQTTBridge bridge;
+  for (auto& pending : bridge._slot_attempt_pending) pending = false;
+  bridge.serviceNetworkPower(clock_ms);
+  bridge.clients[1].fail_stop = true;
+  clock_ms += grace;
+  require(!bridge.serviceNetworkPower(clock_ms), "failed stop did not hold handoff");
+  require(bridge._slot_transport_fault_pending[1] && bridge.hasPendingSlotConnection(),
+          "failed suspension abandoned a running MQTT transport");
+  // A foreground owner wakes networking and a late CONNACK sets a fresh-session
+  // latch. Neither proves the failed SDK stop completed or frees admission.
+  mesh::wifi::networkPower().setHold(mesh::wifi::NetworkPowerHold::WebConfig, true);
+  bridge._slots[1].connected = bridge.clients[1].live = true;
+  bridge._slot_attempt_pending[1] = false;
+  bridge._slot_health_reset_pending[1] = true;
+  require(bridge.serviceNetworkPower(++clock_ms), "foreground owner did not keep radio");
+  bridge.serviceSlotTransportHealth(clock_ms);
+  require(bridge.clients[1].started && bridge.hasPendingSlotConnection(),
+          "foreground wake or late CONNACK released failed-stop reservation");
+  require(!bridge._slots[1].connected && !bridge._network_suspended,
+          "failed transport was advertised healthy or handed station away");
+  require(bridge._slots[1].transport_health.recoveries == 0,
+          "failed SDK stop was counted as recovered");
+  bridge.clients[1].fail_stop = false;
+  bridge.serviceSlotTransportHealth(++clock_ms);
+  require(!bridge.clients[1].started && !bridge.hasPendingSlotConnection(),
+          "proven successful stop did not release ownership");
+  require(bridge._slots[1].transport_health.recoveries == 1,
+          "successful owner cleanup was not counted");
+  require(WiFi.offs == 0 && WiFi.disconnects == 0,
+          "transport recovery stopped foreground WiFi");
+}
+
 static void recoveryHasOneBoundedWindow() {
   for (uint32_t start : {100U, UINT32_MAX - 10000U}) {
     resetEnvironment();
@@ -686,6 +736,7 @@ int main(int argc, char** argv) {
   try {
     if (name == "handoff") failureStopsBothServicesBeforeWifi();
     else if (name == "failed-stop") failedSdkStopWithholdsHandoff();
+    else if (name == "failed-stop-wake") failedSuspendLateConnectCannotAbandonTransportOwnership();
     else if (name == "recovery") recoveryHasOneBoundedWindow();
     else if (name == "owners") foregroundAndEspNowArePreserved();
     else if (name == "background-estimate") automaticClockEstimateCannotWakeSuspendedNetwork();
@@ -769,6 +820,10 @@ class MqttWiFiNetworkPowerTests(unittest.TestCase):
             "void MQTTBridge::refreshNTP(",
             "void MQTTBridge::pollNtpRefresh(",
             "bool MQTTBridge::requestNtpTimeEstimate(",
+            "bool MQTTBridge::hasPendingSlotConnection(",
+            "void MQTTBridge::serviceSlotTransportHealth(",
+            "bool MQTTBridge::recoverSlotTransport(",
+            "void MQTTBridge::updateSlotTransportDiag(",
         ))
         cls.temporary = tempfile.TemporaryDirectory(prefix="meshcore-mqtt-network-power-")
         cls.addClassCleanup(cls.temporary.cleanup)
@@ -815,6 +870,9 @@ class MqttWiFiNetworkPowerTests(unittest.TestCase):
     def test_failed_sdk_transport_stop_keeps_station_owned_and_awake(self):
         self.run_case("failed-stop")
 
+    def test_failed_suspend_late_connect_and_foreground_wake_keep_sdk_reservation(self):
+        self.run_case("failed-stop-wake")
+
     def test_recovery_restarts_once_and_stops_after_bounded_probe_across_wrap(self):
         self.run_case("recovery")