Locate a while warning inside a list condition and restore App state in its test

A list condition is wrapped in an and: with no place in the file, so the
warning now names and locates the first condition in it, and leaves the
location out rather than pointing at the main file. The shutdown test
restores App's state bits, so later tests still yield.

.agents/skills/automations/SKILL.md

@@ -210,8 +210,10 @@ run, so prefer `register_apply_action` wherever the action allows it. Because of
 return fewer objects than the config has entries. A builder that wraps `apply_action_builder` may do other work
 first, but must return its result unchanged and not use it. A builder that must generate its own statements, and
 is registered with `automation.ApplyAction`, can call `apply_statements_action` and still merge; `script.execute`
-is an example. `if`, `choose`, `while` and `repeat` whose bodies are all inline become C++ control flow in the
-same function, and `register_apply_condition` and `lambda:` conditions become plain expressions there.
+is an example. `if`, `choose`, `repeat` and a `while` with trigger arguments whose bodies are all inline become
+C++ control flow in the same function, and `register_apply_condition` and `lambda:` conditions become plain
+expressions there. A `while` without trigger arguments becomes an `ApplyWhileAction` instead, which can yield to the
+main loop; with arguments a loop never yields.
 
 ## Conditions
 

esphome/automation.py

@@ -1,4 +1,4 @@
-from collections.abc import Awaitable, Callable, Sequence
+from collections.abc import Awaitable, Callable, Iterator, Sequence
 from contextvars import ContextVar
 from dataclasses import dataclass, field
 import logging
