Merge remote-tracking branch 'origin/dev' into HEAD

esphome/components/feedback/cover.py

@@ -27,13 +27,13 @@ CONF_ACCELERATION_WAIT_TIME = "acceleration_wait_time"
 CONF_OVERSHOOT_DURATION = "overshoot_duration"
 CONF_OBSTACLE_ROLLBACK = "obstacle_rollback"
 
-endstop_ns = cg.esphome_ns.namespace("feedback")
-FeedbackCover = endstop_ns.class_("FeedbackCover", cover.Cover, cg.Component)
+feedback_ns = cg.esphome_ns.namespace("feedback")
+FeedbackCover = feedback_ns.class_("FeedbackCover", cover.Cover, cg.Component)
 
 
 def validate_infer_endstop(config: ConfigType) -> ConfigType:
-    if config[CONF_INFER_ENDSTOP_FROM_MOVEMENT] is True:
-        if config[CONF_HAS_BUILT_IN_ENDSTOP] is False:
+    if config[CONF_INFER_ENDSTOP_FROM_MOVEMENT]:
+        if not config[CONF_HAS_BUILT_IN_ENDSTOP]:
             raise cv.Invalid(
                 f"{CONF_INFER_ENDSTOP_FROM_MOVEMENT} can only be set if {CONF_HAS_BUILT_IN_ENDSTOP} is also set"
             )
@@ -119,58 +119,52 @@ async def to_code(config: ConfigType) -> None:
     var = await cover.new_cover(config)
     await cg.register_component(var, config)
 
-    # STOP
     await automation.build_automation(
         var.get_stop_trigger(), [], config[CONF_STOP_ACTION]
     )
-
-    # OPEN
     await automation.build_automation(
         var.get_open_trigger(), [], config[CONF_OPEN_ACTION]
     )
-    cg.add(var.set_open_duration(config[CONF_OPEN_DURATION]))
-    await cg.add_variable(var.set_open_endstop, config.get(CONF_OPEN_ENDSTOP))
-    await cg.add_variable(var.set_open_sensor, config.get(CONF_OPEN_SENSOR))
-    await cg.add_variable(
-        var.set_open_obstacle_sensor, config.get(CONF_OPEN_OBSTACLE_SENSOR)
-    )
-
-    # CLOSE
     await automation.build_automation(
         var.get_close_trigger(), [], config[CONF_CLOSE_ACTION]
     )
+    cg.add(var.set_open_duration(config[CONF_OPEN_DURATION]))
     cg.add(var.set_close_duration(config[CONF_CLOSE_DURATION]))
-    await cg.add_variable(var.set_close_endstop, config.get(CONF_CLOSE_ENDSTOP))
-    await cg.add_variable(var.set_close_sensor, config.get(CONF_CLOSE_SENSOR))
-    await cg.add_variable(
-        var.set_close_obstacle_sensor, config.get(CONF_CLOSE_OBSTACLE_SENSOR)
-    )
 
-    # OTHER
+    if CONF_OPEN_ENDSTOP in config or CONF_CLOSE_ENDSTOP in config:
+        cg.add_define("USE_FEEDBACK_COVER_ENDSTOP")
+        await cg.add_variable(var.set_open_endstop, config.get(CONF_OPEN_ENDSTOP))
+        await cg.add_variable(var.set_close_endstop, config.get(CONF_CLOSE_ENDSTOP))
+    # validation requires both movement sensors or neither
+    if CONF_OPEN_SENSOR in config:
+        cg.add_define("USE_FEEDBACK_COVER_MOVEMENT")
+        await cg.add_variable(var.set_open_sensor, config[CONF_OPEN_SENSOR])
+        await cg.add_variable(var.set_close_sensor, config[CONF_CLOSE_SENSOR])
+        cg.add(var.set_infer_endstop(config[CONF_INFER_ENDSTOP_FROM_MOVEMENT]))
+    if CONF_OPEN_OBSTACLE_SENSOR in config or CONF_CLOSE_OBSTACLE_SENSOR in config:
+        cg.add_define("USE_FEEDBACK_COVER_OBSTACLE")
+        await cg.add_variable(
+            var.set_open_obstacle_sensor, config.get(CONF_OPEN_OBSTACLE_SENSOR)
+        )
+        await cg.add_variable(
+            var.set_close_obstacle_sensor, config.get(CONF_CLOSE_OBSTACLE_SENSOR)
+        )
+        cg.add(var.set_obstacle_rollback(config[CONF_OBSTACLE_ROLLBACK]))
+
     cg.add_optional(var.set_max_duration, config.get(CONF_MAX_DURATION))
-
     cg.add(var.set_has_built_in_endstop(config[CONF_HAS_BUILT_IN_ENDSTOP]))
