Simplify the while check and name its condition in one helper

The condition's name and place in the file come from one helper, also
for a list condition, without the private config location formatter.
The fix text is shared by the error and the warning, and the check
stops at the first action that waits.

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
@@ -18,8 +18,10 @@ from esphome.const import (
     CONF_CONDITION,
     CONF_COUNT,
     CONF_ELSE,
+    CONF_FOR,
     CONF_ID,
     CONF_IF,
+    CONF_LAMBDA,
     CONF_THEN,
     CONF_TIME,
     CONF_TIMEOUT,
@@ -1460,14 +1462,112 @@ async def choose_action_to_code(
     return var
 
 
+# Body actions that run code, which may change anything the condition reads
+_WHILE_CODE_ACTIONS = frozenset((CONF_LAMBDA, "script.execute", "component.update"))
+
+
+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 _condition_has_code(value: Any) -> bool:
+    """Whether a condition runs a lambda or depends on time (``for:``), so it can change on its own."""
+    if isinstance(value, Lambda):
+        return True
+    if isinstance(value, dict):
+        return CONF_FOR in value or any(_condition_has_code(v) for v in value.values())
+    if isinstance(value, list):
+        return any(_condition_has_code(v) for v in value)
+    return False
+
+
+# The fix for a while whose body never waits, for the error and the warning
+_WHILE_FIX = "Add '- delay: 100ms' as the last action under this loop's 'then:'."
+
+
+def _condition_name(condition: ConfigType) -> str:
+    """The condition's name, with its place in the file when known.
+
+    A list condition is wrapped in an `and:` that has no place in the file, so its first entry names it.
+    """
+    key = _registry_key(condition)
+    if (
+        not hasattr(key, "esp_range")
+        and isinstance(items := condition[key], list)
+        and items
+    ):
+        key = _registry_key(items[0])
+    if (location := getattr(key, "esp_range", None)) is None:
+        return f"'{key}'"
+    mark = location.start_mark
+    return f"'{key}' ({mark.document}:{mark.line + 1}:{mark.column + 1})"
+
+
+def _validate_while_waits(config: ConfigType) -> ConfigType:
+    """Reject or warn about a while whose body never waits and cannot change its condition.
+
+    Such a loop runs its body again from inside the previous pass and never lets other components run, so
+    a condition only they can change stays true until the device runs out of stack or the watchdog resets it.
+    """
+    body = config[CONF_THEN]
+    # An empty body ends the loop at once
+    if not body or has_non_synchronous_actions(body):
+        return config
+    actions = list(_iter_actions(body))
+    condition = config[CONF_CONDITION]
+    # An action on an id the condition (or its lambda) reads, or a lambda action naming one, may change it
+    if (read := _referenced_ids(condition)) and read & _action_ids(actions):
+        return config
+    if not _condition_has_code(condition) and all(
+        action not in _WHILE_CODE_ACTIONS for action, _ in actions
+    ):
+        raise cv.Invalid(
+            f"The body of this 'while' loop on {_condition_name(condition)} never waits and "
+            "cannot change the condition, so the condition never becomes false and the device "
+            f"would crash. {_WHILE_FIX}",
+            [CONF_THEN],
+        )
+    _LOGGER.warning(
+        "The body of the 'while' loop on %s never waits, so nothing else runs until the body "
+        "changes the condition; if it does not, the device crashes. %s",
+        _condition_name(condition),
+        _WHILE_FIX,
+    )
+    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,
+            }
+        ),
+        _validate_while_waits,
     ),
     synchronous=True,
 )
