Clear the scan-end skip on every promotion

Co-authored-by: pre-commit-ci-lite[bot] <117423508+pre-commit-ci-lite[bot]@users.noreply.github.com>
Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

docker/Dockerfile

@@ -22,7 +22,7 @@ RUN \
     -r /requirements.txt
 
 # Install the ESPHome Device Builder dashboard.
-RUN uv pip install --no-cache-dir esphome-device-builder==1.12.2
+RUN uv pip install --no-cache-dir esphome-device-builder==1.12.3
 
 RUN \
     platformio settings set enable_telemetry No \

esphome/components/ags10/sensor.py

@@ -17,6 +17,9 @@ from esphome.const import (
     UNIT_OHM,
     UNIT_PARTS_PER_BILLION,
 )
+from esphome.core import ID
+from esphome.cpp_generator import MockObj, TemplateArgsType
+from esphome.types import ConfigType
 
 CONF_RESISTANCE = "resistance"
 
@@ -62,7 +65,7 @@ CONFIG_SCHEMA = (
 FINAL_VALIDATE_SCHEMA = i2c.final_validate_device_schema("ags10", max_frequency="15khz")
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     var = cg.new_Pvariable(config[CONF_ID])
     await cg.register_component(var, config)
     await i2c.register_i2c_device(var, config)
@@ -94,7 +97,12 @@ AGS10_NEW_I2C_ADDRESS_SCHEMA = cv.maybe_simple_value(
     AGS10_NEW_I2C_ADDRESS_SCHEMA,
     synchronous=True,
 )
-async def ags10newi2caddress_to_code(config, action_id, template_arg, args):
+async def ags10newi2caddress_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     var = cg.new_Pvariable(action_id, template_arg)
     await cg.register_parented(var, config[CONF_ID])
     address = await cg.templatable(config[CONF_ADDRESS], args, cg.uint8)
@@ -126,7 +134,12 @@ AGS10_SET_ZERO_POINT_SCHEMA = cv.Schema(
     AGS10_SET_ZERO_POINT_SCHEMA,
     synchronous=True,
 )
-async def ags10setzeropoint_to_code(config, action_id, template_arg, args):
+async def ags10setzeropoint_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     var = cg.new_Pvariable(action_id, template_arg)
     await cg.register_parented(var, config[CONF_ID])
     mode = await cg.templatable(

esphome/components/api/api_connection.cpp

@@ -412,15 +412,15 @@ void APIConnection::finalize_iterator_sync_() {
 }
 
 void APIConnection::process_iterator_batch_(ComponentIterator &iterator) {
-  // 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);
+  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();
+  }
 
-  // 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) {
+  // 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) {
     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;
-// Deferred batch size cap during initial state/info sync
-static constexpr size_t MAX_INITIAL_BATCH_SIZE = 34;
+// Maximum number of entities to process in a single batch during initial state/info sending
+static constexpr size_t MAX_INITIAL_PER_BATCH = 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_BATCH_SIZE,
-              "MAX_MESSAGES_PER_BATCH must be >= MAX_INITIAL_BATCH_SIZE");
+static_assert(MAX_MESSAGES_PER_BATCH >= MAX_INITIAL_PER_BATCH,
+              "MAX_MESSAGES_PER_BATCH must be >= MAX_INITIAL_PER_BATCH");
 
 #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_BATCH_SIZE in api_connection.h (enforced by static_assert there)
+// Must be >= MAX_INITIAL_PER_BATCH 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,17 +95,9 @@ 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();
-  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;
+  return this->client_->send_message(resp);
 }
 #endif
 

esphome/components/at581x/__init__.py

@@ -4,6 +4,9 @@ import esphome.codegen as cg
 from esphome.components import i2c
 import esphome.config_validation as cv
 from esphome.const import CONF_FREQUENCY, CONF_ID
+from esphome.core import ID
+from esphome.cpp_generator import MockObj, TemplateArgsType
+from esphome.types import ConfigType
 
 CODEOWNERS = ["@X-Ryl669"]
 DEPENDENCIES = ["i2c"]
@@ -70,7 +73,7 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     var = cg.new_Pvariable(config[CONF_ID])
     await cg.register_component(var, config)
     await i2c.register_i2c_device(var, config)
@@ -91,7 +94,12 @@ AT581XSettingsAction = at581x_ns.class_("AT581XSettingsAction", automation.Actio
     ),
     synchronous=True,
 )
-async def at581x_reset_to_code(config, action_id, template_arg, args):
+async def at581x_reset_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     var = cg.new_Pvariable(action_id, template_arg)
     await cg.register_parented(var, config[CONF_ID])
 
@@ -163,7 +171,12 @@ RADAR_SETTINGS_SCHEMA = cv.Schema(
     RADAR_SETTINGS_SCHEMA,
     synchronous=True,
 )
-async def at581x_settings_to_code(config, action_id, template_arg, args):
+async def at581x_settings_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     var = cg.new_Pvariable(action_id, template_arg)
     await cg.register_parented(var, config[CONF_ID])
 

esphome/components/at581x/switch/__init__.py

@@ -2,6 +2,7 @@ import esphome.codegen as cg
 from esphome.components import switch
 import esphome.config_validation as cv
 from esphome.const import DEVICE_CLASS_SWITCH, ICON_WIFI
+from esphome.types import ConfigType
 
 from .. import CONF_AT581X_ID, AT581XComponent, at581x_ns
 
