diff --git a/esphome/__main__.py b/esphome/__main__.py
index 8c80dab90af..781bcd62885 100644
--- a/esphome/__main__.py
+++ b/esphome/__main__.py
@@ -21,7 +21,7 @@ import argcomplete
 # Note: Do not import modules from esphome.components here, as this would
 # cause them to be loaded before external components are processed, resulting
 # in the built-in version being used instead of the external component one.
-from esphome import const, writer, yaml_util
+from esphome import const
 import esphome.codegen as cg
 from esphome.config import iter_component_configs, read_config, strip_default_ids
 from esphome.const import (
@@ -72,7 +72,12 @@ from esphome.util import (
     run_external_process,
     safe_print,
 )
-from esphome.zeroconf import discover_mdns_devices
+
+# Keep expensive imports (zeroconf, writer, yaml_util, etc.) out of this
+# module's top level. Every `esphome` invocation — including fast paths
+# like `esphome version` — pays the cost of what's imported here before
+# any command runs. Import inside the function that needs it instead.
+# `script/check_import_time.py` enforces a budget in CI.
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -241,6 +246,8 @@ def _discover_mac_suffix_devices() -> list[str] | None:
     """
     if not (has_name_add_mac_suffix() and has_mdns() and has_non_ip_address()):
         return None
+    from esphome.zeroconf import discover_mdns_devices
+
     _LOGGER.info("Discovering devices...")
     if not (discovered := discover_mdns_devices(CORE.name)):
         _LOGGER.warning(
@@ -660,7 +667,7 @@ def run_miniterm(config: ConfigType, port: str, args) -> int:
     return 0
 
 
-def wrap_to_code(name, comp):
+def _wrap_to_code(name, comp, yaml_util):
     coro = coroutine(comp.to_code)
 
     @functools.wraps(comp.to_code)
@@ -680,6 +687,8 @@ def wrap_to_code(name, comp):
 
 
 def write_cpp(config: ConfigType, native_idf: bool = False) -> int:
+    from esphome import writer
+
     if not get_bool_env(ENV_NOGITIGNORE):
         writer.write_gitignore()
 
@@ -691,17 +700,21 @@ def write_cpp(config: ConfigType, native_idf: bool = False) -> int:
 
 
 def generate_cpp_contents(config: ConfigType) -> None:
+    from esphome import yaml_util
+
     _LOGGER.info("Generating C++ source...")
 
     for name, component, conf in iter_component_configs(CORE.config):
         if component.to_code is not None:
-            coro = wrap_to_code(name, component)
+            coro = _wrap_to_code(name, component, yaml_util)
             CORE.add_job(coro, conf)
 
     CORE.flush_tasks()
 
 
 def write_cpp_file(native_idf: bool = False) -> int:
+    from esphome import writer
+
     code_s = indent(CORE.cpp_main_section)
     writer.write_cpp(code_s)
 
@@ -1180,6 +1193,8 @@ def command_wizard(args: ArgsProtocol) -> int | None:
 
 
 def command_config(args: ArgsProtocol, config: ConfigType) -> int | None:
+    from esphome import yaml_util
+
     if not CORE.verbose:
         config = strip_default_ids(config)
     output = yaml_util.dump(config, args.show_secrets)
@@ -1321,6 +1336,8 @@ def command_clean_mqtt(args: ArgsProtocol, config: ConfigType) -> int | None:
 
 
 def command_clean_all(args: ArgsProtocol) -> int | None:
+    from esphome import writer
+
     try:
         writer.clean_all(args.configuration)
     except OSError as err:
@@ -1336,6 +1353,8 @@ def command_version(args: ArgsProtocol) -> int | None:
 
 
 def command_clean(args: ArgsProtocol, config: ConfigType) -> int | None:
+    from esphome import writer
+
     try:
         writer.clean_build()
     except OSError as err:
@@ -1538,6 +1557,8 @@ def command_analyze_memory(args: ArgsProtocol, config: ConfigType) -> int:
 
 
 def command_rename(args: ArgsProtocol, config: ConfigType) -> int | None:
+    from esphome import yaml_util
+
     new_name = args.name
     for c in new_name:
         if c not in ALLOWED_NAME_CHARS:
diff --git a/esphome/components/ens160_base/ens160_base.cpp b/esphome/components/ens160_base/ens160_base.cpp
index e1cee5005cb..42baa68b35d 100644
--- a/esphome/components/ens160_base/ens160_base.cpp
+++ b/esphome/components/ens160_base/ens160_base.cpp
@@ -5,6 +5,15 @@
 // Implementation based on:
 //   https://github.com/sciosense/ENS160_driver
 
+// For best performance, the sensor shall be operated in normal indoor air in the range -5 to 60°C
+// (typical: 25°C); relative humidity: 20 to 80%RH (typical: 50%RH), non-condensing with no aggressive
+// or poisonous gases present. Prolonged exposure to environments outside these conditions can affect
+// performance and lifetime of the sensor.
+// The sensor is designed for indoor use and is not waterproof or dustproof. It should be protected from
+// water, condensation, dust, and aggressive gases. Note that the status will only be stored in non-volatile
+// memory after an initial 24 h of continuous operation. If unpowered before the conclusion of that period,
+// the ENS160 will resume "Initial Start-up" mode after re-powering.
+
 #include "ens160_base.h"
 #include "esphome/core/log.h"
 #include "esphome/core/hal.h"
@@ -14,7 +23,9 @@ namespace ens160_base {
 
 static const char *const TAG = "ens160";
 
-static const uint8_t ENS160_BOOTING = 10;
+// Datasheet specifies 10ms, but some users report that 10ms is not sufficient for the
+// sensor to boot and be ready for commands. 11ms seems to be a safe value.
+static const uint8_t ENS160_BOOTING = 11;
 
 static const uint16_t ENS160_PART_ID = 0x0160;
 
@@ -91,6 +102,8 @@ void ENS160Component::setup() {
     this->mark_failed();
     return;
   }
+  delay(ENS160_BOOTING);
+
   // clear command
   if (!this->write_byte(ENS160_REG_COMMAND, ENS160_COMMAND_NOP)) {
     this->error_code_ = WRITE_FAILED;
@@ -102,6 +115,7 @@ void ENS160Component::setup() {
     this->mark_failed();
     return;
   }
+  delay(ENS160_BOOTING);
 
   // read firmware version
   if (!this->write_byte(ENS160_REG_COMMAND, ENS160_COMMAND_GET_APPVER)) {
@@ -109,6 +123,8 @@ void ENS160Component::setup() {
     this->mark_failed();
     return;
   }
+  delay(ENS160_BOOTING);
+
   uint8_t version_data[3];
   if (!this->read_bytes(ENS160_REG_GPR_READ_4, version_data, 3)) {
     this->error_code_ = READ_FAILED;
@@ -223,7 +239,6 @@ void ENS160Component::update() {
   if (this->aqi_ != nullptr) {
     // remove reserved bits, just in case they are used in future
     data_aqi = ENS160_DATA_AQI & data_aqi;
-
     this->aqi_->publish_state(data_aqi);
   }
 
diff --git a/esphome/components/esp32/core.cpp b/esphome/components/esp32/core.cpp
index 1c631371836..4886745c068 100644
--- a/esphome/components/esp32/core.cpp
+++ b/esphome/components/esp32/core.cpp
@@ -22,7 +22,7 @@ extern "C" __attribute__((weak)) void initArduino() {}
 
 namespace esphome {
 
-void HOT yield() { vPortYield(); }
+// yield(), delay(), micros(), millis_64() inlined in hal.h.
 // Use xTaskGetTickCount() when tick rate is 1 kHz (ESPHome's default via sdkconfig),
 // falling back to esp_timer for non-standard rates. IRAM_ATTR is required because
 // Wiegand and ZyAura call millis() from IRAM_ATTR ISR handlers on ESP32.
@@ -37,15 +37,6 @@ uint32_t IRAM_ATTR HOT millis() {
   return micros_to_millis(static_cast<uint64_t>(esp_timer_get_time()));
 #endif
 }
-// millis_64() stays on esp_timer — a different clock from xTaskGetTickCount(). This is
-// safe because the two are never cross-compared: millis() values are only used for
-// millis()-vs-millis() deltas (feed_wdt, warn_blocking, component start time), while
-// millis_64() is used by the Scheduler and uptime sensors. On ESP32 (USE_NATIVE_64BIT_TIME),
-// Scheduler::millis_64_from_(now) discards the 32-bit now and calls millis_64() directly,
-// so the Scheduler is internally consistent on the esp_timer clock.
-uint64_t HOT millis_64() { return micros_to_millis<uint64_t>(static_cast<uint64_t>(esp_timer_get_time())); }
-void HOT delay(uint32_t ms) { vTaskDelay(ms / portTICK_PERIOD_MS); }
-uint32_t IRAM_ATTR HOT micros() { return (uint32_t) esp_timer_get_time(); }
 void IRAM_ATTR HOT delayMicroseconds(uint32_t us) { delay_microseconds_safe(us); }
 void arch_restart() {
   esp_restart();
diff --git a/esphome/components/esp8266/core.cpp b/esphome/components/esp8266/core.cpp
index c9bedb61be4..9161ca6aaf5 100644
--- a/esphome/components/esp8266/core.cpp
+++ b/esphome/components/esp8266/core.cpp
@@ -15,7 +15,7 @@ extern "C" {
 
 namespace esphome {
 
-void HOT yield() { ::yield(); }
+// yield(), micros(), millis_64() inlined in hal.h.
 // Fast accumulator replacement for Arduino's millis() (~3.3 μs via 4× 64-bit
 // multiplies on the LX106). Tracks a running ms counter from 32-bit
 // system_get_time() deltas using pure 32-bit ops. Installed as __wrap_millis
@@ -66,7 +66,6 @@ uint32_t IRAM_ATTR HOT millis() {
   xt_wsr_ps(ps);
   return result;
 }
-uint64_t millis_64() { return Millis64Impl::compute(millis()); }
 // Poll-based delay that avoids ::delay() — Arduino's __delay has an intra-object
 // call to the original millis() that --wrap can't intercept, so calling ::delay()
 // would keep the slow Arduino millis body alive in IRAM. optimistic_yield still
@@ -85,8 +84,7 @@ void HOT delay(uint32_t ms) {
     optimistic_yield(1000);
   }
 }
-uint32_t IRAM_ATTR HOT micros() { return ::micros(); }
-void IRAM_ATTR HOT delayMicroseconds(uint32_t us) { delay_microseconds_safe(us); }
+// delayMicroseconds(), arch_feed_wdt(), and progmem_read_*() are inlined in hal/hal_esp8266.h.
 void arch_restart() {
   system_restart();
   // restart() doesn't always end execution
@@ -95,17 +93,6 @@ void arch_restart() {
   }
 }
 void arch_init() {}
-void HOT arch_feed_wdt() { system_soft_wdt_feed(); }
-
-uint8_t progmem_read_byte(const uint8_t *addr) {
-  return pgm_read_byte(addr);  // NOLINT
-}
-const char *progmem_read_ptr(const char *const *addr) {
-  return reinterpret_cast<const char *>(pgm_read_ptr(addr));  // NOLINT
-}
-uint16_t progmem_read_uint16(const uint16_t *addr) {
-  return pgm_read_word(addr);  // NOLINT
-}
 uint32_t IRAM_ATTR HOT arch_get_cpu_cycle_count() { return esp_get_cycle_count(); }
 uint32_t arch_get_cpu_freq_hz() { return F_CPU; }
 
diff --git a/esphome/components/espnow/__init__.py b/esphome/components/espnow/__init__.py
index a9624734d02..7861c0affa1 100644
--- a/esphome/components/espnow/__init__.py
+++ b/esphome/components/espnow/__init__.py
@@ -26,9 +26,9 @@ espnow_ns = cg.esphome_ns.namespace("espnow")
 ESPNowComponent = espnow_ns.class_("ESPNowComponent", cg.Component)
 
 # Handler interfaces that other components can use to register callbacks
-ESPNowReceivedPacketHandler = espnow_ns.class_("ESPNowReceivedPacketHandler")
+ESPNowReceivePacketHandler = espnow_ns.class_("ESPNowReceivePacketHandler")
 ESPNowUnknownPeerHandler = espnow_ns.class_("ESPNowUnknownPeerHandler")
-ESPNowBroadcastedHandler = espnow_ns.class_("ESPNowBroadcastedHandler")
+ESPNowBroadcastHandler = espnow_ns.class_("ESPNowBroadcastHandler")
 
 ESPNowRecvInfo = espnow_ns.class_("ESPNowRecvInfo")
 ESPNowRecvInfoConstRef = ESPNowRecvInfo.operator("const").operator("ref")
@@ -48,10 +48,10 @@ OnUnknownPeerTrigger = espnow_ns.class_(
     "OnUnknownPeerTrigger", ESPNowHandlerTrigger, ESPNowUnknownPeerHandler
 )
 OnReceiveTrigger = espnow_ns.class_(
-    "OnReceiveTrigger", ESPNowHandlerTrigger, ESPNowReceivedPacketHandler
+    "OnReceiveTrigger", ESPNowHandlerTrigger, ESPNowReceivePacketHandler
 )
-OnBroadcastedTrigger = espnow_ns.class_(
-    "OnBroadcastedTrigger", ESPNowHandlerTrigger, ESPNowBroadcastedHandler
+OnBroadcastTrigger = espnow_ns.class_(
+    "OnBroadcastTrigger", ESPNowHandlerTrigger, ESPNowBroadcastHandler
 )
 
 
@@ -94,7 +94,7 @@ CONFIG_SCHEMA = cv.All(
             ),
             cv.Optional(CONF_ON_BROADCAST): automation.validate_automation(
                 {
-                    cv.GenerateID(CONF_TRIGGER_ID): cv.declare_id(OnBroadcastedTrigger),
+                    cv.GenerateID(CONF_TRIGGER_ID): cv.declare_id(OnBroadcastTrigger),
                     cv.Optional(CONF_ADDRESS): cv.mac_address,
                 }
             ),
@@ -140,11 +140,11 @@ async def to_code(config):
 
     for on_receive in config.get(CONF_ON_RECEIVE, []):
         trigger = await _trigger_to_code(on_receive)
-        cg.add(var.register_received_handler(trigger))
+        cg.add(var.register_receive_handler(trigger))
 
     for on_receive in config.get(CONF_ON_BROADCAST, []):
         trigger = await _trigger_to_code(on_receive)
-        cg.add(var.register_broadcasted_handler(trigger))
+        cg.add(var.register_broadcast_handler(trigger))
 
 
 # ========================================== A C T I O N S ================================================
diff --git a/esphome/components/espnow/automation.h b/esphome/components/espnow/automation.h
index 0fbb14e3888..9c3c55e4eff 100644
--- a/esphome/components/espnow/automation.h
+++ b/esphome/components/espnow/automation.h
@@ -67,6 +67,7 @@ template<typename... Ts> class SendAction : public Action<Ts...>, public Parente
     }
   }
 
+ protected:
   void play(const Ts &...x) override { /* ignore - see play_complex */
   }
 
@@ -75,7 +76,6 @@ template<typename... Ts> class SendAction : public Action<Ts...>, public Parente
     this->error_.stop();
   }
 
- protected:
   ActionList<Ts...> sent_;
   ActionList<Ts...> error_;
 
@@ -89,7 +89,7 @@ template<typename... Ts> class SendAction : public Action<Ts...>, public Parente
 template<typename... Ts> class AddPeerAction : public Action<Ts...>, public Parented<ESPNowComponent> {
   TEMPLATABLE_VALUE(peer_address_t, address);
 
- public:
+ protected:
   void play(const Ts &...x) override {
     peer_address_t address = this->address_.value(x...);
     this->parent_->add_peer(address.data());
@@ -99,7 +99,7 @@ template<typename... Ts> class AddPeerAction : public Action<Ts...>, public Pare
 template<typename... Ts> class DeletePeerAction : public Action<Ts...>, public Parented<ESPNowComponent> {
   TEMPLATABLE_VALUE(peer_address_t, address);
 
- public:
+ protected:
   void play(const Ts &...x) override {
     peer_address_t address = this->address_.value(x...);
     this->parent_->del_peer(address.data());
@@ -107,8 +107,9 @@ template<typename... Ts> class DeletePeerAction : public Action<Ts...>, public P
 };
 
 template<typename... Ts> class SetChannelAction : public Action<Ts...>, public Parented<ESPNowComponent> {
- public:
   TEMPLATABLE_VALUE(uint8_t, channel)
+
+ protected:
   void play(const Ts &...x) override {
     if (this->parent_->is_wifi_enabled()) {
       return;
@@ -125,9 +126,9 @@ class OnReceiveTrigger : public Trigger<const ESPNowRecvInfo &, const uint8_t *,
     memcpy(this->address_, address.data(), ESP_NOW_ETH_ALEN);
   }
 
-  explicit OnReceiveTrigger() : has_address_(false) {}
+  explicit OnReceiveTrigger() {}
 
-  bool on_received(const ESPNowRecvInfo &info, const uint8_t *data, uint8_t size) override {
+  bool on_receive(const ESPNowRecvInfo &info, const uint8_t *data, uint8_t size) override {
     bool match = !this->has_address_ || (memcmp(this->address_, info.src_addr, ESP_NOW_ETH_ALEN) == 0);
     if (!match)
       return false;
@@ -138,7 +139,7 @@ class OnReceiveTrigger : public Trigger<const ESPNowRecvInfo &, const uint8_t *,
 
  protected:
   bool has_address_{false};
-  uint8_t address_[ESP_NOW_ETH_ALEN];
+  uint8_t address_[ESP_NOW_ETH_ALEN]{};
 };
 class OnUnknownPeerTrigger : public Trigger<const ESPNowRecvInfo &, const uint8_t *, uint8_t>,
                              public ESPNowUnknownPeerHandler {
@@ -148,15 +149,15 @@ class OnUnknownPeerTrigger : public Trigger<const ESPNowRecvInfo &, const uint8_
     return false;  // Return false to continue processing other internal handlers
   }
 };
-class OnBroadcastedTrigger : public Trigger<const ESPNowRecvInfo &, const uint8_t *, uint8_t>,
-                             public ESPNowBroadcastedHandler {
+class OnBroadcastTrigger : public Trigger<const ESPNowRecvInfo &, const uint8_t *, uint8_t>,
+                           public ESPNowBroadcastHandler {
  public:
-  explicit OnBroadcastedTrigger(std::array<uint8_t, ESP_NOW_ETH_ALEN> address) : has_address_(true) {
+  explicit OnBroadcastTrigger(std::array<uint8_t, ESP_NOW_ETH_ALEN> address) : has_address_(true) {
     memcpy(this->address_, address.data(), ESP_NOW_ETH_ALEN);
   }
-  explicit OnBroadcastedTrigger() : has_address_(false) {}
+  explicit OnBroadcastTrigger() {}
 
-  bool on_broadcasted(const ESPNowRecvInfo &info, const uint8_t *data, uint8_t size) override {
+  bool on_broadcast(const ESPNowRecvInfo &info, const uint8_t *data, uint8_t size) override {
     bool match = !this->has_address_ || (memcmp(this->address_, info.src_addr, ESP_NOW_ETH_ALEN) == 0);
     if (!match)
       return false;
@@ -167,7 +168,7 @@ class OnBroadcastedTrigger : public Trigger<const ESPNowRecvInfo &, const uint8_
 
  protected:
   bool has_address_{false};
-  uint8_t address_[ESP_NOW_ETH_ALEN];
+  uint8_t address_[ESP_NOW_ETH_ALEN]{};
 };
 
 }  // namespace esphome::espnow
diff --git a/esphome/components/espnow/espnow_component.cpp b/esphome/components/espnow/espnow_component.cpp
index 282287ca83d..91d44394e8d 100644
--- a/esphome/components/espnow/espnow_component.cpp
+++ b/esphome/components/espnow/espnow_component.cpp
@@ -299,13 +299,13 @@ void ESPNowComponent::loop() {
                    format_hex_pretty_to(hex_buf, packet->packet_.receive.data, packet->packet_.receive.size));
 #endif
           if (memcmp(info.des_addr, ESPNOW_BROADCAST_ADDR, ESP_NOW_ETH_ALEN) == 0) {
-            for (auto *handler : this->broadcasted_handlers_) {
-              if (handler->on_broadcasted(info, packet->packet_.receive.data, packet->packet_.receive.size))
+            for (auto *handler : this->broadcast_handlers_) {
+              if (handler->on_broadcast(info, packet->packet_.receive.data, packet->packet_.receive.size))
                 break;  // If a handler returns true, stop processing further handlers
             }
           } else {
-            for (auto *handler : this->received_handlers_) {
-              if (handler->on_received(info, packet->packet_.receive.data, packet->packet_.receive.size))
+            for (auto *handler : this->receive_handlers_) {
+              if (handler->on_receive(info, packet->packet_.receive.data, packet->packet_.receive.size))
                 break;  // If a handler returns true, stop processing further handlers
             }
           }
diff --git a/esphome/components/espnow/espnow_component.h b/esphome/components/espnow/espnow_component.h
index ee4adc1b4d4..ff9581ec2ff 100644
--- a/esphome/components/espnow/espnow_component.h
+++ b/esphome/components/espnow/espnow_component.h
@@ -31,8 +31,8 @@ using peer_address_t = std::array<uint8_t, ESP_NOW_ETH_ALEN>;
 enum class ESPNowTriggers : uint8_t {
   TRIGGER_NONE = 0,
   ON_NEW_PEER = 1,
-  ON_RECEIVED = 2,
-  ON_BROADCASTED = 3,
+  ON_RECEIVE = 2,
+  ON_BROADCAST = 3,
   ON_SUCCEED = 10,
   ON_FAILED = 11,
 };
@@ -74,18 +74,18 @@ class ESPNowReceivedPacketHandler {
   /// @param data Pointer to the received data payload
   /// @param size Size of the received data in bytes
   /// @return true if the packet was handled, false otherwise
-  virtual bool on_received(const ESPNowRecvInfo &info, const uint8_t *data, uint8_t size) = 0;
+  virtual bool on_receive(const ESPNowRecvInfo &info, const uint8_t *data, uint8_t size) = 0;
 };
-/// Handler interface for receiving broadcasted ESPNow packets
+/// Handler interface for receiving ESPNow broadcast packets
 /// Components should inherit from this class to handle incoming ESPNow data
-class ESPNowBroadcastedHandler {
+class ESPNowBroadcastHandler {
  public:
-  /// Called when a broadcasted ESPNow packet is received
+  /// Called when an ESPNow broadcast packet is received
   /// @param info Information about the received packet (sender MAC, etc.)
   /// @param data Pointer to the received data payload
   /// @param size Size of the received data in bytes
   /// @return true if the packet was handled, false otherwise
-  virtual bool on_broadcasted(const ESPNowRecvInfo &info, const uint8_t *data, uint8_t size) = 0;
+  virtual bool on_broadcast(const ESPNowRecvInfo &info, const uint8_t *data, uint8_t size) = 0;
 };
 
 class ESPNowComponent : public Component {
@@ -136,13 +136,11 @@ class ESPNowComponent : public Component {
   esp_err_t send(const uint8_t *peer_address, const uint8_t *payload, size_t size,
                  const send_callback_t &callback = nullptr);
 
-  void register_received_handler(ESPNowReceivedPacketHandler *handler) { this->received_handlers_.push_back(handler); }
+  void register_receive_handler(ESPNowReceivedPacketHandler *handler) { this->receive_handlers_.push_back(handler); }
   void register_unknown_peer_handler(ESPNowUnknownPeerHandler *handler) {
     this->unknown_peer_handlers_.push_back(handler);
   }
-  void register_broadcasted_handler(ESPNowBroadcastedHandler *handler) {
-    this->broadcasted_handlers_.push_back(handler);
-  }
+  void register_broadcast_handler(ESPNowBroadcastHandler *handler) { this->broadcast_handlers_.push_back(handler); }
 
  protected:
   friend void on_data_received(const esp_now_recv_info_t *info, const uint8_t *data, int size);
@@ -156,8 +154,8 @@ class ESPNowComponent : public Component {
   void send_();
 
   std::vector<ESPNowUnknownPeerHandler *> unknown_peer_handlers_;
-  std::vector<ESPNowReceivedPacketHandler *> received_handlers_;
-  std::vector<ESPNowBroadcastedHandler *> broadcasted_handlers_;
+  std::vector<ESPNowReceivedPacketHandler *> receive_handlers_;
+  std::vector<ESPNowBroadcastHandler *> broadcast_handlers_;
 
   std::vector<ESPNowPeer> peers_{};
 
diff --git a/esphome/components/espnow/packet_transport/espnow_transport.cpp b/esphome/components/espnow/packet_transport/espnow_transport.cpp
index 6e4f606466f..384e3fe2a9a 100644
--- a/esphome/components/espnow/packet_transport/espnow_transport.cpp
+++ b/esphome/components/espnow/packet_transport/espnow_transport.cpp
@@ -26,10 +26,10 @@ void ESPNowTransport::setup() {
            this->peer_address_[5]);
 
   // Register received handler
-  this->parent_->register_received_handler(this);
+  this->parent_->register_receive_handler(this);
 
-  // Register broadcasted handler
-  this->parent_->register_broadcasted_handler(this);
+  // Register broadcast handler
+  this->parent_->register_broadcast_handler(this);
 }
 
 void ESPNowTransport::send_packet(const std::vector<uint8_t> &buf) const {
@@ -56,7 +56,7 @@ void ESPNowTransport::send_packet(const std::vector<uint8_t> &buf) const {
   });
 }
 
-bool ESPNowTransport::on_received(const ESPNowRecvInfo &info, const uint8_t *data, uint8_t size) {
+bool ESPNowTransport::on_receive(const ESPNowRecvInfo &info, const uint8_t *data, uint8_t size) {
   ESP_LOGV(TAG, "Received packet of size %u from %02X:%02X:%02X:%02X:%02X:%02X", size, info.src_addr[0],
            info.src_addr[1], info.src_addr[2], info.src_addr[3], info.src_addr[4], info.src_addr[5]);
 
@@ -71,7 +71,7 @@ bool ESPNowTransport::on_received(const ESPNowRecvInfo &info, const uint8_t *dat
   return false;  // Allow other handlers to run
 }
 
-bool ESPNowTransport::on_broadcasted(const ESPNowRecvInfo &info, const uint8_t *data, uint8_t size) {
+bool ESPNowTransport::on_broadcast(const ESPNowRecvInfo &info, const uint8_t *data, uint8_t size) {
   ESP_LOGV(TAG, "Received broadcast packet of size %u from %02X:%02X:%02X:%02X:%02X:%02X", size, info.src_addr[0],
            info.src_addr[1], info.src_addr[2], info.src_addr[3], info.src_addr[4], info.src_addr[5]);
 
diff --git a/esphome/components/espnow/packet_transport/espnow_transport.h b/esphome/components/espnow/packet_transport/espnow_transport.h
index d85119db7dd..98c33f01fdb 100644
--- a/esphome/components/espnow/packet_transport/espnow_transport.h
+++ b/esphome/components/espnow/packet_transport/espnow_transport.h
@@ -15,7 +15,7 @@ namespace espnow {
 class ESPNowTransport : public packet_transport::PacketTransport,
                         public Parented<ESPNowComponent>,
                         public ESPNowReceivedPacketHandler,
-                        public ESPNowBroadcastedHandler {
+                        public ESPNowBroadcastHandler {
  public:
   void setup() override;
   float get_setup_priority() const override { return setup_priority::AFTER_WIFI; }
@@ -25,8 +25,8 @@ class ESPNowTransport : public packet_transport::PacketTransport,
   }
 
   // ESPNow handler interface
-  bool on_received(const ESPNowRecvInfo &info, const uint8_t *data, uint8_t size) override;
-  bool on_broadcasted(const ESPNowRecvInfo &info, const uint8_t *data, uint8_t size) override;
+  bool on_receive(const ESPNowRecvInfo &info, const uint8_t *data, uint8_t size) override;
+  bool on_broadcast(const ESPNowRecvInfo &info, const uint8_t *data, uint8_t size) override;
 
  protected:
   void send_packet(const std::vector<uint8_t> &buf) const override;
diff --git a/esphome/components/libretiny/core.cpp b/esphome/components/libretiny/core.cpp
index ca46bcb899e..f46abe3b81a 100644
--- a/esphome/components/libretiny/core.cpp
+++ b/esphome/components/libretiny/core.cpp
@@ -3,7 +3,6 @@
 #include "core.h"
 #include "esphome/core/defines.h"
 #include "esphome/core/hal.h"
-#include "esphome/core/time_64.h"
 #include "esphome/core/helpers.h"
 #include "preferences.h"
 
@@ -15,32 +14,7 @@ void loop();
 
 namespace esphome {
 
-void HOT yield() { ::yield(); }
-// Inline the tick read so esphome::millis() matches MillisInternal::get()'s fast
-// path instead of going through the Arduino core's out-of-line ::millis() wrapper.
-//
-// RTL87xx / LN882x (1 kHz): xTaskGetTickCount() is already ms. IRAM_ATTR + ISR
-// dispatch are needed because ISR handlers (e.g. rotary_encoder) call millis().
-//
-// BK72xx (500 Hz): ticks * portTICK_PERIOD_MS (== 2). IRAM_ATTR and ISR dispatch
-// are both unnecessary — the SDK masks FIQ + IRQ during flash writes (see hal.h),
-// so no ISR runs while flash is stalled.
-#if defined(USE_RTL87XX) || defined(USE_LN882X)
-uint32_t IRAM_ATTR HOT millis() {
-  static_assert(configTICK_RATE_HZ == 1000, "millis() fast path requires 1 kHz FreeRTOS tick");
-  return in_isr_context() ? xTaskGetTickCountFromISR() : xTaskGetTickCount();
-}
-#elif defined(USE_BK72XX)
-uint32_t HOT millis() {
-  static_assert(configTICK_RATE_HZ == 500, "BK72xx millis() fast path assumes 500 Hz FreeRTOS tick");
-  return xTaskGetTickCount() * portTICK_PERIOD_MS;
-}
-#else
-uint32_t IRAM_ATTR HOT millis() { return ::millis(); }
-#endif
-uint64_t millis_64() { return Millis64Impl::compute(millis()); }
-uint32_t IRAM_ATTR HOT micros() { return ::micros(); }
-void HOT delay(uint32_t ms) { ::delay(ms); }
+// yield(), delay(), micros(), millis(), millis_64() inlined in hal.h.
 void IRAM_ATTR HOT delayMicroseconds(uint32_t us) { ::delayMicroseconds(us); }
 
 void arch_init() {
diff --git a/esphome/components/remote_base/remote_base.h b/esphome/components/remote_base/remote_base.h
index d73fff2b0a5..e5e923d7806 100644
--- a/esphome/components/remote_base/remote_base.h
+++ b/esphome/components/remote_base/remote_base.h
@@ -164,7 +164,7 @@ class RemoteTransmitterBase : public RemoteComponentBase {
     return TransmitCall(this);
   }
   template<typename Protocol>
-  void transmit(const typename Protocol::ProtocolData &data, uint32_t send_times = 1, uint32_t send_wait = 0) {
+  void transmit(const Protocol::ProtocolData &data, uint32_t send_times = 1, uint32_t send_wait = 0) {
     auto call = this->transmit();
     Protocol().encode(call.get_data(), data);
     call.set_send_times(send_times);
@@ -250,10 +250,10 @@ template<typename T> class RemoteReceiverBinarySensor : public RemoteReceiverBin
   }
 
  public:
-  void set_data(typename T::ProtocolData data) { data_ = data; }
+  void set_data(T::ProtocolData data) { data_ = data; }
 
  protected:
-  typename T::ProtocolData data_;
+  T::ProtocolData data_;
 };
 
 template<typename T>
@@ -278,7 +278,7 @@ class RemoteTransmittable {
 
  protected:
   template<typename Protocol>
-  void transmit_(const typename Protocol::ProtocolData &data, uint32_t send_times = 1, uint32_t send_wait = 0) {
+  void transmit_(const Protocol::ProtocolData &data, uint32_t send_times = 1, uint32_t send_wait = 0) {
     this->transmitter_->transmit<Protocol>(data, send_times, send_wait);
   }
   RemoteTransmitterBase *transmitter_;
diff --git a/esphome/components/rp2040/core.cpp b/esphome/components/rp2040/core.cpp
index b7a90006123..d3dc1cf2bb5 100644
--- a/esphome/components/rp2040/core.cpp
+++ b/esphome/components/rp2040/core.cpp
@@ -13,11 +13,7 @@
 
 namespace esphome {
 
-void HOT yield() { ::yield(); }
-uint64_t millis_64() { return micros_to_millis<uint64_t>(time_us_64()); }
-uint32_t HOT millis() { return micros_to_millis(time_us_64()); }
-void HOT delay(uint32_t ms) { ::delay(ms); }
-uint32_t HOT micros() { return ::micros(); }
+// yield(), delay(), micros(), millis(), millis_64() inlined in hal.h.
 void HOT delayMicroseconds(uint32_t us) { delay_microseconds_safe(us); }
 void arch_restart() {
   watchdog_reboot(0, 0, 10);
diff --git a/esphome/components/watchdog/watchdog.cpp b/esphome/components/watchdog/watchdog.cpp
index 545d83a6791..edf113b0b4a 100644
--- a/esphome/components/watchdog/watchdog.cpp
+++ b/esphome/components/watchdog/watchdog.cpp
@@ -39,9 +39,18 @@ void WatchdogManager::set_timeout_(uint32_t timeout_ms) {
 #ifdef USE_ESP32
   esp_task_wdt_config_t wdt_config = {
       .timeout_ms = timeout_ms,
-      .idle_core_mask = (1U << CONFIG_FREERTOS_NUMBER_OF_CORES) - 1U,
-      .trigger_panic = true,
+      .idle_core_mask = 0,
+      .trigger_panic = false,
   };
+#if CONFIG_ESP_TASK_WDT_CHECK_IDLE_TASK_CPU0
+  wdt_config.idle_core_mask |= (1U << 0U);
+#endif
+#if CONFIG_ESP_TASK_WDT_CHECK_IDLE_TASK_CPU1
+  wdt_config.idle_core_mask |= (1U << 1U);
+#endif
+#if CONFIG_ESP_TASK_WDT_PANIC
+  wdt_config.trigger_panic = true;
+#endif
   esp_task_wdt_reconfigure(&wdt_config);
 #endif  // USE_ESP32
 
diff --git a/esphome/config.py b/esphome/config.py
index 6eb67af58b3..79d0d2b02ba 100644
--- a/esphome/config.py
+++ b/esphome/config.py
@@ -25,7 +25,10 @@ from esphome.const import (
     CONF_SUBSTITUTIONS,
 )
 from esphome.core import CORE, DocumentRange, EsphomeError
-import esphome.core.config as core_config
+
+# `esphome.core.config` is imported lazily at its two use sites below.
+# It pulls in `esphome.automation` and `esphome.config_validation`, which
+# dominate `esphome.__main__` startup cost when loaded eagerly here.
 import esphome.final_validate as fv
 from esphome.helpers import indent
 from esphome.loader import ComponentManifest, get_component, get_platform
@@ -968,6 +971,8 @@ class CoreFinalValidateStep(ConfigValidationStep):
         if result.errors:
             return
 
+        import esphome.core.config as core_config
+
         token = fv.full_config.set(result)
         with result.catch_error([CONF_ESPHOME]):
             if CONF_ESPHOME in result:
@@ -1073,6 +1078,8 @@ def validate_config(
         return result
 
     # 2. Load partial core config
+    import esphome.core.config as core_config
+
     result[CONF_ESPHOME] = config[CONF_ESPHOME]
     result.add_output_path([CONF_ESPHOME], CONF_ESPHOME)
     try:
diff --git a/esphome/core/finite_set_mask.h b/esphome/core/finite_set_mask.h
index 616c69353d7..272337ff76f 100644
--- a/esphome/core/finite_set_mask.h
+++ b/esphome/core/finite_set_mask.h
@@ -55,7 +55,7 @@ template<typename ValueType, int MaxBits> struct DefaultBitPolicy {
 ///
 template<typename ValueType, typename BitPolicy = DefaultBitPolicy<ValueType, 16>> class FiniteSetMask {
  public:
-  using bitmask_t = typename BitPolicy::mask_t;
+  using bitmask_t = BitPolicy::mask_t;
 
   constexpr FiniteSetMask() = default;
 
diff --git a/esphome/core/hal.h b/esphome/core/hal.h
index e4083622b98..e20797cf95d 100644
--- a/esphome/core/hal.h
+++ b/esphome/core/hal.h
@@ -2,125 +2,48 @@
 #include <string>
 #include <cstdint>
 #include "gpio.h"
+#include "esphome/core/defines.h"
+#include "esphome/core/time_64.h"
+#include "esphome/core/time_conversion.h"
 
+// Per-platform HAL bits (IRAM_ATTR / PROGMEM macros, in_isr_context(),
+// inline yield/delay/micros/millis/millis_64 wrappers, ESP8266 progmem
+// helpers) live under esphome/core/hal/ and are dispatched here based on
+// the active USE_* platform define. Each header guards its body with the
+// matching #ifdef USE_<platform> and re-enters namespace esphome {} so it
+// is safe to be re-included.
 #if defined(USE_ESP32)
-#include <esp_attr.h>
-#ifndef PROGMEM
-#define PROGMEM
-#endif
-
+#include "esphome/core/hal/hal_esp32.h"
 #elif defined(USE_ESP8266)
-
-#include <c_types.h>
-#ifndef PROGMEM
-#define PROGMEM ICACHE_RODATA_ATTR
-#endif
-
-#elif defined(USE_RP2040)
-
-#define IRAM_ATTR __attribute__((noinline, long_call, section(".time_critical")))
-#define PROGMEM
-
+#include "esphome/core/hal/hal_esp8266.h"
 #elif defined(USE_LIBRETINY)
-
-// IRAM_ATTR places a function in executable RAM so it is callable from an
-// ISR even while flash is busy (XIP stall, OTA, logger flash write).
-// Each family uses a section its stock linker already routes to RAM:
-// RTL8710B → .image2.ram.text, RTL8720C → .sram.text. LN882H is the
-// exception: its stock linker has no matching glob, so patch_linker.py
-// injects KEEP(*(.sram.text*)) into .flash_copysection at pre-link.
-//
-// BK72xx (all variants) are left as a no-op: their SDK wraps flash
-// operations in GLOBAL_INT_DISABLE() which masks FIQ + IRQ at the CPU for
-// the duration of every write, so no ISR fires while flash is stalled and
-// the race IRAM_ATTR guards against cannot occur. The trade-off is that
-// interrupts are delayed (not dropped) by up to ~20 ms during a sector
-// erase, but that is an SDK-level choice and cannot be changed from this
-// layer.
-#if defined(USE_BK72XX)
-#define IRAM_ATTR
-#elif defined(USE_LIBRETINY_VARIANT_RTL8710B)
-// Stock linker consumes *(.image2.ram.text*) into .ram_image2.text (> BD_RAM).
-#define IRAM_ATTR __attribute__((noinline, section(".image2.ram.text")))
+#include "esphome/core/hal/hal_libretiny.h"
+#elif defined(USE_RP2040)
+#include "esphome/core/hal/hal_rp2040.h"
+#elif defined(USE_HOST)
+#include "esphome/core/hal/hal_host.h"
+#elif defined(USE_ZEPHYR)
+#include "esphome/core/hal/hal_zephyr.h"
 #else
-// RTL8720C: stock linker consumes *(.sram.text*) into .ram.code_text.
-// LN882H: patch_linker.py.script injects *(.sram.text*) into
-// .flash_copysection (> RAM0 AT> FLASH).
-#define IRAM_ATTR __attribute__((noinline, section(".sram.text")))
-#endif
-#define PROGMEM
-
-#else
-
-#define IRAM_ATTR
-#define PROGMEM
-
-#endif
-
-#ifdef USE_ESP32
-#include <freertos/FreeRTOS.h>
-#include <freertos/task.h>
-#endif
-
-#ifdef USE_BK72XX
-// Declared in the Beken FreeRTOS port (portmacro.h) and built in ARM mode so
-// it is callable from Thumb code via interworking. The MRS CPSR instruction
-// is ARM-only and user code here may be built in Thumb, so in_isr_context()
-// defers to this port helper on BK72xx instead of reading CPSR inline.
-extern "C" uint32_t platform_is_in_interrupt_context(void);
+#error "hal.h: not implemented for this platform"
 #endif
 
 namespace esphome {
 
-/// Returns true when executing inside an interrupt handler.
-/// always_inline so callers placed in IRAM keep the detection in IRAM.
-__attribute__((always_inline)) inline bool in_isr_context() {
-#if defined(USE_ESP32)
-  return xPortInIsrContext() != 0;
-#elif defined(USE_ESP8266)
-  // ESP8266 has no reliable single-register ISR detection: PS.INTLEVEL is
-  // non-zero both in a real ISR and when user code masks interrupts.  The
-  // ESP8266 wake path is context-agnostic (wake_loop_impl uses esp_schedule
-  // which is ISR-safe) so this helper is unused on this platform.
-  return false;
-#elif defined(USE_RP2040)
-  uint32_t ipsr;
-  __asm__ volatile("mrs %0, ipsr" : "=r"(ipsr));
-  return ipsr != 0;
-#elif defined(USE_BK72XX)
-  // BK72xx is ARM968E-S (ARM9); see extern declaration above.
-  return platform_is_in_interrupt_context() != 0;
-#elif defined(USE_LIBRETINY)
-  // Cortex-M (AmebaZ, AmebaZ2, LN882H). IPSR is the active exception number;
-  // non-zero means we're in a handler.
-  uint32_t ipsr;
-  __asm__ volatile("mrs %0, ipsr" : "=r"(ipsr));
-  return ipsr != 0;
-#else
-  // Host and any future platform without an ISR concept.
-  return false;
-#endif
-}
-
-void yield();
-uint32_t millis();
-uint64_t millis_64();
-uint32_t micros();
-void delay(uint32_t ms);
+// ESP8266 inlines delayMicroseconds() and arch_feed_wdt() in hal/hal_esp8266.h;
+// every other platform keeps them out-of-line in components/<platform>/core.cpp.
+#ifndef USE_ESP8266
 void delayMicroseconds(uint32_t us);  // NOLINT(readability-identifier-naming)
+void arch_feed_wdt();
+#endif
 void __attribute__((noreturn)) arch_restart();
 void arch_init();
-void arch_feed_wdt();
 uint32_t arch_get_cpu_cycle_count();
 uint32_t arch_get_cpu_freq_hz();
 
-#ifdef USE_ESP8266
-// ESP8266: pgm_read_* does real flash reads on Harvard architecture
-uint8_t progmem_read_byte(const uint8_t *addr);
-const char *progmem_read_ptr(const char *const *addr);
-uint16_t progmem_read_uint16(const uint16_t *addr);
-#else
-// All other platforms: PROGMEM is a no-op, so these are direct dereferences
+#ifndef USE_ESP8266
+// All non-ESP8266 platforms: PROGMEM is a no-op, so these are direct dereferences.
+// ESP8266's out-of-line declarations live in hal/hal_esp8266.h.
 inline uint8_t progmem_read_byte(const uint8_t *addr) { return *addr; }
 inline const char *progmem_read_ptr(const char *const *addr) { return *addr; }
 inline uint16_t progmem_read_uint16(const uint16_t *addr) { return *addr; }
diff --git a/esphome/core/hal/hal_esp32.h b/esphome/core/hal/hal_esp32.h
new file mode 100644
index 00000000000..e755337540d
--- /dev/null
+++ b/esphome/core/hal/hal_esp32.h
@@ -0,0 +1,35 @@
+#pragma once
+
+#ifdef USE_ESP32
+
+#include <cstdint>
+#include <esp_attr.h>
+#include <freertos/FreeRTOS.h>
+#include <freertos/task.h>
+
+#include "esphome/core/time_conversion.h"
+
+#ifndef PROGMEM
+#define PROGMEM
+#endif
+
+namespace esphome {
+
+/// Returns true when executing inside an interrupt handler.
+__attribute__((always_inline)) inline bool in_isr_context() { return xPortInIsrContext() != 0; }
+
+// Forward decl from <esp_timer.h>.
+// NOLINTNEXTLINE(readability-redundant-declaration)
+extern "C" int64_t esp_timer_get_time(void);
+
+__attribute__((always_inline)) inline void yield() { vPortYield(); }
+__attribute__((always_inline)) inline void delay(uint32_t ms) { vTaskDelay(ms / portTICK_PERIOD_MS); }
+__attribute__((always_inline)) inline uint32_t micros() { return static_cast<uint32_t>(esp_timer_get_time()); }
+uint32_t millis();
+__attribute__((always_inline)) inline uint64_t millis_64() {
+  return micros_to_millis<uint64_t>(static_cast<uint64_t>(esp_timer_get_time()));
+}
+
+}  // namespace esphome
+
+#endif  // USE_ESP32
diff --git a/esphome/core/hal/hal_esp8266.h b/esphome/core/hal/hal_esp8266.h
new file mode 100644
index 00000000000..04326a3579c
--- /dev/null
+++ b/esphome/core/hal/hal_esp8266.h
@@ -0,0 +1,65 @@
+#pragma once
+
+#ifdef USE_ESP8266
+
+#include <c_types.h>
+#include <cstdint>
+#include <pgmspace.h>
+
+#include "esphome/core/time_64.h"
+
+#ifndef PROGMEM
+#define PROGMEM ICACHE_RODATA_ATTR
+#endif
+
+// Forward decls from Arduino's <Arduino.h> for the inline wrappers below.
+// NOLINTBEGIN(google-runtime-int,readability-identifier-naming,readability-redundant-declaration)
+extern "C" void yield(void);
+extern "C" void delay(unsigned long ms);
+extern "C" unsigned long micros(void);
+extern "C" unsigned long millis(void);
+// NOLINTEND(google-runtime-int,readability-identifier-naming,readability-redundant-declaration)
+
+// Forward decl from <user_interface.h> for arch_feed_wdt() inline below.
+// NOLINTNEXTLINE(readability-redundant-declaration)
+extern "C" void system_soft_wdt_feed(void);
+
+namespace esphome {
+
+// Forward decl from helpers.h so this header stays cheap.
+// NOLINTNEXTLINE(readability-redundant-declaration)
+void delay_microseconds_safe(uint32_t us);
+
+/// Returns true when executing inside an interrupt handler.
+/// ESP8266 has no reliable single-register ISR detection: PS.INTLEVEL is
+/// non-zero both in a real ISR and when user code masks interrupts.  The
+/// ESP8266 wake path is context-agnostic (wake_loop_impl uses esp_schedule
+/// which is ISR-safe) so this helper is unused on this platform.
+__attribute__((always_inline)) inline bool in_isr_context() { return false; }
+
+__attribute__((always_inline)) inline void yield() { ::yield(); }
+__attribute__((always_inline)) inline uint32_t micros() { return static_cast<uint32_t>(::micros()); }
+void delay(uint32_t ms);
+uint32_t millis();
+__attribute__((always_inline)) inline uint64_t millis_64() { return Millis64Impl::compute(millis()); }
+
+// ESP8266: pgm_read_* does aligned 32-bit flash reads on Harvard architecture.
+// Inline-forward to the platform macros so the wrappers themselves don't
+// occupy IRAM/flash on every call site.
+__attribute__((always_inline)) inline uint8_t progmem_read_byte(const uint8_t *addr) {
+  return pgm_read_byte(addr);  // NOLINT
+}
+__attribute__((always_inline)) inline const char *progmem_read_ptr(const char *const *addr) {
+  return reinterpret_cast<const char *>(pgm_read_ptr(addr));  // NOLINT
+}
+__attribute__((always_inline)) inline uint16_t progmem_read_uint16(const uint16_t *addr) {
+  return pgm_read_word(addr);  // NOLINT
+}
+
+// NOLINTNEXTLINE(readability-identifier-naming)
+__attribute__((always_inline)) inline void delayMicroseconds(uint32_t us) { delay_microseconds_safe(us); }
+__attribute__((always_inline)) inline void arch_feed_wdt() { system_soft_wdt_feed(); }
+
+}  // namespace esphome
+
+#endif  // USE_ESP8266
diff --git a/esphome/core/hal/hal_host.h b/esphome/core/hal/hal_host.h
new file mode 100644
index 00000000000..145fe4ea9c1
--- /dev/null
+++ b/esphome/core/hal/hal_host.h
@@ -0,0 +1,24 @@
+#pragma once
+
+#ifdef USE_HOST
+
+#include <cstdint>
+
+#define IRAM_ATTR
+#define PROGMEM
+
+namespace esphome {
+
+/// Returns true when executing inside an interrupt handler.
+/// Host has no ISR concept.
+__attribute__((always_inline)) inline bool in_isr_context() { return false; }
+
+void yield();
+void delay(uint32_t ms);
+uint32_t micros();
+uint32_t millis();
+uint64_t millis_64();
+
+}  // namespace esphome
+
+#endif  // USE_HOST
diff --git a/esphome/core/hal/hal_libretiny.h b/esphome/core/hal/hal_libretiny.h
new file mode 100644
index 00000000000..e0d92735bbb
--- /dev/null
+++ b/esphome/core/hal/hal_libretiny.h
@@ -0,0 +1,93 @@
+#pragma once
+
+#ifdef USE_LIBRETINY
+
+#include <cstdint>
+
+// For the inline millis() fast paths (xTaskGetTickCount, portTICK_PERIOD_MS).
+#include <FreeRTOS.h>
+#include <task.h>
+
+#include "esphome/core/time_64.h"
+
+// IRAM_ATTR places a function in executable RAM so it is callable from an
+// ISR even while flash is busy (XIP stall, OTA, logger flash write).
+// Each family uses a section its stock linker already routes to RAM:
+// RTL8710B → .image2.ram.text, RTL8720C → .sram.text. LN882H is the
+// exception: its stock linker has no matching glob, so patch_linker.py
+// injects KEEP(*(.sram.text*)) into .flash_copysection at pre-link.
+//
+// BK72xx (all variants) are left as a no-op: their SDK wraps flash
+// operations in GLOBAL_INT_DISABLE() which masks FIQ + IRQ at the CPU for
+// the duration of every write, so no ISR fires while flash is stalled and
+// the race IRAM_ATTR guards against cannot occur. The trade-off is that
+// interrupts are delayed (not dropped) by up to ~20 ms during a sector
+// erase, but that is an SDK-level choice and cannot be changed from this
+// layer.
+#if defined(USE_BK72XX)
+#define IRAM_ATTR
+#elif defined(USE_LIBRETINY_VARIANT_RTL8710B)
+// Stock linker consumes *(.image2.ram.text*) into .ram_image2.text (> BD_RAM).
+#define IRAM_ATTR __attribute__((noinline, section(".image2.ram.text")))
+#else
+// RTL8720C: stock linker consumes *(.sram.text*) into .ram.code_text.
+// LN882H: patch_linker.py.script injects *(.sram.text*) into
+// .flash_copysection (> RAM0 AT> FLASH).
+#define IRAM_ATTR __attribute__((noinline, section(".sram.text")))
+#endif
+#define PROGMEM
+
+#ifdef USE_BK72XX
+// Declared in the Beken FreeRTOS port (portmacro.h) and built in ARM mode so
+// it is callable from Thumb code via interworking. The MRS CPSR instruction
+// is ARM-only and user code here may be built in Thumb, so in_isr_context()
+// defers to this port helper on BK72xx instead of reading CPSR inline.
+extern "C" uint32_t platform_is_in_interrupt_context(void);
+#endif
+
+// Forward decls from Arduino's <Arduino.h> for the inline wrappers below.
+// NOLINTBEGIN(google-runtime-int,readability-identifier-naming,readability-redundant-declaration)
+extern "C" void yield(void);
+extern "C" void delay(unsigned long ms);
+extern "C" unsigned long micros(void);
+extern "C" unsigned long millis(void);
+// NOLINTEND(google-runtime-int,readability-identifier-naming,readability-redundant-declaration)
+
+namespace esphome {
+
+/// Returns true when executing inside an interrupt handler.
+__attribute__((always_inline)) inline bool in_isr_context() {
+#if defined(USE_BK72XX)
+  // BK72xx is ARM968E-S (ARM9); see extern declaration above.
+  return platform_is_in_interrupt_context() != 0;
+#else
+  // Cortex-M (AmebaZ, AmebaZ2, LN882H). IPSR is the active exception number;
+  // non-zero means we're in a handler.
+  uint32_t ipsr;
+  __asm__ volatile("mrs %0, ipsr" : "=r"(ipsr));
+  return ipsr != 0;
+#endif
+}
+
+__attribute__((always_inline)) inline void yield() { ::yield(); }
+__attribute__((always_inline)) inline void delay(uint32_t ms) { ::delay(ms); }
+__attribute__((always_inline)) inline uint32_t micros() { return static_cast<uint32_t>(::micros()); }
+
+// Per-variant millis() fast path — matches MillisInternal::get().
+#if defined(USE_RTL87XX) || defined(USE_LN882X)
+static_assert(configTICK_RATE_HZ == 1000, "millis() fast path requires 1 kHz FreeRTOS tick");
+__attribute__((always_inline)) inline uint32_t millis() {
+  // xTaskGetTickCountFromISR is mandatory in interrupt context per the FreeRTOS API contract.
+  return in_isr_context() ? xTaskGetTickCountFromISR() : xTaskGetTickCount();
+}
+#elif defined(USE_BK72XX)
+static_assert(configTICK_RATE_HZ == 500, "BK72xx millis() fast path assumes 500 Hz FreeRTOS tick");
+__attribute__((always_inline)) inline uint32_t millis() { return xTaskGetTickCount() * portTICK_PERIOD_MS; }
+#else
+__attribute__((always_inline)) inline uint32_t millis() { return static_cast<uint32_t>(::millis()); }
+#endif
+__attribute__((always_inline)) inline uint64_t millis_64() { return Millis64Impl::compute(millis()); }
+
+}  // namespace esphome
+
+#endif  // USE_LIBRETINY
diff --git a/esphome/core/hal/hal_rp2040.h b/esphome/core/hal/hal_rp2040.h
new file mode 100644
index 00000000000..156ff33b863
--- /dev/null
+++ b/esphome/core/hal/hal_rp2040.h
@@ -0,0 +1,40 @@
+#pragma once
+
+#ifdef USE_RP2040
+
+#include <cstdint>
+
+#include "esphome/core/time_conversion.h"
+
+#define IRAM_ATTR __attribute__((noinline, long_call, section(".time_critical")))
+#define PROGMEM
+
+// Forward decls from Arduino's <Arduino.h> for the inline wrappers below.
+// NOLINTBEGIN(google-runtime-int,readability-identifier-naming,readability-redundant-declaration)
+extern "C" void yield(void);
+extern "C" void delay(unsigned long ms);
+extern "C" unsigned long micros(void);
+extern "C" unsigned long millis(void);
+// NOLINTEND(google-runtime-int,readability-identifier-naming,readability-redundant-declaration)
+
+// Forward decl from <pico/time.h>.
+extern "C" uint64_t time_us_64(void);
+
+namespace esphome {
+
+/// Returns true when executing inside an interrupt handler.
+__attribute__((always_inline)) inline bool in_isr_context() {
+  uint32_t ipsr;
+  __asm__ volatile("mrs %0, ipsr" : "=r"(ipsr));
+  return ipsr != 0;
+}
+
+__attribute__((always_inline)) inline void yield() { ::yield(); }
+__attribute__((always_inline)) inline void delay(uint32_t ms) { ::delay(ms); }
+__attribute__((always_inline)) inline uint32_t micros() { return static_cast<uint32_t>(::micros()); }
+__attribute__((always_inline)) inline uint32_t millis() { return micros_to_millis(::time_us_64()); }
+__attribute__((always_inline)) inline uint64_t millis_64() { return micros_to_millis<uint64_t>(::time_us_64()); }
+
+}  // namespace esphome
+
+#endif  // USE_RP2040
diff --git a/esphome/core/hal/hal_zephyr.h b/esphome/core/hal/hal_zephyr.h
new file mode 100644
index 00000000000..e28be5c775d
--- /dev/null
+++ b/esphome/core/hal/hal_zephyr.h
@@ -0,0 +1,24 @@
+#pragma once
+
+#ifdef USE_ZEPHYR
+
+#include <cstdint>
+
+#define IRAM_ATTR
+#define PROGMEM
+
+namespace esphome {
+
+/// Returns true when executing inside an interrupt handler.
+/// Zephyr/nRF52: not currently consulted — wake path is platform-specific.
+__attribute__((always_inline)) inline bool in_isr_context() { return false; }
+
+void yield();
+void delay(uint32_t ms);
+uint32_t micros();
+uint32_t millis();
+uint64_t millis_64();
+
+}  // namespace esphome
+
+#endif  // USE_ZEPHYR
diff --git a/esphome/core/helpers.h b/esphome/core/helpers.h
index b2b07c57a05..355db6c7f4f 100644
--- a/esphome/core/helpers.h
+++ b/esphome/core/helpers.h
@@ -20,6 +20,7 @@
 #include <strings.h>
 
 #include "esphome/core/optional.h"
+#include "esphome/core/time_conversion.h"
 
 // Backward compatibility re-export of heap-allocating helpers.
 // These functions have moved to alloc_helpers.h. External components should
@@ -833,43 +834,9 @@ template<std::integral T> constexpr uint32_t fnv1a_hash_extend(uint32_t hash, T
 constexpr uint32_t fnv1a_hash(const char *str) { return fnv1a_hash_extend(FNV1_OFFSET_BASIS, str); }
 inline uint32_t fnv1a_hash(const std::string &str) { return fnv1a_hash(str.c_str()); }
 
-/// Convert a 64-bit microsecond count to milliseconds without calling
-/// __udivdi3 (software 64-bit divide, ~1200 ns on Xtensa @ 240 MHz).
-///
-/// Returns uint32_t by default (for millis()), or uint64_t when requested
-/// (for millis_64()). The only difference is whether hi * Q is truncated
-/// to 32 bits or widened to 64.
-///
-/// On 32-bit targets, GCC does not optimize 64-bit constant division into a
-/// multiply-by-reciprocal. Since 1000 = 8 * 125, we first right-shift by 3
-/// (free divide-by-8), then use the Euclidean division identity to decompose
-/// the remaining 64-bit divide-by-125 into a single 32-bit division:
-///
-///   floor(us / 1000) = floor(floor(us / 8) / 125)    [exact for integers]
-///   2^32 = Q * 125 + R  (34359738 * 125 + 46)
-///   (hi * 2^32 + lo) / 125 = hi * Q + (hi * R + lo) / 125
-///
-/// GCC optimizes the remaining 32-bit "/ 125U" into a multiply-by-reciprocal
-/// (mulhu + shift), so no division instruction is emitted.
-///
-/// Safe for us up to ~3.2e18 (~101,700 years of microseconds).
-///
-/// See: https://en.wikipedia.org/wiki/Euclidean_division
-/// See: https://ridiculousfish.com/blog/posts/labor-of-division-episode-iii.html
-template<typename ReturnT = uint32_t> inline constexpr ESPHOME_ALWAYS_INLINE ReturnT micros_to_millis(uint64_t us) {
-  constexpr uint32_t d = 125U;
-  constexpr uint32_t q = static_cast<uint32_t>((1ULL << 32) / d);  // 34359738
-  constexpr uint32_t r = static_cast<uint32_t>((1ULL << 32) % d);  // 46
-  // 1000 = 8 * 125; divide-by-8 is a free shift
-  uint64_t x = us >> 3;
-  uint32_t lo = static_cast<uint32_t>(x);
-  uint32_t hi = static_cast<uint32_t>(x >> 32);
-  // Combine remainder term: hi * (2^32 % 125) + lo
-  uint32_t adj = hi * r + lo;
-  // If adj overflowed, the true value is 2^32 + adj; apply the identity again
-  // static_cast<ReturnT>(hi) widens to 64-bit when ReturnT=uint64_t, preserving upper bits of hi*q
-  return static_cast<ReturnT>(hi) * q + (adj < lo ? (adj + r) / d + q : adj / d);
-}
+// micros_to_millis<>() lives in its own lightweight header so hal.h can pull it
+// in for inline millis_64() without forcing every TU that includes hal.h to
+// also include the rest of helpers.h.
 
 /// Return a random 32-bit unsigned integer.
 /// Not thread-safe. Must only be called from the main loop.
diff --git a/esphome/core/scheduler.cpp b/esphome/core/scheduler.cpp
index 11884ce4ba9..57deeab0da3 100644
--- a/esphome/core/scheduler.cpp
+++ b/esphome/core/scheduler.cpp
@@ -35,7 +35,9 @@ static constexpr uint32_t MAX_INTERVAL_DELAY = 5000;
 // Uses a stack buffer to avoid heap allocation
 // Uses ESPHOME_snprintf_P/ESPHOME_PSTR for ESP8266 to keep format strings in flash
 struct SchedulerNameLog {
-  char buffer[20];  // Enough for "id:4294967295" or "hash:0xFFFFFFFF" or "(null)"
+  // Sized for the widest formatted output: "self:0x" + 16 hex digits (64-bit pointer) + nul.
+  // Also covers "id:4294967295", "hash:0xFFFFFFFF", "iid:4294967295", "(null)".
+  char buffer[28];
 
   // Format a scheduler item name for logging
   // Returns pointer to formatted string (either static_name or internal buffer)
@@ -53,9 +55,15 @@ struct SchedulerNameLog {
     } else if (name_type == NameType::NUMERIC_ID) {
       ESPHOME_snprintf_P(buffer, sizeof(buffer), ESPHOME_PSTR("id:%" PRIu32), hash_or_id);
       return buffer;
-    } else {  // NUMERIC_ID_INTERNAL
+    } else if (name_type == NameType::NUMERIC_ID_INTERNAL) {
       ESPHOME_snprintf_P(buffer, sizeof(buffer), ESPHOME_PSTR("iid:%" PRIu32), hash_or_id);
       return buffer;
+    } else {  // SELF_POINTER
+      // static_name carries the void* key for SELF_POINTER (pointer-width union slot).
+      // %p is specified as void* (not const void*), so strip const for the varargs call.
+      ESPHOME_snprintf_P(buffer, sizeof(buffer), ESPHOME_PSTR("self:%p"),
+                         const_cast<void *>(static_cast<const void *>(static_name)));
+      return buffer;
     }
   }
 };
@@ -293,6 +301,27 @@ bool HOT Scheduler::cancel_interval(Component *component, uint32_t id) {
   return this->cancel_item_(component, NameType::NUMERIC_ID, nullptr, id, SchedulerItem::INTERVAL);
 }
 
+// Self-keyed scheduler API. The cancellation key is `self` (typically the caller's `this`),
+// passed through the existing static_name pointer slot. Matching is by raw pointer equality
+// (see matches_item_locked_'s SELF_POINTER branch). No Component pointer is stored, so
+// is_failed() skip and component-based log attribution don't apply.
+void HOT Scheduler::set_timeout(const void *self, uint32_t timeout, std::function<void()> &&func) {
+  this->set_timer_common_(nullptr, SchedulerItem::TIMEOUT, NameType::SELF_POINTER, static_cast<const char *>(self), 0,
+                          timeout, std::move(func));
+}
+void HOT Scheduler::set_interval(const void *self, uint32_t interval, std::function<void()> &&func) {
+  this->set_timer_common_(nullptr, SchedulerItem::INTERVAL, NameType::SELF_POINTER, static_cast<const char *>(self), 0,
+                          interval, std::move(func));
+}
+bool HOT Scheduler::cancel_timeout(const void *self) {
+  return this->cancel_item_(nullptr, NameType::SELF_POINTER, static_cast<const char *>(self), 0,
+                            SchedulerItem::TIMEOUT);
+}
+bool HOT Scheduler::cancel_interval(const void *self) {
+  return this->cancel_item_(nullptr, NameType::SELF_POINTER, static_cast<const char *>(self), 0,
+                            SchedulerItem::INTERVAL);
+}
+
 // Suppress deprecation warnings for RetryResult usage in the still-present (but deprecated) retry implementation.
 // Remove before 2026.8.0 along with all retry code.
 #pragma GCC diagnostic push
diff --git a/esphome/core/scheduler.h b/esphome/core/scheduler.h
index 46b19855c35..7a6be6bea9a 100644
--- a/esphome/core/scheduler.h
+++ b/esphome/core/scheduler.h
@@ -146,22 +146,43 @@ class Scheduler {
   }
 
   // Name storage type discriminator for SchedulerItem
-  // Used to distinguish between static strings, hashed strings, numeric IDs, and internal numeric IDs
+  // Used to distinguish between static strings, hashed strings, numeric IDs, internal numeric IDs,
+  // and self-keyed pointers (caller-supplied `void *`, typically `this`).
   enum class NameType : uint8_t {
-    STATIC_STRING = 0,       // const char* pointer to static/flash storage
-    HASHED_STRING = 1,       // uint32_t FNV-1a hash of a runtime string
-    NUMERIC_ID = 2,          // uint32_t numeric identifier (component-level)
-    NUMERIC_ID_INTERNAL = 3  // uint32_t numeric identifier (core/internal, separate namespace)
+    STATIC_STRING = 0,        // const char* pointer to static/flash storage
+    HASHED_STRING = 1,        // uint32_t FNV-1a hash of a runtime string
+    NUMERIC_ID = 2,           // uint32_t numeric identifier (component-level)
+    NUMERIC_ID_INTERNAL = 3,  // uint32_t numeric identifier (core/internal, separate namespace)
+    SELF_POINTER = 4          // void* caller-supplied key (typically `this`); pointer equality
   };
 
+  /** Self-keyed timeout. The cancellation key is `self` (typically the caller's `this`).
+   *
+   * Use this when the caller schedules at most one timer of a single purpose at a time and
+   * does not need a `Component` for `is_failed()` skip or log source attribution. Lets
+   * small classes drop `Component` inheritance entirely when their only Component dependency
+   * was the per-instance scheduler key.
+   *
+   * NOT applied for self-keyed items:
+   *  - `is_failed()` skip — callbacks always fire (no Component to consult).
+   *  - Log source attribution — logs use a generic "self:0x…" label.
+   *
+   * If you need either of those, use the existing `(Component *, id)` overloads.
+   */
+  void set_timeout(const void *self, uint32_t timeout, std::function<void()> &&func);
+  /// Self-keyed interval. See set_timeout(const void *, ...) for semantics.
+  void set_interval(const void *self, uint32_t interval, std::function<void()> &&func);
+  bool cancel_timeout(const void *self);
+  bool cancel_interval(const void *self);
+
  protected:
   struct SchedulerItem {
     // Ordered by size to minimize padding
     Component *component;
     // Optimized name storage using tagged union - zero heap allocation
     union {
-      const char *static_name;  // For STATIC_STRING (string literals, no allocation)
-      uint32_t hash_or_id;      // For HASHED_STRING or NUMERIC_ID
+      const char *static_name;  // For STATIC_STRING (string literals) and SELF_POINTER (caller's `this`)
+      uint32_t hash_or_id;      // For HASHED_STRING, NUMERIC_ID, and NUMERIC_ID_INTERNAL
     } name_;
     uint32_t interval;
     // Split time to handle millis() rollover. The scheduler combines the 32-bit millis()
@@ -182,19 +203,19 @@ class Scheduler {
     // std::atomic<uint8_t> inlines correctly on all platforms.
     std::atomic<uint8_t> remove{0};
 
-    // Bit-packed fields (4 bits used, 4 bits padding in 1 byte)
-    enum Type : uint8_t { TIMEOUT, INTERVAL } type : 1;
-    NameType name_type_ : 2;  // Discriminator for name_ union (0–3, see NameType enum)
-    bool is_retry : 1;        // True if this is a retry timeout
-                              // 4 bits padding
-#else
-    // Single-threaded or multi-threaded without atomics: can pack all fields together
     // Bit-packed fields (5 bits used, 3 bits padding in 1 byte)
     enum Type : uint8_t { TIMEOUT, INTERVAL } type : 1;
-    bool remove : 1;
-    NameType name_type_ : 2;  // Discriminator for name_ union (0–3, see NameType enum)
+    NameType name_type_ : 3;  // Discriminator for name_ union (0–4, see NameType enum)
     bool is_retry : 1;        // True if this is a retry timeout
                               // 3 bits padding
+#else
+    // Single-threaded or multi-threaded without atomics: can pack all fields together
+    // Bit-packed fields (6 bits used, 2 bits padding in 1 byte)
+    enum Type : uint8_t { TIMEOUT, INTERVAL } type : 1;
+    bool remove : 1;
+    NameType name_type_ : 3;  // Discriminator for name_ union (0–4, see NameType enum)
+    bool is_retry : 1;        // True if this is a retry timeout
+                              // 2 bits padding
 #endif
 
     // Constructor
@@ -228,19 +249,26 @@ class Scheduler {
     SchedulerItem(SchedulerItem &&) = delete;
     SchedulerItem &operator=(SchedulerItem &&) = delete;
 
-    // Helper to get the static name (only valid for STATIC_STRING type)
-    const char *get_name() const { return (name_type_ == NameType::STATIC_STRING) ? name_.static_name : nullptr; }
+    // Helper to get the pointer-slot value (valid for STATIC_STRING and SELF_POINTER types).
+    // Both share the same union member, so callers (e.g. log formatters) can read either uniformly.
+    const char *get_name() const {
+      return (name_type_ == NameType::STATIC_STRING || name_type_ == NameType::SELF_POINTER) ? name_.static_name
+                                                                                             : nullptr;
+    }
 
-    // Helper to get the hash or numeric ID (only valid for HASHED_STRING or NUMERIC_ID types)
-    uint32_t get_name_hash_or_id() const { return (name_type_ != NameType::STATIC_STRING) ? name_.hash_or_id : 0; }
+    // Helper to get the hash or numeric ID (only valid for HASHED_STRING / NUMERIC_ID / NUMERIC_ID_INTERNAL types)
+    uint32_t get_name_hash_or_id() const {
+      return (name_type_ != NameType::STATIC_STRING && name_type_ != NameType::SELF_POINTER) ? name_.hash_or_id : 0;
+    }
 
     // Helper to get the name type
     NameType get_name_type() const { return name_type_; }
 
-    // Set name storage: for STATIC_STRING stores the pointer, for all other types stores hash_or_id.
-    // Both union members occupy the same offset, so only one store is needed.
+    // Set name storage. STATIC_STRING/SELF_POINTER use the static_name pointer slot
+    // (both are pointer-width); other types use hash_or_id. Both union members occupy
+    // the same offset, so only one store is needed.
     void set_name(NameType type, const char *static_name, uint32_t hash_or_id) {
-      if (type == NameType::STATIC_STRING) {
+      if (type == NameType::STATIC_STRING || type == NameType::SELF_POINTER) {
         name_.static_name = static_name;
       } else {
         name_.hash_or_id = hash_or_id;
@@ -367,10 +395,14 @@ class Scheduler {
     // Name type must match
     if (item->get_name_type() != name_type)
       return false;
-    // For static strings, compare the string content; for hash/ID, compare the value
+    // STATIC_STRING: compare string content. SELF_POINTER: raw pointer equality (no strcmp).
+    // Other types: compare hash/ID value.
     if (name_type == NameType::STATIC_STRING) {
       return this->names_match_static_(item->get_name(), static_name);
     }
+    if (name_type == NameType::SELF_POINTER) {
+      return item->name_.static_name == static_name;
+    }
     return item->get_name_hash_or_id() == hash_or_id;
   }
 
diff --git a/esphome/core/time_64.h b/esphome/core/time_64.h
index d82373dbfe9..f66f9afddb8 100644
--- a/esphome/core/time_64.h
+++ b/esphome/core/time_64.h
@@ -6,8 +6,6 @@
 #include <cstdint>
 #include <limits>
 
-#include "esphome/core/helpers.h"
-
 namespace esphome {
 
 class Scheduler;
@@ -24,7 +22,9 @@ class Millis64Impl {
   static uint32_t last_millis;
   static uint16_t millis_major;
 
-  static inline uint64_t ESPHOME_ALWAYS_INLINE compute(uint32_t now) {
+  // Raw __attribute__((always_inline)) (not ESPHOME_ALWAYS_INLINE) so this
+  // header does not need to pull helpers.h.
+  static inline uint64_t __attribute__((always_inline)) compute(uint32_t now) {
     // 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;
 
diff --git a/esphome/core/time_conversion.h b/esphome/core/time_conversion.h
new file mode 100644
index 00000000000..e9060c06267
--- /dev/null
+++ b/esphome/core/time_conversion.h
@@ -0,0 +1,46 @@
+#pragma once
+
+#include <cstdint>
+
+namespace esphome {
+
+/// Convert a 64-bit microsecond count to milliseconds without calling
+/// __udivdi3 (software 64-bit divide, ~1200 ns on Xtensa @ 240 MHz).
+///
+/// Returns uint32_t by default (for millis()), or uint64_t when requested
+/// (for millis_64()). The only difference is whether hi * Q is truncated
+/// to 32 bits or widened to 64.
+///
+/// On 32-bit targets, GCC does not optimize 64-bit constant division into a
+/// multiply-by-reciprocal. Since 1000 = 8 * 125, we first right-shift by 3
+/// (free divide-by-8), then use the Euclidean division identity to decompose
+/// the remaining 64-bit divide-by-125 into a single 32-bit division:
+///
+///   floor(us / 1000) = floor(floor(us / 8) / 125)    [exact for integers]
+///   2^32 = Q * 125 + R  (34359738 * 125 + 46)
+///   (hi * 2^32 + lo) / 125 = hi * Q + (hi * R + lo) / 125
+///
+/// GCC optimizes the remaining 32-bit "/ 125U" into a multiply-by-reciprocal
+/// (mulhu + shift), so no division instruction is emitted.
+///
+/// Safe for us up to ~3.2e18 (~101,700 years of microseconds).
+///
+/// See: https://en.wikipedia.org/wiki/Euclidean_division
+/// See: https://ridiculousfish.com/blog/posts/labor-of-division-episode-iii.html
+template<typename ReturnT = uint32_t>
+__attribute__((always_inline)) inline constexpr ReturnT micros_to_millis(uint64_t us) {
+  constexpr uint32_t d = 125U;
+  constexpr uint32_t q = static_cast<uint32_t>((1ULL << 32) / d);  // 34359738
+  constexpr uint32_t r = static_cast<uint32_t>((1ULL << 32) % d);  // 46
+  // 1000 = 8 * 125; divide-by-8 is a free shift
+  uint64_t x = us >> 3;
+  uint32_t lo = static_cast<uint32_t>(x);
+  uint32_t hi = static_cast<uint32_t>(x >> 32);
+  // Combine remainder term: hi * (2^32 % 125) + lo
+  uint32_t adj = hi * r + lo;
+  // If adj overflowed, the true value is 2^32 + adj; apply the identity again
+  // static_cast<ReturnT>(hi) widens to 64-bit when ReturnT=uint64_t, preserving upper bits of hi*q
+  return static_cast<ReturnT>(hi) * q + (adj < lo ? (adj + r) / d + q : adj / d);
+}
+
+}  // namespace esphome
diff --git a/esphome/external_files.py b/esphome/external_files.py
index fbc261f8e0e..dfabc54f474 100644
--- a/esphome/external_files.py
+++ b/esphome/external_files.py
@@ -175,6 +175,11 @@ def download_content(url: str, path: Path, timeout: int = NETWORK_TIMEOUT) -> by
             headers={"User-agent": f"ESPHome/{__version__} (https://esphome.io)"},
         )
         req.raise_for_status()
+        # `.content` reads the body lazily; chunked-decode, gzip-decode,
+        # and mid-stream connection errors all surface here as
+        # RequestException subclasses, so this needs the same fall-back
+        # treatment as the request itself.
+        data = req.content
     except requests.exceptions.RequestException as e:
         if path.exists():
             _LOGGER.warning(
@@ -185,7 +190,6 @@ def download_content(url: str, path: Path, timeout: int = NETWORK_TIMEOUT) -> by
             return path.read_bytes()
         raise cv.Invalid(f"Could not download from {url}: {e}") from e
 
-    data = req.content
     write_file(path, data)
     _write_etag(path, req.headers.get(ETAG))
     return data
diff --git a/esphome/loader.py b/esphome/loader.py
index 2405fa6f884..d50554f8c93 100644
--- a/esphome/loader.py
+++ b/esphome/loader.py
@@ -9,14 +9,23 @@ import logging
 from pathlib import Path
 import sys
 from types import ModuleType
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from esphome.const import SOURCE_FILE_EXTENSIONS
 from esphome.core import CORE
-import esphome.core.config
-from esphome.cpp_generator import MockObjClass
 from esphome.types import ConfigType
 
+if TYPE_CHECKING:
+    from esphome.cpp_generator import MockObjClass
+
+# `esphome.core.config` is imported lazily in `_lookup_module` when the
+# "esphome" pseudo-component is first resolved. It pulls in
+# `esphome.automation` and `esphome.config_validation`, which together
+# dominate `esphome.__main__` startup cost when loaded eagerly.
+# `esphome.cpp_generator` is similarly avoided at module scope; it pulls
+# in `esphome.yaml_util` and is only needed for the `MockObjClass` type
+# annotation, which is resolved lazily via `TYPE_CHECKING`.
+
 _LOGGER = logging.getLogger(__name__)
 
 
@@ -94,7 +103,7 @@ class ComponentManifest:
         return getattr(self.module, "CODEOWNERS", [])
 
     @property
-    def instance_type(self) -> MockObjClass | None:
+    def instance_type(self) -> "MockObjClass | None":
         return getattr(self.module, "INSTANCE_TYPE", None)
 
     @property
@@ -213,6 +222,13 @@ def _lookup_module(domain: str, exception: bool) -> ComponentManifest | None:
     if domain in _COMPONENT_CACHE:
         return _COMPONENT_CACHE[domain]
 
+    if domain == "esphome":
+        import esphome.core.config
+
+        manif = ComponentManifest(esphome.core.config, recursive_sources=True)
+        _COMPONENT_CACHE[domain] = manif
+        return manif
+
     try:
         module = importlib.import_module(f"esphome.components.{domain}")
     except ImportError as e:
@@ -248,9 +264,6 @@ def get_platform(domain: str, platform: str) -> ComponentManifest | None:
 
 _COMPONENT_CACHE: dict[str, ComponentManifest] = {}
 CORE_COMPONENTS_PATH = (Path(__file__).parent / "components").resolve()
-_COMPONENT_CACHE["esphome"] = ComponentManifest(
-    esphome.core.config, recursive_sources=True
-)
 
 
 def _replace_component_manifest(domain: str, manifest: ComponentManifest) -> None:
diff --git a/script/import_time_budget.json b/script/import_time_budget.json
index 1e656dc9776..af3aa835113 100644
--- a/script/import_time_budget.json
+++ b/script/import_time_budget.json
@@ -1,5 +1,5 @@
 {
   "target_module": "esphome.__main__",
   "margin_pct": 15,
-  "cumulative_us": 123000
+  "cumulative_us": 91000
 }
diff --git a/tests/components/fan/common.yaml b/tests/components/fan/common.yaml
index 099bbfef08e..6cabbd24f8d 100644
--- a/tests/components/fan/common.yaml
+++ b/tests/components/fan/common.yaml
@@ -57,3 +57,34 @@ binary_sensor:
         return true;
       }
       return false;
+
+# Exercise fan.turn_on with various field combinations so the
+# TurnOnAction codegen paths get build coverage.
+button:
+  - platform: template
+    name: "Fan Speed Only"
+    on_press:
+      - fan.turn_on:
+          id: test_fan
+          speed: 2
+  - platform: template
+    name: "Fan Oscillating + Direction"
+    on_press:
+      - fan.turn_on:
+          id: test_fan
+          oscillating: true
+          direction: REVERSE
+  - platform: template
+    name: "Fan All Fields"
+    on_press:
+      - fan.turn_on:
+          id: test_fan
+          oscillating: false
+          speed: 3
+          direction: FORWARD
+  - platform: template
+    name: "Fan Lambda Speed"
+    on_press:
+      - fan.turn_on:
+          id: test_fan
+          speed: !lambda 'return 1;'
diff --git a/tests/components/template/common-base.yaml b/tests/components/template/common-base.yaml
index daa6f53d42f..819eaa8bbfe 100644
--- a/tests/components/template/common-base.yaml
+++ b/tests/components/template/common-base.yaml
@@ -442,6 +442,20 @@ valve:
           state: CLOSED
     stop_action:
       - logger.log: stop_action
+      # Exercise valve.control with various field combinations so the
+      # ControlAction codegen paths get build coverage.
+      - valve.control:
+          id: template_valve
+          stop: true
+      - valve.control:
+          id: template_valve
+          position: 50%
+      - valve.control:
+          id: template_valve
+          state: OPEN
+      - valve.control:
+          id: template_valve
+          position: !lambda 'return 0.25f;'
     optimistic: true
 
 text:
diff --git a/tests/integration/fixtures/fan_turn_on_action.yaml b/tests/integration/fixtures/fan_turn_on_action.yaml
new file mode 100644
index 00000000000..11bf033e48c
--- /dev/null
+++ b/tests/integration/fixtures/fan_turn_on_action.yaml
@@ -0,0 +1,59 @@
+esphome:
+  name: fan-turn-on-action-test
+host:
+api:
+logger:
+  level: DEBUG
+
+globals:
+  - id: test_speed
+    type: int
+    initial_value: "2"
+
+fan:
+  - platform: template
+    id: test_fan
+    name: "Test Fan"
+    has_oscillating: true
+    has_direction: true
+    speed_count: 5
+
+button:
+  # fan.turn_on: speed only
+  - platform: template
+    id: btn_speed
+    name: "Set Speed"
+    on_press:
+      - fan.turn_on:
+          id: test_fan
+          speed: 3
+
+  # fan.turn_on: oscillating + direction (no speed)
+  - platform: template
+    id: btn_oscillate_direction
+    name: "Set Oscillate Direction"
+    on_press:
+      - fan.turn_on:
+          id: test_fan
+          oscillating: true
+          direction: REVERSE
+
+  # fan.turn_on: all three fields
+  - platform: template
+    id: btn_all_fields
+    name: "Set All Fields"
+    on_press:
+      - fan.turn_on:
+          id: test_fan
+          oscillating: false
+          speed: 4
+          direction: FORWARD
+
+  # fan.turn_on: lambda for speed (exercises lambda path)
+  - platform: template
+    id: btn_lambda_speed
+    name: "Lambda Speed"
+    on_press:
+      - fan.turn_on:
+          id: test_fan
+          speed: !lambda "return id(test_speed);"
diff --git a/tests/integration/fixtures/scheduler_self_keyed.yaml b/tests/integration/fixtures/scheduler_self_keyed.yaml
new file mode 100644
index 00000000000..9a691136f36
--- /dev/null
+++ b/tests/integration/fixtures/scheduler_self_keyed.yaml
@@ -0,0 +1,112 @@
+esphome:
+  debug_scheduler: true  # Enable scheduler leak detection
+  name: scheduler-self-keyed-test
+  on_boot:
+    priority: -100
+    then:
+      - logger.log: "Starting scheduler self-keyed tests"
+
+host:
+api:
+logger:
+  level: VERBOSE
+
+globals:
+  - id: tests_done
+    type: bool
+    initial_value: 'false'
+
+script:
+  - id: test_self_keyed
+    then:
+      - logger.log: "Testing self-keyed scheduler API"
+      - lambda: |-
+          // Two distinct keys backed by addresses of static markers — they
+          // must not collide even though both are self-keyed and share no
+          // Component pointer. Static storage gives them stable, unique
+          // addresses for the lifetime of the program.
+          static int key_a_marker = 0;
+          static int key_b_marker = 0;
+          void *key_a = &key_a_marker;
+          void *key_b = &key_b_marker;
+
+          // ---- Test 1: Self-keyed timeout fires ----
+          App.scheduler.set_timeout(key_a, 50, []() {
+            ESP_LOGI("test", "Self timeout A fired");
+          });
+
+          // ---- Test 2: Self-keyed cancel cancels only that key ----
+          App.scheduler.set_timeout(key_b, 100, []() {
+            ESP_LOGE("test", "ERROR: Self timeout B should have been cancelled");
+          });
+          App.scheduler.cancel_timeout(key_b);
+
+          // ---- Test 3: Two independent self keys don't collide ----
+          // Using fresh static markers so neither matches key_a / key_b.
+          static int key_c_marker = 0;
+          static int key_d_marker = 0;
+          void *key_c = &key_c_marker;
+          void *key_d = &key_d_marker;
+          App.scheduler.set_timeout(key_c, 150, []() {
+            ESP_LOGI("test", "Self timeout C fired");
+          });
+          App.scheduler.set_timeout(key_d, 150, []() {
+            ESP_LOGI("test", "Self timeout D fired");
+          });
+
+          // ---- Test 4: Self-keyed and component-keyed don't collide ----
+          // Use a self pointer that happens to look like a Component-attached id.
+          // The scheduler must treat them as separate namespaces.
+          static int shared_marker = 0;
+          void *self_shared = &shared_marker;
+          App.scheduler.set_timeout(self_shared, 200, []() {
+            ESP_LOGI("test", "Self timeout shared fired");
+          });
+          App.scheduler.set_timeout(id(test_sensor), 7777U, 200, []() {
+            ESP_LOGI("test", "Component timeout 7777 fired");
+          });
+
+          // ---- Test 5: Self-keyed interval fires multiple times then cancels ----
+          static int interval_count = 0;
+          static int key_e_marker = 0;
+          void *key_e = &key_e_marker;
+          App.scheduler.set_interval(key_e, 80, [key_e]() {
+            interval_count++;
+            if (interval_count == 2) {
+              ESP_LOGI("test", "Self interval E fired twice");
+              App.scheduler.cancel_interval(key_e);
+            }
+          });
+
+          // ---- Test 6: Re-registering same self-key replaces the timer ----
+          // The old timer must NOT fire; only the new one does.
+          static int key_f_marker = 0;
+          void *key_f = &key_f_marker;
+          App.scheduler.set_timeout(key_f, 250, []() {
+            ESP_LOGE("test", "ERROR: Self timeout F first registration should have been replaced");
+          });
+          App.scheduler.set_timeout(key_f, 300, []() {
+            ESP_LOGI("test", "Self timeout F replacement fired");
+          });
+
+          // Log completion after all timers should have fired
+          App.scheduler.set_timeout(id(test_sensor), 9999U, 1500, []() {
+            ESP_LOGI("test", "All self-keyed tests complete");
+          });
+
+sensor:
+  - platform: template
+    name: Test Sensor
+    id: test_sensor
+    lambda: return 1.0;
+    update_interval: never
+
+interval:
+  - interval: 0.1s
+    then:
+      - if:
+          condition:
+            lambda: 'return id(tests_done) == false;'
+          then:
+            - lambda: 'id(tests_done) = true;'
+            - script.execute: test_self_keyed
diff --git a/tests/integration/fixtures/valve_control_action.yaml b/tests/integration/fixtures/valve_control_action.yaml
new file mode 100644
index 00000000000..4f43d16289c
--- /dev/null
+++ b/tests/integration/fixtures/valve_control_action.yaml
@@ -0,0 +1,69 @@
+esphome:
+  name: valve-control-action-test
+host:
+api:
+logger:
+  level: DEBUG
+
+globals:
+  - id: test_position
+    type: float
+    initial_value: "0.42"
+
+valve:
+  - platform: template
+    name: "Test Valve"
+    id: test_valve
+    has_position: true
+    optimistic: true
+    assumed_state: true
+    open_action:
+      - valve.template.publish:
+          id: test_valve
+          position: 1.0
+    close_action:
+      - valve.template.publish:
+          id: test_valve
+          position: 0.0
+    stop_action:
+      - valve.template.publish:
+          id: test_valve
+          current_operation: IDLE
+
+button:
+  # valve.control: position only
+  - platform: template
+    id: btn_position
+    name: "Set Position"
+    on_press:
+      - valve.control:
+          id: test_valve
+          position: 50%
+
+  # valve.control: state alias for position 1.0
+  - platform: template
+    id: btn_open_state
+    name: "Open State"
+    on_press:
+      - valve.control:
+          id: test_valve
+          state: OPEN
+
+  # valve.control: lambda position (exercises lambda path)
+  - platform: template
+    id: btn_lambda_position
+    name: "Lambda Position"
+    on_press:
+      - valve.control:
+          id: test_valve
+          position: !lambda "return id(test_position);"
+
+  # valve.control: stop only — template valve's stop_action publishes
+  # current_operation: IDLE.
+  - platform: template
+    id: btn_stop
+    name: "Stop Valve"
+    on_press:
+      - valve.control:
+          id: test_valve
+          stop: true
diff --git a/tests/integration/test_fan_turn_on_action.py b/tests/integration/test_fan_turn_on_action.py
new file mode 100644
index 00000000000..bce258cb5cd
--- /dev/null
+++ b/tests/integration/test_fan_turn_on_action.py
@@ -0,0 +1,75 @@
+"""Integration test for fan TurnOnAction.
+
+Tests that fan.turn_on automation actions work correctly across multiple
+field combinations and the lambda path.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+from aioesphomeapi import ButtonInfo, EntityState, FanDirection, FanInfo, FanState
+import pytest
+
+from .state_utils import InitialStateHelper, require_entity
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_fan_turn_on_action(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test fan TurnOnAction with constants and a lambda."""
+    loop = asyncio.get_running_loop()
+    async with run_compiled(yaml_config), api_client_connected() as client:
+        fan_state_future: asyncio.Future[FanState] | None = None
+
+        def on_state(state: EntityState) -> None:
+            if (
+                isinstance(state, FanState)
+                and fan_state_future is not None
+                and not fan_state_future.done()
+            ):
+                fan_state_future.set_result(state)
+
+        async def wait_for_fan_state(timeout: float = 5.0) -> FanState:
+            nonlocal fan_state_future
+            fan_state_future = loop.create_future()
+            try:
+                return await asyncio.wait_for(fan_state_future, timeout)
+            finally:
+                fan_state_future = None
+
+        entities, _ = await client.list_entities_services()
+        initial_state_helper = InitialStateHelper(entities)
+        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
+        await initial_state_helper.wait_for_initial_states()
+
+        require_entity(entities, "test_fan", FanInfo)
+
+        async def press_and_wait(name: str) -> FanState:
+            btn = require_entity(entities, name.lower().replace(" ", "_"), ButtonInfo)
+            client.button_command(btn.key)
+            return await wait_for_fan_state()
+
+        # speed only
+        state = await press_and_wait("Set Speed")
+        assert state.state is True
+        assert state.speed_level == 3
+
+        # oscillating + direction
+        state = await press_and_wait("Set Oscillate Direction")
+        assert state.oscillating is True
+        assert state.direction == FanDirection.REVERSE
+
+        # all three fields
+        state = await press_and_wait("Set All Fields")
+        assert state.oscillating is False
+        assert state.speed_level == 4
+        assert state.direction == FanDirection.FORWARD
+
+        # lambda path: speed computed at runtime (test_speed global = 2)
+        state = await press_and_wait("Lambda Speed")
+        assert state.speed_level == 2
diff --git a/tests/integration/test_scheduler_self_keyed.py b/tests/integration/test_scheduler_self_keyed.py
new file mode 100644
index 00000000000..e0825ea8254
--- /dev/null
+++ b/tests/integration/test_scheduler_self_keyed.py
@@ -0,0 +1,96 @@
+"""Test the self-keyed scheduler API.
+
+Verifies that `Scheduler::set_timeout(const void *, ...)` /
+`set_interval(const void *, ...)` and the matching `cancel_*(const void *)`
+overloads behave correctly: callbacks fire, distinct keys don't collide,
+self-keyed and component-keyed namespaces are independent, and re-registering
+the same key replaces the existing timer.
+"""
+
+import asyncio
+import re
+
+import pytest
+
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_scheduler_self_keyed(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test self-keyed scheduler API."""
+    self_a_fired = asyncio.Event()
+    self_b_error = asyncio.Event()
+    self_c_fired = asyncio.Event()
+    self_d_fired = asyncio.Event()
+    self_shared_fired = asyncio.Event()
+    component_7777_fired = asyncio.Event()
+    self_interval_done = asyncio.Event()
+    self_f_first_error = asyncio.Event()
+    self_f_replacement_fired = asyncio.Event()
+    all_tests_complete = asyncio.Event()
+
+    def on_log_line(line: str) -> None:
+        clean_line = re.sub(r"\x1b\[[0-9;]*m", "", line)
+
+        if "Self timeout A fired" in clean_line:
+            self_a_fired.set()
+        elif "ERROR: Self timeout B" in clean_line:
+            self_b_error.set()
+        elif "Self timeout C fired" in clean_line:
+            self_c_fired.set()
+        elif "Self timeout D fired" in clean_line:
+            self_d_fired.set()
+        elif "Self timeout shared fired" in clean_line:
+            self_shared_fired.set()
+        elif "Component timeout 7777 fired" in clean_line:
+            component_7777_fired.set()
+        elif "Self interval E fired twice" in clean_line:
+            self_interval_done.set()
+        elif "ERROR: Self timeout F first registration" in clean_line:
+            self_f_first_error.set()
+        elif "Self timeout F replacement fired" in clean_line:
+            self_f_replacement_fired.set()
+        elif "All self-keyed tests complete" in clean_line:
+            all_tests_complete.set()
+
+    async with (
+        run_compiled(yaml_config, line_callback=on_log_line),
+        api_client_connected() as client,
+    ):
+        device_info = await client.device_info()
+        assert device_info is not None
+        assert device_info.name == "scheduler-self-keyed-test"
+
+        try:
+            await asyncio.wait_for(all_tests_complete.wait(), timeout=5.0)
+        except TimeoutError:
+            pytest.fail("Not all self-keyed tests completed within 5 seconds")
+
+        # Test 1: self-keyed timeout fires
+        assert self_a_fired.is_set(), "Self timeout A should have fired"
+
+        # Test 2: cancel_timeout(self) actually cancels
+        assert not self_b_error.is_set(), "Self timeout B should have been cancelled"
+
+        # Test 3: distinct self keys don't collide
+        assert self_c_fired.is_set(), "Self timeout C should have fired"
+        assert self_d_fired.is_set(), "Self timeout D should have fired"
+
+        # Test 4: self-keyed and component-keyed namespaces are independent
+        assert self_shared_fired.is_set(), "Self timeout shared should have fired"
+        assert component_7777_fired.is_set(), "Component timeout 7777 should have fired"
+
+        # Test 5: self-keyed interval fires repeatedly and cancels cleanly
+        assert self_interval_done.is_set(), "Self interval E should have fired twice"
+
+        # Test 6: re-registering same self-key replaces the previous timer
+        assert not self_f_first_error.is_set(), (
+            "Self timeout F first registration should have been replaced"
+        )
+        assert self_f_replacement_fired.is_set(), (
+            "Self timeout F replacement should have fired"
+        )
diff --git a/tests/integration/test_valve_control_action.py b/tests/integration/test_valve_control_action.py
new file mode 100644
index 00000000000..d6515b8960e
--- /dev/null
+++ b/tests/integration/test_valve_control_action.py
@@ -0,0 +1,72 @@
+"""Integration test for valve ControlAction.
+
+Tests that valve.control automation actions work correctly across multiple
+field combinations and the lambda path.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+from aioesphomeapi import ButtonInfo, EntityState, ValveInfo, ValveOperation, ValveState
+import pytest
+
+from .state_utils import InitialStateHelper, require_entity
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_valve_control_action(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test valve ControlAction with constants and a lambda."""
+    loop = asyncio.get_running_loop()
+    async with run_compiled(yaml_config), api_client_connected() as client:
+        valve_state_future: asyncio.Future[ValveState] | None = None
+
+        def on_state(state: EntityState) -> None:
+            if (
+                isinstance(state, ValveState)
+                and valve_state_future is not None
+                and not valve_state_future.done()
+            ):
+                valve_state_future.set_result(state)
+
+        async def wait_for_valve_state(timeout: float = 5.0) -> ValveState:
+            nonlocal valve_state_future
+            valve_state_future = loop.create_future()
+            try:
+                return await asyncio.wait_for(valve_state_future, timeout)
+            finally:
+                valve_state_future = None
+
+        entities, _ = await client.list_entities_services()
+        initial_state_helper = InitialStateHelper(entities)
+        client.subscribe_states(initial_state_helper.on_state_wrapper(on_state))
+        await initial_state_helper.wait_for_initial_states()
+
+        require_entity(entities, "test_valve", ValveInfo)
+
+        async def press_and_wait(name: str) -> ValveState:
+            btn = require_entity(entities, name.lower().replace(" ", "_"), ButtonInfo)
+            client.button_command(btn.key)
+            return await wait_for_valve_state()
+
+        # valve.control: position only
+        state = await press_and_wait("Set Position")
+        assert state.position == pytest.approx(0.5, abs=0.01)
+
+        # valve.control: state alias for position 1.0
+        state = await press_and_wait("Open State")
+        assert state.position == pytest.approx(1.0, abs=0.01)
+
+        # valve.control: lambda position (test_position global = 0.42)
+        state = await press_and_wait("Lambda Position")
+        assert state.position == pytest.approx(0.42, abs=0.01)
+
+        # valve.control: stop only — template valve's stop_action publishes
+        # current_operation: IDLE.
+        state = await press_and_wait("Stop Valve")
+        assert state.current_operation == ValveOperation.IDLE
diff --git a/tests/unit_tests/test_external_files.py b/tests/unit_tests/test_external_files.py
index c894f906662..64ef1495817 100644
--- a/tests/unit_tests/test_external_files.py
+++ b/tests/unit_tests/test_external_files.py
@@ -469,6 +469,69 @@ def test_download_content_with_network_error_no_cache_fails(
         external_files.download_content(url, test_file)
 
 
+class _BodyReadErrorResponse:
+    """Stand-in for `requests.Response` whose `.content` raises on access.
+
+    A small dedicated stub avoids mutating `MagicMock`'s class with a
+    `property` (which would leak across every other MagicMock-based test
+    in this file).
+    """
+
+    def __init__(self, exc: Exception) -> None:
+        self._exc = exc
+        self.headers: dict[str, str] = {}
+
+    def raise_for_status(self) -> None:
+        return None
+
+    @property
+    def content(self) -> bytes:
+        raise self._exc
+
+
+def test_download_content_with_body_read_error_uses_cache(
+    mock_has_remote_file_changed: MagicMock,
+    mock_requests_get: MagicMock,
+    setup_core: Path,
+) -> None:
+    """Body-read errors (chunked-decode/gzip-decode/mid-stream connection
+    drop) raise RequestException subclasses on `.content` access, not from
+    `requests.get` itself. They must follow the same fall-back-to-cache
+    path as a connect-time failure.
+    """
+    test_file = setup_core / "cached.txt"
+    cached_content = b"cached content"
+    test_file.write_bytes(cached_content)
+
+    mock_has_remote_file_changed.return_value = True
+    mock_requests_get.return_value = _BodyReadErrorResponse(
+        requests.exceptions.ChunkedEncodingError("body truncated")
+    )
+
+    result = external_files.download_content("https://example.com/file.txt", test_file)
+
+    assert result == cached_content
+
+
+def test_download_content_with_body_read_error_no_cache_fails(
+    mock_has_remote_file_changed: MagicMock,
+    mock_requests_get: MagicMock,
+    setup_core: Path,
+) -> None:
+    """A body-read failure with no cache available must surface as a
+    cv.Invalid, same as a connect-time failure with no cache.
+    """
+    test_file = setup_core / "nonexistent.txt"
+
+    mock_has_remote_file_changed.return_value = True
+    mock_requests_get.return_value = _BodyReadErrorResponse(
+        requests.exceptions.ChunkedEncodingError("body truncated")
+    )
+
+    with pytest.raises(Invalid, match="Could not download from.*body truncated"):
+        external_files.download_content("https://example.com/file.txt", test_file)
+
+
 def test_download_content_skip_external_update_uses_cache(
     mock_has_remote_file_changed: MagicMock,
     mock_requests_get: MagicMock,
diff --git a/tests/unit_tests/test_main.py b/tests/unit_tests/test_main.py
index 8ec9e70cf85..fb8f206a1d2 100644
--- a/tests/unit_tests/test_main.py
+++ b/tests/unit_tests/test_main.py
@@ -2605,7 +2605,7 @@ def test_choose_upload_log_host_discovers_mac_suffix_devices(tmp_path: Path) ->
     }
     with (
         patch(
-            "esphome.__main__.discover_mdns_devices", return_value=discovered
+            "esphome.zeroconf.discover_mdns_devices", return_value=discovered
         ) as mock_discover,
         patch(
             "esphome.__main__.choose_prompt", return_value="mydevice-abc123.local"
@@ -2653,7 +2653,7 @@ def test_choose_upload_log_host_mac_suffix_no_devices_found(
     )
 
     with (
-        patch("esphome.__main__.discover_mdns_devices", return_value={}),
+        patch("esphome.zeroconf.discover_mdns_devices", return_value={}),
         caplog.at_level(logging.WARNING, logger="esphome.__main__"),
         pytest.raises(EsphomeError),
     ):
@@ -2686,7 +2686,7 @@ def test_choose_upload_log_host_default_ota_discovers_mac_suffix(
         "mydevice-def456.local": ["10.0.0.2"],
     }
     with patch(
-        "esphome.__main__.discover_mdns_devices", return_value=discovered
+        "esphome.zeroconf.discover_mdns_devices", return_value=discovered
     ) as mock_discover:
         result = choose_upload_log_host(
             default="OTA",
@@ -2715,7 +2715,7 @@ def test_choose_upload_log_host_default_ota_no_suffix_discovery(
         name="mydevice",
     )
 
-    with patch("esphome.__main__.discover_mdns_devices") as mock_discover:
+    with patch("esphome.zeroconf.discover_mdns_devices") as mock_discover:
         result = choose_upload_log_host(
             default="OTA",
             check_default=None,