Merge branch 'scheduler_pool_v2' into integration

esphome/components/alarm_control_panel/__init__.py

@@ -13,7 +13,7 @@ from esphome.const import (
     CONF_TRIGGER_ID,
     CONF_WEB_SERVER,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 from esphome.core.entity_helpers import entity_duplicate_validator, setup_entity
 from esphome.cpp_generator import MockObjClass
 
@@ -345,6 +345,6 @@ async def alarm_control_panel_is_armed_to_code(
     return cg.new_Pvariable(condition_id, template_arg, paren)
 
 
-@coroutine_with_priority(100.0)
+@coroutine_with_priority(CoroPriority.CORE)
 async def to_code(config):
     cg.add_global(alarm_control_panel_ns.using)

esphome/components/api/__init__.py

@@ -24,7 +24,7 @@ from esphome.const import (
     CONF_TRIGGER_ID,
     CONF_VARIABLES,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 
 DOMAIN = "api"
 DEPENDENCIES = ["network"]
@@ -136,7 +136,7 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-@coroutine_with_priority(40.0)
+@coroutine_with_priority(CoroPriority.WEB)
 async def to_code(config):
     var = cg.new_Pvariable(config[CONF_ID])
     await cg.register_component(var, config)

esphome/components/async_tcp/__init__.py

@@ -8,7 +8,7 @@ from esphome.const import (
     PLATFORM_LN882X,
     PLATFORM_RTL87XX,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 
 CODEOWNERS = ["@esphome/core"]
 
@@ -27,7 +27,7 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-@coroutine_with_priority(200.0)
+@coroutine_with_priority(CoroPriority.NETWORK_TRANSPORT)
 async def to_code(config):
     if CORE.is_esp32 or CORE.is_libretiny:
         # https://github.com/ESP32Async/AsyncTCP

esphome/components/audio_adc/__init__.py

@@ -2,7 +2,7 @@ from esphome import automation
 import esphome.codegen as cg
 import esphome.config_validation as cv
 from esphome.const import CONF_ID, CONF_MIC_GAIN
-from esphome.core import coroutine_with_priority
+from esphome.core import CoroPriority, coroutine_with_priority
 
 CODEOWNERS = ["@kbx81"]
 IS_PLATFORM_COMPONENT = True
@@ -35,7 +35,7 @@ async def audio_adc_set_mic_gain_to_code(config, action_id, template_arg, args):
     return var
 
 
-@coroutine_with_priority(100.0)
+@coroutine_with_priority(CoroPriority.CORE)
 async def to_code(config):
     cg.add_define("USE_AUDIO_ADC")
     cg.add_global(audio_adc_ns.using)

esphome/components/audio_dac/__init__.py

@@ -3,7 +3,7 @@ from esphome.automation import maybe_simple_id
 import esphome.codegen as cg
 import esphome.config_validation as cv
 from esphome.const import CONF_ID, CONF_VOLUME
-from esphome.core import coroutine_with_priority
+from esphome.core import CoroPriority, coroutine_with_priority
 
 CODEOWNERS = ["@kbx81"]
 IS_PLATFORM_COMPONENT = True
@@ -51,7 +51,7 @@ async def audio_dac_set_volume_to_code(config, action_id, template_arg, args):
     return var
 
 
-@coroutine_with_priority(100.0)
+@coroutine_with_priority(CoroPriority.CORE)
 async def to_code(config):
     cg.add_define("USE_AUDIO_DAC")
     cg.add_global(audio_dac_ns.using)

esphome/components/binary_sensor/__init__.py

@@ -59,7 +59,7 @@ from esphome.const import (
     DEVICE_CLASS_VIBRATION,
     DEVICE_CLASS_WINDOW,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 from esphome.core.entity_helpers import entity_duplicate_validator, setup_entity
 from esphome.cpp_generator import MockObjClass
 from esphome.util import Registry
@@ -652,7 +652,7 @@ async def binary_sensor_is_off_to_code(config, condition_id, template_arg, args)
     return cg.new_Pvariable(condition_id, template_arg, paren, False)
 
 
-@coroutine_with_priority(100.0)
+@coroutine_with_priority(CoroPriority.CORE)
 async def to_code(config):
     cg.add_global(binary_sensor_ns.using)
 

esphome/components/button/__init__.py

@@ -17,7 +17,7 @@ from esphome.const import (
     DEVICE_CLASS_RESTART,
     DEVICE_CLASS_UPDATE,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 from esphome.core.entity_helpers import entity_duplicate_validator, setup_entity
 from esphome.cpp_generator import MockObjClass
 
@@ -134,6 +134,6 @@ async def button_press_to_code(config, action_id, template_arg, args):
     return cg.new_Pvariable(action_id, template_arg, paren)
 
 
-@coroutine_with_priority(100.0)
+@coroutine_with_priority(CoroPriority.CORE)
 async def to_code(config):
     cg.add_global(button_ns.using)

esphome/components/captive_portal/__init__.py

@@ -10,7 +10,7 @@ from esphome.const import (
     PLATFORM_LN882X,
     PLATFORM_RTL87XX,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 
 AUTO_LOAD = ["web_server_base", "ota.web_server"]
 DEPENDENCIES = ["wifi"]
@@ -40,7 +40,7 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-@coroutine_with_priority(64.0)
+@coroutine_with_priority(CoroPriority.COMMUNICATION)
 async def to_code(config):
     paren = await cg.get_variable(config[CONF_WEB_SERVER_BASE_ID])
 

esphome/components/climate/__init__.py

@@ -47,7 +47,7 @@ from esphome.const import (
     CONF_VISUAL,
     CONF_WEB_SERVER,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 from esphome.core.entity_helpers import entity_duplicate_validator, setup_entity
 from esphome.cpp_generator import MockObjClass
 
@@ -517,6 +517,6 @@ async def climate_control_to_code(config, action_id, template_arg, args):
     return var
 
 
-@coroutine_with_priority(100.0)
+@coroutine_with_priority(CoroPriority.CORE)
 async def to_code(config):
     cg.add_global(climate_ns.using)

esphome/components/cover/__init__.py

@@ -32,7 +32,7 @@ from esphome.const import (
     DEVICE_CLASS_SHUTTER,
     DEVICE_CLASS_WINDOW,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 from esphome.core.entity_helpers import entity_duplicate_validator, setup_entity
 from esphome.cpp_generator import MockObjClass
 
@@ -263,6 +263,6 @@ async def cover_control_to_code(config, action_id, template_arg, args):
     return var
 
 
-@coroutine_with_priority(100.0)
+@coroutine_with_priority(CoroPriority.CORE)
 async def to_code(config):
     cg.add_global(cover_ns.using)

esphome/components/datetime/__init__.py

@@ -21,7 +21,7 @@ from esphome.const import (
     CONF_WEB_SERVER,
     CONF_YEAR,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 from esphome.core.entity_helpers import entity_duplicate_validator, setup_entity
 from esphome.cpp_generator import MockObjClass
 
@@ -172,7 +172,7 @@ async def new_datetime(config, *args):
     return var
 
 
-@coroutine_with_priority(100.0)
+@coroutine_with_priority(CoroPriority.CORE)
 async def to_code(config):
     cg.add_global(datetime_ns.using)
 

esphome/components/display/__init__.py

@@ -15,7 +15,7 @@ from esphome.const import (
     CONF_UPDATE_INTERVAL,
     SCHEDULER_DONT_RUN,
 )
-from esphome.core import coroutine_with_priority
+from esphome.core import CoroPriority, coroutine_with_priority
 
 IS_PLATFORM_COMPONENT = True
 
@@ -218,7 +218,7 @@ async def display_is_displaying_page_to_code(config, condition_id, template_arg,
     return var
 
 
-@coroutine_with_priority(100.0)
+@coroutine_with_priority(CoroPriority.CORE)
 async def to_code(config):
     cg.add_global(display_ns.using)
     cg.add_define("USE_DISPLAY")

esphome/components/esp32/__init__.py

@@ -855,11 +855,6 @@ async def to_code(config):
 
         cg.add_platformio_option("platform_packages", [conf[CONF_SOURCE]])
 
-        # platformio/toolchain-esp32ulp does not support linux_aarch64 yet and has not been updated for over 2 years
-        # This is espressif's own published version which is more up to date.
-        cg.add_platformio_option(
-            "platform_packages", ["espressif/toolchain-esp32ulp@2.35.0-20220830"]
-        )
         add_idf_sdkconfig_option(f"CONFIG_IDF_TARGET_{variant}", True)
         add_idf_sdkconfig_option(
             f"CONFIG_ESPTOOLPY_FLASHSIZE_{config[CONF_FLASH_SIZE]}", True

esphome/components/esp32_ble_tracker/__init__.py

@@ -30,7 +30,7 @@ from esphome.const import (
     CONF_SERVICE_UUID,
     CONF_TRIGGER_ID,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 from esphome.enum import StrEnum
 from esphome.types import ConfigType
 
@@ -368,7 +368,7 @@ async def to_code(config):
 # This needs to be run as a job with very low priority so that all components have
 # chance to call register_ble_tracker and register_client before the list is checked
 # and added to the global defines list.
-@coroutine_with_priority(-1000)
+@coroutine_with_priority(CoroPriority.FINAL)
 async def _add_ble_features():
     # Add feature-specific defines based on what's needed
     if BLEFeatures.ESP_BT_DEVICE in _required_features:

esphome/components/esp8266/__init__.py

@@ -17,7 +17,7 @@ from esphome.const import (
     PLATFORM_ESP8266,
     ThreadModel,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 from esphome.helpers import copy_file_if_changed
 
 from .boards import BOARDS, ESP8266_LD_SCRIPTS
@@ -176,7 +176,7 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-@coroutine_with_priority(1000)
+@coroutine_with_priority(CoroPriority.PLATFORM)
 async def to_code(config):
     cg.add(esp8266_ns.setup_preferences())
 

esphome/components/esp8266/gpio.py

@@ -17,7 +17,7 @@ from esphome.const import (
     CONF_PULLUP,
     PLATFORM_ESP8266,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 
 from . import boards
 from .const import KEY_BOARD, KEY_ESP8266, KEY_PIN_INITIAL_STATES, esp8266_ns
@@ -188,7 +188,7 @@ async def esp8266_pin_to_code(config):
     return var
 
 
-@coroutine_with_priority(-999.0)
+@coroutine_with_priority(CoroPriority.WORKAROUNDS)
 async def add_pin_initial_states_array():
     # Add includes at the very end, so that they override everything
     initial_states: list[PinInitialState] = CORE.data[KEY_ESP8266][

esphome/components/esphome/ota/__init__.py

@@ -16,7 +16,7 @@ from esphome.const import (
     CONF_SAFE_MODE,
     CONF_VERSION,
 )
-from esphome.core import coroutine_with_priority
+from esphome.core import CoroPriority, coroutine_with_priority
 import esphome.final_validate as fv
 
 _LOGGER = logging.getLogger(__name__)
@@ -121,7 +121,7 @@ CONFIG_SCHEMA = (
 FINAL_VALIDATE_SCHEMA = ota_esphome_final_validate
 
 
-@coroutine_with_priority(52.0)
+@coroutine_with_priority(CoroPriority.COMMUNICATION)
 async def to_code(config):
     var = cg.new_Pvariable(config[CONF_ID])
     cg.add(var.set_port(config[CONF_PORT]))

esphome/components/ethernet/__init__.py

@@ -38,7 +38,12 @@ from esphome.const import (
     KEY_CORE,
     KEY_FRAMEWORK_VERSION,
 )
-from esphome.core import CORE, TimePeriodMilliseconds, coroutine_with_priority
+from esphome.core import (
+    CORE,
+    CoroPriority,
+    TimePeriodMilliseconds,
+    coroutine_with_priority,
+)
 import esphome.final_validate as fv
 
 CONFLICTS_WITH = ["wifi"]
@@ -289,7 +294,7 @@ def phy_register(address: int, value: int, page: int):
     )
 
 
-@coroutine_with_priority(60.0)
+@coroutine_with_priority(CoroPriority.COMMUNICATION)
 async def to_code(config):
     var = cg.new_Pvariable(config[CONF_ID])
     await cg.register_component(var, config)

esphome/components/event/__init__.py

@@ -17,7 +17,7 @@ from esphome.const import (
     DEVICE_CLASS_EMPTY,
     DEVICE_CLASS_MOTION,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 from esphome.core.entity_helpers import entity_duplicate_validator, setup_entity
 from esphome.cpp_generator import MockObjClass
 
@@ -143,6 +143,6 @@ async def event_fire_to_code(config, action_id, template_arg, args):
     return var
 
 
-@coroutine_with_priority(100.0)
+@coroutine_with_priority(CoroPriority.CORE)
 async def to_code(config):
     cg.add_global(event_ns.using)

esphome/components/fan/__init__.py

@@ -31,7 +31,7 @@ from esphome.const import (
     CONF_TRIGGER_ID,
     CONF_WEB_SERVER,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 from esphome.core.entity_helpers import entity_duplicate_validator, setup_entity
 
 IS_PLATFORM_COMPONENT = True
@@ -398,6 +398,6 @@ async def fan_is_on_off_to_code(config, condition_id, template_arg, args):
     return cg.new_Pvariable(condition_id, template_arg, paren)
 
 
-@coroutine_with_priority(100.0)
+@coroutine_with_priority(CoroPriority.CORE)
 async def to_code(config):
     cg.add_global(fan_ns.using)

esphome/components/globals/__init__.py

@@ -8,7 +8,7 @@ from esphome.const import (
     CONF_TYPE,
     CONF_VALUE,
 )
-from esphome.core import coroutine_with_priority
+from esphome.core import CoroPriority, coroutine_with_priority
 
 CODEOWNERS = ["@esphome/core"]
 globals_ns = cg.esphome_ns.namespace("globals")
@@ -35,7 +35,7 @@ CONFIG_SCHEMA = cv.Schema(
 
 
 # Run with low priority so that namespaces are registered first
-@coroutine_with_priority(-100.0)
+@coroutine_with_priority(CoroPriority.LATE)
 async def to_code(config):
     type_ = cg.RawExpression(config[CONF_TYPE])
     restore = config[CONF_RESTORE_VALUE]

esphome/components/host/preferences.h

@@ -42,9 +42,10 @@ class HostPreferences : public ESPPreferences {
     if (len > 255)
       return false;
     this->setup_();
-    if (this->data.count(key) == 0)
+    auto it = this->data.find(key);
+    if (it == this->data.end())
       return false;
-    auto vec = this->data[key];
+    const auto &vec = it->second;
     if (vec.size() != len)
       return false;
     memcpy(data, vec.data(), len);

esphome/components/http_request/ota/__init__.py

@@ -3,7 +3,7 @@ import esphome.codegen as cg
 from esphome.components.ota import BASE_OTA_SCHEMA, OTAComponent, ota_to_code
 import esphome.config_validation as cv
 from esphome.const import CONF_ID, CONF_PASSWORD, CONF_URL, CONF_USERNAME
-from esphome.core import coroutine_with_priority
+from esphome.core import CoroPriority, coroutine_with_priority
 
 from .. import CONF_HTTP_REQUEST_ID, HttpRequestComponent, http_request_ns
 
@@ -40,7 +40,7 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-@coroutine_with_priority(52.0)
+@coroutine_with_priority(CoroPriority.COMMUNICATION)
 async def to_code(config):
     var = cg.new_Pvariable(config[CONF_ID])
     await ota_to_code(var, config)

esphome/components/i2c/__init__.py

@@ -18,7 +18,7 @@ from esphome.const import (
     PLATFORM_RP2040,
     PlatformFramework,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 import esphome.final_validate as fv
 
 LOGGER = logging.getLogger(__name__)
@@ -74,7 +74,7 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-@coroutine_with_priority(1.0)
+@coroutine_with_priority(CoroPriority.BUS)
 async def to_code(config):
     cg.add_global(i2c_ns.using)
     cg.add_define("USE_I2C")

esphome/components/json/__init__.py

@@ -1,6 +1,6 @@
 import esphome.codegen as cg
 import esphome.config_validation as cv
-from esphome.core import coroutine_with_priority
+from esphome.core import CoroPriority, coroutine_with_priority
 
 CODEOWNERS = ["@esphome/core"]
 json_ns = cg.esphome_ns.namespace("json")
@@ -10,7 +10,7 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-@coroutine_with_priority(1.0)
+@coroutine_with_priority(CoroPriority.BUS)
 async def to_code(config):
     cg.add_library("bblanchon/ArduinoJson", "7.4.2")
     cg.add_define("USE_JSON")

esphome/components/light/__init__.py

@@ -37,7 +37,7 @@ from esphome.const import (
     CONF_WEB_SERVER,
     CONF_WHITE,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 from esphome.core.entity_helpers import entity_duplicate_validator, setup_entity
 from esphome.cpp_generator import MockObjClass
 
@@ -283,6 +283,6 @@ async def new_light(config, *args):
     return output_var
 
 
-@coroutine_with_priority(100.0)
+@coroutine_with_priority(CoroPriority.CORE)
 async def to_code(config):
     cg.add_global(light_ns.using)

esphome/components/lock/__init__.py

@@ -13,7 +13,7 @@ from esphome.const import (
     CONF_TRIGGER_ID,
     CONF_WEB_SERVER,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 from esphome.core.entity_helpers import entity_duplicate_validator, setup_entity
 from esphome.cpp_generator import MockObjClass
 
@@ -155,6 +155,6 @@ async def lock_is_off_to_code(config, condition_id, template_arg, args):
     return cg.new_Pvariable(condition_id, template_arg, paren, False)
 
 
-@coroutine_with_priority(100.0)
+@coroutine_with_priority(CoroPriority.CORE)
 async def to_code(config):
     cg.add_global(lock_ns.using)

esphome/components/logger/__init__.py

@@ -51,7 +51,7 @@ from esphome.const import (
     PLATFORM_RTL87XX,
     PlatformFramework,
 )
-from esphome.core import CORE, Lambda, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, Lambda, coroutine_with_priority
 
 CODEOWNERS = ["@esphome/core"]
 logger_ns = cg.esphome_ns.namespace("logger")
@@ -275,7 +275,7 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-@coroutine_with_priority(90.0)
+@coroutine_with_priority(CoroPriority.DIAGNOSTICS)
 async def to_code(config):
     baud_rate = config[CONF_BAUD_RATE]
     level = config[CONF_LEVEL]

esphome/components/mdns/__init__.py

@@ -11,7 +11,7 @@ from esphome.const import (
     CONF_SERVICES,
     PlatformFramework,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 
 CODEOWNERS = ["@esphome/core"]
 DEPENDENCIES = ["network"]
@@ -72,7 +72,7 @@ def mdns_service(
     )
 
 
-@coroutine_with_priority(55.0)
+@coroutine_with_priority(CoroPriority.COMMUNICATION)
 async def to_code(config):
     if config[CONF_DISABLED] is True:
         return

esphome/components/media_player/__init__.py

@@ -14,7 +14,7 @@ from esphome.const import (
 )
 from esphome.core import CORE
 from esphome.core.entity_helpers import entity_duplicate_validator, setup_entity
-from esphome.coroutine import coroutine_with_priority
+from esphome.coroutine import CoroPriority, coroutine_with_priority
 from esphome.cpp_generator import MockObjClass
 
 CODEOWNERS = ["@jesserockz"]
@@ -303,7 +303,7 @@ async def media_player_volume_set_action(config, action_id, template_arg, args):
     return var
 
 
-@coroutine_with_priority(100.0)
+@coroutine_with_priority(CoroPriority.CORE)
 async def to_code(config):
     cg.add_global(media_player_ns.using)
     cg.add_define("USE_MEDIA_PLAYER")

esphome/components/microphone/__init__.py

@@ -12,7 +12,7 @@ from esphome.const import (
     CONF_TRIGGER_ID,
 )
 from esphome.core import CORE
-from esphome.coroutine import coroutine_with_priority
+from esphome.coroutine import CoroPriority, coroutine_with_priority
 
 AUTO_LOAD = ["audio"]
 CODEOWNERS = ["@jesserockz", "@kahrendt"]
@@ -213,7 +213,7 @@ automation.register_condition(
 )(microphone_action)
 
 
-@coroutine_with_priority(100.0)
+@coroutine_with_priority(CoroPriority.CORE)
 async def to_code(config):
     cg.add_global(microphone_ns.using)
     cg.add_define("USE_MICROPHONE")

esphome/components/mqtt/__init__.py

@@ -57,7 +57,7 @@ from esphome.const import (
     PLATFORM_ESP8266,
     PlatformFramework,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 
 DEPENDENCIES = ["network"]
 
@@ -321,7 +321,7 @@ def exp_mqtt_message(config):
     )
 
 
-@coroutine_with_priority(40.0)
+@coroutine_with_priority(CoroPriority.WEB)
 async def to_code(config):
     var = cg.new_Pvariable(config[CONF_ID])
     await cg.register_component(var, config)

esphome/components/network/__init__.py

@@ -2,7 +2,7 @@ import esphome.codegen as cg
 from esphome.components.esp32 import add_idf_sdkconfig_option
 import esphome.config_validation as cv
 from esphome.const import CONF_ENABLE_IPV6, CONF_MIN_IPV6_ADDR_COUNT
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 
 CODEOWNERS = ["@esphome/core"]
 AUTO_LOAD = ["mdns"]
@@ -36,7 +36,7 @@ CONFIG_SCHEMA = cv.Schema(
 )
 
 
-@coroutine_with_priority(201.0)
+@coroutine_with_priority(CoroPriority.NETWORK)
 async def to_code(config):
     cg.add_define("USE_NETWORK")
     if CORE.using_arduino and CORE.is_esp32:

esphome/components/nrf52/__init__.py

@@ -30,7 +30,7 @@ from esphome.const import (
     PLATFORM_NRF52,
     ThreadModel,
 )
-from esphome.core import CORE, EsphomeError, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, EsphomeError, coroutine_with_priority
 from esphome.storage_json import StorageJSON
 from esphome.types import ConfigType
 
@@ -132,7 +132,7 @@ def _final_validate(config):
 FINAL_VALIDATE_SCHEMA = _final_validate
 
 
-@coroutine_with_priority(1000)
+@coroutine_with_priority(CoroPriority.PLATFORM)
 async def to_code(config: ConfigType) -> None:
     """Convert the configuration to code."""
     cg.add_platformio_option("board", config[CONF_BOARD])
@@ -170,7 +170,7 @@ async def to_code(config: ConfigType) -> None:
         CORE.add_job(_dfu_to_code, dfu_config)
 
 
-@coroutine_with_priority(90)
+@coroutine_with_priority(CoroPriority.DIAGNOSTICS)
 async def _dfu_to_code(dfu_config):
     cg.add_define("USE_NRF52_DFU")
     var = cg.new_Pvariable(dfu_config[CONF_ID])

esphome/components/number/__init__.py

@@ -76,7 +76,7 @@ from esphome.const import (
     DEVICE_CLASS_WIND_DIRECTION,
     DEVICE_CLASS_WIND_SPEED,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 from esphome.core.entity_helpers import entity_duplicate_validator, setup_entity
 from esphome.cpp_generator import MockObjClass
 
@@ -321,7 +321,7 @@ async def number_in_range_to_code(config, condition_id, template_arg, args):
     return var
 
 
-@coroutine_with_priority(100.0)
+@coroutine_with_priority(CoroPriority.CORE)
 async def to_code(config):
     cg.add_global(number_ns.using)
 

esphome/components/ota/__init__.py

@@ -10,7 +10,7 @@ from esphome.const import (
     CONF_TRIGGER_ID,
     PlatformFramework,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 
 CODEOWNERS = ["@esphome/core"]
 AUTO_LOAD = ["md5", "safe_mode"]
@@ -82,7 +82,7 @@ BASE_OTA_SCHEMA = cv.Schema(
 )
 
 
-@coroutine_with_priority(54.0)
+@coroutine_with_priority(CoroPriority.COMMUNICATION)
 async def to_code(config):
     cg.add_define("USE_OTA")
 

esphome/components/rp2040/__init__.py

@@ -18,7 +18,7 @@ from esphome.const import (
     PLATFORM_RP2040,
     ThreadModel,
 )
-from esphome.core import CORE, EsphomeError, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, EsphomeError, coroutine_with_priority
 from esphome.helpers import copy_file_if_changed, mkdir_p, read_file, write_file
 
 from .const import KEY_BOARD, KEY_PIO_FILES, KEY_RP2040, rp2040_ns
@@ -159,7 +159,7 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-@coroutine_with_priority(1000)
+@coroutine_with_priority(CoroPriority.PLATFORM)
 async def to_code(config):
     cg.add(rp2040_ns.setup_preferences())
 

esphome/components/safe_mode/__init__.py

@@ -10,7 +10,7 @@ from esphome.const import (
     CONF_TRIGGER_ID,
     KEY_PAST_SAFE_MODE,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 from esphome.cpp_generator import RawExpression
 
 CODEOWNERS = ["@paulmonigatti", "@jsuanet", "@kbx81"]
@@ -53,7 +53,7 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-@coroutine_with_priority(50.0)
+@coroutine_with_priority(CoroPriority.APPLICATION)
 async def to_code(config):
     if not config[CONF_DISABLED]:
         var = cg.new_Pvariable(config[CONF_ID])

esphome/components/select/__init__.py

@@ -16,7 +16,7 @@ from esphome.const import (
     CONF_TRIGGER_ID,
     CONF_WEB_SERVER,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 from esphome.core.entity_helpers import entity_duplicate_validator, setup_entity
 from esphome.cpp_generator import MockObjClass
 
@@ -124,7 +124,7 @@ async def new_select(config, *args, options: list[str]):
     return var
 
 
-@coroutine_with_priority(100.0)
+@coroutine_with_priority(CoroPriority.CORE)
 async def to_code(config):
     cg.add_global(select_ns.using)
 

esphome/components/sensor/__init__.py

@@ -101,7 +101,7 @@ from esphome.const import (
     DEVICE_CLASS_WIND_SPEED,
     ENTITY_CATEGORY_CONFIG,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 from esphome.core.entity_helpers import entity_duplicate_validator, setup_entity
 from esphome.cpp_generator import MockObjClass
 from esphome.util import Registry
@@ -1142,6 +1142,6 @@ def _lstsq(a, b):
     return _mat_dot(_mat_dot(x, a_t), b)
 
 
-@coroutine_with_priority(100.0)
+@coroutine_with_priority(CoroPriority.CORE)
 async def to_code(config):
     cg.add_global(sensor_ns.using)

esphome/components/speaker/__init__.py

@@ -4,7 +4,7 @@ from esphome.components import audio, audio_dac
 import esphome.config_validation as cv
 from esphome.const import CONF_DATA, CONF_ID, CONF_VOLUME
 from esphome.core import CORE
-from esphome.coroutine import coroutine_with_priority
+from esphome.coroutine import CoroPriority, coroutine_with_priority
 
 AUTO_LOAD = ["audio"]
 CODEOWNERS = ["@jesserockz", "@kahrendt"]
@@ -138,7 +138,7 @@ async def speaker_mute_action_to_code(config, action_id, template_arg, args):
     return cg.new_Pvariable(action_id, template_arg, paren)
 
 
-@coroutine_with_priority(100.0)
+@coroutine_with_priority(CoroPriority.CORE)
 async def to_code(config):
     cg.add_global(speaker_ns.using)
     cg.add_define("USE_SPEAKER")

esphome/components/spi/__init__.py

@@ -35,7 +35,7 @@ from esphome.const import (
     PLATFORM_RP2040,
     PlatformFramework,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 import esphome.final_validate as fv
 
 CODEOWNERS = ["@esphome/core", "@clydebarrow"]
@@ -351,7 +351,7 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
-@coroutine_with_priority(1.0)
+@coroutine_with_priority(CoroPriority.BUS)
 async def to_code(configs):
     cg.add_define("USE_SPI")
     cg.add_global(spi_ns.using)

esphome/components/status_led/__init__.py

@@ -2,7 +2,7 @@ from esphome import pins
 import esphome.codegen as cg
 import esphome.config_validation as cv
 from esphome.const import CONF_ID, CONF_PIN
-from esphome.core import coroutine_with_priority
+from esphome.core import CoroPriority, coroutine_with_priority
 
 status_led_ns = cg.esphome_ns.namespace("status_led")
 StatusLED = status_led_ns.class_("StatusLED", cg.Component)
@@ -15,7 +15,7 @@ CONFIG_SCHEMA = cv.Schema(
 ).extend(cv.COMPONENT_SCHEMA)
 
 
-@coroutine_with_priority(80.0)
+@coroutine_with_priority(CoroPriority.STATUS)
 async def to_code(config):
     pin = await cg.gpio_pin_expression(config[CONF_PIN])
     rhs = StatusLED.new(pin)

esphome/components/stepper/__init__.py

@@ -10,7 +10,7 @@ from esphome.const import (
     CONF_SPEED,
     CONF_TARGET,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 
 IS_PLATFORM_COMPONENT = True
 
@@ -178,6 +178,6 @@ async def stepper_set_deceleration_to_code(config, action_id, template_arg, args
     return var
 
 
-@coroutine_with_priority(100.0)
+@coroutine_with_priority(CoroPriority.CORE)
 async def to_code(config):
     cg.add_global(stepper_ns.using)

esphome/components/switch/__init__.py

@@ -21,7 +21,7 @@ from esphome.const import (
     DEVICE_CLASS_OUTLET,
     DEVICE_CLASS_SWITCH,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 from esphome.core.entity_helpers import entity_duplicate_validator, setup_entity
 from esphome.cpp_generator import MockObjClass
 
@@ -230,6 +230,6 @@ async def switch_is_off_to_code(config, condition_id, template_arg, args):
     return cg.new_Pvariable(condition_id, template_arg, paren, False)
 
 
-@coroutine_with_priority(100.0)
+@coroutine_with_priority(CoroPriority.CORE)
 async def to_code(config):
     cg.add_global(switch_ns.using)

esphome/components/text/__init__.py

@@ -13,7 +13,7 @@ from esphome.const import (
     CONF_VALUE,
     CONF_WEB_SERVER,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 from esphome.core.entity_helpers import entity_duplicate_validator, setup_entity
 from esphome.cpp_generator import MockObjClass
 
@@ -149,7 +149,7 @@ async def new_text(
     return var
 
 
-@coroutine_with_priority(100.0)
+@coroutine_with_priority(CoroPriority.CORE)
 async def to_code(config):
     cg.add_global(text_ns.using)
 

esphome/components/text_sensor/__init__.py

@@ -20,7 +20,7 @@ from esphome.const import (
     DEVICE_CLASS_EMPTY,
     DEVICE_CLASS_TIMESTAMP,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 from esphome.core.entity_helpers import entity_duplicate_validator, setup_entity
 from esphome.cpp_generator import MockObjClass
 from esphome.util import Registry
@@ -230,7 +230,7 @@ async def new_text_sensor(config, *args):
     return var
 
 
-@coroutine_with_priority(100.0)
+@coroutine_with_priority(CoroPriority.CORE)
 async def to_code(config):
     cg.add_global(text_sensor_ns.using)
 

esphome/components/time/__init__.py

@@ -26,7 +26,7 @@ from esphome.const import (
     CONF_TIMEZONE,
     CONF_TRIGGER_ID,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -340,7 +340,7 @@ async def register_time(time_var, config):
     await setup_time_core_(time_var, config)
 
 
-@coroutine_with_priority(100.0)
+@coroutine_with_priority(CoroPriority.CORE)
 async def to_code(config):
     if CORE.using_zephyr:
         zephyr_add_prj_conf("POSIX_CLOCK", True)

esphome/components/touchscreen/__init__.py

@@ -13,7 +13,7 @@ from esphome.const import (
     CONF_SWAP_XY,
     CONF_TRANSFORM,
 )
-from esphome.core import coroutine_with_priority
+from esphome.core import CoroPriority, coroutine_with_priority
 
 CODEOWNERS = ["@jesserockz", "@nielsnl68"]
 DEPENDENCIES = ["display"]
@@ -152,7 +152,7 @@ async def register_touchscreen(var, config):
         )
 
 
-@coroutine_with_priority(100.0)
+@coroutine_with_priority(CoroPriority.CORE)
 async def to_code(config):
     cg.add_global(touchscreen_ns.using)
     cg.add_define("USE_TOUCHSCREEN")

esphome/components/update/__init__.py

@@ -14,7 +14,7 @@ from esphome.const import (
     DEVICE_CLASS_FIRMWARE,
     ENTITY_CATEGORY_CONFIG,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 from esphome.core.entity_helpers import entity_duplicate_validator, setup_entity
 from esphome.cpp_generator import MockObjClass
 
@@ -124,7 +124,7 @@ async def new_update(config):
     return var
 
 
-@coroutine_with_priority(100.0)
+@coroutine_with_priority(CoroPriority.CORE)
 async def to_code(config):
     cg.add_global(update_ns.using)
 

esphome/components/valve/__init__.py

@@ -21,7 +21,7 @@ from esphome.const import (
     DEVICE_CLASS_GAS,
     DEVICE_CLASS_WATER,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 from esphome.core.entity_helpers import entity_duplicate_validator, setup_entity
 from esphome.cpp_generator import MockObjClass
 
@@ -233,6 +233,6 @@ async def valve_control_to_code(config, action_id, template_arg, args):
     return var
 
 
-@coroutine_with_priority(100.0)
+@coroutine_with_priority(CoroPriority.CORE)
 async def to_code(config):
     cg.add_global(valve_ns.using)

esphome/components/web_server/__init__.py

@@ -31,7 +31,7 @@ from esphome.const import (
     PLATFORM_LN882X,
     PLATFORM_RTL87XX,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 import esphome.final_validate as fv
 from esphome.types import ConfigType
 
@@ -269,7 +269,7 @@ def add_resource_as_progmem(
     cg.add_global(cg.RawExpression(size_t))
 
 
-@coroutine_with_priority(40.0)
+@coroutine_with_priority(CoroPriority.WEB)
 async def to_code(config):
     paren = await cg.get_variable(config[CONF_WEB_SERVER_BASE_ID])
 

esphome/components/web_server/ota/__init__.py

@@ -3,7 +3,7 @@ from esphome.components.esp32 import add_idf_component
 from esphome.components.ota import BASE_OTA_SCHEMA, OTAComponent, ota_to_code
 import esphome.config_validation as cv
 from esphome.const import CONF_ID
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 
 CODEOWNERS = ["@esphome/core"]
 DEPENDENCIES = ["network", "web_server_base"]
@@ -22,7 +22,7 @@ CONFIG_SCHEMA = (
 )
 
 
-@coroutine_with_priority(52.0)
+@coroutine_with_priority(CoroPriority.COMMUNICATION)
 async def to_code(config):
     var = cg.new_Pvariable(config[CONF_ID])
     await ota_to_code(var, config)

esphome/components/web_server_base/__init__.py

@@ -1,7 +1,7 @@
 import esphome.codegen as cg
 import esphome.config_validation as cv
 from esphome.const import CONF_ID
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 
 CODEOWNERS = ["@esphome/core"]
 DEPENDENCIES = ["network"]
@@ -26,7 +26,7 @@ CONFIG_SCHEMA = cv.Schema(
 )
 
 
-@coroutine_with_priority(65.0)
+@coroutine_with_priority(CoroPriority.COMMUNICATION)
 async def to_code(config):
     var = cg.new_Pvariable(config[CONF_ID])
     await cg.register_component(var, config)

esphome/components/wifi/__init__.py

@@ -44,7 +44,7 @@ from esphome.const import (
     CONF_USERNAME,
     PlatformFramework,
 )
-from esphome.core import CORE, HexInt, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, HexInt, coroutine_with_priority
 import esphome.final_validate as fv
 
 from . import wpa2_eap
@@ -370,7 +370,7 @@ def wifi_network(config, ap, static_ip):
     return ap
 
 
-@coroutine_with_priority(60.0)
+@coroutine_with_priority(CoroPriority.COMMUNICATION)
 async def to_code(config):
     var = cg.new_Pvariable(config[CONF_ID])
     cg.add(var.set_use_address(config[CONF_USE_ADDRESS]))

esphome/core/__init__.py

@@ -29,6 +29,7 @@ from esphome.const import (
 
 # pylint: disable=unused-import
 from esphome.coroutine import (  # noqa: F401
+    CoroPriority,
     FakeAwaitable as _FakeAwaitable,
     FakeEventLoop as _FakeEventLoop,
     coroutine,

esphome/core/config.py

@@ -39,7 +39,7 @@ from esphome.const import (
     PlatformFramework,
     __version__ as ESPHOME_VERSION,
 )
-from esphome.core import CORE, coroutine_with_priority
+from esphome.core import CORE, CoroPriority, coroutine_with_priority
 from esphome.helpers import (
     copy_file_if_changed,
     fnv1a_32bit_hash,
@@ -359,7 +359,7 @@ ARDUINO_GLUE_CODE = """\
 """
 
 
-@coroutine_with_priority(-999.0)
+@coroutine_with_priority(CoroPriority.WORKAROUNDS)
 async def add_arduino_global_workaround():
     # The Arduino framework defined these itself in the global
     # namespace. For the esphome codebase that is not a problem,
@@ -376,7 +376,7 @@ async def add_arduino_global_workaround():
         cg.add_global(cg.RawStatement(line))
 
 
-@coroutine_with_priority(-1000.0)
+@coroutine_with_priority(CoroPriority.FINAL)
 async def add_includes(includes):
     # Add includes at the very end, so that the included files can access global variables
     for include in includes:
@@ -392,7 +392,7 @@ async def add_includes(includes):
             include_file(path, basename)
 
 
-@coroutine_with_priority(-1000.0)
+@coroutine_with_priority(CoroPriority.FINAL)
 async def _add_platformio_options(pio_options):
     # Add includes at the very end, so that they override everything
     for key, val in pio_options.items():
@@ -401,7 +401,7 @@ async def _add_platformio_options(pio_options):
         cg.add_platformio_option(key, val)
 
 
-@coroutine_with_priority(30.0)
+@coroutine_with_priority(CoroPriority.AUTOMATION)
 async def _add_automations(config):
     for conf in config.get(CONF_ON_BOOT, []):
         trigger = cg.new_Pvariable(conf[CONF_TRIGGER_ID], conf.get(CONF_PRIORITY))
@@ -423,7 +423,7 @@ async def _add_automations(config):
 DATETIME_SUBTYPES = {"date", "time", "datetime"}
 
 
-@coroutine_with_priority(-1000.0)
+@coroutine_with_priority(CoroPriority.FINAL)
 async def _add_platform_defines() -> None:
     # Generate compile-time defines for platforms that have actual entities
     # Only add USE_* and count defines when there are entities
@@ -442,7 +442,7 @@ async def _add_platform_defines() -> None:
             cg.add_define(f"USE_{platform_name.upper()}")
 
 
-@coroutine_with_priority(100.0)
+@coroutine_with_priority(CoroPriority.CORE)
 async def to_code(config: ConfigType) -> None:
     cg.add_global(cg.global_ns.namespace("esphome").using)
     # These can be used by user lambdas, put them to default scope

esphome/core/scheduler.cpp

@@ -15,21 +15,18 @@ namespace esphome {
 static const char *const TAG = "scheduler";
 
 // Memory pool configuration constants
-// Pool size of 10 is a balance between memory usage and performance:
+// Pool size of 5 matches typical usage patterns (2-4 active timers)
-// - Small enough to not waste memory on simple configs (1-2 timers)
+// - Minimal memory overhead (~250 bytes on ESP32)
-// - Large enough to handle complex setups with multiple sensors/components
+// - Sufficient for most configs with a couple sensors/components
-// - Prevents system-wide stalls from heap allocation/deallocation that can
+// - Still prevents heap fragmentation and allocation stalls
-//   disrupt task synchronization and cause dropped events
+// - Complex setups with many timers will just allocate beyond the pool
-static constexpr size_t MAX_POOL_SIZE = 10;
+// See https://github.com/esphome/backlog/issues/52
-// Maximum number of cancelled items to keep in the heap before forcing a cleanup.
+static constexpr size_t MAX_POOL_SIZE = 5;
-// Set to 6 to trigger cleanup relatively frequently, ensuring cancelled items are
-// recycled to the pool in a timely manner to maintain pool efficiency.
-static const uint32_t MAX_LOGICALLY_DELETED_ITEMS = 6;
 
-// Ensure MAX_LOGICALLY_DELETED_ITEMS is at least 4 smaller than MAX_POOL_SIZE
+// Maximum number of logically deleted (cancelled) items before forcing cleanup.
-// This guarantees we have room in the pool for recycled items when cleanup occurs
+// Set to 5 to match the pool size - when we have as many cancelled items as our
-static_assert(MAX_LOGICALLY_DELETED_ITEMS + 4 <= MAX_POOL_SIZE,
+// pool can hold, it's time to clean up and recycle them.
-              "MAX_LOGICALLY_DELETED_ITEMS must be at least 4 smaller than MAX_POOL_SIZE");
+static constexpr uint32_t MAX_LOGICALLY_DELETED_ITEMS = 5;
 
 // Half the 32-bit range - used to detect rollovers vs normal time progression
 static constexpr uint32_t HALF_MAX_UINT32 = std::numeric_limits<uint32_t>::max() / 2;
@@ -108,13 +105,13 @@ void HOT Scheduler::set_timer_common_(Component *component, SchedulerItem::Type
     item = std::move(this->scheduler_item_pool_.back());
     this->scheduler_item_pool_.pop_back();
 #ifdef ESPHOME_DEBUG_SCHEDULER
-    ESP_LOGVV(TAG, "Reused item from pool (pool size now: %zu)", this->scheduler_item_pool_.size());
+    ESP_LOGD(TAG, "Reused item from pool (pool size now: %zu)", this->scheduler_item_pool_.size());
 #endif
   } else {
     // Allocate new if pool is empty
     item = make_unique<SchedulerItem>();
 #ifdef ESPHOME_DEBUG_SCHEDULER
-    ESP_LOGVV(TAG, "Allocated new item (pool empty)");
+    ESP_LOGD(TAG, "Allocated new item (pool empty)");
 #endif
   }
   item->component = component;
@@ -384,9 +381,10 @@ void HOT Scheduler::call(uint32_t now) {
       }
 
       const char *name = item->get_name();
-      ESP_LOGD(TAG, "  %s '%s/%s' interval=%" PRIu32 " next_execution in %" PRIu64 "ms at %" PRIu64,
+      bool is_cancelled = is_item_removed_(item.get());
+      ESP_LOGD(TAG, "  %s '%s/%s' interval=%" PRIu32 " next_execution in %" PRIu64 "ms at %" PRIu64 "%s",
                item->get_type_str(), item->get_source(), name ? name : "(null)", item->interval,
-               item->next_execution_ - now_64, item->next_execution_);
+               item->next_execution_ - now_64, item->next_execution_, is_cancelled ? " [CANCELLED]" : "");
 
       old_items.push_back(std::move(item));
     }
@@ -401,8 +399,13 @@ void HOT Scheduler::call(uint32_t now) {
   }
 #endif /* ESPHOME_DEBUG_SCHEDULER */
 
-  // If we have too many items to remove
+  // Cleanup removed items before processing
-  if (this->to_remove_ > MAX_LOGICALLY_DELETED_ITEMS) {
+  // First try to clean items from the top of the heap (fast path)
+  this->cleanup_();
+
+  // If we still have too many cancelled items, do a full cleanup
+  // This only happens if cancelled items are stuck in the middle/bottom of the heap
+  if (this->to_remove_ >= MAX_LOGICALLY_DELETED_ITEMS) {
     // We hold the lock for the entire cleanup operation because:
     // 1. We're rebuilding the entire items_ list, so we need exclusive access throughout
     // 2. Other threads must see either the old state or the new state, not intermediate states
@@ -428,9 +431,6 @@ void HOT Scheduler::call(uint32_t now) {
     std::make_heap(this->items_.begin(), this->items_.end(), SchedulerItem::cmp);
     this->to_remove_ = 0;
   }
-
-  // Cleanup removed items before processing
-  this->cleanup_();
   while (!this->items_.empty()) {
     // use scoping to indicate visibility of `item` variable
     {
@@ -516,7 +516,9 @@ void HOT Scheduler::call(uint32_t now) {
 void HOT Scheduler::process_to_add() {
   LockGuard guard{this->lock_};
   for (auto &it : this->to_add_) {
-    if (it->remove) {
+    if (is_item_removed_(it.get())) {
+      // Recycle cancelled items
+      this->recycle_item_(std::move(it));
       continue;
     }
 
@@ -594,14 +596,28 @@ bool HOT Scheduler::cancel_item_locked_(Component *component, const char *name_c
 
   // Check all containers for matching items
 #ifndef ESPHOME_THREAD_SINGLE
-  // Cancel and immediately recycle items in defer queue
+  // Mark items in defer queue as cancelled (they'll be skipped when processed)
   if (type == SchedulerItem::TIMEOUT) {
-    total_cancelled +=
+    for (auto &item : this->defer_queue_) {
-        this->cancel_and_recycle_from_container_(this->defer_queue_, component, name_cstr, type, match_retry);
+      if (this->matches_item_(item, component, name_cstr, type, match_retry)) {
+        this->mark_item_removed_(item.get());
+        total_cancelled++;
+      }
+    }
   }
 #endif /* not ESPHOME_THREAD_SINGLE */
 
-  // Cancel items in the main heap (can't recycle immediately due to heap structure)
+  // Cancel items in the main heap
+  // Special case: if the last item in the heap matches, we can remove it immediately
+  // (removing the last element doesn't break heap structure)
+  if (!this->items_.empty()) {
+    auto &last_item = this->items_.back();
+    if (this->matches_item_(last_item, component, name_cstr, type, match_retry)) {
+      this->recycle_item_(std::move(this->items_.back()));
+      this->items_.pop_back();
+      total_cancelled++;
+    }
+    // For other items in heap, we can only mark for removal (can't remove from middle of heap)
     for (auto &item : this->items_) {
       if (this->matches_item_(item, component, name_cstr, type, match_retry)) {
         this->mark_item_removed_(item.get());
@@ -609,9 +625,16 @@ bool HOT Scheduler::cancel_item_locked_(Component *component, const char *name_c
         this->to_remove_++;  // Track removals for heap items
       }
     }
+  }
 
-  // Cancel and immediately recycle items in to_add_ since they're not in heap yet
+  // Cancel items in to_add_
-  total_cancelled += this->cancel_and_recycle_from_container_(this->to_add_, component, name_cstr, type, match_retry);
+  for (auto &item : this->to_add_) {
+    if (this->matches_item_(item, component, name_cstr, type, match_retry)) {
+      this->mark_item_removed_(item.get());
+      total_cancelled++;
+      // Don't track removals for to_add_ items
+    }
+  }
 
   return total_cancelled > 0;
 }
@@ -790,11 +813,11 @@ void Scheduler::recycle_item_(std::unique_ptr<SchedulerItem> item) {
     item->clear_dynamic_name();
     this->scheduler_item_pool_.push_back(std::move(item));
 #ifdef ESPHOME_DEBUG_SCHEDULER
-    ESP_LOGVV(TAG, "Recycled item to pool (pool size now: %zu)", this->scheduler_item_pool_.size());
+    ESP_LOGD(TAG, "Recycled item to pool (pool size now: %zu)", this->scheduler_item_pool_.size());
 #endif
   } else {
 #ifdef ESPHOME_DEBUG_SCHEDULER
-    ESP_LOGVV(TAG, "Pool full (size: %zu), deleting item", this->scheduler_item_pool_.size());
+    ESP_LOGD(TAG, "Pool full (size: %zu), deleting item", this->scheduler_item_pool_.size());
 #endif
   }
   // else: unique_ptr will delete the item when it goes out of scope

esphome/core/scheduler.h

@@ -5,7 +5,6 @@
 #include <memory>
 #include <cstring>
 #include <deque>
-#include <array>
 #ifdef ESPHOME_THREAD_MULTI_ATOMICS
 #include <atomic>
 #endif
@@ -217,6 +216,15 @@ class Scheduler {
   // Common implementation for cancel operations
   bool cancel_item_(Component *component, bool is_static_string, const void *name_ptr, SchedulerItem::Type type);
 
+  // Helper to check if two scheduler item names match
+  inline bool HOT names_match_(const char *name1, const char *name2) const {
+    // Check pointer equality first (common for static strings), then string contents
+    // The core ESPHome codebase uses static strings (const char*) for component names,
+    // making pointer comparison effective. The std::string overloads exist only for
+    // compatibility with external components but are rarely used in practice.
+    return (name1 != nullptr && name2 != nullptr) && ((name1 == name2) || (strcmp(name1, name2) == 0));
+  }
+
   // Helper function to check if item matches criteria for cancellation
   inline bool HOT matches_item_(const std::unique_ptr<SchedulerItem> &item, Component *component, const char *name_cstr,
                                 SchedulerItem::Type type, bool match_retry, bool skip_removed = true) const {
@@ -224,19 +232,7 @@ class Scheduler {
         (match_retry && !item->is_retry)) {
       return false;
     }
-    const char *item_name = item->get_name();
+    return this->names_match_(item->get_name(), name_cstr);
-    if (item_name == nullptr) {
-      return false;
-    }
-    // Fast path: if pointers are equal
-    // This is effective because the core ESPHome codebase uses static strings (const char*)
-    // for component names. The std::string overloads exist only for compatibility with
-    // external components, but are rarely used in practice.
-    if (item_name == name_cstr) {
-      return true;
-    }
-    // Slow path: compare string contents
-    return strcmp(name_cstr, item_name) == 0;
   }
 
   // Helper to execute a scheduler item
@@ -295,24 +291,6 @@ class Scheduler {
     return false;
   }
 
-  // Template helper to cancel and recycle items from a container
-  template<typename Container>
-  size_t cancel_and_recycle_from_container_(Container &container, Component *component, const char *name_cstr,
-                                            SchedulerItem::Type type, bool match_retry) {
-    size_t cancelled = 0;
-    for (auto it = container.begin(); it != container.end();) {
-      if (this->matches_item_(*it, component, name_cstr, type, match_retry)) {
-        // Recycle the cancelled item immediately
-        this->recycle_item_(std::move(*it));
-        it = container.erase(it);
-        cancelled++;
-      } else {
-        ++it;
-      }
-    }
-    return cancelled;
-  }
-
   Mutex lock_;
   std::vector<std::unique_ptr<SchedulerItem>> items_;
   std::vector<std::unique_ptr<SchedulerItem>> to_add_;
@@ -325,8 +303,8 @@ class Scheduler {
   // Memory pool for recycling SchedulerItem objects to reduce heap churn.
   // Design decisions:
   // - std::vector is used instead of a fixed array because many systems only need 1-2 scheduler items
-  // - The vector grows dynamically up to MAX_POOL_SIZE (10) only when needed, saving memory on simple setups
+  // - The vector grows dynamically up to MAX_POOL_SIZE (5) only when needed, saving memory on simple setups
-  // - This approach balances memory efficiency for simple configs with performance for complex ones
+  // - Pool size of 5 matches typical usage (2-4 timers) while keeping memory overhead low (~250 bytes on ESP32)
   // - The pool significantly reduces heap fragmentation which is critical because heap allocation/deallocation
   //   can stall the entire system, causing timing issues and dropped events for any components that need
   //   to synchronize between tasks (see https://github.com/esphome/backlog/issues/52)

esphome/coroutine.py

@@ -42,7 +42,10 @@ Here everything is combined in `yield` expressions. You await other coroutines u
 the last `yield` expression defines what is returned.
 """
 
+from __future__ import annotations
+
 from collections.abc import Awaitable, Callable, Generator, Iterator
+import enum
 import functools
 import heapq
 import inspect
@@ -53,6 +56,79 @@ from typing import Any
 _LOGGER = logging.getLogger(__name__)
 
 
+class CoroPriority(enum.IntEnum):
+    """Execution priority stages for ESPHome code generation.
+
+    Higher values run first. These stages ensure proper dependency
+    resolution during code generation.
+    """
+
+    # Platform initialization - must run first
+    # Examples: esp32, esp8266, rp2040
+    PLATFORM = 1000
+
+    # Network infrastructure setup
+    # Examples: network (201)
+    NETWORK = 201
+
+    # Network transport layer
+    # Examples: async_tcp (200)
+    NETWORK_TRANSPORT = 200
+
+    # Core system components
+    # Examples: esphome core, most entity base components (cover, update, datetime,
+    # valve, alarm_control_panel, lock, event, binary_sensor, button, climate, fan,
+    # light, media_player, number, select, sensor, switch, text_sensor, text),
+    # microphone, speaker, audio_dac, touchscreen, stepper
+    CORE = 100
+
+    # Diagnostic and debugging systems
+    # Examples: logger (90)
+    DIAGNOSTICS = 90
+
+    # Status and monitoring systems
+    # Examples: status_led (80)
+    STATUS = 80
+
+    # Communication protocols and services
+    # Examples: web_server_base (65), captive_portal (64), wifi (60), ethernet (60),
+    # mdns (55), ota_updates (54), web_server_ota (52)
+    COMMUNICATION = 60
+
+    # Application-level services
+    # Examples: safe_mode (50)
+    APPLICATION = 50
+
+    # Web and UI services
+    # Examples: web_server (40)
+    WEB = 40
+
+    # Automations and user logic
+    # Examples: esphome core automations (30)
+    AUTOMATION = 30
+
+    # Bus and peripheral setup
+    # Examples: i2c (1)
+    BUS = 1
+
+    # Standard component priority (default)
+    # Components without explicit priority run at 0
+    COMPONENT = 0
+
+    # Components that need others to be registered first
+    # Examples: globals (-100)
+    LATE = -100
+
+    # Platform-specific workarounds and fixes
+    # Examples: add_arduino_global_workaround (-999), esp8266 pin states (-999)
+    WORKAROUNDS = -999
+
+    # Final setup that requires all components to be registered
+    # Examples: add_includes, _add_platformio_options, _add_platform_defines (all -1000),
+    # esp32_ble_tracker feature defines (-1000)
+    FINAL = -1000
+
+
 def coroutine(func: Callable[..., Any]) -> Callable[..., Awaitable[Any]]:
     """Decorator to apply to methods to convert them to ESPHome coroutines."""
     if getattr(func, "_esphome_coroutine", False):
@@ -95,15 +171,16 @@ def coroutine(func: Callable[..., Any]) -> Callable[..., Awaitable[Any]]:
     return coro
 
 
-def coroutine_with_priority(priority: float):
+def coroutine_with_priority(priority: float | CoroPriority):
     """Decorator to apply to functions to convert them to ESPHome coroutines.
 
     :param priority: priority with which to schedule the coroutine, higher priorities run first.
+                     Can be a float or a CoroPriority enum value.
     """
 
     def decorator(func):
         coro = coroutine(func)
-        coro.priority = priority
+        coro.priority = float(priority)
         return coro
 
     return decorator
@@ -173,7 +250,7 @@ class _Task:
         self.iterator = iterator
         self.original_function = original_function
 
-    def with_priority(self, priority: float) -> "_Task":
+    def with_priority(self, priority: float) -> _Task:
         return _Task(priority, self.id_number, self.iterator, self.original_function)
 
     @property

tests/integration/fixtures/host_preferences_save_load.yaml

@@ -0,0 +1,110 @@
+esphome:
+  name: test_device
+  on_boot:
+    - lambda: |-
+        ESP_LOGD("test", "Host preferences test starting");
+
+host:
+
+logger:
+  level: DEBUG
+
+api:
+
+preferences:
+  flash_write_interval: 0s  # Disable automatic saving for test control
+
+switch:
+  - platform: template
+    name: "Test Switch"
+    id: test_switch
+    optimistic: true
+    restore_mode: DISABLED  # Don't auto-restore for test control
+
+number:
+  - platform: template
+    name: "Test Number"
+    id: test_number
+    min_value: 0
+    max_value: 100
+    step: 0.1
+    optimistic: true
+    restore_value: false  # Don't auto-restore for test control
+
+button:
+  - platform: template
+    name: "Save Preferences"
+    on_press:
+      - lambda: |-
+          // Save current values to preferences
+          ESPPreferenceObject switch_pref = global_preferences->make_preference<bool>(0x1234);
+          ESPPreferenceObject number_pref = global_preferences->make_preference<float>(0x5678);
+
+          bool switch_value = id(test_switch).state;
+          float number_value = id(test_number).state;
+
+          if (switch_pref.save(&switch_value)) {
+            ESP_LOGI("test", "Preference saved: key=switch, value=%.1f", switch_value ? 1.0 : 0.0);
+          }
+          if (number_pref.save(&number_value)) {
+            ESP_LOGI("test", "Preference saved: key=number, value=%.1f", number_value);
+          }
+
+          // Force sync to disk
+          global_preferences->sync();
+
+  - platform: template
+    name: "Load Preferences"
+    on_press:
+      - lambda: |-
+          // Load values from preferences
+          ESPPreferenceObject switch_pref = global_preferences->make_preference<bool>(0x1234);
+          ESPPreferenceObject number_pref = global_preferences->make_preference<float>(0x5678);
+
+          // Also try to load non-existent preferences (tests our fix)
+          ESPPreferenceObject fake_pref1 = global_preferences->make_preference<uint32_t>(0x9999);
+          ESPPreferenceObject fake_pref2 = global_preferences->make_preference<uint32_t>(0xAAAA);
+
+          bool switch_value = false;
+          float number_value = 0.0;
+          uint32_t fake_value = 0;
+          int loaded_count = 0;
+
+          // These should not exist and shouldn't create map entries
+          fake_pref1.load(&fake_value);
+          fake_pref2.load(&fake_value);
+
+          if (switch_pref.load(&switch_value)) {
+            id(test_switch).publish_state(switch_value);
+            ESP_LOGI("test", "Preference loaded: key=switch, value=%.1f", switch_value ? 1.0 : 0.0);
+            loaded_count++;
+          } else {
+            ESP_LOGW("test", "Failed to load switch preference");
+          }
+
+          if (number_pref.load(&number_value)) {
+            id(test_number).publish_state(number_value);
+            ESP_LOGI("test", "Preference loaded: key=number, value=%.1f", number_value);
+            loaded_count++;
+          } else {
+            ESP_LOGW("test", "Failed to load number preference");
+          }
+
+          // Log completion message for the test to detect
+          ESP_LOGI("test", "Final load test: loaded %d preferences successfully", loaded_count);
+
+  - platform: template
+    name: "Verify Preferences"
+    on_press:
+      - lambda: |-
+          // Verify current values match what we expect
+          bool switch_value = id(test_switch).state;
+          float number_value = id(test_number).state;
+
+          // After loading, switch should be true (1.0) and number should be 42.5
+          if (switch_value == true && number_value == 42.5) {
+            ESP_LOGI("test", "Preferences verified: values match!");
+          } else {
+            ESP_LOGE("test", "Preferences mismatch: switch=%d, number=%.1f",
+                     switch_value, number_value);
+          }

tests/integration/fixtures/scheduler_pool.yaml

@@ -27,6 +27,9 @@ api:
     - service: run_phase_6
       then:
         - script.execute: test_full_pool_reuse
+    - service: run_phase_7
+      then:
+        - script.execute: test_same_defer_optimization
     - service: run_complete
       then:
         - script.execute: complete_test
@@ -87,7 +90,8 @@ script:
           auto *component = id(test_sensor);
 
           // Multiple sensors with different update intervals
-          App.scheduler.set_interval(component, "temp_sensor", 100, []() {
+          // These should only allocate once and reuse the same item for each interval execution
+          App.scheduler.set_interval(component, "temp_sensor", 10, []() {
             ESP_LOGD("test", "Temperature sensor update");
             id(interval_counter)++;
             if (id(interval_counter) >= 3) {
@@ -96,7 +100,7 @@ script:
             }
           });
 
-          App.scheduler.set_interval(component, "humidity_sensor", 150, []() {
+          App.scheduler.set_interval(component, "humidity_sensor", 15, []() {
             ESP_LOGD("test", "Humidity sensor update");
             id(interval_counter)++;
             if (id(interval_counter) >= 5) {
@@ -105,7 +109,9 @@ script:
             }
           });
 
+          // Only 2 allocations for the intervals, no matter how many times they execute
           id(create_count) += 2;
+          ESP_LOGD("test", "Created 2 intervals - they will reuse same items for each execution");
           ESP_LOGI("test", "Phase 2 complete");
 
   - id: test_communication_patterns
@@ -215,11 +221,14 @@ script:
   - id: test_full_pool_reuse
     then:
       - lambda: |-
-          ESP_LOGI("test", "Phase 6: Testing full pool reuse after Phase 5 items complete");
+          ESP_LOGI("test", "Phase 6: Testing pool size limits after Phase 5 items complete");
 
           // At this point, all Phase 5 timeouts should have completed and been recycled.
-          // The pool should be at or near its maximum size (10).
+          // The pool should be at its maximum size (5).
-          // Creating 10 new items should reuse all from the pool.
+          // Creating 10 new items tests that:
+          // - First 5 items reuse from the pool
+          // - Remaining 5 items allocate new (pool empty)
+          // - Pool doesn't grow beyond MAX_POOL_SIZE of 5
 
           auto *component = id(test_sensor);
           int full_reuse_count = 10;
@@ -235,6 +244,28 @@ script:
           id(create_count) += full_reuse_count;
           ESP_LOGI("test", "Phase 6 complete");
 
+  - id: test_same_defer_optimization
+    then:
+      - lambda: |-
+          ESP_LOGI("test", "Phase 7: Testing same-named defer optimization");
+
+          auto *component = id(test_sensor);
+
+          // Create 10 defers with the same name - should optimize to update callback in-place
+          // This pattern is common in components like ratgdo that repeatedly defer state updates
+          for (int i = 0; i < 10; i++) {
+            App.scheduler.set_timeout(component, "repeated_defer", 0, [i]() {
+              ESP_LOGD("test", "Repeated defer executed with value: %d", i);
+            });
+          }
+
+          // Only the first should allocate, the rest should update in-place
+          // We expect only 1 allocation for all 10 operations
+          id(create_count) += 1;  // Only count 1 since others should be optimized
+
+          ESP_LOGD("test", "Created 10 same-named defers (should only allocate once)");
+          ESP_LOGI("test", "Phase 7 complete");
+
   - id: complete_test
     then:
       - lambda: |-

tests/integration/test_host_preferences.py

@@ -0,0 +1,167 @@
+"""Test host preferences save and load functionality."""
+
+from __future__ import annotations
+
+import asyncio
+import re
+from typing import Any
+
+from aioesphomeapi import ButtonInfo, EntityInfo, NumberInfo, SwitchInfo
+import pytest
+
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+def find_entity_by_name(
+    entities: list[EntityInfo], entity_type: type, name: str
+) -> Any:
+    """Helper to find an entity by type and name."""
+    return next(
+        (e for e in entities if isinstance(e, entity_type) and e.name == name), None
+    )
+
+
+@pytest.mark.asyncio
+async def test_host_preferences_save_load(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test that preferences are correctly saved and loaded after our optimization fix."""
+    loop = asyncio.get_running_loop()
+    log_lines: list[str] = []
+    preferences_saved = loop.create_future()
+    preferences_loaded = loop.create_future()
+    values_match = loop.create_future()
+    final_load_complete = loop.create_future()
+
+    # Patterns to match preference logs
+    save_pattern = re.compile(r"Preference saved: key=(\w+), value=([0-9.]+)")
+    load_pattern = re.compile(r"Preference loaded: key=(\w+), value=([0-9.]+)")
+    verify_pattern = re.compile(r"Preferences verified: values match!")
+    final_load_success_pattern = re.compile(
+        r"Final load test: loaded \d+ preferences successfully"
+    )
+
+    saved_values: dict[str, float] = {}
+    loaded_values: dict[str, float] = {}
+
+    def check_output(line: str) -> None:
+        """Check log output for preference operations."""
+        log_lines.append(line)
+
+        # Look for save operations
+        match = save_pattern.search(line)
+        if match:
+            key = match.group(1)
+            value = float(match.group(2))
+            saved_values[key] = value
+            if len(saved_values) >= 2 and not preferences_saved.done():
+                preferences_saved.set_result(True)
+
+        # Look for load operations
+        match = load_pattern.search(line)
+        if match:
+            key = match.group(1)
+            value = float(match.group(2))
+            loaded_values[key] = value
+            if len(loaded_values) >= 2 and not preferences_loaded.done():
+                preferences_loaded.set_result(True)
+
+        # Look for verification
+        if verify_pattern.search(line) and not values_match.done():
+            values_match.set_result(True)
+
+        # Look for final load test completion
+        if final_load_success_pattern.search(line) and not final_load_complete.done():
+            final_load_complete.set_result(True)
+
+    async with (
+        run_compiled(yaml_config, line_callback=check_output),
+        api_client_connected() as client,
+    ):
+        # Get entity list
+        entities, _ = await client.list_entities_services()
+
+        # Find our test entities using helper
+        test_switch = find_entity_by_name(entities, SwitchInfo, "Test Switch")
+        test_number = find_entity_by_name(entities, NumberInfo, "Test Number")
+        save_button = find_entity_by_name(entities, ButtonInfo, "Save Preferences")
+        load_button = find_entity_by_name(entities, ButtonInfo, "Load Preferences")
+        verify_button = find_entity_by_name(entities, ButtonInfo, "Verify Preferences")
+
+        assert test_switch is not None, "Test Switch not found"
+        assert test_number is not None, "Test Number not found"
+        assert save_button is not None, "Save Preferences button not found"
+        assert load_button is not None, "Load Preferences button not found"
+        assert verify_button is not None, "Verify Preferences button not found"
+
+        # Set initial values
+        client.switch_command(test_switch.key, True)
+        client.number_command(test_number.key, 42.5)
+
+        # Save preferences
+        client.button_command(save_button.key)
+
+        # Wait for save to complete
+        try:
+            await asyncio.wait_for(preferences_saved, timeout=5.0)
+        except TimeoutError:
+            pytest.fail("Preferences not saved within timeout")
+
+        # Verify we saved the expected values
+        assert "switch" in saved_values, f"Switch preference not saved: {saved_values}"
+        assert "number" in saved_values, f"Number preference not saved: {saved_values}"
+        assert saved_values["switch"] == 1.0, (
+            f"Switch value incorrect: {saved_values['switch']}"
+        )
+        assert saved_values["number"] == 42.5, (
+            f"Number value incorrect: {saved_values['number']}"
+        )
+
+        # Change the values to something else
+        client.switch_command(test_switch.key, False)
+        client.number_command(test_number.key, 13.7)
+
+        # Load preferences (should restore the saved values)
+        client.button_command(load_button.key)
+
+        # Wait for load to complete
+        try:
+            await asyncio.wait_for(preferences_loaded, timeout=5.0)
+        except TimeoutError:
+            pytest.fail("Preferences not loaded within timeout")
+
+        # Verify loaded values match saved values
+        assert "switch" in loaded_values, (
+            f"Switch preference not loaded: {loaded_values}"
+        )
+        assert "number" in loaded_values, (
+            f"Number preference not loaded: {loaded_values}"
+        )
+        assert loaded_values["switch"] == saved_values["switch"], (
+            f"Loaded switch value {loaded_values['switch']} doesn't match saved {saved_values['switch']}"
+        )
+        assert loaded_values["number"] == saved_values["number"], (
+            f"Loaded number value {loaded_values['number']} doesn't match saved {saved_values['number']}"
+        )
+
+        # Verify the values were actually restored
+        client.button_command(verify_button.key)
+
+        # Wait for verification
+        try:
+            await asyncio.wait_for(values_match, timeout=5.0)
+        except TimeoutError:
+            pytest.fail("Preference verification failed within timeout")
+
+        # Test that non-existent preferences don't crash (tests our fix)
+        # This will trigger load attempts for keys that don't exist
+        # Our fix should prevent map entries from being created
+        client.button_command(load_button.key)
+
+        # Wait for the final load test to complete
+        try:
+            await asyncio.wait_for(final_load_complete, timeout=5.0)
+        except TimeoutError:
+            pytest.fail("Final load test did not complete within timeout")

tests/integration/test_scheduler_pool.py

@@ -48,6 +48,7 @@ async def test_scheduler_pool(
         4: loop.create_future(),
         5: loop.create_future(),
         6: loop.create_future(),
+        7: loop.create_future(),
     }
 
     def check_output(line: str) -> None:
@@ -69,9 +70,10 @@ async def test_scheduler_pool(
             new_alloc_count += 1
 
         # Track phase completion
-        for phase_num in range(1, 7):
+        for phase_num in range(1, 8):
             if (
                 f"Phase {phase_num} complete" in line
+                and phase_num in phase_futures
                 and not phase_futures[phase_num].done()
             ):
                 phase_futures[phase_num].set_result(True)
@@ -102,6 +104,7 @@ async def test_scheduler_pool(
             "run_phase_4",
             "run_phase_5",
             "run_phase_6",
+            "run_phase_7",
             "run_complete",
         }
         assert expected_services.issubset(service_names), (
@@ -111,7 +114,7 @@ async def test_scheduler_pool(
         # Get service objects
         phase_services = {
             num: next(s for s in services if s.name == f"run_phase_{num}")
-            for num in range(1, 7)
+            for num in range(1, 8)
         }
         complete_service = next(s for s in services if s.name == "run_complete")
 
@@ -146,6 +149,11 @@ async def test_scheduler_pool(
             await asyncio.wait_for(phase_futures[6], timeout=1.0)
             await asyncio.sleep(0.1)  # Let Phase 6 timeouts complete
 
+            # Phase 7: Same-named defer optimization
+            client.execute_service(phase_services[7], {})
+            await asyncio.wait_for(phase_futures[7], timeout=1.0)
+            await asyncio.sleep(0.05)  # Let the single defer execute
+
             # Complete test
             client.execute_service(complete_service, {})
             await asyncio.wait_for(test_complete_future, timeout=0.5)
@@ -166,7 +174,7 @@ async def test_scheduler_pool(
             )
 
     # Verify all test phases ran
-    for phase_num in range(1, 7):
+    for phase_num in range(1, 8):
         assert phase_futures[phase_num].done(), f"Phase {phase_num} did not complete"
 
     # Verify pool behavior
@@ -180,8 +188,8 @@ async def test_scheduler_pool(
                 size = int(match.group(1))
                 max_pool_size = max(max_pool_size, size)
 
-        # Pool can grow up to its maximum of 10
+        # Pool can grow up to its maximum of 5
-        assert max_pool_size <= 10, f"Pool grew beyond maximum ({max_pool_size})"
+        assert max_pool_size <= 5, f"Pool grew beyond maximum ({max_pool_size})"
 
     # Log summary for debugging
     print("\nScheduler Pool Test Summary (Python Orchestrated):")

tests/unit_tests/test_coroutine.py

@@ -0,0 +1,204 @@
+"""Tests for the coroutine module."""
+
+import pytest
+
+from esphome.coroutine import CoroPriority, FakeEventLoop, coroutine_with_priority
+
+
+def test_coro_priority_enum_values() -> None:
+    """Test that CoroPriority enum values match expected priorities."""
+    assert CoroPriority.PLATFORM == 1000
+    assert CoroPriority.NETWORK == 201
+    assert CoroPriority.NETWORK_TRANSPORT == 200
+    assert CoroPriority.CORE == 100
+    assert CoroPriority.DIAGNOSTICS == 90
+    assert CoroPriority.STATUS == 80
+    assert CoroPriority.COMMUNICATION == 60
+    assert CoroPriority.APPLICATION == 50
+    assert CoroPriority.WEB == 40
+    assert CoroPriority.AUTOMATION == 30
+    assert CoroPriority.BUS == 1
+    assert CoroPriority.COMPONENT == 0
+    assert CoroPriority.LATE == -100
+    assert CoroPriority.WORKAROUNDS == -999
+    assert CoroPriority.FINAL == -1000
+
+
+def test_coroutine_with_priority_accepts_float() -> None:
+    """Test that coroutine_with_priority accepts float values."""
+
+    @coroutine_with_priority(100.0)
+    def test_func() -> None:
+        pass
+
+    assert hasattr(test_func, "priority")
+    assert test_func.priority == 100.0
+
+
+def test_coroutine_with_priority_accepts_enum() -> None:
+    """Test that coroutine_with_priority accepts CoroPriority enum values."""
+
+    @coroutine_with_priority(CoroPriority.CORE)
+    def test_func() -> None:
+        pass
+
+    assert hasattr(test_func, "priority")
+    assert test_func.priority == 100.0
+
+
+def test_float_and_enum_are_interchangeable() -> None:
+    """Test that float and CoroPriority enum values produce the same priority."""
+
+    @coroutine_with_priority(100.0)
+    def func_with_float() -> None:
+        pass
+
+    @coroutine_with_priority(CoroPriority.CORE)
+    def func_with_enum() -> None:
+        pass
+
+    assert func_with_float.priority == func_with_enum.priority
+    assert func_with_float.priority == 100.0
+
+
+@pytest.mark.parametrize(
+    ("enum_value", "float_value"),
+    [
+        (CoroPriority.PLATFORM, 1000.0),
+        (CoroPriority.NETWORK, 201.0),
+        (CoroPriority.NETWORK_TRANSPORT, 200.0),
+        (CoroPriority.CORE, 100.0),
+        (CoroPriority.DIAGNOSTICS, 90.0),
+        (CoroPriority.STATUS, 80.0),
+        (CoroPriority.COMMUNICATION, 60.0),
+        (CoroPriority.APPLICATION, 50.0),
+        (CoroPriority.WEB, 40.0),
+        (CoroPriority.AUTOMATION, 30.0),
+        (CoroPriority.BUS, 1.0),
+        (CoroPriority.COMPONENT, 0.0),
+        (CoroPriority.LATE, -100.0),
+        (CoroPriority.WORKAROUNDS, -999.0),
+        (CoroPriority.FINAL, -1000.0),
+    ],
+)
+def test_all_priority_values_are_interchangeable(
+    enum_value: CoroPriority, float_value: float
+) -> None:
+    """Test that all CoroPriority values work correctly with coroutine_with_priority."""
+
+    @coroutine_with_priority(enum_value)
+    def func_with_enum() -> None:
+        pass
+
+    @coroutine_with_priority(float_value)
+    def func_with_float() -> None:
+        pass
+
+    assert func_with_enum.priority == float_value
+    assert func_with_float.priority == float_value
+    assert func_with_enum.priority == func_with_float.priority
+
+
+def test_execution_order_with_enum_priorities() -> None:
+    """Test that execution order is correct when using enum priorities."""
+    execution_order: list[str] = []
+
+    @coroutine_with_priority(CoroPriority.PLATFORM)
+    async def platform_func() -> None:
+        execution_order.append("platform")
+
+    @coroutine_with_priority(CoroPriority.CORE)
+    async def core_func() -> None:
+        execution_order.append("core")
+
+    @coroutine_with_priority(CoroPriority.FINAL)
+    async def final_func() -> None:
+        execution_order.append("final")
+
+    # Create event loop and add jobs
+    loop = FakeEventLoop()
+    loop.add_job(platform_func)
+    loop.add_job(core_func)
+    loop.add_job(final_func)
+
+    # Run all tasks
+    loop.flush_tasks()
+
+    # Check execution order (higher priority runs first)
+    assert execution_order == ["platform", "core", "final"]
+
+
+def test_mixed_float_and_enum_priorities() -> None:
+    """Test that mixing float and enum priorities works correctly."""
+    execution_order: list[str] = []
+
+    @coroutine_with_priority(1000.0)  # Same as PLATFORM
+    async def func1() -> None:
+        execution_order.append("func1")
+
+    @coroutine_with_priority(CoroPriority.CORE)
+    async def func2() -> None:
+        execution_order.append("func2")
+
+    @coroutine_with_priority(-1000.0)  # Same as FINAL
+    async def func3() -> None:
+        execution_order.append("func3")
+
+    # Create event loop and add jobs
+    loop = FakeEventLoop()
+    loop.add_job(func2)
+    loop.add_job(func3)
+    loop.add_job(func1)
+
+    # Run all tasks
+    loop.flush_tasks()
+
+    # Check execution order
+    assert execution_order == ["func1", "func2", "func3"]
+
+
+def test_enum_priority_comparison() -> None:
+    """Test that enum priorities can be compared directly."""
+    assert CoroPriority.PLATFORM > CoroPriority.NETWORK
+    assert CoroPriority.NETWORK > CoroPriority.NETWORK_TRANSPORT
+    assert CoroPriority.NETWORK_TRANSPORT > CoroPriority.CORE
+    assert CoroPriority.CORE > CoroPriority.DIAGNOSTICS
+    assert CoroPriority.DIAGNOSTICS > CoroPriority.STATUS
+    assert CoroPriority.STATUS > CoroPriority.COMMUNICATION
+    assert CoroPriority.COMMUNICATION > CoroPriority.APPLICATION
+    assert CoroPriority.APPLICATION > CoroPriority.WEB
+    assert CoroPriority.WEB > CoroPriority.AUTOMATION
+    assert CoroPriority.AUTOMATION > CoroPriority.BUS
+    assert CoroPriority.BUS > CoroPriority.COMPONENT
+    assert CoroPriority.COMPONENT > CoroPriority.LATE
+    assert CoroPriority.LATE > CoroPriority.WORKAROUNDS
+    assert CoroPriority.WORKAROUNDS > CoroPriority.FINAL
+
+
+def test_custom_priority_between_enum_values() -> None:
+    """Test that custom float priorities between enum values work correctly."""
+    execution_order: list[str] = []
+
+    @coroutine_with_priority(CoroPriority.CORE)  # 100
+    async def core_func() -> None:
+        execution_order.append("core")
+
+    @coroutine_with_priority(95.0)  # Between CORE and DIAGNOSTICS
+    async def custom_func() -> None:
+        execution_order.append("custom")
+
+    @coroutine_with_priority(CoroPriority.DIAGNOSTICS)  # 90
+    async def diag_func() -> None:
+        execution_order.append("diagnostics")
+
+    # Create event loop and add jobs
+    loop = FakeEventLoop()
+    loop.add_job(diag_func)
+    loop.add_job(core_func)
+    loop.add_job(custom_func)
+
+    # Run all tasks
+    loop.flush_tasks()
+
+    # Check execution order
+    assert execution_order == ["core", "custom", "diagnostics"]