Merge branch 'dev' from upstream

Resolved conflicts in:
- esphome/components/api/list_entities.h
- esphome/components/api/subscribe_state.h

Both conflicts were about NOLINT comment style - chose upstream's inline comment format.

CODEOWNERS

@@ -332,6 +332,7 @@ esphome/components/pca6416a/* @Mat931
 esphome/components/pca9554/* @clydebarrow @hwstar
 esphome/components/pcf85063/* @brogon
 esphome/components/pcf8563/* @KoenBreeman
+esphome/components/pi4ioe5v6408/* @jesserockz
 esphome/components/pid/* @OttoWinter
 esphome/components/pipsolar/* @andreashergert1984
 esphome/components/pm1006/* @habbie

esphome/components/api/client.py

@@ -35,8 +35,8 @@ async def async_run_logs(config: dict[str, Any], address: str) -> None:
     port: int = int(conf[CONF_PORT])
     password: str = conf[CONF_PASSWORD]
     noise_psk: str | None = None
-    if CONF_ENCRYPTION in conf:
-        noise_psk = conf[CONF_ENCRYPTION][CONF_KEY]
+    if (encryption := conf.get(CONF_ENCRYPTION)) and (key := encryption.get(CONF_KEY)):
+        noise_psk = key
     _LOGGER.info("Starting log output from %s using esphome API", address)
     cli = APIClient(
         address,

esphome/components/api/list_entities.h

@@ -11,9 +11,8 @@ class APIConnection;
 
 // Macro for generating ListEntitiesIterator handlers
 // Calls schedule_message_ with try_send_*_info
-// NOLINTNEXTLINE(bugprone-macro-parentheses)
 #define LIST_ENTITIES_HANDLER(entity_type, EntityClass, ResponseType) \
-  bool ListEntitiesIterator::on_##entity_type(EntityClass *entity) { \
+  bool ListEntitiesIterator::on_##entity_type(EntityClass *entity) { /* NOLINT(bugprone-macro-parentheses) */ \
     return this->client_->schedule_message_(entity, &APIConnection::try_send_##entity_type##_info, \
                                             ResponseType::MESSAGE_TYPE); \
   }

esphome/components/api/subscribe_state.h

@@ -12,9 +12,8 @@ class APIConnection;
 
 // Macro for generating InitialStateIterator handlers
 // Calls send_*_state
-// NOLINTNEXTLINE(bugprone-macro-parentheses)
 #define INITIAL_STATE_HANDLER(entity_type, EntityClass) \
-  bool InitialStateIterator::on_##entity_type(EntityClass *entity) { \
+  bool InitialStateIterator::on_##entity_type(EntityClass *entity) { /* NOLINT(bugprone-macro-parentheses) */ \
     return this->client_->send_##entity_type##_state(entity); \
   }
 

esphome/components/mqtt/mqtt_backend_esp32.cpp

@@ -6,6 +6,7 @@
 #include <string>
 #include "esphome/core/helpers.h"
 #include "esphome/core/log.h"
+#include "esphome/core/application.h"
 
 namespace esphome {
 namespace mqtt {
@@ -100,9 +101,24 @@ bool MQTTBackendESP32::initialize_() {
     handler_.reset(mqtt_client);
     is_initalized_ = true;
     esp_mqtt_client_register_event(mqtt_client, MQTT_EVENT_ANY, mqtt_event_handler, this);
+#if defined(USE_MQTT_IDF_ENQUEUE)
+    // Create the task only after MQTT client is initialized successfully
+    // Use larger stack size when TLS is enabled
+    size_t stack_size = this->ca_certificate_.has_value() ? TASK_STACK_SIZE_TLS : TASK_STACK_SIZE;
+    xTaskCreate(esphome_mqtt_task, "esphome_mqtt", stack_size, (void *) this, TASK_PRIORITY, &this->task_handle_);
+    if (this->task_handle_ == nullptr) {
+      ESP_LOGE(TAG, "Failed to create MQTT task");
+      // Clean up MQTT client since we can't start the async task
+      handler_.reset();
+      is_initalized_ = false;
+      return false;
+    }
+    // Set the task handle so the queue can notify it
+    this->mqtt_queue_.set_task_to_notify(this->task_handle_);
+#endif
     return true;
   } else {
-    ESP_LOGE(TAG, "Failed to initialize IDF-MQTT");
+    ESP_LOGE(TAG, "Failed to init client");
     return false;
   }
 }
@@ -115,6 +131,26 @@ void MQTTBackendESP32::loop() {
     mqtt_event_handler_(event);
     mqtt_events_.pop();
   }
+
+#if defined(USE_MQTT_IDF_ENQUEUE)
+  // Periodically log dropped messages to avoid blocking during spikes.
+  // During high load, many messages can be dropped in quick succession.
+  // Logging each drop immediately would flood the logs and potentially
+  // cause more drops if MQTT logging is enabled (cascade effect).
+  // Instead, we accumulate the count and log a summary periodically.
+  // IMPORTANT: Don't move this to the scheduler - if drops are due to memory
+  // pressure, the scheduler's heap allocations would make things worse.
+  uint32_t now = App.get_loop_component_start_time();
+  // Handle rollover: (now - last_time) works correctly with unsigned arithmetic
+  // even when now < last_time due to rollover
+  if ((now - this->last_dropped_log_time_) >= DROP_LOG_INTERVAL_MS) {
+    uint16_t dropped = this->mqtt_queue_.get_and_reset_dropped_count();
+    if (dropped > 0) {
+      ESP_LOGW(TAG, "Dropped %u messages (%us)", dropped, DROP_LOG_INTERVAL_MS / 1000);
+    }
+    this->last_dropped_log_time_ = now;
+  }
+#endif
 }
 
 void MQTTBackendESP32::mqtt_event_handler_(const Event &event) {
@@ -188,6 +224,86 @@ void MQTTBackendESP32::mqtt_event_handler(void *handler_args, esp_event_base_t b
   }
 }
 