@@ -1694,10 +1794,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(
@@ -1830,24 +1934,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:

tests/unit_tests/test_automation.py

@@ -40,6 +40,7 @@ from esphome.automation import (
     WhileAction,
     XorCondition,
     _shadow_iteration,
+    _validate_while_waits,
     apply_action_builder,
     apply_statements_action,
     build_action,
@@ -385,6 +386,112 @@ def test_owning_args_swaps_every_string_ref(
     ]
 
 
+_BUTTON_ON = {"binary_sensor.is_on": ID("button")}
+_LOG = [{"logger.log": "hi"}]
+
+
+def _if(condition: ConfigType) -> list[ConfigType]:
+    return [{"if": {**condition, CONF_THEN: _LOG}}]
+
+
+@pytest.mark.parametrize(
+    ("condition", "body", "outcome"),
+    [
+        # Only the main loop could change the condition, and the body never lets it run
+        (_BUTTON_ON, _LOG, "error"),
+        (
+            _BUTTON_ON,
+            [{"binary_sensor.template.publish": {"id": ID("other")}}],
+            "error",
+        ),
+        (_BUTTON_ON, [{"logger.log": {"args": [Lambda("id(button).state")]}}], "error"),
+        # Conditions inside the body only read
+        (_BUTTON_ON, _if({"any": [{"lambda": Lambda("return true;")}]}), "error"),
+        (
+            _BUTTON_ON,
+            [
+                {
+                    "choose": [
+                        {
+                            "if": {"and": [{"lambda": Lambda("return true;")}]},
+                            CONF_THEN: _LOG,
+                        }
+                    ]
+                }
+            ],
+            "error",
+        ),
+        # Ids generated later are still None and match nothing
+        (
+            {"binary_sensor.is_on": {"id": ID(None)}},
+            [{"switch.turn_off": {"id": ID(None)}}],
+            "error",
+        ),
+        # A body that waits, or an empty one, is fine
+        (_BUTTON_ON, [{"delay": "1s"}], None),
+        (_BUTTON_ON, [{"if": {CONF_THEN: [{"delay": "1s"}]}}], None),
+        (_BUTTON_ON, [], None),
+        # The body acts on what the condition reads
+        (_BUTTON_ON, [{"binary_sensor.template.publish": {"id": ID("button")}}], None),
+        (
+            {"lambda": Lambda("return id(s).state;")},
+            [{"switch.turn_off": {"id": ID("s")}}],
+            None,
+        ),
+        (
+            {"lambda": Lambda("return id(n) < 5;")},
+            [{"lambda": Lambda("id(n) += 1;")}],
+            None,
+        ),
+        (
+            _BUTTON_ON,
+            [
+                {
+                    "if": {
+                        CONF_THEN: [
+                            {"lambda": Lambda("id(button).publish_state(false);")}
+                        ]
+                    }
+                }
+            ],
+            None,
+        ),
+        # Code in the condition or the body might still end the loop
+        ({"lambda": Lambda("return id(s).state;")}, _LOG, "warn"),
+        ({"and": [{"lambda": Lambda("return millis() < 5;")}]}, _LOG, "warn"),
+        ({"binary_sensor.is_on": {"id": ID("button"), "for": "1s"}}, _LOG, "warn"),
+        (_BUTTON_ON, [{"lambda": Lambda("id(n) += 1;")}], "warn"),
+        (_BUTTON_ON, [{"script.execute": ID("release")}], "warn"),
+        (_BUTTON_ON, [{"component.update": ID("other")}], "warn"),
+    ],
+)
+def test_validate_while_waits(
+    mock_registry: dict[str, RegistryEntry],
+    caplog: pytest.LogCaptureFixture,
+    condition: ConfigType,
+    body: list[ConfigType],
+    outcome: str | None,
+) -> None:
+    config = {CONF_CONDITION: condition, CONF_THEN: body}
+    if outcome == "error":
+        with pytest.raises(cv.Invalid, match="never waits.*the device would crash"):
+            _validate_while_waits(config)
+        return
+    assert _validate_while_waits(config) is config
+    assert ("never waits" in caplog.text) is (outcome == "warn")
+
+
+def test_validate_while_waits_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."""
+    _validate_while_waits(
+        {CONF_CONDITION: {"and": [{"lambda": Lambda("return true;")}]}, CONF_THEN: _LOG}
+    )
+    assert "on 'lambda'" in caplog.text
+
+
 def test_validate_no_deferring_actions_accepts_synchronous(
     mock_registry: dict[str, RegistryEntry],
 ) -> None: