[core] wake_loop_threadsafe() forces a component-phase iteration

After the Phase A / Phase B split in this PR, an external producer that
called wake_loop_threadsafe() (MQTT RX, USB RX, BLE event, espnow,
camera, mWW, speakers, USB host/CDC, lwip socket, enable_loop_soon_any_context)
only got Phase A — the component phase stayed gated by loop_interval_,
so the producer's component loop() could be delayed by up to
loop_interval_ ms before draining its queued work. That breaks the
long-standing semantic of wake_loop_threadsafe().

Add a wake_request flag set by every wake_loop_* entry point and
exchange-cleared at the gate in Application::loop(). When the flag is
set, force Phase B regardless of loop_interval_.

Storage is conditional on the threading model:
  - ESPHOME_THREAD_MULTI_ATOMICS: std::atomic<uint8_t> (uint8_t, not
    bool, because GCC on Xtensa generates an indirect call for
    atomic<bool> ops — same workaround as scheduler.h)
  - ESPHOME_THREAD_SINGLE / ESPHOME_THREAD_MULTI_NO_ATOMICS: volatile
    uint8_t (8-bit aligned loads/stores are atomic on every supported
    MCU; the platform signal that follows wake_request_set provides the
    cross-thread/cross-core memory barrier)

Helpers (wake_request_set / wake_request_take) are always_inline so
IRAM_ATTR call sites stay in IRAM. Set BEFORE the platform signal so the
consumer is guaranteed to see the flag on its next gate check.

Adds an integration test that raises loop_interval_ to 2s, snapshots a
counting component's loop count, spawns a std::thread that calls
App.wake_loop_threadsafe() after 50ms, and asserts the count increments
inside a 500ms observation window. Without the fix the count would not
move for ~2s.

esphome/core/application.h