+#if defined(USE_MQTT_IDF_ENQUEUE)
+void MQTTBackendESP32::esphome_mqtt_task(void *params) {
+  MQTTBackendESP32 *this_mqtt = (MQTTBackendESP32 *) params;
+
+  while (true) {
+    // Wait for notification indefinitely
+    ulTaskNotifyTake(pdTRUE, portMAX_DELAY);
+
+    // Process all queued items
+    struct QueueElement *elem;
+    while ((elem = this_mqtt->mqtt_queue_.pop()) != nullptr) {
+      if (this_mqtt->is_connected_) {
+        switch (elem->type) {
+          case MQTT_QUEUE_TYPE_SUBSCRIBE:
+            esp_mqtt_client_subscribe(this_mqtt->handler_.get(), elem->topic, elem->qos);
+            break;
+
+          case MQTT_QUEUE_TYPE_UNSUBSCRIBE:
+            esp_mqtt_client_unsubscribe(this_mqtt->handler_.get(), elem->topic);
+            break;
+
+          case MQTT_QUEUE_TYPE_PUBLISH:
+            esp_mqtt_client_publish(this_mqtt->handler_.get(), elem->topic, elem->payload, elem->payload_len, elem->qos,
+                                    elem->retain);
+            break;
+
+          default:
+            ESP_LOGE(TAG, "Invalid operation type from MQTT queue");
+            break;
+        }
+      }
+      this_mqtt->mqtt_event_pool_.release(elem);
+    }
+  }
+
+  // Clean up any remaining items in the queue
+  struct QueueElement *elem;
+  while ((elem = this_mqtt->mqtt_queue_.pop()) != nullptr) {
+    this_mqtt->mqtt_event_pool_.release(elem);
+  }
+
+  // Note: EventPool destructor will clean up the pool itself
+  // Task will delete itself
+  vTaskDelete(nullptr);
+}
+
+bool MQTTBackendESP32::enqueue_(MqttQueueTypeT type, const char *topic, int qos, bool retain, const char *payload,
+                                size_t len) {
+  auto *elem = this->mqtt_event_pool_.allocate();
+
+  if (!elem) {
+    // Queue is full - increment counter but don't log immediately.
+    // Logging here can cause a cascade effect: if MQTT logging is enabled,
+    // each dropped message would generate a log message, which could itself
+    // be sent via MQTT, causing more drops and more logs in a feedback loop
+    // that eventually triggers a watchdog reset. Instead, we log periodically
+    // in loop() to prevent blocking the event loop during spikes.
+    this->mqtt_queue_.increment_dropped_count();
+    return false;
+  }
+
+  elem->type = type;
+  elem->qos = qos;
+  elem->retain = retain;
+
+  // Use the helper to allocate and copy data
+  if (!elem->set_data(topic, payload, len)) {
+    // Allocation failed, return elem to pool
+    this->mqtt_event_pool_.release(elem);
+    // Increment counter without logging to avoid cascade effect during memory pressure
+    this->mqtt_queue_.increment_dropped_count();
+    return false;
+  }
+
+  // Push to queue - always succeeds since we allocated from the pool
+  this->mqtt_queue_.push(elem);
+  return true;
+}
+#endif  // USE_MQTT_IDF_ENQUEUE
+
 }  // namespace mqtt
 }  // namespace esphome
 #endif  // USE_ESP32

esphome/components/mqtt/mqtt_backend_esp32.h

@@ -6,9 +6,14 @@
 
 #include <string>
 #include <queue>
+#include <cstring>
 #include <mqtt_client.h>
+#include <freertos/FreeRTOS.h>
+#include <freertos/task.h>
 #include "esphome/components/network/ip_address.h"
 #include "esphome/core/helpers.h"
