Merge pull request #15675 from esphome/bump-2026.4.0b2

2026.4.0b2

.clang-tidy.hash

@@ -1 +1 @@
-f31f13994768b5b07e29624406c9b053bf4bb26e1623ac2bc1e9d4a9477502d6
+d48687d988ae2a94a9973226df773478a7db1d52133545f07aa05e34fc678dcf

Doxyfile

@@ -48,7 +48,7 @@ PROJECT_NAME           = ESPHome
 # could be handy for archiving the generated documentation or if some version
 # control system is used.
 
-PROJECT_NUMBER         = 2026.4.0b1
+PROJECT_NUMBER         = 2026.4.0b2
 
 # Using the PROJECT_BRIEF tag one can provide an optional one line description
 # for a project that appears at the top of each page and should give viewer a

esphome/components/api/api.proto

@@ -1625,6 +1625,7 @@ message BluetoothLEAdvertisementResponse {
 }
 
 message BluetoothLERawAdvertisement {
+  option (inline_encode) = true;
   uint64 address = 1 [(force) = true];
   sint32 rssi = 2 [(force) = true];
   uint32 address_type = 3 [(max_value) = 4];

esphome/components/api/api_connection.cpp

@@ -52,11 +52,11 @@
 
 namespace esphome::api {
 
-// Read a maximum of 5 messages per loop iteration to prevent starving other components.
+// Maximum messages to read per loop iteration to prevent starving other components.
 // This is a balance between API responsiveness and allowing other components to run.
 // Since each message could contain multiple protobuf messages when using packet batching,
 // this limits the number of messages processed, not the number of TCP packets.
-static constexpr uint8_t MAX_MESSAGES_PER_LOOP = 5;
+static constexpr uint8_t MAX_MESSAGES_PER_LOOP = 10;
 static constexpr uint8_t MAX_PING_RETRIES = 60;
 static constexpr uint16_t PING_RETRY_INTERVAL = 1000;
 static constexpr uint32_t KEEPALIVE_DISCONNECT_TIMEOUT = (KEEPALIVE_TIMEOUT_MS * 5) / 2;
@@ -220,10 +220,17 @@ void APIConnection::loop() {
   }
 
   const uint32_t now = App.get_loop_component_start_time();
-  // Check if socket has data ready before attempting to read
-  if (this->helper_->is_socket_ready()) {
+  // Check if socket has data ready before attempting to read.
+  // Also try reading if we hit the message limit last time — LWIP's rcvevent
+  // (used by is_socket_ready) tracks pbuf dequeues, not bytes. When multiple
+  // messages share a TCP segment, the last message's data stays in LWIP's
+  // lastdata cache after rcvevent hits 0, making is_socket_ready() return false
+  // even though data remains.
+  if (this->helper_->is_socket_ready() || this->flags_.may_have_remaining_data) {
+    this->flags_.may_have_remaining_data = false;
     // Read up to MAX_MESSAGES_PER_LOOP messages per loop to improve throughput
-    for (uint8_t message_count = 0; message_count < MAX_MESSAGES_PER_LOOP; message_count++) {
+    uint8_t message_count = 0;
+    for (; message_count < MAX_MESSAGES_PER_LOOP; message_count++) {
       ReadPacketBuffer buffer;
       err = this->helper_->read_packet(&buffer);
       if (err == APIError::WOULD_BLOCK) {
@@ -245,6 +252,11 @@ void APIConnection::loop() {
           return;
       }
     }
+    // If we hit the limit, there may be more data remaining in LWIP's
+    // lastdata cache that rcvevent doesn't account for.
+    if (message_count == MAX_MESSAGES_PER_LOOP) {
+      this->flags_.may_have_remaining_data = true;
+    }
   }
 
   // Process deferred batch if scheduled and timer has expired
@@ -2086,6 +2098,13 @@ void APIConnection::process_batch_() {
     return;
   }
 
+  // Ensure TCP_NODELAY is on before draining overflow and writing batch data.
+  // Log messages enable Nagle (NODELAY off) to coalesce small packets.
+  // If Nagle is still on when we try to drain, LWIP holds data in the
+  // Nagle buffer, the TCP send buffer stays full, and the overflow
+  // buffer can never drain — blocking the batch write indefinitely.
+  this->helper_->set_nodelay_for_message(false);
+
   // Try to clear buffer first
   if (!this->try_to_clear_buffer(true)) {
     // Can't write now, we'll try again later
@@ -2193,13 +2212,6 @@ void APIConnection::process_batch_multi_(APIBuffer &shared_buf, size_t num_items
       shared_buf.resize(shared_buf.size() + footer_size);
     }
 
-    // Ensure TCP_NODELAY is on before writing batch data.
-    // Log messages enable Nagle (NODELAY off) to coalesce small packets.
-    // Without this, batch data written to the socket sits in LWIP's Nagle
-    // buffer — the remote won't ACK until it sends its own data (e.g. a
-    // ping), which can take 20+ seconds.
-    this->helper_->set_nodelay_for_message(false);
-
     // Send all collected messages
     APIError err = this->helper_->write_protobuf_messages(ProtoWriteBuffer{&shared_buf},
                                                           std::span<const MessageInfo>(message_info, items_processed));

esphome/components/api/api_connection.h

@@ -771,6 +771,7 @@ class APIConnection final : public APIServerConnectionBase {
     uint8_t batch_scheduled : 1;
     uint8_t batch_first_message : 1;          // For batch buffer allocation
     uint8_t should_try_send_immediately : 1;  // True after initial states are sent
+    uint8_t may_have_remaining_data : 1;      // Read loop hit limit, retry without ready check
 #ifdef HAS_PROTO_MESSAGE_DUMP
     uint8_t log_only_mode : 1;
 #endif

esphome/components/api/api_options.proto

@@ -22,6 +22,7 @@ extend google.protobuf.MessageOptions {
     optional bool log = 1039 [default=true];
     optional bool no_delay = 1040 [default=false];
     optional string base_class = 1041;
+    optional bool inline_encode = 1042 [default=false];
 }
 
 extend google.protobuf.FieldOptions {

esphome/components/api/api_pb2.cpp

@@ -2328,40 +2328,37 @@ bool SubscribeBluetoothLEAdvertisementsRequest::decode_varint(uint32_t field_id,
   }
   return true;
 }
-uint8_t *BluetoothLERawAdvertisement::encode(ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_PARAM) const {
-  uint8_t *__restrict__ pos = buffer.get_pos();
-  ProtoEncode::write_raw_byte(pos PROTO_ENCODE_DEBUG_ARG, 8);
-  ProtoEncode::encode_varint_raw_64(pos PROTO_ENCODE_DEBUG_ARG, this->address);
-  ProtoEncode::write_raw_byte(pos PROTO_ENCODE_DEBUG_ARG, 16);
-  ProtoEncode::encode_varint_raw_short(pos PROTO_ENCODE_DEBUG_ARG, encode_zigzag32(this->rssi));
-  if (this->address_type) {
-    ProtoEncode::write_raw_byte(pos PROTO_ENCODE_DEBUG_ARG, 24);
-    ProtoEncode::encode_varint_raw(pos PROTO_ENCODE_DEBUG_ARG, this->address_type);
-  }
-  ProtoEncode::write_raw_byte(pos PROTO_ENCODE_DEBUG_ARG, 34);
-  ProtoEncode::write_raw_byte(pos PROTO_ENCODE_DEBUG_ARG, static_cast<uint8_t>(this->data_len));
-  ProtoEncode::encode_raw(pos PROTO_ENCODE_DEBUG_ARG, this->data, this->data_len);
-  return pos;
-}
-uint32_t BluetoothLERawAdvertisement::calculate_size() const {
-  uint32_t size = 0;
-  size += ProtoSize::calc_uint64_force(1, this->address);
-  size += ProtoSize::calc_sint32_force(1, this->rssi);
-  size += this->address_type ? 2 : 0;
-  size += 2 + this->data_len;
-  return size;
-}
 uint8_t *BluetoothLERawAdvertisementsResponse::encode(ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_PARAM) const {
   uint8_t *__restrict__ pos = buffer.get_pos();
   for (uint16_t i = 0; i < this->advertisements_len; i++) {
-    ProtoEncode::encode_sub_message(pos PROTO_ENCODE_DEBUG_ARG, buffer, 1, this->advertisements[i]);
+    auto &sub_msg = this->advertisements[i];
+    ProtoEncode::write_raw_byte(pos PROTO_ENCODE_DEBUG_ARG, 10);
+    uint8_t *len_pos = pos;
+    ProtoEncode::reserve_byte(pos PROTO_ENCODE_DEBUG_ARG);
+    ProtoEncode::write_raw_byte(pos PROTO_ENCODE_DEBUG_ARG, 8);
+    ProtoEncode::encode_varint_raw_64(pos PROTO_ENCODE_DEBUG_ARG, sub_msg.address);
+    ProtoEncode::write_raw_byte(pos PROTO_ENCODE_DEBUG_ARG, 16);
+    ProtoEncode::encode_varint_raw_short(pos PROTO_ENCODE_DEBUG_ARG, encode_zigzag32(sub_msg.rssi));
+    if (sub_msg.address_type) {
+      ProtoEncode::write_raw_byte(pos PROTO_ENCODE_DEBUG_ARG, 24);
+      ProtoEncode::encode_varint_raw(pos PROTO_ENCODE_DEBUG_ARG, sub_msg.address_type);
+    }
+    ProtoEncode::write_raw_byte(pos PROTO_ENCODE_DEBUG_ARG, 34);
+    ProtoEncode::write_raw_byte(pos PROTO_ENCODE_DEBUG_ARG, static_cast<uint8_t>(sub_msg.data_len));
+    ProtoEncode::encode_raw(pos PROTO_ENCODE_DEBUG_ARG, sub_msg.data, sub_msg.data_len);
+    *len_pos = static_cast<uint8_t>(pos - len_pos - 1);
   }
   return pos;
 }
 uint32_t BluetoothLERawAdvertisementsResponse::calculate_size() const {
   uint32_t size = 0;
   for (uint16_t i = 0; i < this->advertisements_len; i++) {
-    size += ProtoSize::calc_message_force(1, this->advertisements[i].calculate_size());
+    auto &sub_msg = this->advertisements[i];
+    size += 2;
+    size += ProtoSize::calc_uint64_force(1, sub_msg.address);
+    size += ProtoSize::calc_sint32_force(1, sub_msg.rssi);
+    size += sub_msg.address_type ? 2 : 0;
+    size += 2 + sub_msg.data_len;
   }
   return size;
 }

esphome/components/api/api_pb2.h

@@ -1888,8 +1888,6 @@ class BluetoothLERawAdvertisement final : public ProtoMessage {
   uint32_t address_type{0};
   uint8_t data[62]{};
   uint8_t data_len{0};
-  uint8_t *encode(ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_PARAM) const;
-  uint32_t calculate_size() const;
 #ifdef HAS_PROTO_MESSAGE_DUMP
   const char *dump_to(DumpBuffer &out) const override;
 #endif

esphome/components/api/proto.h

@@ -352,6 +352,12 @@ class ProtoEncode {
     PROTO_ENCODE_CHECK_BOUNDS(pos, 1);
     *pos++ = b;
   }
+  /// Reserve one byte for later backpatch (e.g., sub-message length).
+  /// Advances pos past the reserved byte without writing a value.
+  static inline void ESPHOME_ALWAYS_INLINE reserve_byte(uint8_t *__restrict__ &pos PROTO_ENCODE_DEBUG_PARAM) {
+    PROTO_ENCODE_CHECK_BOUNDS(pos, 1);
+    pos++;
+  }
   /// Write raw bytes to the buffer (no tag, no length prefix).
   static inline void ESPHOME_ALWAYS_INLINE encode_raw(uint8_t *__restrict__ &pos PROTO_ENCODE_DEBUG_PARAM,
                                                       const void *data, size_t len) {

esphome/components/canbus/__init__.py

@@ -162,7 +162,6 @@ async def canbus_action_to_code(config, action_id, template_arg, args):
     await cg.register_parented(var, config[CONF_CANBUS_ID])
 
     if (can_id := config.get(CONF_CAN_ID)) is not None:
-        can_id = await cg.templatable(can_id, args, cg.uint32)
         cg.add(var.set_can_id(can_id))
         cg.add(var.set_use_extended_id(config[CONF_USE_EXTENDED_ID]))
 

esphome/components/cc1101/cc1101.cpp

@@ -102,6 +102,8 @@ CC1101Component::CC1101Component() {
   memset(this->pa_table_, 0, sizeof(this->pa_table_));
 }
 
+void IRAM_ATTR CC1101Component::gpio_intr(CC1101Component *arg) { arg->enable_loop_soon_any_context(); }
+
 void CC1101Component::setup() {
   this->spi_setup();
   this->cs_->digital_write(true);
@@ -148,7 +150,12 @@ void CC1101Component::setup() {
   // Defer pin mode setup until after all components have completed setup()
   // This handles the case where remote_transmitter runs after CC1101 and changes pin mode
   if (this->gdo0_pin_ != nullptr) {
-    this->defer([this]() { this->gdo0_pin_->pin_mode(gpio::FLAG_INPUT); });
+    this->defer([this]() {
+      this->gdo0_pin_->pin_mode(gpio::FLAG_INPUT);
+      if (this->state_.PKT_FORMAT == static_cast<uint8_t>(PacketFormat::PACKET_FORMAT_FIFO)) {
+        this->gdo0_pin_->attach_interrupt(&CC1101Component::gpio_intr, this, gpio::INTERRUPT_RISING_EDGE);
+      }
+    });
   }
 }
 
@@ -160,6 +167,7 @@ void CC1101Component::call_listeners_(const std::vector<uint8_t> &packet, float
 }
 
 void CC1101Component::loop() {
+  this->disable_loop();
   if (this->state_.PKT_FORMAT != static_cast<uint8_t>(PacketFormat::PACKET_FORMAT_FIFO) || this->gdo0_pin_ == nullptr ||
       !this->gdo0_pin_->digital_read()) {
     return;
@@ -240,6 +248,7 @@ void CC1101Component::begin_tx() {
   this->write_(Register::PKTCTRL0, 0x32);
   ESP_LOGV(TAG, "Beginning TX sequence");
   if (this->gdo0_pin_ != nullptr) {
+    this->gdo0_pin_->detach_interrupt();
     this->gdo0_pin_->pin_mode(gpio::FLAG_OUTPUT);
   }
   // Transition through IDLE to bypass CCA (Clear Channel Assessment) which can
@@ -669,6 +678,13 @@ void CC1101Component::set_packet_mode(bool value) {
     this->state_.GDO0_CFG = 0x0D;
   }
   if (this->initialized_) {
+    if (this->gdo0_pin_ != nullptr) {
+      if (value) {
+        this->gdo0_pin_->attach_interrupt(&CC1101Component::gpio_intr, this, gpio::INTERRUPT_RISING_EDGE);
+      } else {
+        this->gdo0_pin_->detach_interrupt();
+      }
+    }
     this->write_(Register::PKTCTRL0);
     this->write_(Register::PKTCTRL1);
     this->write_(Register::IOCFG0);

esphome/components/cc1101/cc1101.h

@@ -93,6 +93,7 @@ class CC1101Component : public Component,
 
   // GDO pin for packet reception
   InternalGPIOPin *gdo0_pin_{nullptr};
+  static void IRAM_ATTR gpio_intr(CC1101Component *arg);
 
   // Packet handling
   void call_listeners_(const std::vector<uint8_t> &packet, float freq_offset, float rssi, uint8_t lqi);

esphome/components/esp32/__init__.py

@@ -671,11 +671,12 @@ def _is_framework_url(source: str) -> bool:
 # The default/recommended arduino framework version
 #  - https://github.com/espressif/arduino-esp32/releases
 ARDUINO_FRAMEWORK_VERSION_LOOKUP = {
-    "recommended": cv.Version(3, 3, 7),
-    "latest": cv.Version(3, 3, 7),
-    "dev": cv.Version(3, 3, 7),
+    "recommended": cv.Version(3, 3, 8),
+    "latest": cv.Version(3, 3, 8),
+    "dev": cv.Version(3, 3, 8),
 }
 ARDUINO_PLATFORM_VERSION_LOOKUP = {
+    cv.Version(3, 3, 8): cv.Version(55, 3, 38),
     cv.Version(3, 3, 7): cv.Version(55, 3, 37),
     cv.Version(3, 3, 6): cv.Version(55, 3, 36),
     cv.Version(3, 3, 5): cv.Version(55, 3, 35),
@@ -695,6 +696,7 @@ ARDUINO_PLATFORM_VERSION_LOOKUP = {
 # These versions correspond to pioarduino/esp-idf releases
 # See: https://github.com/pioarduino/esp-idf/releases
 ARDUINO_IDF_VERSION_LOOKUP = {
+    cv.Version(3, 3, 8): cv.Version(5, 5, 4),
     cv.Version(3, 3, 7): cv.Version(5, 5, 3, "1"),
     cv.Version(3, 3, 6): cv.Version(5, 5, 2),
     cv.Version(3, 3, 5): cv.Version(5, 5, 2),
@@ -714,17 +716,15 @@ ARDUINO_IDF_VERSION_LOOKUP = {
 # The default/recommended esp-idf framework version
 #  - https://github.com/espressif/esp-idf/releases
 ESP_IDF_FRAMEWORK_VERSION_LOOKUP = {
-    "recommended": cv.Version(5, 5, 3, "1"),
-    "latest": cv.Version(5, 5, 3, "1"),
+    "recommended": cv.Version(5, 5, 4),
+    "latest": cv.Version(5, 5, 4),
     "dev": cv.Version(5, 5, 4),
 }
 ESP_IDF_PLATFORM_VERSION_LOOKUP = {
     cv.Version(
         6, 0, 0
     ): "https://github.com/pioarduino/platform-espressif32.git#prep_IDF6",
-    cv.Version(
-        5, 5, 4
-    ): "https://github.com/pioarduino/platform-espressif32.git#develop",
+    cv.Version(5, 5, 4): cv.Version(55, 3, 38),
     cv.Version(5, 5, 3, "1"): cv.Version(55, 3, 37),
     cv.Version(5, 5, 3): cv.Version(55, 3, 37),
     cv.Version(5, 5, 2): cv.Version(55, 3, 37),
@@ -744,8 +744,8 @@ ESP_IDF_PLATFORM_VERSION_LOOKUP = {
 # The platform-espressif32 version
 #  - https://github.com/pioarduino/platform-espressif32/releases
 PLATFORM_VERSION_LOOKUP = {
-    "recommended": cv.Version(55, 3, 37),
-    "latest": cv.Version(55, 3, 37),
+    "recommended": cv.Version(55, 3, 38),
+    "latest": cv.Version(55, 3, 38),
     "dev": "https://github.com/pioarduino/platform-espressif32.git#develop",
 }
 

esphome/components/esp32/boards.py

@@ -1960,6 +1960,10 @@ BOARDS = {
         "name": "Hornbill ESP32 Minima",
         "variant": VARIANT_ESP32,
     },
+    "huidu_hd_wf1": {
+        "name": "Huidu HD-WF1",
+        "variant": VARIANT_ESP32S2,
+    },
     "huidu_hd_wf2": {
         "name": "Huidu HD-WF2",
         "variant": VARIANT_ESP32S3,
@@ -2028,6 +2032,10 @@ BOARDS = {
         "name": "LilyGo T-Display-S3",
         "variant": VARIANT_ESP32S3,
     },
+    "lilygo-t-energy-s3": {
+        "name": "LilyGo T-Energy-S3",
+        "variant": VARIANT_ESP32S3,
+    },
     "lilygo-t3-s3": {
         "name": "LilyGo T3-S3",
         "variant": VARIANT_ESP32S3,
@@ -2289,10 +2297,18 @@ BOARDS = {
         "name": "S.ODI Ultra v1",
         "variant": VARIANT_ESP32,
     },
+    "seeed_xiao_esp32_s3_plus": {
+        "name": "Seeed Studio XIAO ESP32S3 Plus",
+        "variant": VARIANT_ESP32S3,
+    },
     "seeed_xiao_esp32c3": {
         "name": "Seeed Studio XIAO ESP32C3",
         "variant": VARIANT_ESP32C3,
     },
+    "seeed_xiao_esp32c5": {
+        "name": "Seeed Studio XIAO ESP32C5",
+        "variant": VARIANT_ESP32C5,
+    },
     "seeed_xiao_esp32c6": {
         "name": "Seeed Studio XIAO ESP32C6",
         "variant": VARIANT_ESP32C6,

esphome/components/gdk101/gdk101.cpp

@@ -7,7 +7,7 @@ namespace gdk101 {
 
 static const char *const TAG = "gdk101";
 static constexpr uint8_t NUMBER_OF_READ_RETRIES = 5;
-static constexpr uint8_t NUMBER_OF_RESET_RETRIES = 10;
+static constexpr uint8_t NUMBER_OF_RESET_RETRIES = 30;
 static constexpr uint32_t RESET_INTERVAL_ID = 0;
 static constexpr uint32_t RESET_INTERVAL_MS = 1000;
 

esphome/components/hbridge/light/__init__.py

@@ -8,7 +8,7 @@ from .. import hbridge_ns
 CODEOWNERS = ["@DotNetDann"]
 
 HBridgeLightOutput = hbridge_ns.class_(
-    "HBridgeLightOutput", cg.PollingComponent, light.LightOutput
+    "HBridgeLightOutput", cg.Component, light.LightOutput
 )
 
 CONFIG_SCHEMA = light.RGB_LIGHT_SCHEMA.extend(

esphome/components/hbridge/light/hbridge_light_output.h

@@ -1,20 +1,17 @@
 #pragma once
 
-#include "esphome/core/component.h"
-#include "esphome/components/output/float_output.h"
 #include "esphome/components/light/light_output.h"
-#include "esphome/core/log.h"
+#include "esphome/components/output/float_output.h"
+#include "esphome/core/component.h"
+#include "esphome/core/helpers.h"
 
 namespace esphome {
 namespace hbridge {
 
-// Using PollingComponent as the updates are more consistent and reduces flickering
-class HBridgeLightOutput : public PollingComponent, public light::LightOutput {
+class HBridgeLightOutput : public Component, public light::LightOutput {
  public:
-  HBridgeLightOutput() : PollingComponent(1) {}
-
-  void set_pina_pin(output::FloatOutput *pina_pin) { pina_pin_ = pina_pin; }
-  void set_pinb_pin(output::FloatOutput *pinb_pin) { pinb_pin_ = pinb_pin; }
+  void set_pina_pin(output::FloatOutput *pina_pin) { this->pina_pin_ = pina_pin; }
+  void set_pinb_pin(output::FloatOutput *pinb_pin) { this->pinb_pin_ = pinb_pin; }
 
   light::LightTraits get_traits() override {
     auto traits = light::LightTraits();
@@ -24,16 +21,16 @@ class HBridgeLightOutput : public PollingComponent, public light::LightOutput {
     return traits;
   }
 
-  void setup() override { this->forward_direction_ = false; }
+  void setup() override { this->disable_loop(); }
 
-  void update() override {
-    // This method runs around 60 times per second
-    // We cannot do the PWM ourselves so we are reliant on the hardware PWM
-    if (!this->forward_direction_) {  // First LED Direction
+  void loop() override {
+    // Only called when both channels are active — alternate H-bridge direction
+    // each iteration to multiplex cold and warm white.
+    if (!this->forward_direction_) {
       this->pina_pin_->set_level(this->pina_duty_);
       this->pinb_pin_->set_level(0);
       this->forward_direction_ = true;
-    } else {  // Second LED Direction
+    } else {
       this->pina_pin_->set_level(0);
       this->pinb_pin_->set_level(this->pinb_duty_);
       this->forward_direction_ = false;
@@ -43,15 +40,32 @@ class HBridgeLightOutput : public PollingComponent, public light::LightOutput {
   float get_setup_priority() const override { return setup_priority::HARDWARE; }
 
   void write_state(light::LightState *state) override {
-    state->current_values_as_cwww(&this->pina_duty_, &this->pinb_duty_, false);
+    float new_pina, new_pinb;
+    state->current_values_as_cwww(&new_pina, &new_pinb, false);
+
+    this->pina_duty_ = new_pina;
+    this->pinb_duty_ = new_pinb;
+
+    if (new_pina != 0.0f && new_pinb != 0.0f) {
+      // Both channels active — need loop to alternate H-bridge direction
+      this->high_freq_.start();
+      this->enable_loop();
+    } else {
+      // Zero or one channel active — drive pins directly, no multiplexing needed
+      this->high_freq_.stop();
+      this->disable_loop();
+      this->pina_pin_->set_level(new_pina);
+      this->pinb_pin_->set_level(new_pinb);
+    }
   }
 
  protected:
   output::FloatOutput *pina_pin_;
   output::FloatOutput *pinb_pin_;
-  float pina_duty_ = 0;
-  float pinb_duty_ = 0;
-  bool forward_direction_ = false;
+  float pina_duty_{0};
+  float pinb_duty_{0};
+  bool forward_direction_{false};
+  HighFrequencyLoopRequester high_freq_;
 };
 
 }  // namespace hbridge

esphome/components/lvgl/__init__.py

@@ -341,7 +341,7 @@ async def to_code(configs):
                 df.LOGGER.info("LVGL will use hardware rotation via display driver")
             else:
                 rotation_type = RotationType.ROTATION_SOFTWARE
-                if get_esp32_variant() == VARIANT_ESP32P4:
+                if CORE.is_esp32 and get_esp32_variant() == VARIANT_ESP32P4:
                     df.LOGGER.info("LVGL will use software rotation (PPA accelerated)")
                 else:
                     df.LOGGER.info("LVGL will use software rotation")

esphome/components/mcp23016/__init__.py

@@ -5,6 +5,7 @@ import esphome.config_validation as cv
 from esphome.const import (
     CONF_ID,
     CONF_INPUT,
+    CONF_INTERRUPT_PIN,
     CONF_INVERTED,
     CONF_MODE,
     CONF_NUMBER,
@@ -24,6 +25,7 @@ CONFIG_SCHEMA = (
     cv.Schema(
         {
             cv.Required(CONF_ID): cv.declare_id(MCP23016),
+            cv.Optional(CONF_INTERRUPT_PIN): pins.internal_gpio_input_pin_schema,
         }
     )
     .extend(cv.COMPONENT_SCHEMA)
@@ -35,6 +37,8 @@ async def to_code(config):
     var = cg.new_Pvariable(config[CONF_ID])
     await cg.register_component(var, config)
     await i2c.register_i2c_device(var, config)
+    if interrupt_pin := config.get(CONF_INTERRUPT_PIN):
+        cg.add(var.set_interrupt_pin(await cg.gpio_pin_expression(interrupt_pin)))
 
 
 def validate_mode(value):

esphome/components/mcp23016/mcp23016.cpp

@@ -24,11 +24,22 @@ void MCP23016::setup() {
 
   // all pins input
   this->write_reg_(MCP23016_IODIR1, 0xFFFF);
+
+  if (this->interrupt_pin_ != nullptr) {
+    this->interrupt_pin_->setup();
+    this->interrupt_pin_->attach_interrupt(&MCP23016::gpio_intr, this, gpio::INTERRUPT_FALLING_EDGE);
+    this->set_invalidate_on_read_(false);
+  }
+  this->disable_loop();
 }
 
+void IRAM_ATTR MCP23016::gpio_intr(MCP23016 *arg) { arg->enable_loop_soon_any_context(); }
 void MCP23016::loop() {
   // Invalidate cache at the start of each loop
   this->reset_pin_cache_();
+  if (this->interrupt_pin_ != nullptr) {
+    this->disable_loop();
+  }
 }
 bool MCP23016::digital_read_hw(uint8_t pin) { return this->read_reg_(MCP23016_GP1, &this->input_mask_); }
 
@@ -37,6 +48,9 @@ void MCP23016::digital_write_hw(uint8_t pin, bool value) { this->update_reg_(pin
 void MCP23016::pin_mode(uint8_t pin, gpio::Flags flags) {
   if (flags == gpio::FLAG_INPUT) {
     this->update_reg_(pin, true, MCP23016_IODIR1);
+    if (this->interrupt_pin_ == nullptr) {
+      this->enable_loop();
+    }
   } else if (flags == gpio::FLAG_OUTPUT) {
     this->update_reg_(pin, false, MCP23016_IODIR1);
   }

esphome/components/mcp23016/mcp23016.h

@@ -35,7 +35,10 @@ class MCP23016 : public Component, public i2c::I2CDevice, public gpio_expander::
 
   float get_setup_priority() const override;
 
+  void set_interrupt_pin(InternalGPIOPin *pin) { this->interrupt_pin_ = pin; }
+
  protected:
+  static void IRAM_ATTR gpio_intr(MCP23016 *arg);
   // Virtual methods from CachedGpioExpander
   bool digital_read_hw(uint8_t pin) override;
   bool digital_read_cache(uint8_t pin) override;
@@ -51,6 +54,7 @@ class MCP23016 : public Component, public i2c::I2CDevice, public gpio_expander::
   uint16_t olat_{0x0000};
   // Cache for input values (16-bit combined for both banks)
   uint16_t input_mask_{0x0000};
+  InternalGPIOPin *interrupt_pin_{nullptr};
 };
 
 class MCP23016GPIOPin : public GPIOPin {

esphome/components/mdns/__init__.py

@@ -170,7 +170,7 @@ async def to_code(config):
             cg.add_library("LEAmDNS", None)
 
     if CORE.is_esp32:
-        add_idf_component(name="espressif/mdns", ref="1.10.0")
+        add_idf_component(name="espressif/mdns", ref="1.11.0")
 
     cg.add_define("USE_MDNS")
 

esphome/components/micro_wake_word/__init__.py

@@ -451,6 +451,8 @@ async def to_code(config):
     ota.request_ota_state_listeners()
 
     esp32.add_idf_component(name="espressif/esp-tflite-micro", ref="1.3.3~1")
+    # Pin esp-nn for stable future builds (esp-tflite-micro depends on esp-nn)
+    esp32.add_idf_component(name="espressif/esp-nn", ref="1.2.1")
 
     cg.add_build_flag("-DTF_LITE_STATIC_MEMORY")
     cg.add_build_flag("-DTF_LITE_DISABLE_X86_NEON")

esphome/components/micro_wake_word/streaming_model.cpp

@@ -29,14 +29,6 @@ void VADModel::log_model_config() {
 bool StreamingModel::load_model_() {
   RAMAllocator<uint8_t> arena_allocator;
 
-  if (this->tensor_arena_ == nullptr) {
-    this->tensor_arena_ = arena_allocator.allocate(this->tensor_arena_size_);
-    if (this->tensor_arena_ == nullptr) {
-      ESP_LOGE(TAG, "Could not allocate the streaming model's tensor arena.");
-      return false;
-    }
-  }
-
   if (this->var_arena_ == nullptr) {
     this->var_arena_ = arena_allocator.allocate(STREAMING_MODEL_VARIABLE_ARENA_SIZE);
     if (this->var_arena_ == nullptr) {
@@ -53,6 +45,26 @@ bool StreamingModel::load_model_() {
     return false;
   }
 
+  // Probe for the actual required tensor arena size if not yet determined
+  if (!this->tensor_arena_size_probed_) {
+    size_t probed_size = this->probe_arena_size_();
+    if (probed_size > 0) {
+      ESP_LOGD(TAG, "Probed tensor arena size: %zu bytes", probed_size);
+      this->tensor_arena_size_ = probed_size;
+    } else {
+      ESP_LOGW(TAG, "Arena size probe failed, using manifest size: %zu bytes", this->tensor_arena_size_);
+    }
+    this->tensor_arena_size_probed_ = true;
+  }
+
+  if (this->tensor_arena_ == nullptr) {
+    this->tensor_arena_ = arena_allocator.allocate(this->tensor_arena_size_);
+    if (this->tensor_arena_ == nullptr) {
+      ESP_LOGE(TAG, "Could not allocate the streaming model's tensor arena.");
+      return false;
+    }
+  }
+
   if (this->interpreter_ == nullptr) {
     this->interpreter_ =
         make_unique<tflite::MicroInterpreter>(tflite::GetModel(this->model_start_), this->streaming_op_resolver_,
@@ -94,6 +106,70 @@ bool StreamingModel::load_model_() {
   return true;
 }
 
+size_t StreamingModel::probe_arena_size_() {
+  RAMAllocator<uint8_t> arena_allocator;
+
+  // Try with the manifest size first, then escalates to 1.5, then 2x if it fails. Different platforms and different
+  // versions of the esp-nn library require different amounts of memory, so the manifest size may not always be correct,
+  // and probing allows us to find the actual required size for the current build and platform. Aligns test sizes to 16
+  // bytes.
+  size_t attempt_sizes[] = {(this->tensor_arena_size_ + 15) & ~15, (this->tensor_arena_size_ * 3 / 2 + 15) & ~15,
+                            (this->tensor_arena_size_ * 2 + 15) & ~15};
+
+  for (size_t attempt_size : attempt_sizes) {
+    uint8_t *probe_arena = arena_allocator.allocate(attempt_size);
+    if (probe_arena == nullptr) {
+      continue;
+    }
+
+    // Verify the model works at all with this arena size
+    auto probe_interpreter = make_unique<tflite::MicroInterpreter>(
+        tflite::GetModel(this->model_start_), this->streaming_op_resolver_, probe_arena, attempt_size, this->mrv_);
+
+    if (probe_interpreter->AllocateTensors() != kTfLiteOk) {
+      probe_interpreter.reset();
+      arena_allocator.deallocate(probe_arena, attempt_size);
+      this->ma_ = tflite::MicroAllocator::Create(this->var_arena_, STREAMING_MODEL_VARIABLE_ARENA_SIZE);
+      this->mrv_ = tflite::MicroResourceVariables::Create(this->ma_, 20);
+      continue;
+    }
+
+    // Try to shrink the arena. Start with arena_used_bytes() + 16 (rounded to 16-byte alignment).
+    // If that works, use it. Otherwise, try midpoints between that and the full size until one succeeds.
+    size_t lower = (probe_interpreter->arena_used_bytes() + 16 + 15) & ~15;
+    probe_interpreter.reset();
+    this->ma_ = tflite::MicroAllocator::Create(this->var_arena_, STREAMING_MODEL_VARIABLE_ARENA_SIZE);
+    this->mrv_ = tflite::MicroResourceVariables::Create(this->ma_, 20);
+
+    size_t upper = attempt_size;
+
+    while (lower < upper) {
+      auto test_interpreter = make_unique<tflite::MicroInterpreter>(
+          tflite::GetModel(this->model_start_), this->streaming_op_resolver_, probe_arena, lower, this->mrv_);
+
+      bool ok = test_interpreter->AllocateTensors() == kTfLiteOk;
+
+      test_interpreter.reset();
+      this->ma_ = tflite::MicroAllocator::Create(this->var_arena_, STREAMING_MODEL_VARIABLE_ARENA_SIZE);
+      this->mrv_ = tflite::MicroResourceVariables::Create(this->ma_, 20);
+
+      if (ok) {
+        // Found a working size smaller than the full arena
+        upper = lower + 16;  // Pad by 16 bytes to be safe for future allocations
+        break;
+      }
+
+      // Try the midpoint between current attempt and full size
+      lower = ((lower + upper) / 2 + 15) & ~15;
+    }
+
+    arena_allocator.deallocate(probe_arena, attempt_size);
+    return upper;
+  }
+
+  return 0;
+}
+
 void StreamingModel::unload_model() {
   this->interpreter_.reset();
 

esphome/components/micro_wake_word/streaming_model.h

@@ -63,6 +63,10 @@ class StreamingModel {
   /// @brief Allocates tensor and variable arenas and sets up the model interpreter
   /// @return True if successful, false otherwise
   bool load_model_();
+  /// @brief Probes the actual required tensor arena size by trial allocation.
+  /// Tries the manifest size first, then 2x if that fails.
+  /// @return The required arena size rounded up to 16-byte alignment, or 0 on failure.
+  size_t probe_arena_size_();
   /// @brief Returns true if successfully registered the streaming model's TensorFlow operations
   bool register_streaming_ops_(tflite::MicroMutableOpResolver<20> &op_resolver);
 
@@ -70,6 +74,7 @@ class StreamingModel {
 
   bool loaded_{false};
   bool enabled_{true};
+  bool tensor_arena_size_probed_{false};
   bool unprocessed_probability_status_{false};
   uint8_t current_stride_step_{0};
   int16_t ignore_windows_{-MIN_SLICES_BEFORE_DETECTION};

esphome/components/packages/__init__.py

@@ -45,6 +45,18 @@ def is_remote_package(package_config: dict) -> bool:
     return CONF_URL in package_config
 
 
+def is_package_definition(value: object) -> bool:
+    """Returns True if the value looks like a package definition rather than a config fragment.
+
+    Package definitions are IncludeFile objects, git URL shorthand strings, or
+    remote package dicts (containing a ``url:`` key).  Config fragments are
+    plain dicts that represent component configuration.
+    """
+    return isinstance(value, (yaml_util.IncludeFile, str)) or (
+        isinstance(value, dict) and is_remote_package(value)
+    )
+
+
 def valid_package_contents(package_config: dict) -> dict:
     """Validate that a package looks like a plausible ESPHome config fragment.
 
@@ -318,11 +330,11 @@ def _walk_packages(
         if not isinstance(packages, dict):
             _walk_package_list(packages, callback, context)
         elif (result := _walk_package_dict(packages, callback, context)) is not None:
-            if not validate_deprecated:
+            if not validate_deprecated or any(
+                is_package_definition(v) for v in packages.values()
+            ):
                 raise result
             # Fallback: treat the dict as a single deprecated package.
-            # Note: this catches *any* cv.Invalid from the callback, which may
-            # mask real validation errors in named package dicts.
             # This block can be removed once the single-package
             # deprecation period (2026.7.0) is over.
             config[CONF_PACKAGES] = [packages]
@@ -461,6 +473,9 @@ class _PackageProcessor:
         self, package_config: dict | str, context_vars: ContextVars | None
     ) -> dict:
         """Resolve a single package and recurse into any nested packages."""
+        from_remote = isinstance(package_config, dict) and is_remote_package(
+            package_config
+        )
         package_config = self.resolve_package(package_config, context_vars)
         self.collect_substitutions(package_config)
 
@@ -470,7 +485,18 @@ class _PackageProcessor:
         # Push context from !include vars on the package root and on the packages key
         context_vars = push_context(package_config, context_vars)
         context_vars = push_context(package_config[CONF_PACKAGES], context_vars)
-        return _walk_packages(package_config, self.process_package, context_vars)
+        # Disable the deprecated single-package fallback for remote
+        # packages.  _process_remote_package returns dicts with
+        # already-resolved values that is_package_definition cannot
+        # distinguish from config fragments, so the fallback would
+        # always fire and mask real errors with wrong paths
+        # (packages->0 instead of packages-><name>).
+        return _walk_packages(
+            package_config,
+            self.process_package,
+            context_vars,
+            validate_deprecated=not from_remote,
+        )
 
 
 def do_packages_pass(

esphome/components/pca6416a/__init__.py

@@ -5,6 +5,7 @@ import esphome.config_validation as cv
 from esphome.const import (
     CONF_ID,
     CONF_INPUT,
+    CONF_INTERRUPT_PIN,
     CONF_INVERTED,
     CONF_MODE,
     CONF_NUMBER,
@@ -25,7 +26,12 @@ PCA6416AGPIOPin = pca6416a_ns.class_(
 
 CONF_PCA6416A = "pca6416a"
 CONFIG_SCHEMA = (
-    cv.Schema({cv.Required(CONF_ID): cv.declare_id(PCA6416AComponent)})
+    cv.Schema(
+        {
+            cv.Required(CONF_ID): cv.declare_id(PCA6416AComponent),
+            cv.Optional(CONF_INTERRUPT_PIN): pins.internal_gpio_input_pin_schema,
+        }
+    )
     .extend(cv.COMPONENT_SCHEMA)
     .extend(i2c.i2c_device_schema(0x21))
 )
@@ -35,6 +41,8 @@ async def to_code(config):
     var = cg.new_Pvariable(config[CONF_ID])
     await cg.register_component(var, config)
     await i2c.register_i2c_device(var, config)
+    if interrupt_pin := config.get(CONF_INTERRUPT_PIN):
+        cg.add(var.set_interrupt_pin(await cg.gpio_pin_expression(interrupt_pin)))
 
 
 def validate_mode(value):

esphome/components/pca6416a/pca6416a.cpp

@@ -49,11 +49,22 @@ void PCA6416AComponent::setup() {
 
   ESP_LOGD(TAG, "Initialization complete. Warning: %d, Error: %d", this->status_has_warning(),
            this->status_has_error());
+
+  if (this->interrupt_pin_ != nullptr) {
+    this->interrupt_pin_->setup();
+    this->interrupt_pin_->attach_interrupt(&PCA6416AComponent::gpio_intr, this, gpio::INTERRUPT_FALLING_EDGE);
+    this->set_invalidate_on_read_(false);
+  }
+  this->disable_loop();
 }
 
+void IRAM_ATTR PCA6416AComponent::gpio_intr(PCA6416AComponent *arg) { arg->enable_loop_soon_any_context(); }
 void PCA6416AComponent::loop() {
   // Invalidate cache at the start of each loop
   this->reset_pin_cache_();
+  if (this->interrupt_pin_ != nullptr) {
+    this->disable_loop();
+  }
 }
 
 void PCA6416AComponent::dump_config() {
@@ -62,6 +73,7 @@ void PCA6416AComponent::dump_config() {
   } else {
     ESP_LOGCONFIG(TAG, "PCA6416A:");
   }
+  LOG_PIN("  Interrupt Pin: ", this->interrupt_pin_);
   LOG_I2C_DEVICE(this)
   if (this->is_failed()) {
     ESP_LOGE(TAG, ESP_LOG_MSG_COMM_FAIL);
@@ -101,6 +113,9 @@ void PCA6416AComponent::pin_mode(uint8_t pin, gpio::Flags flags) {
       this->update_register_(pin, true, pull_dir);
       this->update_register_(pin, false, pull_en);
     }
+    if (this->interrupt_pin_ == nullptr) {
+      this->enable_loop();
+    }
   } else if (flags == (gpio::FLAG_INPUT | gpio::FLAG_PULLUP)) {
     this->update_register_(pin, true, io_dir);
     if (has_pullup_) {
@@ -109,6 +124,9 @@ void PCA6416AComponent::pin_mode(uint8_t pin, gpio::Flags flags) {
     } else {
       ESP_LOGW(TAG, "Your PCA6416A does not support pull-up resistors");
     }
+    if (this->interrupt_pin_ == nullptr) {
+      this->enable_loop();
+    }
   } else if (flags == gpio::FLAG_OUTPUT) {
     this->update_register_(pin, false, io_dir);
   }

esphome/components/pca6416a/pca6416a.h

@@ -24,7 +24,10 @@ class PCA6416AComponent : public Component,
 
   void dump_config() override;
 
+  void set_interrupt_pin(InternalGPIOPin *pin) { this->interrupt_pin_ = pin; }
+
  protected:
+  static void IRAM_ATTR gpio_intr(PCA6416AComponent *arg);
   // Virtual methods from CachedGpioExpander
   bool digital_read_hw(uint8_t pin) override;
   bool digital_read_cache(uint8_t pin) override;
@@ -43,6 +46,7 @@ class PCA6416AComponent : public Component,
   esphome::i2c::ErrorCode last_error_;
   /// Only the PCAL6416A has pull-up resistors
   bool has_pullup_{false};
+  InternalGPIOPin *interrupt_pin_{nullptr};
 };
 
 /// Helper class to expose a PCA6416A pin as an internal input GPIO pin.

esphome/components/rp2040/helpers.cpp

@@ -9,7 +9,7 @@
 #include <WiFi.h>
 #include <pico/cyw43_arch.h>  // For cyw43_arch_lwip_begin/end (LwIPLock)
 #elif defined(USE_ETHERNET)
-#include <LwipEthernet.h>  // For ethernet_arch_lwip_begin/end (LwIPLock)
+#include <lwip_wrap.h>  // For LWIPMutex — LwIPLock mirrors its semantics (see below)
 #include "esphome/components/ethernet/ethernet_component.h"
 #endif
 #include <hardware/structs/rosc.h>
@@ -43,9 +43,18 @@ IRAM_ATTR InterruptLock::~InterruptLock() { restore_interrupts(state_); }
 // main loop, corrupting the shared rx_buf_ pbuf chain (use-after-free, pbuf_cat
 // assertion failures). See esphome#10681.
 //
-// WiFi uses cyw43_arch_lwip_begin/end; Ethernet uses ethernet_arch_lwip_begin/end.
-// Both acquire the async_context recursive mutex to prevent IRQ callbacks from
-// firing during critical sections.
+// WiFi uses cyw43_arch_lwip_begin/end.
+//
+// For wired Ethernet, taking only the async_context lock is NOT enough. The
+// W5500 GPIO IRQ path (LwipIntfDev::_irq) checks arduino-pico's `__inLWIP`
+// counter to decide whether to defer packet processing. If we hold the
+// async_context lock without bumping `__inLWIP`, an interrupt-driven packet
+// arrival re-enters lwIP from IRQ context and corrupts pbufs (the `pbuf_cat`
+// assertion crash on wiznet-w5500-evb-pico). We mirror arduino-pico's
+// LWIPMutex (cores/rp2040/lwip_wrap.h) exactly: bump `__inLWIP`, take the
+// lock, and on release re-unmask any GPIO IRQs that were deferred while we
+// held it. We can't `using LwIPLock = LWIPMutex;` in helpers.h because
+// pulling lwip_wrap.h there poisons many TUs with lwIP types.
 //
 // When neither WiFi nor Ethernet is configured, this is a no-op since
 // there's no network stack and no lwip callbacks to race with.
@@ -53,8 +62,18 @@ IRAM_ATTR InterruptLock::~InterruptLock() { restore_interrupts(state_); }
 LwIPLock::LwIPLock() { cyw43_arch_lwip_begin(); }
 LwIPLock::~LwIPLock() { cyw43_arch_lwip_end(); }
 #elif defined(USE_ETHERNET)
-LwIPLock::LwIPLock() { ethernet_arch_lwip_begin(); }
-LwIPLock::~LwIPLock() { ethernet_arch_lwip_end(); }
+LwIPLock::LwIPLock() {
+  __inLWIP++;
+  ethernet_arch_lwip_begin();
+}
+LwIPLock::~LwIPLock() {
+  ethernet_arch_lwip_end();
+  __inLWIP--;
+  if (__needsIRQEN && !__inLWIP) {
+    __needsIRQEN = false;
+    ethernet_arch_lwip_gpio_unmask();
+  }
+}
 #else
 LwIPLock::LwIPLock() {}
 LwIPLock::~LwIPLock() {}

esphome/components/sx126x/sx126x.cpp

@@ -104,11 +104,17 @@ void SX126x::write_register_(uint16_t reg, uint8_t *data, uint8_t size) {
   delayMicroseconds(SWITCHING_DELAY_US);
 }
 
+void IRAM_ATTR SX126x::gpio_intr(SX126x *arg) { arg->enable_loop_soon_any_context(); }
+
 void SX126x::setup() {
   // setup pins
   this->busy_pin_->setup();
   this->rst_pin_->setup();
   this->dio1_pin_->setup();
+  if (this->dio1_pin_->is_internal()) {
+    static_cast<InternalGPIOPin *>(this->dio1_pin_)
+        ->attach_interrupt(&SX126x::gpio_intr, this, gpio::INTERRUPT_RISING_EDGE);
+  }
 
   // start spi
   this->spi_setup();
@@ -348,6 +354,9 @@ void SX126x::call_listeners_(const std::vector<uint8_t> &packet, float rssi, flo
 }
 
 void SX126x::loop() {
+  if (this->dio1_pin_->is_internal()) {
+    this->disable_loop();
+  }
   if (!this->dio1_pin_->digital_read()) {
     return;
   }

esphome/components/sx126x/sx126x.h

@@ -3,6 +3,7 @@
 #include "esphome/components/spi/spi.h"
 #include "esphome/core/automation.h"
 #include "esphome/core/component.h"
+#include "esphome/core/hal.h"
 #include "sx126x_reg.h"
 #include <utility>
 #include <vector>
@@ -100,6 +101,7 @@ class SX126x : public Component,
   Trigger<std::vector<uint8_t>, float, float> *get_packet_trigger() { return &this->packet_trigger_; }
 
  protected:
+  static void IRAM_ATTR gpio_intr(SX126x *arg);
   void configure_fsk_ook_();
   void configure_lora_();
   void set_packet_params_(uint8_t payload_length);

esphome/components/sx127x/sx127x.cpp

@@ -53,6 +53,8 @@ void SX127x::write_fifo_(const std::vector<uint8_t> &packet) {
   this->disable();
 }
 
+void IRAM_ATTR SX127x::gpio_intr(SX127x *arg) { arg->enable_loop_soon_any_context(); }
+
 void SX127x::setup() {
   // setup reset
   this->rst_pin_->setup();
@@ -60,6 +62,7 @@ void SX127x::setup() {
   // setup dio0
   if (this->dio0_pin_) {
     this->dio0_pin_->setup();
+    this->dio0_pin_->attach_interrupt(&SX127x::gpio_intr, this, gpio::INTERRUPT_RISING_EDGE);
   }
 
   // start spi
@@ -313,6 +316,7 @@ void SX127x::call_listeners_(const std::vector<uint8_t> &packet, float rssi, flo
 }
 
 void SX127x::loop() {
+  this->disable_loop();
   if (this->dio0_pin_ == nullptr || !this->dio0_pin_->digital_read()) {
     return;
   }
@@ -383,7 +387,7 @@ void SX127x::set_mode_(uint8_t modulation, uint8_t mode) {
     if (millis() - start > 20) {
       ESP_LOGE(TAG, "Set mode failure");
       this->mark_failed();
-      break;
+      return;
     }
   }
 }

esphome/components/sx127x/sx127x.h

@@ -4,6 +4,7 @@
 #include "esphome/components/spi/spi.h"
 #include "esphome/core/automation.h"
 #include "esphome/core/component.h"
+#include "esphome/core/hal.h"
 #include <vector>
 
 namespace esphome {
@@ -86,6 +87,7 @@ class SX127x : public Component,
   Trigger<std::vector<uint8_t>, float, float> *get_packet_trigger() { return &this->packet_trigger_; }
 
  protected:
+  static void IRAM_ATTR gpio_intr(SX127x *arg);
   void configure_fsk_ook_();
   void configure_lora_();
   void set_mode_(uint8_t modulation, uint8_t mode);

esphome/components/tca9555/__init__.py

@@ -5,6 +5,7 @@ import esphome.config_validation as cv
 from esphome.const import (
     CONF_ID,
     CONF_INPUT,
+    CONF_INTERRUPT_PIN,
     CONF_INVERTED,
     CONF_MODE,
     CONF_NUMBER,
@@ -27,6 +28,7 @@ CONFIG_SCHEMA = (
     cv.Schema(
         {
             cv.Required(CONF_ID): cv.declare_id(TCA9555Component),
+            cv.Optional(CONF_INTERRUPT_PIN): pins.internal_gpio_input_pin_schema,
         }
     )
     .extend(cv.COMPONENT_SCHEMA)
@@ -38,6 +40,8 @@ async def to_code(config):
     var = cg.new_Pvariable(config[CONF_ID])
     await cg.register_component(var, config)
     await i2c.register_i2c_device(var, config)
+    if interrupt_pin := config.get(CONF_INTERRUPT_PIN):
+        cg.add(var.set_interrupt_pin(await cg.gpio_pin_expression(interrupt_pin)))
 
 
 def validate_mode(value):

esphome/components/tca9555/tca9555.cpp

@@ -24,9 +24,18 @@ void TCA9555Component::setup() {
     this->mark_failed();
     return;
   }
+
+  if (this->interrupt_pin_ != nullptr) {
+    this->interrupt_pin_->setup();
+    this->interrupt_pin_->attach_interrupt(&TCA9555Component::gpio_intr, this, gpio::INTERRUPT_FALLING_EDGE);
+    this->set_invalidate_on_read_(false);
+  }
+  this->disable_loop();
 }
+void IRAM_ATTR TCA9555Component::gpio_intr(TCA9555Component *arg) { arg->enable_loop_soon_any_context(); }
 void TCA9555Component::dump_config() {
   ESP_LOGCONFIG(TAG, "TCA9555:");
+  LOG_PIN("  Interrupt Pin: ", this->interrupt_pin_);
   LOG_I2C_DEVICE(this)
   if (this->is_failed()) {
     ESP_LOGE(TAG, ESP_LOG_MSG_COMM_FAIL);
@@ -36,6 +45,9 @@ void TCA9555Component::pin_mode(uint8_t pin, gpio::Flags flags) {
   if (flags == gpio::FLAG_INPUT) {
     // Set mode mask bit
     this->mode_mask_ |= 1 << pin;
+    if (this->interrupt_pin_ == nullptr) {
+      this->enable_loop();
+    }
   } else if (flags == gpio::FLAG_OUTPUT) {
     // Clear mode mask bit
     this->mode_mask_ &= ~(1 << pin);
@@ -43,7 +55,12 @@ void TCA9555Component::pin_mode(uint8_t pin, gpio::Flags flags) {
   // Write GPIO to enable input mode
   this->write_gpio_modes_();
 }
-void TCA9555Component::loop() { this->reset_pin_cache_(); }
+void TCA9555Component::loop() {
+  this->reset_pin_cache_();
+  if (this->interrupt_pin_ != nullptr) {
+    this->disable_loop();
+  }
+}
 
 bool TCA9555Component::read_gpio_outputs_() {
   if (this->is_failed())

esphome/components/tca9555/tca9555.h

@@ -24,7 +24,10 @@ class TCA9555Component : public Component,
 
   void loop() override;
 
+  void set_interrupt_pin(InternalGPIOPin *pin) { this->interrupt_pin_ = pin; }
+
  protected:
+  static void IRAM_ATTR gpio_intr(TCA9555Component *arg);
   bool digital_read_hw(uint8_t pin) override;
   bool digital_read_cache(uint8_t pin) override;
   void digital_write_hw(uint8_t pin, bool value) override;
@@ -39,6 +42,8 @@ class TCA9555Component : public Component,
   bool read_gpio_modes_();
   bool write_gpio_modes_();
   bool read_gpio_outputs_();
+
+  InternalGPIOPin *interrupt_pin_{nullptr};
 };
 
 /// Helper class to expose a TCA9555 pin as an internal input GPIO pin.

esphome/const.py

@@ -4,7 +4,7 @@ from enum import Enum
 
 from esphome.enum import StrEnum
 
-__version__ = "2026.4.0b1"
+__version__ = "2026.4.0b2"
 
 ALLOWED_NAME_CHARS = "abcdefghijklmnopqrstuvwxyz0123456789-_"
 VALID_SUBSTITUTIONS_CHARACTERS = (

esphome/idf_component.yml

@@ -14,7 +14,7 @@ dependencies:
   espressif/esp32-camera:
     version: 2.1.6
   espressif/mdns:
-    version: 1.10.0
+    version: 1.11.0
   espressif/esp_wifi_remote:
     version: 1.4.0
     rules:

esphome/platformio_api.py

@@ -5,104 +5,15 @@ import os
 from pathlib import Path
 import re
 import subprocess
-import time
-from typing import Any
+import sys
 
 from esphome.const import CONF_COMPILE_PROCESS_LIMIT, CONF_ESPHOME, KEY_CORE
 from esphome.core import CORE, EsphomeError
-from esphome.util import run_external_command, run_external_process
+from esphome.util import run_external_process
 
 _LOGGER = logging.getLogger(__name__)
 
 
-def patch_structhash():
-    # Patch platformio's structhash to not recompile the entire project when files are
-    # removed/added. This might have unintended consequences, but this improves compile
-    # times greatly when adding/removing components and a simple clean build solves
-    # all issues
-    from platformio.run import cli, helpers
-
-    def patched_clean_build_dir(build_dir, *args):
-        from platformio import fs
-        from platformio.project.helpers import get_project_dir
-
-        platformio_ini = Path(get_project_dir()) / "platformio.ini"
-
-        build_dir = Path(build_dir)
-
-        # if project's config is modified
-        if (
-            build_dir.is_dir()
-            and platformio_ini.stat().st_mtime > build_dir.stat().st_mtime
-        ):
-            fs.rmtree(build_dir)
-
-        if not build_dir.is_dir():
-            build_dir.mkdir(parents=True)
-
-    helpers.clean_build_dir = patched_clean_build_dir
-    cli.clean_build_dir = patched_clean_build_dir
-
-
-def patch_file_downloader():
-    """Patch PlatformIO's FileDownloader to retry on PackageException errors.
-
-    PlatformIO's FileDownloader uses HTTPSession which lacks built-in retry
-    for 502/503 errors. We add retries with exponential backoff and close the
-    session between attempts to force a fresh TCP connection, which may route
-    to a different CDN edge node.
-    """
-    from platformio.package.download import FileDownloader
-    from platformio.package.exception import PackageException
-
-    if getattr(FileDownloader.__init__, "_esphome_patched", False):
-        return
-
-    original_init = FileDownloader.__init__
-
-    def patched_init(self, *args: Any, **kwargs: Any) -> None:
-        max_retries = 5
-
-        for attempt in range(max_retries):
-            try:
-                original_init(self, *args, **kwargs)
-                return
-            except PackageException as e:
-                if attempt < max_retries - 1:
-                    # Exponential backoff: 2, 4, 8, 16 seconds
-                    delay = 2 ** (attempt + 1)
-                    _LOGGER.warning(
-                        "Package download failed: %s. "
-                        "Retrying in %d seconds... (attempt %d/%d)",
-                        str(e),
-                        delay,
-                        attempt + 1,
-                        max_retries,
-                    )
-                    # Close the response and session to free resources
-                    # and force a new TCP connection on retry, which may
-                    # route to a different CDN edge node
-                    # pylint: disable=protected-access,broad-except
-                    try:
-                        if (
-                            hasattr(self, "_http_response")
-                            and self._http_response is not None
-                        ):
-                            self._http_response.close()
-                        if hasattr(self, "_http_session"):
-                            self._http_session.close()
-                    except Exception:
-                        pass
-                    # pylint: enable=protected-access,broad-except
-                    time.sleep(delay)
-                else:
-                    # Final attempt - re-raise
-                    raise
-
-    patched_init._esphome_patched = True  # type: ignore[attr-defined]  # pylint: disable=protected-access
-    FileDownloader.__init__ = patched_init
-
-
 IGNORE_LIB_WARNINGS = f"(?:{'|'.join(['Hash', 'Update'])})"
 FILTER_PLATFORMIO_LINES = [
     r"Verbose mode can be enabled via `-v, --verbose` option.*",
@@ -142,20 +53,6 @@ FILTER_PLATFORMIO_LINES = [
 ]
 
 
-class PlatformioLogFilter(logging.Filter):
-    """Filter to suppress noisy platformio log messages."""
-
-    _PATTERN = re.compile(
-        r"|".join(r"(?:" + pattern + r")" for pattern in FILTER_PLATFORMIO_LINES)
-    )
-
-    def filter(self, record: logging.LogRecord) -> bool:
-        # Only filter messages from platformio-related loggers
-        if "platformio" not in record.name.lower():
-            return True
-        return self._PATTERN.match(record.getMessage()) is None
-
-
 def run_platformio_cli(*args, **kwargs) -> str | int:
     os.environ["PLATFORMIO_FORCE_COLOR"] = "true"
     os.environ["PLATFORMIO_BUILD_DIR"] = str(CORE.relative_pioenvs_path().absolute())
@@ -166,30 +63,12 @@ def run_platformio_cli(*args, **kwargs) -> str | int:
     os.environ.setdefault("PYTHONWARNINGS", "ignore::SyntaxWarning")
     # Increase uv retry count to handle transient network errors (default is 3)
     os.environ.setdefault("UV_HTTP_RETRIES", "10")
-    cmd = ["platformio"] + list(args)
+    cmd = [sys.executable, "-m", "esphome.platformio_runner"] + list(args)
 
     if not CORE.verbose:
         kwargs["filter_lines"] = FILTER_PLATFORMIO_LINES
 
-    if os.environ.get("ESPHOME_USE_SUBPROCESS") is not None:
-        return run_external_process(*cmd, **kwargs)
-
-    import platformio.__main__
-
-    patch_structhash()
-    patch_file_downloader()
-
-    # Add log filter to suppress noisy platformio messages
-    log_filter = PlatformioLogFilter() if not CORE.verbose else None
-    if log_filter:
-        for handler in logging.getLogger().handlers:
-            handler.addFilter(log_filter)
-    try:
-        return run_external_command(platformio.__main__.main, *cmd, **kwargs)
-    finally:
-        if log_filter:
-            for handler in logging.getLogger().handlers:
-                handler.removeFilter(log_filter)
+    return run_external_process(*cmd, **kwargs)
 
 
 def run_platformio_cli_run(config, verbose, *args, **kwargs) -> str | int:

esphome/platformio_runner.py

@@ -0,0 +1,114 @@
+"""Subprocess entry point that applies ESPHome's PlatformIO patches.
+
+Invoked via ``python -m esphome.platformio_runner`` instead of
+``python -m platformio`` so that the patches (incremental rebuild
+preservation, download retries) apply inside the subprocess. Running
+PlatformIO in a subprocess keeps its ``sys.path`` mutations and other
+global state from leaking into the ESPHome process.
+"""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+import sys
+import time
+from typing import Any
+
+_LOGGER = logging.getLogger(__name__)
+
+
+def patch_structhash() -> None:
+    """Avoid full rebuilds when files are added or removed.
+
+    PlatformIO clears the build dir whenever its structure hash changes.
+    We replace that with an mtime check against ``platformio.ini`` so
+    incremental builds are preserved unless the project config changed.
+    """
+    from platformio.run import cli, helpers
+
+    def patched_clean_build_dir(build_dir, *_args):
+        from platformio import fs
+        from platformio.project.helpers import get_project_dir
+
+        platformio_ini = Path(get_project_dir()) / "platformio.ini"
+        build_dir = Path(build_dir)
+
+        if (
+            build_dir.is_dir()
+            and platformio_ini.stat().st_mtime > build_dir.stat().st_mtime
+        ):
+            fs.rmtree(build_dir)
+
+        if not build_dir.is_dir():
+            build_dir.mkdir(parents=True)
+
+    helpers.clean_build_dir = patched_clean_build_dir
+    cli.clean_build_dir = patched_clean_build_dir
+
+
+def patch_file_downloader() -> None:
+    """Retry PlatformIO package downloads with exponential backoff.
+
+    PlatformIO's ``FileDownloader`` uses an ``HTTPSession`` without built-in
+    retry for 502/503 errors. We wrap ``__init__`` to retry on
+    ``PackageException`` and close the session between attempts so a new
+    TCP connection can route to a different CDN edge node.
+    """
+    from platformio.package.download import FileDownloader
+    from platformio.package.exception import PackageException
+
+    if getattr(FileDownloader.__init__, "_esphome_patched", False):
+        return
+
+    original_init = FileDownloader.__init__
+
+    def patched_init(self, *args: Any, **kwargs: Any) -> None:
+        max_retries = 5
+
+        for attempt in range(max_retries):
+            try:
+                original_init(self, *args, **kwargs)
+                return
+            except PackageException as e:
+                if attempt < max_retries - 1:
+                    delay = 2 ** (attempt + 1)
+                    _LOGGER.warning(
+                        "Package download failed: %s. "
+                        "Retrying in %d seconds... (attempt %d/%d)",
+                        str(e),
+                        delay,
+                        attempt + 1,
+                        max_retries,
+                    )
+                    # pylint: disable=protected-access,broad-except
+                    try:
+                        if (
+                            hasattr(self, "_http_response")
+                            and self._http_response is not None
+                        ):
+                            self._http_response.close()
+                        if hasattr(self, "_http_session"):
+                            self._http_session.close()
+                    except Exception:
+                        pass
+                    # pylint: enable=protected-access,broad-except
+                    time.sleep(delay)
+                else:
+                    raise
+
+    patched_init._esphome_patched = True  # type: ignore[attr-defined]  # pylint: disable=protected-access
+    FileDownloader.__init__ = patched_init
+
+
+def main() -> int:
+    patch_structhash()
+    patch_file_downloader()
+
+    import platformio.__main__
+
+    return platformio.__main__.main() or 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

platformio.ini

@@ -133,10 +133,10 @@ extra_scripts = post:esphome/components/esp8266/post_build.py.script
 ; This are common settings for the ESP32 (all variants) using Arduino.
 [common:esp32-arduino]
 extends = common:arduino
-platform = https://github.com/pioarduino/platform-espressif32/releases/download/55.03.37/platform-espressif32.zip
+platform = https://github.com/pioarduino/platform-espressif32/releases/download/55.03.38/platform-espressif32.zip
 platform_packages =
-    pioarduino/framework-arduinoespressif32@https://github.com/espressif/arduino-esp32/releases/download/3.3.7/esp32-core-3.3.7.tar.xz
-    pioarduino/framework-espidf@https://github.com/pioarduino/esp-idf/releases/download/v5.5.3.1/esp-idf-v5.5.3.1.tar.xz
+    pioarduino/framework-arduinoespressif32@https://github.com/espressif/arduino-esp32/releases/download/3.3.8/esp32-core-3.3.8.tar.xz
+    pioarduino/framework-espidf@https://github.com/pioarduino/esp-idf/releases/download/v5.5.4/esp-idf-v5.5.4.tar.xz
 
 framework = arduino, espidf  ; Arduino as an ESP-IDF component
 lib_deps =
@@ -169,9 +169,9 @@ extra_scripts = post:esphome/components/esp32/post_build.py.script
 ; This are common settings for the ESP32 (all variants) using IDF.
 [common:esp32-idf]
 extends = common:idf
-platform = https://github.com/pioarduino/platform-espressif32/releases/download/55.03.37/platform-espressif32.zip
+platform = https://github.com/pioarduino/platform-espressif32/releases/download/55.03.38/platform-espressif32.zip
 platform_packages =
-    pioarduino/framework-espidf@https://github.com/pioarduino/esp-idf/releases/download/v5.5.3.1/esp-idf-v5.5.3.1.tar.xz
+    pioarduino/framework-espidf@https://github.com/pioarduino/esp-idf/releases/download/v5.5.4/esp-idf-v5.5.4.tar.xz
 
 framework = espidf
 lib_deps =

pyproject.toml

@@ -20,7 +20,6 @@ classifiers = [
     "Topic :: Home Automation",
 ]
 
-# Python 3.14 is not supported on Windows, see https://github.com/zephyrproject-rtos/windows-curses/issues/76
 requires-python = ">=3.11.0,<3.15"
 
 dynamic = ["dependencies", "optional-dependencies", "version"]

requirements.txt

@@ -12,7 +12,7 @@ platformio==6.1.19
 esptool==5.2.0
 click==8.3.2
 esphome-dashboard==20260408.1
-aioesphomeapi==44.12.0
+aioesphomeapi==44.13.3
 zeroconf==0.148.0
 puremagic==1.30
 ruamel.yaml==0.19.1 # dashboard_import

script/api_protobuf/api_protobuf.py

@@ -60,6 +60,10 @@ FILE_HEADER = """// This file was automatically generated with a tool.
 # Maps enum type name (e.g. ".BluetoothDeviceRequestType") to max enum value.
 _enum_max_values: dict[str, int] = {}
 
+# Populated by main() before message generation.
+# Maps message name (e.g. "BluetoothLERawAdvertisement") to its descriptor.
+_message_desc_map: dict[str, Any] = {}
+
 
 def indent_list(text: str, padding: str = "  ") -> list[str]:
     """Indent each line of the given text with the specified padding."""
@@ -427,6 +431,23 @@ class TypeInfo(ABC):
             Estimated size in bytes including field ID and typical data
         """
 
+    def get_max_encoded_size(self) -> int | None:
+        """Get the maximum possible encoded size in bytes for this field.
+
+        Returns the worst-case encoded size including field ID and maximum
+        possible value encoding. Returns None if the size is unbounded
+        (e.g., variable-length strings without max_data_length).
+
+        Used by (inline_encode) validation to ensure sub-messages fit in a
+        single-byte length varint (< 128 bytes).
+        """
+        return None  # Unbounded by default
+
+
+def _varint_max_size(bits: int) -> int:
+    """Return the maximum varint encoding size for a value with the given number of bits."""
+    return (max(bits, 1) + 6) // 7  # ceil(bits / 7), min 1 byte for varint(0)
+
 
 TYPE_INFO: dict[int, TypeInfo] = {}
 
@@ -514,8 +535,30 @@ def register_type(name: int):
     return func
 
 
+class FixedSizeTypeMixin:
+    """Mixin for types with a known fixed encoded size (float, double, fixed32, fixed64)."""
+
+    def get_max_encoded_size(self) -> int:
+        return self.calculate_field_id_size() + self.get_fixed_size_bytes()
+
+
+class VarintTypeMixin:
+    """Mixin for varint types. Subclasses set _varint_max_bits."""
+
+    _varint_max_bits: int = 64  # Default to worst case
+
+    def get_max_encoded_size(self) -> int:
+        max_val = self.max_value
+        if max_val is not None:
+            return self.calculate_field_id_size() + _varint_max_size(
+                max_val.bit_length() if max_val > 0 else 1
+            )
+        return self.calculate_field_id_size() + _varint_max_size(self._varint_max_bits)
+
+
 @register_type(1)
-class DoubleType(TypeInfo):
+class DoubleType(FixedSizeTypeMixin, TypeInfo):
+    # Unsupported but defined for completeness
     cpp_type = "double"
     default_value = "0.0"
     decode_64bit = "value.as_double()"
@@ -541,7 +584,7 @@ class DoubleType(TypeInfo):
 
 
 @register_type(2)
-class FloatType(TypeInfo):
+class FloatType(FixedSizeTypeMixin, TypeInfo):
     cpp_type = "float"
     default_value = "0.0f"
     decode_32bit = "value.as_float()"
@@ -567,8 +610,9 @@ class FloatType(TypeInfo):
 
 
 @register_type(3)
-class Int64Type(TypeInfo):
+class Int64Type(VarintTypeMixin, TypeInfo):
     cpp_type = "int64_t"
+    _varint_max_bits = 64
     default_value = "0"
     decode_varint = "static_cast<int64_t>(value)"
     encode_func = "encode_int64"
@@ -587,8 +631,9 @@ class Int64Type(TypeInfo):
 
 
 @register_type(4)
-class UInt64Type(TypeInfo):
+class UInt64Type(VarintTypeMixin, TypeInfo):
     cpp_type = "uint64_t"
+    _varint_max_bits = 64
     default_value = "0"
     decode_varint = "value"
     encode_func = "encode_uint64"
@@ -607,8 +652,9 @@ class UInt64Type(TypeInfo):
 
 
 @register_type(5)
-class Int32Type(TypeInfo):
+class Int32Type(VarintTypeMixin, TypeInfo):
     cpp_type = "int32_t"
+    _varint_max_bits = 64  # int32 is sign-extended to 64 bits in protobuf
     default_value = "0"
     decode_varint = "static_cast<int32_t>(value)"
     encode_func = "encode_int32"
@@ -627,7 +673,7 @@ class Int32Type(TypeInfo):
 
 
 @register_type(6)
-class Fixed64Type(TypeInfo):
+class Fixed64Type(FixedSizeTypeMixin, TypeInfo):
     cpp_type = "uint64_t"
     default_value = "0"
     decode_64bit = "value.as_fixed64()"
@@ -653,7 +699,7 @@ class Fixed64Type(TypeInfo):
 
 
 @register_type(7)
-class Fixed32Type(TypeInfo):
+class Fixed32Type(FixedSizeTypeMixin, TypeInfo):
     cpp_type = "uint32_t"
     default_value = "0"
     decode_32bit = "value.as_fixed32()"
@@ -689,7 +735,8 @@ class Fixed32Type(TypeInfo):
 
 
 @register_type(8)
-class BoolType(TypeInfo):
+class BoolType(VarintTypeMixin, TypeInfo):
+    _varint_max_bits = 1
     cpp_type = "bool"
     default_value = "false"
     decode_varint = "value != 0"
@@ -807,6 +854,16 @@ class StringType(TypeInfo):
     def get_estimated_size(self) -> int:
         return self.calculate_field_id_size() + 8  # field ID + 8 bytes typical string
 
+    def get_max_encoded_size(self) -> int | None:
+        max_len = self.max_data_length
+        if max_len is not None:
+            return (
+                self.calculate_field_id_size()
+                + _varint_max_size(max_len.bit_length())
+                + max_len
+            )
+        return None  # Unbounded
+
 
 @register_type(11)
 class MessageType(TypeInfo):
@@ -1122,6 +1179,16 @@ class PointerToStringBufferType(PointerToBufferTypeBase):
     def get_estimated_size(self) -> int:
         return self.calculate_field_id_size() + 8  # field ID + 8 bytes typical string
 
+    def get_max_encoded_size(self) -> int | None:
+        max_len = self.max_data_length
+        if max_len is not None:
+            return (
+                self.calculate_field_id_size()
+                + _varint_max_size(max_len.bit_length())
+                + max_len
+            )
+        return None
+
 
 class PackedBufferTypeInfo(TypeInfo):
     """Type for packed repeated fields that expose raw buffer instead of decoding.
@@ -1299,14 +1366,23 @@ class FixedArrayBytesType(TypeInfo):
             self.calculate_field_id_size() + 1 + 31
         )  # field ID + length byte + typical 31 bytes
 
+    def get_max_encoded_size(self) -> int:
+        # field_id + varint(array_size) + array_size
+        return (
+            self.calculate_field_id_size()
+            + _varint_max_size(self.array_size.bit_length())
+            + self.array_size
+        )
+
     @property
     def wire_type(self) -> WireType:
         return WireType.LENGTH_DELIMITED
 
 
 @register_type(13)
-class UInt32Type(TypeInfo):
+class UInt32Type(VarintTypeMixin, TypeInfo):
     cpp_type = "uint32_t"
+    _varint_max_bits = 32
     default_value = "0"
     decode_varint = "value"
     encode_func = "encode_uint32"
@@ -1328,7 +1404,9 @@ class UInt32Type(TypeInfo):
 
 
 @register_type(14)
-class EnumType(TypeInfo):
+class EnumType(VarintTypeMixin, TypeInfo):
+    _varint_max_bits = 32
+
     @property
     def cpp_type(self) -> str:
         return f"enums::{self._field.type_name[1:]}"
@@ -1379,7 +1457,7 @@ class EnumType(TypeInfo):
 
 
 @register_type(15)
-class SFixed32Type(TypeInfo):
+class SFixed32Type(FixedSizeTypeMixin, TypeInfo):
     cpp_type = "int32_t"
     default_value = "0"
     decode_32bit = "value.as_sfixed32()"
@@ -1405,7 +1483,7 @@ class SFixed32Type(TypeInfo):
 
 
 @register_type(16)
-class SFixed64Type(TypeInfo):
+class SFixed64Type(FixedSizeTypeMixin, TypeInfo):
     cpp_type = "int64_t"
     default_value = "0"
     decode_64bit = "value.as_sfixed64()"
@@ -1431,8 +1509,9 @@ class SFixed64Type(TypeInfo):
 
 
 @register_type(17)
-class SInt32Type(TypeInfo):
+class SInt32Type(VarintTypeMixin, TypeInfo):
     cpp_type = "int32_t"
+    _varint_max_bits = 32  # zigzag encoding keeps it 32-bit
     default_value = "0"
     decode_varint = "decode_zigzag32(static_cast<uint32_t>(value))"
     encode_func = "encode_sint32"
@@ -1451,8 +1530,9 @@ class SInt32Type(TypeInfo):
 
 
 @register_type(18)
-class SInt64Type(TypeInfo):
+class SInt64Type(VarintTypeMixin, TypeInfo):
     cpp_type = "int64_t"
+    _varint_max_bits = 64
     default_value = "0"
     decode_varint = "decode_zigzag64(value)"
     encode_func = "encode_sint64"
@@ -1500,6 +1580,91 @@ def _generate_array_dump_content(
     return o
 
 
+def _is_inline_encode(sub_msg_name: str) -> bool:
+    """Check if a sub-message type has the (inline_encode) option set."""
+    sub_desc = _message_desc_map.get(sub_msg_name)
+    if not sub_desc:
+        return False
+    inline_opt = getattr(pb, "inline_encode", None)
+    if inline_opt is None:
+        return False
+    return get_opt(sub_desc, inline_opt, False)
+
+
+def _generate_inline_encode_block(
+    field_number: int, sub_msg_name: str, element: str
+) -> str:
+    """Generate inline encode code for a sub-message with (inline_encode) = true.
+
+    Instead of calling encode_sub_message (function pointer indirection),
+    this inlines the sub-message's field encoding directly. Uses 1-byte
+    backpatch for the length (validated to be < 128 at generation time).
+
+    Uses a local reference alias 'sub_msg' to avoid issues with this-> replacement
+    on complex element expressions.
+
+    Args:
+        field_number: The parent field number for this sub-message
+        sub_msg_name: The sub-message type name
+        element: C++ expression for the element (e.g., "it" or "this->field[i]")
+    """
+    sub_desc = _message_desc_map[sub_msg_name]
+    tag = (field_number << 3) | 2  # wire type 2 = LENGTH_DELIMITED
+    assert tag < 128, f"inline_encode requires single-byte tag, got {tag}"
+
+    lines = []
+    lines.append(f"auto &sub_msg = {element};")
+    lines.append(f"ProtoEncode::write_raw_byte(pos, {tag});")
+    lines.append("uint8_t *len_pos = pos;")
+    lines.append("ProtoEncode::reserve_byte(pos);")
+
+    # Generate inline field encoding for each sub-message field
+    for field in sub_desc.field:
+        if field.options.deprecated:
+            continue
+        ti = create_field_type_info(field, needs_decode=False, needs_encode=True)
+        encode_line = ti.encode_content
+        # Replace this-> with sub_msg reference for the sub-message fields
+        encode_line = encode_line.replace("this->", "sub_msg.")
+        lines.extend(wrap_with_ifdef(encode_line, get_field_opt(field, pb.field_ifdef)))
+
+    lines.append("*len_pos = static_cast<uint8_t>(pos - len_pos - 1);")
+    return "\n".join(lines)
+
+
+def _generate_inline_size_block(
+    field_number: int, sub_msg_name: str, element: str
+) -> str:
+    """Generate inline size calculation for a sub-message with (inline_encode) = true.
+
+    Uses a local reference alias 'sub_msg' to avoid issues with this-> replacement
+    on complex element expressions like 'this->advertisements[i]'.
+
+    Args:
+        field_number: The parent field number for this sub-message
+        sub_msg_name: The sub-message type name
+        element: C++ expression for the element
+    """
+    sub_desc = _message_desc_map[sub_msg_name]
+
+    lines = []
+    lines.append(f"auto &sub_msg = {element};")
+    # 1 byte tag + 1 byte length (guaranteed < 128 by validation)
+    lines.append("size += 2;")
+
+    for field in sub_desc.field:
+        if field.options.deprecated:
+            continue
+        ti = create_field_type_info(field, needs_decode=False, needs_encode=True)
+        force = get_field_opt(field, pb.force, False)
+        size_line = ti.get_size_calculation(f"sub_msg.{ti.field_name}", force)
+        # Replace hardcoded this-> references (e.g., FixedArrayBytesType uses this->field_len)
+        size_line = size_line.replace("this->", "sub_msg.")
+        lines.extend(wrap_with_ifdef(size_line, get_field_opt(field, pb.field_ifdef)))
+
+    return "\n".join(lines)
+
+
 class FixedArrayRepeatedType(TypeInfo):
     """Special type for fixed-size repeated fields using std::array.
 
@@ -1526,6 +1691,10 @@ class FixedArrayRepeatedType(TypeInfo):
             return f"ProtoEncode::{self._ti.encode_func}(pos, {self.number}, static_cast<uint32_t>({element}), true);"
         # Repeated message elements use encode_sub_message (force=true is default)
         if isinstance(self._ti, MessageType):
+            if _is_inline_encode(self._ti.cpp_type):
+                return _generate_inline_encode_block(
+                    self.number, self._ti.cpp_type, element
+                )
             return f"ProtoEncode::encode_sub_message(pos, buffer, {self.number}, {element});"
         return (
             f"ProtoEncode::{self._ti.encode_func}(pos, {self.number}, {element}, true);"
@@ -1633,8 +1802,19 @@ class FixedArrayRepeatedType(TypeInfo):
             ]
             return f"if ({non_zero_checks}) {{\n" + "\n".join(size_lines) + "\n}"
 
+        is_inline = isinstance(self._ti, MessageType) and _is_inline_encode(
+            self._ti.cpp_type
+        )
+
         # When using a define, always use loop-based approach
         if self.is_define:
+            if is_inline:
+                o = f"for (const auto &it : {name}) {{\n"
+                o += indent(
+                    _generate_inline_size_block(self.number, self._ti.cpp_type, "it")
+                )
+                o += "\n}"
+                return o
             o = f"for (const auto &it : {name}) {{\n"
             o += f"  {self._ti.get_size_calculation('it', True)}\n"
             o += "}"
@@ -1642,6 +1822,14 @@ class FixedArrayRepeatedType(TypeInfo):
 
         # For fixed arrays, we always encode all elements
 
+        if is_inline:
+            o = f"for (const auto &it : {name}) {{\n"
+            o += indent(
+                _generate_inline_size_block(self.number, self._ti.cpp_type, "it")
+            )
+            o += "\n}"
+            return o
+
         # Special case for single-element arrays - no loop needed
         if self.array_size == 1:
             return self._ti.get_size_calculation(f"{name}[0]", True)
@@ -1714,6 +1902,15 @@ class FixedArrayWithLengthRepeatedType(FixedArrayRepeatedType):
 
     def get_size_calculation(self, name: str, force: bool = False) -> str:
         # Calculate size only for active elements
+        if isinstance(self._ti, MessageType) and _is_inline_encode(self._ti.cpp_type):
+            o = f"for (uint16_t i = 0; i < {name}_len; i++) {{\n"
+            o += indent(
+                _generate_inline_size_block(
+                    self.number, self._ti.cpp_type, f"{name}[i]"
+                )
+            )
+            o += "\n}"
+            return o
         o = f"for (uint16_t i = 0; i < {name}_len; i++) {{\n"
         o += f"  {self._ti.get_size_calculation(f'{name}[i]', True)}\n"
         o += "}"
@@ -2222,6 +2419,28 @@ def calculate_message_estimated_size(desc: descriptor.DescriptorProto) -> int:
     return total_size
 
 
+def calculate_message_max_size(desc: descriptor.DescriptorProto) -> int | None:
+    """Calculate the maximum possible encoded size for a message.
+
+    Returns None if any field has unbounded size (e.g., variable-length strings).
+    Used to validate that (inline_encode) messages fit in a single-byte length varint.
+    """
+    total_size = 0
+
+    for field in desc.field:
+        if field.options.deprecated:
+            continue
+
+        ti = create_field_type_info(field, needs_decode=False, needs_encode=True)
+        max_size = ti.get_max_encoded_size()
+        if max_size is None:
+            return None
+
+        total_size += max_size
+
+    return total_size
+
+
 def build_message_type(
     desc: descriptor.DescriptorProto,
     base_class_fields: dict[str, list[descriptor.FieldDescriptorProto]],
@@ -2451,11 +2670,23 @@ def build_message_type(
         prot = "void decode(const uint8_t *buffer, size_t length);"
         public_content.append(prot)
 
+    # Check if this message uses inline_encode — if so, skip generating standalone
+    # encode/calculate_size methods since the encoding is inlined into the parent.
+    inline_opt = getattr(pb, "inline_encode", None)
+    is_inline_only = (
+        message_id is None  # Not a service message (no id)
+        and inline_opt is not None
+        and get_opt(desc, inline_opt, False)
+    )
+
     # Only generate encode method if this message needs encoding and has fields
-    if needs_encode and encode:
+    if needs_encode and encode and not is_inline_only:
         # Add PROTO_ENCODE_DEBUG_ARG after pos in all proto_* calls
         encode_debug = [
-            line.replace("(pos,", "(pos PROTO_ENCODE_DEBUG_ARG,") for line in encode
+            line.replace("(pos,", "(pos PROTO_ENCODE_DEBUG_ARG,").replace(
+                "(pos)", "(pos PROTO_ENCODE_DEBUG_ARG)"
+            )
+            for line in encode
         ]
         o = f"uint8_t *{desc.name}::encode(ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_PARAM) const {{\n"
         o += "  uint8_t *__restrict__ pos = buffer.get_pos();\n"
@@ -2470,7 +2701,7 @@ def build_message_type(
     # If no fields to encode or message doesn't need encoding, the default implementation in ProtoMessage will be used
 
     # Add calculate_size method only if this message needs encoding and has fields
-    if needs_encode and size_calc:
+    if needs_encode and size_calc and not is_inline_only:
         o = f"uint32_t {desc.name}::calculate_size() const {{\n"
         o += "  uint32_t size = 0;\n"
         o += indent("\n".join(size_calc)) + "\n"
@@ -2830,6 +3061,32 @@ def main() -> None:
         if not enum.options.deprecated and enum.value:
             _enum_max_values[f".{enum.name}"] = max(v.number for v in enum.value)
 
+    # Build message descriptor map for inline_encode lookups
+    mt = file.message_type
+    _message_desc_map.update({m.name: m for m in mt if not m.options.deprecated})
+
+    # Validate inline_encode messages fit in single-byte length varint
+    inline_encode_opt = getattr(pb, "inline_encode", None)
+    if inline_encode_opt is not None:
+        for m in mt:
+            if m.options.deprecated:
+                continue
+            if not get_opt(m, inline_encode_opt, False):
+                continue
+            max_size = calculate_message_max_size(m)
+            if max_size is None:
+                raise ValueError(
+                    f"Message '{m.name}' has (inline_encode) = true but contains "
+                    f"fields with unbounded size. Inline encoding requires all "
+                    f"fields to have bounded maximum size."
+                )
+            if max_size >= 128:
+                raise ValueError(
+                    f"Message '{m.name}' has (inline_encode) = true but max "
+                    f"encoded size is {max_size} bytes (>= 128). Inline encoding "
+                    f"requires sub-messages that fit in a single-byte length varint."
+                )
+
     # Build dynamic ifdef mappings early so we can emit USE_API_VARINT64 before includes
     enum_ifdef_map, message_ifdef_map, message_source_map, used_messages = (
         build_type_usage_map(file)
@@ -3048,8 +3305,6 @@ static void dump_bytes_field(DumpBuffer &out, const char *field_name, const uint
 
     content += "\n}  // namespace enums\n\n"
 
-    mt = file.message_type
-
     # Identify empty SOURCE_CLIENT messages that don't need class generation
     for m in mt:
         if m.options.deprecated:

tests/component_tests/packages/test_packages.py

@@ -1,11 +1,18 @@
 """Tests for the packages component."""
 
+import logging
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 import pytest
 
-from esphome.components.packages import CONFIG_SCHEMA, do_packages_pass, merge_packages
+from esphome.components.packages import (
+    CONFIG_SCHEMA,
+    _walk_packages,
+    do_packages_pass,
+    is_package_definition,
+    merge_packages,
+)
 from esphome.components.substitutions import do_substitution_pass
 import esphome.config as config_module
 from esphome.config import resolve_extend_remove
@@ -37,7 +44,7 @@ from esphome.const import (
 )
 from esphome.core import CORE
 from esphome.util import OrderedDict
-from esphome.yaml_util import add_context
+from esphome.yaml_util import IncludeFile, add_context
 
 # Test strings
 TEST_DEVICE_NAME = "test_device_name"
@@ -79,6 +86,44 @@ def packages_pass(config):
     return config
 
 
+_INCLUDE_FILE = "INCLUDE_FILE"
+
+
+@pytest.mark.parametrize(
+    ("value", "expected"),
+    [
+        # IncludeFile objects are package definitions
+        (_INCLUDE_FILE, True),
+        # Git URL shorthand strings are package definitions
+        ("github://esphome/firmware/base.yaml@main", True),
+        # Remote package dicts (with url key) are package definitions
+        ({"url": "https://github.com/esphome/firmware", "file": "base.yaml"}, True),
+        # Plain config dicts are NOT package definitions (they are config fragments)
+        ({"wifi": {"ssid": "test"}}, False),
+        # None is not a package definition
+        (None, False),
+        # Lists are not package definitions
+        ([{"wifi": {"ssid": "test"}}], False),
+        # Empty dicts are not package definitions
+        ({}, False),
+    ],
+    ids=[
+        "include_file",
+        "git_shorthand",
+        "remote_package",
+        "config_fragment",
+        "none",
+        "list",
+        "empty_dict",
+    ],
+)
+def test_is_package_definition(value: object, expected: bool) -> None:
+    """Test that is_package_definition correctly identifies package definitions."""
+    if value is _INCLUDE_FILE:
+        value = MagicMock(spec=IncludeFile)
+    assert is_package_definition(value) is expected
+
+
 def test_package_unused(basic_esphome, basic_wifi) -> None:
     """
     Ensures do_package_pass does not change a config if packages aren't used.
@@ -1107,6 +1152,134 @@ def test_invalid_package_contents_masked_by_deprecation(
         do_packages_pass(config)
 
 
+def test_named_dict_with_include_files_no_false_deprecation_warning(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Package errors in named dicts must not trigger the deprecated fallback."""
+    good_include = MagicMock(spec=IncludeFile)
+    bad_include = MagicMock(spec=IncludeFile)
+
+    config = {
+        CONF_PACKAGES: {
+            "good_pkg": good_include,
+            "bad_pkg": bad_include,
+        },
+    }
+
+    call_count = 0
+
+    def failing_callback(package_config: dict, context: object) -> dict:
+        nonlocal call_count
+        call_count += 1
+        if call_count == 1:
+            # First package processes fine
+            return {CONF_WIFI: {CONF_SSID: "test"}}
+        # Second package has an error (e.g. jinja syntax error)
+        raise cv.Invalid("simulated jinja error in bad_pkg")
+
+    with (
+        caplog.at_level(logging.WARNING),
+        pytest.raises(cv.Invalid, match="simulated jinja error"),
+    ):
+        _walk_packages(config, failing_callback)
+
+    # Must NOT emit the deprecated single-package warning
+    assert "deprecated" not in caplog.text.lower()
+
+
+def test_validate_deprecated_false_raises_directly(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """With validate_deprecated=False, errors raise directly without fallback.
+
+    This is the codepath used for remote packages where _process_remote_package
+    returns already-resolved dicts that is_package_definition cannot detect.
+    """
+    config = {
+        CONF_PACKAGES: {
+            "pkg_a": {CONF_WIFI: {CONF_SSID: "test"}},
+            "pkg_b": {CONF_WIFI: {CONF_SSID: "test2"}},
+        },
+    }
+
+    call_count = 0
+
+    def failing_callback(package_config: dict, context: object) -> dict:
+        nonlocal call_count
+        call_count += 1
+        if call_count == 1:
+            return package_config
+        raise cv.Invalid("nested error")
+
+    with (
+        caplog.at_level(logging.WARNING),
+        pytest.raises(cv.Invalid, match="nested error"),
+    ):
+        _walk_packages(config, failing_callback, validate_deprecated=False)
+
+    assert "deprecated" not in caplog.text.lower()
+
+
+def test_error_on_first_declared_package_still_detected() -> None:
+    """When the first declared package errors, it's the last processed in reverse.
+
+    All other entries are already resolved to dicts, but the failing entry
+    retains its original IncludeFile value since assignment was skipped.
+    """
+    config = {
+        CONF_PACKAGES: {
+            "first_pkg": MagicMock(spec=IncludeFile),
+            "second_pkg": MagicMock(spec=IncludeFile),
+            "third_pkg": MagicMock(spec=IncludeFile),
+        },
+    }
+
+    call_count = 0
+
+    def fail_on_last(package_config: dict, context: object) -> dict:
+        nonlocal call_count
+        call_count += 1
+        # Reverse iteration: third_pkg (1), second_pkg (2), first_pkg (3)
+        if call_count < 3:
+            return {CONF_WIFI: {CONF_SSID: "test"}}
+        raise cv.Invalid("error in first_pkg")
+
+    with pytest.raises(cv.Invalid, match="error in first_pkg"):
+        _walk_packages(config, fail_on_last)
+
+
+def test_deprecated_single_package_fallback_still_works(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """The deprecated single-package form still falls back at the top level.
+
+    When a dict's values are plain config fragments (not package definitions)
+    and the callback fails, the deprecated fallback wraps the dict in a list
+    and retries with a deprecation warning.
+    """
+    config = {
+        CONF_PACKAGES: {
+            CONF_WIFI: {CONF_SSID: "test", CONF_PASSWORD: "secret"},
+        },
+    }
+
+    attempt = 0
+
+    def fail_then_succeed(package_config: dict, context: object) -> dict:
+        nonlocal attempt
+        attempt += 1
+        if attempt == 1:
+            # First attempt: treating as named dict fails
+            raise cv.Invalid("not a valid package")
+        # Second attempt: after fallback wraps as list, succeeds
+        return package_config
+
+    with caplog.at_level(logging.WARNING):
+        _walk_packages(config, fail_then_succeed)
+
+    assert "deprecated" in caplog.text.lower()
+
+
 def test_merge_packages_invalid_nested_type_raises() -> None:
     """Invalid nested packages type during merge raises cv.Invalid."""
     config = {

tests/components/canbus/common.yaml

@@ -50,6 +50,13 @@ button:
   - platform: template
     name: Canbus Actions
     on_press:
+      - canbus.send:
+          can_id: 0x601
+          data: [0, 1, 2]
+      - canbus.send:
+          can_id: 0x1FFFFFFF
+          use_extended_id: true
+          data: [0, 1, 2]
       - canbus.send: "abc"
       - canbus.send: [0, 1, 2]
       - canbus.send: !lambda return {0, 1, 2};

tests/components/lvgl/test.host.yaml

@@ -20,6 +20,7 @@ lvgl:
   - id: lvgl_0
     default_font: space16
     displays: sdl0
+    rotation: 180
     top_layer:
 
   - id: lvgl_1

tests/components/mcp23016/common.yaml

@@ -1,6 +1,10 @@
 mcp23016:
-  i2c_id: i2c_bus
-  id: mcp23016_hub
+  - i2c_id: i2c_bus
+    id: mcp23016_hub
+  - i2c_id: i2c_bus
+    id: mcp23016_hub_int
+    address: 0x21
+    interrupt_pin: ${interrupt_pin}
 
 binary_sensor:
   - platform: gpio

tests/components/mcp23016/test.esp32-idf.yaml

@@ -1,3 +1,6 @@
+substitutions:
+  interrupt_pin: GPIO15
+
 packages:
   i2c: !include ../../test_build_components/common/i2c/esp32-idf.yaml
 

tests/components/mcp23016/test.esp8266-ard.yaml

@@ -1,3 +1,6 @@
+substitutions:
+  interrupt_pin: GPIO15
+
 packages:
   i2c: !include ../../test_build_components/common/i2c/esp8266-ard.yaml
 

tests/components/mcp23016/test.rp2040-ard.yaml

@@ -1,3 +1,6 @@
+substitutions:
+  interrupt_pin: GPIO2
+
 packages:
   i2c: !include ../../test_build_components/common/i2c/rp2040-ard.yaml
 

tests/components/pca6416a/common.yaml

@@ -2,6 +2,10 @@ pca6416a:
   - id: pca6416a_hub
     i2c_id: i2c_bus
     address: 0x21
+  - id: pca6416a_hub_int
+    i2c_id: i2c_bus
+    address: 0x22
+    interrupt_pin: ${interrupt_pin}
 
 binary_sensor:
   - platform: gpio

tests/components/pca6416a/test.esp32-idf.yaml

@@ -1,3 +1,6 @@
+substitutions:
+  interrupt_pin: GPIO15
+
 packages:
   i2c: !include ../../test_build_components/common/i2c/esp32-idf.yaml
 

tests/components/pca6416a/test.esp8266-ard.yaml

@@ -1,3 +1,6 @@
+substitutions:
+  interrupt_pin: GPIO15
+
 packages:
   i2c: !include ../../test_build_components/common/i2c/esp8266-ard.yaml
 

tests/components/pca6416a/test.rp2040-ard.yaml

@@ -1,3 +1,6 @@
+substitutions:
+  interrupt_pin: GPIO2
+
 packages:
   i2c: !include ../../test_build_components/common/i2c/rp2040-ard.yaml
 

tests/components/tca9555/common.yaml

@@ -2,6 +2,10 @@ tca9555:
   - id: tca9555_hub
     i2c_id: i2c_bus
     address: 0x21
+  - id: tca9555_hub_int
+    i2c_id: i2c_bus
+    address: 0x22
+    interrupt_pin: ${interrupt_pin}
 
 binary_sensor:
   - platform: gpio

tests/components/tca9555/test.esp32-idf.yaml

@@ -1,3 +1,6 @@
+substitutions:
+  interrupt_pin: GPIO15
+
 packages:
   i2c: !include ../../test_build_components/common/i2c/esp32-idf.yaml
 

tests/components/tca9555/test.esp8266-ard.yaml

@@ -1,3 +1,6 @@
+substitutions:
+  interrupt_pin: GPIO15
+
 packages:
   i2c: !include ../../test_build_components/common/i2c/esp8266-ard.yaml
 

tests/components/tca9555/test.rp2040-ard.yaml

@@ -1,3 +1,6 @@
+substitutions:
+  interrupt_pin: GPIO2
+
 packages:
   i2c: !include ../../test_build_components/common/i2c/rp2040-ard.yaml
 

tests/unit_tests/conftest.py

@@ -84,9 +84,9 @@ def mock_decode_pc() -> Generator[Mock, None, None]:
 
 
 @pytest.fixture
-def mock_run_external_command() -> Generator[Mock, None, None]:
-    """Mock run_external_command for platformio_api."""
-    with patch("esphome.platformio_api.run_external_command") as mock:
+def mock_run_external_process() -> Generator[Mock, None, None]:
+    """Mock run_external_process for platformio_api."""
+    with patch("esphome.platformio_api.run_external_process") as mock:
         yield mock
 
 

tests/unit_tests/test_platformio_api.py

@@ -1,7 +1,8 @@
 """Tests for platformio_api.py path functions."""
 
+# pylint: disable=protected-access
+
 import json
-import logging
 import os
 from pathlib import Path
 import shutil
@@ -10,7 +11,7 @@ from unittest.mock import MagicMock, Mock, call, patch
 
 import pytest
 
-from esphome import platformio_api
+from esphome import platformio_api, platformio_runner
 from esphome.core import CORE, EsphomeError
 
 
@@ -281,13 +282,13 @@ def test_run_idedata_raises_on_invalid_json(
 
 
 def test_run_platformio_cli_sets_environment_variables(
-    setup_core: Path, mock_run_external_command: Mock
+    setup_core: Path, mock_run_external_process: Mock
 ) -> None:
     """Test run_platformio_cli sets correct environment variables."""
     CORE.build_path = str(setup_core / "build" / "test")
 
     with patch.dict(os.environ, {}, clear=False):
-        mock_run_external_command.return_value = 0
+        mock_run_external_process.return_value = 0
         platformio_api.run_platformio_cli("test", "arg")
 
         # Check environment variables were set
@@ -300,10 +301,12 @@ def test_run_platformio_cli_sets_environment_variables(
         assert "PLATFORMIO_LIBDEPS_DIR" in os.environ
         assert "PYTHONWARNINGS" in os.environ
 
-        # Check command was called correctly
-        mock_run_external_command.assert_called_once()
-        args = mock_run_external_command.call_args[0]
-        assert "platformio" in args
+        # Check command was called correctly — runs PlatformIO as a subprocess
+        # via the esphome.platformio_runner entry point.
+        mock_run_external_process.assert_called_once()
+        args = mock_run_external_process.call_args[0]
+        assert "-m" in args
+        assert "esphome.platformio_runner" in args
         assert "test" in args
         assert "arg" in args
 
@@ -444,7 +447,7 @@ def test_patch_structhash(setup_core: Path) -> None:
         },
     ):
         # Call patch_structhash
-        platformio_api.patch_structhash()
+        platformio_runner.patch_structhash()
 
         # Verify both modules had clean_build_dir patched
         # Check that clean_build_dir was set on both modules
@@ -496,7 +499,7 @@ def test_patched_clean_build_dir_removes_outdated(setup_core: Path) -> None:
         },
     ):
         # Call patch_structhash to install the patched function
-        platformio_api.patch_structhash()
+        platformio_runner.patch_structhash()
 
         # Call the patched function
         mock_helpers.clean_build_dir(str(build_dir), [])
@@ -546,7 +549,7 @@ def test_patched_clean_build_dir_keeps_updated(setup_core: Path) -> None:
         },
     ):
         # Call patch_structhash to install the patched function
-        platformio_api.patch_structhash()
+        platformio_runner.patch_structhash()
 
         # Call the patched function
         mock_helpers.clean_build_dir(str(build_dir), [])
@@ -594,7 +597,7 @@ def test_patched_clean_build_dir_creates_missing(setup_core: Path) -> None:
         },
     ):
         # Call patch_structhash to install the patched function
-        platformio_api.patch_structhash()
+        platformio_runner.patch_structhash()
 
         # Call the patched function
         mock_helpers.clean_build_dir(str(build_dir), [])
@@ -719,7 +722,7 @@ def test_patch_file_downloader_succeeds_first_try() -> None:
             ),
         },
     ):
-        platformio_api.patch_file_downloader()
+        platformio_runner.patch_file_downloader()
 
         from platformio.package.download import FileDownloader
 
@@ -758,7 +761,7 @@ def test_patch_file_downloader_retries_on_failure() -> None:
         ),
         patch("time.sleep") as mock_sleep,
     ):
-        platformio_api.patch_file_downloader()
+        platformio_runner.patch_file_downloader()
 
         from platformio.package.download import FileDownloader
 
@@ -799,7 +802,7 @@ def test_patch_file_downloader_raises_after_max_retries() -> None:
         ),
         patch("time.sleep") as mock_sleep,
     ):
-        platformio_api.patch_file_downloader()
+        platformio_runner.patch_file_downloader()
 
         from platformio.package.download import FileDownloader
 
@@ -847,7 +850,7 @@ def test_patch_file_downloader_closes_session_and_response_between_retries() ->
         ),
         patch("time.sleep"),
     ):
-        platformio_api.patch_file_downloader()
+        platformio_runner.patch_file_downloader()
 
         from platformio.package.download import FileDownloader
 
@@ -882,9 +885,9 @@ def test_patch_file_downloader_idempotent() -> None:
         },
     ):
         # Patch multiple times
-        platformio_api.patch_file_downloader()
-        platformio_api.patch_file_downloader()
-        platformio_api.patch_file_downloader()
+        platformio_runner.patch_file_downloader()
+        platformio_runner.patch_file_downloader()
+        platformio_runner.patch_file_downloader()
 
         from platformio.package.download import FileDownloader
 
@@ -895,19 +898,18 @@ def test_patch_file_downloader_idempotent() -> None:
         assert call_count == 1
 
 
-def test_platformio_log_filter_allows_non_platformio_messages() -> None:
-    """Test that non-platformio logger messages are allowed through."""
-    log_filter = platformio_api.PlatformioLogFilter()
-    record = logging.LogRecord(
-        name="esphome.core",
-        level=logging.INFO,
-        pathname="",
-        lineno=0,
-        msg="Some esphome message",
-        args=(),
-        exc_info=None,
+def _filter_through_redirect(line: str) -> str:
+    """Write a line through RedirectText with FILTER_PLATFORMIO_LINES and return what passes."""
+    import io
+
+    from esphome.util import RedirectText
+
+    captured = io.StringIO()
+    redirect = RedirectText(
+        captured, filter_lines=platformio_api.FILTER_PLATFORMIO_LINES
     )
-    assert log_filter.filter(record) is True
+    redirect.write(line + "\n")
+    return captured.getvalue()
 
 
 @pytest.mark.parametrize(
@@ -930,19 +932,9 @@ def test_platformio_log_filter_allows_non_platformio_messages() -> None:
         "Memory Usage -> https://bit.ly/pio-memory-usage",
     ],
 )
-def test_platformio_log_filter_blocks_noisy_messages(msg: str) -> None:
-    """Test that noisy platformio messages are filtered out."""
-    log_filter = platformio_api.PlatformioLogFilter()
-    record = logging.LogRecord(
-        name="platformio.builder",
-        level=logging.INFO,
-        pathname="",
-        lineno=0,
-        msg=msg,
-        args=(),
-        exc_info=None,
-    )
-    assert log_filter.filter(record) is False
+def test_filter_platformio_lines_blocks_noisy_messages(msg: str) -> None:
+    """Test that noisy platformio output lines are filtered out by RedirectText."""
+    assert _filter_through_redirect(msg) == ""
 
 
 @pytest.mark.parametrize(
@@ -954,39 +946,6 @@ def test_platformio_log_filter_blocks_noisy_messages(msg: str) -> None:
         "warning: unused variable",
     ],
 )
-def test_platformio_log_filter_allows_other_platformio_messages(msg: str) -> None:
-    """Test that non-noisy platformio messages are allowed through."""
-    log_filter = platformio_api.PlatformioLogFilter()
-    record = logging.LogRecord(
-        name="platformio.builder",
-        level=logging.INFO,
-        pathname="",
-        lineno=0,
-        msg=msg,
-        args=(),
-        exc_info=None,
-    )
-    assert log_filter.filter(record) is True
-
-
-@pytest.mark.parametrize(
-    "logger_name",
-    [
-        "PLATFORMIO.builder",
-        "PlatformIO.core",
-        "platformio.run",
-    ],
-)
-def test_platformio_log_filter_case_insensitive_logger_name(logger_name: str) -> None:
-    """Test that platformio logger name matching is case insensitive."""
-    log_filter = platformio_api.PlatformioLogFilter()
-    record = logging.LogRecord(
-        name=logger_name,
-        level=logging.INFO,
-        pathname="",
-        lineno=0,
-        msg="Found 5 compatible libraries",
-        args=(),
-        exc_info=None,
-    )
-    assert log_filter.filter(record) is False
+def test_filter_platformio_lines_allows_other_messages(msg: str) -> None:
+    """Test that non-noisy platformio output lines pass through RedirectText."""
+    assert _filter_through_redirect(msg) == msg + "\n"