@@ -660,12 +660,20 @@ inline void ESPHOME_ALWAYS_INLINE Application::loop() {
   uint32_t loop_tail_start_us = loop_before_end_us;
 #endif
 
-  // Gate the component phase on loop_interval_ or an active high-frequency
-  // request. When a scheduler wake preempts sleep early, this gate keeps the
-  // component phase from running more often than loop_interval_.
+  // Gate the component phase on loop_interval_, an active high-frequency
+  // request, or an explicit wake from a background producer. A scheduler-only
+  // wake (e.g. set_interval firing under a raised loop_interval_) leaves the
+  // component phase gated; an external producer that called wake_loop_*
+  // (MQTT RX, USB RX, BLE event, etc.) needs the component phase to actually
+  // run so its component's loop() can drain the queued work — that is the
+  // long-standing semantic of wake_loop_threadsafe(), and the wake_request
+  // flag preserves it. wake_request_take() exchange-clears the flag; wakes
+  // that arrive during Phase B re-set it and run Phase B again on the next
+  // iteration.
   const bool high_frequency = HighFrequencyLoopRequester::is_high_frequency();
   const uint32_t elapsed = now - this->last_loop_;
-  const bool do_component_phase = high_frequency || (elapsed >= this->loop_interval_);
+  const bool woke = esphome::wake_request_take();
+  const bool do_component_phase = high_frequency || woke || (elapsed >= this->loop_interval_);
 
   if (do_component_phase) {
     this->before_component_phase_();
@@ -716,8 +724,12 @@ inline void ESPHOME_ALWAYS_INLINE Application::loop() {
 #endif
 
   // Compute sleep: bounded by time-until-next-component-phase and the
-  // scheduler's next deadline. When a scheduler event wakes us early we
-  // re-enter loop(), Phase A services it, and the component phase stays gated.
+  // scheduler's next deadline. When a scheduler timer fires it re-enters
+  // loop(), Phase A services it, and the component phase stays gated by
+  // loop_interval_. When a background producer calls wake_loop_threadsafe()
+  // it sets the wake_request flag and wakes select() / the task notification;
+  // the gate above sees the flag and runs Phase B too so the producer's
+  // component can drain its queued work without waiting up to loop_interval_.
   //
   // Re-read HighFrequencyLoopRequester::is_high_frequency() here instead of
   // reusing the cached `high_frequency` captured above: a component calling

esphome/core/wake.cpp

@@ -12,9 +12,22 @@
 
 namespace esphome {
 
+// === Wake-requested flag storage ===
+#ifdef ESPHOME_THREAD_MULTI_ATOMICS
+// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+std::atomic<uint8_t> g_wake_requested{0};
+#else
+// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+volatile uint8_t g_wake_requested = 0;
+#endif
+
 // === ESP32 / LibreTiny — IRAM_ATTR entry points ===
 #if defined(USE_ESP32) || defined(USE_LIBRETINY)
 void IRAM_ATTR wake_loop_isrsafe(BaseType_t *px_higher_priority_task_woken) {
+  // ISR-safe: set flag before notify so the wake is visible on the next gate
+  // check. wake_request_set() is just an aligned 8-bit store / atomic store
+  // and is safe from IRAM.
+  wake_request_set();
   esphome_main_task_notify_from_isr(px_higher_priority_task_woken);
 }
 void IRAM_ATTR wake_loop_any_context() { wake_main_task_any_context(); }
@@ -72,6 +85,9 @@ void wakeable_delay(uint32_t ms) {
 // === Host (UDP loopback socket) ===
 #ifdef USE_HOST
 void wake_loop_threadsafe() {
+  // Set flag before sending so the consumer's gate check on the next loop()
+  // entry observes the wake regardless of select() scheduling.
+  wake_request_set();
   if (App.wake_socket_fd_ >= 0) {
     const char dummy = 1;
     ::send(App.wake_socket_fd_, &dummy, 1, 0);

esphome/core/wake.h

@@ -7,6 +7,10 @@
 #include "esphome/core/defines.h"
 #include "esphome/core/hal.h"
 
+#ifdef ESPHOME_THREAD_MULTI_ATOMICS
+#include <atomic>
+#endif
+
 #if defined(USE_ESP32) || defined(USE_LIBRETINY)
 #include "esphome/core/main_task.h"
 #endif
@@ -25,12 +29,48 @@ namespace esphome {
 extern volatile bool g_main_loop_woke;
 #endif
 
+// === wake_request flag — signals Application::loop() that a producer queued
+// work for some component's loop() to drain (MQTT RX, USB RX, BLE event, etc.)
+// and the component phase should run this tick instead of being held off by
+// the loop_interval_ gate. Set by every wake_loop_* entry point; consumed
+// (via exchange-and-clear) at the gate in Application::loop(). ===
+//
+// std::atomic<uint8_t> rather than std::atomic<bool> because GCC on Xtensa
+// generates an indirect function call for atomic<bool> ops instead of inlining
+// them — same workaround applied in scheduler.h for the SchedulerItem::remove
+// flag. On non-atomic platforms a volatile uint8_t suffices: 8-bit aligned
+// loads/stores are atomic on every supported MCU, and the platform signal
+// that follows wake_request_set() (FreeRTOS task-notify, esp_schedule, socket
+// send) provides the cross-thread/cross-core memory barrier.
+#ifdef ESPHOME_THREAD_MULTI_ATOMICS
+// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+extern std::atomic<uint8_t> g_wake_requested;
+
+__attribute__((always_inline)) inline void wake_request_set() { g_wake_requested.store(1, std::memory_order_release); }
+__attribute__((always_inline)) inline bool wake_request_take() {
+  return g_wake_requested.exchange(0, std::memory_order_acquire) != 0;
+}
+#else
+// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+extern volatile uint8_t g_wake_requested;
+
+__attribute__((always_inline)) inline void wake_request_set() { g_wake_requested = 1; }
+__attribute__((always_inline)) inline bool wake_request_take() {
+  uint8_t v = g_wake_requested;
+  g_wake_requested = 0;
+  return v != 0;
+}
+#endif
+
 // === ESP32 / LibreTiny (FreeRTOS) ===
 #if defined(USE_ESP32) || defined(USE_LIBRETINY)
 
 /// Wake the main loop from any context (ISR or task).
 /// always_inline so callers placed in IRAM keep the whole wake path in IRAM.
 __attribute__((always_inline)) inline void wake_main_task_any_context() {
+  // Set the wake-requested flag BEFORE the task notification so the consumer
+  // (Application::loop() gate) is guaranteed to see it on its next gate check.
+  wake_request_set();
   if (in_isr_context()) {
     BaseType_t px_higher_priority_task_woken = pdFALSE;
     esphome_main_task_notify_from_isr(&px_higher_priority_task_woken);
@@ -50,7 +90,10 @@ __attribute__((always_inline)) inline void wake_main_task_any_context() {
 void wake_loop_isrsafe(BaseType_t *px_higher_priority_task_woken);
 void wake_loop_any_context();
 
-inline void wake_loop_threadsafe() { esphome_main_task_notify(); }
+inline void wake_loop_threadsafe() {
+  wake_request_set();
+  esphome_main_task_notify();
+}
 
 namespace internal {
 inline void wakeable_delay(uint32_t ms) {
@@ -67,6 +110,9 @@ inline void wakeable_delay(uint32_t ms) {
 
 /// Inline implementation — IRAM callers inline this directly.
 inline void ESPHOME_ALWAYS_INLINE wake_loop_impl() {
+  // Set the wake-requested flag BEFORE esp_schedule so the consumer is
+  // guaranteed to see it on its next gate check.
+  wake_request_set();
   g_main_loop_woke = true;
   esp_schedule();
 }
@@ -98,6 +144,9 @@ inline void wakeable_delay(uint32_t ms) {
 #elif defined(USE_RP2040)
 
 inline void wake_loop_any_context() {
+  // Set the wake-requested flag BEFORE the SEV so the consumer is guaranteed
+  // to see it on its next gate check.
+  wake_request_set();
   g_main_loop_woke = true;
   __sev();
 }

tests/integration/fixtures/external_components/wake_test_component/__init__.py

@@ -0,0 +1,19 @@
+import esphome.codegen as cg
+import esphome.config_validation as cv
+from esphome.const import CONF_ID
+
+CODEOWNERS = ["@esphome/tests"]
+
+wake_test_component_ns = cg.esphome_ns.namespace("wake_test_component")
+WakeTestComponent = wake_test_component_ns.class_("WakeTestComponent", cg.Component)
+
+CONFIG_SCHEMA = cv.Schema(
+    {
+        cv.GenerateID(): cv.declare_id(WakeTestComponent),
+    }
+).extend(cv.COMPONENT_SCHEMA)
+
+
+async def to_code(config):
+    var = cg.new_Pvariable(config[CONF_ID])
+    await cg.register_component(var, config)

tests/integration/fixtures/external_components/wake_test_component/wake_test_component.cpp

@@ -0,0 +1,19 @@
+#include "wake_test_component.h"
+#include "esphome/core/application.h"
+#include "esphome/core/log.h"
+#include <chrono>
+#include <thread>
+
+namespace esphome::wake_test_component {
+
+static const char *const TAG = "wake_test_component";
+
+void WakeTestComponent::start_async_wake() {
+  ESP_LOGI(TAG, "Spawning async wake thread (50ms delay)");
+  std::thread([] {
+    std::this_thread::sleep_for(std::chrono::milliseconds(50));
+    App.wake_loop_threadsafe();
+  }).detach();
+}
+
+}  // namespace esphome::wake_test_component

tests/integration/fixtures/external_components/wake_test_component/wake_test_component.h

@@ -0,0 +1,27 @@
+#pragma once
+
+#include "esphome/core/component.h"
+#include <atomic>
+
+namespace esphome::wake_test_component {
+
+class WakeTestComponent : public Component {
+ public:
+  void setup() override {}
+  void loop() override { this->loop_count_.fetch_add(1, std::memory_order_relaxed); }
+
+  int get_loop_count() const { return this->loop_count_.load(std::memory_order_relaxed); }
+
+  // Spawn a detached thread that sleeps briefly then calls
+  // App.wake_loop_threadsafe(). Used by the integration test to verify a
+  // cross-thread wake forces a component-phase iteration even when
+  // loop_interval_ has been raised high enough to gate it off otherwise.
+  void start_async_wake();
+
+  float get_setup_priority() const override { return setup_priority::DATA; }
+
+ protected:
+  std::atomic<int> loop_count_{0};
+};
+
+}  // namespace esphome::wake_test_component

tests/integration/fixtures/wake_loop_forces_phase_b.yaml

@@ -0,0 +1,52 @@
+esphome:
+  name: wake-loop-phase-b
+  on_boot:
+    priority: -100
+    then:
+      - lambda: |-
+          // Raise loop_interval_ to 2000ms. Without the wake-request flag,
+          // a wake_loop_threadsafe() call would only run Phase A (scheduler)
+          // and leave the component phase gated for ~2s.
+          App.set_loop_interval(2000);
+      # Let boot transients settle.
+      - delay: 1000ms
+      - lambda: |-
+          // Snapshot the loop counter, then ask the component to spawn a
+          // background thread that calls App.wake_loop_threadsafe() after
+          // ~50ms. With the fix, that wake forces Phase B on the next tick
+          // and the counter increments well within the 500ms observation
+          // window below.
+          id(count_at_start) = id(wake_counter)->get_loop_count();
+          id(start_time) = millis();
+          id(wake_counter)->start_async_wake();
+          ESP_LOGI("test", "WAKE_STARTED count=%d", id(count_at_start));
+      # Observation window must be much shorter than loop_interval_ (2000ms)
+      # so a "false pass" isn't possible by simply waiting out the gate.
+      - delay: 500ms
+      - lambda: |-
+          int count_now = id(wake_counter)->get_loop_count();
+          int delta = count_now - id(count_at_start);
+          uint32_t elapsed = millis() - id(start_time);
+          ESP_LOGI("test", "WAKE_RESULT delta=%d elapsed=%u", delta, elapsed);
+
+host:
+api:
+logger:
+  level: INFO
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+    components: [wake_test_component]
+
+globals:
+  - id: count_at_start
+    type: int
+    initial_value: "0"
+  - id: start_time
+    type: uint32_t
+    initial_value: "0"
+
+wake_test_component:
+  id: wake_counter

tests/integration/test_wake_loop_forces_phase_b.py

@@ -0,0 +1,76 @@
+"""Test that wake_loop_threadsafe() forces a component-phase iteration.
+
+Regression test for the wake-request flag added to Application::loop()'s
+Phase A / Phase B gate. Background producers (MQTT RX, USB RX, BLE event,
+etc.) call App.wake_loop_threadsafe() expecting their component's loop()
+to drain queued work; if the component phase stays gated by loop_interval_,
+the work waits up to loop_interval_ ms instead of running on the next tick.
+
+Setup:
+- App.set_loop_interval(2000) — a wide gate that would clearly mask the bug.
+- A test component spawns a detached std::thread that sleeps 50 ms and then
+  calls App.wake_loop_threadsafe() from a non-main thread.
+- The on_boot block snapshots the component's loop counter before/after a
+  500 ms observation window.
+
+Without the fix, delta=0 (the gate holds Phase B for ~2 s).
+With the fix, delta>=1 (the wake forces Phase B within one tick of the wake).
+"""
+
+from __future__ import annotations
+
+import asyncio
+from pathlib import Path
+import re
+
+import pytest
+
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_wake_loop_forces_phase_b(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """A wake_loop_threadsafe() call from a background thread must trigger the
+    component phase within the next tick, even when loop_interval_ is raised
+    well above the observation window."""
+    external_components_path = str(
+        Path(__file__).parent / "fixtures" / "external_components"
+    )
+    yaml_config = yaml_config.replace(
+        "EXTERNAL_COMPONENT_PATH", external_components_path
+    )
+
+    loop = asyncio.get_running_loop()
+    result: asyncio.Future[tuple[int, int]] = loop.create_future()
+
+    def on_log_line(line: str) -> None:
+        match = re.search(r"WAKE_RESULT delta=(\d+) elapsed=(\d+)", line)
+        if match and not result.done():
+            result.set_result((int(match.group(1)), int(match.group(2))))
+
+    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 == "wake-loop-phase-b"
+
+        try:
+            delta, elapsed = await asyncio.wait_for(result, timeout=15.0)
+        except TimeoutError:
+            pytest.fail("WAKE_RESULT marker never appeared")
+
+        # Without the fix, delta would be 0 — loop_interval_=2000ms held
+        # Phase B off for the full 500ms observation window. With the fix
+        # the wake from the background thread (~50ms after start) forces
+        # Phase B on the next tick, so the counter increments at least once.
+        assert delta >= 1, (
+            f"wake_loop_threadsafe() from a background thread should force "
+            f"Phase B within the next tick; observed delta={delta} after "
+            f"{elapsed}ms with loop_interval_=2000ms"
+        )