+#include "esphome/core/lock_free_queue.h"
+#include "esphome/core/event_pool.h"
 
 namespace esphome {
 namespace mqtt {
@@ -42,9 +47,79 @@ struct Event {
         error_handle(*event.error_handle) {}
 };
 
+enum MqttQueueTypeT : uint8_t {
+  MQTT_QUEUE_TYPE_NONE = 0,
+  MQTT_QUEUE_TYPE_SUBSCRIBE,
+  MQTT_QUEUE_TYPE_UNSUBSCRIBE,
+  MQTT_QUEUE_TYPE_PUBLISH,
+};
+
+struct QueueElement {
+  char *topic;
+  char *payload;
+  uint16_t payload_len;  // MQTT max payload is 64KiB
+  uint8_t type : 2;
+  uint8_t qos : 2;  // QoS only needs values 0-2
+  uint8_t retain : 1;
+  uint8_t reserved : 3;  // Reserved for future use
+
+  QueueElement() : topic(nullptr), payload(nullptr), payload_len(0), qos(0), retain(0), reserved(0) {}
+
+  // Helper to set topic/payload (uses RAMAllocator)
+  bool set_data(const char *topic_str, const char *payload_data, size_t len) {
+    // Check payload size limit (MQTT max is 64KiB)
+    if (len > std::numeric_limits<uint16_t>::max()) {
+      return false;
+    }
+
+    // Use RAMAllocator with default flags (tries external RAM first, falls back to internal)
+    RAMAllocator<char> allocator;
+
+    // Allocate and copy topic
+    size_t topic_len = strlen(topic_str) + 1;
+    topic = allocator.allocate(topic_len);
+    if (!topic)
+      return false;
+    memcpy(topic, topic_str, topic_len);
+
+    if (payload_data && len) {
+      payload = allocator.allocate(len);
+      if (!payload) {
+        allocator.deallocate(topic, topic_len);
+        topic = nullptr;
+        return false;
+      }
+      memcpy(payload, payload_data, len);
+      payload_len = static_cast<uint16_t>(len);
+    } else {
+      payload = nullptr;
+      payload_len = 0;
+    }
+    return true;
+  }
+
+  // Helper to release (uses RAMAllocator)
+  void release() {
+    RAMAllocator<char> allocator;
+    if (topic) {
+      allocator.deallocate(topic, strlen(topic) + 1);
+      topic = nullptr;
+    }
+    if (payload) {
+      allocator.deallocate(payload, payload_len);
+      payload = nullptr;
+    }
+    payload_len = 0;
+  }
+};
+
 class MQTTBackendESP32 final : public MQTTBackend {
  public:
   static const size_t MQTT_BUFFER_SIZE = 4096;
+  static const size_t TASK_STACK_SIZE = 2048;
+  static const size_t TASK_STACK_SIZE_TLS = 4096;  // Larger stack for TLS operations
+  static const ssize_t TASK_PRIORITY = 5;
+  static const uint8_t MQTT_QUEUE_LENGTH = 30;  // 30*12 bytes = 360
 
   void set_keep_alive(uint16_t keep_alive) final { this->keep_alive_ = keep_alive; }
   void set_client_id(const char *client_id) final { this->client_id_ = client_id; }
@@ -105,15 +180,23 @@ class MQTTBackendESP32 final : public MQTTBackend {
   }
 
   bool subscribe(const char *topic, uint8_t qos) final {
+#if defined(USE_MQTT_IDF_ENQUEUE)
+    return enqueue_(MQTT_QUEUE_TYPE_SUBSCRIBE, topic, qos);
+#else
     return esp_mqtt_client_subscribe(handler_.get(), topic, qos) != -1;
+#endif
+  }
+  bool unsubscribe(const char *topic) final {
+#if defined(USE_MQTT_IDF_ENQUEUE)
+    return enqueue_(MQTT_QUEUE_TYPE_UNSUBSCRIBE, topic);
+#else
+    return esp_mqtt_client_unsubscribe(handler_.get(), topic) != -1;
+#endif
   }
-  bool unsubscribe(const char *topic) final { return esp_mqtt_client_unsubscribe(handler_.get(), topic) != -1; }
 
   bool publish(const char *topic, const char *payload, size_t length, uint8_t qos, bool retain) final {
 #if defined(USE_MQTT_IDF_ENQUEUE)
-    // use the non-blocking version
-    // it can delay sending a couple of seconds but won't block
-    return esp_mqtt_client_enqueue(handler_.get(), topic, payload, length, qos, retain, true) != -1;
+    return enqueue_(MQTT_QUEUE_TYPE_PUBLISH, topic, qos, retain, payload, length);
 #else
     // might block for several seconds, either due to network timeout (10s)
     // or if publishing payloads longer than internal buffer (due to message fragmentation)
@@ -129,6 +212,12 @@ class MQTTBackendESP32 final : public MQTTBackend {
   void set_cl_key(const std::string &key) { cl_key_ = key; }
   void set_skip_cert_cn_check(bool skip_check) { skip_cert_cn_check_ = skip_check; }
 
+  // No destructor needed: ESPHome components live for the entire device runtime.
+  // The MQTT task and queue will run until the device reboots or loses power,
+  // at which point the entire process terminates and FreeRTOS cleans up all tasks.
+  // Implementing a destructor would add complexity and potential race conditions
+  // for a scenario that never occurs in practice.
+
  protected:
   bool initialize_();
   void mqtt_event_handler_(const Event &event);
@@ -160,6 +249,14 @@ class MQTTBackendESP32 final : public MQTTBackend {
   optional<std::string> cl_certificate_;
   optional<std::string> cl_key_;
   bool skip_cert_cn_check_{false};
+#if defined(USE_MQTT_IDF_ENQUEUE)
+  static void esphome_mqtt_task(void *params);
+  EventPool<struct QueueElement, MQTT_QUEUE_LENGTH> mqtt_event_pool_;
+  LockFreeQueue<struct QueueElement, MQTT_QUEUE_LENGTH> mqtt_queue_;
+  TaskHandle_t task_handle_{nullptr};
+  bool enqueue_(MqttQueueTypeT type, const char *topic, int qos = 0, bool retain = false, const char *payload = NULL,
+                size_t len = 0);
+#endif
 
   // callbacks
   CallbackManager<on_connect_callback_t> on_connect_;
@@ -169,6 +266,11 @@ class MQTTBackendESP32 final : public MQTTBackend {
   CallbackManager<on_message_callback_t> on_message_;
   CallbackManager<on_publish_user_callback_t> on_publish_;
   std::queue<Event> mqtt_events_;
+
+#if defined(USE_MQTT_IDF_ENQUEUE)
+  uint32_t last_dropped_log_time_{0};
+  static constexpr uint32_t DROP_LOG_INTERVAL_MS = 10000;  // Log every 10 seconds
+#endif
 };
 
 }  // namespace mqtt

esphome/components/pi4ioe5v6408/__init__.py

@@ -0,0 +1,84 @@
+from esphome import pins
+import esphome.codegen as cg
+from esphome.components import i2c
+import esphome.config_validation as cv
+from esphome.const import (
+    CONF_ID,
+    CONF_INPUT,
+    CONF_INVERTED,
+    CONF_MODE,
+    CONF_NUMBER,
+    CONF_OUTPUT,
+    CONF_PULLDOWN,
+    CONF_PULLUP,
+    CONF_RESET,
+)
+
+AUTO_LOAD = ["gpio_expander"]
+CODEOWNERS = ["@jesserockz"]
+DEPENDENCIES = ["i2c"]
+MULTI_CONF = True
+
+
+pi4ioe5v6408_ns = cg.esphome_ns.namespace("pi4ioe5v6408")
+PI4IOE5V6408Component = pi4ioe5v6408_ns.class_(
+    "PI4IOE5V6408Component", cg.Component, i2c.I2CDevice
+)
+PI4IOE5V6408GPIOPin = pi4ioe5v6408_ns.class_("PI4IOE5V6408GPIOPin", cg.GPIOPin)
+
+CONF_PI4IOE5V6408 = "pi4ioe5v6408"
+
+CONFIG_SCHEMA = (
+    cv.Schema(
+        {
+            cv.Required(CONF_ID): cv.declare_id(PI4IOE5V6408Component),
+            cv.Optional(CONF_RESET, default=True): cv.boolean,
+        }
+    )
+    .extend(cv.COMPONENT_SCHEMA)
+    .extend(i2c.i2c_device_schema(0x43))
+)
+
+
+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)
+
+    cg.add(var.set_reset(config[CONF_RESET]))
+
+
+def validate_mode(value):
+    if not (value[CONF_INPUT] or value[CONF_OUTPUT]):
+        raise cv.Invalid("Mode must be either input or output")
+    if value[CONF_INPUT] and value[CONF_OUTPUT]:
+        raise cv.Invalid("Mode must be either input or output")
+    return value
+
+
+PI4IOE5V6408_PIN_SCHEMA = pins.gpio_base_schema(
+    PI4IOE5V6408GPIOPin,
+    cv.int_range(min=0, max=7),
+    modes=[
+        CONF_INPUT,
+        CONF_OUTPUT,
+        CONF_PULLUP,
+        CONF_PULLDOWN,
+    ],
+    mode_validator=validate_mode,
+).extend(
+    {
+        cv.Required(CONF_PI4IOE5V6408): cv.use_id(PI4IOE5V6408Component),
+    }
+)
+
+
+@pins.PIN_SCHEMA_REGISTRY.register(CONF_PI4IOE5V6408, PI4IOE5V6408_PIN_SCHEMA)
+async def pi4ioe5v6408_pin_schema(config):
+    var = cg.new_Pvariable(config[CONF_ID])
+    await cg.register_parented(var, config[CONF_PI4IOE5V6408])
+
+    cg.add(var.set_pin(config[CONF_NUMBER]))
+    cg.add(var.set_inverted(config[CONF_INVERTED]))
+    cg.add(var.set_flags(pins.gpio_flags_expr(config[CONF_MODE])))
+    return var

esphome/components/pi4ioe5v6408/pi4ioe5v6408.cpp

@@ -0,0 +1,171 @@
+#include "pi4ioe5v6408.h"
+#include "esphome/core/log.h"
+
+namespace esphome {
+namespace pi4ioe5v6408 {
+
+static const uint8_t PI4IOE5V6408_REGISTER_DEVICE_ID = 0x01;
+static const uint8_t PI4IOE5V6408_REGISTER_IO_DIR = 0x03;
+static const uint8_t PI4IOE5V6408_REGISTER_OUT_SET = 0x05;
+static const uint8_t PI4IOE5V6408_REGISTER_OUT_HIGH_IMPEDENCE = 0x07;
+static const uint8_t PI4IOE5V6408_REGISTER_IN_DEFAULT_STATE = 0x09;
+static const uint8_t PI4IOE5V6408_REGISTER_PULL_ENABLE = 0x0B;
+static const uint8_t PI4IOE5V6408_REGISTER_PULL_SELECT = 0x0D;
+static const uint8_t PI4IOE5V6408_REGISTER_IN_STATE = 0x0F;
+static const uint8_t PI4IOE5V6408_REGISTER_INTERRUPT_ENABLE_MASK = 0x11;
+static const uint8_t PI4IOE5V6408_REGISTER_INTERRUPT_STATUS = 0x13;
+
+static const char *const TAG = "pi4ioe5v6408";
+
+void PI4IOE5V6408Component::setup() {
+  ESP_LOGCONFIG(TAG, "Running setup");
+  if (this->reset_) {
+    this->reg(PI4IOE5V6408_REGISTER_DEVICE_ID) |= 0b00000001;
+    this->reg(PI4IOE5V6408_REGISTER_OUT_HIGH_IMPEDENCE) = 0b00000000;
+  } else {
+    if (!this->read_gpio_modes_()) {
+      this->mark_failed();
+      ESP_LOGE(TAG, "Failed to read GPIO modes");
+      return;
+    }
+    if (!this->read_gpio_outputs_()) {
+      this->mark_failed();
+      ESP_LOGE(TAG, "Failed to read GPIO outputs");
+      return;
+    }
+  }
+}
+void PI4IOE5V6408Component::dump_config() {
+  ESP_LOGCONFIG(TAG, "PI4IOE5V6408:");
+  LOG_I2C_DEVICE(this)
+  if (this->is_failed()) {
+    ESP_LOGE(TAG, ESP_LOG_MSG_COMM_FAIL);
+  }
+}
+void PI4IOE5V6408Component::pin_mode(uint8_t pin, gpio::Flags flags) {
+  if (flags & gpio::FLAG_OUTPUT) {
+    // Set mode mask bit
+    this->mode_mask_ |= 1 << pin;
+  } else if (flags & gpio::FLAG_INPUT) {
+    // Clear mode mask bit
+    this->mode_mask_ &= ~(1 << pin);
+    if (flags & gpio::FLAG_PULLUP) {
+      this->pull_up_down_mask_ |= 1 << pin;
+      this->pull_enable_mask_ |= 1 << pin;
+    } else if (flags & gpio::FLAG_PULLDOWN) {
+      this->pull_up_down_mask_ &= ~(1 << pin);
+      this->pull_enable_mask_ |= 1 << pin;
+    }
+  }
+  // Write GPIO to enable input mode
+  this->write_gpio_modes_();
+}
+
+void PI4IOE5V6408Component::loop() { this->reset_pin_cache_(); }
+
+bool PI4IOE5V6408Component::read_gpio_outputs_() {
+  if (this->is_failed())
+    return false;
+
+  uint8_t data;
+  if (!this->read_byte(PI4IOE5V6408_REGISTER_OUT_SET, &data)) {
+    this->status_set_warning("Failed to read output register");
+    return false;
+  }
+  this->output_mask_ = data;
+  this->status_clear_warning();
+  return true;
+}
+
+bool PI4IOE5V6408Component::read_gpio_modes_() {
+  if (this->is_failed())
+    return false;
+
+  uint8_t data;
+  if (!this->read_byte(PI4IOE5V6408_REGISTER_IO_DIR, &data)) {
+    this->status_set_warning("Failed to read GPIO modes");
+    return false;
+  }
+#if ESPHOME_LOG_LEVEL >= ESPHOME_LOG_LEVEL_VERBOSE
+  ESP_LOGV(TAG, "Read GPIO modes: 0b" BYTE_TO_BINARY_PATTERN, BYTE_TO_BINARY(data));
+#endif
+  this->mode_mask_ = data;
+  this->status_clear_warning();
+  return true;
+}
+
+bool PI4IOE5V6408Component::digital_read_hw(uint8_t pin) {
+  if (this->is_failed())
+    return false;
+
+  uint8_t data;
+  if (!this->read_byte(PI4IOE5V6408_REGISTER_IN_STATE, &data)) {
+    this->status_set_warning("Failed to read GPIO state");
+    return false;
+  }
+  this->input_mask_ = data;
+  this->status_clear_warning();
+  return true;
+}
+
+void PI4IOE5V6408Component::digital_write_hw(uint8_t pin, bool value) {
+  if (this->is_failed())
+    return;
+
+  if (value) {
+    this->output_mask_ |= (1 << pin);
+  } else {
+    this->output_mask_ &= ~(1 << pin);
+  }
+  if (!this->write_byte(PI4IOE5V6408_REGISTER_OUT_SET, this->output_mask_)) {
+    this->status_set_warning("Failed to write output register");
+    return;
+  }
+#if ESPHOME_LOG_LEVEL >= ESPHOME_LOG_LEVEL_VERBOSE
+  ESP_LOGV(TAG, "Wrote GPIO output: 0b" BYTE_TO_BINARY_PATTERN, BYTE_TO_BINARY(this->output_mask_));
+#endif
+  this->status_clear_warning();
+}
+
+bool PI4IOE5V6408Component::write_gpio_modes_() {
+  if (this->is_failed())
+    return false;
+
+  if (!this->write_byte(PI4IOE5V6408_REGISTER_IO_DIR, this->mode_mask_)) {
+    this->status_set_warning("Failed to write GPIO modes");
+    return false;
+  }
+  if (!this->write_byte(PI4IOE5V6408_REGISTER_PULL_SELECT, this->pull_up_down_mask_)) {
+    this->status_set_warning("Failed to write GPIO pullup/pulldown");
+    return false;
+  }
+  if (!this->write_byte(PI4IOE5V6408_REGISTER_PULL_ENABLE, this->pull_enable_mask_)) {
+    this->status_set_warning("Failed to write GPIO pull enable");
+    return false;
+  }
+#if ESPHOME_LOG_LEVEL >= ESPHOME_LOG_LEVEL_VERBOSE
+  ESP_LOGV(TAG,
+           "Wrote GPIO modes: 0b" BYTE_TO_BINARY_PATTERN "\n"
+           "Wrote GPIO pullup/pulldown: 0b" BYTE_TO_BINARY_PATTERN "\n"
+           "Wrote GPIO pull enable: 0b" BYTE_TO_BINARY_PATTERN,
+           BYTE_TO_BINARY(this->mode_mask_), BYTE_TO_BINARY(this->pull_up_down_mask_),
+           BYTE_TO_BINARY(this->pull_enable_mask_));
+#endif
+  this->status_clear_warning();
+  return true;
+}
+
+bool PI4IOE5V6408Component::digital_read_cache(uint8_t pin) { return (this->input_mask_ & (1 << pin)); }
+
+float PI4IOE5V6408Component::get_setup_priority() const { return setup_priority::IO; }
+
+void PI4IOE5V6408GPIOPin::setup() { this->pin_mode(this->flags_); }
+void PI4IOE5V6408GPIOPin::pin_mode(gpio::Flags flags) { this->parent_->pin_mode(this->pin_, flags); }
+bool PI4IOE5V6408GPIOPin::digital_read() { return this->parent_->digital_read(this->pin_) != this->inverted_; }
+void PI4IOE5V6408GPIOPin::digital_write(bool value) {
+  this->parent_->digital_write(this->pin_, value != this->inverted_);
+}
+std::string PI4IOE5V6408GPIOPin::dump_summary() const { return str_sprintf("%u via PI4IOE5V6408", this->pin_); }
+
+}  // namespace pi4ioe5v6408
+}  // namespace esphome

esphome/components/pi4ioe5v6408/pi4ioe5v6408.h

@@ -0,0 +1,70 @@
+#pragma once
+
+#include "esphome/components/gpio_expander/cached_gpio.h"
+#include "esphome/components/i2c/i2c.h"
+#include "esphome/core/component.h"
+#include "esphome/core/hal.h"
+
+namespace esphome {
+namespace pi4ioe5v6408 {
+class PI4IOE5V6408Component : public Component,
+                              public i2c::I2CDevice,
+                              public gpio_expander::CachedGpioExpander<uint8_t, 8> {
+ public:
+  PI4IOE5V6408Component() = default;
+
+  void setup() override;
+  void pin_mode(uint8_t pin, gpio::Flags flags);
+
+  float get_setup_priority() const override;
+  void dump_config() override;
+  void loop() override;
+
+  /// Indicate if the component should reset the state during setup
+  void set_reset(bool reset) { this->reset_ = reset; }
+
+ protected:
+  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;
+
+  /// Mask for the pin mode - 1 means output, 0 means input
+  uint8_t mode_mask_{0x00};
+  /// The mask to write as output state - 1 means HIGH, 0 means LOW
+  uint8_t output_mask_{0x00};
+  /// The state read in digital_read_hw - 1 means HIGH, 0 means LOW
+  uint8_t input_mask_{0x00};
+  /// The mask to write as input buffer state - 1 means enabled, 0 means disabled
+  uint8_t pull_enable_mask_{0x00};
+  /// The mask to write as pullup state - 1 means pullup, 0 means pulldown
+  uint8_t pull_up_down_mask_{0x00};
+
+  bool reset_{true};
+
+  bool read_gpio_modes_();
+  bool write_gpio_modes_();
+  bool read_gpio_outputs_();
+};
+
+class PI4IOE5V6408GPIOPin : public GPIOPin, public Parented<PI4IOE5V6408Component> {
+ public:
+  void setup() override;
+  void pin_mode(gpio::Flags flags) override;
+  bool digital_read() override;
+  void digital_write(bool value) override;
+  std::string dump_summary() const override;
+
+  void set_pin(uint8_t pin) { this->pin_ = pin; }
+  void set_inverted(bool inverted) { this->inverted_ = inverted; }
+  void set_flags(gpio::Flags flags) { this->flags_ = flags; }
+
+  gpio::Flags get_flags() const override { return this->flags_; }
+
+ protected:
+  uint8_t pin_;
+  bool inverted_;
+  gpio::Flags flags_;
+};
+
+}  // namespace pi4ioe5v6408
+}  // namespace esphome

esphome/components/uart/uart_component_esp_idf.cpp

@@ -42,7 +42,7 @@ uart_config_t IDFUARTComponent::get_config_() {
       break;
   }
 
-  uart_config_t uart_config;
+  uart_config_t uart_config{};
   uart_config.baud_rate = this->baud_rate_;
   uart_config.data_bits = data_bits;
   uart_config.parity = parity;

esphome/components/web_server/web_server.cpp

@@ -1638,12 +1638,15 @@ void WebServer::handle_event_request(AsyncWebServerRequest *request, const UrlMa
   request->send(404);
 }
 
+static std::string get_event_type(event::Event *event) { return event->last_event_type ? *event->last_event_type : ""; }
+
 std::string WebServer::event_state_json_generator(WebServer *web_server, void *source) {
-  return web_server->event_json((event::Event *) (source), *(((event::Event *) (source))->last_event_type),
-                                DETAIL_STATE);
+  auto *event = static_cast<event::Event *>(source);
+  return web_server->event_json(event, get_event_type(event), DETAIL_STATE);
 }
 std::string WebServer::event_all_json_generator(WebServer *web_server, void *source) {
-  return web_server->event_json((event::Event *) (source), *(((event::Event *) (source))->last_event_type), DETAIL_ALL);
+  auto *event = static_cast<event::Event *>(source);
+  return web_server->event_json(event, get_event_type(event), DETAIL_ALL);
 }
 std::string WebServer::event_json(event::Event *obj, const std::string &event_type, JsonDetail start_config) {
   return json::build_json([this, obj, event_type, start_config](JsonObject root) {

esphome/components/web_server_idf/web_server_idf.cpp

@@ -292,21 +292,38 @@ void AsyncEventSource::handleRequest(AsyncWebServerRequest *request) {
 }
 
 void AsyncEventSource::loop() {
-  for (auto *ses : this->sessions_) {
-    ses->loop();
+  // Clean up dead sessions safely
+  // This follows the ESP-IDF pattern where free_ctx marks resources as dead
+  // and the main loop handles the actual cleanup to avoid race conditions
+  auto it = this->sessions_.begin();
+  while (it != this->sessions_.end()) {
+    auto *ses = *it;
+    // If the session has a dead socket (marked by destroy callback)
+    if (ses->fd_.load() == 0) {
+      ESP_LOGD(TAG, "Removing dead event source session");
+      it = this->sessions_.erase(it);
+      delete ses;  // NOLINT(cppcoreguidelines-owning-memory)
+    } else {
+      ses->loop();
+      ++it;
+    }
   }
 }
 
 void AsyncEventSource::try_send_nodefer(const char *message, const char *event, uint32_t id, uint32_t reconnect) {
   for (auto *ses : this->sessions_) {
-    ses->try_send_nodefer(message, event, id, reconnect);
+    if (ses->fd_.load() != 0) {  // Skip dead sessions
+      ses->try_send_nodefer(message, event, id, reconnect);
+    }
   }
 }
 
 void AsyncEventSource::deferrable_send_state(void *source, const char *event_type,
                                              message_generator_t *message_generator) {
   for (auto *ses : this->sessions_) {
-    ses->deferrable_send_state(source, event_type, message_generator);
+    if (ses->fd_.load() != 0) {  // Skip dead sessions
+      ses->deferrable_send_state(source, event_type, message_generator);
+    }
   }
 }
 
@@ -331,7 +348,7 @@ AsyncEventSourceResponse::AsyncEventSourceResponse(const AsyncWebServerRequest *
   req->free_ctx = AsyncEventSourceResponse::destroy;
 
   this->hd_ = req->handle;
-  this->fd_ = httpd_req_to_sockfd(req);
+  this->fd_.store(httpd_req_to_sockfd(req));
 
   // Configure reconnect timeout and send config
   // this should always go through since the tcp send buffer is empty on connect
@@ -362,8 +379,10 @@ AsyncEventSourceResponse::AsyncEventSourceResponse(const AsyncWebServerRequest *
 
 void AsyncEventSourceResponse::destroy(void *ptr) {
   auto *rsp = static_cast<AsyncEventSourceResponse *>(ptr);
-  rsp->server_->sessions_.erase(rsp);
-  delete rsp;  // NOLINT(cppcoreguidelines-owning-memory)
+  ESP_LOGD(TAG, "Event source connection closed (fd: %d)", rsp->fd_.load());
+  // Mark as dead by setting fd to 0 - will be cleaned up in the main loop
+  rsp->fd_.store(0);
+  // Note: We don't delete or remove from set here to avoid race conditions
 }
 
 // helper for allowing only unique entries in the queue
@@ -403,9 +422,11 @@ void AsyncEventSourceResponse::process_buffer_() {
     return;
   }
 
-  int bytes_sent = httpd_socket_send(this->hd_, this->fd_, event_buffer_.c_str() + event_bytes_sent_,
+  int bytes_sent = httpd_socket_send(this->hd_, this->fd_.load(), event_buffer_.c_str() + event_bytes_sent_,
                                      event_buffer_.size() - event_bytes_sent_, 0);
   if (bytes_sent == HTTPD_SOCK_ERR_TIMEOUT || bytes_sent == HTTPD_SOCK_ERR_FAIL) {
+    // Socket error - just return, the connection will be closed by httpd
+    // and our destroy callback will be called
     return;
   }
   event_bytes_sent_ += bytes_sent;
@@ -425,7 +446,7 @@ void AsyncEventSourceResponse::loop() {
 
 bool AsyncEventSourceResponse::try_send_nodefer(const char *message, const char *event, uint32_t id,
                                                 uint32_t reconnect) {
-  if (this->fd_ == 0) {
+  if (this->fd_.load() == 0) {
     return false;
   }
 

esphome/components/web_server_idf/web_server_idf.h

@@ -4,6 +4,7 @@
 #include "esphome/core/defines.h"
 #include <esp_http_server.h>
 
+#include <atomic>
 #include <functional>
 #include <list>
 #include <map>
@@ -271,7 +272,7 @@ class AsyncEventSourceResponse {
   static void destroy(void *p);
   AsyncEventSource *server_;
   httpd_handle_t hd_{};
-  int fd_{};
+  std::atomic<int> fd_{};
   std::vector<DeferredEvent> deferred_queue_;
   esphome::web_server::WebServer *web_server_;
   std::unique_ptr<esphome::web_server::ListEntitiesIterator> entities_iterator_;

esphome/core/application.cpp

@@ -380,7 +380,7 @@ void Application::enable_pending_loops_() {
 
     // Clear the pending flag and enable the loop
     component->pending_enable_loop_ = false;
-    ESP_LOGD(TAG, "%s loop enabled from ISR", component->get_component_source());
+    ESP_LOGVV(TAG, "%s loop enabled from ISR", component->get_component_source());
     component->component_state_ &= ~COMPONENT_STATE_MASK;
     component->component_state_ |= COMPONENT_STATE_LOOP;
 

esphome/core/component.cpp

@@ -165,7 +165,7 @@ void Component::mark_failed() {
 }
 void Component::disable_loop() {
   if ((this->component_state_ & COMPONENT_STATE_MASK) != COMPONENT_STATE_LOOP_DONE) {
-    ESP_LOGD(TAG, "%s loop disabled", this->get_component_source());
+    ESP_LOGVV(TAG, "%s loop disabled", this->get_component_source());
     this->component_state_ &= ~COMPONENT_STATE_MASK;
     this->component_state_ |= COMPONENT_STATE_LOOP_DONE;
     App.disable_component_loop_(this);
@@ -173,7 +173,7 @@ void Component::disable_loop() {
 }
 void Component::enable_loop() {
   if ((this->component_state_ & COMPONENT_STATE_MASK) == COMPONENT_STATE_LOOP_DONE) {
-    ESP_LOGD(TAG, "%s loop enabled", this->get_component_source());
+    ESP_LOGVV(TAG, "%s loop enabled", this->get_component_source());
     this->component_state_ &= ~COMPONENT_STATE_MASK;
     this->component_state_ |= COMPONENT_STATE_LOOP;
     App.enable_component_loop_(this);

requirements.txt

@@ -13,7 +13,7 @@ platformio==6.1.18  # When updating platformio, also update /docker/Dockerfile
 esptool==4.9.0
 click==8.1.7
 esphome-dashboard==20250514.0
-aioesphomeapi==33.1.1
+aioesphomeapi==34.0.0
 zeroconf==0.147.0
 puremagic==1.29
 ruamel.yaml==0.18.14 # dashboard_import

tests/components/pi4ioe5v6408/common.yaml

@@ -0,0 +1,22 @@
+i2c:
+  id: i2c_pi4ioe5v6408
+  sda: ${i2c_sda}
+  scl: ${i2c_scl}
+
+pi4ioe5v6408:
+  id: pi4ioe1
+  address: 0x44
+
+switch:
+  - platform: gpio
+    id: switch1
+    pin:
+      pi4ioe5v6408: pi4ioe1
+      number: 0
+
+binary_sensor:
+  - platform: gpio
+    id: sensor1
+    pin:
+      pi4ioe5v6408: pi4ioe1
+      number: 1

tests/components/pi4ioe5v6408/test.esp32-ard.yaml

@@ -0,0 +1,5 @@
+substitutions:
+  i2c_sda: GPIO21
+  i2c_scl: GPIO22
+
+<<: !include common.yaml

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

@@ -0,0 +1,5 @@
+substitutions:
+  i2c_sda: GPIO21
+  i2c_scl: GPIO22
+
+<<: !include common.yaml

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

@@ -0,0 +1,5 @@
+substitutions:
+  i2c_sda: GPIO4
+  i2c_scl: GPIO5
+
+<<: !include common.yaml