-
-    if CONF_ASSUMED_STATE in config:
-        cg.add(var.set_assumed_state(config[CONF_ASSUMED_STATE]))
-    else:
-        cg.add(
-            var.set_assumed_state(
-                not (
-                    (CONF_CLOSE_ENDSTOP in config and CONF_OPEN_ENDSTOP in config)
-                    or config[CONF_INFER_ENDSTOP_FROM_MOVEMENT]
-                )
-            )
+    assumed_state = config.get(CONF_ASSUMED_STATE)
+    if assumed_state is None:
+        assumed_state = not (
+            (CONF_CLOSE_ENDSTOP in config and CONF_OPEN_ENDSTOP in config)
+            or config[CONF_INFER_ENDSTOP_FROM_MOVEMENT]
         )
-
+    cg.add(var.set_assumed_state(assumed_state))
     cg.add(var.set_update_interval(config[CONF_UPDATE_INTERVAL]))
-    cg.add(var.set_infer_endstop(config[CONF_INFER_ENDSTOP_FROM_MOVEMENT]))
-    if CONF_DIRECTION_CHANGE_WAIT_TIME in config:
-        cg.add(
-            var.set_direction_change_waittime(config[CONF_DIRECTION_CHANGE_WAIT_TIME])
-        )
+    cg.add_optional(
+        var.set_direction_change_waittime, config.get(CONF_DIRECTION_CHANGE_WAIT_TIME)
+    )
     cg.add(var.set_acceleration_wait_time(config[CONF_ACCELERATION_WAIT_TIME]))
     if (overshoot := config[CONF_OVERSHOOT_DURATION]).total_milliseconds:
         cg.add_define("USE_FEEDBACK_COVER_OVERSHOOT")
         cg.add(var.set_overshoot_duration(overshoot))
-    cg.add(var.set_obstacle_rollback(config[CONF_OBSTACLE_ROLLBACK]))

esphome/components/feedback/feedback_cover.cpp

@@ -22,15 +22,15 @@ void FeedbackCover::setup() {
   }
   this->current_operation = COVER_OPERATION_IDLE;
 
-#ifdef USE_BINARY_SENSOR
-  // if available, get position from endstop sensors
+#ifdef USE_FEEDBACK_COVER_ENDSTOP
   if (this->open_endstop_ != nullptr && this->open_endstop_->state) {
     this->position = COVER_OPEN;
   } else if (this->close_endstop_ != nullptr && this->close_endstop_->state) {
     this->position = COVER_CLOSED;
   }
+#endif
 
-  // if available, get moving state from sensors
+#ifdef USE_FEEDBACK_COVER_MOVEMENT
   if (this->open_feedback_ != nullptr && this->open_feedback_->state) {
     this->current_operation = COVER_OPERATION_OPENING;
   } else if (this->close_feedback_ != nullptr && this->close_feedback_->state) {
@@ -51,30 +51,36 @@ CoverTraits FeedbackCover::get_traits() {
 }
 
 void FeedbackCover::dump_config() {
-  LOG_COVER("", "Endstop Cover", this);
-  ESP_LOGCONFIG(TAG, "  Open Duration: %.1fs", this->open_duration_ / 1e3f);
-#ifdef USE_BINARY_SENSOR
+  LOG_COVER("", "Feedback Cover", this);
+  ESP_LOGCONFIG(TAG,
+                "  Open Duration: %.1fs\n"
+                "  Close Duration: %.1fs",
+                this->open_duration_ / 1e3f, this->close_duration_ / 1e3f);
+#ifdef USE_FEEDBACK_COVER_ENDSTOP
   LOG_BINARY_SENSOR("  ", "Open Endstop", this->open_endstop_);
-  LOG_BINARY_SENSOR("  ", "Open Feedback", this->open_feedback_);
-  LOG_BINARY_SENSOR("  ", "Open Obstacle", this->open_obstacle_);
-#endif
-  ESP_LOGCONFIG(TAG, "  Close Duration: %.1fs", this->close_duration_ / 1e3f);
-#ifdef USE_BINARY_SENSOR
   LOG_BINARY_SENSOR("  ", "Close Endstop", this->close_endstop_);
+#endif
+#ifdef USE_FEEDBACK_COVER_MOVEMENT
+  LOG_BINARY_SENSOR("  ", "Open Feedback", this->open_feedback_);
   LOG_BINARY_SENSOR("  ", "Close Feedback", this->close_feedback_);
+#endif
+#ifdef USE_FEEDBACK_COVER_OBSTACLE
+  LOG_BINARY_SENSOR("  ", "Open Obstacle", this->open_obstacle_);
   LOG_BINARY_SENSOR("  ", "Close Obstacle", this->close_obstacle_);
 #endif
   if (this->has_built_in_endstop_) {
     ESP_LOGCONFIG(TAG, "  Has builtin endstop: YES");
   }
+#ifdef USE_FEEDBACK_COVER_MOVEMENT
   if (this->infer_endstop_) {
     ESP_LOGCONFIG(TAG, "  Infer endstop from movement: YES");
   }
+#endif
   if (this->max_duration_ < UINT32_MAX) {
     ESP_LOGCONFIG(TAG, "  Max Duration: %.1fs", this->max_duration_ / 1e3f);
   }
-  if (this->direction_change_waittime_.has_value()) {
-    ESP_LOGCONFIG(TAG, "  Direction change wait time: %.1fs", *this->direction_change_waittime_ / 1e3f);
+  if (this->direction_change_waittime_ != UINT32_MAX) {
+    ESP_LOGCONFIG(TAG, "  Direction change wait time: %.1fs", this->direction_change_waittime_ / 1e3f);
   }
   if (this->acceleration_wait_time_) {
     ESP_LOGCONFIG(TAG, "  Acceleration wait time: %.1fs", this->acceleration_wait_time_ / 1e3f);
@@ -84,60 +90,51 @@ void FeedbackCover::dump_config() {
     ESP_LOGCONFIG(TAG, "  Overshoot duration: %.1fs", this->overshoot_duration_ / 1e3f);
   }
 #endif
-#ifdef USE_BINARY_SENSOR
-  if (this->obstacle_rollback_ && (this->open_obstacle_ != nullptr || this->close_obstacle_ != nullptr)) {
+#ifdef USE_FEEDBACK_COVER_OBSTACLE
+  if (this->obstacle_rollback_) {
     ESP_LOGCONFIG(TAG, "  Obstacle rollback: %.1f%%", this->obstacle_rollback_ * 100);
   }
 #endif
 }
 
-#ifdef USE_BINARY_SENSOR
-
+#ifdef USE_FEEDBACK_COVER_MOVEMENT
 void FeedbackCover::set_open_sensor(binary_sensor::BinarySensor *open_feedback) {
   this->open_feedback_ = open_feedback;
-
-  // setup callbacks to react to sensor changes
-  open_feedback->add_on_state_callback([this](bool state) {
-    ESP_LOGD(TAG, "'%s' - Open feedback '%s'.", LOG_STR_ARG(this->get_log_name()),
-             state ? LOG_STR_LITERAL("STARTED") : LOG_STR_LITERAL("ENDED"));
-    this->recompute_position_();
-    if (!state && this->infer_endstop_ && this->current_trigger_operation_ == COVER_OPERATION_OPENING) {
-      this->endstop_reached_(true);
-    }
-    this->set_current_operation_(state ? COVER_OPERATION_OPENING : COVER_OPERATION_IDLE, false);
-  });
+  open_feedback->add_on_state_callback([this](bool state) { this->on_feedback_(true, state); });
 }
 
 void FeedbackCover::set_close_sensor(binary_sensor::BinarySensor *close_feedback) {
   this->close_feedback_ = close_feedback;
-
-  close_feedback->add_on_state_callback([this](bool state) {
-    ESP_LOGD(TAG, "'%s' - Close feedback '%s'.", LOG_STR_ARG(this->get_log_name()),
-             state ? LOG_STR_LITERAL("STARTED") : LOG_STR_LITERAL("ENDED"));
-    this->recompute_position_();
-    if (!state && this->infer_endstop_ && this->current_trigger_operation_ == COVER_OPERATION_CLOSING) {
-      this->endstop_reached_(false);
-    }
-
-    this->set_current_operation_(state ? COVER_OPERATION_CLOSING : COVER_OPERATION_IDLE, false);
-  });
+  close_feedback->add_on_state_callback([this](bool state) { this->on_feedback_(false, state); });
 }
 
+void FeedbackCover::on_feedback_(bool open, bool state) {
+  const CoverOperation operation = open ? COVER_OPERATION_OPENING : COVER_OPERATION_CLOSING;
+  ESP_LOGD(TAG, "'%s' - %s feedback '%s'.", LOG_STR_ARG(this->get_log_name()),
+           open ? LOG_STR_LITERAL("Open") : LOG_STR_LITERAL("Close"),
+           state ? LOG_STR_LITERAL("STARTED") : LOG_STR_LITERAL("ENDED"));
+  this->recompute_position_();
+  if (!state && this->infer_endstop_ && this->current_trigger_operation_ == operation) {
+    this->endstop_reached_(open);
+  }
+  this->set_current_operation_(state ? operation : COVER_OPERATION_IDLE, false);
+}
+#endif
+
+#ifdef USE_FEEDBACK_COVER_ENDSTOP
 void FeedbackCover::set_open_endstop(binary_sensor::BinarySensor *open_endstop) {
   this->open_endstop_ = open_endstop;
   open_endstop->add_on_state_callback([this](bool state) {
-    if (state) {
+    if (state)
       this->endstop_reached_(true);
-    }
   });
 }
 
 void FeedbackCover::set_close_endstop(binary_sensor::BinarySensor *close_endstop) {
   this->close_endstop_ = close_endstop;
   close_endstop->add_on_state_callback([this](bool state) {
-    if (state) {
+    if (state)
       this->endstop_reached_(false);
-    }
   });
 }
 #endif
@@ -147,22 +144,22 @@ void FeedbackCover::endstop_reached_(bool open_endstop) {
 
   this->position = open_endstop ? COVER_OPEN : COVER_CLOSED;
 
-  // only act if endstop activated while moving in the right direction, in case we are coming back
-  // from a position slightly past the endpoint
+  // Only when moving towards it; the cover may be returning from just past it
   if (this->current_trigger_operation_ == (open_endstop ? COVER_OPERATION_OPENING : COVER_OPERATION_CLOSING)) {
     float dur = (now - this->start_dir_time_) / 1e3f;
     ESP_LOGD(TAG, "'%s' - %s endstop reached. Took %.1fs.", LOG_STR_ARG(this->get_log_name()),
              open_endstop ? LOG_STR_LITERAL("Open") : LOG_STR_LITERAL("Close"), dur);
 
-    // if there is no external mechanism, stop the cover
     if (!this->has_built_in_endstop_) {
       this->start_direction_(COVER_OPERATION_IDLE);
     } else {
       this->set_current_operation_(COVER_OPERATION_IDLE, true);
     }
+    // Without movement sensors the stop above already published
+    if (!this->has_movement_sensors_())
+      return;
   }
 
-  // always sync position and publish
   this->publish_state();
   this->last_publish_time_ = now;
 }
@@ -170,69 +167,54 @@ void FeedbackCover::endstop_reached_(bool open_endstop) {
 void FeedbackCover::set_current_operation_(cover::CoverOperation operation, bool is_triggered) {
   if (is_triggered) {
     this->current_trigger_operation_ = operation;
+    // With movement sensors they set the actual operation
+    if (this->has_movement_sensors_())
+      return;
   }
 
-  // if it is setting the actual operation (not triggered one) or
-  // if we don't have moving sensor, we operate in optimistic mode, assuming actions take place immediately
-  // thus, triggered operation always sets current operation.
-  // otherwise, current operation comes from sensor, and may differ from requested operation
-  // this might be from delays or complex actions, or because the movement was not trigger by the component
-  // but initiated externally
-
-#ifdef USE_BINARY_SENSOR
-  if (!is_triggered || (this->open_feedback_ == nullptr || this->close_feedback_ == nullptr))
-#endif
-  {
-    const uint32_t now = App.get_loop_component_start_time();
-    this->current_operation = operation;
-    this->start_dir_time_ = this->last_recompute_time_ = now;
-    this->publish_state();
-    this->last_publish_time_ = now;
-  }
-}
-
-#ifdef USE_BINARY_SENSOR
-void FeedbackCover::set_close_obstacle_sensor(binary_sensor::BinarySensor *close_obstacle) {
-  this->close_obstacle_ = close_obstacle;
-
-  close_obstacle->add_on_state_callback([this](bool state) {
-    if (state && (this->current_operation == COVER_OPERATION_CLOSING ||
-                  this->current_trigger_operation_ == COVER_OPERATION_CLOSING)) {
-      ESP_LOGD(TAG, "'%s' - Close obstacle detected.", LOG_STR_ARG(this->get_log_name()));
-      this->start_direction_(COVER_OPERATION_IDLE);
-
-      if (this->obstacle_rollback_) {
-        this->target_position_ = clamp(this->position + this->obstacle_rollback_, COVER_CLOSED, COVER_OPEN);
-        this->start_direction_(COVER_OPERATION_OPENING);
-      }
-    }
-  });
+  const uint32_t now = App.get_loop_component_start_time();
+  this->current_operation = operation;
+  if (operation != COVER_OPERATION_IDLE)
+    this->enable_loop();
+  this->start_dir_time_ = this->last_recompute_time_ = now;
+  this->publish_state();
+  this->last_publish_time_ = now;
 }
 
+#ifdef USE_FEEDBACK_COVER_OBSTACLE
 void FeedbackCover::set_open_obstacle_sensor(binary_sensor::BinarySensor *open_obstacle) {
   this->open_obstacle_ = open_obstacle;
+  open_obstacle->add_on_state_callback([this](bool state) { this->on_obstacle_(true, state); });
+}
 
-  open_obstacle->add_on_state_callback([this](bool state) {
-    if (state && (this->current_operation == COVER_OPERATION_OPENING ||
-                  this->current_trigger_operation_ == COVER_OPERATION_OPENING)) {
-      ESP_LOGD(TAG, "'%s' - Open obstacle detected.", LOG_STR_ARG(this->get_log_name()));
-      this->start_direction_(COVER_OPERATION_IDLE);
+void FeedbackCover::set_close_obstacle_sensor(binary_sensor::BinarySensor *close_obstacle) {
+  this->close_obstacle_ = close_obstacle;
+  close_obstacle->add_on_state_callback([this](bool state) { this->on_obstacle_(false, state); });
+}
 
-      if (this->obstacle_rollback_) {
-        this->target_position_ = clamp(this->position - this->obstacle_rollback_, COVER_CLOSED, COVER_OPEN);
-        this->start_direction_(COVER_OPERATION_CLOSING);
-      }
-    }
-  });
+void FeedbackCover::on_obstacle_(bool open, bool state) {
+  const CoverOperation operation = open ? COVER_OPERATION_OPENING : COVER_OPERATION_CLOSING;
+  if (!state || (this->current_operation != operation && this->current_trigger_operation_ != operation))
+    return;
+  ESP_LOGD(TAG, "'%s' - %s obstacle detected.", LOG_STR_ARG(this->get_log_name()),
+           open ? LOG_STR_LITERAL("Open") : LOG_STR_LITERAL("Close"));
+  this->start_direction_(COVER_OPERATION_IDLE);
+
+  if (this->obstacle_rollback_) {
+    const float rollback = open ? -this->obstacle_rollback_ : this->obstacle_rollback_;
+    this->target_position_ = clamp(this->position + rollback, COVER_CLOSED, COVER_OPEN);
+    this->start_direction_(open ? COVER_OPERATION_CLOSING : COVER_OPERATION_OPENING);
+  }
 }
 #endif
 
 void FeedbackCover::loop() {
-  if (this->current_operation == COVER_OPERATION_IDLE)
+  if (this->current_operation == COVER_OPERATION_IDLE) {
+    this->disable_loop();
     return;
+  }
   const uint32_t now = App.get_loop_component_start_time();
 
-  // Recompute position every loop cycle
   this->recompute_position_();
 
   // if we initiated the move, check if we reached position or max time
@@ -264,9 +246,7 @@ void FeedbackCover::loop() {
     }
   }
 
-  // update current position at requested interval, regardless of who started the movement
-  // so that we also update UI if there was an external movement
-  // don't save intermediate positions
+  // Publish intermediate positions without saving them
   if (now - this->last_publish_time_ > this->update_interval_) {
     this->publish_state(false);
     this->last_publish_time_ = now;
@@ -274,11 +254,9 @@ void FeedbackCover::loop() {
 }
 
 void FeedbackCover::control(const CoverCall &call) {
-  // stop action logic
   if (call.get_stop()) {
     this->start_direction_(COVER_OPERATION_IDLE);
   } else if (call.get_toggle().has_value()) {
-    // toggle action logic: OPEN - STOP - CLOSE
     if (this->current_trigger_operation_ != COVER_OPERATION_IDLE) {
       this->start_direction_(COVER_OPERATION_IDLE);
     } else {
@@ -294,28 +272,15 @@ void FeedbackCover::control(const CoverCall &call) {
     auto pos_opt = call.get_position();
     if (!pos_opt.has_value())
       return;
-    // go to position action
     auto pos = *pos_opt;
     if (pos == this->position) {
-      // already at target,
-
-      // for covers with built in end stop, if we don't have sensors we should send the command again
-      // to make sure the assumed state is not wrong
-      if (this->has_built_in_endstop_ && ((pos == COVER_OPEN
-#ifdef USE_BINARY_SENSOR
-                                           && this->open_endstop_ == nullptr
-#endif
-                                           && !this->infer_endstop_) ||
-                                          (pos == COVER_CLOSED
-#ifdef USE_BINARY_SENSOR
-                                           && this->close_endstop_ == nullptr
-#endif
-                                           && !this->infer_endstop_))) {
+      // Resend in case the assumed state is wrong
+      if (this->has_built_in_endstop_ &&
+          ((pos == COVER_OPEN && !this->has_endstop_(true)) || (pos == COVER_CLOSED && !this->has_endstop_(false)))) {
         this->target_position_ = pos;
         this->start_direction_(pos == COVER_CLOSED ? COVER_OPERATION_CLOSING : COVER_OPERATION_OPENING);
       } else if (this->current_operation != COVER_OPERATION_IDLE ||
                  this->current_trigger_operation_ != COVER_OPERATION_IDLE) {
-        // if we are moving, stop
         this->start_direction_(COVER_OPERATION_IDLE);
       }
     } else {
@@ -326,7 +291,7 @@ void FeedbackCover::control(const CoverCall &call) {
 }
 
 void FeedbackCover::stop_prev_trigger_() {
-  if (this->direction_change_waittime_.has_value()) {
+  if (this->direction_change_waittime_ != UINT32_MAX) {
     this->cancel_timeout(DIRECTION_CHANGE_TIMEOUT_ID);
   }
   if (this->prev_command_trigger_ != nullptr) {
@@ -336,9 +301,6 @@ void FeedbackCover::stop_prev_trigger_() {
 }
 
 bool FeedbackCover::is_at_target_() const {
-  // if initiated externally, current operation might be different from
-  // operation that was triggered, thus evaluate position against what was asked
-
   switch (this->current_trigger_operation_) {
     case COVER_OPERATION_OPENING:
       return this->position >= this->target_position_;
@@ -350,10 +312,10 @@ bool FeedbackCover::is_at_target_() const {
       return true;
   }
 }
+
 void FeedbackCover::start_direction_(CoverOperation dir) {
   Trigger<> *trig;
-
-#ifdef USE_BINARY_SENSOR
+#ifdef USE_FEEDBACK_COVER_OBSTACLE
   binary_sensor::BinarySensor *obstacle{nullptr};
 #endif
 
@@ -369,14 +331,14 @@ void FeedbackCover::start_direction_(CoverOperation dir) {
     case COVER_OPERATION_OPENING:
       this->last_operation_ = dir;
       trig = &this->open_trigger_;
-#ifdef USE_BINARY_SENSOR
+#ifdef USE_FEEDBACK_COVER_OBSTACLE
       obstacle = this->open_obstacle_;
 #endif
       break;
     case COVER_OPERATION_CLOSING:
       this->last_operation_ = dir;
       trig = &this->close_trigger_;
-#ifdef USE_BINARY_SENSOR
+#ifdef USE_FEEDBACK_COVER_OBSTACLE
       obstacle = this->close_obstacle_;
 #endif
       break;
@@ -386,9 +348,8 @@ void FeedbackCover::start_direction_(CoverOperation dir) {
 
   this->stop_prev_trigger_();
 
-#ifdef USE_BINARY_SENSOR
-  // check if there is an obstacle to start the new operation -> abort without any change
-  // the case when an obstacle appears while moving is handled in the callback
+#ifdef USE_FEEDBACK_COVER_OBSTACLE
+  // An obstacle already present aborts the start
   if (obstacle != nullptr && obstacle->state) {
     ESP_LOGD(TAG, "'%s' - %s obstacle detected. Action not started.", LOG_STR_ARG(this->get_log_name()),
              dir == COVER_OPERATION_OPENING ? LOG_STR_LITERAL("Open") : LOG_STR_LITERAL("Close"));
@@ -396,14 +357,12 @@ void FeedbackCover::start_direction_(CoverOperation dir) {
   }
 #endif
 
-  // if we are moving and need to move in the opposite direction
-  // check if we have a wait time
-  if (this->direction_change_waittime_.has_value() && dir != COVER_OPERATION_IDLE &&
+  if (this->direction_change_waittime_ != UINT32_MAX && dir != COVER_OPERATION_IDLE &&
       this->current_operation != COVER_OPERATION_IDLE && dir != this->current_operation) {
-    const uint32_t waittime = *this->direction_change_waittime_;
     ESP_LOGD(TAG, "'%s' - Reversing direction.", LOG_STR_ARG(this->get_log_name()));
     this->start_direction_(COVER_OPERATION_IDLE);
-    this->set_timeout(DIRECTION_CHANGE_TIMEOUT_ID, waittime, [this, dir]() { this->start_direction_(dir); });
+    this->set_timeout(DIRECTION_CHANGE_TIMEOUT_ID, this->direction_change_waittime_,
+                      [this, dir]() { this->start_direction_(dir); });
   } else {
     this->set_current_operation_(dir, true);
     this->prev_command_trigger_ = trig;
@@ -425,42 +384,24 @@ void FeedbackCover::recompute_position_() {
   float min_pos;
   float max_pos;
 
-  // endstop sensors update position from their callbacks, and sets the fully open/close value
-  // If we have endstop, estimation never reaches the fully open/closed state.
-  // but if movement continues past corresponding endstop (inertia), keep the fully open/close state
-
+  // With an endstop the estimate stops short of the end; the endstop callback sets it
   switch (this->current_operation) {
     case COVER_OPERATION_OPENING:
       dir = 1.0f;
       action_dur = this->open_duration_;
       min_pos = COVER_CLOSED;
-      max_pos = (
-#ifdef USE_BINARY_SENSOR
-                    this->open_endstop_ != nullptr ||
-#endif
-                    this->infer_endstop_) &&
-                        this->position < COVER_OPEN
-                    ? 0.99f
-                    : COVER_OPEN;
+      max_pos = this->has_endstop_(true) && this->position < COVER_OPEN ? 0.99f : COVER_OPEN;
       break;
     case COVER_OPERATION_CLOSING:
       dir = -1.0f;
       action_dur = this->close_duration_;
-      min_pos = (
-#ifdef USE_BINARY_SENSOR
-                    this->close_endstop_ != nullptr ||
-#endif
-                    this->infer_endstop_) &&
-                        this->position > COVER_CLOSED
-                    ? 0.01f
-                    : COVER_CLOSED;
+      min_pos = this->has_endstop_(false) && this->position > COVER_CLOSED ? 0.01f : COVER_CLOSED;
       max_pos = COVER_OPEN;
       break;
     default:
       return;
   }
 
-  // check if we have an acceleration_wait_time, and remove from position computation
   if (now - this->start_dir_time_ > this->acceleration_wait_time_) {
     uint32_t accel_end_time = this->start_dir_time_ + this->acceleration_wait_time_;
     uint32_t effective_start;

esphome/components/feedback/feedback_cover.h

@@ -20,13 +20,19 @@ class FeedbackCover final : public cover::Cover, public Component {
   Trigger<> *get_close_trigger() { return &this->close_trigger_; }
   Trigger<> *get_stop_trigger() { return &this->stop_trigger_; }
 
-#ifdef USE_BINARY_SENSOR
+#ifdef USE_FEEDBACK_COVER_ENDSTOP
   void set_open_endstop(binary_sensor::BinarySensor *open_endstop);
-  void set_open_sensor(binary_sensor::BinarySensor *open_feedback);
-  void set_open_obstacle_sensor(binary_sensor::BinarySensor *open_obstacle);
   void set_close_endstop(binary_sensor::BinarySensor *close_endstop);
+#endif
+#ifdef USE_FEEDBACK_COVER_MOVEMENT
+  void set_open_sensor(binary_sensor::BinarySensor *open_feedback);
   void set_close_sensor(binary_sensor::BinarySensor *close_feedback);
+  void set_infer_endstop(bool infer_endstop) { this->infer_endstop_ = infer_endstop; }
+#endif
+#ifdef USE_FEEDBACK_COVER_OBSTACLE
+  void set_open_obstacle_sensor(binary_sensor::BinarySensor *open_obstacle);
   void set_close_obstacle_sensor(binary_sensor::BinarySensor *close_obstacle);
+  void set_obstacle_rollback(float obstacle_rollback) { this->obstacle_rollback_ = obstacle_rollback; }
 #endif
   void set_open_duration(uint32_t duration) { this->open_duration_ = duration; }
   void set_close_duration(uint32_t duration) { this->close_duration_ = duration; }
@@ -36,9 +42,7 @@ class FeedbackCover final : public cover::Cover, public Component {
 #ifdef USE_FEEDBACK_COVER_OVERSHOOT
   void set_overshoot_duration(uint32_t overshoot_duration) { this->overshoot_duration_ = overshoot_duration; }
 #endif
-  void set_obstacle_rollback(float obstacle_rollback) { this->obstacle_rollback_ = obstacle_rollback; }
   void set_update_interval(uint32_t interval) { this->update_interval_ = interval; }
-  void set_infer_endstop(bool infer_endstop) { this->infer_endstop_ = infer_endstop; }
   void set_direction_change_waittime(uint32_t waittime) { this->direction_change_waittime_ = waittime; }
   void set_acceleration_wait_time(uint32_t waittime) { this->acceleration_wait_time_ = waittime; }
 
@@ -53,15 +57,43 @@ class FeedbackCover final : public cover::Cover, public Component {
   void endstop_reached_(bool open_endstop);
   void recompute_position_();
   void set_current_operation_(cover::CoverOperation operation, bool is_triggered);
+#ifdef USE_FEEDBACK_COVER_MOVEMENT
+  void on_feedback_(bool open, bool state);
+#endif
+#ifdef USE_FEEDBACK_COVER_OBSTACLE
+  void on_obstacle_(bool open, bool state);
+#endif
 
-#ifdef USE_BINARY_SENSOR
+  bool has_endstop_(bool open) const {
+#ifdef USE_FEEDBACK_COVER_ENDSTOP
+    if ((open ? this->open_endstop_ : this->close_endstop_) != nullptr)
+      return true;
+#endif
+#ifdef USE_FEEDBACK_COVER_MOVEMENT
+    return this->infer_endstop_;
+#else
+    return false;
+#endif
+  }
+  bool has_movement_sensors_() const {
+#ifdef USE_FEEDBACK_COVER_MOVEMENT
+    return this->open_feedback_ != nullptr && this->close_feedback_ != nullptr;
+#else
+    return false;
+#endif
+  }
+
+#ifdef USE_FEEDBACK_COVER_ENDSTOP
   binary_sensor::BinarySensor *open_endstop_{nullptr};
   binary_sensor::BinarySensor *close_endstop_{nullptr};
+#endif
+#ifdef USE_FEEDBACK_COVER_MOVEMENT
   binary_sensor::BinarySensor *open_feedback_{nullptr};
   binary_sensor::BinarySensor *close_feedback_{nullptr};
+#endif
+#ifdef USE_FEEDBACK_COVER_OBSTACLE
   binary_sensor::BinarySensor *open_obstacle_{nullptr};
   binary_sensor::BinarySensor *close_obstacle_{nullptr};
-
 #endif
   Trigger<> open_trigger_;
   Trigger<> close_trigger_;
@@ -73,12 +105,13 @@ class FeedbackCover final : public cover::Cover, public Component {
 #ifdef USE_FEEDBACK_COVER_OVERSHOOT
   uint32_t overshoot_duration_{0};
 #endif
-  optional<uint32_t> direction_change_waittime_{};
+  uint32_t direction_change_waittime_{UINT32_MAX};
   uint32_t acceleration_wait_time_{0};
   bool has_built_in_endstop_{false};
   bool assumed_state_{false};
+#ifdef USE_FEEDBACK_COVER_MOVEMENT
   bool infer_endstop_{false};
-  float obstacle_rollback_{0};
+#endif
 
   cover::CoverOperation last_operation_{cover::COVER_OPERATION_OPENING};
   cover::CoverOperation current_trigger_operation_{cover::COVER_OPERATION_IDLE};
@@ -90,6 +123,9 @@ class FeedbackCover final : public cover::Cover, public Component {
   uint32_t start_overshoot_time_{0};
 #endif
   float target_position_{0};
+#ifdef USE_FEEDBACK_COVER_OBSTACLE
+  float obstacle_rollback_{0};
+#endif
   uint32_t update_interval_{1000};
 };
 

esphome/components/web_server/__init__.py

@@ -614,25 +614,9 @@ async def to_code(config: ConfigType) -> None:
 
 
 def FILTER_SOURCE_FILES() -> list[str]:
-    """Filter out web_server_v1.cpp when version is not 1, the ESP32 event stream elsewhere, and the server on host."""
+    """Filter out web_server_v1.cpp when version is not 1."""
     files_to_filter: list[str] = []
 
-    # The host has no web server; it builds only sse_chunk.cpp, for its unit tests
-    if CORE.is_host:
-        files_to_filter.extend(
-            (
-                "event_source.cpp",
-                "list_entities.cpp",
-                "web_server.cpp",
-                "web_server_v1.cpp",
-            )
-        )
-        return files_to_filter
-
-    # The /events implementation is ESP32 only
-    if not CORE.is_esp32:
-        files_to_filter.extend(("event_source.cpp", "sse_chunk.cpp"))
-
     # web_server_v1.cpp is only needed when version is 1
     config = CORE.config.get(DOMAIN, {})
     if config.get(CONF_VERSION, 2) != 1:

esphome/components/web_server/event_source.cpp

@@ -1,491 +0,0 @@
-#include "event_source.h"
-#if defined(USE_ESP32) && defined(USE_WEBSERVER)
-
-#include <algorithm>
-#include <cstring>
-#include <iterator>
-
-#include "esphome/core/application.h"
-#include "esphome/core/helpers.h"
-#include "esphome/core/log.h"
-#include "web_server.h"
-
-namespace esphome::web_server {
-
-ESPHOME_LOG_TAG(TAG, "web_server_idf");
-
-// Only prepare_json_() may hold the JSON arena: every other frame in this file is capped below one
-// arena. Measured at -Os on GCC 14; newer toolchains stay checked on purpose, older ones skip it.
-#if defined(__GNUC__) && !defined(__clang__) && __GNUC__ >= 14 && defined(__OPTIMIZE_SIZE__)
-#pragma GCC diagnostic error "-Wstack-usage=2048"
-#endif
-
-AsyncEventSource::~AsyncEventSource() {
-  LockGuard guard{this->pending_mutex_};
-  for (auto *vec : {&this->sessions_, &this->pending_sessions_}) {
-    for (auto *ses : *vec) {
-      delete ses;  // NOLINT(cppcoreguidelines-owning-memory)
-    }
-  }
-}
-
-void AsyncEventSource::handleRequest(AsyncWebServerRequest *request) {
-  // Httpd task: set up the live httpd_req_t and park the session; main loop does the rest.
-  // NOLINTNEXTLINE(cppcoreguidelines-owning-memory,clang-analyzer-cplusplus.NewDeleteLeaks)
-  auto *rsp = new AsyncEventSourceResponse(request, this, this->web_server_);
-  {
-    LockGuard guard{this->pending_mutex_};
-    this->pending_sessions_.push_back(rsp);
-    this->has_pending_sessions_.store(true, std::memory_order_release);
-  }
-  this->web_server_->enable_loop_soon_any_context();
-}
-
-// clang-analyzer traces a false-positive leak path from loop() through
-// adopt_pending_sessions_main_loop_() into start_session_main_loop_() and
-// finally ArduinoJson. Suppress along the entire in-our-code call chain.
-// NOLINTBEGIN(clang-analyzer-cplusplus.NewDeleteLeaks)
-bool AsyncEventSource::loop() {
-  // Fast path: one atomic load per tick. Slow path is out-of-line on connect.
-  if (this->has_pending_sessions_.load(std::memory_order_acquire)) {
-    this->adopt_pending_sessions_main_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
-  for (size_t i = 0; i < this->sessions_.size();) {
-    auto *ses = this->sessions_[i];
-    // If the session has a dead socket (marked by destroy callback)
-    if (ses->safe_to_delete()) {
-      // destroy() already logged the close with the fd; don't double-log here.
-      delete ses;  // NOLINT(cppcoreguidelines-owning-memory)
-      // Remove by swapping with last element (O(1) removal, order doesn't matter for sessions)
-      this->sessions_[i] = this->sessions_.back();
-      this->sessions_.pop_back();
-    } else {
-      ses->loop();
-      ++i;
-    }
-  }
-  // A session still waiting for httpd to commit its context keeps the loop alive too
-  return !this->sessions_.empty() || this->has_pending_sessions_.load(std::memory_order_acquire);
-}
-
-void AsyncEventSource::adopt_pending_sessions_main_loop_() {
-  std::vector<AsyncEventSourceResponse *> incoming;
-  {
-    LockGuard guard{this->pending_mutex_};
-    incoming.swap(this->pending_sessions_);
-    this->has_pending_sessions_.store(false, std::memory_order_relaxed);
-  }
-  for (auto *rsp : incoming) {
-    // Already disconnected? Drop it; skip session start on a dead session.
-    if (rsp->safe_to_delete()) {
-      delete rsp;  // NOLINT(cppcoreguidelines-owning-memory)
-      continue;
-    }
-    // httpd commits the session context only after the creating handler returns, so stay
-    // pending until then; httpd_req_cleanup() always commits it or calls destroy()
-    if (!rsp->is_attached()) {
-      LockGuard guard{this->pending_mutex_};
-      this->pending_sessions_.push_back(rsp);
-      this->has_pending_sessions_.store(true, std::memory_order_release);
-      continue;
-    }
-    this->sessions_.push_back(rsp);
-    rsp->start_session_main_loop_();
-  }
-}
-// NOLINTEND(clang-analyzer-cplusplus.NewDeleteLeaks)
-
-void AsyncEventSource::try_send_nodefer(const char *message, size_t message_len, const char *event, uint32_t id,
-                                        uint32_t reconnect) {
-  for (auto *ses : this->sessions_) {
-    if (ses->is_connected()) {
-      ses->try_send_nodefer(message, message_len, event, id, reconnect);
-    }
-  }
-}
-
-void AsyncEventSource::deferrable_send_state(void *source, bool detail_all, message_generator_t *message_generator) {
-  // Skip if no connected clients to avoid unnecessary processing
-  if (this->empty())
-    return;
-  for (auto *ses : this->sessions_) {
-    if (ses->is_connected()) {
-      ses->deferrable_send_state(source, detail_all, message_generator);
-    }
-  }
-}
-
-AsyncEventSourceResponse::AsyncEventSourceResponse(AsyncWebServerRequest *request, AsyncEventSource *server,
-                                                   WebServer *ws)
-    : server_(server), web_server_(ws), entities_iterator_(ws, this) {
-  // Httpd task only. start_session_main_loop_() sends the greeting and starts the iterator.
-  this->begin(request, SSE_STATUS, SSE_CONTENT_TYPE, SSE_HEADERS, SSE_FIRST_CHUNK);
-}
-static_assert(web_server_base::stream_head_iov_count(std::size(SSE_HEADERS) + WEB_SERVER_DEFAULT_HEADERS_COUNT) <=
-                  web_server_base::STREAM_HEAD_MAX_IOV,
-              "the event stream head must go out in one write");
-
-// NOLINTBEGIN(clang-analyzer-cplusplus.NewDeleteLeaks) false positive with ArduinoJson
-void AsyncEventSourceResponse::start_session_main_loop_() {
-  auto *ws = this->web_server_;
-
-  // The tcp send buffer is empty on connect. A refusal is a closing session or a failed tail
-  // allocation; nothing retries the greeting, so close and let the client reconnect.
-  auto message = ws->get_config_json();
-  if (!this->try_send_nodefer(message.c_str(), message.size(), "ping", millis(), 30000)) {
-    ESP_LOGW(TAG, "Config not sent to fd %d", this->log_id());
-    this->close();
-    return;
-  }
-
-#ifdef USE_WEBSERVER_SORTING
-  for (auto &group : ws->sorting_groups_) {
-    json::JsonBuilder builder;
-    JsonObject root = builder.root();
-    root["name"] = json::linked(group.second.name);
-    root["sorting_weight"] = group.second.weight;
-    message = builder.serialize();
-
-    // a (very) large number of these should be able to be queued initially without defer
-    // since the only thing in the send buffer at this point is the initial ping/config.
-    // A refusal means the socket is full or closing; the remaining groups are not sent. The
-    // session stays up on purpose: partial grouping beats a reconnect loop on a slow link.
-    if (!this->try_send_nodefer(message.c_str(), message.size(), "sorting_group")) {
-      ESP_LOGW(TAG, "Sorting groups not sent to fd %d", this->log_id());
-      break;
-    }
-  }
-#endif
-
-  this->entities_iterator_.begin(ws->include_internal_);
-}
-// NOLINTEND(clang-analyzer-cplusplus.NewDeleteLeaks)
-
-// helper for allowing only unique entries in the queue
-void AsyncEventSourceResponse::deq_push_back_with_dedup_(void *source, message_generator_t *message_generator) {
-  DeferredEvent item(source, message_generator);
-
-  // Use range-based for loop instead of std::find_if to reduce template instantiation overhead and binary size
-  for (auto &event : this->deferred_queue_) {
-    if (event == item) {
-      return;  // Already in queue, no need to update since items are equal
-    }
-  }
-  this->deferred_queue_.push_back(item);
-}
-
-void AsyncEventSourceResponse::process_deferred_queue_() {
-  if (this->is_closing()) {
-    return;
-  }
-  while (!deferred_queue_.empty()) {
-    DeferredEvent &de = deferred_queue_.front();
-    if (this->send_json_(de.source_, de.message_generator_)) {
-      if (this->is_closing() || deferred_queue_.empty()) {
-        return;
-      }
-      // O(n) but memory efficiency is more important than speed here which is why std::vector was chosen
-      deferred_queue_.erase(deferred_queue_.begin());
-    } else {
-      break;
-    }
-  }
-}
-
-void AsyncEventSourceResponse::release_buffers_() {
-  std::vector<DeferredEvent>().swap(this->deferred_queue_);
-  this->tail_.reset();
-  this->tail_cap_ = 0;
-  this->tail_len_ = 0;
-  this->tail_sent_ = 0;
-}
-
-void AsyncEventSourceResponse::drain_tail_() {
-  if (this->sending_ || this->is_closing() || this->tail_len_ == 0) {
-    return;
-  }
-  SendGuard guard{*this};
-
-  const size_t remaining = this->tail_len_ - this->tail_sent_;
-  const ssize_t sent = this->write(this->tail_.get() + this->tail_sent_, remaining);
-  if (sent <= 0) {
-    return;  // write() owns the stall clock and the close
-  }
-  this->no_memory_logged_ = false;
-  this->tail_sent_ += sent;
-  if (this->tail_sent_ < this->tail_len_) {
-    ESP_LOGV(TAG, "Partial send: %zd/%zu bytes (total: %u/%u)", sent, remaining, this->tail_sent_, this->tail_len_);
-    return;
-  }
-  // Fully sent; the storage stays for the next stall
-  this->tail_len_ = 0;
-}
-
-bool AsyncEventSourceResponse::reserve_tail_(size_t len) {
-  if (this->tail_cap_ >= len) {
-    return true;
-  }
-  if (len > TAIL_MAX_SIZE) {
-    return false;
-  }
-  // Nothing is pending while the tail grows, so free the old block first. PREFER_INTERNAL keeps
-  // the tail where plain new put it.
-  this->tail_.reset();
-  this->tail_ = RAMAllocator<uint8_t>(RAMAllocator<uint8_t>::PREFER_INTERNAL).make_unique_array_for_overwrite(len);
-  this->tail_cap_ = this->tail_ ? len : 0;
-  return this->tail_cap_ != 0;
-}
-
-bool AsyncEventSourceResponse::stash_chunk_(const char *prefix, size_t prefix_len, const char *message,
-                                            size_t message_len, size_t total, size_t sent) {
-  // A log event of nothing but line breaks is the largest chunk; reserve_tail_() must never
-  // refuse it for size, only for memory
-  static_assert(SSE_SEP_LEN * LOG_EVENT_MAX_LEN + PREFIX_BUF_SIZE + SSE_SUFFIX_LEN <= TAIL_MAX_SIZE,
-                "the log cut in web_server.h must keep a worst case log event inside the tail ceiling");
-  if (!this->reserve_tail_(total)) {
-    if (sent != 0) {
-      // Part of the chunk is on the wire, so the stream is broken and the client has to go
-      ESP_LOGW(TAG, "Cannot buffer a %zu byte chunk, closing", total);
-      this->close();
-    } else {
-      this->tail_alloc_failed_(total);  // nothing on the wire, the caller retries on the stall clock
-    }
-    return false;
-  }
-  uint8_t *dst = this->tail_.get();
-  std::memcpy(dst, prefix, prefix_len);
-  dst += prefix_len;
-  for_each_chunk_piece(
-      message, message_len,
-      [](void *ctx, const char *piece, size_t len) {
-        auto &out = *static_cast<uint8_t **>(ctx);
-        std::memcpy(out, piece, len);
-        out += len;
-      },
-      &dst);
-  this->tail_len_ = total;
-  this->tail_sent_ = sent;
-  return true;
-}
-
-void AsyncEventSourceResponse::loop() {
-  AsyncStreamResponse::loop();
-  if (this->is_closing()) {
-    if (this->tail_cap_ != 0 || this->deferred_queue_.capacity() != 0) {
-      this->release_buffers_();
-    }
-    return;
-  }
-  drain_tail_();
-  process_deferred_queue_();
-  if (this->is_closing())
-    return;
-  // One step per loop; refusals retry next pass
-  this->entities_iterator_.try_advance(1);
-}
-
-#if defined(__GNUC__) && !defined(__clang__) && __GNUC__ >= 14 && defined(__OPTIMIZE_SIZE__)
-#pragma GCC diagnostic push
-#pragma GCC diagnostic ignored "-Wstack-usage="  // the one frame that holds the JSON arena
-#endif
-AsyncEventSourceResponse::JsonPrepareResult AsyncEventSourceResponse::prepare_json_(void *source,
-                                                                                    message_generator_t *generator,
-                                                                                    char *buffer, size_t *size_out) {
-  // Log lines emitted while the generator runs are dropped for this session: on the large path they would fill
-  // tail_ before it is overwritten, and on the small path they would go out ahead of the event being built.
-  SendGuard guard{*this};
-  json::JsonArena<json::JSON_ARENA_SIZE> arena;
-  json::JsonBuilder builder(&arena);
-  generator(this->web_server_, source, builder);
-  *size_out = builder.serialize_to(buffer, JSON_BUF_SIZE);
-  if (*size_out < JSON_BUF_SIZE) {
-    return JsonPrepareResult::JSON_PREPARE_RESULT_BUFFER;
-  }
-
-  char prefix[PREFIX_BUF_SIZE];
-  const size_t prefix_len = build_chunk_prefix(prefix, sizeof(prefix), "state", 0, 0, true);
-
-  // Nothing reaches the wire until the complete document and its framing are owned by the caller.
-  size_t json_len = 0;
-  size_t cap = std::max<size_t>(JSON_BUF_SIZE * 2, this->tail_cap_);
-  for (;;) {
-    if (!this->reserve_tail_(cap)) {
-      *size_out = cap;
-      return JsonPrepareResult::JSON_PREPARE_RESULT_NO_MEMORY;
-    }
-    const size_t room = cap - prefix_len - SSE_SUFFIX_LEN;
-    json_len = builder.serialize_to(reinterpret_cast<char *>(this->tail_.get()) + prefix_len, room);
-    if (json_len < room) {
-      break;
-    }
-    if (cap >= TAIL_MAX_SIZE) {
-      *size_out = room;
-      return JsonPrepareResult::JSON_PREPARE_RESULT_TOO_LARGE;
-    }
-    cap = std::min<size_t>(cap * 2, TAIL_MAX_SIZE);
-  }
-
-  const size_t total = prefix_len + json_len + SSE_SUFFIX_LEN;
-  write_chunk_header(prefix, total - CHUNK_HDR_LEN - CHUNK_END_LEN);
-  uint8_t *dst = this->tail_.get();
-  std::memcpy(dst, prefix, prefix_len);
-  std::memcpy(dst + prefix_len + json_len, SSE_SUFFIX, SSE_SUFFIX_LEN);
-  this->tail_len_ = total;
-  this->tail_sent_ = 0;
-  return JsonPrepareResult::JSON_PREPARE_RESULT_TAIL;
-}
-#if defined(__GNUC__) && !defined(__clang__) && __GNUC__ >= 14 && defined(__OPTIMIZE_SIZE__)
-#pragma GCC diagnostic pop
-#endif
-
-bool AsyncEventSourceResponse::send_json_(void *source, message_generator_t *generator) {
-  if (!this->ready_to_send_()) {
-    return false;
-  }
-  char buffer[JSON_BUF_SIZE];
-  size_t size;
-  const auto result = this->prepare_json_(source, generator, buffer, &size);
-  switch (result) {
-    case JsonPrepareResult::JSON_PREPARE_RESULT_BUFFER:
-      return this->try_send_nodefer(buffer, size, "state");
-    case JsonPrepareResult::JSON_PREPARE_RESULT_TAIL:
-      this->drain_tail_();
-      return true;
-    case JsonPrepareResult::JSON_PREPARE_RESULT_NO_MEMORY: {
-      SendGuard guard{*this};
-      this->tail_alloc_failed_(size);
-      return false;
-    }
-    case JsonPrepareResult::JSON_PREPARE_RESULT_TOO_LARGE: {
-      SendGuard guard{*this};
-      ESP_LOGW(TAG, "State event does not fit %zu bytes, dropped", size);
-      this->tail_.reset();
-      this->tail_cap_ = 0;
-      this->stall_.clear();
-      this->no_memory_logged_ = false;
-      return true;
-    }
-  }
-  return false;
-}
-
-void AsyncEventSourceResponse::tail_alloc_failed_(size_t cap) {
-  // Same stall clock as a socket that stops draining, so a session cannot retry forever. A refused write may
-  // already have started it, so the warning keeps its own flag.
-  if (!this->no_memory_logged_) {
-    this->no_memory_logged_ = true;
-    ESP_LOGW(TAG, "No memory for a %zu byte chunk", cap);
-  }
-  const uint32_t now = App.get_loop_component_start_time();
-  if (!this->stall_.is_running()) {
-    this->stall_.start(now);
-    return;
-  }
-  if (this->stall_.is_expired(now, SEND_STALL_TIMEOUT_MS)) {
-    ESP_LOGW(TAG, "Closing EventSource after %" PRIu32 " ms without memory", this->stall_.elapsed(now));
-    this->close();
-  }
-}
-
-bool AsyncEventSourceResponse::ready_to_send_() {
-  if (this->sending_ || !this->is_connected() || this->is_closing()) {
-    return false;
-  }
-  drain_tail_();
-  return !this->is_closing() && this->tail_len_ == 0;
-}
-
-bool AsyncEventSourceResponse::try_send_nodefer(const char *message, size_t message_len, const char *event, uint32_t id,
-                                                uint32_t reconnect) {
-  if (!this->ready_to_send_()) {
-    return false;
-  }
-  SendGuard guard{*this};
-
-  // Everything after the prefix goes out straight from the caller's buffer
-  char prefix[PREFIX_BUF_SIZE];
-  const size_t prefix_len = build_chunk_prefix(prefix, sizeof(prefix), event, id, reconnect, message != nullptr);
-  if (message == nullptr && prefix_len == CHUNK_HDR_LEN) {
-    return true;  // Match ESPAsyncWebServer: nothing to send
-  }
-  if (prefix_len >= PREFIX_BUF_SIZE - 1) {
-    // The appenders truncate silently, which would put a malformed event on the wire
-    ESP_LOGW(TAG, "Event name too long, dropped");
-    return true;
-  }
-
-  // Gather list: the prefix, then the data lines and their separators from the message
-  struct Gather {
-    web_server_base::StreamIovec iov[MAX_SEND_IOV];  // left uninitialized on purpose
-    int iovcnt{1};
-    size_t total{0};
-    bool fits{true};
-  } g;
-  g.total = prefix_len;
-  for_each_chunk_piece(
-      message, message_len,
-      [](void *ctx, const char *piece, size_t len) {
-        auto &g = *static_cast<Gather *>(ctx);
-        g.total += len;
-        if (len == 0) {
-          return;
-        }
-        if (g.iovcnt == MAX_SEND_IOV) {
-          g.fits = false;
-          return;
-        }
-        g.iov[g.iovcnt++] = {const_cast<char *>(piece), len};
-      },
-      &g);
-  // The header and the terminator are not part of the chunk length
-  write_chunk_header(prefix, g.total - CHUNK_HDR_LEN - CHUNK_END_LEN);
-  g.iov[0] = {prefix, prefix_len};
-
-  // A message with more lines than the list holds skips straight to the tail
-  const ssize_t sent = g.fits ? this->write(g.iov, g.iovcnt) : 0;
-  if (sent < 0) {
-    return false;
-  }
-  if (sent > 0) {
-    this->no_memory_logged_ = false;
-  }
-  if (static_cast<size_t>(sent) == g.total) {
-    return true;
-  }
-  // The caller's buffers do not outlive this call, so keep the chunk and continue from loop()
-  return this->stash_chunk_(prefix, prefix_len, message, message_len, g.total, sent);
-}
-
-void AsyncEventSourceResponse::deferrable_send_state(void *source, bool detail_all,
-                                                     message_generator_t *message_generator) {
-  // Let the detail_all listing finish before bare state events, so the web UI never shows unnamed entries and
-  // initial connect sends less
-  if (!detail_all && !this->entities_iterator_.completed())
-    return;
-
-  drain_tail_();
-  process_deferred_queue_();
-
-  if (this->is_closing()) {
-    return;
-  }
-
-  if (this->tail_len_ != 0 || !deferred_queue_.empty()) {
-    // outgoing event buffer or deferred queue still not empty which means downstream tcp send buffer full, no point
-    // trying to send first
-    deq_push_back_with_dedup_(source, message_generator);
-  } else {
-    // A send error closes the session and clears the queue; nothing is queued after that
-    if (!this->send_json_(source, message_generator) && !this->is_closing()) {
-      deq_push_back_with_dedup_(source, message_generator);
-    }
-  }
-}
-
-}  // namespace esphome::web_server
-
-#endif  // USE_ESP32 && USE_WEBSERVER

esphome/components/web_server/event_source.h

@@ -1,176 +0,0 @@
-#pragma once
-
-#include "esphome/core/defines.h"
-#if defined(USE_ESP32) && defined(USE_WEBSERVER)
-
-#include "esphome/components/json/json_util.h"
-#include "esphome/components/web_server_base/stream_response.h"
-#include "esphome/core/helpers.h"
-#include "esphome/core/string_ref.h"
-#include "list_entities.h"
-#include "sse_chunk.h"
-
-#include <atomic>
-#include <cstdint>
-#include <vector>
-
-namespace esphome::web_server {
-
-class WebServer;
-class AsyncEventSource;
-class AsyncEventSourceResponse;
-
-using message_generator_t = void(WebServer *, void *, json::JsonBuilder &);
-
-/*
-  This class holds a pointer to the source component that wants to publish a state event, and a pointer to a function
-  that will lazily generate that event.  The two pointers allow dedup in the deferred queue if multiple publishes for
-  the same component are backed up, and take up only two pointers of memory.  The entry in the deferred queue (a
-  std::vector) is the DeferredEvent instance itself (not a pointer to one elsewhere in heap) so still only two pointers
-  per entry (and no heap fragmentation).  Even 100 backed up events (you'd have to have at least 100 sensors publishing
-  because of dedup) would take up only 0.8 kB.
-*/
-struct DeferredEvent {
-  friend class AsyncEventSourceResponse;
-
- protected:
-  void *source_;
-  message_generator_t *message_generator_;
-
- public:
-  DeferredEvent(void *source, message_generator_t *message_generator)
-      : source_(source), message_generator_(message_generator) {}
-  bool operator==(const DeferredEvent &test) const {
-    return (source_ == test.source_ && message_generator_ == test.message_generator_);
-  }
-};
-static_assert(sizeof(DeferredEvent) == sizeof(void *) + sizeof(message_generator_t *),
-              "DeferredEvent should have no padding");
-
-class AsyncEventSourceResponse : public web_server_base::AsyncStreamResponse {
-  friend class AsyncEventSource;
-
- public:
-  bool try_send_nodefer(const char *message, size_t message_len, const char *event = nullptr, uint32_t id = 0,
-                        uint32_t reconnect = 0);
-  // detail_all events go out while entities are still being listed; state events wait until listing is done
-  void deferrable_send_state(void *source, bool detail_all, message_generator_t *message_generator);
-  // The old event name argument would otherwise convert to true
-  void deferrable_send_state(void *source, const char *event_type, message_generator_t *message_generator) = delete;
-  void loop();
-
- protected:
-  AsyncEventSourceResponse(AsyncWebServerRequest *request, AsyncEventSource *server, WebServer *ws);
-
-  // Main-loop only: sends initial ping/config/sorting_groups, starts entity iterator.
-  void start_session_main_loop_();
-
-  void deq_push_back_with_dedup_(void *source, message_generator_t *message_generator);
-  void process_deferred_queue_();
-  // A new chunk may go out: not re-entered from a log line, session alive, tail empty
-  bool ready_to_send_();
-  // Push what is left of the chunk in tail_ to the socket
-  void drain_tail_();
-  // Grow tail_ to hold len bytes, kept at its high-water mark. False on OOM.
-  bool reserve_tail_(size_t len);
-  // Keep the whole chunk in tail_ and continue from sent; false when the tail cannot be allocated.
-  bool stash_chunk_(const char *prefix, size_t prefix_len, const char *message, size_t message_len, size_t total,
-                    size_t sent);
-  // Send a state event; JSON too large for the stack buffer is serialized into tail_ instead
-  bool send_json_(void *source, message_generator_t *generator);
-  enum class JsonPrepareResult : uint8_t {
-    JSON_PREPARE_RESULT_BUFFER,
-    JSON_PREPARE_RESULT_TAIL,
-    JSON_PREPARE_RESULT_NO_MEMORY,
-    JSON_PREPARE_RESULT_TOO_LARGE,
-  };
-  // Keep the arena out of the caller's frame so it is gone before the state send.
-  // size_out: BUFFER = JSON length, NO_MEMORY = requested tail capacity, TOO_LARGE = available JSON room;
-  // unused for TAIL.
-  JsonPrepareResult __attribute__((noinline)) prepare_json_(void *source, message_generator_t *generator, char *buffer,
-                                                            size_t *size_out);
-  // Warn once per stall, and close the session once the stall timeout passes with no memory for the tail
-  void tail_alloc_failed_(size_t cap);
-
-  // A log line emitted inside a send re-enters try_send_nodefer on this session; refuse it
-  struct SendGuard {
-    AsyncEventSourceResponse &owner;
-    explicit SendGuard(AsyncEventSourceResponse &owner) : owner(owner) { owner.sending_ = true; }
-    ~SendGuard() { this->owner.sending_ = false; }
-  };
-  // Frees the tail and the queue's storage once the session is closing
-  void release_buffers_();
-
-  AsyncEventSource *server_;
-  std::vector<DeferredEvent> deferred_queue_;
-  WebServer *web_server_;
-  ListEntitiesIterator entities_iterator_;
-  // One chunk the socket did not take whole, allocated on the first stall; the only heap use
-  // on the send path
-  RAMUniquePtr<uint8_t[]> tail_;
-  uint16_t tail_cap_{0};
-  uint16_t tail_len_{0};  // Zero means nothing pending
-  uint16_t tail_sent_{0};
-  bool sending_{false};
-  // The no-memory warning went out during the current stall; cleared on send progress
-  bool no_memory_logged_{false};
-  // The longest multi line log message in the tree (a climate dump_config) has 22 lines; a
-  // longer one goes through the tail
-  static constexpr size_t MAX_SEND_LINES = 22;
-  static constexpr size_t MAX_SEND_IOV = 1 + 2 * MAX_SEND_LINES;
-  // Chunk header, retry/id/event lines and the first "data: "
-  static constexpr size_t PREFIX_BUF_SIZE = 128;
-
-  // Stack buffer for a state event's JSON; a larger document is serialized into the tail
-  static constexpr size_t JSON_BUF_SIZE = 1024;
-  // Same ceiling JsonBuilder::serialize() applies (max_heap_size in json_util.cpp); a larger
-  // document is dropped before anything is on the wire
-  static constexpr size_t JSON_MAX_SIZE = 5120;
-  // Most RAM a stalled session keeps: the largest state document plus any accepted framing
-  static constexpr size_t TAIL_MAX_SIZE = JSON_MAX_SIZE + PREFIX_BUF_SIZE + SSE_SUFFIX_LEN;
-};
-
-class AsyncEventSource : public AsyncWebHandler {
-  friend class AsyncEventSourceResponse;
-
- public:
-  AsyncEventSource(StringRef url, WebServer *ws) : url_(url), web_server_(ws) {}
-  ~AsyncEventSource() override;
-
-  // NOLINTNEXTLINE(readability-identifier-naming)
-  bool canHandle(AsyncWebServerRequest *request) const override {
-    if (request->method() != HTTP_GET)
-      return false;
-    char url_buf[AsyncWebServerRequest::URL_BUF_SIZE];
-    return request->url_to(url_buf) == this->url_;
-  }
-  // NOLINTNEXTLINE(readability-identifier-naming)
-  void handleRequest(AsyncWebServerRequest *request) override;
-  void try_send_nodefer(const char *message, size_t message_len, const char *event = nullptr, uint32_t id = 0,
-                        uint32_t reconnect = 0);
-  void deferrable_send_state(void *source, bool detail_all, message_generator_t *message_generator);
-  // The old event name argument would otherwise convert to true
-  void deferrable_send_state(void *source, const char *event_type, message_generator_t *message_generator) = delete;
-  /// Returns true if there are sessions remaining (including pending cleanup).
-  bool loop();
-  bool empty() { return this->count() == 0; }
-
-  size_t count() const { return this->sessions_.size(); }
-
- protected:
-  // Cold path: move sessions from pending_sessions_ into sessions_ and greet each one.
-  void __attribute__((noinline, cold)) adopt_pending_sessions_main_loop_();
-
-  StringRef url_;  // Must outlive this object (string literal)
-  // Main-loop only. Vector: SSE sessions are 1-5 connections, linear search beats set.
-  std::vector<AsyncEventSourceResponse *> sessions_;
-  // Httpd-task intake; guarded by pending_mutex_, gated by has_pending_sessions_.
-  std::vector<AsyncEventSourceResponse *> pending_sessions_;
-  Mutex pending_mutex_;
-  WebServer *web_server_;
-  std::atomic<bool> has_pending_sessions_{false};
-};
-
-}  // namespace esphome::web_server
-
-#endif  // USE_ESP32 && USE_WEBSERVER

esphome/components/web_server/list_entities.cpp

@@ -8,8 +8,14 @@
 
 namespace esphome::web_server {
 
+#ifdef USE_ESP32
+ListEntitiesIterator::ListEntitiesIterator(const WebServer *ws, AsyncEventSource *es) : web_server_(ws), events_(es) {}
+#elif USE_ARDUINO
+ListEntitiesIterator::ListEntitiesIterator(const WebServer *ws, DeferredUpdateEventSource *es)
+    : web_server_(ws), events_(es) {}
+#endif
 void ListEntitiesIterator::send_all_(void *source, message_generator_t *generator) {
-  this->session_->deferrable_send_state(source, true, generator);
+  this->events_->deferrable_send_state(source, true, generator);
 }
 
 #ifdef USE_BINARY_SENSOR

esphome/components/web_server/list_entities.h

@@ -1,28 +1,33 @@
 #pragma once
 
 #include "esphome/core/defines.h"
-#if defined(USE_WEBSERVER) && (defined(USE_ESP32) || defined(USE_ARDUINO))
+#ifdef USE_WEBSERVER
 #include "esphome/core/component.h"
 #include "esphome/core/component_iterator.h"
 namespace esphome::json {
 class JsonBuilder;
 }  // namespace esphome::json
 
+namespace esphome::web_server_idf {
+#ifdef USE_ESP32
+class AsyncEventSource;
+#endif
+}  // namespace esphome::web_server_idf
+
 namespace esphome::web_server {
 
-// The /events session the iterator lists entities to
-#ifdef USE_ESP32
-class AsyncEventSourceResponse;
-using EventSession = AsyncEventSourceResponse;
-#elif defined(USE_ARDUINO)
+#if !defined(USE_ESP32) && defined(USE_ARDUINO)
 class DeferredUpdateEventSource;
-using EventSession = DeferredUpdateEventSource;
 #endif
 class WebServer;
 
 class ListEntitiesIterator final : public ComponentIterator {
  public:
-  ListEntitiesIterator(const WebServer *ws, EventSession *session) : web_server_(ws), session_(session) {}
+#ifdef USE_ESP32
+  ListEntitiesIterator(const WebServer *ws, esphome::web_server_idf::AsyncEventSource *es);
+#elif defined(USE_ARDUINO)
+  ListEntitiesIterator(const WebServer *ws, DeferredUpdateEventSource *es);
+#endif
 
 // Entity overrides (generated from entity_types.h).
 // Implementations live in list_entities.cpp.
@@ -40,7 +45,11 @@ class ListEntitiesIterator final : public ComponentIterator {
   void send_all_(void *source, void (*generator)(WebServer *, void *, json::JsonBuilder &));
 
   const WebServer *web_server_;
-  EventSession *session_;
+#ifdef USE_ESP32
+  esphome::web_server_idf::AsyncEventSource *events_;
+#elif USE_ARDUINO
+  DeferredUpdateEventSource *events_;
+#endif
 };
 
 }  // namespace esphome::web_server

esphome/components/web_server/web_server.cpp

@@ -245,7 +245,7 @@ void DeferredUpdateEventSource::process_deferred_queue_() {
       deferred_queue_.erase(deferred_queue_.begin());
       this->consecutive_send_failures_ = 0;  // Reset failure count on successful send
     } else {
-      // NOTE: Similar logic exists in web_server_idf/web_server_idf.cpp in AsyncStreamResponse::write().
+      // NOTE: Similar logic exists in web_server_idf/web_server_idf.cpp in AsyncEventSourceResponse::drain_tail_().
       // The close mechanisms are platform-specific (this path calls close() directly; the IDF path is time-based and
       // closes through HTTPD to preserve session ownership), but both drop a client after roughly 20 seconds without
       // send progress. Keep that stall policy in sync when changing either side.

esphome/components/web_server/web_server.h

@@ -5,11 +5,7 @@
 #include "esphome/components/json/json_util.h"
 #include "esphome/components/web_server_base/web_server_base.h"
 #include "esphome/components/web_server_base/captive_dns.h"
-// The host only builds sse_chunk.cpp for unit tests, but other tests can define USE_WEBSERVER there
-#if defined(USE_WEBSERVER) && (defined(USE_ESP32) || defined(USE_ARDUINO))
-#ifdef USE_ESP32
-#include "event_source.h"
-#endif
+#ifdef USE_WEBSERVER
 #include "esphome/core/component.h"
 #include "esphome/core/entity_base.h"
 #include "esphome/core/entity_includes.h"

esphome/components/web_server_base/__init__.py

@@ -105,9 +105,9 @@ async def to_code(config: ConfigType) -> None:
     var = cg.new_Pvariable(config[CONF_ID])
     cg.add(cg.RawExpression(f"{web_server_base_ns}::global_web_server_base = {var}"))
 
-    # Default headers added in init(): web_server_idf's StaticVector, or the library's DefaultHeaders
-    cg.add_define("WEB_SERVER_DEFAULT_HEADERS_COUNT", 1)
     if CORE.is_esp32:
+        # Count for StaticVector in web_server_idf - matches headers added in init()
+        cg.add_define("WEB_SERVER_DEFAULT_HEADERS_COUNT", 1)
         return
 
     # ESP32 uses IDF web server (early return above), so this is for other Arduino platforms

esphome/components/web_server_base/stream_head.h

@@ -1,126 +0,0 @@
-#pragma once
-
-#include <cstddef>
-#include <cstdint>
-#include <cstring>
-#include <span>
-
-#include <strings.h>
-
-// Pure helpers of AsyncStreamResponse, kept free of httpd and lwIP so they build on the host.
-// The iovec type is a template parameter: lwIP defines its own on the device, POSIX on the host.
-
-namespace esphome::web_server_base {
-
-struct HttpHeader {
-  const char *name;
-  const char *value;
-};
-
-/// Drops the first sent bytes from a gather list after a write; a short write resumes inside the
-/// element it stopped in. count reaches 0 once everything went out.
-template<typename Iovec> void advance_iovecs(Iovec *&iov, int &count, size_t sent) {
-  while (count > 0 && sent >= iov->iov_len) {
-    sent -= iov->iov_len;
-    ++iov;
-    --count;
-  }
-  if (count > 0) {
-    iov->iov_base = static_cast<char *>(iov->iov_base) + sent;
-    iov->iov_len -= sent;
-  }
-}
-
-/// Pieces in a head: 4 for the status line and Content-Type, 4 per header, the blank line and the body
-constexpr size_t stream_head_iov_count(size_t header_count) { return 4 + 4 * header_count + 2; }
-
-/// Pieces per batch: a head with up to four headers goes out in one write; 192 bytes on 32-bit
-static constexpr int STREAM_HEAD_MAX_IOV = 24;
-
-/// True if headers holds a header named name, compared case insensitively
-inline bool has_header(std::span<const HttpHeader> headers, const char *name) {
-  for (const auto &header : headers) {
-    if (strcasecmp(header.name, name) == 0) {
-      return true;
-    }
-  }
-  return false;
-}
-
-/// Gathers the head of a stream response: "HTTP/1.1 <status>", Content-Type, headers, the default_headers that
-/// headers does not replace, the blank line and body (may be null). send(Iovec *iov, int count) gets each full batch of
-/// STREAM_HEAD_MAX_IOV pieces and then the rest; once it returns false the later pieces are skipped. Returns false if
-/// any send failed.
-template<typename Iovec, typename Send>
-bool write_stream_head(Send &&send, const char *status, const char *content_type, std::span<const HttpHeader> headers,
-                       std::span<const HttpHeader> default_headers, const char *body) {
-  struct Batch {
-    Send &send;
-    int count{0};
-    bool ok{true};
-    Iovec iov[STREAM_HEAD_MAX_IOV];  // left uninitialized on purpose
-
-    void flush() {
-      if (this->ok && this->count > 0) {
-        this->ok = this->send(this->iov, this->count);
-      }
-      this->count = 0;
-    }
-    void add(const char *buf, size_t len) {
-      if (this->count == STREAM_HEAD_MAX_IOV) {
-        this->flush();
-      }
-      this->iov[this->count++] = {const_cast<char *>(buf), len};
-    }
-    void add(const char *str) { this->add(str, strlen(str)); }
-    void add_header(const HttpHeader &header) {
-      this->add("\r\n", 2);
-      this->add(header.name);
-      this->add(": ", 2);
-      this->add(header.value);
-    }
-  } batch{send};
-
-  batch.add("HTTP/1.1 ", 9);
-  batch.add(status);
-  batch.add("\r\nContent-Type: ", 16);
-  batch.add(content_type);
-  for (const auto &header : headers) {
-    batch.add_header(header);
-  }
-  for (const auto &header : default_headers) {
-    if (!has_header(headers, header.name)) {
-      batch.add_header(header);
-    }
-  }
-  batch.add("\r\n\r\n", 4);
-  if (body != nullptr) {
-    batch.add(body);
-  }
-  batch.flush();
-  return batch.ok;
-}
-
-/// How long sends have made no progress. Zero means not running, so a start at millis() == 0 records 1.
-/// The comparison is wraparound safe across a millis() rollover.
-class StallClock {
- public:
-  /// Starts the clock; a running clock keeps its start time.
-  void start(uint32_t now) {
-    if (this->started_ms_ == 0) {
-      this->started_ms_ = now != 0 ? now : 1;
-    }
-  }
-  void clear() { this->started_ms_ = 0; }
-  bool is_running() const { return this->started_ms_ != 0; }
-  /// True once the clock has run for timeout_ms
-  bool is_expired(uint32_t now, uint32_t timeout_ms) const {
-    return this->started_ms_ != 0 && static_cast<int32_t>(now - (this->started_ms_ + timeout_ms)) >= 0;
-  }
-  uint32_t elapsed(uint32_t now) const { return now - this->started_ms_; }
-
- protected:
-  uint32_t started_ms_{0};
-};
-
-}  // namespace esphome::web_server_base

esphome/components/web_server_base/stream_response.h

@@ -1,16 +0,0 @@
-#pragma once
-
-#include "esphome/core/defines.h"
-
-// The long lived stream response the /events layer writes to, for the platform's web server
-#ifdef USE_ESP32
-#include "esphome/components/web_server_idf/web_server_idf.h"
-
-#include <sys/socket.h>
-
-namespace esphome::web_server_base {
-using web_server_idf::AsyncStreamResponse;
-// One entry of the gather list AsyncStreamResponse::write() takes
-using StreamIovec = ::iovec;
-}  // namespace esphome::web_server_base
-#endif

esphome/components/web_server_idf/sse_chunk.cpp

@@ -5,7 +5,7 @@
 
 #include "esphome/core/helpers.h"
 
-namespace esphome::web_server {
+namespace esphome::web_server_idf {
 
 void for_each_chunk_piece(const char *message, size_t message_len, ChunkPieceSink sink, void *ctx) {
   if (message == nullptr) {
@@ -63,4 +63,4 @@ void write_chunk_header(char *buf, size_t chunk_len) {
   buf[9] = '\n';
 }
 
-}  // namespace esphome::web_server
+}  // namespace esphome::web_server_idf

esphome/components/web_server_idf/sse_chunk.h

@@ -3,9 +3,7 @@
 #include <cstddef>
 #include <cstdint>
 
-#include "esphome/components/web_server_base/stream_head.h"
-
-namespace esphome::web_server {
+namespace esphome::web_server_idf {
 
 // Wire framing of one Server-Sent-Events event inside the chunked /events response:
 //   "%08x\r\n" + fields + "data: " line ["\r\ndata: " line]... "\r\n" + "\r\n" + "\r\n"
@@ -24,17 +22,6 @@ constexpr size_t SSE_SEP_LEN = sizeof(SSE_SEP) - 1;
 constexpr char SSE_SUFFIX[] = "\r\n\r\n\r\n";
 constexpr size_t SSE_SUFFIX_LEN = sizeof(SSE_SUFFIX) - 1;
 
-// Status, Content-Type and headers of the /events response head
-constexpr char SSE_STATUS[] = "200 OK";
-constexpr char SSE_CONTENT_TYPE[] = "text/event-stream";
-constexpr web_server_base::HttpHeader SSE_HEADERS[] = {
-    {"Transfer-Encoding", "chunked"},
-    {"Cache-Control", "no-cache"},
-    {"Connection", "keep-alive"},
-};
-// Hand-framed first chunk holding an empty event (one blank line), as httpd_resp_send_chunk() sent it before
-constexpr char SSE_FIRST_CHUNK[] = "2\r\n\r\n\r\n";
-
 // Receives one piece of the chunk; ctx is whatever the caller passed to for_each_chunk_piece()
 using ChunkPieceSink = void (*)(void *ctx, const char *piece, size_t len);
 
@@ -50,4 +37,4 @@ size_t build_chunk_prefix(char *buf, size_t size, const char *event, uint32_t id
 // Fills the header placeholder at the start of buf once the chunk length is known
 void write_chunk_header(char *buf, size_t chunk_len);
 
-}  // namespace esphome::web_server
+}  // namespace esphome::web_server_idf

esphome/components/web_server_idf/web_server_idf.cpp

@@ -17,6 +17,7 @@
 #include <freertos/task.h>
 
 #include "utils.h"
+#include "sse_chunk.h"
 #include "web_server_idf.h"
 
 #ifdef USE_WEBSERVER_AUTH_DIGEST
@@ -29,15 +30,17 @@
 #include "multipart.h"  // For parse_multipart_boundary and other utils
 #endif
 
+#ifdef USE_WEBSERVER
+#include "esphome/components/web_server/web_server.h"
+#include "esphome/components/web_server/list_entities.h"
+#endif  // USE_WEBSERVER
+
 // Include socket headers after Arduino headers to avoid IPADDR_NONE/INADDR_NONE macro conflicts
 #include <cerrno>
 #include <sys/socket.h>
 
 namespace esphome::web_server_idf {
 
-using web_server_base::advance_iovecs;
-using web_server_base::write_stream_head;
-
 // Status strings not provided by esp_http_server.h
 #ifndef HTTPD_401
 #define HTTPD_401 "401 Unauthorized"
@@ -52,10 +55,13 @@ using web_server_base::write_stream_head;
 #define HTTPD_422 "422 Unprocessable Entity"
 #endif
 
+#define CRLF_STR "\r\n"
+#define CRLF_LEN (sizeof(CRLF_STR) - 1)
+
 ESPHOME_LOG_TAG(TAG, "web_server_idf");
 
-// Every frame in this file is capped below one JSON arena.
-// Measured at -Os on GCC 14; newer toolchains stay checked on purpose, older ones skip it.
+// Only prepare_json_() may hold the JSON arena: every other frame in this file is capped below one
+// arena. Measured at -Os on GCC 14; newer toolchains stay checked on purpose, older ones skip it.
 #if defined(__GNUC__) && !defined(__clang__) && __GNUC__ >= 14 && defined(__OPTIMIZE_SIZE__)
 #pragma GCC diagnostic error "-Wstack-usage=2048"
 #endif
@@ -134,8 +140,6 @@ void AsyncWebServer::safe_close_with_shutdown(httpd_handle_t hd, int sockfd) {
   //
   // See: https://github.com/esphome/esphome-webserver/issues/163
 
-  AsyncStreamResponse::socket_closing(hd, sockfd);
-
   // Attempt shutdown - ignore errors as socket may already be closed
   shutdown(sockfd, SHUT_RD);
 
@@ -732,84 +736,224 @@ void AsyncResponseStream::printf(const char *fmt, ...) {
   this->print(str);
 }
 
-void AsyncStreamResponse::begin(AsyncWebServerRequest *request, const char *status, const char *content_type,
-                                std::span<const HttpHeader> headers, const char *body) {
-  httpd_req_t *req = *request;
-  // Same head httpd_resp_send_chunk() writes, without its first chunk. Blocking sends: the session's non-blocking
-  // send override is not installed yet.
-  const int fd = httpd_req_to_sockfd(req);
-  const auto &defaults = DefaultHeaders::Instance().headers_;
-  const bool head_sent = write_stream_head<struct iovec>(
-      [fd](struct iovec *iov, int count) {
-        while (count > 0) {
-          struct msghdr msg{};
-          msg.msg_iov = iov;
-          msg.msg_iovlen = count;
-          const ssize_t sent = sendmsg(fd, &msg, 0);
-          if (sent <= 0) {
-            return false;
-          }
-          advance_iovecs(iov, count, sent);
-        }
-        return true;
-      },
-      status, content_type, headers, std::span<const HttpHeader>(defaults.data(), defaults.size()), body);
-
-  req->sess_ctx = this;
-  req->free_ctx = AsyncStreamResponse::destroy;
-
-  this->hd_ = req->handle;
-  this->fd_.store(fd);
-
-  // Use non-blocking send to prevent watchdog timeouts when TCP buffers are full
-  httpd_sess_set_send_override(this->hd_, fd, nonblocking_send);
-
-  if (!head_sent) {
-    // A partly sent head must not be continued by later writes. close() is main loop only, so shut the socket
-    // down here; the first write() then fails and closes the stream, and httpd frees the session as usual.
-    shutdown(fd, SHUT_RDWR);
+#ifdef USE_WEBSERVER
+AsyncEventSource::~AsyncEventSource() {
+  LockGuard guard{this->pending_mutex_};
+  for (auto *vec : {&this->sessions_, &this->pending_sessions_}) {
+    for (auto *ses : *vec) {
+      delete ses;  // NOLINT(cppcoreguidelines-owning-memory)
+    }
   }
 }
 
-// Held around a stream's ownership check and send, and while its socket is marked closed.
-// Created at boot, so no allocation can fail once requests arrive.
-static Mutex stream_send_mutex;  // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
-
-void AsyncStreamResponse::socket_closing(httpd_handle_t hd, int fd) {
-  // Only streams set a session context
-  auto *rsp = static_cast<AsyncStreamResponse *>(httpd_sess_get_ctx(hd, fd));
-  if (rsp == nullptr)
-    return;
-  LockGuard guard{stream_send_mutex};
-  rsp->socket_closed_ = true;
+void AsyncEventSource::handleRequest(AsyncWebServerRequest *request) {
+  // Httpd task: set up the live httpd_req_t and park the session; main loop does the rest.
+  // NOLINTNEXTLINE(cppcoreguidelines-owning-memory,clang-analyzer-cplusplus.NewDeleteLeaks)
+  auto *rsp = new AsyncEventSourceResponse(request, this, this->web_server_);
+  {
+    LockGuard guard{this->pending_mutex_};
+    this->pending_sessions_.push_back(rsp);
+    this->has_pending_sessions_.store(true, std::memory_order_release);
+  }
+  this->web_server_->enable_loop_soon_any_context();
 }
 
-void AsyncStreamResponse::destroy(void *ptr) {
-  auto *rsp = static_cast<AsyncStreamResponse *>(ptr);
+// clang-analyzer traces a false-positive leak path from loop() through
+// adopt_pending_sessions_main_loop_() into start_session_main_loop_() and
+// finally ArduinoJson. Suppress along the entire in-our-code call chain.
+// NOLINTBEGIN(clang-analyzer-cplusplus.NewDeleteLeaks)
+bool AsyncEventSource::loop() {
+  // Fast path: one atomic load per tick. Slow path is out-of-line on connect.
+  if (this->has_pending_sessions_.load(std::memory_order_acquire)) {
+    this->adopt_pending_sessions_main_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
+  for (size_t i = 0; i < this->sessions_.size();) {
+    auto *ses = this->sessions_[i];
+    // If the session has a dead socket (marked by destroy callback)
+    if (ses->safe_to_delete_()) {
+      // destroy() already logged the close with the fd; don't double-log here.
+      delete ses;  // NOLINT(cppcoreguidelines-owning-memory)
+      // Remove by swapping with last element (O(1) removal, order doesn't matter for sessions)
+      this->sessions_[i] = this->sessions_.back();
+      this->sessions_.pop_back();
+    } else {
+      ses->loop();
+      ++i;
+    }
+  }
+  // A session still waiting for httpd to commit its context keeps the loop alive too
+  return !this->sessions_.empty() || this->has_pending_sessions_.load(std::memory_order_acquire);
+}
+
+void AsyncEventSource::adopt_pending_sessions_main_loop_() {
+  std::vector<AsyncEventSourceResponse *> incoming;
+  {
+    LockGuard guard{this->pending_mutex_};
+    incoming.swap(this->pending_sessions_);
+    this->has_pending_sessions_.store(false, std::memory_order_relaxed);
+  }
+  for (auto *rsp : incoming) {
+    // Already disconnected? Drop it; skip session start on a dead session.
+    if (rsp->safe_to_delete_()) {
+      delete rsp;  // NOLINT(cppcoreguidelines-owning-memory)
+      continue;
+    }
+    // httpd commits the session context only after the creating handler returns, so stay
+    // pending until then; httpd_req_cleanup() always commits it or calls destroy()
+    if (httpd_sess_get_ctx(rsp->hd_, rsp->fd_.load()) != rsp) {
+      LockGuard guard{this->pending_mutex_};
+      this->pending_sessions_.push_back(rsp);
+      this->has_pending_sessions_.store(true, std::memory_order_release);
+      continue;
+    }
+    this->sessions_.push_back(rsp);
+    rsp->start_session_main_loop_();
+  }
+}
+// NOLINTEND(clang-analyzer-cplusplus.NewDeleteLeaks)
+
+void AsyncEventSource::try_send_nodefer(const char *message, size_t message_len, const char *event, uint32_t id,
+                                        uint32_t reconnect) {
+  for (auto *ses : this->sessions_) {
+    if (ses->fd_.load() != 0) {  // Skip dead sessions
+      ses->try_send_nodefer(message, message_len, event, id, reconnect);
+    }
+  }
+}
+
+void AsyncEventSource::deferrable_send_state(void *source, bool detail_all, message_generator_t *message_generator) {
+  // Skip if no connected clients to avoid unnecessary processing
+  if (this->empty())
+    return;
+  for (auto *ses : this->sessions_) {
+    if (ses->fd_.load() != 0) {  // Skip dead sessions
+      ses->deferrable_send_state(source, detail_all, message_generator);
+    }
+  }
+}
+
+AsyncEventSourceResponse::AsyncEventSourceResponse(const AsyncWebServerRequest *request,
+                                                   esphome::web_server_idf::AsyncEventSource *server,
+                                                   esphome::web_server::WebServer *ws)
+    : server_(server), web_server_(ws), entities_iterator_(ws, server) {
+  // Httpd task only. start_session_main_loop_() sends the greeting and starts the iterator.
+  httpd_req_t *req = *request;
+
+  httpd_resp_set_status(req, HTTPD_200);
+  httpd_resp_set_type(req, "text/event-stream");
+  httpd_resp_set_hdr(req, "Cache-Control", "no-cache");
+  httpd_resp_set_hdr(req, "Connection", "keep-alive");
+
+  DefaultHeaders::Instance().apply_(req);
+
+  httpd_resp_send_chunk(req, CRLF_STR, CRLF_LEN);
+
+  req->sess_ctx = this;
+  req->free_ctx = AsyncEventSourceResponse::destroy;
+
+  this->hd_ = req->handle;
+  this->fd_.store(httpd_req_to_sockfd(req));
+
+  // Use non-blocking send to prevent watchdog timeouts when TCP buffers are full
+  httpd_sess_set_send_override(this->hd_, this->fd_.load(), nonblocking_send);
+}
+
+// NOLINTBEGIN(clang-analyzer-cplusplus.NewDeleteLeaks) false positive with ArduinoJson
+void AsyncEventSourceResponse::start_session_main_loop_() {
+  auto *ws = this->web_server_;
+
+  // The tcp send buffer is empty on connect. A refusal is a closing session or a failed tail
+  // allocation; nothing retries the greeting, so close and let the client reconnect.
+  auto message = ws->get_config_json();
+  if (!this->try_send_nodefer(message.c_str(), message.size(), "ping", millis(), 30000)) {
+    ESP_LOGW(TAG, "Config not sent to fd %d", this->fd_.load());
+    this->request_close_();
+    return;
+  }
+
+#ifdef USE_WEBSERVER_SORTING
+  for (auto &group : ws->sorting_groups_) {
+    json::JsonBuilder builder;
+    JsonObject root = builder.root();
+    root["name"] = json::linked(group.second.name);
+    root["sorting_weight"] = group.second.weight;
+    message = builder.serialize();
+
+    // a (very) large number of these should be able to be queued initially without defer
+    // since the only thing in the send buffer at this point is the initial ping/config.
+    // A refusal means the socket is full or closing; the remaining groups are not sent. The
+    // session stays up on purpose: partial grouping beats a reconnect loop on a slow link.
+    if (!this->try_send_nodefer(message.c_str(), message.size(), "sorting_group")) {
+      ESP_LOGW(TAG, "Sorting groups not sent to fd %d", this->fd_.load());
+      break;
+    }
+  }
+#endif
+
+  this->entities_iterator_.begin(ws->include_internal_);
+}
+// NOLINTEND(clang-analyzer-cplusplus.NewDeleteLeaks)
+
+void AsyncEventSourceResponse::destroy(void *ptr) {
+  auto *rsp = static_cast<AsyncEventSourceResponse *>(ptr);
   int fd = rsp->fd_.exchange(0);  // Atomically get and clear fd
-  ESP_LOGD(TAG, "Stream connection closed (fd: %d)", fd);
-  // Mark as dead - the owner deletes it from the main loop once safe_to_delete().
+  ESP_LOGD(TAG, "Event source connection closed (fd: %d)", fd);
+  // Mark as dead - will be cleaned up in the main loop
+  // Note: We don't delete or remove from set here to avoid race conditions
   // httpd will call our custom close_fn (safe_close_with_shutdown) which handles
   // shutdown() before close() to prevent lwIP race conditions
 }
 
-void AsyncStreamResponse::close() {
+// helper for allowing only unique entries in the queue
+void AsyncEventSourceResponse::deq_push_back_with_dedup_(void *source, message_generator_t *message_generator) {
+  DeferredEvent item(source, message_generator);
+
+  // Use range-based for loop instead of std::find_if to reduce template instantiation overhead and binary size
+  for (auto &event : this->deferred_queue_) {
+    if (event == item) {
+      return;  // Already in queue, no need to update since items are equal
+    }
+  }
+  this->deferred_queue_.push_back(item);
+}
+
+void AsyncEventSourceResponse::process_deferred_queue_() {
+  if (this->close_requested_) {
+    return;
+  }
+  while (!deferred_queue_.empty()) {
+    DeferredEvent &de = deferred_queue_.front();
+    if (this->send_json_(de.source_, de.message_generator_)) {
+      if (this->close_requested_ || deferred_queue_.empty()) {
+        return;
+      }
+      // O(n) but memory efficiency is more important than speed here which is why std::vector was chosen
+      deferred_queue_.erase(deferred_queue_.begin());
+    } else {
+      break;
+    }
+  }
+}
+
+void AsyncEventSourceResponse::request_close_() {
   if (!this->close_requested_) {
     this->close_requested_ = true;
+    this->deferred_queue_.clear();
+    this->tail_.reset();
+    this->tail_cap_ = 0;
+    this->tail_len_ = 0;
+    this->tail_sent_ = 0;
     this->next_close_attempt_ms_ = App.get_loop_component_start_time();
   }
+
   this->process_close_();
 }
 
-void AsyncStreamResponse::loop() {
-  if (this->close_requested_) {
-    this->process_close_();
-  } else {
-    this->refresh_lru_();
-  }
-}
-
-void AsyncStreamResponse::process_close_() {
+void AsyncEventSourceResponse::process_close_() {
   if (!this->close_requested_ || this->close_work_queued_.load(std::memory_order_acquire)) {
     return;
   }
@@ -827,7 +971,7 @@ void AsyncStreamResponse::process_close_() {
   // httpd_sess_trigger_close() queues only a reusable fd/session slot and can
   // therefore close a new client if the original peer disconnects meanwhile.
   this->close_work_queued_.store(true, std::memory_order_release);
-  const esp_err_t err = httpd_queue_work(this->hd_, &AsyncStreamResponse::close_session_work, this);
+  const esp_err_t err = httpd_queue_work(this->hd_, &AsyncEventSourceResponse::close_session_work, this);
   this->next_close_attempt_ms_ = now + (err == ESP_OK ? CLOSE_CONFIRM_INTERVAL_MS : CLOSE_RETRY_INTERVAL_MS);
   if (err == ESP_OK) {
     return;
@@ -835,17 +979,17 @@ void AsyncStreamResponse::process_close_() {
 
   this->close_work_queued_.store(false, std::memory_order_release);
   if (!this->close_retry_warning_logged_) {
-    ESP_LOGW(TAG, "Failed to queue stream close (%s); retrying", esp_err_to_name(err));
+    ESP_LOGW(TAG, "Failed to queue EventSource close (%s); retrying", esp_err_to_name(err));
     this->close_retry_warning_logged_ = true;
   }
 }
 
-void AsyncStreamResponse::refresh_lru_() {
-  // httpd only advances a session's LRU counter when a request arrives, and a stream receives one
+void AsyncEventSourceResponse::refresh_lru_() {
+  // httpd only advances a session's LRU counter when a request arrives, and an event stream receives one
   // request ever, so the purge that frees a socket for a new client would evict the live streams first.
   // The HTTPD task owns the session table, so the refresh runs there.
   // A stalled stream is the socket the purge should take, and a zeroed fd has no session left to refresh.
-  if (this->stall_.is_running() || this->fd_.load() == 0)
+  if (this->send_failure_started_ms_ != 0 || this->fd_.load() == 0)
     return;
   const uint32_t now = App.get_loop_component_start_time();
   if (now - this->last_lru_refresh_ms_ < LRU_REFRESH_INTERVAL_MS ||
@@ -854,13 +998,13 @@ void AsyncStreamResponse::refresh_lru_() {
   }
   this->last_lru_refresh_ms_ = now;  // a full queue also waits for the next interval
   this->lru_work_queued_.store(true, std::memory_order_release);
-  if (httpd_queue_work(this->hd_, &AsyncStreamResponse::refresh_lru_work, this) != ESP_OK) {
+  if (httpd_queue_work(this->hd_, &AsyncEventSourceResponse::refresh_lru_work, this) != ESP_OK) {
     this->lru_work_queued_.store(false, std::memory_order_release);
   }
 }
 
-void AsyncStreamResponse::refresh_lru_work(void *arg) {
-  auto *response = static_cast<AsyncStreamResponse *>(arg);
+void AsyncEventSourceResponse::refresh_lru_work(void *arg) {
+  auto *response = static_cast<AsyncEventSourceResponse *>(arg);
   const int fd = response->fd_.load();
   if (response->owns_session_(fd)) {
     httpd_sess_update_lru_counter(response->hd_, fd);
@@ -869,8 +1013,8 @@ void AsyncStreamResponse::refresh_lru_work(void *arg) {
   response->lru_work_queued_.store(false, std::memory_order_release);
 }
 
-void AsyncStreamResponse::close_session_work(void *arg) {
-  auto *response = static_cast<AsyncStreamResponse *>(arg);
+void AsyncEventSourceResponse::close_session_work(void *arg) {
+  auto *response = static_cast<AsyncEventSourceResponse *>(arg);
   const int fd = response->fd_.load();
   if (response->owns_session_(fd)) {
     // The HTTPD task remains the session owner. Shutting the socket down makes
@@ -882,61 +1026,320 @@ void AsyncStreamResponse::close_session_work(void *arg) {
   response->close_work_queued_.store(false, std::memory_order_release);
 }
 
-ssize_t AsyncStreamResponse::write(struct iovec *iov, int iovcnt) {
-  if (this->close_requested_) {
-    return -1;
-  }
-  // Nothing to send is not a stall
-  size_t total = 0;
-  for (int i = 0; i < iovcnt; i++) {
-    total += iov[i].iov_len;
-  }
-  if (total == 0) {
-    return 0;
-  }
-  // httpd closes the socket before it clears the session, so the fd number may already be a new client's.
+ssize_t AsyncEventSourceResponse::send_(struct iovec *iov, int iovcnt) {
+  // httpd frees a session before closing its socket, so the fd may already be a new client's.
   // Treated as would-block; the stall timer ends a session that never becomes ours again.
   const int fd = this->fd_.load();
-  ssize_t sent = 0;
-  int err = 0;
-  {
-    // Held across the check and the send, so socket_closing() cannot close the socket in between
-    LockGuard guard{stream_send_mutex};
-    if (!this->socket_closed_ && this->owns_session_(fd)) {
-      struct msghdr msg{};
-      msg.msg_iov = iov;
-      msg.msg_iovlen = iovcnt;
-      sent = sendmsg(fd, &msg, MSG_DONTWAIT);
-      err = errno;
-    }
+  if (!this->owns_session_(fd)) {
+    return 0;
   }
-  if (sent > 0) {
-    this->stall_.clear();
+  struct msghdr msg{};
+  msg.msg_iov = iov;
+  msg.msg_iovlen = iovcnt;
+  const ssize_t sent = sendmsg(fd, &msg, MSG_DONTWAIT);
+  if (sent >= 0) {
     return sent;
   }
-  if (sent < 0 && err != EAGAIN && err != EWOULDBLOCK) {
-    ESP_LOGD(TAG, "send error: errno %d", err);
-    // A hard error is not a timeout, so is_stalled() must not report one
-    this->stall_.clear();
-    this->close();
-    return -1;
+  const int err = errno;
+  if (err == EAGAIN || err == EWOULDBLOCK) {
+    return 0;
   }
-  // Socket buffer full, try again later; a socket that stays full past the stall timeout is closed
-  const uint32_t now = App.get_loop_component_start_time();
-  this->stall_.start(now);
-  if (this->stall_.is_expired(now, SEND_STALL_TIMEOUT_MS)) {
-    ESP_LOGW(TAG, "Closing stuck stream after %" PRIu32 " ms without send progress", this->stall_.elapsed(now));
-    this->close();
-    return -1;
-  }
-  return 0;
+  ESP_LOGD(TAG, "send error: errno %d", err);
+  this->request_close_();
+  return -1;
 }
 
-ssize_t AsyncStreamResponse::write(const void *data, size_t len) {
-  struct iovec iov = {const_cast<void *>(data), len};
-  return this->write(&iov, 1);
+void AsyncEventSourceResponse::drain_tail_() {
+  if (this->sending_ || this->close_requested_ || this->tail_len_ == 0) {
+    return;
+  }
+  SendGuard guard{*this};
+
+  const size_t remaining = this->tail_len_ - this->tail_sent_;
+  struct iovec iov = {this->tail_.get() + this->tail_sent_, remaining};
+  const ssize_t sent = this->send_(&iov, 1);
+  if (sent < 0) {
+    return;
+  }
+  if (sent == 0) {
+    // Socket buffer full, try again later
+    // NOTE: Similar logic exists in web_server/web_server.cpp in DeferredUpdateEventSource::process_deferred_queue_().
+    // The IDF path is intentionally time-based and closes through HTTPD to preserve session ownership.
+    const uint32_t now = App.get_loop_component_start_time();
+    if (this->send_failure_started_ms_ == 0) {
+      this->send_failure_started_ms_ = now != 0 ? now : 1;  // Reserve zero for no stall.
+    }
+    if (static_cast<int32_t>(now - (this->send_failure_started_ms_ + SEND_STALL_TIMEOUT_MS)) >= 0) {
+      ESP_LOGW(TAG, "Closing stuck EventSource connection after %" PRIu32 " ms without send progress",
+               now - this->send_failure_started_ms_);
+      this->request_close_();
+    }
+    return;
+  }
+
+  this->send_failure_started_ms_ = 0;
+  this->tail_sent_ += sent;
+  if (this->tail_sent_ < this->tail_len_) {
+    ESP_LOGV(TAG, "Partial send: %zd/%zu bytes (total: %u/%u)", sent, remaining, this->tail_sent_, this->tail_len_);
+    return;
+  }
+  // Fully sent; the storage stays for the next stall
+  this->tail_len_ = 0;
 }
 
+bool AsyncEventSourceResponse::reserve_tail_(size_t len) {
+  if (this->tail_cap_ >= len) {
+    return true;
+  }
+  if (len > TAIL_MAX_SIZE) {
+    return false;
+  }
+  // Nothing is pending while the tail grows, so free the old block first. PREFER_INTERNAL keeps
+  // the tail where plain new put it.
+  this->tail_.reset();
+  this->tail_ = RAMAllocator<uint8_t>(RAMAllocator<uint8_t>::PREFER_INTERNAL).make_unique_array_for_overwrite(len);
+  this->tail_cap_ = this->tail_ ? len : 0;
+  return this->tail_cap_ != 0;
+}
+
+bool AsyncEventSourceResponse::stash_chunk_(const char *prefix, size_t prefix_len, const char *message,
+                                            size_t message_len, size_t total, size_t sent) {
+  // A log event of nothing but line breaks is the largest chunk; reserve_tail_() must never
+  // refuse it for size, only for memory
+  static_assert(SSE_SEP_LEN * web_server::LOG_EVENT_MAX_LEN + PREFIX_BUF_SIZE + SSE_SUFFIX_LEN <= TAIL_MAX_SIZE,
+                "the log cut in web_server.h must keep a worst case log event inside the tail ceiling");
+  if (!this->reserve_tail_(total)) {
+    if (sent != 0) {
+      // Part of the chunk is on the wire, so the stream is broken and the client has to go
+      ESP_LOGW(TAG, "Cannot buffer a %zu byte chunk, closing", total);
+      this->request_close_();
+    } else {
+      this->tail_alloc_failed_(total);  // nothing on the wire, the caller retries on the stall clock
+    }
+    return false;
+  }
+  uint8_t *dst = this->tail_.get();
+  std::memcpy(dst, prefix, prefix_len);
+  dst += prefix_len;
+  for_each_chunk_piece(
+      message, message_len,
+      [](void *ctx, const char *piece, size_t len) {
+        auto &out = *static_cast<uint8_t **>(ctx);
+        std::memcpy(out, piece, len);
+        out += len;
+      },
+      &dst);
+  this->tail_len_ = total;
+  this->tail_sent_ = sent;
+  return true;
+}
+
+void AsyncEventSourceResponse::loop() {
+  if (this->close_requested_) {
+    this->process_close_();
+    return;
+  }
+  this->refresh_lru_();
+  drain_tail_();
+  process_deferred_queue_();
+  if (this->close_requested_)
+    return;
+  // One step per loop; refusals retry next pass
+  this->entities_iterator_.try_advance(1);
+}
+
+#if defined(__GNUC__) && !defined(__clang__) && __GNUC__ >= 14 && defined(__OPTIMIZE_SIZE__)
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wstack-usage="  // the one frame that holds the JSON arena
+#endif
+AsyncEventSourceResponse::JsonPrepareResult AsyncEventSourceResponse::prepare_json_(void *source,
+                                                                                    message_generator_t *generator,
+                                                                                    char *buffer, size_t *size_out) {
+  // Log lines emitted while the generator runs are dropped for this session: on the large path they would fill
+  // tail_ before it is overwritten, and on the small path they would go out ahead of the event being built.
+  SendGuard guard{*this};
+  json::JsonArena<json::JSON_ARENA_SIZE> arena;
+  json::JsonBuilder builder(&arena);
+  generator(this->web_server_, source, builder);
+  *size_out = builder.serialize_to(buffer, JSON_BUF_SIZE);
+  if (*size_out < JSON_BUF_SIZE) {
+    return JsonPrepareResult::JSON_PREPARE_RESULT_BUFFER;
+  }
+
+  char prefix[PREFIX_BUF_SIZE];
+  const size_t prefix_len = build_chunk_prefix(prefix, sizeof(prefix), "state", 0, 0, true);
+
+  // Nothing reaches the wire until the complete document and its framing are owned by the caller.
+  size_t json_len = 0;
+  size_t cap = std::max<size_t>(JSON_BUF_SIZE * 2, this->tail_cap_);
+  for (;;) {
+    if (!this->reserve_tail_(cap)) {
+      *size_out = cap;
+      return JsonPrepareResult::JSON_PREPARE_RESULT_NO_MEMORY;
+    }
+    const size_t room = cap - prefix_len - SSE_SUFFIX_LEN;
+    json_len = builder.serialize_to(reinterpret_cast<char *>(this->tail_.get()) + prefix_len, room);
+    if (json_len < room) {
+      break;
+    }
+    if (cap >= TAIL_MAX_SIZE) {
+      *size_out = room;
+      return JsonPrepareResult::JSON_PREPARE_RESULT_TOO_LARGE;
+    }
+    cap = std::min<size_t>(cap * 2, TAIL_MAX_SIZE);
+  }
+
+  const size_t total = prefix_len + json_len + SSE_SUFFIX_LEN;
+  write_chunk_header(prefix, total - CHUNK_HDR_LEN - CHUNK_END_LEN);
+  uint8_t *dst = this->tail_.get();
+  std::memcpy(dst, prefix, prefix_len);
+  std::memcpy(dst + prefix_len + json_len, SSE_SUFFIX, SSE_SUFFIX_LEN);
+  this->tail_len_ = total;
+  this->tail_sent_ = 0;
+  return JsonPrepareResult::JSON_PREPARE_RESULT_TAIL;
+}
+#if defined(__GNUC__) && !defined(__clang__) && __GNUC__ >= 14 && defined(__OPTIMIZE_SIZE__)
+#pragma GCC diagnostic pop
+#endif
+
+bool AsyncEventSourceResponse::send_json_(void *source, message_generator_t *generator) {
+  if (!this->ready_to_send_()) {
+    return false;
+  }
+  char buffer[JSON_BUF_SIZE];
+  size_t size;
+  const auto result = this->prepare_json_(source, generator, buffer, &size);
+  switch (result) {
+    case JsonPrepareResult::JSON_PREPARE_RESULT_BUFFER:
+      return this->try_send_nodefer(buffer, size, "state");
+    case JsonPrepareResult::JSON_PREPARE_RESULT_TAIL:
+      this->drain_tail_();
+      return true;
+    case JsonPrepareResult::JSON_PREPARE_RESULT_NO_MEMORY: {
+      SendGuard guard{*this};
+      this->tail_alloc_failed_(size);
+      return false;
+    }
+    case JsonPrepareResult::JSON_PREPARE_RESULT_TOO_LARGE: {
+      SendGuard guard{*this};
+      ESP_LOGW(TAG, "State event does not fit %zu bytes, dropped", size);
+      this->tail_.reset();
+      this->tail_cap_ = 0;
+      this->send_failure_started_ms_ = 0;
+      return true;
+    }
+  }
+  return false;
+}
+
+void AsyncEventSourceResponse::tail_alloc_failed_(size_t cap) {
+  // Same stall clock as a socket that stops draining, so a session cannot retry forever
+  const uint32_t now = App.get_loop_component_start_time();
+  if (this->send_failure_started_ms_ == 0) {
+    this->send_failure_started_ms_ = now != 0 ? now : 1;  // Reserve zero for no stall.
+    ESP_LOGW(TAG, "No memory for a %zu byte chunk", cap);
+    return;
+  }
+  if (static_cast<int32_t>(now - (this->send_failure_started_ms_ + SEND_STALL_TIMEOUT_MS)) >= 0) {
+    ESP_LOGW(TAG, "Closing EventSource after %" PRIu32 " ms without memory", now - this->send_failure_started_ms_);
+    this->request_close_();
+  }
+}
+
+bool AsyncEventSourceResponse::ready_to_send_() {
+  if (this->sending_ || this->fd_.load() == 0 || this->close_requested_) {
+    return false;
+  }
+  drain_tail_();
+  return !this->close_requested_ && this->tail_len_ == 0;
+}
+
+bool AsyncEventSourceResponse::try_send_nodefer(const char *message, size_t message_len, const char *event, uint32_t id,
+                                                uint32_t reconnect) {
+  if (!this->ready_to_send_()) {
+    return false;
+  }
+  SendGuard guard{*this};
+
+  // Everything after the prefix goes out straight from the caller's buffer
+  char prefix[PREFIX_BUF_SIZE];
+  const size_t prefix_len = build_chunk_prefix(prefix, sizeof(prefix), event, id, reconnect, message != nullptr);
+  if (message == nullptr && prefix_len == CHUNK_HDR_LEN) {
+    return true;  // Match ESPAsyncWebServer: nothing to send
+  }
+  if (prefix_len >= PREFIX_BUF_SIZE - 1) {
+    // The appenders truncate silently, which would put a malformed event on the wire
+    ESP_LOGW(TAG, "Event name too long, dropped");
+    return true;
+  }
+
+  // Gather list: the prefix, then the data lines and their separators from the message
+  struct Gather {
+    struct iovec iov[MAX_SEND_IOV];  // left uninitialized on purpose
+    int iovcnt{1};
+    size_t total{0};
+    bool fits{true};
+  } g;
+  g.total = prefix_len;
+  for_each_chunk_piece(
+      message, message_len,
+      [](void *ctx, const char *piece, size_t len) {
+        auto &g = *static_cast<Gather *>(ctx);
+        g.total += len;
+        if (len == 0) {
+          return;
+        }
+        if (g.iovcnt == MAX_SEND_IOV) {
+          g.fits = false;
+          return;
+        }
+        g.iov[g.iovcnt++] = {const_cast<char *>(piece), len};
+      },
+      &g);
+  // The header and the terminator are not part of the chunk length
+  write_chunk_header(prefix, g.total - CHUNK_HDR_LEN - CHUNK_END_LEN);
+  g.iov[0] = {prefix, prefix_len};
+
+  // A message with more lines than the list holds skips straight to the tail
+  const ssize_t sent = g.fits ? this->send_(g.iov, g.iovcnt) : 0;
+  if (sent < 0) {
+    return false;
+  }
+  if (static_cast<size_t>(sent) == g.total) {
+    this->send_failure_started_ms_ = 0;  // progress, whichever stall clock was running
+    return true;
+  }
+  // The caller's buffers do not outlive this call, so keep the chunk and continue from loop()
+  return this->stash_chunk_(prefix, prefix_len, message, message_len, g.total, sent);
+}
+
+void AsyncEventSourceResponse::deferrable_send_state(void *source, bool detail_all,
+                                                     message_generator_t *message_generator) {
+  // Let the detail_all listing finish before bare state events, so the web UI never shows unnamed entries and
+  // initial connect sends less
+  if (!detail_all && !this->entities_iterator_.completed())
+    return;
+
+  drain_tail_();
+  process_deferred_queue_();
+
+  if (this->close_requested_) {
+    return;
+  }
+
+  if (this->tail_len_ != 0 || !deferred_queue_.empty()) {
+    // outgoing event buffer or deferred queue still not empty which means downstream tcp send buffer full, no point
+    // trying to send first
+    deq_push_back_with_dedup_(source, message_generator);
+  } else {
+    // A send error closes the session and clears the queue; nothing is queued after that
+    if (!this->send_json_(source, message_generator) && !this->close_requested_) {
+      deq_push_back_with_dedup_(source, message_generator);
+    }
+  }
+}
+#endif
+
 #ifdef USE_WEBSERVER_OTA
 esp_err_t AsyncWebServer::handle_multipart_upload_(httpd_req_t *r, const char *content_type) {
   // Parse boundary and create reader

esphome/components/web_server_idf/web_server_idf.h

@@ -15,14 +15,21 @@
 #include <utility>
 #include <vector>
 
-#include "esphome/components/web_server_base/stream_head.h"
+#ifdef USE_WEBSERVER
+#include "esphome/components/json/json_util.h"
+#include "esphome/components/web_server/list_entities.h"
+#include "sse_chunk.h"
+#endif
 
 struct iovec;  // NOLINT(readability-identifier-naming) - forward decl of lwip's gather list entry
 
-namespace esphome::web_server_idf {
-
-using web_server_base::HttpHeader;
-using web_server_base::StallClock;
+namespace esphome {
+#ifdef USE_WEBSERVER
+namespace web_server {
+class WebServer;
+};  // namespace web_server
+#endif
+namespace web_server_idf {
 
 class AsyncWebParameter {
  public:
@@ -184,7 +191,6 @@ class AsyncWebServerRequest {
   void onDisconnect(std::function<void()> fn) { this->on_disconnect_ = std::move(fn); }
 #endif
 
-  /// sess_ctx is reserved for AsyncStreamResponse: the server's close_fn treats any session context as one.
   operator httpd_req_t *() const { return this->req_; }
   optional<std::string> get_header(const char *name) const;
   /// Copy a header value into buf without heap allocation.
@@ -263,85 +269,204 @@ class AsyncWebHandler {
   virtual bool isRequestHandlerTrivial() const { return true; }
 };
 
-/// A response that keeps its connection open and is written from the main loop, such as an event stream or
-/// an audio stream. Use it directly or derive from it. Call loop() and delete the object only through the type
-/// it was created as: subclasses hide loop() instead of overriding it, and the destructor is not virtual.
-///
-/// Lifecycle:
-///  1. Httpd task, inside handleRequest(): create it and call begin(), then hand the pointer to the main loop.
-///  2. Main loop: once is_attached() is true, write() may go through. Call loop() on every pass.
-///  3. Main loop: delete it once safe_to_delete() is true. httpd owns the socket throughout.
-class AsyncStreamResponse {
- public:
-  /// Httpd task, once. Sends the status line (for example HTTPD_200), Content-Type, headers, the default headers
-  /// and body (a C string or null, sent as is), then makes this object the session context. A chunked response
-  /// passes {"Transfer-Encoding", "chunked"} as its first header and frames each chunk itself.
-  void begin(AsyncWebServerRequest *request, const char *status, const char *content_type,
-             std::span<const HttpHeader> headers, const char *body = nullptr);
+#ifdef USE_WEBSERVER
+class AsyncEventSource;
+class AsyncEventSourceResponse;
 
-  /// Main loop. Non-blocking gather write. Returns the bytes accepted, which may be fewer than offered: the caller
-  /// keeps the rest and continues from there. 0 means the socket is full or not ours yet; -1 means the stream failed
-  /// and is closing. Refused writes for SEND_STALL_TIMEOUT_MS in a row also close it.
-  ssize_t write(struct iovec *iov, int iovcnt);
-  ssize_t write(const void *data, size_t len);
-  /// Main loop. Shuts the connection down through httpd; safe to call more than once.
-  void close();
-  /// Main loop, every pass. Keeps the session fresh in httpd's LRU and retries a pending close.
-  void loop();
-  /// Httpd task, from the server's close_fn: no write() reaches the socket after this returns.
-  static void socket_closing(httpd_handle_t hd, int fd);
+using message_generator_t = void(esphome::web_server::WebServer *, void *, json::JsonBuilder &);
 
-  /// httpd holds this response as the session context; begin() has sent the whole head by then.
-  bool is_attached() const { return this->owns_session_(this->fd_.load()); }
-  /// The client is still connected; httpd has not freed the session.
-  bool is_connected() const { return this->fd_.load() != 0; }
-  bool is_closing() const { return this->close_requested_; }
-  /// The socket fd, for log lines only
-  int log_id() const { return this->fd_.load(); }
-  /// Sends are making no progress; after write() returns -1 this tells a send timeout from a send error.
-  bool is_stalled() const { return this->stall_.is_running(); }
-  /// The session is gone and no queued httpd work still references this object.
-  bool safe_to_delete() const {
-    return this->fd_.load() == 0 && !this->close_work_queued_.load(std::memory_order_acquire) &&
-           !this->lru_work_queued_.load(std::memory_order_acquire);
-  }
-
-  static constexpr uint32_t SEND_STALL_TIMEOUT_MS = 20000;
+/*
+  This class holds a pointer to the source component that wants to publish a state event, and a pointer to a function
+  that will lazily generate that event.  The two pointers allow dedup in the deferred queue if multiple publishes for
+  the same component are backed up, and take up only two pointers of memory.  The entry in the deferred queue (a
+  std::vector) is the DeferredEvent instance itself (not a pointer to one elsewhere in heap) so still only two pointers
+  per entry (and no heap fragmentation).  Even 100 backed up events (you'd have to have at least 100 sensors publishing
+  because of dedup) would take up only 0.8 kB.
+*/
+struct DeferredEvent {
+  friend class AsyncEventSourceResponse;
 
  protected:
+  void *source_;
+  message_generator_t *message_generator_;
+
+ public:
+  DeferredEvent(void *source, message_generator_t *message_generator)
+      : source_(source), message_generator_(message_generator) {}
+  bool operator==(const DeferredEvent &test) const {
+    return (source_ == test.source_ && message_generator_ == test.message_generator_);
+  }
+};
+static_assert(sizeof(DeferredEvent) == sizeof(void *) + sizeof(message_generator_t *),
+              "DeferredEvent should have no padding");
+
+class AsyncEventSourceResponse {
+  friend class AsyncEventSource;
+
+ public:
+  bool try_send_nodefer(const char *message, size_t message_len, const char *event = nullptr, uint32_t id = 0,
+                        uint32_t reconnect = 0);
+  // detail_all events go out while entities are still being listed; state events wait until listing is done
+  void deferrable_send_state(void *source, bool detail_all, message_generator_t *message_generator);
+  // The old event name argument would otherwise convert to true
+  void deferrable_send_state(void *source, const char *event_type, message_generator_t *message_generator) = delete;
+  void loop();
+
+ protected:
+  AsyncEventSourceResponse(const AsyncWebServerRequest *request, esphome::web_server_idf::AsyncEventSource *server,
+                           esphome::web_server::WebServer *ws);
+
+  // Main-loop only: sends initial ping/config/sorting_groups, starts entity iterator.
+  void start_session_main_loop_();
+
+  void deq_push_back_with_dedup_(void *source, message_generator_t *message_generator);
+  void process_deferred_queue_();
+  // A new chunk may go out: not re-entered from a log line, session alive, tail empty
+  bool ready_to_send_();
+  // Non-blocking gather write. Returns bytes written, 0 on would-block or a socket that is not
+  // ours, -1 after requesting the close on any other error.
+  ssize_t send_(struct iovec *iov, int iovcnt);
+  // Push what is left of the chunk in tail_ to the socket; owns the stall timer.
+  void drain_tail_();
+  // Grow tail_ to hold len bytes, kept at its high-water mark. False on OOM.
+  bool reserve_tail_(size_t len);
+  // Keep the whole chunk in tail_ and continue from sent; false when the tail cannot be allocated.
+  bool stash_chunk_(const char *prefix, size_t prefix_len, const char *message, size_t message_len, size_t total,
+                    size_t sent);
+  // Send a state event; JSON too large for the stack buffer is serialized into tail_ instead
+  bool send_json_(void *source, message_generator_t *generator);
+  enum class JsonPrepareResult : uint8_t {
+    JSON_PREPARE_RESULT_BUFFER,
+    JSON_PREPARE_RESULT_TAIL,
+    JSON_PREPARE_RESULT_NO_MEMORY,
+    JSON_PREPARE_RESULT_TOO_LARGE,
+  };
+  // Keep the arena out of the caller's frame so it is gone before the state send.
+  // size_out: BUFFER = JSON length, NO_MEMORY = requested tail capacity, TOO_LARGE = available JSON room;
+  // unused for TAIL.
+  JsonPrepareResult __attribute__((noinline)) prepare_json_(void *source, message_generator_t *generator, char *buffer,
+                                                            size_t *size_out);
+  // Warn once, and close the session once the stall timeout passes with no memory for the tail
+  void tail_alloc_failed_(size_t cap);
+  void request_close_();
+
+  // A log line emitted inside a send re-enters try_send_nodefer on this session; refuse it
+  struct SendGuard {
+    AsyncEventSourceResponse &owner;
+    explicit SendGuard(AsyncEventSourceResponse &owner) : owner(owner) { owner.sending_ = true; }
+    ~SendGuard() { this->owner.sending_ = false; }
+  };
   void process_close_();
   static void close_session_work(void *arg);
   void refresh_lru_();
   static void refresh_lru_work(void *arg);
-  // httpd closes the socket before it clears the session, so this can still match a closed fd; write() also checks
-  // socket_closed_
+  // httpd frees a session before closing its socket, so the fd may already belong to a new client
   bool owns_session_(int fd) const { return fd != 0 && httpd_sess_get_ctx(this->hd_, fd) == this; }
-  static void destroy(void *p);
 
+  // Deletable only after destroy() zeroed fd_ and no queued HTTPD work still references this object.
+  bool safe_to_delete_() const {
+    return this->fd_.load() == 0 && !this->close_work_queued_.load(std::memory_order_acquire) &&
+           !this->lru_work_queued_.load(std::memory_order_acquire);
+  }
+
+  static void destroy(void *p);
+  AsyncEventSource *server_;
   httpd_handle_t hd_{};
   std::atomic<int> fd_{};
-  // Runs while sends make no progress; a running clock keeps its start time
-  StallClock stall_;
+  std::vector<DeferredEvent> deferred_queue_;
+  esphome::web_server::WebServer *web_server_;
+  esphome::web_server::ListEntitiesIterator entities_iterator_;
+  // One chunk the socket did not take whole, allocated on the first stall; the only heap use
+  // on the send path
+  RAMUniquePtr<uint8_t[]> tail_;
+  uint32_t send_failure_started_ms_{0};  // Zero means no send stall in progress.
   uint32_t next_close_attempt_ms_{0};
   uint32_t last_lru_refresh_ms_{0};
+  uint16_t tail_cap_{0};
+  uint16_t tail_len_{0};  // Zero means nothing pending
+  uint16_t tail_sent_{0};
   // Set on the main loop before queueing close work, cleared by the HTTPD-task callback when done.
   std::atomic<bool> close_work_queued_{false};
   std::atomic<bool> lru_work_queued_{false};
-  // Guarded by the stream send mutex
-  bool socket_closed_{false};
   // Main-loop only; the HTTPD task never reads or writes these flags.
   bool close_requested_{false};
   bool close_retry_warning_logged_{false};
+  bool sending_{false};
+  // The longest multi line log message in the tree (a climate dump_config) has 22 lines; a
+  // longer one goes through the tail
+  static constexpr size_t MAX_SEND_LINES = 22;
+  static constexpr size_t MAX_SEND_IOV = 1 + 2 * MAX_SEND_LINES;
+  // Chunk header, retry/id/event lines and the first "data: "
+  static constexpr size_t PREFIX_BUF_SIZE = 128;
 
+  // Stack buffer for a state event's JSON; a larger document is serialized into the tail
+  static constexpr size_t JSON_BUF_SIZE = 1024;
+  // Same ceiling JsonBuilder::serialize() applies (max_heap_size in json_util.cpp); a larger
+  // document is dropped before anything is on the wire
+  static constexpr size_t JSON_MAX_SIZE = 5120;
+  // Most RAM a stalled session keeps: the largest state document plus any accepted framing
+  static constexpr size_t TAIL_MAX_SIZE = JSON_MAX_SIZE + PREFIX_BUF_SIZE + SSE_SUFFIX_LEN;
+  static constexpr uint32_t SEND_STALL_TIMEOUT_MS = 20000;
   static constexpr uint32_t CLOSE_RETRY_INTERVAL_MS = 250;
   static constexpr uint32_t CLOSE_CONFIRM_INTERVAL_MS = 1000;
   static constexpr uint32_t LRU_REFRESH_INTERVAL_MS = 1000;
 };
 
+using AsyncEventSourceClient = AsyncEventSourceResponse;
+
+class AsyncEventSource : public AsyncWebHandler {
+  friend class AsyncEventSourceResponse;
+
+ public:
+  AsyncEventSource(StringRef url, esphome::web_server::WebServer *ws) : url_(url), web_server_(ws) {}
+  ~AsyncEventSource() override;
+
+  // NOLINTNEXTLINE(readability-identifier-naming)
+  bool canHandle(AsyncWebServerRequest *request) const override {
+    if (request->method() != HTTP_GET)
+      return false;
+    char url_buf[AsyncWebServerRequest::URL_BUF_SIZE];
+    return request->url_to(url_buf) == this->url_;
+  }
+  // NOLINTNEXTLINE(readability-identifier-naming)
+  void handleRequest(AsyncWebServerRequest *request) override;
+  void try_send_nodefer(const char *message, size_t message_len, const char *event = nullptr, uint32_t id = 0,
+                        uint32_t reconnect = 0);
+  void deferrable_send_state(void *source, bool detail_all, message_generator_t *message_generator);
+  // The old event name argument would otherwise convert to true
+  void deferrable_send_state(void *source, const char *event_type, message_generator_t *message_generator) = delete;
+  /// Returns true if there are sessions remaining (including pending cleanup).
+  bool loop();
+  bool empty() { return this->count() == 0; }
+
+  size_t count() const { return this->sessions_.size(); }
+
+ protected:
+  // Cold path: move sessions from pending_sessions_ into sessions_ and greet each one.
+  void __attribute__((noinline, cold)) adopt_pending_sessions_main_loop_();
+
+  StringRef url_;  // Must outlive this object (string literal)
+  // Main-loop only. Vector: SSE sessions are 1-5 connections, linear search beats set.
+  std::vector<AsyncEventSourceResponse *> sessions_;
+  // Httpd-task intake; guarded by pending_mutex_, gated by has_pending_sessions_.
+  std::vector<AsyncEventSourceResponse *> pending_sessions_;
+  Mutex pending_mutex_;
+  esphome::web_server::WebServer *web_server_;
+  std::atomic<bool> has_pending_sessions_{false};
+};
+#endif  // USE_WEBSERVER
+
+struct HttpHeader {
+  const char *name;
+  const char *value;
+};
+
 class DefaultHeaders {
   friend class AsyncWebServerRequest;
-  friend class AsyncStreamResponse;
   friend class AsyncWebServerResponse;
+#ifdef USE_WEBSERVER
+  friend class AsyncEventSourceResponse;
+#endif
   static_assert(WEB_SERVER_DEFAULT_HEADERS_COUNT <= 8, "AsyncWebServerResponse::replaced_defaults_ holds 8 bits");
 
  public:
@@ -359,7 +484,8 @@ class DefaultHeaders {
   void apply_(httpd_req_t *req, uint8_t skip_mask = 0) const;
 };
 
-}  // namespace esphome::web_server_idf
+}  // namespace web_server_idf
+}  // namespace esphome
 
 using namespace esphome::web_server_idf;  // NOLINT(google-global-names-in-headers)
 

esphome/core/defines.h

@@ -109,6 +109,9 @@
 #define USE_EZSP_PROXY_TAP
 #define USE_FAN
 #define USE_FONT
+#define USE_FEEDBACK_COVER_ENDSTOP
+#define USE_FEEDBACK_COVER_MOVEMENT
+#define USE_FEEDBACK_COVER_OBSTACLE
 #define USE_FEEDBACK_COVER_OVERSHOOT
 #define USE_GPIO_BINARY_SENSOR_INTERRUPT
 #define USE_GPIO_SWITCH_INTERLOCK

tests/components/feedback/test-minimal.esp8266-ard.yaml

@@ -0,0 +1,13 @@
+# No endstop, movement or obstacle sensors, so their code is compiled out
+cover:
+  - platform: feedback
+    id: feedback_minimal
+    name: Minimal Feedback Cover
+    open_action:
+      - logger.log: open
+    open_duration: 10s
+    close_action:
+      - logger.log: close
+    close_duration: 10s
+    stop_action:
+      - logger.log: stop

tests/components/web_server/test_stream_head.cpp

@@ -1,300 +0,0 @@
-#ifdef USE_HOST
-#include <gtest/gtest.h>
-
-#include <sys/uio.h>
-
-#include <cstdint>
-#include <cstdio>
-#include <string>
-#include <vector>
-
-#include "esphome/components/web_server/sse_chunk.h"
-#include "esphome/components/web_server_base/stream_head.h"
-
-namespace esphome::web_server::testing {
-
-using namespace web_server_base;  // NOLINT(google-build-using-namespace)
-
-// Collects every batch write_stream_head() sends
-struct Capture {
-  std::string bytes;
-  std::vector<int> batch_sizes;
-  int attempts{0};
-  int fail_after_batches{-1};
-
-  bool operator()(struct iovec *iov, int count) {
-    this->attempts++;
-    if (static_cast<int>(this->batch_sizes.size()) == this->fail_after_batches) {
-      return false;
-    }
-    this->batch_sizes.push_back(count);
-    for (int i = 0; i < count; i++) {
-      this->bytes.append(static_cast<const char *>(iov[i].iov_base), iov[i].iov_len);
-    }
-    return true;
-  }
-};
-
-static std::string build_head(Capture &cap, const char *status, const char *content_type,
-                              std::span<const HttpHeader> headers, std::span<const HttpHeader> defaults,
-                              const char *body) {
-  EXPECT_TRUE(write_stream_head<struct iovec>(cap, status, content_type, headers, defaults, body));
-  return cap.bytes;
-}
-
-// What httpd_resp_send_chunk() put on the wire for its first chunk (esp_http_server/src/httpd_txrx.c in
-// ESP-IDF 5.5): the chunked head with Transfer-Encoding, each httpd_resp_set_hdr() header, the blank line,
-// then "%lx\r\n", the chunk data and "\r\n".
-static std::string reference_send_chunk(const char *status, const char *content_type,
-                                        const std::vector<HttpHeader> &resp_hdrs, const char *buf, size_t buf_len) {
-  char head[256];
-  std::string out(
-      head, snprintf(head, sizeof(head), "HTTP/1.1 %s\r\nContent-Type: %s\r\nTransfer-Encoding: chunked\r\n", status,
-                     content_type));
-  for (const auto &hdr : resp_hdrs) {
-    out.append(hdr.name).append(": ").append(hdr.value).append("\r\n");
-  }
-  out.append("\r\n");
-  char len_str[10];
-  out.append(len_str, snprintf(len_str, sizeof(len_str), "%lx\r\n", static_cast<long>(buf_len)));
-  out.append(buf, buf_len);
-  out.append("\r\n");
-  return out;
-}
-
-static constexpr HttpHeader DEFAULTS[] = {
-    {"Access-Control-Allow-Origin", "*"},
-    {"Access-Control-Allow-Private-Network", "true"},
-};
-
-TEST(StreamHead, SseHeadMatchesHttpdRespSendChunk) {
-  // The old event source set Cache-Control and Connection, then the default headers, and sent "\r\n" as a chunk
-  const std::string expected = reference_send_chunk(
-      "200 OK", "text/event-stream",
-      {{"Cache-Control", "no-cache"}, {"Connection", "keep-alive"}, DEFAULTS[0], DEFAULTS[1]}, "\r\n", 2);
-  Capture cap;
-  EXPECT_EQ(build_head(cap, SSE_STATUS, SSE_CONTENT_TYPE, SSE_HEADERS, DEFAULTS, SSE_FIRST_CHUNK), expected);
-}
-
-TEST(StreamHead, HeaderReplacesDefaultOfSameName) {
-  // A default the caller also sets, in any case, goes out once with the caller's value
-  static constexpr HttpHeader HEADERS[] = {
-      {"access-control-allow-origin", "http://example.com"},
-  };
-  Capture cap;
-  EXPECT_EQ(build_head(cap, "200 OK", "text/plain", HEADERS, DEFAULTS, nullptr),
-            "HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\naccess-control-allow-origin: http://example.com"
-            "\r\nAccess-Control-Allow-Private-Network: true\r\n\r\n");
-}
-
-TEST(StreamHead, RawHeadWithCustomHeadersAndNoBody) {
-  static constexpr HttpHeader HEADERS[] = {
-      {"Cache-Control", "no-store"},
-      {"X-Stream", "audio"},
-  };
-  Capture cap;
-  EXPECT_EQ(build_head(cap, "200 OK", "audio/wav", HEADERS, {}, nullptr), "HTTP/1.1 200 OK\r\n"
-                                                                          "Content-Type: audio/wav\r\n"
-                                                                          "Cache-Control: no-store\r\n"
-                                                                          "X-Stream: audio\r\n"
-                                                                          "\r\n");
-}
-
-TEST(StreamHead, BareHead) {
-  Capture cap;
-  EXPECT_EQ(build_head(cap, "204 No Content", "text/plain", {}, {}, nullptr),
-            "HTTP/1.1 204 No Content\r\nContent-Type: text/plain\r\n\r\n");
-  EXPECT_EQ(cap.batch_sizes, std::vector<int>{5});
-}
-
-TEST(StreamHead, ManyHeadersSplitIntoBatchesWithSameBytes) {
-  // 4 + 12 * 4 + 2 = 54 pieces, more than two batches hold
-  static constexpr HttpHeader HEADERS[] = {
-      {"X-Header-0", "value-0"}, {"X-Header-1", "value-1"}, {"X-Header-2", "value-2"}, {"X-Header-3", "value-3"},
-      {"X-Header-4", "value-4"}, {"X-Header-5", "value-5"}, {"X-Header-6", "value-6"}, {"X-Header-7", "value-7"},
-      {"X-Header-8", "value-8"}, {"X-Header-9", "value-9"},
-  };
-  const std::span<const HttpHeader> headers{HEADERS};
-
-  Capture batched;
-  const std::string bytes = build_head(batched, "200 OK", "text/plain", headers, DEFAULTS, "body");
-  EXPECT_EQ(batched.batch_sizes, (std::vector<int>{STREAM_HEAD_MAX_IOV, STREAM_HEAD_MAX_IOV, 6}));
-
-  std::string expected = "HTTP/1.1 200 OK\r\nContent-Type: text/plain";
-  const auto append_headers = [&expected](std::span<const HttpHeader> list) {
-    for (const auto &h : list) {
-      expected.append("\r\n").append(h.name).append(": ").append(h.value);
-    }
-  };
-  append_headers(headers);
-  append_headers(DEFAULTS);
-  expected.append("\r\n\r\nbody");
-  EXPECT_EQ(bytes, expected);
-}
-
-TEST(StreamHead, BatchBoundary) {
-  // 4 + 4 * 4 + 1 + 1 = 22 pieces fit one batch; a fifth header makes 4 + 5 * 4 = 24 before the blank line
-  static constexpr HttpHeader HEADERS[] = {{"A", "1"}, {"B", "2"}, {"C", "3"}, {"D", "4"}, {"E", "5"}};
-  Capture four;
-  build_head(four, "200 OK", "text/plain", std::span(HEADERS, 4), {}, "x");
-  EXPECT_EQ(four.batch_sizes, std::vector<int>{22});
-  EXPECT_EQ(stream_head_iov_count(4), 22u);
-  Capture five;
-  const std::string bytes = build_head(five, "200 OK", "text/plain", HEADERS, {}, "x");
-  EXPECT_EQ(five.batch_sizes, (std::vector<int>{STREAM_HEAD_MAX_IOV, 2}));
-  EXPECT_EQ(bytes, "HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nA: 1\r\nB: 2\r\nC: 3\r\nD: 4\r\nE: 5\r\n\r\nx");
-}
-
-TEST(StreamHead, FailedSendSkipsLaterBatches) {
-  // Three batches; the second fails, so the third is never sent
-  std::vector<HttpHeader> headers(12, HttpHeader{"X", "y"});
-  Capture full;
-  build_head(full, "200 OK", "text/plain", headers, {}, nullptr);
-  ASSERT_EQ(full.batch_sizes.size(), 3u);
-  Capture cap;
-  cap.fail_after_batches = 1;
-  EXPECT_FALSE(write_stream_head<struct iovec>(cap, "200 OK", "text/plain", headers, {}, nullptr));
-  EXPECT_EQ(cap.attempts, 2);
-  EXPECT_EQ(cap.batch_sizes.size(), 1u);
-}
-
-// Three elements of 3, 4 and 5 bytes
-struct IovFixture {
-  char a[4] = "abc";
-  char b[5] = "defg";
-  char c[6] = "hijkl";
-  struct iovec iov[3] = {{a, 3}, {b, 4}, {c, 5}};
-  struct iovec *cur = iov;
-  int count = 3;
-
-  std::string rest() const {
-    std::string out;
-    for (int i = 0; i < this->count; i++) {
-      out.append(static_cast<const char *>(this->cur[i].iov_base), this->cur[i].iov_len);
-    }
-    return out;
-  }
-};
-
-TEST(AdvanceIovecs, ZeroLengthWriteChangesNothing) {
-  IovFixture f;
-  advance_iovecs(f.cur, f.count, 0);
-  EXPECT_EQ(f.cur, f.iov);
-  EXPECT_EQ(f.count, 3);
-  EXPECT_EQ(f.rest(), "abcdefghijkl");
-}
-
-TEST(AdvanceIovecs, PartialInsideFirstElement) {
-  IovFixture f;
-  advance_iovecs(f.cur, f.count, 2);
-  EXPECT_EQ(f.cur, f.iov);
-  EXPECT_EQ(f.count, 3);
-  EXPECT_EQ(f.cur->iov_len, 1u);
-  EXPECT_EQ(f.rest(), "cdefghijkl");
-}
-
-TEST(AdvanceIovecs, ExactlyOnElementBoundary) {
-  IovFixture f;
-  advance_iovecs(f.cur, f.count, 3);
-  EXPECT_EQ(f.cur, &f.iov[1]);
-  EXPECT_EQ(f.count, 2);
-  EXPECT_EQ(f.rest(), "defghijkl");
-}
-
-TEST(AdvanceIovecs, AcrossElementBoundaries) {
-  IovFixture f;
-  advance_iovecs(f.cur, f.count, 9);
-  EXPECT_EQ(f.cur, &f.iov[2]);
-  EXPECT_EQ(f.count, 1);
-  EXPECT_EQ(f.rest(), "jkl");
-}
-
-TEST(AdvanceIovecs, RepeatedShortWrites) {
-  IovFixture f;
-  std::string sent;
-  while (f.count > 0) {
-    const std::string before = f.rest();
-    advance_iovecs(f.cur, f.count, 2);
-    sent += before.substr(0, before.size() - f.rest().size());
-  }
-  EXPECT_EQ(sent, "abcdefghijkl");
-}
-
-TEST(AdvanceIovecs, EverythingWritten) {
-  IovFixture f;
-  advance_iovecs(f.cur, f.count, 12);
-  EXPECT_EQ(f.count, 0);
-  EXPECT_EQ(f.rest(), "");
-}
-
-TEST(AdvanceIovecs, SkipsZeroLengthElements) {
-  char a[] = "ab";
-  char c[] = "cd";
-  struct iovec iov[3] = {{a, 2}, {nullptr, 0}, {c, 2}};
-  struct iovec *cur = iov;
-  int count = 3;
-  advance_iovecs(cur, count, 2);
-  EXPECT_EQ(cur, &iov[2]);
-  EXPECT_EQ(count, 1);
-}
-
-static constexpr uint32_t TIMEOUT = 20000;
-
-TEST(StallClock, NotStarted) {
-  StallClock clock;
-  EXPECT_FALSE(clock.is_running());
-  EXPECT_FALSE(clock.is_expired(0, TIMEOUT));
-  EXPECT_FALSE(clock.is_expired(UINT32_MAX, TIMEOUT));
-}
-
-TEST(StallClock, ExpiresAtExactlyTheTimeout) {
-  StallClock clock;
-  clock.start(1000);
-  EXPECT_TRUE(clock.is_running());
-  EXPECT_FALSE(clock.is_expired(1000, TIMEOUT));
-  EXPECT_FALSE(clock.is_expired(1000 + TIMEOUT - 1, TIMEOUT));
-  EXPECT_TRUE(clock.is_expired(1000 + TIMEOUT, TIMEOUT));
-  EXPECT_TRUE(clock.is_expired(1000 + TIMEOUT + 1, TIMEOUT));
-  EXPECT_EQ(clock.elapsed(1000 + TIMEOUT), TIMEOUT);
-}
-
-TEST(StallClock, RunningClockKeepsItsStart) {
-  StallClock clock;
-  clock.start(1000);
-  clock.start(5000);
-  EXPECT_TRUE(clock.is_expired(1000 + TIMEOUT, TIMEOUT));
-}
-
-TEST(StallClock, Cleared) {
-  StallClock clock;
-  clock.start(1000);
-  clock.clear();
-  EXPECT_FALSE(clock.is_running());
-  EXPECT_FALSE(clock.is_expired(1000 + TIMEOUT, TIMEOUT));
-  clock.start(50000);
-  EXPECT_FALSE(clock.is_expired(50000 + TIMEOUT - 1, TIMEOUT));
-  EXPECT_TRUE(clock.is_expired(50000 + TIMEOUT, TIMEOUT));
-}
-
-TEST(StallClock, StartAtZeroStillRuns) {
-  StallClock clock;
-  clock.start(0);
-  EXPECT_TRUE(clock.is_running());
-  // Recorded as 1, so it expires one millisecond late
-  EXPECT_FALSE(clock.is_expired(TIMEOUT, TIMEOUT));
-  EXPECT_TRUE(clock.is_expired(TIMEOUT + 1, TIMEOUT));
-}
-
-TEST(StallClock, AcrossMillisWraparound) {
-  StallClock clock;
-  const uint32_t start = UINT32_MAX - 5000;
-  clock.start(start);
-  EXPECT_FALSE(clock.is_expired(UINT32_MAX, TIMEOUT));
-  EXPECT_FALSE(clock.is_expired(start + TIMEOUT - 1, TIMEOUT));  // wrapped to 14998
-  EXPECT_TRUE(clock.is_expired(start + TIMEOUT, TIMEOUT));
-  EXPECT_EQ(clock.elapsed(start + TIMEOUT), TIMEOUT);
-}
-
-}  // namespace esphome::web_server::testing
-
-#endif  // USE_HOST

tests/components/web_server_idf/test_sse_chunk.cpp

@@ -6,9 +6,9 @@
 #include <random>
 #include <string>
 
-#include "esphome/components/web_server/sse_chunk.h"
+#include "esphome/components/web_server_idf/sse_chunk.h"
 
-namespace esphome::web_server::testing {
+namespace esphome::web_server_idf::testing {
 
 // The chunk as try_send_nodefer lays it out: header, fields, first "data: ", then the pieces
 static std::string build_chunk(const char *message, size_t message_len, const char *event, uint32_t id,
@@ -128,4 +128,4 @@ TEST(SseChunk, MatchesReferenceOnRandomMessages) {
   }
 }
 
-}  // namespace esphome::web_server::testing
+}  // namespace esphome::web_server_idf::testing

tests/integration/fixtures/feedback_cover_loop_wake.yaml

@@ -0,0 +1,69 @@
+esphome:
+  name: feedback-loop-wake
+
+host:
+api:
+  batch_delay: 0ms
+logger:
+  level: DEBUG
+
+# Stand in for a motor started outside ESPHome
+switch:
+  - platform: template
+    name: "External Open"
+    id: external_open
+    optimistic: true
+  - platform: template
+    name: "External Close"
+    id: external_close
+    optimistic: true
+
+binary_sensor:
+  - platform: template
+    id: open_moving
+    internal: true
+    condition:
+      switch.is_on: external_open
+  - platform: template
+    id: close_moving
+    internal: true
+    condition:
+      switch.is_on: external_close
+
+cover:
+  - platform: feedback
+    name: "Time Cover"
+    open_action:
+      - logger.log: "time_cover open"
+    close_action:
+      - logger.log: "time_cover close"
+    stop_action:
+      - logger.log: "time_cover stop"
+    open_duration: 2s
+    close_duration: 2s
+
+  - platform: feedback
+    name: "Max Duration Cover"
+    open_action:
+      - logger.log: "max_cover open"
+    close_action:
+      - logger.log: "max_cover close"
+    stop_action:
+      - logger.log: "max_cover stop"
+    open_duration: 3s
+    close_duration: 3s
+    max_duration: 1s
+
+  - platform: feedback
+    name: "Sensor Cover"
+    open_action:
+      - logger.log: "sensor_cover open"
+    close_action:
+      - logger.log: "sensor_cover close"
+    stop_action:
+      - logger.log: "sensor_cover stop"
+    open_duration: 4s
+    close_duration: 4s
+    open_sensor: open_moving
+    close_sensor: close_moving
+    update_interval: 200ms

tests/integration/test_feedback_cover_loop_wake.py

@@ -0,0 +1,100 @@
+"""Integration test that every way a feedback cover starts moving wakes its idle loop."""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Awaitable
+
+from aioesphomeapi import CoverInfo, CoverOperation, CoverState, EntityState, SwitchInfo
+import pytest
+
+from .log_utils import LineWaiter
+from .state_utils import StateWaiter, require_entity, subscribe_after_initial_states
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+def _expect_idle(waiter: StateWaiter, cover: CoverInfo) -> Awaitable[EntityState]:
+    return waiter.expect(
+        lambda s: (
+            isinstance(s, CoverState)
+            and s.key == cover.key
+            and s.current_operation == CoverOperation.IDLE
+        ),
+        label=f"{cover.name} IDLE",
+    )
+
+
+@pytest.mark.asyncio
+async def test_feedback_cover_loop_wake(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test that target position, max_duration and movement sensors still work with the loop idled."""
+    lines = LineWaiter()
+    max_stopped = lines.expect("Max duration reached")
+    async with (
+        run_compiled(yaml_config, line_callback=lines.callback),
+        api_client_connected() as client,
+    ):
+        sensor_states: list[CoverState] = []
+        entities, _ = await client.list_entities_services()
+        time_cover = require_entity(entities, "time_cover", CoverInfo)
+        max_cover = require_entity(entities, "max_duration_cover", CoverInfo)
+        sensor_cover = require_entity(entities, "sensor_cover", CoverInfo)
+
+        def on_state(state: EntityState) -> None:
+            if isinstance(state, CoverState) and state.key == sensor_cover.key:
+                sensor_states.append(state)
+
+        sub = await subscribe_after_initial_states(client, entities, on_state=on_state)
+        waiter = sub.waiter
+
+        # Positions are restored from the previous run, so pick targets away from them
+        start = sub.initial_states[time_cover.key].position
+        target = 0.2 if start >= 0.5 else 0.8
+
+        # Only loop() stops a time-only cover at its target
+        done = _expect_idle(waiter, time_cover)
+        client.cover_command(key=time_cover.key, position=target)
+        state = await done
+        assert state.position == pytest.approx(target, abs=0.05)
+
+        # Stop mid move, then a new command must move it again and stop at the target
+        stopped = _expect_idle(waiter, time_cover)
+        client.cover_command(key=time_cover.key, position=start)
+        await asyncio.sleep(0.3)
+        client.cover_command(key=time_cover.key, stop=True)
+        mid = (await stopped).position
+        assert mid != pytest.approx(target, abs=0.05)
+        done = _expect_idle(waiter, time_cover)
+        client.cover_command(key=time_cover.key, position=target)
+        state = await done
+        assert state.position == pytest.approx(target, abs=0.05)
+
+        # 1s max_duration against a 3s travel
+        max_start = sub.initial_states[max_cover.key].position
+        max_target = 0.0 if max_start >= 0.5 else 1.0
+        done = _expect_idle(waiter, max_cover)
+        client.cover_command(key=max_cover.key, position=max_target)
+        state = await done
+        await asyncio.wait_for(max_stopped, 1.0)
+        assert abs(state.position - max_start) == pytest.approx(1 / 3, abs=0.1)
+
+        # Moved by the movement sensor alone, no command
+        sensor_start = sub.initial_states[sensor_cover.key].position
+        switch_name = "external_close" if sensor_start >= 0.5 else "external_open"
+        switch = require_entity(entities, switch_name, SwitchInfo)
+        sensor_states.clear()
+        client.switch_command(switch.key, True)
+        await asyncio.sleep(1.2)
+        settled = _expect_idle(waiter, sensor_cover)
+        client.switch_command(switch.key, False)
+        await settled
+        moving = {
+            round(s.position, 3)
+            for s in sensor_states
+            if s.current_operation != CoverOperation.IDLE
+        }
+        # One state from the sensor callback, the rest are loop() updates every 200ms
+        assert len(moving) >= 4, sensor_states