diff --git a/.ai/instructions.md b/.ai/instructions.md
index a7e08f9c4d..86f554e9ce 100644
--- a/.ai/instructions.md
+++ b/.ai/instructions.md
@@ -239,6 +239,123 @@ This document provides essential context for AI models interacting with this pro
               var = await switch.new_switch(config)
           ```
 
+*   **Automations (Triggers, Actions, Conditions):**
+
+    Automations have three building blocks: **Triggers** (fire when something happens), **Actions** (do something), and **Conditions** (check if something is true).
+
+    *   **Triggers -- Callback method (preferred):**
+
+        Use `build_callback_automation()` for simple triggers. This eliminates the need for a C++ Trigger class by using a lightweight pointer-sized forwarder struct registered directly as a callback. No `CONF_TRIGGER_ID` in the schema.
+
+        **Python:**
+        ```python
+        from esphome import automation
+
+        CONFIG_SCHEMA = cv.Schema({
+            cv.GenerateID(): cv.declare_id(MyComponent),
+            cv.Optional(CONF_ON_STATE): automation.validate_automation({}),
+        }).extend(cv.COMPONENT_SCHEMA)
+
+        async def to_code(config):
+            var = cg.new_Pvariable(config[CONF_ID])
+            await cg.register_component(var, config)
+            for conf in config.get(CONF_ON_STATE, []):
+                await automation.build_callback_automation(
+                    var, "add_on_state_callback", [(bool, "x")], conf
+                )
+        ```
+
+        `build_callback_automation` arguments: `parent`, `callback_method` (C++ method name), `args` (template args as `[(type, name)]` tuples), `config`, and optional `forwarder` (defaults to `TriggerForwarder<Ts...>`).
+
+        For boolean filtering (e.g. `on_press`/`on_release`), use built-in forwarders with `args=[]`:
+        ```python
+        for conf_key, forwarder in (
+            (CONF_ON_PRESS, automation.TriggerOnTrueForwarder),
+            (CONF_ON_RELEASE, automation.TriggerOnFalseForwarder),
+        ):
+            for conf in config.get(conf_key, []):
+                await automation.build_callback_automation(
+                    var, "add_on_state_callback", [], conf, forwarder=forwarder
+                )
+        ```
+
+        **C++ -- no trigger class needed.** The callback registration method must be templatized to accept both `std::function` and lightweight forwarder structs (which avoid heap allocation):
+        ```cpp
+        class MyComponent : public Component {
+         public:
+          // Must be a template -- accepts both std::function and pointer-sized forwarder structs
+          template<typename F> void add_on_state_callback(F &&callback) {
+            this->state_callback_.add(std::forward<F>(callback));
+          }
+         protected:
+          // Use CallbackManager when callbacks are always registered (e.g. core components)
+          CallbackManager<void(bool)> state_callback_;
+          // Use LazyCallbackManager when callbacks are often not registered -- saves 8 bytes
+          // (nullptr vs empty std::vector) per instance when no callbacks are added
+          // LazyCallbackManager<void(bool)> state_callback_;
+        };
+        ```
+
+    *   **Triggers -- Trigger class method:**
+
+        Use `build_automation()` with a `Trigger<Ts...>` subclass only when the forwarder needs **mutable state beyond a single `Automation*` pointer** (e.g. edge detection tracking previous state, timing logic).
+
+        **Python:**
+        ```python
+        TurnOnTrigger = my_ns.class_("TurnOnTrigger", automation.Trigger.template())
+
+        CONFIG_SCHEMA = cv.Schema({
+            cv.Optional(CONF_ON_TURN_ON): automation.validate_automation(
+                {cv.GenerateID(CONF_TRIGGER_ID): cv.declare_id(TurnOnTrigger)}
+            ),
+        })
+
+        async def to_code(config):
+            for conf in config.get(CONF_ON_TURN_ON, []):
+                trigger = cg.new_Pvariable(conf[CONF_TRIGGER_ID], var)
+                await automation.build_automation(trigger, [], conf)
+        ```
+
+        **C++:**
+        ```cpp
+        class TurnOnTrigger : public Trigger<> {
+         public:
+          explicit TurnOnTrigger(MyComponent *parent) : last_on_{false} {
+            parent->add_on_state_callback([this](bool state) {
+              if (state && !this->last_on_)
+                this->trigger();
+              this->last_on_ = state;
+            });
+          }
+         protected:
+          bool last_on_;
+        };
+        ```
+
+    *   **Actions:**
+        ```cpp
+        template<typename... Ts> class MyAction : public Action<Ts...> {
+         public:
+          explicit MyAction(MyComponent *parent) : parent_(parent) {}
+          void play(const Ts &...) override { this->parent_->do_something(); }
+         protected:
+          MyComponent *parent_;
+        };
+        ```
+        Register with `@automation.register_action("my_component.do_something", MyAction, schema, synchronous=True)`. Use `synchronous=True` for actions that run to completion inside `play()` without deferring. Use `synchronous=False` if the action may suspend/defer execution (e.g. `delay`, `wait_until`, `script.wait`) or store trigger arguments for later use.
+
+    *   **Conditions:**
+        ```cpp
+        template<typename... Ts> class MyCondition : public Condition<Ts...> {
+         public:
+          explicit MyCondition(MyComponent *parent) : parent_(parent) {}
+          bool check(const Ts &...) override { return this->parent_->is_active(); }
+         protected:
+          MyComponent *parent_;
+        };
+        ```
+        Register with `@automation.register_condition("my_component.is_active", MyCondition, schema)`.
+
 *   **Configuration Validation:**
     *   **Common Validators:** `cv.int_`, `cv.float_`, `cv.string`, `cv.boolean`, `cv.int_range(min=0, max=100)`, `cv.positive_int`, `cv.percentage`.
     *   **Complex Validation:** `cv.All(cv.string, cv.Length(min=1, max=50))`, `cv.Any(cv.int_, cv.string)`.
@@ -274,10 +391,39 @@ This document provides essential context for AI models interacting with this pro
     *   **Component Tests:** YAML-based compilation tests are located in `tests/`. The structure is as follows:
         ```
         tests/