@@ -22,7 +23,7 @@ CONFIG_SCHEMA = switch.switch_schema(
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     at581x_component = await cg.get_variable(config[CONF_AT581X_ID])
     s = await switch.new_switch(config)
     await cg.register_parented(s, config[CONF_AT581X_ID])

esphome/components/ble_device_base/__init__.py

@@ -206,32 +206,36 @@ def validate_scan_parameters(config: ConfigType) -> ConfigType:
     interval = config[CONF_INTERVAL]
     window = config[CONF_WINDOW]
 
-    if window > interval:
-        raise cv.Invalid(
-            f"Scan window ({window}) needs to be smaller than scan interval ({interval})"
-        )
+    # Labels are reused in every error below; the optional one names its key.
+    windows = [("Scan window", window)]
+    if (connection_window := config.get(CONF_CONNECTION_SCAN_WINDOW)) is not None:
+        windows.append((CONF_CONNECTION_SCAN_WINDOW, connection_window))
+
+    for name, value in windows:
+        if value > interval:
+            raise cv.Invalid(
+                f"{name} ({value}) needs to be smaller than scan interval ({interval})"
+            )
 
     # BLE scan interval/window are programmed in 0.625 ms units as a 16-bit value; the
     # controller only accepts 2.5 ms .. 10240 ms (0x0004 .. 0x4000). Reject out-of-range
     # values here instead of letting the unit conversion silently overflow.
-    for name, value in (("interval", interval), ("window", window)):
+    for name, value in (("Scan interval", interval), *windows):
         if value.total_microseconds < 2500 or value.total_microseconds > 10_240_000:
-            raise cv.Invalid(
-                f"Scan {name} ({value}) must be between 2.5 ms and 10240 ms"
-            )
+            raise cv.Invalid(f"{name} ({value}) must be between 2.5 ms and 10240 ms")
 
     # Validate what actually reaches the controller: both values are truncated to
     # whole 0.625 ms units, so a window/interval pair that differs by less than one
     # unit collapses to the same value — silently programming a 100 % duty cycle
     # (radio permanently on) from a config that asked for less.
     interval_units = to_ble_units(interval)
-    window_units = to_ble_units(window)
-    if window_units == interval_units and window < interval:
-        raise cv.Invalid(
-            f"Scan window ({window}) and interval ({interval}) both truncate to "
-            f"{interval_units} x 0.625 ms, which the controller scans at a 100 % duty "
-            f"cycle. Separate them by at least 0.625 ms."
-        )
+    for name, value in windows:
+        if to_ble_units(value) == interval_units and value < interval:
+            raise cv.Invalid(
+                f"{name} ({value}) and interval ({interval}) both truncate to "
+                f"{interval_units} x 0.625 ms, which the controller scans at a 100 % duty "
+                f"cycle. Separate them by at least 0.625 ms."
+            )
 
     if interval.total_microseconds * 3 > duration.total_microseconds:
         raise cv.Invalid(
@@ -247,11 +251,14 @@ def validate_scan_parameters(config: ConfigType) -> ConfigType:
 # their own; also the fallback for esp32's conditional default.
 DEFAULT_SCAN_WINDOW = "30ms"
 
+CONF_CONNECTION_SCAN_WINDOW = "connection_scan_window"
+
 
 def scan_parameters_schema(
     interval_default: str,
     *,
     window_default: str | Callable[[], TimePeriod] = DEFAULT_SCAN_WINDOW,
+    connection_window: bool = False,
 ) -> cv.All:
     """Build the scan_parameters value schema shared by all BLE trackers.
 
@@ -263,7 +270,9 @@ def scan_parameters_schema(
     can adjust it once sibling keys are resolved). The `active` option
     (default on) is unconditional: active scanning is part of the tracker
     contract — every current proxy client assumes it, so a passive-only
-    tracker must not share this schema.
+    tracker must not share this schema. connection_window opts in to the
+    `connection_scan_window` option for trackers that can fall back to a
+    smaller window while a GATT connection is active.
     """
     schema = {
         cv.Optional(CONF_DURATION, default="5min"): cv.positive_time_period_seconds,
@@ -272,6 +281,8 @@ def scan_parameters_schema(
         cv.Optional(CONF_CONTINUOUS, default=True): cv.boolean,
         cv.Optional(CONF_ACTIVE, default=True): cv.boolean,
     }
+    if connection_window:
+        schema[cv.Optional(CONF_CONNECTION_SCAN_WINDOW)] = cv.positive_time_period
     return cv.All(cv.Schema(schema), validate_scan_parameters)
 
 

esphome/components/canbus/__init__.py

@@ -1,10 +1,13 @@
 import re
+from typing import Any
 
 from esphome import automation
 import esphome.codegen as cg
 import esphome.config_validation as cv
 from esphome.const import CONF_DATA, CONF_ID, CONF_TRIGGER_ID
 from esphome.core import CORE, ID
+from esphome.cpp_generator import MockObj, TemplateArgsType
+from esphome.types import ConfigType
 
 CODEOWNERS = ["@mvturnho", "@danielschramm"]
 IS_PLATFORM_COMPONENT = True
@@ -18,7 +21,7 @@ CONF_BIT_RATE = "bit_rate"
 CONF_ON_FRAME = "on_frame"
 
 
-def validate_id(config):
+def validate_id(config: ConfigType) -> ConfigType:
     if CONF_CAN_ID in config:
         can_id = config[CONF_CAN_ID]
         id_ext = config[CONF_USE_EXTENDED_ID]
@@ -27,7 +30,7 @@ def validate_id(config):
     return config
 
 
-def validate_raw_data(value):
+def validate_raw_data(value: Any) -> bytes | list:
     if isinstance(value, str):
         return value.encode("utf-8")
     if isinstance(value, list):
@@ -71,7 +74,7 @@ CAN_SPEEDS = {
 }
 
 
-def get_rate(value):
+def get_rate(value: str) -> int:
     match = re.match(r"(\d+)(?:K(\d+)?)?BPS", value, re.IGNORECASE)
     if not match:
         raise ValueError(f"Invalid rate format: {value}")
@@ -103,7 +106,7 @@ CANBUS_SCHEMA = cv.Schema(
 CANBUS_SCHEMA.add_extra(validate_id)
 
 
-async def setup_canbus_core_(var, config):
+async def setup_canbus_core_(var: MockObj, config: ConfigType) -> None:
     await cg.register_component(var, config)
     cg.add(var.set_can_id([config[CONF_CAN_ID]]))
     cg.add(var.set_use_extended_id([config[CONF_USE_EXTENDED_ID]]))
@@ -134,7 +137,7 @@ async def setup_canbus_core_(var, config):
         )
 
 
-async def register_canbus(var, config):
+async def register_canbus(var: MockObj, config: ConfigType) -> None:
     if not CORE.has_id(config[CONF_ID]):
         var = cg.new_Pvariable(config[CONF_ID], var)
     await setup_canbus_core_(var, config)
@@ -157,7 +160,12 @@ async def register_canbus(var, config):
     ),
     synchronous=True,
 )
-async def canbus_action_to_code(config, action_id, template_arg, args):
+async def canbus_action_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     var = cg.new_Pvariable(action_id, template_arg)
     await cg.register_parented(var, config[CONF_CANBUS_ID])
 

esphome/components/daly_bms/__init__.py

@@ -2,6 +2,7 @@ import esphome.codegen as cg
 from esphome.components import uart
 import esphome.config_validation as cv
 from esphome.const import CONF_ADDRESS, CONF_ID
+from esphome.types import ConfigType
 
 CODEOWNERS = ["@s1lvi0"]
 MULTI_CONF = True
@@ -26,7 +27,7 @@ CONFIG_SCHEMA = (
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     var = cg.new_Pvariable(config[CONF_ID])
     await cg.register_component(var, config)
     await uart.register_uart_device(var, config)

esphome/components/daly_bms/binary_sensor.py

@@ -1,6 +1,8 @@
 import esphome.codegen as cg
 from esphome.components import binary_sensor
 import esphome.config_validation as cv
+from esphome.cpp_generator import MockObj
+from esphome.types import ConfigType
 
 from . import CONF_BMS_DALY_ID, DalyBmsComponent
 
@@ -27,13 +29,13 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-async def setup_conf(config, key, hub):
+async def setup_conf(config: ConfigType, key: str, hub: MockObj) -> None:
     if sensor_config := config.get(key):
         var = await binary_sensor.new_binary_sensor(sensor_config)
         cg.add(getattr(hub, f"set_{key}_binary_sensor")(var))
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     hub = await cg.get_variable(config[CONF_BMS_DALY_ID])
     for key in TYPES:
         await setup_conf(config, key, hub)

esphome/components/daly_bms/sensor.py

@@ -23,6 +23,8 @@ from esphome.const import (
     UNIT_PERCENT,
     UNIT_VOLT,
 )
+from esphome.cpp_generator import MockObj
+from esphome.types import ConfigType
 
 from . import CONF_BMS_DALY_ID, DalyBmsComponent
 
@@ -222,13 +224,13 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-async def setup_conf(config, key, hub):
+async def setup_conf(config: ConfigType, key: str, hub: MockObj) -> None:
     if sensor_config := config.get(key):
         sens = await sensor.new_sensor(sensor_config)
         cg.add(getattr(hub, f"set_{key}_sensor")(sens))
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     hub = await cg.get_variable(config[CONF_BMS_DALY_ID])
     for key in TYPES:
         await setup_conf(config, key, hub)

esphome/components/daly_bms/text_sensor.py

@@ -2,6 +2,8 @@ import esphome.codegen as cg
 from esphome.components import text_sensor
 import esphome.config_validation as cv
 from esphome.const import CONF_STATUS
+from esphome.cpp_generator import MockObj
+from esphome.types import ConfigType
 
 from . import CONF_BMS_DALY_ID, DalyBmsComponent
 
@@ -23,13 +25,13 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-async def setup_conf(config, key, hub):
+async def setup_conf(config: ConfigType, key: str, hub: MockObj) -> None:
     if sensor_config := config.get(key):
         sens = await text_sensor.new_text_sensor(sensor_config)
         cg.add(getattr(hub, f"set_{key}_text_sensor")(sens))
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     hub = await cg.get_variable(config[CONF_BMS_DALY_ID])
     for key in TYPES:
         await setup_conf(config, key, hub)

esphome/components/deep_sleep/__init__.py

@@ -38,7 +38,8 @@ from esphome.const import (
     PLATFORM_NRF52,
     PlatformFramework,
 )
-from esphome.core import CORE
+from esphome.core import CORE, ID
+from esphome.cpp_generator import MockObj, TemplateArgsType
 from esphome.types import ConfigType
 
 WAKEUP_PINS = {
@@ -174,7 +175,7 @@ def validate_config(config: ConfigType) -> ConfigType:
     return config
 
 
-def _validate_ex1_wakeup_mode(value):
+def _validate_ex1_wakeup_mode(value: str) -> str:
     if value == "ALL_LOW":
         esp32.only_on_variant(supported=[VARIANT_ESP32], msg_prefix="ALL_LOW")(value)
     if value == "ANY_LOW":
@@ -345,7 +346,7 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     var = cg.new_Pvariable(config[CONF_ID])
     await cg.register_component(var, config)
 
@@ -458,7 +459,12 @@ DEEP_SLEEP_ENTER_SCHEMA = cv.All(
     DEEP_SLEEP_ENTER_SCHEMA,
     synchronous=True,
 )
-async def deep_sleep_enter_to_code(config, action_id, template_arg, args):
+async def deep_sleep_enter_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     paren = await cg.get_variable(config[CONF_ID])
     var = cg.new_Pvariable(action_id, template_arg, paren)
     if CONF_SLEEP_DURATION in config:
@@ -487,7 +493,12 @@ async def deep_sleep_enter_to_code(config, action_id, template_arg, args):
     automation.maybe_simple_id(DEEP_SLEEP_ACTION_SCHEMA),
     synchronous=True,
 )
-async def deep_sleep_action_to_code(config, action_id, template_arg, args):
+async def deep_sleep_action_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     var = cg.new_Pvariable(action_id, template_arg)
     await cg.register_parented(var, config[CONF_ID])
     return var

esphome/components/ds1307/time.py

@@ -3,6 +3,9 @@ import esphome.codegen as cg
 from esphome.components import i2c, time
 import esphome.config_validation as cv
 from esphome.const import CONF_ID
+from esphome.core import ID
+from esphome.cpp_generator import MockObj, TemplateArgsType
+from esphome.types import ConfigType
 
 CODEOWNERS = ["@badbadc0ffee"]
 DEPENDENCIES = ["i2c"]
@@ -29,7 +32,12 @@ CONFIG_SCHEMA = time.TIME_SCHEMA.extend(
     ),
     synchronous=True,
 )
-async def ds1307_write_time_to_code(config, action_id, template_arg, args):
+async def ds1307_write_time_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     var = cg.new_Pvariable(action_id, template_arg)
     await cg.register_parented(var, config[CONF_ID])
     return var
@@ -45,13 +53,18 @@ async def ds1307_write_time_to_code(config, action_id, template_arg, args):
     ),
     synchronous=True,
 )
-async def ds1307_read_time_to_code(config, action_id, template_arg, args):
+async def ds1307_read_time_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     var = cg.new_Pvariable(action_id, template_arg)
     await cg.register_parented(var, config[CONF_ID])
     return var
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     var = cg.new_Pvariable(config[CONF_ID])
 
     await cg.register_component(var, config)

esphome/components/emontx/sensor/__init__.py