@@ -20,6 +20,7 @@ from esphome.const import (
     CONF_ELSE,
     CONF_ID,
     CONF_IF,
+    CONF_LAMBDA,
     CONF_THEN,
     CONF_TIME,
     CONF_TIMEOUT,
@@ -324,8 +325,10 @@ class _CollectedAction(MockObj):
 
 _COLLECTED_ACTION = _CollectedAction("esphome__collected_action", "->")
 _RUNNING_PARAM = "esphome__running"
-# One statement of its own, so a function that nothing can stop can leave them all out
+# Its own statement, matched exactly: _direct_function drops it and a while step rewrites it to _STOP_STEP
 _STOP_CHECK = f"if (!{_RUNNING_PARAM})\n  return;"
+# The same for a while step, where a stop ends the loop like a false condition
+_STOP_STEP = f"if (!{_RUNNING_PARAM})\n  return false;"
 
 
 def _is_collecting(id_: ID) -> bool:
@@ -989,6 +992,7 @@ LambdaAction = cg.esphome_ns.class_("LambdaAction", Action)
 StatelessLambdaAction = cg.esphome_ns.class_("StatelessLambdaAction", Action)
 IfAction = cg.esphome_ns.class_("IfAction", Action)
 WhileAction = cg.esphome_ns.class_("WhileAction", Action)
+ApplyWhileAction = cg.esphome_ns.class_("ApplyWhileAction", Action)
 RepeatAction = cg.esphome_ns.class_("RepeatAction", Action)
 WaitUntilAction = cg.esphome_ns.class_("WaitUntilAction", Action, cg.Component)
 UpdateComponentAction = cg.esphome_ns.class_("UpdateComponentAction", Action)
@@ -1388,6 +1392,7 @@ def _validate_choose_else(choices: list[ConfigType]) -> list[ConfigType]:
     return choices
 
 
+# _iter_actions knows this shape: a list of choices whose if: holds a condition
 @register_action(
     CONF_CHOOSE,
     IfAction,
@@ -1460,14 +1465,71 @@ async def choose_action_to_code(
     return var
 
 
+def _is_id_reference(id_: Any) -> bool:
+    # An id generated later is still None during validation
+    return isinstance(id_, ID) and not id_.is_declaration and id_.id is not None
+
+
+def _action_ids(actions: list[tuple[str, Any]]) -> set[str]:
+    """The ids the actions act on: their ``id`` option or shorthand id, and every id a lambda action names."""
+    ids: set[str] = set()
+    for key, options in actions:
+        if key == CONF_LAMBDA:
+            ids |= _referenced_ids(options)
+        elif _is_id_reference(
+            target := options.get(CONF_ID) if isinstance(options, dict) else options
+        ):
+            ids.add(target.id)
+    return ids
+
+
+def _referenced_ids(value: Any) -> set[str]:
+    # Lazy: importing esphome.config here would load it for every entity component
+    from esphome.config import iter_ids
+
+    return {id_.id for id_, _ in iter_ids(value) if _is_id_reference(id_)}
+
+
+def _warn_while_without_wait(config: ConfigType) -> ConfigType:
+    # Warn when the body never waits and acts on nothing the condition (or its lambda) reads, the
+    # ids a lambda action names included. An empty body ends the loop at once.
+    body = config[CONF_THEN]
+    if not body:
+        return config
+    actions = list(_iter_actions(body))
+    if any(_is_deferring(key) for key, _ in actions):
+        return config
+    if not _referenced_ids(config[CONF_CONDITION]) & _action_ids(actions):
+        # Lazy, see _referenced_ids
+        from esphome.config import _format_location
+
+        condition = config[CONF_CONDITION]
+        # A list condition is wrapped in an `and:` that has no place in the file
+        while not hasattr(key := _registry_key(condition), "esp_range") and (
+            isinstance(items := condition[key], list) and items
+        ):
+            condition = items[0]
+        location = getattr(key, "esp_range", None)
+        _LOGGER.warning(
+            "%sthe body of this 'while' loop on '%s' never waits, so the loop holds up "
+            "the device until the body changes the condition. Add a 'delay' inside the loop.",
+            f"{_format_location(location.start_mark)}: " if location else "",
+            key,
+        )
+    return config
+
+
 @register_action(
     "while",
     WhileAction,
-    cv.Schema(
-        {
-            cv.Required(CONF_CONDITION): validate_potentially_and_condition,
-            cv.Required(CONF_THEN): validate_action_list,
-        }
+    cv.All(
+        cv.Schema(
+            {
+                cv.Required(CONF_CONDITION): validate_potentially_and_condition,
+                cv.Required(CONF_THEN): validate_action_list,
+            }
+        ),
+        _warn_while_without_wait,
     ),
     synchronous=True,
 )
@@ -1482,17 +1544,41 @@ async def while_action_to_code(
     actions = await build_list(config[CONF_THEN], template_arg, args)
     # An empty body exits the loop at once, unlike an empty C++ loop body
     if _is_collecting(action_id) and actions and _plans_inline(actions):
-        return _collect_control_flow(
+        check = (
+            f"const bool esphome__condition = {_condition_expression(condition, args)};"
+        )
+        if args:
+            # Never yields, see WHILE_YIELD_AFTER_MS
+            return _collect_control_flow(
+                action_id,
+                [
+                    "while (true) {",
+                    check,
+                    _STOP_CHECK,
+                    "if (!esphome__condition)\n  break;",
+                    *_plan_statements(actions),
+                    _STOP_CHECK,
+                    "}",
+                ],
+            )
+        # A step function the ApplyWhileAction runs once per pass
+        return _new_running_action(
             action_id,
+            ApplyWhileAction,
+            cg.bool_,
             [
-                "while (true) {",
-                f"const bool esphome__condition = {_condition_expression(condition, args)};",
-                _STOP_CHECK,
-                "if (!esphome__condition)\n  break;",
-                *_plan_statements(actions),
-                _STOP_CHECK,
-                "}",
+                check,
+                f"if (!{_RUNNING_PARAM} || !esphome__condition)\n  return false;",
+                # Every stop check, also those of nested inline if and repeat, is its own line
+                *(
+                    _STOP_STEP if line == _STOP_CHECK else line
+                    for line in _plan_statements(actions)
+                ),
+                # The last body action can stop the automation too
+                f"return {_RUNNING_PARAM} != 0;",
             ],
+            template_arg,
+            args,
         )
     var = cg.new_Pvariable(action_id, template_arg, _condition_object(condition))
     for action in _emit_action_plan(actions, template_arg, args):
@@ -1694,10 +1780,14 @@ def _check_collected(result: MockObj, collector: _InlineCollector) -> None:
         )
 
 
+def _registry_key(full_config: ConfigType) -> str:
+    """The action or condition name of a validated entry."""
+    return next(key for key in full_config if key != CONF_TYPE_ID)
+
+
 def _builder_name(full_config: ConfigType) -> str:
     """For an error: the id is usually generated, so the registry key tells the author where to look."""
-    name = next(key for key in full_config if key != CONF_TYPE_ID)
-    return f"'{name}' ({full_config[CONF_TYPE_ID]})"
+    return f"'{_registry_key(full_config)}' ({full_config[CONF_TYPE_ID]})"
 
 
 async def _build_condition_or_inline(
@@ -1743,16 +1833,31 @@ def _emit_inline_run(
     """
     if len(run) == 1 and run[0].emit is not None:
         return run[0].emit()
-    statements = _run_statements(run)
-    id_ = run[0].id_.copy()
-    id_.type = ApplyRunAction
+    return _new_running_action(
+        run[0].id_,
+        ApplyRunAction,
+        cg.void,
+        _run_statements(run),
+        template_arg,
+        args,
+    )
+
+
+def _new_running_action(
+    source_id: ID,
+    action_type: MockObjClass,
+    return_type: SafeExpType,
+    statements: list[str],
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
+    """An ``action_type`` object running a generated function that reads ``num_running_``."""
+    id_ = source_id.copy()
+    id_.type = action_type
     fn = apply_static_function(
         id_,
-        cg.void,
-        [
-            (cg.RawExpression("const int &"), _RUNNING_PARAM),
-            *_apply_params(args),
-        ],
+        return_type,
+        [(cg.RawExpression("const int &"), _RUNNING_PARAM), *_apply_params(args)],
         statements,
     )
     return cg.new_Pvariable(id_, cg.TemplateArguments(fn, *template_arg))
@@ -1830,24 +1935,44 @@ def has_non_synchronous_actions(actions: ConfigType) -> bool:
     option that shares a name with an action, such as the ``delay:`` option of
     ``speaker_source.set_playlist_delay``, is not mistaken for that action.
     """
+    return any(_is_deferring(key) for key, _ in _iter_actions(actions))
+
+
+def _is_deferring(key: str) -> bool:
+    return key in ACTION_REGISTRY and not ACTION_REGISTRY[key].synchronous
+
+
+def _iter_actions(actions: Any) -> Iterator[tuple[str, Any]]:
+    """Yield ``(key, options)`` of every action in an action list, a single action or an automation.
+
+    Nested action lists (``then:``, ``else:``, ...) are searched; other options, including conditions, are not
+    taken for actions.
+    """
     if isinstance(actions, list):
-        return any(has_non_synchronous_actions(item) for item in actions)
-    if isinstance(actions, dict):
+        for item in actions:
+            yield from _iter_actions(item)
+    elif isinstance(actions, dict):
         for key, options in actions.items():
-            if key in ACTION_REGISTRY and not ACTION_REGISTRY[key].synchronous:
-                return True
-            if _options_have_non_synchronous_actions(options):
-                return True
-    return False
+            yield key, options
+            if key == CONF_CHOOSE:
+                # A list of choices, not of actions
+                for choice in options:
+                    yield from _iter_option_actions(choice)
+            else:
+                yield from _iter_option_actions(options)
 
 
-def _options_have_non_synchronous_actions(options: Any) -> bool:
-    """Search an action's options for nested action lists, without treating option keys as actions."""
+# Options that hold conditions, not actions, though and:/or: are lists
+_CONDITION_OPTIONS = frozenset((CONF_CONDITION, CONF_ANY, CONF_ALL, CONF_IF))
+
+
+def _iter_option_actions(options: Any) -> Iterator[tuple[str, Any]]:
     if isinstance(options, list):
-        return has_non_synchronous_actions(options)
-    if isinstance(options, dict):
-        return any(_options_have_non_synchronous_actions(v) for v in options.values())
-    return False
+        yield from _iter_actions(options)
+    elif isinstance(options, dict):
+        for key, value in options.items():
+            if key not in _CONDITION_OPTIONS:
+                yield from _iter_option_actions(value)
 
 
 def owning_args(args: TemplateArgsType, conf: ConfigType) -> TemplateArgsType:

esphome/core/application.cpp

@@ -291,6 +291,7 @@ bool Application::any_component_has_status_flag_(uint8_t flag) const {
 
 void Application::reboot() {
   ESP_LOGI(TAG, "Forcing a reboot");
+  this->app_state_ |= APP_STATE_SHUTTING_DOWN;
   for (auto &component : std::ranges::reverse_view(this->components_)) {
     component->on_shutdown();
   }
@@ -309,6 +310,7 @@ void Application::safe_shutdown(uint32_t timeout_ms) {
 }
 
 void Application::run_safe_shutdown_hooks() {
+  this->app_state_ |= APP_STATE_SHUTTING_DOWN;
   for (auto &component : std::ranges::reverse_view(this->components_)) {
     component->on_safe_shutdown();
   }

esphome/core/application.h

@@ -318,7 +318,7 @@ class Application {
   /// Return the public app state status bits (STATUS_LED_* only).
   /// Internal bookkeeping bits like APP_STATE_SETUP_COMPLETE are masked
   /// out so external readers (status_led components, etc.) never see them.
-  uint8_t get_app_state() const { return this->app_state_ & ~APP_STATE_SETUP_COMPLETE; }
+  uint8_t get_app_state() const { return this->app_state_ & ~(APP_STATE_SETUP_COMPLETE | APP_STATE_SHUTTING_DOWN); }
 
   /// True once Application::setup() has finished walking all components
   /// and finalized the initial status flags. Before this point, the
@@ -328,6 +328,10 @@ class Application {
   /// (bit 6) to avoid costing additional RAM.
   bool is_setup_complete() const { return (this->app_state_ & APP_STATE_SETUP_COMPLETE) != 0; }
 
+  /// True once shutdown started (reboot or safe shutdown); timeouts set from then on never run.
+  /// Stored as a free bit on app_state_ (bit 7) to avoid costing additional RAM.
+  bool is_shutting_down() const { return (this->app_state_ & APP_STATE_SHUTTING_DOWN) != 0; }
+
 // Helper macro for entity getter method declarations
 #ifdef USE_DEVICES
 #define GET_ENTITY_METHOD(entity_type, entity_name, entities_member) \

esphome/core/automation.h

@@ -595,6 +595,7 @@ template<typename... Ts> class Trigger {
 };
 
 template<typename... Ts> class ActionList;
+class ActionTimer;
 
 template<typename... Ts> class Action {
  public:
@@ -624,6 +625,8 @@ template<typename... Ts> class Action {
 
  protected:
   friend ActionList<Ts...>;
+  // Resumes a while that gave the main loop a turn
+  friend ActionTimer;
 
   /// For a branch owner (if, while), play() runs when the branch finishes.
   virtual void play(const Ts &...x) = 0;

esphome/core/base_automation.h

@@ -11,6 +11,7 @@
 
 #include <array>
 #include <list>
+#include <memory>
 #include <tuple>
 #include <type_traits>
 #include <vector>
@@ -145,6 +146,31 @@ template<typename... Ts> class ForCondition : public Condition<Ts...>, public Co
   uint32_t last_inactive_{0};
 };
 
+/// A while with no trigger arguments gives the main loop a turn after this long when its body never waits,
+/// so entity states can change. With arguments it never yields: they may only be valid while the trigger runs.
+inline constexpr uint32_t WHILE_YIELD_AFTER_MS = 20;
+
+/// Timeouts named by an action; the one scheduler friend, for delay and while.
+class ActionTimer {
+  template<typename... Ts> friend class DelayAction;
+  template<typename... Ts> friend class WhileAction;
+  friend class ApplyWhileActionBase;
+
+  /// See Scheduler::set_action_timeout_.
+  static void set(const void *self, uint32_t delay, std::function<void()> &&func, bool parallel) {
+    App.scheduler.set_action_timeout_(self, delay, std::move(func), parallel);
+  }
+  /// For a while without arguments: once its loop started at `start` ran too long, resumes it through play() on
+  /// the next main loop pass and returns true. Parallel, as a yielding run has no timeout of its own pending.
+  static bool yield_if_due(Action<> *self, uint32_t start) {
+    // No later pass runs the timeout once shutdown started, so a loop then runs to its end
+    if (App.is_shutting_down() || millis() - start <= WHILE_YIELD_AFTER_MS)
+      return false;
+    set(self, 0, [self]() { self->play(); }, /* parallel= */ true);
+    return true;
+  }
+};
+
 class StartupTrigger : public Trigger<>, public Component {
  public:
   explicit StartupTrigger(float setup_priority) : setup_priority_(setup_priority) {}
@@ -205,37 +231,17 @@ template<typename... Ts> class DelayAction : public Action<Ts...> {
 
     // If num_running_ > 1, we have multiple instances running in parallel
     // In single/restart/queued modes, only one instance runs at a time
-    // Parallel mode uses skip_cancel=true to allow multiple delays to coexist
+    // Parallel mode passes parallel=true to allow multiple delays to coexist
     // WARNING: This can accumulate delays if scripts are triggered faster than they complete!
     // Users should set max_runs on parallel scripts to limit concurrent executions.
     // Issue #10264: This is a workaround for parallel script delays interfering with each other.
 
-    // Optimization: For no-argument delays (most common case), use direct lambda
-    // to avoid overhead from capturing arguments by value
-    if constexpr (sizeof...(Ts) == 0) {
-      auto f = [this]() { this->play_next_(); };
-      App.scheduler.set_timer_common_(
-          /* component= */ nullptr, Scheduler::SchedulerItem::TIMEOUT, Scheduler::NameType::SELF_POINTER,
-          /* static_name= */ reinterpret_cast<const char *>(this), /* hash_or_id= */ 0, this->delay_.value(),
-          std::move(f),
-          /* skip_cancel= */ this->num_running_ > 1,
-          // Record the owning script (if any) so the blocking warning can name it; propagates across
-          // chained delays via the scheduler.
-          /* source= */ App.get_current_source());
-    } else {
-      // For delays with arguments, capture by value to preserve argument values
-      // Arguments must be copied because original references may be invalid after delay
-      // `mutable` is required so captured copies of non-const reference args (e.g. std::string&)
-      // are passed as non-const lvalues to play_next_(const Ts&...) where Ts may be `T&`
-      auto f = [this, x...]() mutable { this->play_next_(x...); };
-      App.scheduler.set_timer_common_(
-          /* component= */ nullptr, Scheduler::SchedulerItem::TIMEOUT, Scheduler::NameType::SELF_POINTER,
-          /* static_name= */ reinterpret_cast<const char *>(this), /* hash_or_id= */ 0, this->delay_.value(x...),
-          std::move(f),
-          /* skip_cancel= */ this->num_running_ > 1,
-          // See the no-argument branch above: record the owning script for log attribution.
-          /* source= */ App.get_current_source());
-    }
+    // Arguments must be copied because original references may be invalid after delay
+    // `mutable` is required so captured copies of non-const reference args (e.g. std::string&)
+    // are passed as non-const lvalues to play_next_(const Ts&...) where Ts may be `T&`
+    ActionTimer::set(
+        this, this->delay_.value(x...), [this, x...]() mutable { this->play_next_(x...); },
+        /* parallel= */ this->num_running_ > 1);
   }
 
   void play(const Ts &...x) override { /* ignore - see play_complex */ }
@@ -360,6 +366,38 @@ template<bool HasElse, typename... Ts> class IfAction : public Action<Ts...> {
   [[no_unique_address]] std::conditional_t<HasElse, ActionList<Ts...>, NoElse> else_;
 };
 
+/// Shared by every ApplyWhileAction, so each step function adds only its play() and vtable.
+class ApplyWhileActionBase : public Action<> {
+ public:
+  void play_complex() override {
+    this->num_running_++;
+    this->play();
+  }
+
+  void stop() override { App.scheduler.cancel_timeout(this); }
+
+ protected:
+  using StepFn = bool (*)(const int &running);
+
+  void run_(StepFn step) {
+    // Any other type would bind the step's const int & to a temporary that never sees a stop
+    static_assert(std::is_same_v<decltype(this->num_running_), int>);
+    const uint32_t start = millis();
+    while (step(this->num_running_)) {
+      if (ActionTimer::yield_if_due(this, start))
+        return;
+    }
+    this->play_next_();
+  }
+};
+
+/// A while with no trigger arguments whose condition and body codegen merged into one step function.
+/// Step runs the body once and returns true, or returns false once the condition is false or the run stopped.
+template<auto Step> class ApplyWhileAction final : public ApplyWhileActionBase {
+ public:
+  void play() override { this->run_(Step); }
+};
+
 template<typename... Ts> class WhileAction : public Action<Ts...> {
  public:
   WhileAction(Condition<Ts...> *condition) : condition_(condition) {}
@@ -375,40 +413,74 @@ template<typename... Ts> class WhileAction : public Action<Ts...> {
 
   void play_complex(const Ts &...x) final {
     this->num_running_++;
-    // Initial condition check
-    if (!this->condition_->check(x...)) {
-      // If new condition check failed, stop loop if running
-      this->then_.stop();
-      this->play_next_(x...);
-      return;
-    }
-
-    // The condition may have stopped this automation
-    if (this->num_running_ == 0)
-      return;
-    // An empty body cannot resume the loop
+    // An empty body cannot resume the loop; the condition still runs once, as it may stop the automation
     if (this->then_.empty()) {
+      this->condition_->check(x...);
       this->play_next_(x...);
       return;
     }
-    this->then_.play(x...);
+    this->run_loop_(x...);
   }
 
-  // Branch-end resume hook: loop again or continue
+  // Branch-end resume hook
   void play(const Ts &...x) override {
-    // The condition may stop this automation
-    if (this->num_running_ > 0 && this->condition_->check(x...) && this->num_running_ > 0) {
-      this->then_.play(x...);
-    } else {
-      this->play_next_(x...);
+    // A body that ended without waiting is looped by run_loop_(); mark it ended. Another run's end passes other
+    // argument storage, and a second end in one pass loops on its own.
+    Body *body = this->body_;
+    if (body != nullptr && !body->ended && body->args == Keys{std::addressof(x)...}) {
+      body->ended = true;
+      return;
     }
+    this->run_loop_(x...);
   }
 
-  void stop() override { this->then_.stop(); }
+  void stop() override {
+    this->then_.stop();
+    if constexpr (sizeof...(Ts) == 0)
+      App.scheduler.cancel_timeout(this);
+  }
 
  protected:
+  // The argument storage, which tells a run from another run of this action
+  using Keys = std::array<const void *, sizeof...(Ts)>;
+  // The body running synchronously, on the stack of run_loop_(); play() marks it ended. Per action, not
+  // shared, so loops of different actions never see each other's.
+  struct Body {
+    Keys args;
+    bool ended;
+  };
+
+  // Runs the body in a loop, not by recursion, while it ends without waiting
+  void run_loop_(const Ts &...x) {
+    [[maybe_unused]] uint32_t start{};
+    if constexpr (sizeof...(Ts) == 0)
+      start = millis();
+    const Keys keys{std::addressof(x)...};
+    while (true) {
+      // The condition may stop this automation
+      if (this->num_running_ == 0 || !this->condition_->check(x...) || this->num_running_ == 0) {
+        this->play_next_(x...);
+        return;
+      }
+      // On the stack, as the body can start this action again
+      Body body{keys, false};
+      Body *outer = this->body_;
+      this->body_ = &body;
+      this->then_.play(x...);
+      this->body_ = outer;
+      // A body that waits resumes through play() later
+      if (!body.ended)
+        return;
+      if constexpr (sizeof...(Ts) == 0) {
+        if (this->num_running_ > 0 && ActionTimer::yield_if_due(this, start))
+          return;
+      }
+    }
+  }
+
   Condition<Ts...> *condition_;
   ActionList<Ts...> then_;
+  Body *body_{nullptr};
 };
 
 // Forward declaration for RepeatLoopContinuation

esphome/core/component.h

@@ -98,6 +98,9 @@ inline constexpr uint8_t COMPONENT_HAS_LOOP = 0x20;
 // decide whether to propagate clears to App.app_state_. Never set on a
 // Component's component_state_.
 inline constexpr uint8_t APP_STATE_SETUP_COMPLETE = 0x40;
+// Bit 7 on Application::app_state_ (ONLY) — set once shutdown starts, as the
+// scheduler then no longer runs.
+inline constexpr uint8_t APP_STATE_SHUTTING_DOWN = 0x80;
 
 inline constexpr uint8_t WARN_IF_BLOCKING_OVER_CS = 5U;  // 50ms in centiseconds (1cs = 10ms)
 

esphome/core/scheduler.cpp

@@ -110,6 +110,11 @@ uint32_t Scheduler::calculate_interval_offset_(uint32_t delay) {
   return static_cast<uint32_t>((static_cast<uint64_t>(random_uint32()) * max_offset) >> 32);
 }
 
+void Scheduler::set_action_timeout_(const void *self, uint32_t delay, std::function<void()> &&func, bool parallel) {
+  this->set_timer_common_(nullptr, SchedulerItem::TIMEOUT, NameType::SELF_POINTER, static_cast<const char *>(self), 0,
+                          delay, std::move(func), parallel, App.get_current_source());
+}
+
 // Common implementation for both timeout and interval
 // name_type determines storage type: STATIC_STRING uses static_name, others use hash_or_id
 void HOT Scheduler::set_timer_common_(Component *component, SchedulerItem::Type type, NameType name_type,

esphome/core/scheduler.h

@@ -18,11 +18,11 @@ namespace esphome {
 class Component;
 
 class Scheduler {
-  // Allow DelayAction to call set_timer_common_ with skip_cancel=true for parallel script delays.
+  // Allow ActionTimer (delay and while) to call set_action_timeout_, which skips cancel for parallel runs.
   // This is needed to fix issue #10264 where parallel scripts with delays interfere with each other.
   // We use friend instead of a public API because skip_cancel is dangerous - it can cause delays
   // to accumulate and overload the scheduler if misused.
-  template<typename... Ts> friend class DelayAction;
+  friend class ActionTimer;
 
  public:
   /** Set a timeout with a const char* name.
@@ -266,6 +266,10 @@ class Scheduler {
     }
   };
 
+  // A timeout named by the action `self`; `parallel` keeps the timeouts of other runs of it.
+  // Records the owning script, if any, so the blocking warning can name it.
+  void set_action_timeout_(const void *self, uint32_t delay, std::function<void()> &&func, bool parallel);
+
   // Common implementation for both timeout and interval
   // name_type determines storage type: STATIC_STRING uses static_name, others use hash_or_id
   // `source` is stored (in the union slot) only for SELF_POINTER items; ignored otherwise.

tests/components/climate/common-triggers.yaml

@@ -19,6 +19,19 @@ climate:
       - logger.log:
           format: "Climate mode %d"
           args: [x.mode]
+      # A while with an argument that cannot be copied (Climate &): the inline and the WhileAction form
+      - while:
+          condition:
+            lambda: return false;
+          then:
+            - lambda: ESP_LOGD("test", "inline body");
+      - while:
+          condition:
+            lambda: return false;
+          then:
+            - wait_until:
+                condition:
+                  lambda: return true;
     on_control:
       - lambda: |-
           if (x.get_mode().has_value())

tests/components/core/test_action_branch_owner.cpp

@@ -1,4 +1,5 @@
 #include <gtest/gtest.h>
+#include <utility>
 #include <vector>
 
 #include "esphome/core/automation.h"
@@ -79,10 +80,10 @@ class StopOnCheckCondition : public Condition<> {
 };
 
 // Turns its condition off after a number of checks
-class CountdownCondition : public Condition<> {
+template<typename... Ts> class CountdownCondition : public Condition<Ts...> {
  public:
   explicit CountdownCondition(int passes) : passes_(passes) {}
-  bool check() override { return this->passes_-- > 0; }
+  bool check(const Ts &.../*x*/) override { return this->passes_-- > 0; }
 
  protected:
   int passes_;
@@ -177,7 +178,7 @@ TEST(ActionBranchOwner, PerItemIfAndWhileResumeAfterTheirBranch) {
   EXPECT_EQ(log, (Log{1, 2, 4, 3, 4}));
 
   log.clear();
-  CountdownCondition loop_cond(2);
+  CountdownCondition<> loop_cond(2);
   WhileAction<> loop(&loop_cond);
   DeferredAction wait(&log, 5);
   RecordAction after2(&log, 6);
@@ -341,7 +342,7 @@ TEST(ActionBranchOwner, EmptyBranchesFinishImmediately) {
 
 TEST(ActionBranchOwner, WhileRechecksEachPassAndExits) {
   Log log;
-  CountdownCondition cond(3);
+  CountdownCondition<> cond(3);
   WhileAction<> loop(&cond);
   RecordAction body(&log, 1);
   RecordAction after(&log, 2);
@@ -379,7 +380,7 @@ TEST(ActionBranchOwner, StopMidBranchClearsRunning) {
 
 TEST(ActionBranchOwner, NestedIfInsideWhileResumesEachOwner) {
   Log log;
-  CountdownCondition loop_cond(2);
+  CountdownCondition<> loop_cond(2);
   FlagCondition if_cond;
   WhileAction<> loop(&loop_cond);
   IfAction<false> inner(&if_cond);
@@ -400,4 +401,246 @@ TEST(ActionBranchOwner, NestedIfInsideWhileResumesEachOwner) {
   EXPECT_EQ(top.num_running(), 0);
 }
 
+// Counts the passes of a body that never waits
+class CountAction : public Action<int> {
+ public:
+  void play(const int &x) override { this->total += x; }
+  int total{0};
+};
+
+// Enough passes to overflow the stack if each one nested another call chain
+TEST(ActionBranchOwner, WhileBodyThatNeverWaitsLoopsInPlace) {
+  static constexpr int PASSES = 200000;
+  CountdownCondition<int> cond(PASSES);
+  WhileAction<int> loop(&cond);
+  CountAction body, after;
+  loop.add_then({&body});
+  ActionList<int> list;
+  list.add_actions({&loop, &after});
+
+  list.play(1);
+  // With an argument the loop never yields, so it ends before play() returns
+  EXPECT_EQ(body.total, PASSES);
+  EXPECT_EQ(after.total, 1);
+  EXPECT_FALSE(list.is_running());
+}
+
+// Keeps the first argument by reference, like wait_until does for an entity, and copies the second
+class MixedHandoffAction : public Action<const int &, int> {
+ public:
+  void play_complex(const int &entity, const int &value) override {
+    this->num_running_++;
+    this->finish();
+    this->entity_ = &entity;
+    this->value_ = value;
+    this->waiting_ = true;
+  }
+  void play(const int & /*entity*/, const int & /*value*/) override {}
+  void finish() {
+    if (this->waiting_) {
+      this->waiting_ = false;
+      int released = this->value_;
+      this->play_next_(*this->entity_, released);
+    }
+  }
+
+ protected:
+  const int *entity_{nullptr};
+  int value_{0};
+  bool waiting_{false};
+};
+
+class OncePerSecondValueCondition : public Condition<const int &, int> {
+ public:
+  bool check(const int & /*entity*/, const int &x) override { return !std::exchange(this->seen_[x], true); }
+
+ protected:
+  bool seen_[4]{};
+};
+
+class RecordSecondAction : public Action<const int &, int> {
+ public:
+  explicit RecordSecondAction(Log *log) : log_(log) {}
+  void play(const int & /*entity*/, const int &x) override { this->log_->push_back(x); }
+
+ protected:
+  Log *log_;
+};
+
+// The end of another run's body inside this run's body must resume that run, not end this one. Runs on the same
+// entity share its storage; the copied argument still tells them apart.
+TEST(ActionBranchOwner, WhileTellsRunsApartBySharedAndCopiedArguments) {
+  OncePerSecondValueCondition cond;
+  WhileAction<const int &, int> loop(&cond);
+  MixedHandoffAction wait;
+  Log log;
+  RecordSecondAction after(&log);
+  loop.add_then({&wait});
+  ActionList<const int &, int> list;
+  list.add_actions({&loop, &after});
+
+  const int entity = 0;
+  list.play(entity, 1);
+  list.play(entity, 2);
+  EXPECT_EQ(log, (Log{1}));
+  wait.finish();
+  EXPECT_EQ(log, (Log{1, 2}));
+  EXPECT_FALSE(list.is_running());
+}
+
+class CountPassAction : public Action<> {
+ public:
+  void play() override { this->passes++; }
+  int passes{0};
+};
+
+// Without arguments a wait_until timeout of one run can end inside another run's pass. Runs without arguments
+// are alike: whichever frame takes that end loops once for it, so every pass and every run still continues once.
+TEST(ActionBranchOwner, WhileWithoutArgumentsKeepsRunsWhenAWaitTimesOutInside) {
+  CountdownCondition<> cond(3);
+  WhileAction<> loop(&cond);
+  CountPassAction pass;
+  FlagCondition never;
+  never.value = false;
+  WaitUntilAction<> wait(&never);
+  wait.set_timeout_value([]() -> uint32_t { return 10; });
+  Log log;
+  RecordAction after(&log, 1);
+  loop.add_then({&pass, &wait});
+  ActionList<> list;
+  list.add_actions({&loop, &after});
+
+  list.play();  // run A waits
+  delay(15);
+  list.play();  // run B's pass releases A, whose wait timed out, inside it
+  delay(15);
+  wait.loop();  // the remaining waits time out
+  EXPECT_EQ(pass.passes, 3);
+  EXPECT_EQ(log, (Log{1, 1}));
+  EXPECT_FALSE(list.is_running());
+}
+
+// Releases the run that waits here, then lets the current run go on at once: two ends in one pass
+class ReleaseAndPassAction : public Action<> {
+ public:
+  void play_complex() override {
+    this->num_running_++;
+    if (this->waiting_) {
+      this->waiting_ = false;
+      this->play_next_();
+    }
+    if (this->pass) {
+      this->play_next_();
+      return;
+    }
+    this->waiting_ = true;
+  }
+  void play() override {}
+  bool pass{false};
+
+ protected:
+  bool waiting_{false};
+};
+
+// Without arguments both ends of one pass mark the same body; the second must still continue its run
+TEST(ActionBranchOwner, WhileWithoutArgumentsContinuesBothEndsOfOnePass) {
+  CountdownCondition<> cond(2);
+  WhileAction<> loop(&cond);
+  ReleaseAndPassAction wait;
+  Log log;
+  RecordAction after(&log, 1);
+  loop.add_then({&wait});
+  ActionList<> list;
+  list.add_actions({&loop, &after});
+
+  list.play();  // run A waits
+  wait.pass = true;
+  list.play();  // run B's pass releases A and ends itself
+  EXPECT_EQ(log, (Log{1, 1}));
+  EXPECT_FALSE(list.is_running());
+}
+
+static int step_passes = 0;                // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
+static ActionList<> *step_list = nullptr;  // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
+static int step_stop_on = 0;               // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
+
+// Like a generated step: false once the condition is false or the run stopped
+static bool count_step(const int &running) {
+  if (!running || step_passes >= 3)
+    return false;
+  if (++step_passes == step_stop_on)
+    step_list->stop();
+  return true;
+}
+
+// Runs an ApplyWhileAction of count_step that stops its run on pass `stop_on` (0 for never)
+static Log run_apply_while(int stop_on) {
+  step_passes = 0;
+  step_stop_on = stop_on;
+  Log log;
+  ApplyWhileAction<count_step> loop;
+  RecordAction after(&log, 1);
+  ActionList<> list;
+  list.add_actions({&loop, &after});
+  step_list = &list;
+  list.play();
+  step_list = nullptr;
+  EXPECT_FALSE(list.is_running());
+  return log;
+}
+
+TEST(ActionBranchOwner, ApplyWhileRunsStepsThenContinues) {
+  EXPECT_EQ(run_apply_while(0), (Log{1}));
+  EXPECT_EQ(step_passes, 3);
+}
+
+TEST(ActionBranchOwner, ApplyWhileStoppedInAStepEndsTheRun) {
+  EXPECT_TRUE(run_apply_while(2).empty());
+  EXPECT_EQ(step_passes, 2);
+}
+
+static int slow_passes = 0;  // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
+
+// Two passes that together take longer than the yield time
+static bool slow_step(const int &running) {
+  if (!running || slow_passes >= 2)
+    return false;
+  delay(WHILE_YIELD_AFTER_MS);
+  slow_passes++;
+  return true;
+}
+
+// Never constructed: names Application's protected state bits through a member pointer
+struct AppStateAccess : Application {
+  static uint8_t Application::*state() { return &AppStateAccess::app_state_; }
+};
+
+// Restores App's state bits, so a shutdown started by a test does not reach later tests
+class AppStateRestore {
+ public:
+  AppStateRestore() : saved_(App.*AppStateAccess::state()) {}
+  ~AppStateRestore() { App.*AppStateAccess::state() = this->saved_; }
+
+ protected:
+  uint8_t saved_;
+};
+
+// Nothing runs a timeout once shutdown started, so a loop then must not yield
+TEST(ActionBranchOwner, WhileDoesNotYieldOnceShutdownStarted) {
+  AppStateRestore restore;
+  App.run_safe_shutdown_hooks();
+  ASSERT_TRUE(App.is_shutting_down());
+  Log log;
+  ApplyWhileAction<slow_step> loop;
+  RecordAction after(&log, 1);
+  ActionList<> list;
+  list.add_actions({&loop, &after});
+
+  list.play();
+  EXPECT_EQ(slow_passes, 2);
+  EXPECT_EQ(log, (Log{1}));
+}
+
+TEST(ActionBranchOwner, ShutdownStateIsRestoredAfterTheTest) { EXPECT_FALSE(App.is_shutting_down()); }
+
 }  // namespace esphome::core::testing

tests/integration/fixtures/while_without_wait.yaml

@@ -0,0 +1,134 @@
+esphome:
+  name: test-while-without-wait
+
+host:
+
+logger:
+
+globals:
+  - id: count
+    type: int
+    initial_value: "0"
+
+binary_sensor:
+  - platform: template
+    id: held
+    name: Held
+
+script:
+  - id: release_later
+    then:
+      - delay: 300ms
+      - binary_sensor.template.publish:
+          id: held
+          state: false
+
+  # No arguments, inline body: ApplyWhileAction, which yields
+  - id: inline_noarg
+    then:
+      - while:
+          condition:
+            binary_sensor.is_on: held
+          then:
+            - lambda: id(count) += 1;
+      - logger.log: "INLINE_NOARG done"
+  # No arguments, a body with its own action object (wait_until ends at once): WhileAction<>, which yields
+  - id: object_noarg
+    then:
+      - while:
+          condition:
+            binary_sensor.is_on: held
+          then:
+            - lambda: id(count) += 1;
+            - wait_until:
+                condition:
+                  lambda: return true;
+      - logger.log: "OBJECT_NOARG done"
+
+  # Busy bodies longer than the yield time. Without arguments the caller runs first; with arguments
+  # the loop never yields and ends before the caller goes on.
+  - id: busy_inline_noarg
+    then:
+      - lambda: id(count) = 0;
+      - while: &busy_inline
+          condition:
+            lambda: return id(count) < 2;
+          then:
+            - lambda: |-
+                delay(25);
+                id(count) += 1;
+      - logger.log: "BUSY_INLINE_NOARG done"
+  - id: busy_object_noarg
+    then:
+      - lambda: id(count) = 0;
+      - while: &busy_object
+          condition:
+            lambda: return id(count) < 2;
+          then:
+            - lambda: |-
+                delay(25);
+                id(count) += 1;
+            - wait_until:
+                condition:
+                  lambda: return true;
+      - logger.log: "BUSY_OBJECT_NOARG done"
+  - id: busy_inline_str
+    parameters:
+      tag: string
+    then:
+      - lambda: id(count) = 0;
+      - while: *busy_inline
+      - logger.log:
+          format: "BUSY_INLINE_STR done %s"
+          args: [tag.c_str()]
+  - id: busy_object_str
+    parameters:
+      tag: string
+    then:
+      - lambda: id(count) = 0;
+      - while: *busy_object
+      - logger.log:
+          format: "BUSY_OBJECT_STR done %s"
+          args: [tag.c_str()]
+
+api:
+  actions:
+    - action: run_entity_loops
+      then:
+        - binary_sensor.template.publish:
+            id: held
+            state: true
+        - script.execute: release_later
+        - script.execute: inline_noarg
+        - script.execute: object_noarg
+    - action: run_busy_loops
+      then:
+        - script.execute: busy_inline_noarg
+        - logger.log: "BUSY_INLINE_NOARG caller"
+        - script.execute: busy_object_noarg
+        - logger.log: "BUSY_OBJECT_NOARG caller"
+        - delay: 200ms
+        - script.execute:
+            id: busy_inline_str
+            tag: a
+        - logger.log: "BUSY_INLINE_STR caller"
+        - script.execute:
+            id: busy_object_str
+            tag: b
+        - logger.log: "BUSY_OBJECT_STR caller"
+    - action: run_stop_while_yielded
+      then:
+        - binary_sensor.template.publish:
+            id: held
+            state: true
+        - script.execute: inline_noarg
+        - script.execute: object_noarg
+        - delay: 100ms
+        - script.stop: inline_noarg
+        - script.stop: object_noarg
+        - delay: 100ms
+        - lambda: |-
+            ESP_LOGI("test", "STOP running %d %d", id(inline_noarg).is_running(), id(object_noarg).is_running());
+        - binary_sensor.template.publish:
+            id: held
+            state: false

tests/integration/test_automation_merged_actions.py

@@ -74,8 +74,7 @@ async def test_automation_merged_actions(
         assert count("LOOP_STOP condition then") == 0
         assert count("LOOP_STOP condition after") == 0
 
-        # With an inline body, a nested run whose while condition is false does not end the
-        # outer run. WhileAction, still used for a deferring body, stops its body there.
+        # A nested run whose while condition is false does not end the outer run
         await client.execute_service(service["run_while_reentry"], {})
         await waiter.wait_for("WHILE_REENTER running", timeout=5.0)
         assert count("WHILE_REENTER after nested") == 1

tests/integration/test_while_without_wait.py

@@ -0,0 +1,56 @@
+"""A while whose body never waits ends without a crash or hang; it yields only without arguments."""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from .log_utils import LineWaiter
+from .state_utils import require_service
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_while_without_wait(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    waiter = LineWaiter()
+
+    def index(text: str) -> int:
+        return next(i for i, line in enumerate(waiter.lines) if text in line)
+
+    async with (
+        run_compiled(yaml_config, line_callback=waiter.callback),
+        api_client_connected() as client,
+    ):
+        _, services = await client.list_entities_services()
+
+        # The condition only turns false when the main loop runs
+        await client.execute_service(require_service(services, "run_entity_loops"), {})
+        await waiter.wait_count("INLINE_NOARG done", 1, 5.0, "inline loop did not end")
+        await waiter.wait_count("OBJECT_NOARG done", 1, 5.0, "object loop did not end")
+
+        # A busy loop yields after the first slow pass, unless it has arguments
+        await client.execute_service(require_service(services, "run_busy_loops"), {})
+        await waiter.wait_count(
+            "BUSY_OBJECT_STR caller", 1, 5.0, "busy loops did not end"
+        )
+        for name in ("BUSY_INLINE_NOARG", "BUSY_OBJECT_NOARG"):
+            assert index(f"{name} caller") < index(f"{name} done"), name
+        for name in ("BUSY_INLINE_STR", "BUSY_OBJECT_STR"):
+            assert index(f"{name} done") < index(f"{name} caller"), name
+
+        # A stop while the loop waits for its next turn ends it for good
+        done_before = waiter.count("NOARG done")
+        await client.execute_service(
+            require_service(services, "run_stop_while_yielded"), {}
+        )
+        await waiter.wait_count("STOP running 0 0", 1, 5.0, "stopped loops still run")
+        await asyncio.sleep(0.2)
+        assert waiter.count("NOARG done") == done_before
+
+        # Still answering after every loop
+        await client.device_info()

tests/unit_tests/test_automation.py

@@ -40,6 +40,7 @@ from esphome.automation import (
     WhileAction,
     XorCondition,
     _shadow_iteration,
+    _warn_while_without_wait,
     apply_action_builder,
     apply_statements_action,
     build_action,
@@ -106,7 +107,7 @@ from esphome.core import (
 )
 from esphome.cpp_generator import LambdaExpression, MockObj, RawExpression
 from esphome.schema_extractors import SCHEMA_EXTRACT
-from esphome.types import ConfigType
+from esphome.types import ConfigType, TemplateArgsType
 from esphome.util import Registry, RegistryEntry
 from tests.component_tests.helpers import get_define_value
 
@@ -385,6 +386,125 @@ def test_owning_args_swaps_every_string_ref(
     ]
 
 
+_BUTTON_ON = {"binary_sensor.is_on": ID("button")}
+
+
+@pytest.mark.parametrize(
+    ("condition", "body", "warns"),
+    [
+        (_BUTTON_ON, [{"logger.log": "hi"}], True),
+        (_BUTTON_ON, [{"delay": "1s"}], False),
+        (_BUTTON_ON, [{"if": {CONF_THEN: [{"delay": "1s"}]}}], False),
+        # A lambda condition that only reads state the body leaves alone still hangs the loop
+        ({"lambda": Lambda("return id(s).state;")}, [{"logger.log": "hi"}], True),
+        # A lambda action naming what the condition reads can change it, such as a counter
+        (
+            {"lambda": Lambda("return id(n) < 5;")},
+            [{"lambda": Lambda("id(n) += 1;")}],
+            False,
+        ),
+        # A lambda action that names nothing the condition reads cannot
+        (_BUTTON_ON, [{"lambda": Lambda("id(n) += 1;")}], True),
+        # A templated value only reads, even of the condition's own entity
+        (_BUTTON_ON, [{"logger.log": {"args": [Lambda("id(button).state")]}}], True),
+        # A lambda condition inside the body only reads, also in an and:, which holds a list
+        (
+            _BUTTON_ON,
+            [
+                {
+                    "if": {
+                        CONF_CONDITION: {"and": [{"lambda": Lambda("return true;")}]},
+                        CONF_THEN: [{"logger.log": "hi"}],
+                    }
+                }
+            ],
+            True,
+        ),
+        # And inside an if's any: or a choice's if:
+        (
+            _BUTTON_ON,
+            [
+                {
+                    "if": {
+                        "any": [{"lambda": Lambda("return true;")}],
+                        CONF_THEN: [{"logger.log": "hi"}],
+                    }
+                }
+            ],
+            True,
+        ),
+        (
+            _BUTTON_ON,
+            [
+                {
+                    "choose": [
+                        {
+                            "if": {"and": [{"lambda": Lambda("return true;")}]},
+                            CONF_THEN: [{"logger.log": "hi"}],
+                        }
+                    ]
+                }
+            ],
+            True,
+        ),
+        # Ids generated later are still None and match nothing
+        (
+            {"binary_sensor.is_on": {"id": ID(None)}},
+            [{"switch.turn_off": {"id": ID(None)}}],
+            True,
+        ),
+        # A lambda action nested in the body still counts
+        (
+            _BUTTON_ON,
+            [
+                {
+                    "if": {
+                        CONF_THEN: [
+                            {"lambda": Lambda("id(button).publish_state(false);")}
+                        ]
+                    }
+                }
+            ],
+            False,
+        ),
+        # The body acts on an entity the condition's lambda reads
+        (
+            {"lambda": Lambda("return id(s).state;")},
+            [{"switch.turn_off": {"id": ID("s")}}],
+            False,
+        ),
+        # The body can turn the condition's own entity off
+        (_BUTTON_ON, [{"binary_sensor.template.publish": {"id": ID("button")}}], False),
+        (_BUTTON_ON, [{"binary_sensor.template.publish": {"id": ID("other")}}], True),
+        (_BUTTON_ON, [], False),
+    ],
+)
+def test_warn_while_without_wait(
+    mock_registry: dict[str, RegistryEntry],
+    caplog: pytest.LogCaptureFixture,
+    condition: ConfigType,
+    body: list[ConfigType],
+    warns: bool,
+) -> None:
+    config = {CONF_CONDITION: condition, CONF_THEN: body}
+    assert _warn_while_without_wait(config) is config
+    assert ("never waits, so the loop holds up the device" in caplog.text) is warns
+    if warns:
+        # Names the loop by its condition
+        assert f"on '{next(iter(condition))}'" in caplog.text
+
+
+def test_warn_while_names_the_condition_inside_a_list(
+    mock_registry: dict[str, RegistryEntry],
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A list condition is wrapped in an `and:`; the warning names the condition in it."""
+    _warn_while_without_wait(
+        {CONF_CONDITION: {"and": [_BUTTON_ON]}, CONF_THEN: [{"logger.log": "hi"}]}
+    )
+    assert "on 'binary_sensor.is_on'" in caplog.text
+
+
 def test_validate_no_deferring_actions_accepts_synchronous(
     mock_registry: dict[str, RegistryEntry],
 ) -> None:
@@ -3083,22 +3203,24 @@ async def test_not_condition_is_an_object_when_its_if_keeps_one() -> None:
     assert "new(act_if) IfAction<false>(neg);" in main
 
 
+def _while_action(*body: ConfigType) -> ConfigType:
+    return {
+        "while": {CONF_CONDITION: _inline_condition(), CONF_THEN: list(body)},
+        CONF_TYPE_ID: ID(
+            "act_w", is_declaration=True, type=WhileAction, is_manual=False
+        ),
+    }
+
+
 @pytest.mark.asyncio
 @pytest.mark.usefixtures("apply_actions")
-async def test_inline_while_checks_condition_and_stops() -> None:
-    config = [
-        {
-            "while": {
-                CONF_CONDITION: _inline_condition(),
-                CONF_THEN: [_action("my.a", "act_a")],
-            },
-            CONF_TYPE_ID: ID(
-                "act_w", is_declaration=True, type=WhileAction, is_manual=False
-            ),
-        }
-    ]
-
-    result = await build_action_list(config, TEMPLATE_ARG, [])
+async def test_inline_while_with_arguments_is_a_cpp_loop() -> None:
+    """With arguments it never yields, so it stays one C++ loop in the merged function."""
+    result = await build_action_list(
+        [_while_action(_action("my.a", "act_a"))],
+        cg.TemplateArguments(cg.int_),
+        [(cg.int_, "x")],
+    )
 
     assert [str(action) for action in result] == ["act_w"]
     assert "WhileAction" not in CORE.cpp_main_section
@@ -3110,18 +3232,57 @@ async def test_inline_while_checks_condition_and_stops() -> None:
 
 @pytest.mark.asyncio
 @pytest.mark.usefixtures("apply_actions")
-async def test_inline_while_with_empty_body_keeps_its_object() -> None:
-    """An empty body exits a WhileAction at once, which a C++ loop would not do."""
-    config = [
-        {
-            "while": {CONF_CONDITION: _inline_condition(), CONF_THEN: []},
-            CONF_TYPE_ID: ID(
-                "act_w", is_declaration=True, type=WhileAction, is_manual=False
-            ),
-        }
-    ]
+async def test_inline_while_without_arguments_is_a_step_action() -> None:
+    """Without arguments it is an ApplyWhileAction, which can give the main loop a turn."""
+    body = _while_action(_action("my.a", "act_a"), _action("my.b", "act_b"))
+    result = await build_action_list([body], TEMPLATE_ARG, [])
 
-    await build_action_list(config, TEMPLATE_ARG, [])
+    assert [str(action) for action in result] == ["act_w"]
+    assert "new(act_w) ApplyWhileAction<esphome__act_w__fn>();" in CORE.cpp_main_section
+    text = CORE.cpp_global_section
+    assert "static bool esphome__act_w__fn(const int & esphome__running)" in text
+    assert "while (true)" not in text
+    # A stop between body actions ends the loop like a false condition
+    stop = text.index("if (!esphome__running)\n", text.index("::comp->do_a();"))
+    assert stop < text.index("return false;", stop) < text.index("::comp->do_b();")
+    assert "return;" not in text
+    # The last body action can stop the run too
+    assert text.index("::comp->do_b();") < text.index("return esphome__running != 0;")
+
+
+@pytest.mark.asyncio
+@pytest.mark.usefixtures("apply_actions")
+async def test_step_action_replaces_nested_stop_checks() -> None:
+    """A stop check inside a nested inline if must also return false from the bool step."""
+    body = _while_action(_if_action(_action("my.a", "act_a"), _action("my.b", "act_b")))
+
+    await build_action_list([body], TEMPLATE_ARG, [])
+
+    text = CORE.cpp_global_section
+    assert "ApplyWhileAction<esphome__act_w__fn>" in CORE.cpp_main_section
+    assert "return;" not in text
+    assert text.count("return false;") >= 3
+
+
+@pytest.mark.asyncio
+@pytest.mark.usefixtures("apply_actions")
+@pytest.mark.parametrize("args", [[], [(cg.int_, "x")]])
+async def test_while_with_an_object_body_is_a_while_action(
+    args: TemplateArgsType,
+) -> None:
+    """A body that keeps an object, here a delay, needs WhileAction with or without arguments."""
+    template_arg = cg.TemplateArguments(*[type_ for type_, _ in args])
+
+    await build_action_list([_while_action(_delay("act_d"))], template_arg, args)
+
+    assert f"new(act_w) WhileAction{template_arg}(" in CORE.cpp_main_section
+
+
+@pytest.mark.asyncio
+@pytest.mark.usefixtures("apply_actions")
+async def test_while_with_empty_body_keeps_its_object() -> None:
+    """An empty body exits a WhileAction at once, which a C++ loop or a step would not do."""
+    await build_action_list([_while_action()], TEMPLATE_ARG, [])
 
     assert "new(act_w) WhileAction<>(" in CORE.cpp_main_section