-        ├── test_build_components/ # Base test configurations
-        └── components/[component]/ # Component-specific tests
+        ├── test_build_components/
+        │   └── common/          # Shared bus packages (uart, i2c, spi, etc.)
+        │       ├── uart/        # UART at default baud rate
+        │       ├── uart_115200/ # UART at 115200 baud
+        │       ├── i2c/         # I2C bus
+        │       └── spi/         # SPI bus
+        └── components/[component]/
+            ├── common.yaml      # Component-only config (no bus definitions)
+            ├── test.esp32-idf.yaml
+            ├── test.esp8266-ard.yaml
+            └── test.rp2040-ard.yaml
         ```
         Run them using `script/test_build_components`. Use `-c <component>` to test specific components and `-t <target>` for specific platforms.
+
+    *   **Test Grouping with Packages:** Components that use shared bus packages can be grouped together in CI to reduce build count. **Never define buses (uart, i2c, spi, modbus) directly in test YAML files** — always use packages from `test_build_components/common/`:
+        ```yaml
+        # test.esp32-idf.yaml — use packages for buses
+        packages:
+          uart: !include ../../test_build_components/common/uart_115200/esp32-idf.yaml
+
+        <<: !include common.yaml
+        ```
+        ```yaml
+        # common.yaml — component config only, NO bus definitions
+        my_component:
+          id: my_instance
+
+        sensor:
+          - platform: my_component
+            name: My Sensor
+        ```
+        Components that define buses directly are flagged as "NEEDS MIGRATION" and cannot be grouped, increasing CI build time.
+
     *   **Testing All Components Together:** To verify that all components can be tested together without ID conflicts or configuration issues, use:
         ```bash
         ./script/test_component_grouping.py -e config --all
@@ -417,6 +563,30 @@ This document provides essential context for AI models interacting with this pro
 
         Note: Avoiding heap allocation after `setup()` is always required regardless of component type. The prioritization above is about the effort spent on container optimization (e.g., migrating from `std::vector` to `StaticVector`).
 
+        **Callback Managers:**
+
+        ESPHome provides two callback manager types in `esphome/core/helpers.h` for the observer pattern. Both support `std::function`, lambdas, and lightweight forwarder structs via their templatized `add()` method.
+
+        | Type | Idle overhead (32-bit) | When to use |
+        |------|----------------------|-------------|
+        | `CallbackManager<void(Ts...)>` | 12 bytes (empty `std::vector`) | Callbacks are always or almost always registered |
+        | `LazyCallbackManager<void(Ts...)>` | 4 bytes (`nullptr`) | Callbacks are often not registered (common case) |
+
+        `LazyCallbackManager` is a drop-in replacement for `CallbackManager` that defers allocation until the first callback is added. Prefer it for entity-level callbacks where most instances have no subscribers.
+
+        **Important:** Registration methods that add to a callback manager **must always be templatized** to accept both `std::function` and pointer-sized forwarder structs (used by `build_callback_automation`). Never use `std::function` in the method signature:
+        ```cpp
+        // Bad -- forces heap allocation for forwarder structs
+        void add_on_state_callback(std::function<void(bool)> &&callback) {
+          this->state_callback_.add(std::move(callback));
+        }
+
+        // Good -- accepts any callable without forcing std::function wrapping
+        template<typename F> void add_on_state_callback(F &&callback) {
+          this->state_callback_.add(std::forward<F>(callback));
+        }
+        ```
+
     *   **State Management:** Use `CORE.data` for component state that needs to persist during configuration generation. Avoid module-level mutable globals.
 
         **Bad Pattern (Module-Level Globals):**
