[api] Fix loop stall that triggers the task watchdog when entity list sends block (#18577)

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