Merge branch 'integration' of https://github.com/esphome/esphome into integration

CODEOWNERS

@@ -587,6 +587,7 @@ esphome/components/xl9535/* @mreditor97
 esphome/components/xpt2046/touchscreen/* @nielsnl68 @numo68
 esphome/components/xxtea/* @clydebarrow
 esphome/components/zephyr/* @tomaszduda23
+esphome/components/zephyr_mcumgr/ota/* @tomaszduda23
 esphome/components/zhlt01/* @cfeenstra1024
 esphome/components/zigbee/* @tomaszduda23
 esphome/components/zio_ultrasonic/* @kahrendt

esphome/components/api/api.proto

@@ -1604,11 +1604,11 @@ message BluetoothLEAdvertisementResponse {
 }
 
 message BluetoothLERawAdvertisement {
-  uint64 address = 1;
-  sint32 rssi = 2;
+  uint64 address = 1 [(force) = true];
+  sint32 rssi = 2 [(force) = true];
   uint32 address_type = 3;
 
-  bytes data = 4 [(fixed_array_size) = 62];
+  bytes data = 4 [(fixed_array_size) = 62, (force) = true];
 }
 
 message BluetoothLERawAdvertisementsResponse {

esphome/components/api/api_buffer.cpp

@@ -0,0 +1,13 @@
+#include "api_buffer.h"
+
+namespace esphome::api {
+
+void APIBuffer::grow_(size_t n) {
+  auto new_data = make_buffer(n);
+  if (this->size_)
+    std::memcpy(new_data.get(), this->data_.get(), this->size_);
+  this->data_ = std::move(new_data);
+  this->capacity_ = n;
+}
+
+}  // namespace esphome::api

esphome/components/api/api_buffer.h

@@ -0,0 +1,67 @@
+#pragma once
+
+#include <cstdint>
+#include <cstring>
+#include <memory>
+
+#include "esphome/core/defines.h"
+#include "esphome/core/helpers.h"
+
+namespace esphome::api {
+
+/// Helper to use make_unique_for_overwrite where available (skips zero-fill),
+/// falling back to make_unique on older GCC (ESP8266, BK72xx, LN882x).
+inline std::unique_ptr<uint8_t[]> make_buffer(size_t n) {
+#if defined(USE_ESP8266) || defined(USE_BK72XX) || defined(USE_LN882X)
+  return std::make_unique<uint8_t[]>(n);
+#else
+  return std::make_unique_for_overwrite<uint8_t[]>(n);
+#endif
+}
+
+/// Byte buffer that skips zero-initialization on resize().
+///
+/// std::vector<uint8_t>::resize() zero-fills new bytes via memset. For the
+/// shared protobuf write buffer, every byte is overwritten by the encoder,
+/// making the zero-fill pure waste. For the receive buffer, bytes are
+/// overwritten by socket reads.
+///
+/// Designed for bulk clear/resize/overwrite patterns. grow_() allocates
+/// exactly the requested size (no growth factor) since callers resize to
+/// known sizes rather than appending incrementally.
+///
+/// Safe because: callers always write exactly the number of bytes they
+/// resize for. In the protobuf write path, debug_check_bounds_ validates
+/// writes in debug builds.
+class APIBuffer {
+ public:
+  void clear() { this->size_ = 0; }
+  inline void reserve(size_t n) ESPHOME_ALWAYS_INLINE {
+    if (n > this->capacity_)
+      this->grow_(n);
+  }
+  inline void resize(size_t n) ESPHOME_ALWAYS_INLINE {
+    this->reserve(n);
+    this->size_ = n;  // no zero-fill
+  }
+  uint8_t *data() { return this->data_.get(); }
+  const uint8_t *data() const { return this->data_.get(); }
+  size_t size() const { return this->size_; }
+  bool empty() const { return this->size_ == 0; }
+  uint8_t &operator[](size_t i) { return this->data_[i]; }
+  const uint8_t &operator[](size_t i) const { return this->data_[i]; }
+  /// Release all memory (equivalent to std::vector swap trick).
+  void release() {
+    this->data_.reset();
+    this->size_ = 0;
+    this->capacity_ = 0;
+  }
+
+ protected:
+  void grow_(size_t n);
+  std::unique_ptr<uint8_t[]> data_;
+  size_t size_{0};
+  size_t capacity_{0};
+};
+
+}  // namespace esphome::api

esphome/components/api/api_connection.cpp

@@ -1571,7 +1571,7 @@ bool APIConnection::send_ping_response_() {
 }
 
 bool APIConnection::send_device_info_response_() {
-  DeviceInfoResponse resp{};
+  DeviceInfoResponse resp;
   resp.name = StringRef(App.get_name());
   resp.friendly_name = StringRef(App.get_friendly_name());
 #ifdef USE_AREAS
@@ -2083,7 +2083,7 @@ void APIConnection::process_batch_() {
 
 // Separated from process_batch_() so the single-message fast path gets a minimal
 // stack frame without the MAX_MESSAGES_PER_BATCH * sizeof(MessageInfo) array.
-void APIConnection::process_batch_multi_(ProtoByteBuffer &shared_buf, size_t num_items, uint8_t header_padding,
+void APIConnection::process_batch_multi_(APIBuffer &shared_buf, size_t num_items, uint8_t header_padding,
                                          uint8_t footer_size) {
   // Ensure MessageInfo remains trivially destructible for our placement new approach
   static_assert(std::is_trivially_destructible<MessageInfo>::value,

esphome/components/api/api_connection.h

@@ -278,7 +278,7 @@ class APIConnection final : public APIServerConnectionBase {
     }
   }
 
-  void prepare_first_message_buffer(ProtoByteBuffer &shared_buf, size_t header_padding, size_t total_size) {
+  void prepare_first_message_buffer(APIBuffer &shared_buf, size_t header_padding, size_t total_size) {
     shared_buf.clear();
     // Reserve space for header padding + message + footer
     // - Header padding: space for protocol headers (7 bytes for Noise, 6 for Plaintext)
@@ -289,7 +289,7 @@ class APIConnection final : public APIServerConnectionBase {
   }
 
   // Convenience overload - computes frame overhead internally
-  void prepare_first_message_buffer(ProtoByteBuffer &shared_buf, size_t payload_size) {
+  void prepare_first_message_buffer(APIBuffer &shared_buf, size_t payload_size) {
     const uint8_t header_padding = this->helper_->frame_header_padding();
     const uint8_t footer_size = this->helper_->frame_footer_size();
     this->prepare_first_message_buffer(shared_buf, header_padding, payload_size + header_padding + footer_size);
@@ -669,7 +669,7 @@ class APIConnection final : public APIServerConnectionBase {
 
   bool schedule_batch_();
   void process_batch_();
-  void process_batch_multi_(ProtoByteBuffer &shared_buf, size_t num_items, uint8_t header_padding, uint8_t footer_size)
+  void process_batch_multi_(APIBuffer &shared_buf, size_t num_items, uint8_t header_padding, uint8_t footer_size)
       __attribute__((noinline));
   void clear_batch_() {
     this->deferred_batch_.clear();

esphome/components/api/api_frame_helper.h

@@ -5,10 +5,10 @@
 #include <memory>
 #include <span>
 #include <utility>
-#include <vector>
 
 #include "esphome/core/defines.h"
 #ifdef USE_API
+#include "esphome/components/api/api_buffer.h"
 #include "esphome/components/socket/socket.h"
 #include "esphome/core/application.h"
 #include "esphome/core/log.h"
@@ -178,8 +178,7 @@ class APIFrameHelper {
     // rx_buf_len_ tracks bytes read so far; if non-zero, we're mid-frame
     // and clearing would lose partially received data.
     if (this->rx_buf_len_ == 0) {
-      // Use swap trick since shrink_to_fit() is non-binding and may be ignored
-      std::vector<uint8_t>().swap(this->rx_buf_);
+      this->rx_buf_.release();
     }
   }
 
@@ -206,9 +205,6 @@ class APIFrameHelper {
 
   // Common socket write error handling
   APIError handle_socket_write_error_();
-  template<typename StateEnum>
-  APIError write_raw_(const struct iovec *iov, int iovcnt, socket::Socket *socket, std::vector<uint8_t> &tx_buf,
-                      const std::string &info, StateEnum &state, StateEnum failed_state);
 
   // Socket ownership (4 bytes on 32-bit, 8 bytes on 64-bit)
   std::unique_ptr<socket::Socket> socket_;
@@ -234,7 +230,7 @@ class APIFrameHelper {
 
   // Containers (size varies, but typically 12+ bytes on 32-bit)
   std::array<std::unique_ptr<SendBuffer>, API_MAX_SEND_QUEUE> tx_buf_;
-  std::vector<uint8_t> rx_buf_;
+  APIBuffer rx_buf_;
 
   // Client name buffer - stores name from Hello message or initial peername
   char client_name_[CLIENT_INFO_NAME_MAX_LEN]{};

esphome/components/api/api_frame_helper_noise.cpp

@@ -579,8 +579,7 @@ APIError APINoiseFrameHelper::init_handshake_() {
   if (aerr != APIError::OK)
     return aerr;
   // set_prologue copies it into handshakestate, so we can get rid of it now
-  // Use swap idiom to actually release memory (= {} only clears size, not capacity)
-  std::vector<uint8_t>().swap(prologue_);
+  prologue_.release();
 
   err = noise_handshakestate_start(handshake_);
   aerr = handle_noise_error_(err, LOG_STR("noise_handshakestate_start"), APIError::HANDSHAKESTATE_SETUP_FAILED);

esphome/components/api/api_frame_helper_noise.h

@@ -43,8 +43,8 @@ class APINoiseFrameHelper final : public APIFrameHelper {
   // Reference to noise context (4 bytes on 32-bit)
   APINoiseContext &ctx_;
 
-  // Vector (12 bytes on 32-bit)
-  std::vector<uint8_t> prologue_;
+  // Buffer for noise handshake prologue (released after handshake)
+  APIBuffer prologue_;
 
   // NoiseProtocolId (size depends on implementation)
   NoiseProtocolId nid_;

esphome/components/api/api_options.proto

@@ -90,4 +90,10 @@ extend google.protobuf.FieldOptions {
     //   - uint16_t <field>_length_{0};
     //   - uint16_t <field>_count_{0};
     optional bool packed_buffer = 50015 [default=false];
+
+    // force: Always encode this field, even when its value equals the proto3 default.
+    // Skips the zero/empty check in calculate_size() and encode(), using the _force
+    // variant of the calc_ method. Use on fields that are almost always non-default
+    // to eliminate dead branches on hot paths.
+    optional bool force = 50016 [default=false];
 }

esphome/components/api/api_pb2.cpp

@@ -139,7 +139,7 @@ void DeviceInfoResponse::encode(ProtoWriteBuffer &buffer) const {
 #endif
 #ifdef USE_SERIAL_PROXY
   for (const auto &it : this->serial_proxies) {
-    buffer.encode_message(25, it);
+    buffer.encode_sub_message(25, it);
   }
 #endif
 }
@@ -2245,17 +2245,17 @@ bool SubscribeBluetoothLEAdvertisementsRequest::decode_varint(uint32_t field_id,
   return true;
 }
 void BluetoothLERawAdvertisement::encode(ProtoWriteBuffer &buffer) const {
-  buffer.encode_uint64(1, this->address);
-  buffer.encode_sint32(2, this->rssi);
+  buffer.encode_uint64(1, this->address, true);
+  buffer.encode_sint32(2, this->rssi, true);
   buffer.encode_uint32(3, this->address_type);
-  buffer.encode_bytes(4, this->data, this->data_len);
+  buffer.encode_bytes(4, this->data, this->data_len, true);
 }
 uint32_t BluetoothLERawAdvertisement::calculate_size() const {
   uint32_t size = 0;
-  size += ProtoSize::calc_uint64(1, this->address);
-  size += ProtoSize::calc_sint32(1, this->rssi);
+  size += ProtoSize::calc_uint64_force(1, this->address);
+  size += ProtoSize::calc_sint32_force(1, this->rssi);
   size += ProtoSize::calc_uint32(1, this->address_type);
-  size += ProtoSize::calc_length(1, this->data_len);
+  size += ProtoSize::calc_length_force(1, this->data_len);
   return size;
 }
 void BluetoothLERawAdvertisementsResponse::encode(ProtoWriteBuffer &buffer) const {

esphome/components/api/api_server.h

@@ -2,6 +2,7 @@
 
 #include "esphome/core/defines.h"
 #ifdef USE_API
+#include "api_buffer.h"
 #include "api_noise_context.h"
 #include "api_pb2.h"
 #include "api_pb2_service.h"
@@ -65,7 +66,7 @@ class APIServer : public Component,
   void set_max_connections(uint8_t max_connections) { this->max_connections_ = max_connections; }
 
   // Get reference to shared buffer for API connections
-  ProtoByteBuffer &get_shared_buffer_ref() { return shared_write_buffer_; }
+  APIBuffer &get_shared_buffer_ref() { return shared_write_buffer_; }
 
 #ifdef USE_API_NOISE
   bool save_noise_psk(psk_t psk, bool make_active = true);
@@ -276,7 +277,7 @@ class APIServer : public Component,
   // Not pre-allocated: all send paths call prepare_first_message_buffer() which
   // reserves the exact needed size. Pre-allocating here would cause heap fragmentation
   // since the buffer would almost always reallocate on first use.
-  ProtoByteBuffer shared_write_buffer_;
+  APIBuffer shared_write_buffer_;
 #ifdef USE_API_HOMEASSISTANT_STATES
   std::vector<HomeAssistantStateSubscription> state_subs_;
 #endif

esphome/components/api/proto.cpp

@@ -8,16 +8,18 @@ namespace esphome::api {
 
 static const char *const TAG = "api.proto";
 
-void ProtoByteBuffer::grow_(size_t n) {
-  auto new_data = make_buffer(n);
-  if (this->size_)
-    std::memcpy(new_data.get(), this->data_.get(), this->size_);
-  this->data_ = std::move(new_data);
-  this->capacity_ = n;
-}
-
 uint32_t ProtoSize::varint_slow(uint32_t value) { return varint_wide(value); }
 
+void ProtoWriteBuffer::encode_varint_raw_slow_(uint32_t value) {
+  do {
+    this->debug_check_bounds_(1);
+    *this->pos_++ = static_cast<uint8_t>(value | 0x80);
+    value >>= 7;
+  } while (value > 0x7F);
+  this->debug_check_bounds_(1);
+  *this->pos_++ = static_cast<uint8_t>(value);
+}
+
 #ifdef USE_API_VARINT64
 optional<ProtoVarInt> ProtoVarInt::parse_wide(const uint8_t *buffer, uint32_t len, uint32_t *consumed,
                                               uint32_t result32) {

esphome/components/api/proto.h

@@ -1,6 +1,7 @@
 #pragma once
 
 #include "api_pb2_defines.h"
+#include "api_buffer.h"
 #include "esphome/core/component.h"
 #include "esphome/core/helpers.h"
 #include "esphome/core/log.h"
@@ -8,7 +9,6 @@
 
 #include <cassert>
 #include <cstring>
-#include <memory>
 #include <vector>
 
 #ifdef ESPHOME_LOG_HAS_VERY_VERBOSE
@@ -236,60 +236,17 @@ class Proto32Bit {
 
 // NOTE: Proto64Bit class removed - wire type 1 (64-bit fixed) not supported
 
-/// Helper to use make_unique_for_overwrite where available (skips zero-fill),
-/// falling back to make_unique on older GCC (ESP8266, BK72xx, LN882x).
-inline std::unique_ptr<uint8_t[]> make_buffer(size_t n) {
-#if defined(USE_ESP8266) || defined(USE_BK72XX) || defined(USE_LN882X)
-  return std::make_unique<uint8_t[]>(n);
-#else
-  return std::make_unique_for_overwrite<uint8_t[]>(n);
-#endif
-}
-
-/// Byte buffer that skips zero-initialization on resize().
-///
-/// std::vector<uint8_t>::resize() zero-fills new bytes via memset. The shared
-/// protobuf write buffer is clear()'d before every message, so resize() always
-/// grows from size 0 — memsetting the entire requested region. Every byte is
-/// then overwritten by the encoder, making the zero-fill pure waste.
-///
-/// Safe because: the encoder writes exactly calculate_size() bytes, the frame
-/// helper sends exactly those bytes, and debug_check_bounds_ validates writes
-/// in debug builds. No byte is ever read before being written.
-class ProtoByteBuffer {
- public:
-  void clear() { this->size_ = 0; }
-  inline void reserve(size_t n) ESPHOME_ALWAYS_INLINE {
-    if (n > this->capacity_)
-      this->grow_(n);
-  }
-  inline void resize(size_t n) ESPHOME_ALWAYS_INLINE {
-    this->reserve(n);
-    this->size_ = n;  // no zero-fill
-  }
-  uint8_t *data() { return this->data_.get(); }
-  const uint8_t *data() const { return this->data_.get(); }
-  size_t size() const { return this->size_; }
-
- protected:
-  void grow_(size_t n);
-  std::unique_ptr<uint8_t[]> data_;
-  size_t size_{0};
-  size_t capacity_{0};
-};
-
 class ProtoWriteBuffer {
  public:
-  ProtoWriteBuffer(ProtoByteBuffer *buffer) : buffer_(buffer), pos_(buffer->data() + buffer->size()) {}
-  ProtoWriteBuffer(ProtoByteBuffer *buffer, size_t write_pos) : buffer_(buffer), pos_(buffer->data() + write_pos) {}
-  void encode_varint_raw(uint32_t value) {
-    while (value > 0x7F) {
+  ProtoWriteBuffer(APIBuffer *buffer) : buffer_(buffer), pos_(buffer->data() + buffer->size()) {}
+  ProtoWriteBuffer(APIBuffer *buffer, size_t write_pos) : buffer_(buffer), pos_(buffer->data() + write_pos) {}
+  inline void ESPHOME_ALWAYS_INLINE encode_varint_raw(uint32_t value) {
+    if (value < 128) [[likely]] {
       this->debug_check_bounds_(1);
-      *this->pos_++ = static_cast<uint8_t>(value | 0x80);
-      value >>= 7;
+      *this->pos_++ = static_cast<uint8_t>(value);
+      return;
     }
-    this->debug_check_bounds_(1);
-    *this->pos_++ = static_cast<uint8_t>(value);
+    this->encode_varint_raw_slow_(value);
   }
   void encode_varint_raw_64(uint64_t value) {
     while (value > 0x7F) {
@@ -417,9 +374,12 @@ class ProtoWriteBuffer {
   // Non-template core for encode_optional_sub_message.
   void encode_optional_sub_message(uint32_t field_id, uint32_t nested_size, const void *value,
                                    void (*encode_fn)(const void *, ProtoWriteBuffer &));
-  ProtoByteBuffer *get_buffer() const { return buffer_; }
+  APIBuffer *get_buffer() const { return buffer_; }
 
  protected:
+  // Slow path for encode_varint_raw values >= 128, outlined to keep fast path small
+  void encode_varint_raw_slow_(uint32_t value) __attribute__((noinline));
+
 #ifdef ESPHOME_DEBUG_API
   void debug_check_bounds_(size_t bytes, const char *caller = __builtin_FUNCTION());
   void debug_check_encode_size_(uint32_t field_id, uint32_t expected, ptrdiff_t actual);
@@ -427,7 +387,7 @@ class ProtoWriteBuffer {
   void debug_check_bounds_([[maybe_unused]] size_t bytes) {}
 #endif
 
-  ProtoByteBuffer *buffer_;
+  APIBuffer *buffer_;
   uint8_t *pos_;
 };
 
@@ -554,7 +514,7 @@ class ProtoSize {
    * @return The number of bytes needed to encode the value
    */
   static constexpr inline uint32_t ESPHOME_ALWAYS_INLINE varint(uint32_t value) {
-    if (value < 128)
+    if (value < 128) [[likely]]
       return 1;  // Fast path: 7 bits, most common case
     if (__builtin_is_constant_evaluated())
       return varint_wide(value);

esphome/components/ble_nus/ble_nus.cpp

@@ -100,16 +100,17 @@ size_t BLENUS::available() {
 #endif
 }
 
-void BLENUS::flush() {
+uart::FlushResult BLENUS::flush() {
   constexpr uint32_t timeout_5sec = 5000;
   uint32_t start = millis();
   while (atomic_get(&this->tx_status_) != TX_DISABLED && !ring_buf_is_empty(&global_ble_tx_ring_buf)) {
     if (millis() - start > timeout_5sec) {
       ESP_LOGW(TAG, "Flush timeout");
-      return;
+      return uart::FlushResult::TIMEOUT;
     }
     delay(1);
   }
+  return uart::FlushResult::SUCCESS;
 }
 
 void BLENUS::connected(bt_conn *conn, uint8_t err) {

esphome/components/ble_nus/ble_nus.h

@@ -26,7 +26,7 @@ class BLENUS : public uart::UARTComponent, public Component {
   bool peek_byte(uint8_t *data) override;
   bool read_array(uint8_t *data, size_t len) override;
   size_t available() override;
-  void flush() override;
+  uart::FlushResult flush() override;
   void check_logger_conflict() override {}
   void set_expose_log(bool expose_log) { this->expose_log_ = expose_log; }
 #ifdef USE_LOGGER

esphome/components/i2s_audio/__init__.py

@@ -281,6 +281,8 @@ async def to_code(config):
 
     if use_legacy():
         cg.add_define("USE_I2S_LEGACY")
+        # Legacy I2S API lives in the "driver" shim component (driver/i2s.h)
+        include_builtin_idf_component("driver")
 
     # Helps avoid callbacks being skipped due to processor load
     add_idf_sdkconfig_option("CONFIG_I2S_ISR_IRAM_SAFE", True)

esphome/components/modbus/modbus.cpp

@@ -11,6 +11,11 @@ static const char *const TAG = "modbus";
 // Maximum bytes to log for Modbus frames (truncated if larger)
 static constexpr size_t MODBUS_MAX_LOG_BYTES = 64;
 
+// Approximate bits per character on the wire (depends on parity/stop bit config)
+static constexpr uint32_t MODBUS_BITS_PER_CHAR = 11;
+// Milliseconds per second
+static constexpr uint32_t MS_PER_SEC = 1000;
+
 void Modbus::setup() {
   if (this->flow_control_pin_ != nullptr) {
     this->flow_control_pin_->setup();
@@ -19,10 +24,17 @@ void Modbus::setup() {
   this->frame_delay_ms_ =
       std::max(2,  // 1750us minimum per spec - rounded up to 2ms.
                    // 3.5 characters * 11 bits per character * 1000ms/sec / (bits/sec) (Standard modbus frame delay)
-               (uint16_t) (3.5 * 11 * 1000 / this->parent_->get_baud_rate()) + 1);
+               (uint16_t) (3.5 * MODBUS_BITS_PER_CHAR * MS_PER_SEC / this->parent_->get_baud_rate()) + 1);
 
+  // When rx_full_threshold is configured (non-zero), the UART has a hardware FIFO with a
+  // meaningful threshold (e.g., ESP32 native UART), so we can calculate a precise delay.
+  // Otherwise (e.g., USB UART), use 50ms to handle data arriving in chunks.
+  static constexpr uint16_t DEFAULT_LONG_RX_BUFFER_DELAY_MS = 50;
+  size_t rx_threshold = this->parent_->get_rx_full_threshold();
   this->long_rx_buffer_delay_ms_ =
-      (this->parent_->get_rx_full_threshold() * 11 * 1000 / this->parent_->get_baud_rate()) + 1;
+      rx_threshold != uart::UARTComponent::RX_FULL_THRESHOLD_UNSET
+          ? (rx_threshold * MODBUS_BITS_PER_CHAR * MS_PER_SEC / this->parent_->get_baud_rate()) + 1
+          : DEFAULT_LONG_RX_BUFFER_DELAY_MS;
 }
 
 void Modbus::loop() {
@@ -290,7 +302,7 @@ void Modbus::send_next_frame_() {
     this->last_send_tx_offset_ = 0;
   } else {
     this->write_array(frame.data.get(), frame.size);
-    this->last_send_tx_offset_ = frame.size * 11 * 1000 / this->parent_->get_baud_rate() + 1;
+    this->last_send_tx_offset_ = frame.size * MODBUS_BITS_PER_CHAR * MS_PER_SEC / this->parent_->get_baud_rate() + 1;
   }
 
 #if ESPHOME_LOG_LEVEL >= ESPHOME_LOG_LEVEL_VERBOSE

esphome/components/nrf52/__init__.py

@@ -27,10 +27,15 @@ from esphome.components.zephyr.const import (
 )
 import esphome.config_validation as cv
 from esphome.const import (
+    CONF_ADVANCED,
     CONF_BOARD,
+    CONF_DISABLED,
+    CONF_ENABLE_OTA_ROLLBACK,
     CONF_FRAMEWORK,
     CONF_ID,
+    CONF_OTA,
     CONF_RESET_PIN,
+    CONF_SAFE_MODE,
     CONF_VERSION,
     CONF_VOLTAGE,
     KEY_CORE,
@@ -41,6 +46,7 @@ from esphome.const import (
     ThreadModel,
 )
 from esphome.core import CORE, CoroPriority, EsphomeError, coroutine_with_priority
+import esphome.final_validate as fv
 from esphome.storage_json import StorageJSON
 from esphome.types import ConfigType
 
@@ -133,6 +139,7 @@ CONF_UICR_ERASE = "uicr_erase"
 
 VOLTAGE_LEVELS = [1.8, 2.1, 2.4, 2.7, 3.0, 3.3]
 
+
 CONFIG_SCHEMA = cv.All(
     _detect_bootloader,
     set_core_data,
@@ -156,9 +163,19 @@ CONFIG_SCHEMA = cv.All(
                     cv.Optional(CONF_UICR_ERASE, default=False): cv.boolean,
                 }
             ),
-            cv.Optional(CONF_FRAMEWORK, default={CONF_VERSION: "2.6.1-a"}): cv.Schema(
+            cv.Optional(
+                CONF_FRAMEWORK,
+                default={},
+            ): cv.Schema(
                 {
-                    cv.Required(CONF_VERSION): cv.string_strict,
+                    cv.Optional(CONF_VERSION, default="2.6.1-a"): cv.string_strict,
+                    cv.Optional(CONF_ADVANCED, default={}): cv.Schema(
+                        {
+                            cv.Optional(
+                                CONF_ENABLE_OTA_ROLLBACK, default=True
+                            ): cv.boolean,
+                        }
+                    ),
                 }
             ),
             cv.GenerateID(CONF_CDC_ACM): cv.declare_id(CdcAcm),
@@ -181,6 +198,24 @@ def _final_validate(config):
         _LOGGER.warning(
             "Selected generic Adafruit bootloader. The board might crash. Consider settings `bootloader:`"
         )
+    full_config = fv.full_config.get()
+    conf = config[CONF_FRAMEWORK]
+    advanced = conf[CONF_ADVANCED]
+
+    if advanced[CONF_ENABLE_OTA_ROLLBACK]:
+        # "disabled: false" means safe mode *is* enabled.
+        safe_mode_config = full_config.get(CONF_SAFE_MODE, {CONF_DISABLED: True})
+        safe_mode_enabled = not safe_mode_config[CONF_DISABLED]
+        ota_enabled = CONF_OTA in full_config
+        # Both need to be enabled for rollback to work
+        if not (ota_enabled and safe_mode_enabled):
+            # But only warn if ota is even possible
+            if ota_enabled:
+                _LOGGER.warning(
+                    "OTA rollback requires safe_mode, disabling rollback support"
+                )
+            # disable the rollback feature anyway since it can't be used.
+            advanced[CONF_ENABLE_OTA_ROLLBACK] = False
 
 
 FINAL_VALIDATE_SCHEMA = _final_validate
@@ -247,6 +282,11 @@ async def to_code(config: ConfigType) -> None:
         if reg0_config[CONF_UICR_ERASE]:
             cg.add_define("USE_NRF52_UICR_ERASE")
 
+    conf = config[CONF_FRAMEWORK]
+    advanced = conf[CONF_ADVANCED]
+    # Enable OTA rollback support
+    if advanced[CONF_ENABLE_OTA_ROLLBACK]:
+        cg.add_define("USE_OTA_ROLLBACK")
     # c++ support
     if framework_ver < cv.Version(2, 9, 2):
         zephyr_add_prj_conf("CPLUSPLUS", True)
@@ -259,7 +299,7 @@ async def to_code(config: ConfigType) -> None:
     zephyr_add_prj_conf("WDT_DISABLE_AT_BOOT", False)
     # disable console
     zephyr_add_prj_conf("UART_CONSOLE", False)
-    zephyr_add_prj_conf("CONSOLE", False)
+    zephyr_add_prj_conf("CONSOLE", False, False)
     # use NFC pins as GPIO
     if framework_ver < cv.Version(2, 9, 2):
         zephyr_add_prj_conf("NFCT_PINS_AS_GPIOS", True)
@@ -352,22 +392,42 @@ def _upload_using_platformio(
 def upload_program(config: ConfigType, args, host: str) -> bool:
     from esphome.__main__ import check_permissions, get_port_type
 
-    result = 0
-    handled = False
+    mcumgr_device: str | None = None
 
     if get_port_type(host) == "SERIAL":
         check_permissions(host)
-        result = _upload_using_platformio(config, host, ["-t", "upload"])
-        handled = True
+        if zephyr_data()[KEY_BOOTLOADER] == BOOTLOADER_MCUBOOT:
+            mcumgr_device = host
+        else:
+            result = _upload_using_platformio(config, host, ["-t", "upload"])
+            if result != 0:
+                raise EsphomeError(f"Upload failed with result: {result}")
+            return True  # Handled: platformio serial upload
 
     if host == "PYOCD":
         result = _upload_using_platformio(config, host, ["-t", "flash_pyocd"])
-        handled = True
+        if result != 0:
+            raise EsphomeError(f"Upload failed with result: {result}")
+        return True  # Handled: platformio PYOCD upload
 
-    if result != 0:
-        raise EsphomeError(f"Upload failed with result: {result}")
+    # Deferred imports: bleak/smpclient are heavy, only load for BLE/mcumgr paths
+    from .ble_logger import is_mac_address
+    from .ota import smpmgr_scan, smpmgr_upload
 
-    return handled
+    if host == "BLE":
+        mcumgr_device = asyncio.run(smpmgr_scan(CORE.name))
+
+    if is_mac_address(host):
+        mcumgr_device = host
+
+    if mcumgr_device:
+        firmware = Path(
+            CORE.relative_pioenvs_path(CORE.name, "zephyr", "app_update.bin")
+        ).resolve()
+        asyncio.run(smpmgr_upload(mcumgr_device, firmware))
+        return True  # Handled: mcumgr OTA upload
+
+    return False  # Not handled: let caller try default upload methods
 
 
 def show_logs(config: ConfigType, args, devices: list[str]) -> bool:
@@ -375,7 +435,7 @@ def show_logs(config: ConfigType, args, devices: list[str]) -> bool:
     from .ble_logger import is_mac_address, logger_connect, logger_scan
 
     if devices[0] == "BLE":
-        ble_device = asyncio.run(logger_scan(CORE.config["esphome"]["name"]))
+        ble_device = asyncio.run(logger_scan(CORE.name))
         if ble_device:
             address = ble_device.address
         else:

esphome/components/nrf52/ota.py

@@ -0,0 +1,164 @@
+import asyncio
+from dataclasses import asdict
+import json
+import logging
+from pathlib import Path
+
+from bleak import BleakScanner
+from bleak.exc import BleakDeviceNotFoundError
+from smp.exceptions import SMPBadStartDelimiter
+from smpclient import SMPClient
+from smpclient.generics import error, success
+from smpclient.mcuboot import IMAGE_TLV, ImageInfo, MCUBootImageError, TLVNotFound
+from smpclient.requests.image_management import ImageStatesRead, ImageStatesWrite
+from smpclient.requests.os_management import ResetWrite
+from smpclient.transport import SMPTransportDisconnected
+from smpclient.transport.ble import (
+    SMPBLETransport,
+    SMPBLETransportDeviceNotFound,
+    SMPBLETransportException,
+)
+from smpclient.transport.serial import SMPSerialTransport
+
+from esphome.core import EsphomeError
+from esphome.espota2 import ProgressBar
+
+from .ble_logger import is_mac_address
+
+SMP_SERVICE_UUID = "8D53DC1D-1DB7-4CD3-868B-8A527460AA84"
+BLE_SCAN_TIMEOUT = 10.0  # seconds
+RESET_DELAY = 2.0  # seconds to wait before reset, allows on_end action to execute
+
+_LOGGER = logging.getLogger(__name__)
+
+
+def _json_state(o: object) -> object:
+    """JSON serializer for SMP image state objects."""
+    if isinstance(o, (bytes, bytearray)):
+        return o.hex()
+    if hasattr(o, "hex"):
+        return o.hex()
+    if hasattr(o, "__dict__"):
+        return vars(o)
+    return str(o)
+
+
+async def smpmgr_scan(name: str) -> str:
+    _LOGGER.info("Scanning bluetooth for %s...", name)
+    for device in await BleakScanner.discover(
+        timeout=BLE_SCAN_TIMEOUT, service_uuids=[SMP_SERVICE_UUID]
+    ):
+        if device.name == name:
+            return device.address
+    raise EsphomeError(f"BLE device {name} with OTA service not found")
+
+
+async def smpmgr_upload(device: str, firmware: Path) -> None:
+    try:
+        await _smpmgr_upload(device, firmware)
+    except SMPTransportDisconnected as exc:
+        raise EsphomeError(f"{device} was disconnected.") from exc
+    except SMPBLETransportDeviceNotFound as exc:
+        raise EsphomeError(f"{device} was not found.") from exc
+
+
+def _get_image_tlv_sha256(file: Path) -> bytes:
+    _LOGGER.info("Checking image: %s", str(file))
+    try:
+        image_info = ImageInfo.load_file(str(file))
+        _LOGGER.info(
+            "Image header:\n%s", json.dumps(asdict(image_info.header), indent=2)
+        )
+        _LOGGER.debug(str(image_info))
+    except MCUBootImageError as exc:
+        raise EsphomeError("Inspection of FW image failed") from exc
+    except FileNotFoundError as exc:
+        raise EsphomeError(
+            f"Firmware image file not found: {file}. Build with zephyr_mcumgr enabled"
+        ) from exc
+
+    try:
+        image_tlv_sha256 = image_info.get_tlv(IMAGE_TLV.SHA256)
+        _LOGGER.info("Image tlv sha256: %s", image_tlv_sha256)
+    except TLVNotFound as exc:
+        raise EsphomeError("Could not find IMAGE_TLV_SHA256 in image.") from exc
+    return image_tlv_sha256.value
+
+
+async def _smpmgr_upload(device: str, firmware: Path) -> None:
+    image_tlv_sha256 = _get_image_tlv_sha256(firmware)
+
+    if is_mac_address(device):
+        smp_client = SMPClient(SMPBLETransport(), device)
+    else:
+        smp_client = SMPClient(SMPSerialTransport(), device)
+
+    _LOGGER.info("Connecting %s...", device)
+    try:
+        await smp_client.connect()
+    except BleakDeviceNotFoundError as exc:
+        raise EsphomeError(f"Device {device} not found") from exc
+    except SMPBLETransportException as exc:
+        raise EsphomeError(f"Connection error with {device}") from exc
+
+    _LOGGER.info("Connected %s...", device)
+    try:
+        await _smpmgr_upload_connected(smp_client, device, firmware, image_tlv_sha256)
+    finally:
+        await smp_client.disconnect()
+
+
+async def _smpmgr_upload_connected(
+    smp_client: SMPClient, device: str, firmware: Path, image_tlv_sha256: bytes
+) -> None:
+    try:
+        image_state = await smp_client.request(ImageStatesRead(), 2.5)
+    except (SMPBadStartDelimiter, TimeoutError) as exc:
+        raise EsphomeError(f"mcumgr is not supported by device ({device})") from exc
+
+    already_uploaded = False
+
+    if error(image_state):
+        raise EsphomeError(f"Failed to read image state from {device}: {image_state}")
+    if success(image_state):
+        if len(image_state.images) == 0:
+            _LOGGER.warning("No images on device!")
+        for image in image_state.images:
+            _LOGGER.info(
+                "Image state:\n%s",
+                json.dumps(image, indent=2, default=_json_state),
+            )
+            if image.active and not image.confirmed:
+                raise EsphomeError("No free slot. Testing mode but not confirmed yet.")
+            if image.hash == image_tlv_sha256:
+                if already_uploaded:
+                    raise EsphomeError("Both slots have the same image already")
+                if image.confirmed:
+                    raise EsphomeError("The same image already confirmed")
+                _LOGGER.warning("The same image already uploaded")
+                already_uploaded = True
+
+    if not already_uploaded:
+        with open(firmware, "rb") as file:
+            image = file.read()
+            upload_size = len(image)
+            progress = ProgressBar()
+            progress.update(0)
+            try:
+                async for offset in smp_client.upload(image):
+                    progress.update(offset / upload_size)
+            finally:
+                progress.done()
+
+    _LOGGER.info("Mark image for testing")
+    r = await smp_client.request(ImageStatesWrite(hash=image_tlv_sha256), 1.0)
+
+    if error(r):
+        raise EsphomeError(f"Failed to mark image for testing on {device}: {r}")
+
+    await asyncio.sleep(RESET_DELAY)
+    _LOGGER.info("Reset")
+    r = await smp_client.request(ResetWrite(), 1.0)
+
+    if error(r):
+        raise EsphomeError(f"Failed to reset {device}: {r}")

esphome/components/ota/__init__.py

@@ -18,7 +18,14 @@ from esphome.coroutine import CoroPriority
 OTA_STATE_LISTENER_KEY = "ota_state_listener"
 
 CODEOWNERS = ["@esphome/core"]
-AUTO_LOAD = ["md5", "safe_mode"]
+
+
+def AUTO_LOAD() -> list[str]:
+    components = ["safe_mode"]
+    if not CORE.using_zephyr:
+        components.extend(["md5"])
+    return components
+
 
 IS_PLATFORM_COMPONENT = True
 

esphome/components/rf_bridge/rf_bridge.cpp

@@ -145,6 +145,9 @@ void RFBridgeComponent::loop() {
     }
     avail -= to_read;
     for (size_t i = 0; i < to_read; i++) {
+      if (this->rx_buffer_.size() > MAX_RX_BUFFER_SIZE) {
+        this->rx_buffer_.clear();
+      }
       if (this->parse_bridge_byte_(buf[i])) {
         ESP_LOGVV(TAG, "Parsed: 0x%02X", buf[i]);
         this->last_bridge_byte_ = now;

esphome/components/rf_bridge/rf_bridge.h

@@ -30,6 +30,7 @@ static const uint8_t RF_CODE_RFIN_BUCKET = 0xB1;
 static const uint8_t RF_CODE_BEEP = 0xC0;
 static const uint8_t RF_CODE_STOP = 0x55;
 static const uint8_t RF_DEBOUNCE = 200;
+static const size_t MAX_RX_BUFFER_SIZE = 512;
 
 struct RFBridgeData {
   uint16_t sync;

esphome/components/safe_mode/safe_mode.cpp

@@ -9,10 +9,14 @@
 #include <cinttypes>
 #include <cstdio>
 
-#if defined(USE_ESP32) && defined(USE_OTA_ROLLBACK)
+#ifdef USE_OTA_ROLLBACK
+#ifdef USE_ZEPHYR
+#include <zephyr/dfu/mcuboot.h>
+#elif defined(USE_ESP32)
 #include <esp_ota_ops.h>
 #include <esp_system.h>
 #endif
+#endif
 
 namespace esphome::safe_mode {
 
@@ -66,9 +70,16 @@ float SafeModeComponent::get_setup_priority() const { return setup_priority::AFT
 void SafeModeComponent::mark_successful() {
   this->clean_rtc();
   this->boot_successful_ = true;
-#if defined(USE_ESP32) && defined(USE_OTA_ROLLBACK)
+#if defined(USE_OTA_ROLLBACK)
+// Mark OTA partition as valid to prevent rollback
+#if defined(USE_ZEPHYR)
+  if (!boot_is_img_confirmed()) {
+    boot_write_img_confirmed();
+  }
+#elif defined(USE_ESP32)
   // Mark OTA partition as valid to prevent rollback
   esp_ota_mark_app_valid_cancel_rollback();
+#endif
 #endif
   // Disable loop since we no longer need to check
   this->disable_loop();

esphome/components/sensirion_common/i2c_sensirion.cpp

@@ -12,7 +12,7 @@ static const char *const TAG = "sensirion_i2c";
 static const size_t BUFFER_STACK_SIZE = 16;
 
 bool SensirionI2CDevice::read_data(uint16_t *data, const uint8_t len) {
-  const uint8_t num_bytes = len * 3;
+  const size_t num_bytes = len * 3;
   uint8_t buf[num_bytes];
 
   this->last_error_ = this->read(buf, num_bytes);

esphome/components/uart/__init__.py

@@ -183,6 +183,7 @@ UART_PARITY_OPTIONS = {
     "ODD": UARTParityOptions.UART_CONFIG_PARITY_ODD,
 }
 
+CONF_FLUSH_TIMEOUT = "flush_timeout"
 CONF_RX_FULL_THRESHOLD = "rx_full_threshold"
 CONF_RX_TIMEOUT = "rx_timeout"
 
@@ -266,6 +267,9 @@ CONFIG_SCHEMA = cv.All(
             cv.SplitDefault(CONF_RX_TIMEOUT, esp32=2): cv.All(
                 cv.only_on_esp32, cv.validate_bytes, cv.int_range(min=0, max=92)
             ),
+            cv.Optional(CONF_FLUSH_TIMEOUT): cv.All(
+                cv.only_on_esp32, cv.positive_time_period_milliseconds
+            ),
             cv.Optional(CONF_STOP_BITS, default=1): cv.one_of(1, 2, int=True),
             cv.Optional(CONF_DATA_BITS, default=8): cv.int_range(min=5, max=8),
             cv.Optional(CONF_PARITY, default="NONE"): cv.enum(
@@ -345,6 +349,8 @@ async def to_code(config):
             )
         cg.add(var.set_rx_full_threshold(config[CONF_RX_FULL_THRESHOLD]))
         cg.add(var.set_rx_timeout(config[CONF_RX_TIMEOUT]))
+        if CONF_FLUSH_TIMEOUT in config:
+            cg.add(var.set_flush_timeout(config[CONF_FLUSH_TIMEOUT]))
     cg.add(var.set_stop_bits(config[CONF_STOP_BITS]))
     cg.add(var.set_data_bits(config[CONF_DATA_BITS]))
     cg.add(var.set_parity(config[CONF_PARITY]))

esphome/components/uart/uart.h

@@ -45,7 +45,7 @@ class UARTDevice {
 
   size_t available() { return this->parent_->available(); }
 
-  void flush() { this->parent_->flush(); }
+  FlushResult flush() { return this->parent_->flush(); }
 
   // Compat APIs
   int read() {

esphome/components/uart/uart_component.h

@@ -29,8 +29,18 @@ enum UARTDirection {
 
 const LogString *parity_to_str(UARTParityOptions parity);
 
+/// Result of a flush() call.
+enum class FlushResult {
+  SUCCESS,          ///< Confirmed: all bytes left the TX FIFO.
+  TIMEOUT,          ///< Confirmed: timed out before TX completed.
+  FAILED,           ///< Confirmed: driver or hardware error.
+  ASSUMED_SUCCESS,  ///< Platform cannot report result; success is assumed.
+};
+
 class UARTComponent {
  public:
+  static constexpr size_t RX_FULL_THRESHOLD_UNSET = 0;
+
   // Writes an array of bytes to the UART bus.
   // @param data A vector of bytes to be written.
   void write_array(const std::vector<uint8_t> &data) { this->write_array(&data[0], data.size()); }
@@ -72,7 +82,13 @@ class UARTComponent {
   virtual size_t available() = 0;
 
   // Pure virtual method to block until all bytes have been written to the UART bus.
-  virtual void flush() = 0;
+  // @return FlushResult indicating whether the flush was confirmed, timed out, failed, or assumed successful.
+  virtual FlushResult flush() = 0;
+
+  // Sets the maximum time to wait for TX to drain during flush().
+  // Only meaningful on ESP32 (IDF). Other platforms ignore this value.
+  // @param flush_timeout_ms Timeout in milliseconds; 0 means wait indefinitely.
+  virtual void set_flush_timeout(uint32_t flush_timeout_ms) {}
 
   // Sets the TX (transmit) pin for the UART bus.
   // @param tx_pin Pointer to the internal GPIO pin used for transmission.
@@ -187,7 +203,9 @@ class UARTComponent {
   InternalGPIOPin *rx_pin_{};
   InternalGPIOPin *flow_control_pin_{};
   size_t rx_buffer_size_{};
-  size_t rx_full_threshold_{1};
+  // ESP32 (both Arduino and ESP-IDF) always sets this at codegen time via set_rx_full_threshold().
+  // Other platforms (USB UART, Arduino, etc.) leave it unset.
+  size_t rx_full_threshold_{RX_FULL_THRESHOLD_UNSET};
   size_t rx_timeout_{0};
   uint32_t baud_rate_{0};
   uint8_t stop_bits_{0};

esphome/components/uart/uart_component_esp8266.cpp

@@ -213,13 +213,14 @@ size_t ESP8266UartComponent::available() {
     return this->sw_serial_->available();
   }
 }
-void ESP8266UartComponent::flush() {
+FlushResult ESP8266UartComponent::flush() {
   ESP_LOGVV(TAG, "    Flushing");
   if (this->hw_serial_ != nullptr) {
     this->hw_serial_->flush();
   } else {
     this->sw_serial_->flush();
   }
+  return FlushResult::ASSUMED_SUCCESS;
 }
 void ESP8266SoftwareSerial::setup(InternalGPIOPin *tx_pin, InternalGPIOPin *rx_pin, uint32_t baud_rate,
                                   uint8_t stop_bits, uint32_t data_bits, UARTParityOptions parity,

esphome/components/uart/uart_component_esp8266.h

@@ -58,7 +58,7 @@ class ESP8266UartComponent : public UARTComponent, public Component {
   bool read_array(uint8_t *data, size_t len) override;
 
   size_t available() override;
-  void flush() override;
+  FlushResult flush() override;
 
   uint32_t get_config();
 

esphome/components/uart/uart_component_esp_idf.cpp

@@ -230,6 +230,9 @@ void IDFUARTComponent::dump_config() {
                   "  RX Timeout: %u",
                   this->rx_buffer_size_, this->rx_full_threshold_, this->rx_timeout_);
   }
+  if (this->flush_timeout_ms_ > 0) {
+    ESP_LOGCONFIG(TAG, "  Flush Timeout: %" PRIu32 " ms", this->flush_timeout_ms_);
+  }
   ESP_LOGCONFIG(TAG,
                 "  Baud Rate: %" PRIu32 " baud\n"
                 "  Data Bits: %u\n"
@@ -332,9 +335,15 @@ size_t IDFUARTComponent::available() {
   return available;
 }
 
-void IDFUARTComponent::flush() {
+FlushResult IDFUARTComponent::flush() {
   ESP_LOGVV(TAG, "    Flushing");
-  uart_wait_tx_done(this->uart_num_, portMAX_DELAY);
+  TickType_t ticks = this->flush_timeout_ms_ == 0 ? portMAX_DELAY : pdMS_TO_TICKS(this->flush_timeout_ms_);
+  esp_err_t err = uart_wait_tx_done(this->uart_num_, ticks);
+  if (err == ESP_OK)
+    return FlushResult::SUCCESS;
+  if (err == ESP_ERR_TIMEOUT)
+    return FlushResult::TIMEOUT;
+  return FlushResult::FAILED;
 }
 
 void IDFUARTComponent::check_logger_conflict() {}

esphome/components/uart/uart_component_esp_idf.h

@@ -31,7 +31,9 @@ class IDFUARTComponent : public UARTComponent, public Component {
   bool read_array(uint8_t *data, size_t len) override;
 
   size_t available() override;
-  void flush() override;
+  FlushResult flush() override;
+
+  void set_flush_timeout(uint32_t flush_timeout_ms) override { this->flush_timeout_ms_ = flush_timeout_ms; }
 
   uint8_t get_hw_serial_number() { return this->uart_num_; }
 
@@ -57,6 +59,7 @@ class IDFUARTComponent : public UARTComponent, public Component {
 
   bool has_peek_{false};
   uint8_t peek_byte_;
+  uint32_t flush_timeout_ms_{0};  ///< 0 means wait indefinitely (portMAX_DELAY).
 
 #ifdef USE_UART_WAKE_LOOP_ON_RX
   // ISR callback for UART RX data notification — wakes the main loop directly.

esphome/components/uart/uart_component_host.cpp

@@ -274,12 +274,13 @@ size_t HostUartComponent::available() {
   return result;
 };
 
-void HostUartComponent::flush() {
+FlushResult HostUartComponent::flush() {
   if (this->file_descriptor_ == -1) {
-    return;
+    return FlushResult::ASSUMED_SUCCESS;
   }
   tcflush(this->file_descriptor_, TCIOFLUSH);
   ESP_LOGV(TAG, "    Flushing");
+  return FlushResult::ASSUMED_SUCCESS;
 }
 
 void HostUartComponent::update_error_(const std::string &error) {

esphome/components/uart/uart_component_host.h

@@ -18,7 +18,7 @@ class HostUartComponent : public UARTComponent, public Component {
   bool peek_byte(uint8_t *data) override;
   bool read_array(uint8_t *data, size_t len) override;
   size_t available() override;
-  void flush() override;
+  FlushResult flush() override;
   void set_name(std::string port_name) { port_name_ = port_name; };
 
  protected:

esphome/components/uart/uart_component_libretiny.cpp

@@ -170,9 +170,10 @@ bool LibreTinyUARTComponent::read_array(uint8_t *data, size_t len) {
 }
 
 size_t LibreTinyUARTComponent::available() { return this->serial_->available(); }
-void LibreTinyUARTComponent::flush() {
+FlushResult LibreTinyUARTComponent::flush() {
   ESP_LOGVV(TAG, "    Flushing");
   this->serial_->flush();
+  return FlushResult::ASSUMED_SUCCESS;
 }
 
 void LibreTinyUARTComponent::check_logger_conflict() {

esphome/components/uart/uart_component_libretiny.h

@@ -22,7 +22,7 @@ class LibreTinyUARTComponent : public UARTComponent, public Component {
   bool read_array(uint8_t *data, size_t len) override;
 
   size_t available() override;
-  void flush() override;
+  FlushResult flush() override;
 
   uint16_t get_config();
 

esphome/components/uart/uart_component_rp2040.cpp

@@ -187,9 +187,10 @@ bool RP2040UartComponent::read_array(uint8_t *data, size_t len) {
   return true;
 }
 size_t RP2040UartComponent::available() { return this->serial_->available(); }
-void RP2040UartComponent::flush() {
+FlushResult RP2040UartComponent::flush() {
   ESP_LOGVV(TAG, "    Flushing");
   this->serial_->flush();
+  return FlushResult::ASSUMED_SUCCESS;
 }
 
 }  // namespace esphome::uart

esphome/components/uart/uart_component_rp2040.h

@@ -25,7 +25,7 @@ class RP2040UartComponent : public UARTComponent, public Component {
   bool read_array(uint8_t *data, size_t len) override;
 
   size_t available() override;
-  void flush() override;
+  FlushResult flush() override;
 
   uint16_t get_config();
 

esphome/components/usb_cdc_acm/usb_cdc_acm.h

@@ -82,7 +82,7 @@ class USBCDCACMInstance : public uart::UARTComponent, public Parented<USBCDCACMC
   bool peek_byte(uint8_t *data) override;
   bool read_array(uint8_t *data, size_t len) override;
   size_t available() override;
-  void flush() override;
+  uart::FlushResult flush() override;
 
  protected:
   void check_logger_conflict() override;

esphome/components/usb_cdc_acm/usb_cdc_acm_esp32.cpp

@@ -326,10 +326,10 @@ size_t USBCDCACMInstance::available() {
   return waiting + (this->has_peek_ ? 1 : 0);
 }
 
-void USBCDCACMInstance::flush() {
+uart::FlushResult USBCDCACMInstance::flush() {
   // Wait for TX ring buffer to be empty
   if (this->usb_tx_ringbuf_ == nullptr) {
-    return;
+    return uart::FlushResult::ASSUMED_SUCCESS;
   }
 
   UBaseType_t waiting = 1;
@@ -341,7 +341,12 @@ void USBCDCACMInstance::flush() {
   }
 
   // Also wait for USB to finish transmitting
-  tinyusb_cdcacm_write_flush(static_cast<tinyusb_cdcacm_itf_t>(this->itf_), pdMS_TO_TICKS(100));
+  esp_err_t err = tinyusb_cdcacm_write_flush(static_cast<tinyusb_cdcacm_itf_t>(this->itf_), pdMS_TO_TICKS(100));
+  if (err == ESP_OK)
+    return uart::FlushResult::SUCCESS;
+  if (err == ESP_ERR_TIMEOUT)
+    return uart::FlushResult::TIMEOUT;
+  return uart::FlushResult::FAILED;
 }
 
 void USBCDCACMInstance::check_logger_conflict() {}

esphome/components/usb_uart/__init__.py

@@ -1,7 +1,7 @@
 import esphome.codegen as cg
 from esphome.components import socket
 from esphome.components.const import CONF_DATA_BITS, CONF_PARITY, CONF_STOP_BITS
-from esphome.components.uart import CONF_DEBUG_PREFIX, UARTComponent
+from esphome.components.uart import CONF_DEBUG_PREFIX, CONF_FLUSH_TIMEOUT, UARTComponent
 from esphome.components.usb_host import register_usb_client, usb_device_schema
 import esphome.config_validation as cv
 from esphome.const import (
@@ -91,6 +91,9 @@ def channel_schema(channels, baud_rate_required):
                             cv.Optional(CONF_DUMMY_RECEIVER, default=False): cv.boolean,
                             cv.Optional(CONF_DEBUG, default=False): cv.boolean,
                             cv.Optional(CONF_DEBUG_PREFIX, default=""): cv.string,
+                            cv.Optional(
+                                CONF_FLUSH_TIMEOUT, default="100ms"
+                            ): cv.positive_time_period_milliseconds,
                         }
                     )
                 ),
@@ -129,6 +132,7 @@ async def to_code(config):
             cg.add(chvar.set_parity(channel[CONF_PARITY]))
             cg.add(chvar.set_baud_rate(channel[CONF_BAUD_RATE]))
             cg.add(chvar.set_dummy_receiver(channel[CONF_DUMMY_RECEIVER]))
+            cg.add(chvar.set_flush_timeout(channel[CONF_FLUSH_TIMEOUT]))
             cg.add(chvar.set_debug(channel[CONF_DEBUG]))
             if channel[CONF_DEBUG_PREFIX]:
                 cg.add(chvar.set_debug_prefix(channel[CONF_DEBUG_PREFIX]))

esphome/components/usb_uart/usb_uart.cpp

@@ -166,17 +166,20 @@ void USBUartChannel::write_array(const uint8_t *data, size_t len) {
   this->parent_->start_output(this);
 }
 
-void USBUartChannel::flush() {
+uart::FlushResult USBUartChannel::flush() {
   // Spin until the output queue is drained and the last USB transfer completes.
   // Safe to call from the main loop only.
-  // The 100 ms timeout guards against a device that stops responding mid-flush;
+  // The flush_timeout_ms_ timeout guards against a device that stops responding mid-flush;
   // in that case the main loop is blocked for the full duration.
-  uint32_t start = millis();  // 100 ms safety timeout
-  while ((!this->output_queue_.empty() || this->output_started_.load()) && millis() - start < 100) {
+  uint32_t start = millis();
+  while ((!this->output_queue_.empty() || this->output_started_.load()) && millis() - start < this->flush_timeout_ms_) {
     // Kick start_output() in case data arrived but no transfer is in flight yet.
     this->parent_->start_output(this);
     yield();
   }
+  if (!this->output_queue_.empty() || this->output_started_.load())
+    return uart::FlushResult::TIMEOUT;
+  return uart::FlushResult::SUCCESS;
 }
 
 bool USBUartChannel::peek_byte(uint8_t *data) {
@@ -257,10 +260,12 @@ void USBUartComponent::dump_config() {
                   "    Data Bits: %u\n"
                   "    Parity: %s\n"
                   "    Stop bits: %s\n"
+                  "    Flush Timeout: %" PRIu32 " ms\n"
                   "    Debug: %s\n"
                   "    Dummy receiver: %s",
                   channel->index_, channel->baud_rate_, channel->data_bits_, PARITY_NAMES[channel->parity_],
-                  STOP_BITS_NAMES[channel->stop_bits_], YESNO(channel->debug_), YESNO(channel->dummy_receiver_));
+                  STOP_BITS_NAMES[channel->stop_bits_], channel->flush_timeout_ms_, YESNO(channel->debug_),
+                  YESNO(channel->dummy_receiver_));
   }
 }
 void USBUartComponent::start_input(USBUartChannel *channel) {

esphome/components/usb_uart/usb_uart.h

@@ -110,12 +110,13 @@ class USBUartChannel : public uart::UARTComponent, public Parented<USBUartCompon
   bool peek_byte(uint8_t *data) override;
   bool read_array(uint8_t *data, size_t len) override;
   size_t available() override { return this->input_buffer_.get_available(); }
-  void flush() override;
+  uart::FlushResult flush() override;
   void check_logger_conflict() override {}
   void set_parity(UARTParityOptions parity) { this->parity_ = parity; }
   void set_debug(bool debug) { this->debug_ = debug; }
   void set_dummy_receiver(bool dummy_receiver) { this->dummy_receiver_ = dummy_receiver; }
   void set_debug_prefix(const char *prefix) { this->debug_prefix_ = StringRef(prefix); }
+  void set_flush_timeout(uint32_t flush_timeout_ms) override { this->flush_timeout_ms_ = flush_timeout_ms; }
 
   /// Register a callback invoked immediately after data is pushed to the input ring buffer.
   /// Called from USBUartComponent::loop() in the main loop context.
@@ -124,23 +125,23 @@ class USBUartChannel : public uart::UARTComponent, public Parented<USBUartCompon
   void set_rx_callback(std::function<void()> cb) { this->rx_callback_ = std::move(cb); }
 
  protected:
-  // Larger structures first for better alignment
+  // Larger structures first (8+ bytes)
   RingBuffer input_buffer_;
   LockFreeQueue<UsbOutputChunk, USB_OUTPUT_CHUNK_COUNT> output_queue_;
   EventPool<UsbOutputChunk, USB_OUTPUT_CHUNK_COUNT> output_pool_;
   std::function<void()> rx_callback_{};
   CdcEps cdc_dev_{};
-  // Enum (likely 4 bytes)
+  StringRef debug_prefix_{};
+  // 4-byte fields
   UARTParityOptions parity_{UART_CONFIG_PARITY_NONE};
-  // Group atomics together (each 1 byte)
+  uint32_t flush_timeout_ms_{100};
+  // 1-byte fields (no padding between groups)
   std::atomic<bool> input_started_{true};
   std::atomic<bool> output_started_{true};
   std::atomic<bool> initialised_{false};
-  // Group regular bytes together to minimize padding
   const uint8_t index_;
   bool debug_{};
   bool dummy_receiver_{};
-  StringRef debug_prefix_{};
 };
 
 class USBUartComponent : public usb_host::USBClient {

esphome/components/vbus/vbus.cpp

@@ -87,6 +87,9 @@ void VBus::loop() {
           this->state_ = 0;
           ESP_LOGD(TAG, "P1 empty message");
         }
+      } else if (this->buffer_.size() > 15) {
+        ESP_LOGW(TAG, "Unknown protocol 0x%02x, discarding", this->protocol_);
+        this->state_ = 0;
       }
       continue;
     }

esphome/components/weikai/weikai.cpp

@@ -433,15 +433,16 @@ void WeikaiChannel::write_array(const uint8_t *buffer, size_t length) {
   this->reg(0).write_fifo(const_cast<uint8_t *>(buffer), length);
 }
 
-void WeikaiChannel::flush() {
+uart::FlushResult WeikaiChannel::flush() {
   uint32_t const start_time = millis();
   while (this->tx_fifo_is_not_empty_()) {  // wait until buffer empty
     if (millis() - start_time > 200) {
       ESP_LOGW(TAG, "WARNING flush timeout - still %d bytes not sent after 200 ms", this->tx_in_fifo_());
-      return;
+      return uart::FlushResult::TIMEOUT;
     }
     yield();  // reschedule our thread to avoid blocking
   }
+  return uart::FlushResult::SUCCESS;
 }
 
 size_t WeikaiChannel::xfer_fifo_to_buffer_() {

esphome/components/weikai/weikai.h

@@ -380,7 +380,7 @@ class WeikaiChannel : public uart::UARTComponent {
   /// @details If we refer to Serial.flush() in Arduino it says: ** Waits for the transmission of outgoing serial data
   /// to complete. (Prior to Arduino 1.0, this the method was removing any buffered incoming serial data.). ** Therefore
   /// we wait until all bytes are gone with a timeout of 100 ms
-  void flush() override;
+  uart::FlushResult flush() override;
 
  protected:
   friend class WeikaiComponent;

esphome/components/zephyr_mcumgr/ota/__init__.py

@@ -0,0 +1,141 @@
+import esphome.codegen as cg
+from esphome.components.ota import BASE_OTA_SCHEMA, OTAComponent, ota_to_code
+from esphome.components.zephyr import (
+    zephyr_add_cdc_acm,
+    zephyr_add_overlay,
+    zephyr_add_prj_conf,
+    zephyr_data,
+)
+from esphome.components.zephyr.const import BOOTLOADER_MCUBOOT, KEY_BOOTLOADER
+import esphome.config_validation as cv
+from esphome.const import CONF_HARDWARE_UART, CONF_ID, Framework
+from esphome.core import CORE, coroutine_with_priority
+from esphome.coroutine import CoroPriority
+from esphome.types import ConfigType
+
+CODEOWNERS = ["@tomaszduda23"]
+DEPENDENCIES = ["zephyr"]
+
+ZephyrMcumgrOTAComponent = cg.esphome_ns.namespace("zephyr_mcumgr").class_(
+    "OTAComponent", OTAComponent
+)
+
+CONF_BLE = "ble"
+CONF_TRANSPORT = "transport"
+
+
+def _validate_transport(conf: ConfigType) -> ConfigType:
+    transport = conf[CONF_TRANSPORT]
+    if transport[CONF_BLE] or CONF_HARDWARE_UART in transport:
+        return conf
+    raise cv.Invalid(
+        f"At least one transport protocol has to be enabled. Set '{CONF_BLE}: true' or '{CONF_HARDWARE_UART}'"
+    )
+
+
+UARTS = {
+    "CDC": ("cdc_acm_uart0", 0),
+    "CDC1": ("cdc_acm_uart1", 1),
+    "UART0": ("uart0", -1),
+    "UART1": ("uart1", -1),
+}
+
+
+CONFIG_SCHEMA = cv.All(
+    cv.Schema(
+        {
+            cv.GenerateID(): cv.declare_id(ZephyrMcumgrOTAComponent),
+            cv.Optional(CONF_TRANSPORT, default={CONF_BLE: True}): cv.Schema(
+                {
+                    cv.Optional(CONF_BLE, default=False): cv.boolean,
+                    cv.Optional(
+                        CONF_HARDWARE_UART,
+                    ): cv.one_of(*UARTS, upper=True),
+                }
+            ),
+        }
+    )
+    .extend(BASE_OTA_SCHEMA)
+    .extend(cv.COMPONENT_SCHEMA),
+    _validate_transport,
+    cv.only_with_framework(Framework.ZEPHYR),
+)
+
+
+def _validate_mcumgr_bootloader(config: ConfigType) -> None:
+    bootloader = zephyr_data()[KEY_BOOTLOADER]
+    if bootloader != BOOTLOADER_MCUBOOT:
+        raise cv.Invalid(f"'{bootloader}' bootloader does not support OTA")
+
+
+KEY_ZEPHYR_BLE_SERVER = "zephyr_ble_server"
+
+
+def _validate_ble_server(config: ConfigType) -> None:
+    if (
+        config[CONF_TRANSPORT][CONF_BLE]
+        and KEY_ZEPHYR_BLE_SERVER not in CORE.loaded_integrations
+    ):
+        raise cv.Invalid(f"'{KEY_ZEPHYR_BLE_SERVER}' component is required for BLE OTA")
+
+
+def _final_validate(config: ConfigType) -> None:
+    _validate_mcumgr_bootloader(config)
+    _validate_ble_server(config)
+
+
+FINAL_VALIDATE_SCHEMA = _final_validate
+
+
+@coroutine_with_priority(CoroPriority.OTA_UPDATES)
+async def to_code(config: ConfigType) -> None:
+    var = cg.new_Pvariable(config[CONF_ID])
+    await ota_to_code(var, config)
+
+    await cg.register_component(var, config)
+
+    zephyr_add_prj_conf("NET_BUF", True)
+    zephyr_add_prj_conf("ZCBOR", True)
+    zephyr_add_prj_conf("MCUMGR", True)
+
+    zephyr_add_prj_conf("MCUMGR_GRP_IMG", True)
+
+    zephyr_add_prj_conf("IMG_MANAGER", True)
+    zephyr_add_prj_conf("STREAM_FLASH", True)
+    zephyr_add_prj_conf("FLASH_MAP", True)
+    zephyr_add_prj_conf("FLASH", True)
+
+    zephyr_add_prj_conf("IMG_ERASE_PROGRESSIVELY", True)
+
+    zephyr_add_prj_conf("BOOTLOADER_MCUBOOT", True)
+
+    zephyr_add_prj_conf("MCUMGR_MGMT_NOTIFICATION_HOOKS", True)
+    zephyr_add_prj_conf("MCUMGR_GRP_IMG_STATUS_HOOKS", True)
+    zephyr_add_prj_conf("MCUMGR_GRP_IMG_UPLOAD_CHECK_HOOK", True)
+    transport = config[CONF_TRANSPORT]
+    if transport[CONF_BLE]:
+        zephyr_add_prj_conf("MCUMGR_TRANSPORT_BT", True)
+        zephyr_add_prj_conf("MCUMGR_TRANSPORT_BT_REASSEMBLY", True)
+
+        zephyr_add_prj_conf("MCUMGR_GRP_OS", True)
+        zephyr_add_prj_conf("MCUMGR_GRP_OS_MCUMGR_PARAMS", True)
+
+        zephyr_add_prj_conf("NCS_SAMPLE_MCUMGR_BT_OTA_DFU_SPEEDUP", True)
+    if CONF_HARDWARE_UART in transport:
+        uart = UARTS[transport[CONF_HARDWARE_UART]]
+        uart_name = uart[0]
+        cdc_id = uart[1]
+        if cdc_id >= 0:
+            zephyr_add_cdc_acm(config, cdc_id)
+        zephyr_add_prj_conf("MCUMGR_TRANSPORT_UART", True)
+        zephyr_add_prj_conf("BASE64", True)
+        zephyr_add_prj_conf("CONSOLE", True)
+        zephyr_add_overlay(
+            f"""
+                / {{
+                    chosen {{
+                        zephyr,uart-mcumgr = &{uart_name};
+                    }};
+                }};
+                """
+        )

esphome/components/zephyr_mcumgr/ota/ota_zephyr_mcumgr.cpp

@@ -0,0 +1,143 @@
+#ifdef USE_ZEPHYR
+#include "ota_zephyr_mcumgr.h"
+#include "esphome/core/log.h"
+#include "esphome/core/hal.h"
+#include <zephyr/sys/math_extras.h>
+#include <zephyr/usb/usb_device.h>
+#include <zephyr/dfu/mcuboot.h>
+
+// It should be from below header but there is problem with internal includes.
+// #include <zephyr/mgmt/mcumgr/grp/img_mgmt/img_mgmt.h>
+// NOLINTBEGIN(readability-identifier-naming,google-runtime-int)
+struct img_mgmt_upload_action {
+  /** The total size of the image. */
+  unsigned long long size;
+};
+
+struct img_mgmt_upload_req {
+  uint32_t image; /* 0 by default */
+  size_t off;     /* SIZE_MAX if unspecified */
+};
+// NOLINTEND(readability-identifier-naming,google-runtime-int)
+
+namespace esphome::zephyr_mcumgr {
+
+static_assert(sizeof(struct img_mgmt_upload_action) == 8, "ABI mismatch");
+static_assert(sizeof(struct img_mgmt_upload_req) == 8, "ABI mismatch");
+static_assert(offsetof(struct img_mgmt_upload_req, image) == 0, "ABI mismatch");
+static_assert(offsetof(struct img_mgmt_upload_req, off) == 4, "ABI mismatch");
+
+static const char *const TAG = "zephyr_mcumgr";
+static OTAComponent *global_ota_component;  // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
+
+static enum mgmt_cb_return mcumgr_img_mgmt_cb(uint32_t event, enum mgmt_cb_return prev_status, int32_t *rc,
+                                              uint16_t *group, bool *abort_more, void *data, size_t data_size) {
+  if (MGMT_EVT_OP_IMG_MGMT_DFU_CHUNK == event) {
+    const img_mgmt_upload_check &upload = *static_cast<img_mgmt_upload_check *>(data);
+    global_ota_component->update_chunk(upload);
+  } else if (MGMT_EVT_OP_IMG_MGMT_DFU_STARTED == event) {
+    global_ota_component->update_started();
+  } else if (MGMT_EVT_OP_IMG_MGMT_DFU_CHUNK_WRITE_COMPLETE == event) {
+    global_ota_component->update_chunk_wrote();
+  } else if (MGMT_EVT_OP_IMG_MGMT_DFU_PENDING == event) {
+    global_ota_component->update_pending();
+  } else if (MGMT_EVT_OP_IMG_MGMT_DFU_STOPPED == event) {
+    global_ota_component->update_stopped();
+  } else {
+    ESP_LOGD(TAG, "MCUmgr Image Management Event with the %d ID", u32_count_trailing_zeros(MGMT_EVT_GET_ID(event)));
+  }
+  return MGMT_CB_OK;
+}
+
+OTAComponent::OTAComponent() { global_ota_component = this; }
+
+void OTAComponent::setup() {
+  this->img_mgmt_callback_.callback = mcumgr_img_mgmt_cb;
+  this->img_mgmt_callback_.event_id = MGMT_EVT_OP_IMG_MGMT_ALL;
+  mgmt_callback_register(&this->img_mgmt_callback_);
+#ifdef CONFIG_USB_DEVICE_STACK
+  usb_enable(nullptr);
+#endif
+// Handle OTA rollback: mark partition valid immediately unless USE_OTA_ROLLBACK is enabled,
+// in which case safe_mode will mark it valid after confirming successful boot.
+#ifndef USE_OTA_ROLLBACK
+  if (!boot_is_img_confirmed()) {
+    boot_write_img_confirmed();
+  }
+#endif
+}
+
+#ifdef ESPHOME_LOG_HAS_CONFIG
+static const char *swap_type_str(uint8_t type) {
+  switch (type) {
+    case BOOT_SWAP_TYPE_NONE:
+      return "none";
+    case BOOT_SWAP_TYPE_TEST:
+      return "test";
+    case BOOT_SWAP_TYPE_PERM:
+      return "perm";
+    case BOOT_SWAP_TYPE_REVERT:
+      return "revert";
+    case BOOT_SWAP_TYPE_FAIL:
+      return "fail";
+  }
+
+  return "unknown";
+}
+#endif
+
+void OTAComponent::dump_config() {
+  ESP_LOGCONFIG(TAG,
+                "Over-The-Air Updates:\n"
+                "  swap type after reboot: %s\n"
+                "  image confirmed: %s",
+                swap_type_str(mcuboot_swap_type()), YESNO(boot_is_img_confirmed()));
+}
+
+void OTAComponent::update_chunk(const img_mgmt_upload_check &upload) {
+  float percentage = (upload.req->off * 100.0f) / upload.action->size;
+  this->defer([this, percentage]() { this->percentage_ = percentage; });
+}
+
+void OTAComponent::update_started() {
+  this->defer([this]() {
+    ESP_LOGD(TAG, "Starting update");
+#ifdef USE_OTA_STATE_LISTENER
+    this->notify_state_(ota::OTA_STARTED, 0.0f, 0);
+#endif
+  });
+}
+
+void OTAComponent::update_chunk_wrote() {
+  uint32_t now = millis();
+  if (now - this->last_progress_ > 1000) {
+    this->last_progress_ = now;
+    this->defer([this]() {
+      ESP_LOGD(TAG, "OTA in progress: %0.1f%%", this->percentage_);
+#ifdef USE_OTA_STATE_LISTENER
+      this->notify_state_(ota::OTA_IN_PROGRESS, this->percentage_, 0);
+#endif
+    });
+  }
+}
+
+void OTAComponent::update_pending() {
+  this->defer([this]() {
+    ESP_LOGD(TAG, "OTA pending");
+#ifdef USE_OTA_STATE_LISTENER
+    this->notify_state_(ota::OTA_COMPLETED, 100.0f, 0);
+#endif
+  });
+}
+
+void OTAComponent::update_stopped() {
+  this->defer([this]() {
+    ESP_LOGD(TAG, "OTA stopped");
+#ifdef USE_OTA_STATE_LISTENER
+    this->notify_state_(ota::OTA_ERROR, 0.0f, static_cast<uint8_t>(ota::OTA_RESPONSE_ERROR_UNKNOWN));
+#endif
+  });
+}
+
+}  // namespace esphome::zephyr_mcumgr
+#endif

esphome/components/zephyr_mcumgr/ota/ota_zephyr_mcumgr.h

@@ -0,0 +1,29 @@
+#pragma once
+#include "esphome/core/defines.h"
+#ifdef USE_ZEPHYR
+#include "esphome/components/ota/ota_backend.h"
+#include <zephyr/mgmt/mcumgr/mgmt/callbacks.h>
+
+struct img_mgmt_upload_check;
+
+namespace esphome::zephyr_mcumgr {
+
+class OTAComponent : public ota::OTAComponent {
+ public:
+  OTAComponent();
+  void setup() override;
+  void dump_config() override;
+  void update_chunk(const img_mgmt_upload_check &upload);
+  void update_started();
+  void update_chunk_wrote();
+  void update_pending();
+  void update_stopped();
+
+ protected:
+  uint32_t last_progress_ = 0;
+  float percentage_ = 0;
+  mgmt_callback img_mgmt_callback_{};
+};
+
+}  // namespace esphome::zephyr_mcumgr
+#endif

esphome/core/defines.h

@@ -368,6 +368,7 @@
 #define USE_NRF52_DFU
 #define USE_NRF52_REG0_VOUT 5
 #define USE_NRF52_UICR_ERASE
+#define USE_OTA_ROLLBACK
 #define USE_SOFTDEVICE_ID 7
 #define USE_SOFTDEVICE_VERSION 1
 #define USE_ZIGBEE

requirements.txt

@@ -12,7 +12,7 @@ platformio==6.1.19
 esptool==5.2.0
 click==8.3.1
 esphome-dashboard==20260210.0
-aioesphomeapi==44.4.0
+aioesphomeapi==44.5.0
 zeroconf==0.148.0
 puremagic==1.30
 ruamel.yaml==0.19.1 # dashboard_import
@@ -23,6 +23,7 @@ resvg-py==0.2.6
 freetype-py==2.5.1
 jinja2==3.1.6
 bleak==2.1.1
+smpclient==6.0.0
 requests==2.32.5
 
 # esp-idf >= 5.0 requires this

script/api_protobuf/api_protobuf.py

@@ -151,6 +151,11 @@ class TypeInfo(ABC):
         """Check if the field is repeated."""
         return self._field.label == FieldDescriptorProto.LABEL_REPEATED
 
+    @property
+    def force(self) -> bool:
+        """Check if this field should always be encoded (skip zero/empty check)."""
+        return get_field_opt(self._field, pb.force, False)
+
     @property
     def wire_type(self) -> WireType:
         """Get the wire type for the field."""
@@ -218,6 +223,8 @@ class TypeInfo(ABC):
 
     @property
     def encode_content(self) -> str:
+        if self.force:
+            return f"buffer.{self.encode_func}({self.number}, this->{self.field_name}, true);"
         return f"buffer.{self.encode_func}({self.number}, this->{self.field_name});"
 
     encode_func = None
@@ -413,6 +420,8 @@ class DoubleType(TypeInfo):
 
     def get_size_calculation(self, name: str, force: bool = False) -> str:
         field_id_size = self.calculate_field_id_size()
+        if force:
+            return f"size += {field_id_size + self.get_fixed_size_bytes()};"
         return f"size += ProtoSize::calc_fixed64({field_id_size}, {name});"
 
     def get_fixed_size_bytes(self) -> int:
@@ -437,6 +446,8 @@ class FloatType(TypeInfo):
 
     def get_size_calculation(self, name: str, force: bool = False) -> str:
         field_id_size = self.calculate_field_id_size()
+        if force:
+            return f"size += {field_id_size + self.get_fixed_size_bytes()};"
         return f"size += ProtoSize::calc_float({field_id_size}, {name});"
 
     def get_fixed_size_bytes(self) -> int:
@@ -521,6 +532,8 @@ class Fixed64Type(TypeInfo):
 
     def get_size_calculation(self, name: str, force: bool = False) -> str:
         field_id_size = self.calculate_field_id_size()
+        if force:
+            return f"size += {field_id_size + self.get_fixed_size_bytes()};"
         return f"size += ProtoSize::calc_fixed64({field_id_size}, {name});"
 
     def get_fixed_size_bytes(self) -> int:
@@ -545,6 +558,8 @@ class Fixed32Type(TypeInfo):
 
     def get_size_calculation(self, name: str, force: bool = False) -> str:
         field_id_size = self.calculate_field_id_size()
+        if force:
+            return f"size += {field_id_size + self.get_fixed_size_bytes()};"
         return f"size += ProtoSize::calc_fixed32({field_id_size}, {name});"
 
     def get_fixed_size_bytes(self) -> int:
@@ -607,6 +622,8 @@ class StringType(TypeInfo):
     @property
     def encode_content(self) -> str:
         # Use the StringRef
+        if self.force:
+            return f"buffer.encode_string({self.number}, this->{self.field_name}_ref_, true);"
         return f"buffer.encode_string({self.number}, this->{self.field_name}_ref_);"
 
     def dump(self, name):
@@ -694,7 +711,7 @@ class MessageType(TypeInfo):
 
     @property
     def encode_content(self) -> str:
-        # Singular message fields skip encoding when empty
+        # encode_sub_message always encodes (uses backpatch), no force needed
         return f"buffer.{self.encode_func}({self.number}, this->{self.field_name});"
 
     @property
@@ -771,6 +788,8 @@ class BytesType(TypeInfo):
 
     @property
     def encode_content(self) -> str:
+        if self.force:
+            return f"buffer.encode_bytes({self.number}, this->{self.field_name}_ptr_, this->{self.field_name}_len_, true);"
         return f"buffer.encode_bytes({self.number}, this->{self.field_name}_ptr_, this->{self.field_name}_len_);"
 
     def dump(self, name: str) -> str:
@@ -876,6 +895,8 @@ class PointerToBytesBufferType(PointerToBufferTypeBase):
 
     @property
     def encode_content(self) -> str:
+        if self.force:
+            return f"buffer.encode_bytes({self.number}, this->{self.field_name}, this->{self.field_name}_len, true);"
         return f"buffer.encode_bytes({self.number}, this->{self.field_name}, this->{self.field_name}_len);"
 
     @property
@@ -923,6 +944,10 @@ class PointerToStringBufferType(PointerToBufferTypeBase):
 
     @property
     def encode_content(self) -> str:
+        if self.force:
+            return (
+                f"buffer.encode_string({self.number}, this->{self.field_name}, true);"
+            )
         return f"buffer.encode_string({self.number}, this->{self.field_name});"
 
     @property
@@ -1086,6 +1111,8 @@ class FixedArrayBytesType(TypeInfo):
 
     @property
     def encode_content(self) -> str:
+        if self.force:
+            return f"buffer.encode_bytes({self.number}, this->{self.field_name}, this->{self.field_name}_len, true);"
         return f"buffer.encode_bytes({self.number}, this->{self.field_name}, this->{self.field_name}_len);"
 
     def dump(self, name: str) -> str:
@@ -1159,6 +1186,8 @@ class EnumType(TypeInfo):
 
     @property
     def encode_content(self) -> str:
+        if self.force:
+            return f"buffer.{self.encode_func}({self.number}, static_cast<uint32_t>(this->{self.field_name}), true);"
         return f"buffer.{self.encode_func}({self.number}, static_cast<uint32_t>(this->{self.field_name}));"
 
     def dump(self, name: str) -> str:
@@ -1192,6 +1221,8 @@ class SFixed32Type(TypeInfo):
 
     def get_size_calculation(self, name: str, force: bool = False) -> str:
         field_id_size = self.calculate_field_id_size()
+        if force:
+            return f"size += {field_id_size + self.get_fixed_size_bytes()};"
         return f"size += ProtoSize::calc_sfixed32({field_id_size}, {name});"
 
     def get_fixed_size_bytes(self) -> int:
@@ -1216,6 +1247,8 @@ class SFixed64Type(TypeInfo):
 
     def get_size_calculation(self, name: str, force: bool = False) -> str:
         field_id_size = self.calculate_field_id_size()
+        if force:
+            return f"size += {field_id_size + self.get_fixed_size_bytes()};"
         return f"size += ProtoSize::calc_sfixed64({field_id_size}, {name});"
 
     def get_fixed_size_bytes(self) -> int:
@@ -2134,7 +2167,8 @@ def build_message_type(
             encode.extend(wrap_with_ifdef(ti.encode_content, field_ifdef))
             size_calc.extend(
                 wrap_with_ifdef(
-                    ti.get_size_calculation(f"this->{ti.field_name}"), field_ifdef
+                    ti.get_size_calculation(f"this->{ti.field_name}", ti.force),
+                    field_ifdef,
                 )
             )
 

script/helpers_zephyr.py

@@ -28,6 +28,22 @@ extern "C" void zboss_signal_handler() {};
 CONFIG_NEWLIB_LIBC=y
 CONFIG_BT=y
 CONFIG_ADC=y
+#mcumgr begin
+CONFIG_NET_BUF=y
+CONFIG_ZCBOR=y
+CONFIG_MCUMGR=y
+CONFIG_MCUMGR_GRP_IMG=y
+CONFIG_IMG_MANAGER=y
+CONFIG_STREAM_FLASH=y
+CONFIG_FLASH_MAP=y
+CONFIG_FLASH=y
+CONFIG_IMG_ERASE_PROGRESSIVELY=y
+CONFIG_BOOTLOADER_MCUBOOT=y
+CONFIG_MCUMGR_MGMT_NOTIFICATION_HOOKS=y
+CONFIG_MCUMGR_GRP_IMG_STATUS_HOOKS=y
+CONFIG_MCUMGR_GRP_IMG_UPLOAD_CHECK_HOOK=y
+CONFIG_MCUMGR_TRANSPORT_UART=y
+#mcumgr end
 #zigbee begin
 CONFIG_ZIGBEE=y
 CONFIG_CRYPTO=y

tests/components/ld2450/common.h

@@ -16,7 +16,7 @@ class MockUARTComponent : public uart::UARTComponent {
   MOCK_METHOD(bool, read_array, (uint8_t * data, size_t len), (override));
   MOCK_METHOD(bool, peek_byte, (uint8_t * data), (override));
   MOCK_METHOD(size_t, available, (), (override));
-  MOCK_METHOD(void, flush, (), (override));
+  MOCK_METHOD(uart::FlushResult, flush, (), (override));
   MOCK_METHOD(void, check_logger_conflict, (), (override));
 };
 

tests/components/ota/test.nrf52-mcumgr.yaml

@@ -0,0 +1,27 @@
+zephyr_ble_server:
+
+ota:
+  - platform: zephyr_mcumgr
+    transport:
+      ble: true
+      hardware_uart: CDC
+    on_begin:
+      then:
+        - logger.log: "OTA start"
+    on_progress:
+      then:
+        - logger.log:
+            format: "OTA progress %0.1f%%"
+            args: ["x"]
+    on_end:
+      then:
+        - logger.log: "OTA end"
+    on_error:
+      then:
+        - logger.log:
+            format: "OTA update error %d"
+            args: ["x"]
+    on_state_change:
+      then:
+        lambda: >-
+          ESP_LOGD("ota", "State %d", state);

tests/components/uart/common.h

@@ -30,7 +30,7 @@ class MockUARTComponent : public UARTComponent {
   MOCK_METHOD(bool, read_array, (uint8_t * data, size_t len), (override));
   MOCK_METHOD(bool, peek_byte, (uint8_t * data), (override));
   MOCK_METHOD(size_t, available, (), (override));
-  MOCK_METHOD(void, flush, (), (override));
+  MOCK_METHOD(FlushResult, flush, (), (override));
   MOCK_METHOD(void, check_logger_conflict, (), (override));
 };
 

tests/integration/fixtures/external_components/uart_mock/__init__.py

@@ -62,6 +62,7 @@ CONFIG_INJECT_RX_SCHEMA = cv.maybe_simple_value(
     {
         cv.GenerateID(): cv.use_id(MockUartComponent),
         cv.Required("data"): cv.templatable(validate_raw_data),
+        cv.Optional(CONF_DELAY): cv.positive_time_period_milliseconds,
     },
     key=CONF_DATA,
 )
@@ -87,7 +88,7 @@ CONFIG_SCHEMA = cv.Schema(
         cv.GenerateID(): cv.declare_id(MockUartComponent),
         cv.Required(CONF_BAUD_RATE): cv.int_range(min=1),
         cv.Optional(CONF_RX_BUFFER_SIZE, default=256): cv.validate_bytes,
-        cv.Optional(CONF_RX_FULL_THRESHOLD, default=10): cv.int_range(min=1, max=120),
+        cv.Optional(CONF_RX_FULL_THRESHOLD): cv.int_range(min=1, max=120),
         cv.Optional(CONF_RX_TIMEOUT, default=2): cv.int_range(min=0, max=92),
         cv.Optional(CONF_STOP_BITS, default=1): cv.one_of(1, 2, int=True),
         cv.Optional(CONF_DATA_BITS, default=8): cv.int_range(min=5, max=8),
@@ -126,6 +127,8 @@ async def inject_rx_to_code(config, action_id, template_arg, args):
         arr_id = ID(f"{action_id}_data", is_declaration=True, type=cg.uint8)
         arr = cg.static_const_array(arr_id, cg.ArrayInitializer(*data))
         cg.add(var.set_data_static(arr, len(data)))
+    if CONF_DELAY in config:
+        cg.add(var.set_delay(config[CONF_DELAY]))
     return var
 
 
@@ -135,7 +138,8 @@ async def to_code(config):
 
     cg.add(var.set_baud_rate(config[CONF_BAUD_RATE]))
     cg.add(var.set_rx_buffer_size(config[CONF_RX_BUFFER_SIZE]))
-    cg.add(var.set_rx_full_threshold(config[CONF_RX_FULL_THRESHOLD]))
+    if CONF_RX_FULL_THRESHOLD in config:
+        cg.add(var.set_rx_full_threshold(config[CONF_RX_FULL_THRESHOLD]))
     cg.add(var.set_rx_timeout(config[CONF_RX_TIMEOUT]))
     cg.add(var.set_stop_bits(config[CONF_STOP_BITS]))
     cg.add(var.set_data_bits(config[CONF_DATA_BITS]))

tests/integration/fixtures/external_components/uart_mock/automation.h

@@ -22,18 +22,30 @@ template<typename... Ts> class MockUartInjectRXAction : public Action<Ts...>, pu
     this->len_ = len;  // Length >= 0 indicates static mode
   }
 
+  void set_delay(uint32_t delay_ms) { this->delay_ms_ = delay_ms; }
+
   void play(const Ts &...x) override {
     if (this->len_ >= 0) {
       // Static mode: use pointer and length
-      this->parent_->inject_to_rx_buffer(this->code_.data, static_cast<size_t>(this->len_));
+      if (this->delay_ms_ > 0) {
+        std::vector<uint8_t> data(this->code_.data, this->code_.data + this->len_);
+        this->parent_->inject_to_rx_buffer_delayed(data, this->delay_ms_);
+      } else {
+        this->parent_->inject_to_rx_buffer(this->code_.data, static_cast<size_t>(this->len_));
+      }
     } else {
       // Template mode: call function
       auto val = this->code_.func(x...);
-      this->parent_->inject_to_rx_buffer(val);
+      if (this->delay_ms_ > 0) {
+        this->parent_->inject_to_rx_buffer_delayed(val, this->delay_ms_);
+      } else {
+        this->parent_->inject_to_rx_buffer(val);
+      }
     }
   }
 
  protected:
+  uint32_t delay_ms_{0};
   ssize_t len_{-1};  // -1 = template mode, >=0 = static mode with length
   union Code {
     std::vector<uint8_t> (*func)(Ts...);  // Function pointer (stateless lambdas)

tests/integration/fixtures/external_components/uart_mock/uart_mock.cpp

@@ -53,6 +53,15 @@ void MockUartComponent::loop() {
     }
   }
 
+  // Process staged RX - deliver bytes whose delay has elapsed
+  uint32_t now_ms = millis();
+  while (!this->staged_rx_.empty() && (static_cast<int32_t>(now_ms - this->staged_rx_.front().available_at_ms) >= 0)) {
+    auto &staged = this->staged_rx_.front();
+    ESP_LOGD(TAG, "Delivering %zu staged RX bytes", staged.data.size());
+    this->inject_to_rx_buffer(staged.data);
+    this->staged_rx_.pop_front();
+  }
+
   // Process delayed responses
   for (auto &response : this->responses_) {
     if (response.delay_ms > 0 && response.last_match_ms > 0 && now - response.last_match_ms >= response.delay_ms) {
@@ -144,8 +153,9 @@ bool MockUartComponent::read_array(uint8_t *data, size_t len) {
 
 size_t MockUartComponent::available() { return this->rx_buffer_.size(); }
 
-void MockUartComponent::flush() {
+uart::FlushResult MockUartComponent::flush() {
   // Nothing to flush in mock
+  return uart::FlushResult::ASSUMED_SUCCESS;
 }
 
 void MockUartComponent::set_rx_full_threshold(size_t rx_full_threshold) {
@@ -208,4 +218,15 @@ void MockUartComponent::inject_to_rx_buffer(const std::vector<uint8_t> &data) {
   }
 }
 
+void MockUartComponent::inject_to_rx_buffer_delayed(const std::vector<uint8_t> &data, uint32_t delay_ms) {
+  if (!data.empty() && data.size() <= 64) {
+    char hex_buf[format_hex_pretty_size(64)];
+    ESP_LOGD(TAG, "Staging %zu RX bytes with %ums delay: %s", data.size(), delay_ms,
+             format_hex_pretty_to(hex_buf, sizeof(hex_buf), data.data(), data.size()));
+  } else if (data.size() > 64) {
+    ESP_LOGD(TAG, "Staging %zu RX bytes with %ums delay (too large to log inline)", data.size(), delay_ms);
+  }
+  this->staged_rx_.push_back({data, millis() + delay_ms});
+}
+
 }  // namespace esphome::uart_mock

tests/integration/fixtures/external_components/uart_mock/uart_mock.h

@@ -28,7 +28,7 @@ class MockUartComponent : public uart::UARTComponent, public Component {
   bool peek_byte(uint8_t *data) override;
   bool read_array(uint8_t *data, size_t len) override;
   size_t available() override;
-  void flush() override;
+  uart::FlushResult flush() override;
   void set_rx_full_threshold(size_t rx_full_threshold) override;
   void set_rx_timeout(size_t rx_timeout) override;
 
@@ -43,6 +43,8 @@ class MockUartComponent : public uart::UARTComponent, public Component {
   void set_tx_hook(std::function<void(const std::vector<uint8_t> &)> &&cb) { this->tx_hook_ = std::move(cb); }
   void inject_to_rx_buffer(const std::vector<uint8_t> &data);
   void inject_to_rx_buffer(const uint8_t *data, size_t len);
+  // Stage bytes for delayed delivery - simulates transport-level latency (e.g., USB packets)
+  void inject_to_rx_buffer_delayed(const std::vector<uint8_t> &data, uint32_t delay_ms);
 
  protected:
   void check_logger_conflict() override {}
@@ -82,6 +84,14 @@ class MockUartComponent : public uart::UARTComponent, public Component {
   };
   std::vector<PeriodicRx> periodic_rx_;
 
+  // Staged RX - bytes that are pending delivery after a delay
+  // Simulates transport-level latency (e.g., USB packet delivery)
+  struct StagedRx {
+    std::vector<uint8_t> data;
+    uint32_t available_at_ms;  // millis() time when bytes become available
+  };
+  std::deque<StagedRx> staged_rx_;
+
   // Observability
   uint32_t tx_count_{0};
   uint32_t rx_count_{0};

tests/integration/fixtures/uart_mock_ld2450.yaml

@@ -0,0 +1,213 @@
+esphome:
+  name: uart-mock-ld2450-test
+
+host:
+api:
+logger:
+  level: VERBOSE
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+
+# Dummy uart entry to satisfy ld2450's DEPENDENCIES = ["uart"]
+# The actual UART bus used is the uart_mock component below
+uart:
+  baud_rate: 115200
+  port: /dev/null
+
+uart_mock:
+  id: mock_uart
+  baud_rate: 256000
+  auto_start: false
+  responses:
+    # Catch-all response: match any command footer (04 03 02 01).
+    # Returns a generic ACK to unblock setup commands.
+    #
+    # Response layout:
+    #   [0-3]  FD FC FB FA = header
+    #   [4-5]  04 00       = length 4
+    #   [6]    FF          = cmd (handled as CMD_ENABLE_CONF)
+    #   [7]    01          = status (ACK)
+    #   [8-9]  00 00       = error = 0
+    #   [10-13] 04 03 02 01 = footer
+    - expect_tx: [0x04, 0x03, 0x02, 0x01]
+      inject_rx:
+        [
+          0xFD, 0xFC, 0xFB, 0xFA,
+          0x04, 0x00,
+          0xFF, 0x01,
+          0x00, 0x00,
+          0x04, 0x03, 0x02, 0x01,
+        ]
+
+  injections:
+    # Phase 1 (t=100ms): Valid LD2450 periodic data frame - happy path
+    # The buffer is clean at this point, so this frame should parse correctly.
+    #
+    # Target 1: X=-500mm, Y=1000mm, Speed=-50mm/s (approaching), Res=320mm
+    #   X: magnitude=500 (0x01F4), negative → high=0x01, low=0xF4
+    #   Y: magnitude=1000 (0x03E8), positive → high=0x83, low=0xE8
+    #   Speed: raw=5, negative (approaching) → high=0x00, low=0x05, decoded=-50mm/s
+    #   Resolution: 320 → low=0x40, high=0x01
+    #   Distance: sqrt(500²+1000²) = sqrt(1250000) ≈ 1118mm
+    #
+    # Target 2: X=200mm, Y=500mm, Speed=0 (stationary), Res=100mm
+    #   X: magnitude=200 (0x00C8), positive → high=0x80, low=0xC8
+    #   Y: magnitude=500 (0x01F4), positive → high=0x81, low=0xF4
+    #   Speed: 0 → 0x00, 0x00
+    #   Resolution: 100 → low=0x64, high=0x00
+    #   Distance: sqrt(200²+500²) = sqrt(290000) ≈ 538mm
+    #
+    # Target 3: No target (all zeros)
+    #   Distance: 0 → sensors publish unknown/NaN
+    #
+    # Counts: target_count=2, moving_target_count=1, still_target_count=1
+    #
+    # Frame layout (30 bytes):
+    #   [0-3]   AA FF 03 00 = periodic data header
+    #   [4-11]  Target 1 (8 bytes): X_L X_H Y_L Y_H SPD_L SPD_H RES_L RES_H
+    #   [12-19] Target 2 (8 bytes)
+    #   [20-27] Target 3 (8 bytes)
+    #   [28-29] 55 CC = periodic data footer
+    - delay: 100ms
+      inject_rx:
+        [
+          0xAA, 0xFF, 0x03, 0x00,
+          0xF4, 0x01, 0xE8, 0x83, 0x05, 0x00, 0x40, 0x01,
+          0xC8, 0x80, 0xF4, 0x81, 0x00, 0x00, 0x64, 0x00,
+          0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+          0x55, 0xCC,
+        ]
+
+    # Phase 2 (t=300ms): Garbage bytes
+    # LD2450's readline_ does NOT reject bytes at position 0 (unlike LD2412),
+    # so these bytes accumulate in the buffer. buffer_pos_ goes from 0 to 7.
+    - delay: 200ms
+      inject_rx: [0xDE, 0xAD, 0xBE, 0xEF, 0x00, 0x11, 0x22]
+
+    # Phase 3 (t=400ms): Truncated frame (header + partial data, no footer)
+    # More bytes accumulating in the buffer without a footer match.
+    # After this, buffer_pos_ = 7 + 8 = 15.
+    - delay: 100ms
+      inject_rx: [0xAA, 0xFF, 0x03, 0x00, 0x01, 0x02, 0x03, 0x04]
+
+    # Phase 4 (t=600ms): Overflow - inject 50 bytes of 0xFF (MAX_LINE_LENGTH=45)
+    # Buffer has 15 bytes from phases 2+3.
+    # Overflow math: need 29 bytes to fill positions 15-43 (buffer_pos_=44),
+    # then byte 30 at position 44 = MAX_LINE_LENGTH-1 triggers overflow.
+    # After overflow, buffer_pos_ = 0. Remaining 0xFF bytes accumulate
+    # until another overflow or until valid data arrives.
+    - delay: 200ms
+      inject_rx:
+        [
+          0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
+          0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
+          0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
+          0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
+          0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
+        ]
+
+    # Phase 5 (t=700ms): Valid frame after overflow - recovery test
+    # Buffer was reset by overflow. This valid frame should parse correctly.
+    #
+    # Target 1: X=300mm, Y=400mm, Speed=30mm/s (moving away), Res=100mm
+    #   X: magnitude=300 (0x012C), positive → high=0x81, low=0x2C
+    #   Y: magnitude=400 (0x0190), positive → high=0x81, low=0x90
+    #   Speed: raw=3, positive (moving away) → high=0x80, low=0x03, decoded=30mm/s
+    #   Resolution: 100 → low=0x64, high=0x00
+    #   Distance: sqrt(300²+400²) = 500mm
+    #
+    # Target 2 & 3: No target (all zeros)
+    # Counts: target_count=1, moving_target_count=1, still_target_count=0
+    - delay: 100ms
+      inject_rx:
+        [
+          0xAA, 0xFF, 0x03, 0x00,
+          0x2C, 0x81, 0x90, 0x81, 0x03, 0x80, 0x64, 0x00,
+          0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+          0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+          0x55, 0xCC,
+        ]
+
+ld2450:
+  id: ld2450_dev
+  uart_id: mock_uart
+
+sensor:
+  - platform: ld2450
+    ld2450_id: ld2450_dev
+    target_count:
+      name: "Target Count"
+      filters: &sensor_filters
+        - timeout:
+            timeout: 50ms
+            value: last
+        - throttle_with_priority: 50ms
+    still_target_count:
+      name: "Still Target Count"
+      filters: *sensor_filters
+    moving_target_count:
+      name: "Moving Target Count"
+      filters: *sensor_filters
+    target_1:
+      x:
+        name: "Target 1 X"
+        filters: *sensor_filters
+      y:
+        name: "Target 1 Y"
+        filters: *sensor_filters
+      speed:
+        name: "Target 1 Speed"
+        filters: *sensor_filters
+      distance:
+        name: "Target 1 Distance"
+        filters: *sensor_filters
+      resolution:
+        name: "Target 1 Resolution"
+        filters: *sensor_filters
+      angle:
+        name: "Target 1 Angle"
+        filters: *sensor_filters
+    target_2:
+      x:
+        name: "Target 2 X"
+        filters: *sensor_filters
+      y:
+        name: "Target 2 Y"
+        filters: *sensor_filters
+      speed:
+        name: "Target 2 Speed"
+        filters: *sensor_filters
+      distance:
+        name: "Target 2 Distance"
+        filters: *sensor_filters
+
+binary_sensor:
+  - platform: ld2450
+    ld2450_id: ld2450_dev
+    has_target:
+      name: "Has Target"
+      filters: &binary_sensor_filters
+        - settle: 50ms
+    has_moving_target:
+      name: "Has Moving Target"
+      filters: *binary_sensor_filters
+    has_still_target:
+      name: "Has Still Target"
+      filters: *binary_sensor_filters
+
+text_sensor:
+  - platform: ld2450
+    ld2450_id: ld2450_dev
+    target_1:
+      direction:
+        name: "Target 1 Direction"
+
+button:
+  - platform: template
+    name: "Start Scenario"
+    id: start_scenario_btn
+    on_press:
+      - lambda: 'id(mock_uart).start_scenario();'

tests/integration/fixtures/uart_mock_modbus_no_threshold.yaml

@@ -0,0 +1,64 @@
+esphome:
+  name: uart-mock-modbus-no-thresh
+
+host:
+api:
+logger:
+  level: VERBOSE
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+
+# Dummy uart entry to satisfy modbus's DEPENDENCIES = ["uart"]
+# The actual UART bus used is the uart_mock component below
+uart:
+  baud_rate: 115200
+  port: /dev/null
+
+# Simulate a non-hardware UART (e.g., USB UART) by not setting rx_full_threshold.
+# This leaves it at the default sentinel value (0), triggering the 50ms fallback timeout.
+uart_mock:
+  - id: virtual_uart_dev
+    baud_rate: 9600
+    auto_start: false
+    debug:
+    on_tx:
+      - then:
+          - if:
+              condition: #Read 80 input registers on device 2, starting at address 0 (SDM meter request)
+                lambda: "return data == std::vector<uint8_t>({0x02,0x04,0x00,0x00,0x00,0x50,0xF0,0x05});"
+              then:
+                - uart_mock.inject_rx: # First USB packet: SDM meter response part 1
+                    !lambda return {0x02,0x04,0xA0,0x43,0x73,0x19,0x9A,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,
+                    0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,
+                    0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,
+                    0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x3F,0x80,0x00,0x00,0x00,0x00,0x00,0x00,0x00,
+                    0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,
+                    0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,
+                    0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00};
+                - uart_mock.inject_rx: # Second USB packet: rest of response (staged with 40ms latency)
+                    delay: 40ms
+                    data: !lambda return{0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x00,
+                      0x00,0x00,0x00,0x00,0x00,0x00,0x00,0x42,0x6F,0xCC,0xCD,0x43,0x7C,0xB8,0x10,0x3D,0x38,0x51,0xEC,
+                      0x43,0x81,0x1B,0xE7,0x3B,0x03,0x12,0x6F,0x50,0x1B};
+
+modbus:
+  uart_id: virtual_uart_dev
+  turnaround_time: 10ms
+
+sensor:
+  - platform: sdm_meter
+    address: 2
+    update_interval: 1s
+    phase_a:
+      voltage:
+        name: sdm_voltage
+
+button:
+  - platform: template
+    name: "Start Scenario"
+    id: start_scenario_btn
+    on_press:
+      - lambda: 'id(virtual_uart_dev).start_scenario();'

tests/integration/state_utils.py

@@ -13,6 +13,7 @@ from aioesphomeapi import (
     EntityInfo,
     EntityState,
     SensorState,
+    TextSensorState,
 )
 
 _LOGGER = logging.getLogger(__name__)
@@ -244,12 +245,13 @@ class InitialStateHelper:
 
 
 class SensorStateCollector:
-    """Collects sensor and binary sensor state updates and provides wait helpers.
+    """Collects sensor, binary sensor, and text sensor state updates with wait helpers.
 
     Usage:
         collector = SensorStateCollector(
             sensor_names=["moving_distance", "still_distance"],
             binary_sensor_names=["has_target"],
+            text_sensor_names=["direction"],
         )
         # Use collector.on_state as the callback (or wrap it)
         client.subscribe_states(helper.on_state_wrapper(collector.on_state))
@@ -259,18 +261,23 @@ class SensorStateCollector:
 
         # Access collected states
         assert collector.sensor_states["moving_distance"][0] == approx(100.0)
+        assert collector.text_sensor_states["direction"][0] == "Approaching"
     """
 
     def __init__(
         self,
         sensor_names: list[str],
         binary_sensor_names: list[str] | None = None,
+        text_sensor_names: list[str] | None = None,
         entities: list[EntityInfo] | None = None,
     ) -> None:
         self.sensor_states: dict[str, list[float]] = {name: [] for name in sensor_names}
         self.binary_states: dict[str, list[bool]] = {
             name: [] for name in (binary_sensor_names or [])
         }
+        self.text_sensor_states: dict[str, list[str]] = {
+            name: [] for name in (text_sensor_names or [])
+        }
         self._key_to_sensor: dict[int, str] = {}
         self._waiters: list[tuple[Callable[[], bool], asyncio.Future[bool]]] = []
 
@@ -279,7 +286,11 @@ class SensorStateCollector:
 
     def build_key_mapping(self, entities: list[EntityInfo]) -> None:
         """Build key-to-name mapping from entities. Sorted by descending length."""
-        all_names = list(self.sensor_states.keys()) + list(self.binary_states.keys())
+        all_names = (
+            list(self.sensor_states.keys())
+            + list(self.binary_states.keys())
+            + list(self.text_sensor_states.keys())
+        )
         all_names.sort(key=len, reverse=True)
         self._key_to_sensor = build_key_to_entity_mapping(entities, all_names)
 
@@ -295,6 +306,11 @@ class SensorStateCollector:
             if sensor_name and sensor_name in self.binary_states:
                 self.binary_states[sensor_name].append(state.state)
                 self._check_waiters()
+        elif isinstance(state, TextSensorState) and not state.missing_state:
+            sensor_name = self._key_to_sensor.get(state.key)
+            if sensor_name and sensor_name in self.text_sensor_states:
+                self.text_sensor_states[sensor_name].append(state.state)
+                self._check_waiters()
 
     def _check_waiters(self) -> None:
         """Check all pending waiters and resolve any whose condition is met."""
@@ -303,9 +319,11 @@ class SensorStateCollector:
                 future.set_result(True)
 
     def _all_have_values(self) -> bool:
-        """Check if all sensor and binary sensor lists have at least one value."""
-        return all(len(v) >= 1 for v in self.sensor_states.values()) and all(
-            len(v) >= 1 for v in self.binary_states.values()
+        """Check if all sensor, binary sensor, and text sensor lists have at least one value."""
+        return (
+            all(len(v) >= 1 for v in self.sensor_states.values())
+            and all(len(v) >= 1 for v in self.binary_states.values())
+            and all(len(v) >= 1 for v in self.text_sensor_states.values())
         )
 
     async def wait_for_all(self, timeout: float = 3.0) -> None:

tests/integration/test_uart_mock_ld2450.py

@@ -0,0 +1,214 @@
+"""Integration test for LD2450 component with mock UART.
+
+Tests:
+test_uart_mock_ld2450:
+  1. Happy path - valid periodic data frame publishes correct target sensor values
+  2. Multi-target tracking - verifies target count, moving/still counts
+  3. Target coordinate decoding - signed X/Y coordinates with sign-magnitude encoding
+  4. Speed decoding - approaching (negative) and stationary (zero) targets
+  5. Distance calculation - computed from X/Y via sqrt(x²+y²)
+  6. Direction text sensor - "Approaching" for negative speed target
+  7. Garbage resilience - random bytes don't crash the component
+  8. Truncated frame handling - partial frame doesn't corrupt state
+  9. Buffer overflow recovery - overflow resets the parser
+  10. Post-overflow parsing - next valid frame after overflow is parsed correctly
+  11. TX logging - verifies LD2450 sends expected setup commands
+"""
+
+from __future__ import annotations
+
+import asyncio
+from pathlib import Path
+
+from aioesphomeapi import ButtonInfo
+import pytest
+
+from .state_utils import InitialStateHelper, SensorStateCollector, find_entity
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_uart_mock_ld2450(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test LD2450 data parsing with happy path, garbage, overflow, and recovery."""
+    # Replace external component path placeholder
+    external_components_path = str(
+        Path(__file__).parent / "fixtures" / "external_components"
+    )
+    yaml_config = yaml_config.replace(
+        "EXTERNAL_COMPONENT_PATH", external_components_path
+    )
+
+    loop = asyncio.get_running_loop()
+
+    # Track overflow warning in logs
+    overflow_seen = loop.create_future()
+
+    # Track TX data logged by the mock for assertions
+    tx_log_lines: list[str] = []
+
+    def line_callback(line: str) -> None:
+        if "Max command length exceeded" in line and not overflow_seen.done():
+            overflow_seen.set_result(True)
+        # Capture all TX log lines from uart_mock
+        if "uart_mock" in line and "TX " in line:
+            tx_log_lines.append(line)
+
+    collector = SensorStateCollector(
+        sensor_names=[
+            "target_1_x",
+            "target_1_y",
+            "target_1_speed",
+            "target_1_distance",
+            "target_1_resolution",
+            "target_1_angle",
+            "target_2_x",
+            "target_2_y",
+            "target_2_speed",
+            "target_2_distance",
+            "target_count",
+            "still_target_count",
+            "moving_target_count",
+        ],
+        binary_sensor_names=[
+            "has_target",
+            "has_moving_target",
+            "has_still_target",
+        ],
+        text_sensor_names=[
+            "target_1_direction",
+        ],
+    )
+
+    # Signal when we see recovery frame values (target 1 distance ≈ 500mm)
+    recovery_received = collector.add_waiter(
+        lambda: (
+            pytest.approx(500.0, abs=1.0)
+            in collector.sensor_states["target_1_distance"]
+        )
+    )
+
+    async with (
+        run_compiled(yaml_config, line_callback=line_callback),
+        api_client_connected() as client,
+    ):
+        entities, _ = await client.list_entities_services()
+        collector.build_key_mapping(entities)
+
+        # Set up initial state helper
+        initial_state_helper = InitialStateHelper(entities)
+        client.subscribe_states(
+            initial_state_helper.on_state_wrapper(collector.on_state)
+        )
+
+        try:
+            await initial_state_helper.wait_for_initial_states()
+        except TimeoutError:
+            pytest.fail("Timeout waiting for initial states")
+
+        # Start the UART mock scenario now that we're subscribed
+        start_btn = find_entity(entities, "start_scenario", ButtonInfo)
+        assert start_btn is not None, "Start Scenario button not found"
+        client.button_command(start_btn.key)
+
+        # Wait for Phase 1 - all sensors and binary sensors have at least one value
+        try:
+            await collector.wait_for_all(timeout=5.0)
+        except TimeoutError:
+            pytest.fail(
+                f"Timeout waiting for Phase 1 frame. Received:\n"
+                f"  sensor_states: {collector.sensor_states}\n"
+                f"  binary_states: {collector.binary_states}\n"
+                f"  text_states: {collector.text_sensor_states}"
+            )
+
+        # Phase 1 values:
+        # Target 1: X=-500, Y=1000, Speed=-50 (approaching), Res=320
+        #   Distance = sqrt(500²+1000²) ≈ 1118mm
+        assert collector.sensor_states["target_1_x"][0] == pytest.approx(-500.0)
+        assert collector.sensor_states["target_1_y"][0] == pytest.approx(1000.0)
+        assert collector.sensor_states["target_1_speed"][0] == pytest.approx(-50.0)
+        assert collector.sensor_states["target_1_resolution"][0] == pytest.approx(320.0)
+        # Distance computed from X/Y
+        assert collector.sensor_states["target_1_distance"][0] == pytest.approx(
+            1118.0, abs=1.0
+        )
+
+        # Target 2: X=200, Y=500, Speed=0 (stationary), Res=100
+        #   Distance = sqrt(200²+500²) ≈ 538mm
+        assert collector.sensor_states["target_2_x"][0] == pytest.approx(200.0)
+        assert collector.sensor_states["target_2_y"][0] == pytest.approx(500.0)
+        assert collector.sensor_states["target_2_speed"][0] == pytest.approx(0.0)
+        assert collector.sensor_states["target_2_distance"][0] == pytest.approx(
+            538.0, abs=1.0
+        )
+
+        # Target counts: 2 targets total, 1 moving, 1 still
+        assert collector.sensor_states["target_count"][0] == pytest.approx(2.0)
+        assert collector.sensor_states["moving_target_count"][0] == pytest.approx(1.0)
+        assert collector.sensor_states["still_target_count"][0] == pytest.approx(1.0)
+
+        # Binary sensors: all true (targets detected)
+        assert collector.binary_states["has_target"][0] is True
+        assert collector.binary_states["has_moving_target"][0] is True
+        assert collector.binary_states["has_still_target"][0] is True
+
+        # Direction text sensor: Target 1 is approaching (speed < 0)
+        assert collector.text_sensor_states["target_1_direction"][0] == "Approaching"
+
+        # Wait for the recovery frame (Phase 5) to be parsed
+        # This proves the component survived garbage + truncated + overflow
+        try:
+            await asyncio.wait_for(recovery_received, timeout=5.0)
+        except TimeoutError:
+            pytest.fail(
+                f"Timeout waiting for recovery frame. Received:\n"
+                f"  sensor_states: {collector.sensor_states}"
+            )
+
+        # Verify overflow warning was logged
+        assert overflow_seen.done(), (
+            "Expected 'Max command length exceeded' warning in logs"
+        )
+
+        # Verify LD2450 sent setup commands (TX logging)
+        assert len(tx_log_lines) > 0, "Expected TX log lines from uart_mock"
+        tx_data = " ".join(tx_log_lines)
+        # Verify command frame header appears (FD:FC:FB:FA)
+        assert "FD:FC:FB:FA" in tx_data, (
+            "Expected LD2450 command frame header FD:FC:FB:FA in TX log"
+        )
+        # Verify command frame footer appears (04:03:02:01)
+        assert "04:03:02:01" in tx_data, (
+            "Expected LD2450 command frame footer 04:03:02:01 in TX log"
+        )
+
+        # Recovery frame values (Phase 5, after overflow):
+        # Target 1: X=300, Y=400, Distance=500, Speed=30 (moving away)
+        # target_count=1, moving=1, still=0
+        recovery_idx = next(
+            i
+            for i, v in enumerate(collector.sensor_states["target_1_distance"])
+            if v == pytest.approx(500.0, abs=1.0)
+        )
+        assert collector.sensor_states["target_1_x"][recovery_idx] == pytest.approx(
+            300.0
+        )
+        assert collector.sensor_states["target_1_y"][recovery_idx] == pytest.approx(
+            400.0
+        )
+        assert collector.sensor_states["target_1_speed"][recovery_idx] == pytest.approx(
+            30.0
+        )
+        assert collector.sensor_states["target_count"][recovery_idx] == pytest.approx(
+            1.0
+        )
+        assert collector.sensor_states["moving_target_count"][
+            recovery_idx
+        ] == pytest.approx(1.0)
+        assert collector.sensor_states["still_target_count"][
+            recovery_idx
+        ] == pytest.approx(0.0)

tests/integration/test_uart_mock_modbus.py

@@ -5,6 +5,10 @@ test_uart_mock_modbus :
   1. Read a single register and parse successfully (basic_register)
   2. Read multiple registers from SDM meter and parse successfully (sdm_voltage), with some intermediate delay to simulate UART buffer time.
 
+test_uart_mock_modbus_no_threshold :
+  Test modbus with no rx_full_threshold set (simulating USB UART / non-hardware UART).
+  Verifies the 50ms fallback timeout handles chunked data with USB packet gaps.
+
 """
 
 from __future__ import annotations
@@ -218,3 +222,78 @@ async def test_uart_mock_modbus_timing(
                 f"Timeout waiting for SDM voltage change. Received sensor states:\n"
                 f"  sdm_voltage: {sensor_states['sdm_voltage']}\n"
             )
+
+
+@pytest.mark.asyncio
+async def test_uart_mock_modbus_no_threshold(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test modbus with no rx_full_threshold (simulating USB UART).
+
+    Without the 50ms fallback timeout, the chunked response with a 40ms gap
+    between USB packets would cause a false timeout and CRC failure cascade.
+    """
+    # Replace external component path placeholder
+    external_components_path = str(
+        Path(__file__).parent / "fixtures" / "external_components"
+    )
+    yaml_config = yaml_config.replace(
+        "EXTERNAL_COMPONENT_PATH", external_components_path
+    )
+
+    loop = asyncio.get_running_loop()
+
+    # Track sensor state updates (after initial state is swallowed)
+    sensor_states: dict[str, list[float]] = {
+        "sdm_voltage": [],
+    }
+
+    voltage_changed = loop.create_future()
+
+    def on_state(state: EntityState) -> None:
+        if isinstance(state, SensorState) and not state.missing_state:
+            sensor_name = key_to_sensor.get(state.key)
+            if sensor_name and sensor_name in sensor_states:
+                sensor_states[sensor_name].append(state.state)
+                # Check if this is a good voltage reading (243V)
+                if (
+                    sensor_name == "sdm_voltage"
+                    and state.state > 200.0
+                    and not voltage_changed.done()
+                ):
+                    voltage_changed.set_result(True)
+
+    async with (
+        run_compiled(yaml_config),
+        api_client_connected() as client,
+    ):
+        entities, _ = await client.list_entities_services()
+
+        # Build key mappings for all sensor types
+        all_names = list(sensor_states.keys())
+        key_to_sensor = build_key_to_entity_mapping(entities, all_names)
+
+        # Set up initial state helper
+        initial_state_helper = InitialStateHelper(entities)
+        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
+
+        try:
+            await initial_state_helper.wait_for_initial_states()
+        except TimeoutError:
+            pytest.fail("Timeout waiting for initial states")
+
+        # Start the UART mock scenario now that we're subscribed
+        start_btn = find_entity(entities, "start_scenario", ButtonInfo)
+        assert start_btn is not None, "Start Scenario button not found"
+        client.button_command(start_btn.key)
+
+        # Wait for voltage to be updated with successful parse
+        try:
+            await asyncio.wait_for(voltage_changed, timeout=2.0)
+        except TimeoutError:
+            pytest.fail(
+                f"Timeout waiting for SDM voltage change. Received sensor states:\n"
+                f"  sdm_voltage: {sensor_states['sdm_voltage']}\n"
+            )