@@ -68,6 +68,7 @@ PATTERN_CONFIGS = {
     "PULSE": {
         CONF_UNIT_OF_MEASUREMENT: UNIT_PULSES,
         CONF_DEVICE_CLASS: DEVICE_CLASS_ENERGY,
+        CONF_STATE_CLASS: STATE_CLASS_TOTAL_INCREASING,
         CONF_ACCURACY_DECIMALS: 0,
     },
     "PF": {
@@ -78,12 +79,13 @@ PATTERN_CONFIGS = {
     },
 }
 
-# Create a base schema that's flexible for any tag
-BASE_SCHEMA = sensor.sensor_schema(
-    EmonTxSensor,
-    state_class=STATE_CLASS_MEASUREMENT,
-    accuracy_decimals=0,
-).extend(
+# BASE_SCHEMA intentionally omits state_class and accuracy_decimals defaults.
+# Passing them to sensor_schema() would register them via cv.Optional(key, default=...),
+# making them always present in the validated config dict and preventing
+# apply_tag_defaults from overriding them with the correct per-prefix values.
+# They are injected by apply_tag_defaults below, after running through
+# sensor.validate_state_class() so the value is code-generation-ready.
+BASE_SCHEMA = sensor.sensor_schema(EmonTxSensor).extend(
     {
         cv.GenerateID(CONF_EMONTX_ID): cv.use_id(EmonTx),
         cv.Required(CONF_TAG_NAME): cv.string,
@@ -91,34 +93,43 @@ BASE_SCHEMA = sensor.sensor_schema(
 )
 
 
+def _apply_defaults(config: ConfigType, defaults: dict) -> None:
+    """Inject defaults into config, skipping keys already set by the user.
+    state_class values are run through validate_state_class so they are
+    code-generation-ready, matching what sensor_schema() would normally do."""
+    for key, value in defaults.items():
+        if key not in config:
+            if key == CONF_STATE_CLASS:
+                value = sensor.validate_state_class(value)
+            config[key] = value
+
+
 def apply_tag_defaults(config: ConfigType) -> ConfigType:
     """Apply defaults based on tag prefix if applicable, but don't restrict any tags."""
     tag = config[CONF_TAG_NAME]
 
-    # Skip if tag is too short
-    if len(tag) < 2:
-        return config
+    if len(tag) >= 2:
+        tag_upper = tag.upper()
 
-    # Check if this tag starts with a known prefix
-    tag_upper = tag.upper()
+        for pattern, pattern_config in PATTERN_CONFIGS.items():
+            if tag_upper.startswith(pattern):
+                _apply_defaults(config, pattern_config)
+                return config
 
-    for pattern, pattern_config in PATTERN_CONFIGS.items():
-        if tag_upper.startswith(pattern):
-            # Apply pattern defaults if not overridden by user
-            for key, value in pattern_config.items():
-                if key not in config:
-                    config[key] = value
+        # Only apply defaults for known prefixes with numeric indices (e.g. E1, V2, T3)
+        prefix = tag_upper[0]
+        if prefix in SENSOR_CONFIGS and tag[1:].isdigit():
+            _apply_defaults(config, SENSOR_CONFIGS[prefix])
             return config
 
-    # Only apply defaults for known prefixes with numeric indices
-    prefix = tag_upper[0]
-    if prefix in SENSOR_CONFIGS and len(tag) > 1 and tag[1:].isdigit():
-        # Apply defaults for known tag types, but only if not overridden by user
-        defaults = SENSOR_CONFIGS[prefix]
-        for key, value in defaults.items():
-            if key not in config:
-                config[key] = value
-
+    # Fall back to generic defaults for tags with no known prefix
+    _apply_defaults(
+        config,
+        {
+            CONF_STATE_CLASS: STATE_CLASS_MEASUREMENT,
+            CONF_ACCURACY_DECIMALS: 0,
+        },
+    )
     return config
 
 

esphome/components/esp32_ble_tracker/__init__.py

@@ -7,6 +7,7 @@ import logging
 from esphome import automation
 import esphome.codegen as cg
 from esphome.components import ble_device_base, esp32_ble, ota
+from esphome.components.ble_device_base import CONF_CONNECTION_SCAN_WINDOW
 from esphome.components.const import CONF_ON_SCAN_END, CONF_SCAN_PARAMETERS, CONF_WINDOW
 from esphome.components.esp32 import (
     add_idf_sdkconfig_option,
@@ -38,7 +39,8 @@ from esphome.const import (
     CONF_SERVICE_UUID,
     CONF_TRIGGER_ID,
 )
-from esphome.core import CORE, CoroPriority, TimePeriod, coroutine_with_priority
+from esphome.core import CORE, ID, CoroPriority, TimePeriod, coroutine_with_priority
+from esphome.cpp_generator import MockObj, TemplateArgsType
 from esphome.enum import StrEnum
 from esphome.types import ConfigType
 
@@ -72,8 +74,9 @@ def _get_required_features() -> set[BLEFeatures]:
 
 # Slot counters sizing the tracker's StaticVector storage; one request per
 # registered listener or client.
+CLIENT_COUNT_DEFINE = "ESPHOME_ESP32_BLE_TRACKER_CLIENT_COUNT"
 _request_listener_slot = cg.slot_counter("ESPHOME_ESP32_BLE_TRACKER_LISTENER_COUNT")
-_request_client_slot = cg.slot_counter("ESPHOME_ESP32_BLE_TRACKER_CLIENT_COUNT")
+_request_client_slot = cg.slot_counter(CLIENT_COUNT_DEFINE)
 
 
 def register_ble_features(features: set[BLEFeatures]) -> None:
@@ -146,6 +149,7 @@ class TrackerData:
     """Per-run validation state, namespaced under DOMAIN in CORE.data."""
 
     scan_window_defaulted: bool = False
+    connection_window_injected: bool = False
 
 
 def _get_data() -> TrackerData:
@@ -174,17 +178,34 @@ def _raise_defaulted_scan_window(config: ConfigType) -> ConfigType:
     honors the window strictly (>= 5.5.5); without the arbiter a full-duty
     scan would starve wifi outright, and a user-set window is never touched.
     Raising to the interval cannot invalidate the already-validated
-    parameters, so no re-validation is needed.
+    parameters, so no re-validation is needed. The connection window is
+    checked against the window here, after the raise.
     """
+    params = config[CONF_SCAN_PARAMETERS]
     if (
         _get_data().scan_window_defaulted
         and config.get(CONF_SOFTWARE_COEXISTENCE)
         and idf_version() >= IDF_SCAN_WINDOW_FIX_VERSION
     ):
-        params = config[CONF_SCAN_PARAMETERS]
         # Copy so the config dump shows a plain value instead of a YAML
         # anchor/alias pair pointing at the interval.
         params[CONF_WINDOW] = copy.copy(params[CONF_INTERVAL])
+        # Arm the connection-time fallback unless the user set one. Injected
+        # after validation; safe because it equals the validated window default.
+        if CONF_CONNECTION_SCAN_WINDOW not in params:
+            params[CONF_CONNECTION_SCAN_WINDOW] = cv.positive_time_period(
+                ble_device_base.DEFAULT_SCAN_WINDOW
+            )
+            _get_data().connection_window_injected = True
+    if (
+        connection_window := params.get(CONF_CONNECTION_SCAN_WINDOW)
+    ) is not None and connection_window > params[CONF_WINDOW]:
+        # A larger value would widen the scan during connections.
+        raise cv.Invalid(
+            f"{CONF_CONNECTION_SCAN_WINDOW} ({connection_window}) needs to be "
+            f"smaller than the scan window ({params[CONF_WINDOW]})",
+            path=[CONF_SCAN_PARAMETERS, CONF_CONNECTION_SCAN_WINDOW],
+        )
     return config
 
 
@@ -193,7 +214,7 @@ def _raise_defaulted_scan_window(config: ConfigType) -> ConfigType:
 # window/interval pairs that collapse to the same 0.625 ms unit count.
 # The window default is conditional (see _scan_window_default above).
 SCAN_PARAMETERS_SCHEMA = ble_device_base.scan_parameters_schema(
-    "320ms", window_default=_scan_window_default
+    "320ms", window_default=_scan_window_default, connection_window=True
 )
 
 # Codegen helpers are owned by ble_device_base; kept under the historical names
@@ -262,7 +283,7 @@ ESP_BLE_DEVICE_SCHEMA = cv.Schema(
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     # Register the loggers this component needs
     esp32_ble.register_bt_logger(BTLoggers.BLE_SCAN)
 
@@ -287,6 +308,25 @@ async def to_code(config):
     cg.add(var.set_scan_duration(params[CONF_DURATION]))
     cg.add(var.set_scan_interval(ble_device_base.to_ble_units(params[CONF_INTERVAL])))
     cg.add(var.set_scan_window(ble_device_base.to_ble_units(params[CONF_WINDOW])))
+    if (connection_window := params.get(CONF_CONNECTION_SCAN_WINDOW)) is not None:
+        # Emitted at FINAL so a scan-only build, where the guarded C++ path
+        # compiles out, skips the call entirely.
+        window_units = ble_device_base.to_ble_units(connection_window)
+
+        @coroutine_with_priority(CoroPriority.FINAL)
+        async def _emit_connection_scan_window() -> None:
+            if cg.get_slot_count(CLIENT_COUNT_DEFINE):
+                cg.add(var.set_connection_scan_window(window_units))
+            elif not _get_data().connection_window_injected:
+                # Warn only for a user-set value; the injected default drops silently.
+                _LOGGER.warning(
+                    "'%s' has no effect because this build has no BLE client "
+                    "components (for example bluetooth_proxy with active "
+                    "connections, or ble_client)",
+                    CONF_CONNECTION_SCAN_WINDOW,
+                )
+
+        CORE.add_job(_emit_connection_scan_window)
     cg.add(var.set_scan_active(params[CONF_ACTIVE]))
     cg.add(var.set_scan_continuous(params[CONF_CONTINUOUS]))
 
@@ -360,7 +400,7 @@ async def to_code(config):
 # chance to call register_ble_tracker and register_client before the list is checked
 # and added to the global defines list.
 @coroutine_with_priority(CoroPriority.FINAL)
-async def _add_ble_features():
+async def _add_ble_features() -> None:
     # Add feature-specific defines based on what's needed
     required_features = _get_required_features()
     # Sensors registered through the neutral ble_device_base path (BLEHub) need
@@ -389,8 +429,11 @@ ESP32_BLE_START_SCAN_ACTION_SCHEMA = cv.Schema(
     synchronous=True,
 )
 async def esp32_ble_tracker_start_scan_action_to_code(
-    config, action_id, template_arg, args
-):
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     paren = await cg.get_variable(config[CONF_ID])
     var = cg.new_Pvariable(action_id, template_arg, paren)
     template_ = await cg.templatable(config[CONF_CONTINUOUS], args, cg.bool_)
@@ -414,8 +457,11 @@ ESP32_BLE_STOP_SCAN_ACTION_SCHEMA = automation.maybe_simple_id(
     synchronous=True,
 )
 async def esp32_ble_tracker_stop_scan_action_to_code(
-    config, action_id, template_arg, args
-):
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     var = cg.new_Pvariable(action_id, template_arg)
     await cg.register_parented(var, config[CONF_ID])
     return var

esphome/components/esp32_ble_tracker/esp32_ble_tracker.cpp

@@ -122,6 +122,9 @@ void ESP32BLETracker::loop() {
   // - start_scan_(): scanner_state_ becomes IDLE via set_scanner_state_() in cleanup_scan_state_()
   // - try_promote_discovered_clients_(): client enters DISCOVERED via set_state(), or
   //   connecting client finishes (state change), or scanner reaches RUNNING/IDLE
+  // - connection-window restart: scan_params_ is only written in start_scan_()
+  //   (which changes scanner state via set_scanner_state_()), and
+  //   counts.active/disconnecting only change on client state changes
   //
   // All conditions that affect the logic below are tied to state changes that increment
   // state_version_, so the fast path is safe.
@@ -144,6 +147,19 @@ void ESP32BLETracker::loop() {
       (this->scan_set_param_failed_ && this->scanner_state_ == ScannerState::RUNNING)) {
     this->handle_scanner_failure_();
   }
+
+#ifdef ESPHOME_ESP32_BLE_TRACKER_CLIENT_COUNT
+  // The programmed window no longer matches the connection state (typically
+  // the last connection dropped): restart so the right window applies now
+  // instead of at the end of the scan period. Continuous only (a user-started
+  // scan would not restart); !disconnecting matches the restart gate below.
+  if (this->scanner_state_ == ScannerState::RUNNING && this->scan_continuous_ && !counts.disconnecting &&
+      this->scan_params_.scan_window != this->desired_scan_window_(counts.active)) {
+    // Same logical scan period continues: no on_scan_end sweeps for this
+    // restart. Only armed when the stop was issued.
+    this->skip_next_scan_end_ = this->stop_scan_();
+  }
+#endif
   /*
 
     Avoid starting the scanner if:
@@ -195,19 +211,23 @@ void ESP32BLETracker::stop_scan() {
   // reason at D themselves, and the user-facing stop action is deliberate.
   ESP_LOGV(TAG, "Stopping scan.");
   this->scan_continuous_ = false;
+#ifdef ESPHOME_ESP32_BLE_TRACKER_CLIENT_COUNT
+  // The window-change restart is abandoned with continuous scanning.
+  this->skip_next_scan_end_ = false;
+#endif
   this->stop_scan_();
 }
 
 void ESP32BLETracker::ble_before_disabled_event_handler() { this->stop_scan_(); }
 
-void ESP32BLETracker::stop_scan_() {
+bool ESP32BLETracker::stop_scan_() {
   if (this->scanner_state_ != ScannerState::RUNNING && this->scanner_state_ != ScannerState::FAILED) {
     // IDLE means there is nothing to stop; STOPPING means a stop is already in
     // flight and will finish on its own. Neither is an error.
     if (this->scanner_state_ != ScannerState::IDLE && this->scanner_state_ != ScannerState::STOPPING) {
       ESP_LOGE(TAG, "Cannot stop scan: %s", this->scanner_state_to_string_(this->scanner_state_));
     }
-    return;
+    return false;
   }
   // Reset timeout state machine when stopping scan
   this->scan_timeout_state_ = ScanTimeoutState::INACTIVE;
@@ -215,8 +235,9 @@ void ESP32BLETracker::stop_scan_() {
   esp_err_t err = esp_ble_gap_stop_scanning();
   if (err != ESP_OK) {
     ESP_LOGE(TAG, "esp_ble_gap_stop_scanning failed: %d", err);
-    return;
+    return false;
   }
+  return true;
 }
 
 void ESP32BLETracker::start_scan_(bool first) {
@@ -230,16 +251,11 @@ void ESP32BLETracker::start_scan_(bool first) {
   }
   this->set_scanner_state_(ScannerState::STARTING);
   ESP_LOGV(TAG, "Starting scan, set scanner state to STARTING.");
-  if (!first) {
-#ifdef ESPHOME_ESP32_BLE_TRACKER_LISTENER_COUNT
-    for (auto *listener : this->listeners_)
-      listener->on_scan_end();
+  if (!first)
+    this->notify_scan_end_();
+#ifdef ESPHOME_ESP32_BLE_TRACKER_CLIENT_COUNT
+  this->skip_next_scan_end_ = false;
 #endif
-#ifdef ESPHOME_BLE_DEVICE_BASE_LISTENER_COUNT
-    for (auto *listener : this->neutral_listeners_)
-      listener->on_scan_end();
-#endif
-  }
 #ifdef USE_ESP32_BLE_DEVICE
   this->discovered_log_.clear();
 #endif
@@ -247,7 +263,17 @@ void ESP32BLETracker::start_scan_(bool first) {
   this->scan_params_.own_addr_type = BLE_ADDR_TYPE_PUBLIC;
   this->scan_params_.scan_filter_policy = BLE_SCAN_FILTER_ALLOW_ALL;
   this->scan_params_.scan_interval = this->scan_interval_;
-  this->scan_params_.scan_window = this->scan_window_;
+#ifdef ESPHOME_ESP32_BLE_TRACKER_CLIENT_COUNT
+  // Count fresh: an automation can start a scan before loop() refreshes the counts.
+  const uint32_t window = this->desired_scan_window_(this->count_client_states_().active);
+  if (window != this->scan_window_) {
+    // Guarantee the connection airtime instead of scanning wall to wall.
+    ESP_LOGV(TAG, "Connection active, using %" PRIu32 " unit scan window", window);
+  }
+#else
+  const uint32_t window = this->scan_window_;
+#endif
+  this->scan_params_.scan_window = window;
 
   // Start timeout monitoring in loop() instead of using scheduler
   // This prevents false reboots when the loop is blocked
@@ -408,6 +434,11 @@ void ESP32BLETracker::dump_config() {
                 "  Continuous Scanning: %s",
                 this->scan_duration_, this->scan_interval_ * 0.625f, this->scan_window_ * 0.625f,
                 this->scan_active_ ? "ACTIVE" : "PASSIVE", YESNO(this->scan_continuous_));
+#ifdef ESPHOME_ESP32_BLE_TRACKER_CLIENT_COUNT
+  if (this->connection_scan_window_ != 0) {
+    ESP_LOGCONFIG(TAG, "  Connection Scan Window: %.1f ms", this->connection_scan_window_ * 0.625f);
+  }
+#endif
   ESP_LOGCONFIG(TAG,
                 "  Scanner State: %s\n"
                 "  Connecting: %d, discovered: %d, disconnecting: %d, active: %d",
@@ -487,6 +518,18 @@ void ESP32BLETracker::cleanup_scan_state_(bool is_stop_complete) {
   // Reset timeout state machine instead of cancelling scheduler timeout
   this->scan_timeout_state_ = ScanTimeoutState::INACTIVE;
 
+  this->notify_scan_end_();
+
+  this->set_scanner_state_(ScannerState::IDLE);
+}
+
+void ESP32BLETracker::notify_scan_end_() {
+#ifdef ESPHOME_ESP32_BLE_TRACKER_CLIENT_COUNT
+  // Window-change restart continues the same scan period; the flag stays set
+  // across the stop and is cleared by the restart in start_scan_.
+  if (this->skip_next_scan_end_)
+    return;
+#endif
 #ifdef ESPHOME_ESP32_BLE_TRACKER_LISTENER_COUNT
   for (auto *listener : this->listeners_)
     listener->on_scan_end();
@@ -495,8 +538,6 @@ void ESP32BLETracker::cleanup_scan_state_(bool is_stop_complete) {
   for (auto *listener : this->neutral_listeners_)
     listener->on_scan_end();
 #endif
-
-  this->set_scanner_state_(ScannerState::IDLE);
 }
 
 void ESP32BLETracker::handle_scanner_failure_() {
@@ -534,6 +575,8 @@ void ESP32BLETracker::try_promote_discovered_clients_() {
     }
 
     ESP_LOGD(TAG, "Promoting client to connect");
+    // A connect ends the scan period a window-change restart was continuing.
+    this->skip_next_scan_end_ = false;
 #ifdef USE_ESP32_BLE_SOFTWARE_COEXISTENCE
     this->update_coex_preference_(true);
 #endif

esphome/components/esp32_ble_tracker/esp32_ble_tracker.h

@@ -169,6 +169,9 @@ class ESP32BLETracker final : public Component,
   void set_scan_duration(uint32_t scan_duration) { scan_duration_ = scan_duration; }
   void set_scan_interval(uint32_t scan_interval) { scan_interval_ = scan_interval; }
   void set_scan_window(uint32_t scan_window) { scan_window_ = scan_window; }
+#ifdef ESPHOME_ESP32_BLE_TRACKER_CLIENT_COUNT
+  void set_connection_scan_window(uint32_t scan_window) { connection_scan_window_ = scan_window; }
+#endif
   void set_scan_active(bool scan_active) { scan_active_ = scan_active; }
   bool get_scan_active() const { return scan_active_; }
   void set_scan_continuous(bool scan_continuous) { scan_continuous_ = scan_continuous; }
@@ -226,7 +229,10 @@ class ESP32BLETracker final : public Component,
   ScannerState get_scanner_state() const { return this->scanner_state_; }
 
  protected:
-  void stop_scan_();
+  /// Returns true when a stop was issued to the controller.
+  bool stop_scan_();
+  /// Fire on_scan_end on every listener unless a window-change restart suppressed it.
+  void notify_scan_end_();
   /// Start a single scan by setting up the parameters and doing some esp-idf calls.
   void start_scan_(bool first);
   /// Called when a `ESP_GAP_BLE_SCAN_RESULT_EVT` event is received.
@@ -313,6 +319,15 @@ class ESP32BLETracker final : public Component,
   uint32_t scan_duration_;
   uint32_t scan_interval_;
   uint32_t scan_window_;
+#ifdef ESPHOME_ESP32_BLE_TRACKER_CLIENT_COUNT
+  /// Window used while a GATT connection is active; set by the user, or
+  /// defaulted when the window was raised to full duty (0 = no fallback).
+  uint32_t connection_scan_window_{0};
+  /// The window to scan at for the given number of active GATT connections.
+  uint32_t desired_scan_window_(uint8_t active) const {
+    return (this->connection_scan_window_ != 0 && active > 0) ? this->connection_scan_window_ : this->scan_window_;
+  }
+#endif
   esp_bt_status_t scan_start_failed_{ESP_BT_STATUS_SUCCESS};
   esp_bt_status_t scan_set_param_failed_{ESP_BT_STATUS_SUCCESS};
 
@@ -330,15 +345,20 @@ class ESP32BLETracker final : public Component,
   /// state_version_ to detect if any state changed since last iteration.
   uint8_t last_processed_version_{0};
   ScannerState scanner_state_{ScannerState::IDLE};
-  bool scan_continuous_;
-  bool scan_active_;
+  // Packed 1-bit flags.
+  bool scan_continuous_ : 1;
+  bool scan_active_ : 1;
 #ifdef USE_OTA_STATE_LISTENER
-  bool scan_continuous_before_ota_{false};
+  bool scan_continuous_before_ota_ : 1 {false};
+#endif
+  bool ble_was_disabled_ : 1 {true};
+  bool parse_advertisements_ : 1 {false};
+#ifdef ESPHOME_ESP32_BLE_TRACKER_CLIENT_COUNT
+  /// Suppress the window-change restart's on_scan_end sweeps (stop and start).
+  bool skip_next_scan_end_ : 1 {false};
 #endif
-  bool ble_was_disabled_{true};
-  bool parse_advertisements_{false};
 #ifdef USE_ESP32_BLE_SOFTWARE_COEXISTENCE
-  bool coex_prefer_ble_{false};
+  bool coex_prefer_ble_ : 1 {false};
 #endif
   // Scan timeout state machine
   enum class ScanTimeoutState : uint8_t {
@@ -346,10 +366,10 @@ class ESP32BLETracker final : public Component,
     MONITORING,     // Actively monitoring for timeout
     EXCEEDED_WAIT,  // Timeout exceeded, waiting one loop before reboot
   };
+  ScanTimeoutState scan_timeout_state_{ScanTimeoutState::INACTIVE};
   uint32_t scan_start_time_{0};
   /// Precomputed timeout value: scan_duration_ * 2000
   uint32_t scan_timeout_ms_{0};
-  ScanTimeoutState scan_timeout_state_{ScanTimeoutState::INACTIVE};
 };
 
 // NOLINTNEXTLINE

esphome/components/esp32_hosted/update/esp32_hosted_update.cpp

@@ -135,6 +135,10 @@ void Esp32HostedUpdate::setup() {
   // Publish state
   this->status_clear_error();
   this->publish_state();
+  // Defer so the automation runs on the main loop after setup, not during App.setup()
+  if (this->state_ == update::UPDATE_STATE_AVAILABLE && this->update_available_trigger_) {
+    this->defer([this]() { this->update_available_trigger_->trigger(this->update_info_); });
+  }
 #else
   // HTTP mode: check every 10s until network is ready (max 6 attempts)
   // Only if update interval is > 1 minute to avoid redundant checks
@@ -185,6 +189,8 @@ void Esp32HostedUpdate::check() {
     return;
   }
 
+  const bool was_available = this->state_ == update::UPDATE_STATE_AVAILABLE;
+
   // Compare versions
   if (this->update_info_.latest_version.empty() ||
       this->update_info_.latest_version == this->update_info_.current_version) {
@@ -197,6 +203,9 @@ void Esp32HostedUpdate::check() {
   this->update_info_.progress = 0.0f;
   this->status_clear_error();
   this->publish_state();
+  if (this->state_ == update::UPDATE_STATE_AVAILABLE && !was_available && this->update_available_trigger_) {
+    this->update_available_trigger_->trigger(this->update_info_);
+  }
 #endif
 }
 

esphome/components/ethernet/__init__.py

@@ -48,10 +48,12 @@ from esphome.const import (
 )
 from esphome.core import (
     CORE,
+    ID,
     CoroPriority,
     TimePeriodMilliseconds,
     coroutine_with_priority,
 )
+from esphome.cpp_generator import MockObj, TemplateArgsType
 import esphome.final_validate as fv
 from esphome.types import ConfigType
 
@@ -276,7 +278,7 @@ def _validate_spi_interface(config: ConfigType) -> ConfigType:
     return config
 
 
-def _validate(config):
+def _validate(config: ConfigType) -> ConfigType:
     if CONF_USE_ADDRESS not in config:
         if CONF_MANUAL_IP in config:
             use_address = str(config[CONF_MANUAL_IP][CONF_STATIC_IP])
@@ -441,7 +443,7 @@ GENERIC_SCHEMA = cv.All(
 )
 
 
-def _spi_schema(default_clock: str = "26.67MHz", max_clock: int = int(80e6)):
+def _spi_schema(default_clock: str = "26.67MHz", max_clock: int = int(80e6)) -> cv.All:
     return cv.All(
         BASE_SCHEMA.extend(
             cv.Schema(
@@ -517,7 +519,7 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-def _final_validate_spi(config):
+def _final_validate_spi(config: ConfigType) -> None:
     if not CORE.is_esp32:
         return  # SPI interface validation is ESP32-only
     if config[CONF_TYPE] not in SPI_ETHERNET_TYPES:
@@ -537,7 +539,7 @@ def _final_validate_spi(config):
                     )
 
 
-def manual_ip(config):
+def manual_ip(config: ConfigType) -> cg.StructInitializer:
     return cg.StructInitializer(
         ManualIP,
         ("static_ip", ip_address_literal(config[CONF_STATIC_IP])),
@@ -548,7 +550,7 @@ def manual_ip(config):
     )
 
 
-def phy_register(address: int, value: int, page: int):
+def phy_register(address: int, value: int, page: int) -> cg.StructInitializer:
     return cg.StructInitializer(
         PHYRegister,
         ("address", address),
@@ -558,7 +560,7 @@ def phy_register(address: int, value: int, page: int):
 
 
 @coroutine_with_priority(CoroPriority.COMMUNICATION)
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     var = cg.new_Pvariable(config[CONF_ID])
 
     # Apply network priority before register_component (which emits the user's
@@ -610,7 +612,7 @@ async def to_code(config):
     CORE.add_job(final_step)
 
 
-async def _to_code_esp32(var: cg.Pvariable, config: ConfigType) -> None:
+async def _to_code_esp32(var: cg.MockObj, config: ConfigType) -> None:
     from esphome.components.esp32 import (
         add_idf_component,
         add_idf_sdkconfig_option,
@@ -698,7 +700,7 @@ async def _to_code_esp32(var: cg.Pvariable, config: ConfigType) -> None:
         add_idf_component(name=component.name, ref=component.version)
 
 
-async def _to_code_rp2040(var: cg.Pvariable, config: ConfigType) -> None:
+async def _to_code_rp2040(var: cg.MockObj, config: ConfigType) -> None:
     cg.add(var.set_clk_pin(config[CONF_CLK_PIN]))
     cg.add(var.set_miso_pin(config[CONF_MISO_PIN]))
     cg.add(var.set_mosi_pin(config[CONF_MOSI_PIN]))
@@ -793,7 +795,7 @@ FINAL_VALIDATE_SCHEMA = _final_validate
 
 
 @coroutine_with_priority(CoroPriority.FINAL)
-async def final_step():
+async def final_step() -> None:
     """Final code generation step to configure optional Ethernet features."""
     if ip_state_count := CORE.data.get(ETHERNET_IP_STATE_LISTENERS_KEY, 0):
         cg.add_define("USE_ETHERNET_IP_STATE_LISTENERS")
@@ -845,7 +847,12 @@ def _filter_source_files() -> list[str]:
 FILTER_SOURCE_FILES = _filter_source_files
 
 
-async def _new_pvariable_to_code(config, id_, template_arg, args):
+async def _new_pvariable_to_code(
+    config: ConfigType,
+    id_: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     return cg.new_Pvariable(id_, template_arg)
 
 

esphome/components/hdc302x/sensor.py

@@ -1,3 +1,5 @@
+from typing import Any
+
 from esphome import automation
 from esphome.automation import maybe_simple_id
 import esphome.codegen as cg
@@ -16,6 +18,9 @@ from esphome.const import (
     UNIT_CELSIUS,
     UNIT_PERCENT,
 )
+from esphome.core import ID
+from esphome.cpp_generator import MockObj, TemplateArgsType
+from esphome.types import ConfigType
 
 DEPENDENCIES = ["i2c"]
 
@@ -62,7 +67,7 @@ CONFIG_SCHEMA = (
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     var = cg.new_Pvariable(config[CONF_ID])
     await cg.register_component(var, config)
     await i2c.register_i2c_device(var, config)
@@ -86,7 +91,7 @@ HDC302X_HEATER_POWER_MAP = {
 }
 
 
-def heater_power_value(value):
+def heater_power_value(value: Any) -> cv.Lambda | int:
     """Accept enum names or raw uint16 values"""
     if isinstance(value, cv.Lambda):
         return value
@@ -119,7 +124,12 @@ HDC302X_HEATER_ON_ACTION_SCHEMA = maybe_simple_id(
     HDC302X_HEATER_ON_ACTION_SCHEMA,
     synchronous=True,
 )
-async def hdc302x_heater_on_to_code(config, action_id, template_arg, args):
+async def hdc302x_heater_on_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     var = cg.new_Pvariable(action_id, template_arg)
     await cg.register_parented(var, config[CONF_ID])
     template_ = await cg.templatable(config[CONF_POWER], args, cg.uint16)
@@ -135,7 +145,12 @@ async def hdc302x_heater_on_to_code(config, action_id, template_arg, args):
     HDC302X_ACTION_SCHEMA,
     synchronous=True,
 )
-async def hdc302x_heater_off_to_code(config, action_id, template_arg, args):
+async def hdc302x_heater_off_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     var = cg.new_Pvariable(action_id, template_arg)
     await cg.register_parented(var, config[CONF_ID])
     return var

esphome/components/htu21d/sensor.py

@@ -17,6 +17,9 @@ from esphome.const import (
     UNIT_EMPTY,
     UNIT_PERCENT,
 )
+from esphome.core import ID
+from esphome.cpp_generator import MockObj, TemplateArgsType
+from esphome.types import ConfigType
 
 DEPENDENCIES = ["i2c"]
 
@@ -63,7 +66,7 @@ CONFIG_SCHEMA = (
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     var = cg.new_Pvariable(config[CONF_ID])
     await cg.register_component(var, config)
     await i2c.register_i2c_device(var, config)
@@ -95,7 +98,12 @@ async def to_code(config):
     ),
     synchronous=True,
 )
-async def set_heater_level_to_code(config, action_id, template_arg, args):
+async def set_heater_level_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     var = cg.new_Pvariable(action_id, template_arg)
     await cg.register_parented(var, config[CONF_ID])
     level_ = await cg.templatable(config[CONF_LEVEL], args, cg.uint8)
@@ -115,7 +123,12 @@ async def set_heater_level_to_code(config, action_id, template_arg, args):
     ),
     synchronous=True,
 )
-async def set_heater_to_code(config, action_id, template_arg, args):
+async def set_heater_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     var = cg.new_Pvariable(action_id, template_arg)
     await cg.register_parented(var, config[CONF_ID])
     status_ = await cg.templatable(config[CONF_STATUS], args, cg.bool_)

esphome/components/ld6002b/__init__.py

@@ -60,7 +60,7 @@ FINAL_VALIDATE_SCHEMA = uart.final_validate_device_schema(
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     var = cg.new_Pvariable(config[CONF_ID])
     await cg.register_component(var, config)
     await uart.register_uart_device(var, config)

esphome/components/ld6002b/binary_sensor.py

@@ -2,6 +2,7 @@ import esphome.codegen as cg
 from esphome.components import binary_sensor
 import esphome.config_validation as cv
 from esphome.const import CONF_TARGET, DEVICE_CLASS_OCCUPANCY
+from esphome.types import ConfigType
 
 from . import LD6002BComponent
 from .const import AREA_COUNT, CONF_LD6002B_ID, MAX_TARGETS
@@ -36,7 +37,7 @@ CONFIG_SCHEMA = (
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     hub = await cg.get_variable(config[CONF_LD6002B_ID])
 
     if target_config := config.get(CONF_TARGET):

esphome/components/ld6002b/button/__init__.py

@@ -129,7 +129,7 @@ BUTTON_MAP = {
 }
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     for key, button_type in BUTTON_MAP.items():
         if button_config := config.get(key):
             b = cg.new_Pvariable(button_config[CONF_ID], button_type)

esphome/components/ld6002b/number/__init__.py

@@ -136,7 +136,7 @@ def final_validate(config: ConfigType) -> None:
 FINAL_VALIDATE_SCHEMA = final_validate
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     hub = await cg.get_variable(config[CONF_LD6002B_ID])
 
     for key, number_type, setter, min_value, max_value, step in (

esphome/components/ld6002b/select/__init__.py

@@ -2,6 +2,7 @@ import esphome.codegen as cg
 from esphome.components import select
 import esphome.config_validation as cv
 from esphome.const import CONF_AREA_ID, CONF_SENSITIVITY, ENTITY_CATEGORY_CONFIG
+from esphome.types import ConfigType
 
 from .. import LD6002BComponent, ld6002b_ns
 from ..const import CONF_INSTALLATION_MODE, CONF_LD6002B_ID, CONF_TRIGGER_SPEED
@@ -64,7 +65,7 @@ SELECT_MAP = (
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     hub = await cg.get_variable(config[CONF_LD6002B_ID])
 
     for key, select_type, setter, options in SELECT_MAP:

esphome/components/ld6002b/sensor.py

@@ -9,6 +9,7 @@ from esphome.const import (
     STATE_CLASS_MEASUREMENT,
     UNIT_METER,
 )
+from esphome.types import ConfigType
 
 from . import LD6002BComponent
 from .const import (
@@ -150,7 +151,7 @@ CONFIG_SCHEMA = (
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     hub = await cg.get_variable(config[CONF_LD6002B_ID])
 
     if target_count_config := config.get(CONF_TARGET_COUNT):

esphome/components/ld6002b/switch/__init__.py

@@ -2,6 +2,7 @@ import esphome.codegen as cg
 from esphome.components import switch
 import esphome.config_validation as cv
 from esphome.const import DEVICE_CLASS_SWITCH, ENTITY_CATEGORY_CONFIG
+from esphome.types import ConfigType
 
 from .. import LD6002BComponent, ld6002b_ns
 from ..const import (
@@ -46,7 +47,7 @@ CONFIG_SCHEMA = cv.Schema(
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     hub = await cg.get_variable(config[CONF_LD6002B_ID])
 
     for key, switch_type, setter in (

esphome/components/ld6002b/text_sensor.py

@@ -2,6 +2,7 @@ import esphome.codegen as cg
 from esphome.components import text_sensor
 import esphome.config_validation as cv
 from esphome.const import ENTITY_CATEGORY_DIAGNOSTIC
+from esphome.types import ConfigType
 
 from . import LD6002BComponent
 from .const import CONF_LD6002B_ID, CONF_OTA_VERSION, CONF_WORK_MODE
@@ -21,7 +22,7 @@ CONFIG_SCHEMA = cv.Schema(
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     hub = await cg.get_variable(config[CONF_LD6002B_ID])
     if work_mode_config := config.get(CONF_WORK_MODE):
         sens = await text_sensor.new_text_sensor(work_mode_config)

esphome/components/m5stack_8angle/__init__.py

@@ -2,6 +2,7 @@ import esphome.codegen as cg
 from esphome.components import i2c
 import esphome.config_validation as cv
 from esphome.const import CONF_ID
+from esphome.types import ConfigType
 
 DEPENDENCIES = ["i2c"]
 CODEOWNERS = ["@rnauber"]
@@ -26,7 +27,7 @@ CONFIG_SCHEMA = cv.Schema(
 ).extend(i2c.i2c_device_schema(0x43))
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     var = cg.new_Pvariable(config[CONF_ID])
     await cg.register_component(var, config)
     await i2c.register_i2c_device(var, config)

esphome/components/m5stack_8angle/binary_sensor/__init__.py

@@ -1,6 +1,7 @@
 import esphome.codegen as cg
 from esphome.components import binary_sensor
 import esphome.config_validation as cv
+from esphome.types import ConfigType
 
 from .. import CONF_M5STACK_8ANGLE_ID, M5Stack8AngleComponent, m5stack_8angle_ns
 
@@ -22,7 +23,7 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     hub = await cg.get_variable(config[CONF_M5STACK_8ANGLE_ID])
     sens = await binary_sensor.new_binary_sensor(config)
     cg.add(sens.set_parent(hub))

esphome/components/m5stack_8angle/light/__init__.py

@@ -2,6 +2,7 @@ import esphome.codegen as cg
 from esphome.components import light
 import esphome.config_validation as cv
 from esphome.const import CONF_OUTPUT_ID
+from esphome.types import ConfigType
 
 from .. import CONF_M5STACK_8ANGLE_ID, M5Stack8AngleComponent, m5stack_8angle_ns
 
@@ -21,7 +22,7 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     hub = await cg.get_variable(config[CONF_M5STACK_8ANGLE_ID])
     lights = cg.new_Pvariable(config[CONF_OUTPUT_ID])
     await light.register_light(lights, config)

esphome/components/m5stack_8angle/sensor/__init__.py

@@ -8,6 +8,7 @@ from esphome.const import (
     ICON_ROTATE_RIGHT,
     STATE_CLASS_MEASUREMENT,
 )
+from esphome.types import ConfigType
 
 from .. import (
     CONF_M5STACK_8ANGLE_ID,
@@ -55,7 +56,7 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     var = await sensor.new_sensor(config)
     await cg.register_component(var, config)
     await cg.register_parented(var, config[CONF_M5STACK_8ANGLE_ID])

esphome/components/modbus/__init__.py

@@ -8,8 +8,10 @@ import esphome.codegen as cg
 from esphome.components import uart
 import esphome.config_validation as cv
 from esphome.const import CONF_ADDRESS, CONF_DISABLE_CRC, CONF_FLOW_CONTROL_PIN, CONF_ID
+from esphome.cpp_generator import MockObj
 from esphome.cpp_helpers import gpio_pin_expression
 import esphome.final_validate as fv
+from esphome.types import ConfigType
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -84,7 +86,7 @@ CONFIG_SCHEMA = cv.typed_schema(
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     cg.add_global(modbus_ns.using)
     var = cg.new_Pvariable(config[CONF_ID])
     await cg.register_component(var, config)
@@ -112,7 +114,9 @@ def _validate_server_address(value: Any) -> int:
     return address
 
 
-def modbus_device_schema(default_address, role: Literal["client", "server"] = "client"):
+def modbus_device_schema(
+    default_address: int | None, role: Literal["client", "server"] = "client"
+) -> cv.Schema:
     hub_type = ModbusClient if role == "client" else ModbusServer
     address_validator = _validate_server_address if role == "server" else cv.hex_uint8_t
     schema = {
@@ -127,14 +131,14 @@ def modbus_device_schema(default_address, role: Literal["client", "server"] = "c
 
 def final_validate_modbus_device(
     name: str, *, role: Literal["server", "client"] | None = None
-):
-    def validate_role(value):
+) -> cv.Schema:
+    def validate_role(value: str) -> str:
         assert role in MODBUS_ROLES
         if value != role:
             raise cv.Invalid(f"Component {name} requires role to be {role}")
         return value
 
-    def validate_hub(hub_config):
+    def validate_hub(hub_config: ConfigType) -> ConfigType:
         hub_schema = {}
         if role is not None:
             hub_schema[cv.Required(CONF_ROLE)] = validate_role
@@ -147,19 +151,19 @@ def final_validate_modbus_device(
     )
 
 
-async def register_modbus_client_device(var, config):
+async def register_modbus_client_device(var: MockObj, config: ConfigType) -> None:
     parent = await cg.get_variable(config[CONF_MODBUS_ID])
     cg.add(var.set_parent(parent))
     cg.add(var.set_address(config[CONF_ADDRESS]))
 
 
-async def register_modbus_server_device(var, config):
+async def register_modbus_server_device(var: MockObj, config: ConfigType) -> None:
     parent = await cg.get_variable(config[CONF_MODBUS_ID])
     cg.add(var.set_address(config[CONF_ADDRESS]))
     cg.add(parent.register_device(var))
 
 
-async def register_modbus_device(var, config):
+async def register_modbus_device(var: MockObj, config: ConfigType) -> None:
     # Remove before 2026.12.0
     _LOGGER.warning(
         "'register_modbus_device' is deprecated, use 'register_modbus_client_device' "

esphome/components/openthread/__init__.py

@@ -1,3 +1,5 @@
+from typing import Any
+
 from esphome import automation
 import esphome.codegen as cg
 from esphome.components.esp32 import (
@@ -31,10 +33,12 @@ from esphome.const import (
 )
 from esphome.core import (
     CORE,
+    ID,
     CoroPriority,
     TimePeriodMilliseconds,
     coroutine_with_priority,
 )
+from esphome.cpp_generator import MockObj, TemplateArgsType
 import esphome.final_validate as fv
 from esphome.types import ConfigType
 
@@ -76,7 +80,7 @@ CONF_DEVICE_TYPES = [
 ]
 
 
-def _validate_txpower(value):
+def _validate_txpower(value: Any) -> int | float:
     if CORE.is_esp32:
         variant = get_esp32_variant()
 
@@ -90,7 +94,7 @@ def _validate_txpower(value):
     return value  # Unsupported, fail later with clear error
 
 
-def set_sdkconfig_options(config):
+def set_sdkconfig_options(config: ConfigType) -> None:
     # and expose options for using SPI/UART RCPs
     add_idf_sdkconfig_option("CONFIG_IEEE802154_ENABLED", True)
     add_idf_sdkconfig_option("CONFIG_OPENTHREAD_RADIO_NATIVE", True)
@@ -180,7 +184,7 @@ def _validate(config: ConfigType) -> ConfigType:
     return config
 
 
-def _require_vfs_select(config):
+def _require_vfs_select(config: ConfigType) -> ConfigType:
     """Register VFS select requirement during config validation."""
     # OpenThread uses esp_vfs_eventfd which requires VFS select support (ESP32 only)
     if CORE.is_esp32:
@@ -188,7 +192,7 @@ def _require_vfs_select(config):
     return config
 
 
-def _validate_platform(config):
+def _validate_platform(config: ConfigType) -> ConfigType:
     if CORE.using_zephyr:
         return config
     return only_on_variant(
@@ -203,7 +207,7 @@ def _validate_platform(config):
     )(config)
 
 
-def _validate_tlv_hex(value):
+def _validate_tlv_hex(value: Any) -> str:
     s = cv.string_strict(value)
     if len(s) % 2 != 0:
         raise cv.Invalid("TLV must have an even number of hex characters")
@@ -242,7 +246,7 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-def _final_validate(_):
+def _final_validate(_: ConfigType) -> None:
     full_config = fv.full_config.get()
     network_config = full_config.get("network", {})
     if not network_config.get(CONF_ENABLE_IPV6, False):
@@ -274,7 +278,7 @@ FILTER_SOURCE_FILES = filter_source_files_from_platform(
 
 
 @coroutine_with_priority(CoroPriority.COMMUNICATION)
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     # Re-enable openthread IDF component (excluded by default)
     if CORE.is_esp32:
         include_builtin_idf_component("openthread")
@@ -339,7 +343,12 @@ POLL_PERIOD_ACTION_SCHEMA = automation.maybe_conf(
     POLL_PERIOD_ACTION_SCHEMA,
     synchronous=True,
 )
-async def openthread_poll_period_action_to_code(config, action_id, template_arg, args):
+async def openthread_poll_period_action_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     paren = await cg.get_variable(config[CONF_ID])
     var = cg.new_Pvariable(action_id, template_arg, paren)
     template_ = await cg.templatable(config[CONF_POLL_PERIOD], args, cg.uint32)

esphome/components/pulse_counter/sensor.py

@@ -1,3 +1,5 @@
+from typing import Any
+
 from esphome import automation, pins
 import esphome.codegen as cg
 from esphome.components import sensor
@@ -19,7 +21,9 @@ from esphome.const import (
     UNIT_PULSES,
     UNIT_PULSES_PER_MINUTE,
 )
-from esphome.core import CORE
+from esphome.core import CORE, ID
+from esphome.cpp_generator import MockObj, TemplateArgsType
+from esphome.types import ConfigType
 
 CONF_USE_PCNT = "use_pcnt"
 
@@ -42,7 +46,7 @@ SetTotalPulsesAction = pulse_counter_ns.class_(
 )
 
 
-def validate_internal_filter(value):
+def validate_internal_filter(value: ConfigType) -> ConfigType:
     use_pcnt = value.get(CONF_USE_PCNT)
     if CORE.is_esp8266 and use_pcnt:
         raise cv.Invalid(
@@ -63,7 +67,7 @@ def validate_internal_filter(value):
     return value
 
 
-def validate_pulse_counter_pin(value):
+def validate_pulse_counter_pin(value: Any) -> ConfigType:
     value = pins.internal_gpio_input_pin_schema(value)
     if CORE.is_esp8266 and value[CONF_NUMBER] >= 16:
         raise cv.Invalid(
@@ -72,7 +76,7 @@ def validate_pulse_counter_pin(value):
     return value
 
 
-def validate_count_mode(value):
+def validate_count_mode(value: ConfigType) -> ConfigType:
     rising_edge = value[CONF_RISING_EDGE]
     falling_edge = value[CONF_FALLING_EDGE]
     if rising_edge == "DISABLE" and falling_edge == "DISABLE":
@@ -126,7 +130,7 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     use_pcnt = config.get(CONF_USE_PCNT)
     if CORE.is_esp32 and use_pcnt:
         include_builtin_idf_component("esp_driver_pcnt")
@@ -157,7 +161,12 @@ async def to_code(config):
     ),
     synchronous=True,
 )
-async def set_total_action_to_code(config, action_id, template_arg, args):
+async def set_total_action_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     paren = await cg.get_variable(config[CONF_ID])
     var = cg.new_Pvariable(action_id, template_arg, paren)
     template_ = await cg.templatable(config[CONF_VALUE], args, cg.uint32)

esphome/components/pulse_meter/sensor.py

@@ -1,3 +1,5 @@
+from typing import Any
+
 from esphome import automation, pins
 import esphome.codegen as cg
 from esphome.components import sensor
@@ -17,7 +19,9 @@ from esphome.const import (
     UNIT_PULSES,
     UNIT_PULSES_PER_MINUTE,
 )
-from esphome.core import CORE
+from esphome.core import CORE, ID, TimePeriodMicroseconds
+from esphome.cpp_generator import MockObj, TemplateArgsType
+from esphome.types import ConfigType
 
 CODEOWNERS = ["@stevebaxter", "@cstaahl", "@TrentHouliston"]
 
@@ -37,18 +41,18 @@ FILTER_MODES = {
 SetTotalPulsesAction = pulse_meter_ns.class_("SetTotalPulsesAction", automation.Action)
 
 
-def validate_internal_filter(value):
+def validate_internal_filter(value: Any) -> TimePeriodMicroseconds:
     return cv.positive_time_period_microseconds(value)
 
 
-def validate_timeout(value):
+def validate_timeout(value: Any) -> TimePeriodMicroseconds:
     value = cv.positive_time_period_microseconds(value)
     if value.total_minutes > 70:
         raise cv.Invalid("Maximum timeout is 70 minutes")
     return value
 
 
-def validate_pulse_meter_pin(value):
+def validate_pulse_meter_pin(value: Any) -> ConfigType:
     value = pins.internal_gpio_input_pin_schema(value)
     if CORE.is_esp8266 and value[CONF_NUMBER] >= 16:
         raise cv.Invalid(
@@ -81,7 +85,7 @@ CONFIG_SCHEMA = sensor.sensor_schema(
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     var = await sensor.new_sensor(config)
     await cg.register_component(var, config)
 
@@ -107,7 +111,12 @@ async def to_code(config):
     ),
     synchronous=True,
 )
-async def set_total_action_to_code(config, action_id, template_arg, args):
+async def set_total_action_to_code(
+    config: ConfigType,
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
     paren = await cg.get_variable(config[CONF_ID])
     var = cg.new_Pvariable(action_id, template_arg, paren)
     template_ = await cg.templatable(config[CONF_VALUE], args, cg.uint32)

esphome/components/shelly_dimmer/light.py

@@ -1,6 +1,7 @@
 import hashlib
 from pathlib import Path
 import re
+from typing import Any
 
 from esphome import external_files, pins
 import esphome.codegen as cg
@@ -66,7 +67,7 @@ KNOWN_FIRMWARE = {
 }
 
 
-def parse_firmware_version(value):
+def parse_firmware_version(value: str) -> tuple[int, int]:
     match = re.fullmatch(r"(\d+)\.(\d+)", value)
     if match is None:
         raise ValueError(f"Not a valid version number {value}")
@@ -154,7 +155,7 @@ def _extract_firmware_ref(entry: ConfigType) -> RemoteFile | None:
 PREFETCH_FILES = external_files.single_stage_prefetch(_extract_firmware_ref)
 
 
-def validate_firmware(value):
+def validate_firmware(value: ConfigType) -> ConfigType:
     config = value.copy()
     if CONF_URL not in config:
         try:
@@ -167,14 +168,14 @@ def validate_firmware(value):
     return config
 
 
-def validate_sha256(value):
+def validate_sha256(value: Any) -> str:
     value = cv.string(value)
     if not re.fullmatch(r"[0-9a-fA-F]{64}", value):
         raise ValueError(f"Not a valid SHA256 hex string: {value}")
     return value
 
 
-def validate_version(value):
+def validate_version(value: str) -> str:
     parse_firmware_version(value)
     return value
 
@@ -231,7 +232,7 @@ FINAL_VALIDATE_SCHEMA = uart.final_validate_device_schema(
 )
 
 
-async def to_code(config):
+async def to_code(config: ConfigType) -> None:
     fw_hex = get_firmware(config[CONF_FIRMWARE])
     fw_major, fw_minor = parse_firmware_version(config[CONF_FIRMWARE][CONF_VERSION])
 

esphome/components/web_server/list_entities.h

@@ -35,6 +35,7 @@ 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_();
-  // One step per loop; refusals retry next pass
-  this->entities_iterator_.try_advance(1);
+  if (!this->entities_iterator_.completed())
+    this->entities_iterator_.advance();
 }
 
 void DeferredUpdateEventSource::deferrable_send_state(void *source, const char *event_type,
@@ -321,6 +321,12 @@ 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_();
-  // One step per loop; refusals retry next pass
-  this->entities_iterator_.try_advance(1);
+  if (!this->entities_iterator_.completed())
+    this->entities_iterator_.advance();
 }
 
 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;
 }
 
-bool ComponentIterator::advance_step_() {
+void ComponentIterator::advance() {
   switch (this->state_) {
     case IteratorState::NONE:
       // not started
-      return false;
+      return;
     case IteratorState::BEGIN:
       if (this->on_begin()) {
         advance_platform_();
-        return true;
       }
-      return false;
+      break;
 
 // Entity iterator cases (generated from entity_types.h)
 // NOLINTBEGIN(bugprone-macro-parentheses)
 #define ENTITY_TYPE_(type, singular, plural, count, upper) \
   case IteratorState::upper: \
-    return this->process_platform_item_(App.get_##plural(), &ComponentIterator::on_##singular);
+    this->process_platform_item_(App.get_##plural(), &ComponentIterator::on_##singular); \
+    break;
 #define ENTITY_CONTROLLER_TYPE_(type, singular, plural, count, upper, callback) \
   ENTITY_TYPE_(type, singular, plural, count, upper)
 #include "esphome/core/entity_types.h"
@@ -48,29 +48,26 @@ bool ComponentIterator::advance_step_() {
 
 #ifdef USE_API_USER_DEFINED_ACTIONS
     case IteratorState::SERVICE:
-      return this->process_platform_item_(api::global_api_server->get_user_services(), &ComponentIterator::on_service);
+      this->process_platform_item_(api::global_api_server->get_user_services(), &ComponentIterator::on_service);
+      break;
 #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)) {
-        return false;
+      if (camera_instance != nullptr && (!camera_instance->is_internal() || this->include_internal_)) {
+        this->on_camera(camera_instance);
       }
       advance_platform_();
-      return true;
-    }
+    } break;
 #endif
 
     case IteratorState::MAX:
       if (this->on_end()) {
         this->state_ = IteratorState::NONE;
-        return true;
       }
-      return false;
+      return;
   }
-  return false;
 }
 
 bool ComponentIterator::on_end() { return true; }

esphome/core/component_iterator.h

@@ -30,23 +30,7 @@ class RadioFrequency;
 class ComponentIterator {
  public:
   void begin(bool include_internal = false);
-  /// 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); }
+  void advance();
   bool completed() const { return this->state_ == IteratorState::NONE; }
   virtual bool on_begin();
 // Pure virtual entity callbacks (generated from entity_types.h)
@@ -89,34 +73,23 @@ 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 yield_requested_ : 1 {false};
-  bool include_internal_ : 1 {false};
+  bool include_internal_{false};
 
   template<typename Container>
-  bool process_platform_item_(const Container &items,
+  void process_platform_item_(const Container &items,
                               bool (ComponentIterator::*on_item)(typename Container::value_type)) {
     if (this->at_ >= items.size()) {
       this->advance_platform_();
-      return true;
+    } else {
+      typename Container::value_type item = items[this->at_];
+      if ((item->is_internal() && !this->include_internal_) || (this->*on_item)(item)) {
+        this->at_++;
+      }
     }
-    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_();
 };
 

requirements.txt

@@ -12,7 +12,7 @@ pyserial==3.5
 platformio==6.1.19
 esptool==5.3.1
 click==8.3.3
-aioesphomeapi==45.12.0
+aioesphomeapi==45.13.1
 aiohappyeyeballs==2.7.1  # Happy Eyeballs for requests downloads; already pulled in by aioesphomeapi
 zeroconf==0.150.0
 puremagic==2.2.0

tests/component_tests/emontx/test_sensor_defaults.py

@@ -0,0 +1,100 @@
+"""Tests for emontx sensor tag defaults."""
+
+import pytest
+
+from esphome.components import sensor
+from esphome.components.emontx.sensor import CONFIG_SCHEMA, apply_tag_defaults
+from esphome.const import (
+    CONF_ACCURACY_DECIMALS,
+    CONF_STATE_CLASS,
+    STATE_CLASS_MEASUREMENT,
+    STATE_CLASS_TOTAL_INCREASING,
+)
+
+
+def _resolve_via_config_schema(tag: str) -> dict:
+    """Run a minimal config through the real CONFIG_SCHEMA pipeline, the
+    same path a user's YAML goes through."""
+    return CONFIG_SCHEMA(
+        {"tag_name": tag, "emontx_id": "my_emontx", "name": f"{tag} sensor"}
+    )
+
+
+def test_config_schema_applies_tag_default_state_class():
+    """If sensor_schema(state_class=...) is reintroduced, the schema-level
+    default wins over apply_tag_defaults' per-prefix value, and E1 would
+    resolve to measurement instead of total_increasing. Driving the real
+    CONFIG_SCHEMA (not just apply_tag_defaults) catches that, since
+    sensor_schema() runs before apply_tag_defaults in the cv.All() chain.
+    """
+    result = _resolve_via_config_schema("E1")
+    assert result[CONF_STATE_CLASS] == sensor.validate_state_class(
+        STATE_CLASS_TOTAL_INCREASING
+    )
+
+
+def test_config_schema_applies_tag_default_accuracy_decimals():
+    """Same root cause as the state_class regression: reintroducing
+    sensor_schema(accuracy_decimals=...) would make V1 resolve to the
+    schema-level default instead of the prefix-specific value of 2.
+    """
+    result = _resolve_via_config_schema("V1")
+    assert result[CONF_ACCURACY_DECIMALS] == 2
+
+
+def _make_config(tag: str) -> dict:
+    """Minimal config dict with only tag_name set — no overrides."""
+    return {"tag_name": tag}
+
+
+@pytest.mark.parametrize(
+    ("tag", "expected_state_class", "expected_decimals"),
+    [
+        # Known numeric-index prefixes
+        ("E1", STATE_CLASS_TOTAL_INCREASING, 0),
+        ("E12", STATE_CLASS_TOTAL_INCREASING, 0),
+        ("P1", STATE_CLASS_MEASUREMENT, 0),
+        ("V1", STATE_CLASS_MEASUREMENT, 2),
+        ("I1", STATE_CLASS_MEASUREMENT, 2),
+        ("T1", STATE_CLASS_MEASUREMENT, 2),
+        # Known patterns
+        ("PULSE1", STATE_CLASS_TOTAL_INCREASING, 0),
+        ("PULSE12", STATE_CLASS_TOTAL_INCREASING, 0),
+        ("PF1", STATE_CLASS_MEASUREMENT, 2),
+        # Unknown / free-form tags fall back to generic defaults
+        ("CUSTOM1", STATE_CLASS_MEASUREMENT, 0),
+        ("X", STATE_CLASS_MEASUREMENT, 0),
+    ],
+)
+def test_apply_tag_defaults(tag, expected_state_class, expected_decimals):
+    """apply_tag_defaults must inject the correct state_class and accuracy_decimals
+    for each tag type when no user overrides are present."""
+    config = _make_config(tag)
+    result = apply_tag_defaults(config)
+
+    assert result[CONF_STATE_CLASS] == sensor.validate_state_class(expected_state_class)
+    assert result[CONF_ACCURACY_DECIMALS] == expected_decimals
+
+
+@pytest.mark.parametrize(
+    ("tag", "user_state_class", "user_decimals"),
+    [
+        # User overrides must not be clobbered by defaults
+        ("E1", STATE_CLASS_MEASUREMENT, 3),
+        ("PULSE1", STATE_CLASS_MEASUREMENT, 1),
+        ("V1", STATE_CLASS_TOTAL_INCREASING, 0),
+        ("CUSTOM1", STATE_CLASS_TOTAL_INCREASING, 4),
+    ],
+)
+def test_apply_tag_defaults_respects_user_overrides(
+    tag, user_state_class, user_decimals
+):
+    """apply_tag_defaults must not overwrite values already set by the user."""
+    config = _make_config(tag)
+    config[CONF_STATE_CLASS] = sensor.validate_state_class(user_state_class)
+    config[CONF_ACCURACY_DECIMALS] = user_decimals
+
+    result = apply_tag_defaults(config)
+
+    assert result[CONF_STATE_CLASS] == sensor.validate_state_class(user_state_class)
+    assert result[CONF_ACCURACY_DECIMALS] == user_decimals

tests/component_tests/esp32_ble_tracker/config/scan_window_explicit.yaml

@@ -0,0 +1,19 @@
+esphome:
+  name: scan-window-explicit
+
+esp32:
+  board: esp32dev
+  framework:
+    type: esp-idf
+
+wifi:
+  ssid: MySSID
+
+esp32_ble_tracker:
+  scan_parameters:
+    window: 30ms
+
+bluetooth_proxy:
+  active: true
+
+api:

tests/component_tests/esp32_ble_tracker/config/scan_window_raised.yaml

@@ -0,0 +1,17 @@
+esphome:
+  name: scan-window-raised
+
+esp32:
+  board: esp32dev
+  framework:
+    type: esp-idf
+
+wifi:
+  ssid: MySSID
+
+esp32_ble_tracker:
+
+bluetooth_proxy:
+  active: true
+
+api:

tests/component_tests/esp32_ble_tracker/config/scan_window_scan_only.yaml

@@ -0,0 +1,12 @@
+esphome:
+  name: scan-window-scan-only
+
+esp32:
+  board: esp32dev
+  framework:
+    type: esp-idf
+
+wifi:
+  ssid: MySSID
+
+esp32_ble_tracker:

tests/component_tests/esp32_ble_tracker/config/scan_window_user_set_scan_only.yaml

@@ -0,0 +1,14 @@
+esphome:
+  name: scan-window-user-scan-only
+
+esp32:
+  board: esp32dev
+  framework:
+    type: esp-idf
+
+wifi:
+  ssid: MySSID
+
+esp32_ble_tracker:
+  scan_parameters:
+    connection_scan_window: 20ms

tests/component_tests/esp32_ble_tracker/test_scan_window_default.py

@@ -12,11 +12,12 @@ arbiter a full-duty scan would starve wifi, so the 30 ms default is kept.
 from __future__ import annotations
 
 from collections.abc import Callable
+from pathlib import Path
 
 import pytest
 
 from esphome import config_validation as cv
-from esphome.components.ble_device_base import to_ble_units
+from esphome.components.ble_device_base import CONF_CONNECTION_SCAN_WINDOW, to_ble_units
 from esphome.components.const import CONF_SCAN_PARAMETERS, CONF_WINDOW
 from esphome.components.esp32 import KEY_IDF_VERSION
 from esphome.components.esp32_ble_tracker import (
@@ -120,3 +121,103 @@ def test_short_interval_without_window_still_rejected(
     stage_esp32("5.5.5", wifi=True)
     with pytest.raises(cv.Invalid, match="needs to be smaller than scan interval"):
         _scan_params({"scan_parameters": {"interval": "20ms"}})
+
+
+# The connection-time fallback window: while a GATT connection is active the
+# scanner drops from a raised full-duty window back to this value so the
+# connection gets guaranteed airtime.
+
+
+def test_raise_arms_connection_scan_window_default(
+    stage_esp32: Callable[..., None],
+) -> None:
+    stage_esp32("5.5.5", wifi=True)
+    params = _scan_params({})
+    assert params[CONF_WINDOW] == params[CONF_INTERVAL]
+    assert to_ble_units(params[CONF_CONNECTION_SCAN_WINDOW]) == 48
+
+
+def test_user_connection_scan_window_survives_raise(
+    stage_esp32: Callable[..., None],
+) -> None:
+    stage_esp32("5.5.5", wifi=True)
+    params = _scan_params({"scan_parameters": {"connection_scan_window": "60ms"}})
+    assert params[CONF_WINDOW] == params[CONF_INTERVAL]
+    assert to_ble_units(params[CONF_CONNECTION_SCAN_WINDOW]) == 96
+
+
+def test_unraised_window_gets_no_connection_scan_window_default(
+    stage_esp32: Callable[..., None],
+) -> None:
+    stage_esp32("5.5.4", wifi=True)
+    assert CONF_CONNECTION_SCAN_WINDOW not in _scan_params({})
+
+
+def test_connection_scan_window_above_interval_rejected(
+    stage_esp32: Callable[..., None],
+) -> None:
+    stage_esp32("5.5.5", wifi=True)
+    with pytest.raises(
+        cv.Invalid, match="connection_scan_window .* needs to be smaller"
+    ):
+        _scan_params({"scan_parameters": {"connection_scan_window": "400ms"}})
+
+
+def test_connection_scan_window_above_window_rejected(
+    stage_esp32: Callable[..., None],
+) -> None:
+    """A connection window above the (post-raise) window would widen the scan
+    during connections; the reject runs after the raise so a fallback below a
+    raised window still validates (covered by the survives-raise test)."""
+    stage_esp32("5.5.5", wifi=True)
+    with pytest.raises(
+        cv.Invalid, match="connection_scan_window .* needs to be smaller"
+    ):
+        _scan_params(
+            {"scan_parameters": {"window": "30ms", "connection_scan_window": "300ms"}}
+        )
+
+
+def test_connection_scan_window_truncation_collapse_rejected(
+    stage_esp32: Callable[..., None],
+) -> None:
+    """A connection window that truncates into the interval's 0.625 ms unit
+    would silently program a full-duty scan during connections."""
+    stage_esp32("5.5.5", wifi=True)
+    with pytest.raises(cv.Invalid, match="connection_scan_window .* both truncate"):
+        _scan_params(
+            {
+                "scan_parameters": {
+                    "interval": "320.5ms",
+                    "connection_scan_window": "320.2ms",
+                }
+            }
+        )
+
+
+@pytest.mark.parametrize(
+    ("config_file", "window_call", "connection_call", "warns"),
+    [
+        # Raised window with GATT clients: the injected fallback is emitted.
+        ("scan_window_raised.yaml", "set_scan_window(512)", True, False),
+        # Explicit window: nothing injected.
+        ("scan_window_explicit.yaml", "set_scan_window(48)", False, False),
+        # Scan-only build compiles the path out: the injected default is
+        # dropped silently, a user-set value warns.
+        ("scan_window_scan_only.yaml", "set_scan_window(512)", False, False),
+        ("scan_window_user_set_scan_only.yaml", "set_scan_window(512)", False, True),
+    ],
+)
+def test_connection_scan_window_codegen(
+    generate_main: Callable[[str | Path], str],
+    component_config_path: Callable[[str], Path],
+    caplog: pytest.LogCaptureFixture,
+    config_file: str,
+    window_call: str,
+    connection_call: bool,
+    warns: bool,
+) -> None:
+    main_cpp = generate_main(component_config_path(config_file))
+    assert window_call in main_cpp
+    assert ("set_connection_scan_window(48)" in main_cpp) == connection_call
+    assert ("'connection_scan_window' has no effect" in caplog.text) == warns

tests/components/camera/__init__.py

@@ -1,11 +0,0 @@
-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

@@ -1,79 +0,0 @@
-#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

@@ -1,11 +0,0 @@
-# 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

@@ -1,195 +0,0 @@
-#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/components/emontx/test.esp32-idf.yaml

@@ -1,4 +1,3 @@
 packages:
   uart_115200: !include ../../test_build_components/common/uart_115200/esp32-idf.yaml
-
-<<: !include common.yaml
+  emontx: !include common.yaml

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

@@ -1,4 +1,3 @@
 packages:
   uart_115200: !include ../../test_build_components/common/uart_115200/esp8266-ard.yaml
-
-<<: !include common.yaml
+  emontx: !include common.yaml

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

@@ -1,4 +1,3 @@
 packages:
   uart_115200: !include ../../test_build_components/common/uart_115200/rp2040-ard.yaml
-
-<<: !include common.yaml
+  emontx: !include common.yaml

tests/components/emontx/validate.esp32-idf.yaml

@@ -0,0 +1,73 @@
+packages:
+  uart_115200: !include ../../test_build_components/common/uart_115200/esp32-idf.yaml
+  emontx: !include common.yaml
+
+# Validate that each sensor type gets the correct default state_class,
+# unit_of_measurement, device_class, and accuracy_decimals when NO overrides
+# are provided. The values are intentionally omitted so apply_tag_defaults is
+# exercised, not the user-override path.
+
+sensor:
+  # Energy sensor (E prefix): expects state_class=total_increasing, unit=Wh,
+  # device_class=energy, accuracy_decimals=0
+  - platform: emontx
+    tag_name: E1
+    name: Energy 1
+    emontx_id: test_emontx
+
+  # Power sensor (P prefix): expects state_class=measurement, unit=W,
+  # device_class=power, accuracy_decimals=0
+  - platform: emontx
+    tag_name: P1
+    name: Power 1
+    emontx_id: test_emontx
+
+  # Voltage sensor (V prefix): expects state_class=measurement, unit=V,
+  # device_class=voltage, accuracy_decimals=2
+  - platform: emontx
+    tag_name: V1
+    name: Voltage 1
+    emontx_id: test_emontx
+
+  # Current sensor (I prefix): expects state_class=measurement, unit=A,
+  # device_class=current, accuracy_decimals=2
+  - platform: emontx
+    tag_name: I1
+    name: Current 1
+    emontx_id: test_emontx
+
+  # Temperature sensor (T prefix): expects state_class=measurement, unit=°C,
+  # device_class=temperature, accuracy_decimals=2
+  - platform: emontx
+    tag_name: T1
+    name: Temperature 1
+    emontx_id: test_emontx
+
+  # Pulse sensor (PULSE pattern): expects state_class=total_increasing,
+  # unit=pulses, device_class=energy, accuracy_decimals=0
+  - platform: emontx
+    tag_name: PULSE1
+    name: Pulse 1
+    emontx_id: test_emontx
+
+  # Power factor sensor (PF pattern): expects state_class=measurement,
+  # device_class=power_factor, accuracy_decimals=2
+  - platform: emontx
+    tag_name: PF1
+    name: Power Factor 1
+    emontx_id: test_emontx
+
+  # Unknown tag: no prefix match, falls back to state_class=measurement,
+  # accuracy_decimals=0
+  - platform: emontx
+    tag_name: CUSTOM1
+    name: Custom sensor
+    emontx_id: test_emontx
+
+  # User override: verify that explicit values are respected and not clobbered
+  - platform: emontx
+    tag_name: E2
+    name: Energy 2 (user override)
+    emontx_id: test_emontx
+    state_class: measurement
+    accuracy_decimals: 3

tests/components/esp32_hosted/test-embedded.esp32-p4-idf.yaml

@@ -6,3 +6,6 @@ update:
     type: embedded
     path: $component_dir/test_firmware.bin
     sha256: de2f256064a0af797747c2b97505dc0b9f3df0de4f489eac731c23ae9ca9cc31
+    on_update_available:
+      then:
+        - logger.log: "Coprocessor update available"

tests/components/esp32_hosted/test-http.esp32-p4-idf.yaml

@@ -8,3 +8,6 @@ update:
     type: http
     source: https://esphome.github.io/esp-hosted-firmware/manifest/esp32c6.json
     update_interval: 6h
+    on_update_available:
+      then:
+        - logger.log: "Coprocessor update available"

tests/components/sendspin/common-action.yaml

@@ -1,6 +1,6 @@
 # `sendspin.switch` action enables the controller role, so we use a standalone test
 packages:
-  base: !include common.yaml
+  sendspin: !include common.yaml
 
 wifi:
   on_connect:

tests/components/sendspin/common-ethernet.yaml

@@ -0,0 +1,5 @@
+packages:
+  sendspin_hub: !include common-hub.yaml
+
+ethernet:
+  type: OPENETH

tests/components/sendspin/common-hub.yaml

@@ -0,0 +1,6 @@
+psram:
+  mode: quad
+
+sendspin:
+  id: sendspin_hub_id
+  task_stack_in_psram: true

tests/components/sendspin/common-media_player.yaml

@@ -1,4 +1,5 @@
-<<: !include common.yaml
+packages:
+  sendspin: !include common.yaml
 
 media_player:
   - platform: sendspin

tests/components/sendspin/common-media_source.yaml

@@ -1,4 +1,5 @@
-<<: !include common.yaml
+packages:
+  sendspin: !include common.yaml
 
 media_source:
   - platform: sendspin

tests/components/sendspin/common-sensor.yaml

@@ -1,4 +1,5 @@
-<<: !include common.yaml
+packages:
+  sendspin: !include common.yaml
 
 sensor:
   - platform: sendspin

tests/components/sendspin/common-text_sensor.yaml

@@ -1,4 +1,5 @@
-<<: !include common.yaml
+packages:
+  sendspin: !include common.yaml
 
 text_sensor:
   - platform: sendspin

tests/components/sendspin/common.yaml

@@ -1,9 +1,5 @@
+packages:
+  sendspin_hub: !include common-hub.yaml
+
 wifi:
   ap:
-
-psram:
-  mode: quad
-
-sendspin:
-  id: sendspin_hub_id
-  task_stack_in_psram: true

tests/components/sendspin/test-action.esp32-idf.yaml

@@ -1 +1,2 @@
-<<: !include common-action.yaml
+packages:
+  sendspin: !include common-action.yaml

tests/components/sendspin/test-ethernet.esp32-idf.yaml

@@ -1,9 +1,2 @@
-ethernet:
-  type: OPENETH
-
-psram:
-  mode: quad
-
-sendspin:
-  id: sendspin_hub_id
-  task_stack_in_psram: true
+packages:
+  sendspin: !include common-ethernet.yaml

tests/components/sendspin/test-media_player.esp32-idf.yaml

@@ -1 +1,2 @@
-<<: !include common-media_player.yaml
+packages:
+  sendspin: !include common-media_player.yaml

tests/components/sendspin/test-media_source.esp32-idf.yaml

@@ -1 +1,2 @@
-<<: !include common-media_source.yaml
+packages:
+  sendspin: !include common-media_source.yaml

tests/components/sendspin/test-sensor.esp32-idf.yaml

@@ -1 +1,2 @@
-<<: !include common-sensor.yaml
+packages:
+  sendspin: !include common-sensor.yaml

tests/components/sendspin/test-text_sensor.esp32-idf.yaml

@@ -1 +1,2 @@
-<<: !include common-text_sensor.yaml
+packages:
+  sendspin: !include common-text_sensor.yaml

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

@@ -1 +1,2 @@
-<<: !include common.yaml
+packages:
+  sendspin: !include common.yaml

tests/integration/README.md

@@ -7,7 +7,6 @@ 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
@@ -348,7 +347,6 @@ 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

@@ -1,23 +0,0 @@
-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

@@ -1,20 +0,0 @@
-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

@@ -1,55 +0,0 @@
-#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

@@ -1,21 +0,0 @@
-#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

@@ -1,148 +0,0 @@
-"""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

@@ -1,110 +0,0 @@
-"""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