diff --git a/esphome/components/api/api_frame_helper.cpp b/esphome/components/api/api_frame_helper.cpp
index 6d3bd51b58..f46693a4e8 100644
--- a/esphome/components/api/api_frame_helper.cpp
+++ b/esphome/components/api/api_frame_helper.cpp
@@ -103,7 +103,8 @@ const LogString *api_error_to_logstr(APIError err) {
 APIError APIFrameHelper::drain_overflow_and_handle_errors_() {
   if (this->overflow_buf_.try_drain(this->socket_.get()) == -1) {
     int err = errno;
-    if (this->check_socket_write_err_(err) != APIError::WOULD_BLOCK) {
+    if (err != EWOULDBLOCK && err != EAGAIN) {
+      this->state_ = State::FAILED;
       HELPER_LOG("Socket write failed with errno %d", err);
       return APIError::SOCKET_WRITE_FAILED;
     }
@@ -111,45 +112,51 @@ APIError APIFrameHelper::drain_overflow_and_handle_errors_() {
   return APIError::OK;
 }
 
-// Write data to socket, overflow to backlog buffer if LWIP TCP send buffer is full.
-// Returns OK if all data was sent or successfully queued.
-// Returns SOCKET_WRITE_FAILED on hard error (sets state to FAILED).
-APIError APIFrameHelper::write_raw_(const struct iovec *iov, int iovcnt, uint16_t total_write_len) {
+// Single-buffer write path: wraps in iovec and delegates.
+APIError APIFrameHelper::write_raw_buf_(const void *data, uint16_t len, ssize_t sent) {
+  struct iovec iov = {const_cast<void *>(data), len};
+  return this->write_raw_iov_(&iov, 1, len, sent);
+}
+
+// Handles partial writes, errors, and overflow buffering.
+// Called when the inline fast path in the header couldn't complete the write,
+// or directly from cold paths (handshake, error handling).
+// sent == -1 means either the fast path write returned -1, or there was overflow backlog.
+APIError APIFrameHelper::write_raw_iov_(const struct iovec *iov, int iovcnt, uint16_t total_write_len, ssize_t sent) {
 #ifdef HELPER_LOG_PACKETS
   for (int i = 0; i < iovcnt; i++) {
     LOG_PACKET_SENDING(reinterpret_cast<uint8_t *>(iov[i].iov_base), iov[i].iov_len);
   }
 #endif
 
-  uint16_t skip = 0;
-
-  // Drain any existing backlog first
-  if (!this->overflow_buf_.empty()) [[unlikely]] {
-    APIError err = this->drain_overflow_and_handle_errors_();
-    if (err != APIError::OK)
-      return err;
-  }
-
-  // If backlog is clear, try direct send
-  if (this->overflow_buf_.empty()) [[likely]] {
-    ssize_t sent =
-        (iovcnt == 1) ? this->socket_->write(iov[0].iov_base, iov[0].iov_len) : this->socket_->writev(iov, iovcnt);
-
-    if (sent == -1) [[unlikely]] {
+  if (sent == -1) {
+    // Either the fast path got -1, or we were called with overflow backlog
+    if (!this->overflow_buf_.empty()) {
+      // Drain existing backlog first
+      APIError err = this->drain_overflow_and_handle_errors_();
+      if (err != APIError::OK)
+        return err;
+      // Try again after drain
+      if (this->overflow_buf_.empty()) {
+        sent =
+            (iovcnt == 1) ? this->socket_->write(iov[0].iov_base, iov[0].iov_len) : this->socket_->writev(iov, iovcnt);
+        if (sent == static_cast<ssize_t>(total_write_len))
+          return APIError::OK;
+      }
+    }
+    if (sent == -1) {
       int err = errno;
-      if (this->check_socket_write_err_(err) != APIError::WOULD_BLOCK) {
+      if (err != EWOULDBLOCK && err != EAGAIN) {
+        this->state_ = State::FAILED;
         HELPER_LOG("Socket write failed with errno %d", err);
         return APIError::SOCKET_WRITE_FAILED;
       }
-    } else if (static_cast<uint16_t>(sent) >= total_write_len) [[likely]] {
-      return APIError::OK;
-    } else {
-      skip = static_cast<uint16_t>(sent);
+      sent = 0;  // Treat WOULD_BLOCK as zero bytes sent
     }
   }
 
   // Queue unsent data into overflow buffer
-  if (!this->overflow_buf_.enqueue_iov(iov, iovcnt, total_write_len, skip)) {
+  if (!this->overflow_buf_.enqueue_iov(iov, iovcnt, total_write_len, static_cast<uint16_t>(sent))) {
     HELPER_LOG("Overflow buffer full, dropping connection");
     this->state_ = State::FAILED;
     return APIError::SOCKET_WRITE_FAILED;
diff --git a/esphome/components/api/api_frame_helper.h b/esphome/components/api/api_frame_helper.h
index 72ccf8aa56..94007eeb3d 100644
--- a/esphome/components/api/api_frame_helper.h
+++ b/esphome/components/api/api_frame_helper.h
@@ -161,17 +161,13 @@ class APIFrameHelper {
       this->nodelay_counter_ = 0;
     }
   }
-  APIError write_protobuf_packet(uint8_t type, ProtoWriteBuffer buffer) {
-    // Resize buffer to include footer space if needed (e.g. Noise MAC)
-    if (frame_footer_size_)
-      buffer.get_buffer()->resize(buffer.get_buffer()->size() + frame_footer_size_);
-    MessageInfo msg{type, 0,
-                    static_cast<uint16_t>(buffer.get_buffer()->size() - frame_header_padding_ - frame_footer_size_)};
-    return write_protobuf_messages(buffer, std::span<const MessageInfo>(&msg, 1));
-  }
-  // Write multiple protobuf messages in a single operation
-  // messages contains (message_type, offset, length) for each message in the buffer
-  // The buffer contains all messages with appropriate padding before each
+  // Write a single protobuf message - the hot path (87-100% of all writes).
+  // Caller must ensure state is DATA before calling.
+  virtual APIError write_protobuf_packet(uint8_t type, ProtoWriteBuffer buffer) = 0;
+  // Write multiple protobuf messages in a single batched operation.
+  // Caller must ensure state is DATA and messages is not empty.
+  // messages contains (message_type, offset, length) for each message in the buffer.
+  // The buffer contains all messages with appropriate padding before each.
   virtual APIError write_protobuf_messages(ProtoWriteBuffer buffer, std::span<const MessageInfo> messages) = 0;
   // Get the frame header padding required by this protocol
   uint8_t frame_header_padding() const { return frame_header_padding_; }
@@ -196,17 +192,32 @@ class APIFrameHelper {
   // Returns OK for transient errors (WOULD_BLOCK), SOCKET_WRITE_FAILED for hard errors.
   APIError drain_overflow_and_handle_errors_();
 
-  // Common implementation for writing raw data to socket
-  APIError write_raw_(const struct iovec *iov, int iovcnt, uint16_t total_write_len);
-
-  // Check if a socket write errno is a hard error (not WOULD_BLOCK/EAGAIN).
-  // Returns WOULD_BLOCK for transient errors, SOCKET_WRITE_FAILED for hard errors.
-  APIError check_socket_write_err_(int err) {
-    if (err == EWOULDBLOCK || err == EAGAIN)
-      return APIError::WOULD_BLOCK;
-    this->state_ = State::FAILED;
-    return APIError::SOCKET_WRITE_FAILED;
+  // Inlined write methods — used by hot paths (write_protobuf_packet, write_protobuf_messages)
+  // These inline the fast path (overflow empty + full write) and tail-call the out-of-line
+  // slow path only on failure/partial write.
+  inline APIError ESPHOME_ALWAYS_INLINE write_raw_fast_buf_(const void *data, uint16_t len) {
+    ssize_t sent = -1;
+    if (this->overflow_buf_.empty()) [[likely]] {
+      sent = this->socket_->write(data, len);
+      if (sent == static_cast<ssize_t>(len)) [[likely]]
+        return APIError::OK;
+    }
+    return this->write_raw_buf_(data, len, sent);
   }
+  inline APIError ESPHOME_ALWAYS_INLINE write_raw_fast_iov_(const struct iovec *iov, int iovcnt,
+                                                            uint16_t total_write_len) {
+    ssize_t sent = -1;
+    if (this->overflow_buf_.empty()) [[likely]] {
+      sent = this->socket_->writev(iov, iovcnt);
+      if (sent == static_cast<ssize_t>(total_write_len)) [[likely]]
+        return APIError::OK;
+    }
+    return this->write_raw_iov_(iov, iovcnt, total_write_len, sent);
+  }
+
+  // Out-of-line write paths: handle partial writes, errors, overflow buffering
+  APIError write_raw_buf_(const void *data, uint16_t len, ssize_t sent = -1);
+  APIError write_raw_iov_(const struct iovec *iov, int iovcnt, uint16_t total_write_len, ssize_t sent = -1);
 
   // Socket ownership (4 bytes on 32-bit, 8 bytes on 64-bit)
   std::unique_ptr<socket::Socket> socket_;
diff --git a/esphome/components/api/api_frame_helper_noise.cpp b/esphome/components/api/api_frame_helper_noise.cpp
index 5b60cee143..0c0c069b53 100644
--- a/esphome/components/api/api_frame_helper_noise.cpp
+++ b/esphome/components/api/api_frame_helper_noise.cpp
@@ -488,9 +488,32 @@ APIError APINoiseFrameHelper::encrypt_noise_message_(uint8_t *buf_start, const M
   return APIError::OK;
 }
 
-// Outlined multi-message path to keep the single-message fast path's stack frame small.
-APIError __attribute__((noinline))
-APINoiseFrameHelper::write_protobuf_messages_batch_(uint8_t *buffer_data, std::span<const MessageInfo> messages) {
+APIError APINoiseFrameHelper::write_protobuf_packet(uint8_t type, ProtoWriteBuffer buffer) {
+#ifdef ESPHOME_DEBUG_API
+  assert(this->state_ == State::DATA);
+#endif
+
+  // Resize buffer to include footer space for Noise MAC
+  if (frame_footer_size_)
+    buffer.get_buffer()->resize(buffer.get_buffer()->size() + frame_footer_size_);
+
+  MessageInfo msg{type, 0,
+                  static_cast<uint16_t>(buffer.get_buffer()->size() - frame_header_padding_ - frame_footer_size_)};
+  uint8_t *buf_start = buffer.get_buffer()->data();
+  struct iovec iov;
+  APIError aerr = this->encrypt_noise_message_(buf_start, msg, iov);
+  if (aerr != APIError::OK)
+    return aerr;
+  return this->write_raw_fast_buf_(iov.iov_base, static_cast<uint16_t>(iov.iov_len));
+}
+
+APIError APINoiseFrameHelper::write_protobuf_messages(ProtoWriteBuffer buffer, std::span<const MessageInfo> messages) {
+#ifdef ESPHOME_DEBUG_API
+  assert(this->state_ == State::DATA);
+  assert(!messages.empty());
+#endif
+
+  uint8_t *buffer_data = buffer.get_buffer()->data();
   StaticVector<struct iovec, MAX_MESSAGES_PER_BATCH> iovs;
   uint16_t total_write_len = 0;
 
@@ -504,33 +527,7 @@ APINoiseFrameHelper::write_protobuf_messages_batch_(uint8_t *buffer_data, std::s
     total_write_len += iov.iov_len;
   }
 
-  return this->write_raw_(iovs.data(), iovs.size(), total_write_len);
-}
-
-APIError APINoiseFrameHelper::write_protobuf_messages(ProtoWriteBuffer buffer, std::span<const MessageInfo> messages) {
-  APIError aerr = this->check_data_state_();
-  if (aerr != APIError::OK)
-    return aerr;
-
-  if (messages.empty()) {
-    return APIError::OK;
-  }
-
-  uint8_t *buffer_data = buffer.get_buffer()->data();
-
-  if (messages.size() == 1) [[likely]] {
-    // Peeled first iteration: single-message case (most common path via write_protobuf_packet)
-    // avoids StaticVector stack allocation and loop overhead
-    const auto &first = messages[0];
-    struct iovec iov;
-    aerr = this->encrypt_noise_message_(buffer_data + first.offset, first, iov);
-    if (aerr != APIError::OK)
-      return aerr;
-    return this->write_raw_(&iov, 1, static_cast<uint16_t>(iov.iov_len));
-  }
-
-  // Multiple messages: outlined to avoid large stack frame on single-message path
-  return this->write_protobuf_messages_batch_(buffer_data, messages);
+  return this->write_raw_fast_iov_(iovs.data(), iovs.size(), total_write_len);
 }
 
 APIError APINoiseFrameHelper::write_frame_(const uint8_t *data, uint16_t len) {
@@ -539,16 +536,16 @@ APIError APINoiseFrameHelper::write_frame_(const uint8_t *data, uint16_t len) {
   header[1] = (uint8_t) (len >> 8);
   header[2] = (uint8_t) len;
 
+  if (len == 0) {
+    return this->write_raw_buf_(header, 3);
+  }
   struct iovec iov[2];
   iov[0].iov_base = header;
   iov[0].iov_len = 3;
-  if (len == 0) {
-    return this->write_raw_(iov, 1, 3);  // Just header
-  }
   iov[1].iov_base = const_cast<uint8_t *>(data);
   iov[1].iov_len = len;
 
-  return this->write_raw_(iov, 2, 3 + len);  // Header + data
+  return this->write_raw_iov_(iov, 2, 3 + len);
 }
 
 /** Initiate the data structures for the handshake.
diff --git a/esphome/components/api/api_frame_helper_noise.h b/esphome/components/api/api_frame_helper_noise.h
index 7a81d2032e..e56006d955 100644
--- a/esphome/components/api/api_frame_helper_noise.h
+++ b/esphome/components/api/api_frame_helper_noise.h
@@ -22,6 +22,7 @@ class APINoiseFrameHelper final : public APIFrameHelper {
   APIError init() override;
   APIError loop() override;
   APIError read_packet(ReadPacketBuffer *buffer) override;
+  APIError write_protobuf_packet(uint8_t type, ProtoWriteBuffer buffer) override;
   APIError write_protobuf_messages(ProtoWriteBuffer buffer, std::span<const MessageInfo> messages) override;
 
  protected:
@@ -29,7 +30,6 @@ class APINoiseFrameHelper final : public APIFrameHelper {
   APIError try_read_frame_();
   APIError write_frame_(const uint8_t *data, uint16_t len);
   APIError encrypt_noise_message_(uint8_t *buf_start, const MessageInfo &msg, struct iovec &iov_out);
-  APIError write_protobuf_messages_batch_(uint8_t *buffer_data, std::span<const MessageInfo> messages);
   APIError init_handshake_();
   APIError check_handshake_finished_();
   void send_explicit_handshake_reject_(const LogString *reason);
diff --git a/esphome/components/api/api_frame_helper_plaintext.cpp b/esphome/components/api/api_frame_helper_plaintext.cpp
index 64c8882afa..6be20f89f3 100644
--- a/esphome/components/api/api_frame_helper_plaintext.cpp
+++ b/esphome/components/api/api_frame_helper_plaintext.cpp
@@ -205,7 +205,6 @@ APIError APIPlaintextFrameHelper::read_packet(ReadPacketBuffer *buffer) {
       // Make sure to tell the remote that we don't
       // understand the indicator byte so it knows
       // we do not support it.
-      struct iovec iov[1];
       // The \x00 first byte is the marker for plaintext.
       //
       // The remote will know how to handle the indicator byte,
@@ -220,14 +219,12 @@ APIError APIPlaintextFrameHelper::read_packet(ReadPacketBuffer *buffer) {
                                                 "Bad indicator byte";
       char msg[INDICATOR_MSG_SIZE];
       memcpy_P(msg, MSG_PROGMEM, INDICATOR_MSG_SIZE);
-      iov[0].iov_base = (void *) msg;
+      this->write_raw_buf_(msg, INDICATOR_MSG_SIZE);
 #else
       static const char MSG[] = "\x00"
                                 "Bad indicator byte";
-      iov[0].iov_base = (void *) MSG;
+      this->write_raw_buf_(MSG, INDICATOR_MSG_SIZE);
 #endif
-      iov[0].iov_len = INDICATOR_MSG_SIZE;
-      this->write_raw_(iov, 1, INDICATOR_MSG_SIZE);
     }
     return aerr;
   }
@@ -239,8 +236,8 @@ APIError APIPlaintextFrameHelper::read_packet(ReadPacketBuffer *buffer) {
 }
 // Write plaintext header into pre-allocated padding before payload.
 // Returns pointer to start of frame (header + payload are contiguous).
-static inline uint8_t *write_plaintext_header(uint8_t *buf_start, const MessageInfo &msg,
-                                              uint8_t frame_header_padding) {
+ESPHOME_ALWAYS_INLINE static inline uint8_t *write_plaintext_header(uint8_t *buf_start, const MessageInfo &msg,
+                                                                    uint8_t frame_header_padding) {
   // Calculate varint sizes for header layout using inline ternary to avoid varint_slow call overhead
   uint8_t size_varint_len = msg.payload_size < ProtoSize::VARINT_THRESHOLD_1_BYTE
                                 ? 1
@@ -285,11 +282,27 @@ static inline uint8_t *write_plaintext_header(uint8_t *buf_start, const MessageI
   return buf_start + header_offset;
 }
 
-// Outlined multi-message path to keep the single-message fast path's stack frame small.
-// The StaticVector<iovec, MAX_MESSAGES_PER_BATCH> would force a ~300-byte stack frame
-// even when only sending one message if it were in the same function.
-APIError __attribute__((noinline))
-APIPlaintextFrameHelper::write_protobuf_messages_batch_(uint8_t *buffer_data, std::span<const MessageInfo> messages) {
+APIError APIPlaintextFrameHelper::write_protobuf_packet(uint8_t type, ProtoWriteBuffer buffer) {
+#ifdef ESPHOME_DEBUG_API
+  assert(this->state_ == State::DATA);
+#endif
+
+  MessageInfo msg{type, 0, static_cast<uint16_t>(buffer.get_buffer()->size() - frame_header_padding_)};
+  uint8_t *buffer_data = buffer.get_buffer()->data();
+  uint8_t *msg_start = write_plaintext_header(buffer_data, msg, frame_header_padding_);
+  uint8_t msg_header_len = static_cast<uint8_t>(buffer_data + frame_header_padding_ - msg_start);
+  uint16_t msg_len = static_cast<uint16_t>(msg_header_len + msg.payload_size);
+  return this->write_raw_fast_buf_(msg_start, msg_len);
+}
+
+APIError APIPlaintextFrameHelper::write_protobuf_messages(ProtoWriteBuffer buffer,
+                                                          std::span<const MessageInfo> messages) {
+#ifdef ESPHOME_DEBUG_API
+  assert(this->state_ == State::DATA);
+  assert(!messages.empty());
+#endif
+
+  uint8_t *buffer_data = buffer.get_buffer()->data();
   StaticVector<struct iovec, MAX_MESSAGES_PER_BATCH> iovs;
   uint16_t total_write_len = 0;
   const uint8_t padding = frame_header_padding_;
@@ -302,34 +315,7 @@ APIPlaintextFrameHelper::write_protobuf_messages_batch_(uint8_t *buffer_data, st
     total_write_len += msg_len;
   }
 
-  return write_raw_(iovs.data(), iovs.size(), total_write_len);
-}
-
-APIError APIPlaintextFrameHelper::write_protobuf_messages(ProtoWriteBuffer buffer,
-                                                          std::span<const MessageInfo> messages) {
-  APIError aerr = this->check_data_state_();
-  if (aerr != APIError::OK)
-    return aerr;
-
-  if (messages.empty()) {
-    return APIError::OK;
-  }
-
-  uint8_t *buffer_data = buffer.get_buffer()->data();
-
-  if (messages.size() == 1) [[likely]] {
-    // Peeled first iteration: single-message case (most common path via write_protobuf_packet)
-    // avoids StaticVector stack allocation and loop overhead
-    const auto &first = messages[0];
-    uint8_t *first_start = write_plaintext_header(buffer_data + first.offset, first, frame_header_padding_);
-    uint8_t first_header_len = static_cast<uint8_t>((buffer_data + first.offset + frame_header_padding_) - first_start);
-    size_t first_len = static_cast<size_t>(first_header_len + first.payload_size);
-    struct iovec iov = {first_start, first_len};
-    return write_raw_(&iov, 1, static_cast<uint16_t>(first_len));
-  }
-
-  // Multiple messages: outlined to avoid large stack frame on single-message path
-  return write_protobuf_messages_batch_(buffer_data, messages);
+  return this->write_raw_fast_iov_(iovs.data(), iovs.size(), total_write_len);
 }
 
 }  // namespace esphome::api
diff --git a/esphome/components/api/api_frame_helper_plaintext.h b/esphome/components/api/api_frame_helper_plaintext.h
index 84311cb93a..96d47e9c7b 100644
--- a/esphome/components/api/api_frame_helper_plaintext.h
+++ b/esphome/components/api/api_frame_helper_plaintext.h
@@ -19,11 +19,11 @@ class APIPlaintextFrameHelper final : public APIFrameHelper {
   APIError init() override;
   APIError loop() override;
   APIError read_packet(ReadPacketBuffer *buffer) override;
+  APIError write_protobuf_packet(uint8_t type, ProtoWriteBuffer buffer) override;
   APIError write_protobuf_messages(ProtoWriteBuffer buffer, std::span<const MessageInfo> messages) override;
 
  protected:
   APIError try_read_frame_();
-  APIError write_protobuf_messages_batch_(uint8_t *buffer_data, std::span<const MessageInfo> messages);
 
   // Group 2-byte aligned types
   uint16_t rx_header_parsed_type_ = 0;
diff --git a/esphome/components/ch422g/ch422g.cpp b/esphome/components/ch422g/ch422g.cpp
index 5f5e848c76..fc856cd563 100644
--- a/esphome/components/ch422g/ch422g.cpp
+++ b/esphome/components/ch422g/ch422g.cpp
@@ -124,12 +124,6 @@ bool CH422GComponent::write_outputs_() {
 
 float CH422GComponent::get_setup_priority() const { return setup_priority::IO; }
 
-#ifdef USE_LOOP_PRIORITY
-// Run our loop() method very early in the loop, so that we cache read values
-// before other components call our digital_read() method.
-float CH422GComponent::get_loop_priority() const { return 9.0f; }  // Just after WIFI
-#endif
-
 void CH422GGPIOPin::pin_mode(gpio::Flags flags) { this->parent_->pin_mode(this->pin_, flags); }
 bool CH422GGPIOPin::digital_read() { return this->parent_->digital_read(this->pin_) ^ this->inverted_; }
 
diff --git a/esphome/components/ch422g/ch422g.h b/esphome/components/ch422g/ch422g.h
index 1b96568209..6e6bdad64a 100644
--- a/esphome/components/ch422g/ch422g.h
+++ b/esphome/components/ch422g/ch422g.h
@@ -23,9 +23,6 @@ class CH422GComponent : public Component, public i2c::I2CDevice {
   void pin_mode(uint8_t pin, gpio::Flags flags);
 
   float get_setup_priority() const override;
-#ifdef USE_LOOP_PRIORITY
-  float get_loop_priority() const override;
-#endif
   void dump_config() override;
 
  protected:
diff --git a/esphome/components/ch423/ch423.cpp b/esphome/components/ch423/ch423.cpp
index 805d8df877..8424d130b4 100644
--- a/esphome/components/ch423/ch423.cpp
+++ b/esphome/components/ch423/ch423.cpp
@@ -129,12 +129,6 @@ bool CH423Component::write_outputs_() {
 
 float CH423Component::get_setup_priority() const { return setup_priority::IO; }
 
-#ifdef USE_LOOP_PRIORITY
-// Run our loop() method very early in the loop, so that we cache read values
-// before other components call our digital_read() method.
-float CH423Component::get_loop_priority() const { return 9.0f; }  // Just after WIFI
-#endif
-
 void CH423GPIOPin::pin_mode(gpio::Flags flags) { this->parent_->pin_mode(this->pin_, flags); }
 bool CH423GPIOPin::digital_read() { return this->parent_->digital_read(this->pin_) ^ this->inverted_; }
 
diff --git a/esphome/components/ch423/ch423.h b/esphome/components/ch423/ch423.h
index d85648a8f9..d384971a72 100644
--- a/esphome/components/ch423/ch423.h
+++ b/esphome/components/ch423/ch423.h
@@ -22,9 +22,6 @@ class CH423Component : public Component, public i2c::I2CDevice {
   void pin_mode(uint8_t pin, gpio::Flags flags);
 
   float get_setup_priority() const override;
-#ifdef USE_LOOP_PRIORITY
-  float get_loop_priority() const override;
-#endif
   void dump_config() override;
 
  protected:
diff --git a/esphome/components/deep_sleep/deep_sleep_component.cpp b/esphome/components/deep_sleep/deep_sleep_component.cpp
index 0511518419..3dd1b70930 100644
--- a/esphome/components/deep_sleep/deep_sleep_component.cpp
+++ b/esphome/components/deep_sleep/deep_sleep_component.cpp
@@ -40,12 +40,6 @@ void DeepSleepComponent::loop() {
     this->begin_sleep();
 }
 
-#ifdef USE_LOOP_PRIORITY
-float DeepSleepComponent::get_loop_priority() const {
-  return -100.0f;  // run after everything else is ready
-}
-#endif
-
 void DeepSleepComponent::set_sleep_duration(uint32_t time_ms) { this->sleep_duration_ = uint64_t(time_ms) * 1000; }
 
 void DeepSleepComponent::set_run_duration(uint32_t time_ms) { this->run_duration_ = time_ms; }
diff --git a/esphome/components/deep_sleep/deep_sleep_component.h b/esphome/components/deep_sleep/deep_sleep_component.h
index 14713d51a1..9090f91876 100644
--- a/esphome/components/deep_sleep/deep_sleep_component.h
+++ b/esphome/components/deep_sleep/deep_sleep_component.h
@@ -113,9 +113,6 @@ class DeepSleepComponent : public Component {
   void setup() override;
   void dump_config() override;
   void loop() override;
-#ifdef USE_LOOP_PRIORITY
-  float get_loop_priority() const override;
-#endif
   float get_setup_priority() const override;
 
   /// Helper to enter deep sleep mode
diff --git a/esphome/components/esp8266/__init__.py b/esphome/components/esp8266/__init__.py
index 2728533723..552e842575 100644
--- a/esphome/components/esp8266/__init__.py
+++ b/esphome/components/esp8266/__init__.py
@@ -43,14 +43,14 @@ from .const import (
 from .gpio import PinInitialState, add_pin_initial_states_array
 
 CONF_ENABLE_SCANF_FLOAT = "enable_scanf_float"
-# Matches scanf/sscanf calls with float format specifiers:
-#   %f, %.2f, %6.2f  - basic float formats
-#   %lf, %Lf         - double/long double
-#   %e, %E, %g, %G   - scientific/general notation
-#   %*f               - assignment suppression
-#   %8lf              - width + length modifier
-# Uses [\s\S]*? to match across newlines in multi-line lambdas.
-_SCANF_FLOAT_RE = re.compile(r"scanf\s*\([\s\S]*?%[*\d.]*[hlL]*[feEgGaA]")
+# Heuristically matches scanf/sscanf calls with float format specifiers.
+# Standard scanf float conversions: %f %F %e %E %g %G %a %A
+# With optional modifiers: %*f (suppression), %8f (width), %lf %Lf (length)
+# Also matches non-standard patterns like %.2f as a heuristic — these are
+# invalid in scanf but users may write them by analogy with printf.
+# Uses [^;]*? to stay within a single statement, preventing false positives
+# from e.g. sscanf(buf, "%d", &x); printf("%f", val);
+_SCANF_FLOAT_RE = re.compile(r"scanf\s*\([^;]*?%[*\d.]*[hlL]*[feEgGaAF]")
 
 CODEOWNERS = ["@esphome/core"]
 _LOGGER = logging.getLogger(__name__)
@@ -213,7 +213,7 @@ CONFIG_SCHEMA = cv.All(
             cv.Optional(CONF_ENABLE_SERIAL): cv.boolean,
             cv.Optional(CONF_ENABLE_SERIAL1): cv.boolean,
             cv.Optional(CONF_ENABLE_FULL_PRINTF, default=False): cv.boolean,
-            cv.Optional(CONF_ENABLE_SCANF_FLOAT, default=False): cv.boolean,
+            cv.Optional(CONF_ENABLE_SCANF_FLOAT): cv.boolean,
         }
     ),
     set_core_data,
@@ -234,8 +234,8 @@ async def to_code(config):
     cg.add_define("ESPHOME_VARIANT", "ESP8266")
     cg.add_define(ThreadModel.SINGLE)
 
-    enable_scanf_float = config[CONF_ENABLE_SCANF_FLOAT]
-    if not enable_scanf_float and _lambdas_use_scanf_float(CORE.config):
+    enable_scanf_float = config.get(CONF_ENABLE_SCANF_FLOAT)
+    if enable_scanf_float is None and _lambdas_use_scanf_float(CORE.config):
         enable_scanf_float = True
         _LOGGER.warning(
             "Lambda uses scanf with a float format specifier; "
diff --git a/esphome/components/pca9554/pca9554.cpp b/esphome/components/pca9554/pca9554.cpp
index d94767ef07..adc7bc0fb5 100644
--- a/esphome/components/pca9554/pca9554.cpp
+++ b/esphome/components/pca9554/pca9554.cpp
@@ -122,11 +122,6 @@ bool PCA9554Component::write_register_(uint8_t reg, uint16_t value) {
 
 float PCA9554Component::get_setup_priority() const { return setup_priority::IO; }
 
-#ifdef USE_LOOP_PRIORITY
-// Run our loop() method early to invalidate cache before any other components access the pins
-float PCA9554Component::get_loop_priority() const { return 9.0f; }  // Just after WIFI
-#endif
-
 void PCA9554GPIOPin::setup() { pin_mode(flags_); }
 void PCA9554GPIOPin::pin_mode(gpio::Flags flags) { this->parent_->pin_mode(this->pin_, flags); }
 bool PCA9554GPIOPin::digital_read() { return this->parent_->digital_read(this->pin_) != this->inverted_; }
diff --git a/esphome/components/pca9554/pca9554.h b/esphome/components/pca9554/pca9554.h
index 6dd15ccb4b..1d877f9ce2 100644
--- a/esphome/components/pca9554/pca9554.h
+++ b/esphome/components/pca9554/pca9554.h
@@ -23,10 +23,6 @@ class PCA9554Component : public Component,
 
   float get_setup_priority() const override;
 
-#ifdef USE_LOOP_PRIORITY
-  float get_loop_priority() const override;
-#endif
-
   void dump_config() override;
 
   void set_pin_count(size_t pin_count) { this->pin_count_ = pin_count; }
diff --git a/esphome/components/pcf8574/pcf8574.cpp b/esphome/components/pcf8574/pcf8574.cpp
index fa9496e7e4..d3ec31436d 100644
--- a/esphome/components/pcf8574/pcf8574.cpp
+++ b/esphome/components/pcf8574/pcf8574.cpp
@@ -99,11 +99,6 @@ bool PCF8574Component::write_gpio_() {
 }
 float PCF8574Component::get_setup_priority() const { return setup_priority::IO; }
 
-#ifdef USE_LOOP_PRIORITY
-// Run our loop() method early to invalidate cache before any other components access the pins
-float PCF8574Component::get_loop_priority() const { return 9.0f; }  // Just after WIFI
-#endif
-
 void PCF8574GPIOPin::setup() { pin_mode(flags_); }
 void PCF8574GPIOPin::pin_mode(gpio::Flags flags) { this->parent_->pin_mode(this->pin_, flags); }
 bool PCF8574GPIOPin::digital_read() { return this->parent_->digital_read(this->pin_) != this->inverted_; }
diff --git a/esphome/components/pcf8574/pcf8574.h b/esphome/components/pcf8574/pcf8574.h
index 23bccc26c9..b039173789 100644
--- a/esphome/components/pcf8574/pcf8574.h
+++ b/esphome/components/pcf8574/pcf8574.h
@@ -26,9 +26,6 @@ class PCF8574Component : public Component,
   void pin_mode(uint8_t pin, gpio::Flags flags);
 
   float get_setup_priority() const override;
-#ifdef USE_LOOP_PRIORITY
-  float get_loop_priority() const override;
-#endif
 
   void dump_config() override;
 
diff --git a/esphome/components/status_led/light/status_led_light.h b/esphome/components/status_led/light/status_led_light.h
index a5c98d90d4..3a745e0017 100644
--- a/esphome/components/status_led/light/status_led_light.h
+++ b/esphome/components/status_led/light/status_led_light.h
@@ -30,9 +30,6 @@ class StatusLEDLightOutput : public light::LightOutput, public Component {
   void dump_config() override;
 
   float get_setup_priority() const override { return setup_priority::HARDWARE; }
-#ifdef USE_LOOP_PRIORITY
-  float get_loop_priority() const override { return 50.0f; }
-#endif
 
  protected:
   GPIOPin *pin_{nullptr};
diff --git a/esphome/components/status_led/status_led.cpp b/esphome/components/status_led/status_led.cpp
index 93a8d4b38e..a792110eeb 100644
--- a/esphome/components/status_led/status_led.cpp
+++ b/esphome/components/status_led/status_led.cpp
@@ -28,9 +28,6 @@ void StatusLED::loop() {
   }
 }
 float StatusLED::get_setup_priority() const { return setup_priority::HARDWARE; }
-#ifdef USE_LOOP_PRIORITY
-float StatusLED::get_loop_priority() const { return 50.0f; }
-#endif
 
 }  // namespace status_led
 }  // namespace esphome
diff --git a/esphome/components/status_led/status_led.h b/esphome/components/status_led/status_led.h
index f262eb260c..a4b5db93d7 100644
--- a/esphome/components/status_led/status_led.h
+++ b/esphome/components/status_led/status_led.h
@@ -14,9 +14,6 @@ class StatusLED : public Component {
   void dump_config() override;
   void loop() override;
   float get_setup_priority() const override;
-#ifdef USE_LOOP_PRIORITY
-  float get_loop_priority() const override;
-#endif
 
  protected:
   GPIOPin *pin_;
diff --git a/esphome/components/wifi/wifi_component.cpp b/esphome/components/wifi/wifi_component.cpp
index f58a9b4a46..3ed243134c 100644
--- a/esphome/components/wifi/wifi_component.cpp
+++ b/esphome/components/wifi/wifi_component.cpp
@@ -970,12 +970,6 @@ void WiFiComponent::set_ap(const WiFiAP &ap) {
 }
 #endif  // USE_WIFI_AP
 
-#ifdef USE_LOOP_PRIORITY
-float WiFiComponent::get_loop_priority() const {
-  return 10.0f;  // before other loop components
-}
-#endif
-
 void WiFiComponent::init_sta(size_t count) { this->sta_.init(count); }
 void WiFiComponent::add_sta(const WiFiAP &ap) { this->sta_.push_back(ap); }
 void WiFiComponent::clear_sta() {
diff --git a/esphome/components/wifi/wifi_component.h b/esphome/components/wifi/wifi_component.h
index 55e532c37d..8dfe5fa7af 100644
--- a/esphome/components/wifi/wifi_component.h
+++ b/esphome/components/wifi/wifi_component.h
@@ -463,10 +463,6 @@ class WiFiComponent final : public Component {
   void restart_adapter();
   /// WIFI setup_priority.
   float get_setup_priority() const override;
-#ifdef USE_LOOP_PRIORITY
-  float get_loop_priority() const override;
-#endif
-
   /// Reconnect WiFi if required.
   void loop() override;
 
diff --git a/esphome/core/application.cpp b/esphome/core/application.cpp
index ce15aed1e2..5cb8a5bb24 100644
--- a/esphome/core/application.cpp
+++ b/esphome/core/application.cpp
@@ -99,12 +99,6 @@ void Application::setup() {
     if (component->can_proceed())
       continue;
 
-#ifdef USE_LOOP_PRIORITY
-    // Sort components 0 through i by loop priority
-    insertion_sort_by_priority<decltype(this->components_.begin()), &Component::get_loop_priority>(
-        this->components_.begin(), this->components_.begin() + i + 1);
-#endif
-
     do {
       uint8_t new_app_state = STATUS_LED_WARNING;
       uint32_t now = millis();
diff --git a/esphome/core/component.cpp b/esphome/core/component.cpp
index c81e357808..00a36fce3d 100644
--- a/esphome/core/component.cpp
+++ b/esphome/core/component.cpp
@@ -87,10 +87,6 @@ static constexpr uint16_t WARN_IF_BLOCKING_INCREMENT_MS =
 // Threshold in ms (computed from centiseconds constant in component.h)
 static constexpr uint32_t WARN_IF_BLOCKING_OVER_MS = static_cast<uint32_t>(WARN_IF_BLOCKING_OVER_CS) * 10U;
 
-#ifdef USE_LOOP_PRIORITY
-float Component::get_loop_priority() const { return 0.0f; }
-#endif
-
 float Component::get_setup_priority() const { return setup_priority::DATA; }
 
 void Component::setup() {}
diff --git a/esphome/core/component.h b/esphome/core/component.h
index 86c25cf09c..c390a205f0 100644
--- a/esphome/core/component.h
+++ b/esphome/core/component.h
@@ -122,16 +122,6 @@ class Component {
 
   void set_setup_priority(float priority);
 
-  /** priority of loop(). higher -> executed earlier
-   *
-   * Defaults to 0.
-   *
-   * @return The loop priority of this component
-   */
-#ifdef USE_LOOP_PRIORITY
-  virtual float get_loop_priority() const;
-#endif
-
   void call();
 
   virtual void on_shutdown() {}
diff --git a/esphome/core/defines.h b/esphome/core/defines.h
index 58fbc64624..d95fb6f090 100644
--- a/esphome/core/defines.h
+++ b/esphome/core/defines.h
@@ -358,7 +358,6 @@
 
 #ifdef USE_RP2040
 #define USE_ARDUINO_VERSION_CODE VERSION_CODE(3, 3, 0)
-#define USE_LOOP_PRIORITY
 #define USE_RP2040_CRASH_HANDLER
 #define USE_HTTP_REQUEST_RESPONSE
 #define USE_I2C
diff --git a/esphome/core/entity_base.h b/esphome/core/entity_base.h
index d03326ccf8..5a69c9dd09 100644
--- a/esphome/core/entity_base.h
+++ b/esphome/core/entity_base.h
@@ -307,9 +307,10 @@ void log_entity_unit_of_measurement(const char *tag, const char *prefix, const E
  *   - get_trigger_on_initial_state(): return whether callbacks should fire on the first state
  *
  * Subclasses may override set_new_state() to add behavior (logging, notifications) after calling
- * the base implementation. Since set_new_state() is virtual, callers like invalidate_state() and
- * send_state_internal() dispatch through the vtable to the subclass override in the .cpp,
- * avoiding template code bloat at inline call sites.
+ * the base implementation. Since set_new_state() is virtual, callers like invalidate_state()
+ * dispatch through the vtable to the subclass override in the .cpp, avoiding template code
+ * bloat at inline call sites. Subclasses may also add a fast-path dedup check before calling
+ * set_new_state() to skip virtual dispatch entirely when the state hasn't changed.
  *
  * Callback behavior:
  *   - full_state_callbacks_: fired on every change, receives optional<T> previous and current
diff --git a/tests/integration/test_uart_mock_ld2410.py b/tests/integration/test_uart_mock_ld2410.py
index ce0e1bb7ec..88d6f2cbac 100644
--- a/tests/integration/test_uart_mock_ld2410.py
+++ b/tests/integration/test_uart_mock_ld2410.py
@@ -73,9 +73,16 @@ async def test_uart_mock_ld2410(
         ],
     )
 
-    # Signal when we see recovery frame values
+    # Signal when we see ALL recovery frame values to avoid race where some
+    # arrive after the waiter fires but before we index into the lists
     recovery_received = collector.add_waiter(
-        lambda: pytest.approx(50.0) in collector.sensor_states["moving_distance"]
+        lambda: (
+            pytest.approx(50.0) in collector.sensor_states["moving_distance"]
+            and pytest.approx(75.0) in collector.sensor_states["still_distance"]
+            and pytest.approx(100.0) in collector.sensor_states["moving_energy"]
+            and pytest.approx(80.0) in collector.sensor_states["still_energy"]
+            and pytest.approx(127.0) in collector.sensor_states["detection_distance"]
+        )
     )
 
     async with (
diff --git a/tests/unit_tests/components/test_esp8266.py b/tests/unit_tests/components/test_esp8266.py
new file mode 100644
index 0000000000..318fd2d889
--- /dev/null
+++ b/tests/unit_tests/components/test_esp8266.py
@@ -0,0 +1,62 @@
+"""Tests for ESP8266 component."""
+
+import pytest
+
+from esphome.components.esp8266 import lambdas_use_scanf_float
+from esphome.core import Lambda
+from esphome.types import ConfigType
+
+
+@pytest.mark.parametrize(
+    ("src", "expected"),
+    [
+        # Basic float formats
+        ('sscanf(buf, "%f", &v)', True),
+        ('sscanf(buf, "%F", &v)', True),
+        ('sscanf(buf, "%e", &v)', True),
+        ('sscanf(buf, "%E", &v)', True),
+        ('sscanf(buf, "%g", &v)', True),
+        ('sscanf(buf, "%G", &v)', True),
+        ('sscanf(buf, "%a", &v)', True),
+        ('sscanf(buf, "%A", &v)', True),
+        # With modifiers
+        ('sscanf(buf, "%lf", &v)', True),
+        ('sscanf(buf, "%Lf", &v)', True),
+        ('sscanf(buf, "%8lf", &v)', True),
+        ('sscanf(buf, "%*f")', True),
+        ('sscanf(buf, "%.2f", &v)', True),
+        # Mixed formats
+        ('sscanf(buf, "%d,%f", &a, &b)', True),
+        # fscanf and std::sscanf
+        ('fscanf(fp, "%f", &v)', True),
+        ('std::sscanf(buf, "%f", &v)', True),
+        # Multi-line
+        ('sscanf(buf,\n"%f", &v)', True),
+        # No float format
+        ('sscanf(buf, "%d", &v)', False),
+        ('sscanf(buf, "%s", s)', False),
+        # printf not scanf
+        ('printf("%f", val)', False),
+        # %f in a different statement after scanf
+        ('sscanf(buf, "%d", &x); printf("%f", val);', False),
+        # scanf %f in comment only
+        ('// sscanf(buf, "%f", &v)\nsscanf(buf, "%d", &x)', False),
+        ('/* sscanf(buf, "%f") */\nsscanf(buf, "%d", &x)', False),
+    ],
+)
+def test_lambdas_use_scanf_float(src: str, expected: bool) -> None:
+    """Test scanf float detection in lambda source."""
+    config: ConfigType = {"test": [Lambda(src)]}
+    assert lambdas_use_scanf_float(config) is expected
+
+
+def test_lambdas_use_scanf_float_no_lambdas() -> None:
+    """Test with config containing no lambdas."""
+    config: ConfigType = {"key": "value", "list": [1, 2]}
+    assert lambdas_use_scanf_float(config) is False
+
+
+def test_lambdas_use_scanf_float_nested() -> None:
+    """Test detection in deeply nested config."""
+    config: ConfigType = {"a": {"b": {"c": [Lambda('sscanf(buf, "%f", &v)')]}}}
+    assert lambdas_use_scanf_float(config) is True