[core] Fix clang-tidy naming for static test helper

esphome/components/api/api_connection.cpp

@@ -412,15 +412,15 @@ void APIConnection::finalize_iterator_sync_() {
 }
 
 void APIConnection::process_iterator_batch_(ComponentIterator &iterator) {
-  size_t initial_size = this->deferred_batch_.size();
-  size_t max_batch = MAX_INITIAL_PER_BATCH;
-  while (!iterator.completed() && (this->deferred_batch_.size() - initial_size) < max_batch) {
-    iterator.advance();
-  }
+  // Budget by remaining batch capacity so a pass cannot overfill the batch;
+  // stops early on a refused send and resumes next loop pass
+  size_t batch_size = this->deferred_batch_.size();
+  if (batch_size < MAX_INITIAL_BATCH_SIZE)
+    iterator.try_advance(MAX_INITIAL_BATCH_SIZE - batch_size);
 
-  // If the batch is full, process it immediately
-  // Note: iterator.advance() already calls schedule_batch_() via schedule_message_()
-  if (this->deferred_batch_.size() >= max_batch) {
+  // Flush immediately once enough is queued (not guaranteed every pass);
+  // partial batches go out via the batch timer or finalize_iterator_sync_()
+  if (this->deferred_batch_.size() >= MAX_INITIAL_BATCH_SIZE) {
     this->process_batch_();
   }
 }

esphome/components/api/api_connection.h

@@ -53,11 +53,11 @@ void log_dropped_message(const char *tag, int line, const LogString *what);
 
 // Keepalive timeout in milliseconds
 static constexpr uint32_t KEEPALIVE_TIMEOUT_MS = 60000;
-// Maximum number of entities to process in a single batch during initial state/info sending
-static constexpr size_t MAX_INITIAL_PER_BATCH = 34;
+// Deferred batch size cap during initial state/info sync
+static constexpr size_t MAX_INITIAL_BATCH_SIZE = 34;
 // Verify MAX_MESSAGES_PER_BATCH (defined in api_frame_helper.h) can hold the initial batch
-static_assert(MAX_MESSAGES_PER_BATCH >= MAX_INITIAL_PER_BATCH,
-              "MAX_MESSAGES_PER_BATCH must be >= MAX_INITIAL_PER_BATCH");
+static_assert(MAX_MESSAGES_PER_BATCH >= MAX_INITIAL_BATCH_SIZE,
+              "MAX_MESSAGES_PER_BATCH must be >= MAX_INITIAL_BATCH_SIZE");
 
 #ifdef USE_BENCHMARK
 class APIConnection;

esphome/components/api/api_frame_helper.h

@@ -36,7 +36,7 @@ static constexpr uint16_t MAX_MESSAGE_SIZE = 32768;  // 32 KiB for ESP32 and oth
 static constexpr uint16_t RX_BUF_NULL_TERMINATOR = 1;
 
 // Maximum number of messages to batch in a single write operation
-// Must be >= MAX_INITIAL_PER_BATCH in api_connection.h (enforced by static_assert there)
+// Must be >= MAX_INITIAL_BATCH_SIZE in api_connection.h (enforced by static_assert there)
 static constexpr size_t MAX_MESSAGES_PER_BATCH = 34;
 
 // Max client name length (e.g., "Home Assistant 2026.1.0.dev0" = 28 chars)

esphome/components/api/list_entities.cpp

@@ -95,9 +95,17 @@ bool ListEntitiesIterator::on_end() { return this->client_->send_list_info_done(
 ListEntitiesIterator::ListEntitiesIterator(APIConnection *client) : client_(client) {}
 
 #ifdef USE_API_USER_DEFINED_ACTIONS
+// Yield after every Nth service; bounds direct (non-batched) writes per loop pass
+static constexpr uint8_t SERVICE_YIELD_INTERVAL = 3;
+
 bool ListEntitiesIterator::on_service(UserServiceDescriptor *service) {
   auto resp = service->encode_list_service_response();
-  return this->client_->send_message(resp);
+  if (!this->client_->send_message(resp))
+    return false;
+  // at_ is this service's index
+  if ((this->at_ + 1) % SERVICE_YIELD_INTERVAL == 0)
+    this->yield_after_step_();
+  return true;
 }
 #endif
 

esphome/components/web_server/list_entities.h

@@ -35,7 +35,6 @@ class ListEntitiesIterator final : public ComponentIterator {
 #undef ENTITY_TYPE_
 #undef ENTITY_CONTROLLER_TYPE_
   // NOLINTEND(bugprone-macro-parentheses)
-  bool completed() { return this->state_ == IteratorState::NONE; }
 
  protected:
   const WebServer *web_server_;

esphome/components/web_server/web_server.cpp

@@ -214,8 +214,8 @@ void DeferredUpdateEventSource::process_deferred_queue_() {
 
 void DeferredUpdateEventSource::loop() {
   process_deferred_queue_();
-  if (!this->entities_iterator_.completed())
-    this->entities_iterator_.advance();
+  // One step per loop; refusals retry next pass
+  this->entities_iterator_.try_advance(1);
 }
 
 void DeferredUpdateEventSource::deferrable_send_state(void *source, const char *event_type,
@@ -321,12 +321,6 @@ void DeferredUpdateEventSourceList::on_client_connect_(DeferredUpdateEventSource
 #endif
 
     source->entities_iterator_.begin(ws->include_internal_);
-
-    // just dump them all up-front and take advantage of the deferred queue
-    //     on second thought that takes too long, but leaving the commented code here for debug purposes
-    // while(!source->entities_iterator_.completed()) {
-    //  source->entities_iterator_.advance();
-    //}
   });
 }
 

esphome/components/web_server_idf/web_server_idf.cpp

@@ -935,8 +935,8 @@ void AsyncEventSourceResponse::process_buffer_() {
 void AsyncEventSourceResponse::loop() {
   process_buffer_();
   process_deferred_queue_();
-  if (!this->entities_iterator_.completed())
-    this->entities_iterator_.advance();
+  // One step per loop; refusals retry next pass
+  this->entities_iterator_.try_advance(1);
 }
 
 bool AsyncEventSourceResponse::try_send_nodefer(const char *message, size_t message_len, const char *event, uint32_t id,

esphome/core/component_iterator.cpp

@@ -22,23 +22,23 @@ void ComponentIterator::advance_platform_() {
   this->at_ = 0;
 }
 
-void ComponentIterator::advance() {
+bool ComponentIterator::advance_step_() {
   switch (this->state_) {
     case IteratorState::NONE:
       // not started
-      return;
+      return false;
     case IteratorState::BEGIN:
       if (this->on_begin()) {
         advance_platform_();
+        return true;
       }
-      break;
+      return false;
 
 // Entity iterator cases (generated from entity_types.h)
 // NOLINTBEGIN(bugprone-macro-parentheses)
 #define ENTITY_TYPE_(type, singular, plural, count, upper) \
   case IteratorState::upper: \
-    this->process_platform_item_(App.get_##plural(), &ComponentIterator::on_##singular); \
-    break;
+    return this->process_platform_item_(App.get_##plural(), &ComponentIterator::on_##singular);
 #define ENTITY_CONTROLLER_TYPE_(type, singular, plural, count, upper, callback) \
   ENTITY_TYPE_(type, singular, plural, count, upper)
 #include "esphome/core/entity_types.h"
@@ -48,26 +48,29 @@ void ComponentIterator::advance() {
 
 #ifdef USE_API_USER_DEFINED_ACTIONS
     case IteratorState::SERVICE:
-      this->process_platform_item_(api::global_api_server->get_user_services(), &ComponentIterator::on_service);
-      break;
+      return this->process_platform_item_(api::global_api_server->get_user_services(), &ComponentIterator::on_service);
 #endif
 
 #ifdef USE_CAMERA
     case IteratorState::CAMERA: {
       camera::Camera *camera_instance = camera::Camera::instance();
-      if (camera_instance != nullptr && (!camera_instance->is_internal() || this->include_internal_)) {
-        this->on_camera(camera_instance);
+      if (camera_instance != nullptr && (!camera_instance->is_internal() || this->include_internal_) &&
+          !this->on_camera(camera_instance)) {
+        return false;
       }
       advance_platform_();
-    } break;
+      return true;
+    }
 #endif
 
     case IteratorState::MAX:
       if (this->on_end()) {
         this->state_ = IteratorState::NONE;
+        return true;
       }
-      return;
+      return false;
   }
+  return false;
 }
 
 bool ComponentIterator::on_end() { return true; }

esphome/core/component_iterator.h

@@ -30,7 +30,23 @@ class RadioFrequency;
 class ComponentIterator {
  public:
   void begin(bool include_internal = false);
-  void advance();
+  /// Run up to max_steps iteration steps; stops early when iteration
+  /// completes or a callback refuses (that step is retried on the next
+  /// call). Inline so an idle (completed) iterator costs one compare, no call.
+  ESPHOME_ALWAYS_INLINE void try_advance(size_t max_steps) {
+    size_t steps = 0;
+    while (steps < max_steps && !this->completed()) {
+      this->yield_requested_ = false;
+      if (!this->advance_step_())
+        break;
+      steps++;
+      if (this->yield_requested_)
+        break;
+    }
+  }
+  // Remove before 2027.3.0
+  ESPDEPRECATED("Use try_advance() instead. Removed in 2027.3.0", "2026.8.1")
+  void advance() { this->try_advance(1); }
   bool completed() const { return this->state_ == IteratorState::NONE; }
   virtual bool on_begin();
 // Pure virtual entity callbacks (generated from entity_types.h)
@@ -73,23 +89,34 @@ class ComponentIterator {
 #endif
     MAX,
   };
+  /// End the current try_advance() pass after this step; lets callbacks
+  /// that write directly to the socket cap direct writes per pass.
+  void yield_after_step_() { this->yield_requested_ = true; }
+
   uint16_t at_{0};  // Supports up to 65,535 entities per type
   IteratorState state_{IteratorState::NONE};
-  bool include_internal_{false};
+  bool yield_requested_ : 1 {false};
+  bool include_internal_ : 1 {false};
 
   template<typename Container>
-  void process_platform_item_(const Container &items,
+  bool process_platform_item_(const Container &items,
                               bool (ComponentIterator::*on_item)(typename Container::value_type)) {
     if (this->at_ >= items.size()) {
       this->advance_platform_();
-    } else {
-      typename Container::value_type item = items[this->at_];
-      if ((item->is_internal() && !this->include_internal_) || (this->*on_item)(item)) {
-        this->at_++;
-      }
+      return true;
     }
+    typename Container::value_type item = items[this->at_];
+    if ((item->is_internal() && !this->include_internal_) || (this->*on_item)(item)) {
+      this->at_++;
+      return true;
+    }
+    return false;
   }
 
+  /// One iteration step; false if no progress was made (callback refused
+  /// or iterator not running).
+  bool advance_step_();
+
   void advance_platform_();
 };
 

tests/components/camera/__init__.py

@@ -0,0 +1,11 @@
+import esphome.codegen as cg
+from tests.testing_helpers import ComponentManifestOverride
+
+
+def override_manifest(manifest: ComponentManifestOverride) -> None:
+    # No host camera platform exists to emit USE_CAMERA; define it here so
+    # the iterator CAMERA state compiles into the test binary.
+    async def to_code_testing(config):
+        cg.add_define("USE_CAMERA")
+
+    manifest.to_code = to_code_testing

tests/components/camera/test_component_iterator_camera.cpp

@@ -0,0 +1,79 @@
+#include <gtest/gtest.h>
+
+#include "esphome/core/component_iterator.h"
+
+#ifdef USE_CAMERA
+#include "esphome/components/camera/camera.h"
+
+namespace esphome::testing {
+
+class StubCamera : public camera::Camera {
+ public:
+  void add_listener(camera::CameraListener *listener) override {}
+  camera::CameraImageReader *create_image_reader() override { return nullptr; }
+  void request_image(camera::CameraRequester requester) override {}
+  void start_stream(camera::CameraRequester requester) override {}
+  void stop_stream(camera::CameraRequester requester) override {}
+};
+
+// Iterator that accepts everything except the camera, which can refuse a
+// configurable number of times. The CAMERA state is a singleton path
+// distinct from process_platform_item_; this pins the same contract:
+// a refused camera is re-offered, never skipped.
+class CameraRefusingIterator : public ComponentIterator {
+ public:
+// NOLINTBEGIN(bugprone-macro-parentheses)
+#define ENTITY_TYPE_(type, singular, plural, count, upper) \
+  bool on_##singular(type *obj) override { return true; }
+#define ENTITY_CONTROLLER_TYPE_(type, singular, plural, count, upper, callback) \
+  ENTITY_TYPE_(type, singular, plural, count, upper)
+#include "esphome/core/entity_types.h"
+#undef ENTITY_TYPE_
+#undef ENTITY_CONTROLLER_TYPE_
+  // NOLINTEND(bugprone-macro-parentheses)
+
+  bool on_camera(camera::Camera *obj) override {
+    this->camera_calls++;
+    if (this->camera_refusals > 0) {
+      this->camera_refusals--;
+      return false;
+    }
+    return true;
+  }
+
+  int camera_calls{0};
+  int camera_refusals{0};
+};
+
+// Far above the fixed number of iterator states
+static constexpr size_t BIG_BUDGET = 1000;
+
+class ComponentIteratorCameraTest : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    // Constructing a Camera installs the process-wide singleton
+    static StubCamera stub_camera;
+    ASSERT_EQ(camera::Camera::instance(), &stub_camera);
+  }
+};
+
+TEST_F(ComponentIteratorCameraTest, RefusedCameraIsReofferedNotSkipped) {
+  CameraRefusingIterator it;
+  it.camera_refusals = 2;
+  it.begin();
+  // Runs until the camera refuses, which stops the pass
+  it.try_advance(BIG_BUDGET);
+  EXPECT_EQ(it.camera_calls, 1);
+  EXPECT_FALSE(it.completed());
+  // The camera is re-offered once per call, not skipped
+  it.try_advance(BIG_BUDGET);
+  EXPECT_EQ(it.camera_calls, 2);
+  EXPECT_FALSE(it.completed());
+  // Once accepted, the iteration completes
+  it.try_advance(BIG_BUDGET);
+  EXPECT_TRUE(it.completed());
+  EXPECT_EQ(it.camera_calls, 3);
+}
+
+}  // namespace esphome::testing
+#endif  // USE_CAMERA

tests/components/core/benchmark.yaml

@@ -0,0 +1,11 @@
+# Pulls in sensor so entity iteration paths compile (USE_SENSOR);
+# tests register their own instances. Plain yaml.safe_load, no ESPHome tags.
+# An alphabetically-earlier component's sensor: block shadows this one in
+# combined builds; the tests' sensor-count ASSERT catches a capacity drop.
+sensor:
+  - platform: template
+    id: bench_sensor_a
+    name: "Bench A"
+  - platform: template
+    id: bench_sensor_b
+    name: "Bench B"

tests/components/core/test_component_iterator.cpp

@@ -0,0 +1,195 @@
+#include <gtest/gtest.h>
+
+#include "esphome/core/component_iterator.h"
+
+#ifdef USE_SENSOR
+#include "esphome/components/sensor/sensor.h"
+#include "esphome/core/application.h"
+#endif
+
+namespace esphome::testing {
+
+// Iterator whose begin/end callbacks can refuse a configurable number of
+// times; all entity callbacks accept (any registered entities are accepted).
+class RefusingIterator : public ComponentIterator {
+ public:
+// NOLINTBEGIN(bugprone-macro-parentheses)
+#define ENTITY_TYPE_(type, singular, plural, count, upper) \
+  bool on_##singular(type *obj) override { return true; }
+#define ENTITY_CONTROLLER_TYPE_(type, singular, plural, count, upper, callback) \
+  ENTITY_TYPE_(type, singular, plural, count, upper)
+#include "esphome/core/entity_types.h"
+#undef ENTITY_TYPE_
+#undef ENTITY_CONTROLLER_TYPE_
+  // NOLINTEND(bugprone-macro-parentheses)
+
+  bool on_begin() override { return step(this->begin_calls, this->begin_refusals); }
+  bool on_end() override { return step(this->end_calls, this->end_refusals); }
+
+  int begin_calls{0};
+  int end_calls{0};
+  int begin_refusals{0};
+  int end_refusals{0};
+
+ protected:
+  static bool step(int &calls, int &refusals) {
+    calls++;
+    if (refusals > 0) {
+      refusals--;
+      return false;
+    }
+    return true;
+  }
+};
+
+// Far above the fixed number of iterator states
+static constexpr size_t BIG_BUDGET = 1000;
+
+TEST(ComponentIterator, NotRunningMakesNoProgress) {
+  RefusingIterator it;
+  it.try_advance(BIG_BUDGET);
+  EXPECT_TRUE(it.completed());
+  EXPECT_EQ(it.begin_calls, 0);
+  EXPECT_EQ(it.end_calls, 0);
+}
+
+TEST(ComponentIterator, CompletesInOneCallWithoutRefusals) {
+  RefusingIterator it;
+  it.begin();
+  it.try_advance(BIG_BUDGET);
+  EXPECT_TRUE(it.completed());
+  EXPECT_EQ(it.begin_calls, 1);
+  EXPECT_EQ(it.end_calls, 1);
+}
+
+TEST(ComponentIterator, StepBudgetIsHonored) {
+  RefusingIterator it;
+  it.begin();
+  it.try_advance(1);
+  EXPECT_EQ(it.begin_calls, 1);
+  EXPECT_EQ(it.end_calls, 0);
+  EXPECT_FALSE(it.completed());
+}
+
+TEST(ComponentIterator, RefusedStepStopsBatchAndRetriesSameStep) {
+  RefusingIterator it;
+  it.end_refusals = 3;
+  it.begin();
+  // First call runs until the refused end step, which stops the pass
+  it.try_advance(BIG_BUDGET);
+  EXPECT_EQ(it.end_calls, 1);
+  EXPECT_FALSE(it.completed());
+  // The refused step is retried once per call, not skipped
+  it.try_advance(BIG_BUDGET);
+  it.try_advance(BIG_BUDGET);
+  EXPECT_EQ(it.end_calls, 3);
+  EXPECT_FALSE(it.completed());
+  // Once accepted, the iteration completes
+  it.try_advance(BIG_BUDGET);
+  EXPECT_TRUE(it.completed());
+  EXPECT_EQ(it.end_calls, 4);
+}
+
+TEST(ComponentIterator, RefusedBeginStopsBatchAndRetries) {
+  RefusingIterator it;
+  it.begin_refusals = 2;
+  it.begin();
+  it.try_advance(BIG_BUDGET);
+  it.try_advance(BIG_BUDGET);
+  EXPECT_EQ(it.begin_calls, 2);
+  EXPECT_FALSE(it.completed());
+  it.try_advance(BIG_BUDGET);
+  EXPECT_TRUE(it.completed());
+  EXPECT_EQ(it.begin_calls, 3);
+}
+
+// The deprecated advance() wrapper must keep the legacy once-per-loop
+// pattern working during the deprecation window.
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wdeprecated-declarations"
+TEST(ComponentIterator, DeprecatedAdvanceKeepsLegacyPatternWorking) {
+  RefusingIterator it;
+  it.end_refusals = 2;
+  it.begin();
+  size_t guard = 0;
+  while (!it.completed() && guard++ < BIG_BUDGET) {
+    it.advance();
+  }
+  EXPECT_TRUE(it.completed());
+  // Two refused end steps were retried, then accepted
+  EXPECT_EQ(it.end_calls, 3);
+}
+#pragma GCC diagnostic pop
+
+#ifdef USE_SENSOR
+// Iterator whose sensor callback can refuse or yield; pins the per-item
+// contract: a refused item is re-offered with at_ unchanged, never skipped.
+class ItemRefusingIterator : public RefusingIterator {
+ public:
+  bool on_sensor(sensor::Sensor *obj) override {
+    this->last_sensor = obj;
+    if (!step(this->sensor_calls, this->sensor_refusals))
+      return false;
+    if (this->yield_on_sensor)
+      this->yield_after_step_();
+    return true;
+  }
+  sensor::Sensor *last_sensor{nullptr};
+  int sensor_calls{0};
+  int sensor_refusals{0};
+  bool yield_on_sensor{false};
+};
+
+class ComponentIteratorSensorTest : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    static sensor::Sensor sensor_a;
+    static sensor::Sensor sensor_b;
+    static bool registered = false;
+    if (!registered) {
+      App.register_sensor(&sensor_a);
+      App.register_sensor(&sensor_b);
+      registered = true;
+    }
+    // StaticVector drops silently when full; fail the fixture, not the contract
+    ASSERT_EQ(App.get_sensors().size(), 2u) << "benchmark.yaml sensor count too small";
+  }
+};
+
+TEST_F(ComponentIteratorSensorTest, RefusedItemIsReofferedNotSkipped) {
+  ItemRefusingIterator it;
+  it.sensor_refusals = 2;
+  it.begin();
+  // Runs until the first sensor refuses
+  it.try_advance(BIG_BUDGET);
+  EXPECT_EQ(it.sensor_calls, 1);
+  EXPECT_FALSE(it.completed());
+  // The refused item is re-offered, not skipped
+  it.try_advance(BIG_BUDGET);
+  EXPECT_EQ(it.sensor_calls, 2);
+  sensor::Sensor *refused = it.last_sensor;
+  // Once accepted, iteration continues through the second sensor to the end
+  it.try_advance(BIG_BUDGET);
+  EXPECT_TRUE(it.completed());
+  EXPECT_NE(it.last_sensor, refused);
+  EXPECT_EQ(it.sensor_calls, 4);
+}
+
+TEST_F(ComponentIteratorSensorTest, YieldAfterStepEndsPassAndResumes) {
+  ItemRefusingIterator it;
+  it.yield_on_sensor = true;
+  it.begin();
+  // The pass ends right after the first sensor despite a big budget
+  it.try_advance(BIG_BUDGET);
+  EXPECT_EQ(it.sensor_calls, 1);
+  EXPECT_FALSE(it.completed());
+  // The next pass ends after the second sensor
+  it.try_advance(BIG_BUDGET);
+  EXPECT_EQ(it.sensor_calls, 2);
+  // Remaining states then run to completion in one pass
+  it.try_advance(BIG_BUDGET);
+  EXPECT_TRUE(it.completed());
+}
+#endif  // USE_SENSOR
+
+}  // namespace esphome::testing

tests/integration/README.md

@@ -7,6 +7,7 @@ This directory contains end-to-end integration tests for ESPHome, focusing on te
 - `conftest.py` - Common fixtures and utilities
 - `const.py` - Constants used throughout the integration tests
 - `types.py` - Type definitions for fixtures and functions
+- `raw_api_client.py` - Minimal plaintext api client whose reads happen only on request (for backpressure tests)
 - `state_utils.py` - State handling utilities (e.g., `InitialStateHelper`, `find_entity`, `require_entity`)
 - `fixtures/` - YAML configuration files for tests
 - `test_*.py` - Individual test files
@@ -347,6 +348,7 @@ Create C++ components in `fixtures/external_components/` for:
 - Custom entity behaviors
 - Scheduler testing
 - Memory management tests
+- Deterministic network backpressure (`sndbuf_pin_component` pins socket send buffers; assert on its log line to prove the pin took effect)
 
 ##### Log Line Monitoring
 ```python

tests/integration/fixtures/api_list_entities_backpressure.yaml

@@ -0,0 +1,23 @@
+esphome:
+  name: api-backpressure-test
+
+host:
+
+api:
+  # Smallest queue so a non-draining client blocks the send path quickly
+  max_send_queue: 1
+  actions:
+# GENERATED_ACTIONS
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+    components: [sndbuf_pin_component]
+
+# Pins the device's socket send buffers for deterministic TCP backpressure
+sndbuf_pin_component:
+  buffer_size: SERVER_SNDBUF
+
+logger:
+  level: DEBUG

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

@@ -0,0 +1,20 @@
+import esphome.codegen as cg
+import esphome.config_validation as cv
+from esphome.const import CONF_BUFFER_SIZE, CONF_ID
+
+DEPENDENCIES = ["api"]
+
+sndbuf_pin_ns = cg.esphome_ns.namespace("sndbuf_pin")
+SndbufPinComponent = sndbuf_pin_ns.class_("SndbufPinComponent", cg.Component)
+
+CONFIG_SCHEMA = cv.Schema(
+    {
+        cv.GenerateID(): cv.declare_id(SndbufPinComponent),
+        cv.Required(CONF_BUFFER_SIZE): cv.int_range(min=1),
+    }
+).extend(cv.COMPONENT_SCHEMA)
+
+
+async def to_code(config):
+    var = cg.new_Pvariable(config[CONF_ID], config[CONF_BUFFER_SIZE])
+    await cg.register_component(var, config)

tests/integration/fixtures/external_components/sndbuf_pin_component/sndbuf_pin_component.cpp

@@ -0,0 +1,55 @@
+#include "sndbuf_pin_component.h"
+
+#include <netinet/in.h>
+#include <sys/socket.h>
+#include <cerrno>
+
+#include "esphome/components/api/api_server.h"
+#include "esphome/core/log.h"
+
+namespace esphome::sndbuf_pin {
+
+static const char *const TAG = "sndbuf_pin";
+
+// Skip stdio; scan the low fd range where the listeners land
+static constexpr int FIRST_USER_FD = 3;
+static constexpr int MAX_FD_SCAN = 128;
+
+void SndbufPinComponent::setup() {
+  int pinned = 0;
+  for (int fd = FIRST_USER_FD; fd < MAX_FD_SCAN; fd++) {
+    int type = 0;
+    socklen_t len = sizeof(type);
+    if (::getsockopt(fd, SOL_SOCKET, SO_TYPE, &type, &len) != 0 || type != SOCK_STREAM)
+      continue;
+    struct sockaddr_in addr {};
+    socklen_t addr_len = sizeof(addr);
+    if (::getsockname(fd, reinterpret_cast<struct sockaddr *>(&addr), &addr_len) != 0) {
+      ESP_LOGW(TAG, "fd %d: getsockname failed, errno %d", fd, errno);
+      continue;
+    }
+    if (ntohs(addr.sin_port) != api::global_api_server->get_port())
+      continue;
+    if (::setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &this->buffer_size_, sizeof(this->buffer_size_)) != 0) {
+      ESP_LOGW(TAG, "fd %d: SO_SNDBUF pin failed, errno %d", fd, errno);
+      continue;
+    }
+    int applied = 0;
+    len = sizeof(applied);
+    if (::getsockopt(fd, SOL_SOCKET, SO_SNDBUF, &applied, &len) != 0 || applied < this->buffer_size_) {
+      // Linux doubles the requested value; anything below it means clamped
+      ESP_LOGW(TAG, "fd %d: SO_SNDBUF readback %d below requested %d", fd, applied, this->buffer_size_);
+      continue;
+    }
+    // Tests assert on this line; accepted sockets inherit the pinned size
+    ESP_LOGD(TAG, "fd %d port %d: SO_SNDBUF pinned to %d (effective %d)", fd, ntohs(addr.sin_port), this->buffer_size_,
+             applied);
+    pinned++;
+  }
+  if (pinned == 0) {
+    ESP_LOGE(TAG, "api listener socket was not pinned");
+    this->mark_failed();
+  }
+}
+
+}  // namespace esphome::sndbuf_pin

tests/integration/fixtures/external_components/sndbuf_pin_component/sndbuf_pin_component.h

@@ -0,0 +1,21 @@
+#pragma once
+
+#include "esphome/core/component.h"
+
+namespace esphome::sndbuf_pin {
+
+// Test-only (host): pins SO_SNDBUF on every open TCP socket so integration
+// tests get deterministic backpressure; an explicit SO_SNDBUF also disables
+// kernel autotuning, and accepted sockets inherit it from the listener.
+class SndbufPinComponent : public Component {
+ public:
+  explicit SndbufPinComponent(int buffer_size) : buffer_size_(buffer_size) {}
+  void setup() override;
+  // After the api server so its listening socket exists
+  float get_setup_priority() const override { return setup_priority::LATE; }
+
+ protected:
+  int buffer_size_;
+};
+
+}  // namespace esphome::sndbuf_pin

tests/integration/raw_api_client.py

@@ -0,0 +1,148 @@
+"""Minimal plaintext native-api client over a raw socket.
+
+Reads only when told to, so tests control when the TCP pipe backs up toward
+the device; payloads are skipped and only message types are counted.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from collections import Counter
+import socket
+from typing import Self
+
+from aioesphomeapi import api_pb2
+import aioesphomeapi.core as api_core
+from google.protobuf import message
+
+from .const import LOCALHOST
+
+# Message type ids are protocol constants; derive them from aioesphomeapi so
+# they cannot drift from the client library in use.
+MESSAGE_TYPE_OF = {cls: num for num, cls in api_core.MESSAGE_TYPE_TO_PROTO.items()}
+
+_READ_CHUNK = 4096
+
+
+def encode_varint(value: int) -> bytes:
+    out = bytearray()
+    while True:
+        byte = value & 0x7F
+        value >>= 7
+        if value:
+            out.append(byte | 0x80)
+        else:
+            out.append(byte)
+            return bytes(out)
+
+
+def decode_varint(buf: bytearray, pos: int) -> tuple[int, int] | None:
+    """Decode one varint at pos; return (value, new_pos) or None if short."""
+    value = shift = 0
+    while pos < len(buf):
+        byte = buf[pos]
+        pos += 1
+        value |= (byte & 0x7F) << shift
+        if not byte & 0x80:
+            return value, pos
+        shift += 7
+    return None
+
+
+def encode_frame(msg_type: int, payload: bytes) -> bytes:
+    """Encode one plaintext api frame: 0x00, payload length, message type."""
+    return b"\x00" + encode_varint(len(payload)) + encode_varint(msg_type) + payload
+
+
+class FrameParser:
+    """Incremental parser for the plaintext api frame stream."""
+
+    def __init__(self) -> None:
+        self._buf = bytearray()
+
+    def feed(self, data: bytes) -> list[int]:
+        self._buf.extend(data)
+        types: list[int] = []
+        while (msg_type := self._try_parse()) is not None:
+            types.append(msg_type)
+        return types
+
+    def _try_parse(self) -> int | None:
+        buf = self._buf
+        if not buf:
+            return None
+        assert buf[0] == 0, f"expected plaintext frame, got indicator {buf[0]}"
+        if (size_decoded := decode_varint(buf, 1)) is None:
+            return None
+        size, pos = size_decoded
+        if (type_decoded := decode_varint(buf, pos)) is None:
+            return None
+        msg_type, pos = type_decoded
+        if len(buf) - pos < size:
+            return None
+        del buf[: pos + size]
+        return msg_type
+
+
+class RawApiClient:
+    """Plaintext api client whose reads happen only on request."""
+
+    def __init__(self, port: int, recv_buffer_size: int | None = None) -> None:
+        self._port = port
+        self._parser = FrameParser()
+        self.bytes_received = 0
+        self.frame_counts: Counter[int] = Counter()
+        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        try:
+            if recv_buffer_size is not None:
+                sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, recv_buffer_size)
+                # Kernels may round up (Linux doubles) but must not clamp below
+                applied = sock.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF)
+                assert applied >= recv_buffer_size, (
+                    f"SO_RCVBUF clamped to {applied}, requested {recv_buffer_size}"
+                )
+            sock.setblocking(False)
+        except Exception:
+            sock.close()
+            raise
+        self._sock = sock
+
+    async def __aenter__(self) -> Self:
+        return self
+
+    async def __aexit__(self, *exc_info: object) -> None:
+        self.close()
+
+    async def connect(self, client_info: str = "raw-api-client") -> None:
+        """Connect and complete the Hello handshake (no auth step since 2026.1.0)."""
+        loop = asyncio.get_running_loop()
+        await loop.sock_connect(self._sock, (LOCALHOST, self._port))
+        hello = api_pb2.HelloRequest()
+        hello.client_info = client_info
+        hello.api_version_major = 1
+        hello.api_version_minor = 10
+        await self.send_message(hello)
+        await self.read_until_frame(MESSAGE_TYPE_OF[api_pb2.HelloResponse])
+
+    async def send_message(self, msg: message.Message) -> None:
+        loop = asyncio.get_running_loop()
+        await loop.sock_sendall(
+            self._sock,
+            encode_frame(MESSAGE_TYPE_OF[type(msg)], msg.SerializeToString()),
+        )
+
+    async def read_until_frame(self, msg_type: int, timeout: float = 10.0) -> None:
+        """Read until at least one frame of msg_type has been received."""
+        loop = asyncio.get_running_loop()
+
+        async def _read_loop() -> None:
+            while not self.frame_counts[msg_type]:
+                data = await loop.sock_recv(self._sock, _READ_CHUNK)
+                assert data, "server closed the connection unexpectedly"
+                self.bytes_received += len(data)
+                self.frame_counts.update(self._parser.feed(data))
+
+        await asyncio.wait_for(_read_loop(), timeout)
+
+    def close(self) -> None:
+        self._sock.close()

tests/integration/test_api_list_entities_backpressure.py

@@ -0,0 +1,110 @@
+"""A client that stops reading the entity listing must not starve other clients.
+
+Service responses are sent directly (not via the deferred batch), so a full
+TCP pipe makes the send path refuse; the drive loop now lives in
+try_advance(), which stops on refusal instead of retrying forever. Not a
+before/after regression test: pre-fix builds survive here because the
+refusal path yields and pumps the socket each retry.
+
+The sndbuf_pin_component fixture pins the device's send buffers so the pipe
+fills deterministically regardless of kernel autotuning; the test waits for
+its log line before proceeding.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+from aioesphomeapi import api_pb2
+import pytest
+
+from .raw_api_client import MESSAGE_TYPE_OF, RawApiClient
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+SERVICES_RESPONSE = MESSAGE_TYPE_OF[api_pb2.ListEntitiesServicesResponse]
+LIST_DONE_RESPONSE = MESSAGE_TYPE_OF[api_pb2.ListEntitiesDoneResponse]
+
+# Both ends of the pipe are pinned small; only tens of KB fit in the kernel
+RECV_BUFFER_SIZE = 4096
+SERVER_SNDBUF = 8192  # substituted into the fixture yaml
+# Logged by the sndbuf_pin_component fixture when it pins a socket
+SNDBUF_PIN_LOG = "SO_SNDBUF pinned to"
+# One response (~6.4 KB) must stay smaller than the pinned send buffer; an
+# oversized message parks in the overflow buffer and reports as sent.
+ARGS_PER_SERVICE = 8
+ARG_NAME_LEN = 800
+# ~160 KB listing versus a tens-of-KB pipe guarantees a mid-services block
+NUM_SERVICES = 25
+assert ARGS_PER_SERVICE * ARG_NAME_LEN < SERVER_SNDBUF
+# The pipe fills in well under a second
+STALL_SECONDS = 0.5
+# Well above pipe capacity, well below the listing size
+MIN_DRAINED_BYTES = 60_000
+
+
+def _generated_actions() -> str:
+    """Build the api actions block: services with long argument names."""
+    lines: list[str] = []
+    for i in range(NUM_SERVICES):
+        lines.append(f"    - action: backpressure_service_{i:04d}")
+        lines.append("      variables:")
+        for j in range(ARGS_PER_SERVICE):
+            prefix = f"arg_{i:04d}_{j:02d}_"
+            lines.append(
+                f"        {prefix}{'x' * (ARG_NAME_LEN - len(prefix))}: string"
+            )
+        lines.append("      then:")
+        lines.append("        - logger.log: service called")
+    return "\n".join(lines)
+
+
+@pytest.mark.asyncio
+async def test_api_list_entities_backpressure(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+    unused_tcp_port: int,
+) -> None:
+    """A stalled reader mid-services must not block other api clients."""
+    assert "# GENERATED_ACTIONS" in yaml_config
+    config = yaml_config.replace("# GENERATED_ACTIONS", _generated_actions())
+    config = config.replace("SERVER_SNDBUF", str(SERVER_SNDBUF))
+
+    pin_applied = asyncio.Event()
+
+    def _on_log_line(line: str) -> None:
+        if SNDBUF_PIN_LOG in line:
+            pin_applied.set()
+
+    async with run_compiled(config, line_callback=_on_log_line):
+        # Fails loudly if the pin never applied
+        await asyncio.wait_for(pin_applied.wait(), 10)
+
+        async with RawApiClient(
+            unused_tcp_port, recv_buffer_size=RECV_BUFFER_SIZE
+        ) as stalled:
+            await stalled.connect(client_info="backpressure-stall-client")
+            await stalled.send_message(api_pb2.ListEntitiesRequest())
+            # The client now stops reading entirely.
+
+            # Let the server run against the full pipe
+            await asyncio.sleep(STALL_SECONDS)
+
+            # Other clients must still be served while the first is blocked
+            async with api_client_connected(timeout=20) as client:
+                device_info = await asyncio.wait_for(client.device_info(), 20)
+                assert device_info.name == "api-backpressure-test"
+                _, services = await asyncio.wait_for(
+                    client.list_entities_services(), 30
+                )
+                assert len(services) == NUM_SERVICES
+
+            # Fixture-size guard: the listing must dwarf the pinned pipe
+            before = stalled.bytes_received
+            await stalled.read_until_frame(LIST_DONE_RESPONSE, timeout=60)
+            drained = stalled.bytes_received - before
+            assert drained > MIN_DRAINED_BYTES, (
+                f"only {drained} bytes drained; the listing never backed up"
+            )
+            assert stalled.frame_counts[SERVICES_RESPONSE] == NUM_SERVICES
+            assert stalled.frame_counts[LIST_DONE_RESPONSE] == 1