[core] Fix wait_until crash when re-entered from its own continuation (#17571)

esphome/core/base_automation.h

@@ -502,6 +502,9 @@ template<typename... Ts> class WaitUntilAction : public Action<Ts...>, public Co
 
   void stop() override {
     this->var_queue_.clear();
+    // Tell any process_queue_() call further down the stack that the items it is
+    // still holding were cancelled
+    this->stop_generation_++;
     this->disable_loop();
   }
 
@@ -511,33 +514,57 @@ template<typename... Ts> class WaitUntilAction : public Action<Ts...>, public Co
   }
 
  protected:
+  using QueueItem = std::tuple<uint32_t, optional<uint32_t>, std::tuple<Ts...>>;
+
   // Helper: Process queue, triggering completed items and removing them
   // Returns true if queue still has pending items
   bool process_queue_(uint32_t now) {
-    // Process each queued wait_until and remove completed ones
-    this->var_queue_.remove_if([&](auto &queued) {
-      auto start = std::get<uint32_t>(queued);
-      auto timeout = std::get<optional<uint32_t>>(queued);
-      auto &var = std::get<std::tuple<Ts...>>(queued);
+    // Completed items run the rest of the action chain synchronously, and that chain
+    // can re-enter this same action (e.g. a script with mode: restart that executes
+    // itself) and add to or clear var_queue_. Iterating the member list directly would
+    // then corrupt it, so move it aside and iterate a local list instead.
+    std::list<QueueItem> queue;
+    queue.swap(this->var_queue_);
+    std::list<QueueItem> pending;
+    while (!queue.empty()) {
+      auto it = queue.begin();
+      auto start = std::get<uint32_t>(*it);
+      auto timeout = std::get<optional<uint32_t>>(*it);
 
       // Check if timeout has expired
       auto expired = timeout && (now - start) >= *timeout;
 
       // Keep waiting if not expired and condition not met
-      if (!expired && !this->condition_->check_tuple(var)) {
-        return false;
+      if (!expired && !this->condition_->check_tuple(std::get<std::tuple<Ts...>>(*it))) {
+        pending.splice(pending.end(), queue, it);
+        continue;
       }
 
-      // Condition met or timed out - trigger next action
-      this->play_next_tuple_(var);
-      return true;
-    });
+      // Condition met or timed out - trigger the next action. Keep the item in a local
+      // holder so its arguments stay valid while the chain runs, without any nested
+      // process_queue_() call being able to see (and fire) it again.
+      std::list<QueueItem> completed;
+      completed.splice(completed.begin(), queue, it);
+      uint8_t generation = this->stop_generation_;
+      this->play_next_tuple_(std::get<std::tuple<Ts...>>(completed.front()));
+      if (generation != this->stop_generation_) {
+        // stop() ran inside the chain - the items still held locally were cancelled
+        pending.clear();
+        break;
+      }
+    }
+
+    // Re-entrant continuations may have enqueued new waits into var_queue_; put the
+    // older still-waiting items back in front of them to keep FIFO firing order
+    this->var_queue_.splice(this->var_queue_.begin(), pending);
 
     return !this->var_queue_.empty();
   }
 
   Condition<Ts...> *condition_;
-  std::list<std::tuple<uint32_t, optional<uint32_t>, std::tuple<Ts...>>> var_queue_{};
+  std::list<QueueItem> var_queue_{};
+  // Bumped by stop() so process_queue_() can detect a stop from inside play_next_tuple_()
+  uint8_t stop_generation_{0};
 };
 
 template<typename... Ts> class UpdateComponentAction : public Action<Ts...> {

tests/integration/fixtures/wait_until_reentrant_restart.yaml

@@ -0,0 +1,86 @@
+esphome:
+  name: wait-until-reentrant-restart
+
+host:
+
+api:
+  actions:
+    - action: start_self_restart
+      then:
+        - script.execute: retry_script
+    - action: start_stop_during_wait
+      then:
+        - globals.set:
+            id: gate_open
+            value: 'false'
+        # num 0 is a blocker: its condition never becomes true, so it is still
+        # waiting (already checked and set aside) when num 1 stops the script -
+        # it must be cancelled, not restored, so its timeout must never fire
+        - script.execute:
+            id: waiter
+            num: 0
+        - script.execute:
+            id: waiter
+            num: 1
+        - script.execute:
+            id: waiter
+            num: 2
+        - script.execute:
+            id: waiter
+            num: 3
+        # Give all three instances time to queue in the same wait_until
+        - delay: 100ms
+        - globals.set:
+            id: gate_open
+            value: 'true'
+        - delay: 200ms
+        - logger.log: "stop test complete"
+
+logger:
+  level: DEBUG
+
+globals:
+  - id: attempt
+    type: int
+    initial_value: '0'
+  - id: gate_open
+    type: bool
+    initial_value: 'false'
+
+script:
+  # Self-restart retry pattern: when the wait_until times out, the rest of the
+  # script runs synchronously from inside the wait queue processing and restarts
+  # this same script - re-entering the same WaitUntilAction while it is still
+  # processing its queue. This used to corrupt the queue and crash.
+  - id: retry_script
+    mode: restart
+    then:
+      - wait_until:
+          condition:
+            lambda: 'return false;'
+          timeout: 20ms
+      - lambda: |-
+          id(attempt) += 1;
+          ESP_LOGD("test", "attempt %d done", id(attempt));
+      - if:
+          condition:
+            lambda: 'return id(attempt) < 5;'
+          then:
+            - script.execute: retry_script
+          else:
+            - logger.log: "retry test complete"
+
+  # Parallel waiters all queued in the same wait_until; the first one to pass the
+  # gate stops the script from its continuation, cancelling the other waiters
+  # while the queue is still being processed.
+  - id: waiter
+    mode: parallel
+    parameters:
+      num: int
+    then:
+      - wait_until:
+          condition:
+            lambda: 'return num != 0 && id(gate_open);'
+          timeout: 1s
+      - lambda: 'ESP_LOGD("test", "gate passed %d", num);'
+      - script.stop: waiter

tests/integration/test_wait_until_reentrant_restart.py

@@ -0,0 +1,89 @@
+"""Integration test for wait_until queue reentrancy.
+
+When a wait_until completes, the rest of the action chain runs synchronously
+from inside the wait queue processing. That chain can re-enter the very same
+WaitUntilAction - for example a script with mode: restart that executes itself
+as a retry pattern, or a waiter that stops its own script. Both used to mutate
+the std::list while it was being iterated, corrupting it and crashing the
+device (Guru Meditation StoreProhibited in _M_transfer).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import re
+
+import pytest
+
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_wait_until_reentrant_restart(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test that re-entering a wait_until from its own continuation is safe."""
+    retry_complete = asyncio.Event()
+    stop_complete = asyncio.Event()
+
+    attempt_pattern = re.compile(r"attempt (\d+) done")
+    gate_pattern = re.compile(r"gate passed (\d+)")
+
+    attempts: list[int] = []
+    gate_passed: list[int] = []
+
+    def check_output(line: str) -> None:
+        """Check log output for expected messages."""
+        if mo := attempt_pattern.search(line):
+            attempts.append(int(mo.group(1)))
+        elif mo := gate_pattern.search(line):
+            gate_passed.append(int(mo.group(1)))
+        elif "retry test complete" in line:
+            retry_complete.set()
+        elif "stop test complete" in line:
+            stop_complete.set()
+
+    async with (
+        run_compiled(yaml_config, line_callback=check_output),
+        api_client_connected() as client,
+    ):
+        device_info = await client.device_info()
+        assert device_info is not None
+        assert device_info.name == "wait-until-reentrant-restart"
+
+        _, services = await client.list_entities_services()
+        self_restart_service = next(
+            (s for s in services if s.name == "start_self_restart"), None
+        )
+        assert self_restart_service is not None, "start_self_restart not found"
+        stop_service = next(
+            (s for s in services if s.name == "start_stop_during_wait"), None
+        )
+        assert stop_service is not None, "start_stop_during_wait not found"
+
+        # Scenario 1: the wait_until timeout continuation restarts its own
+        # script five times, re-entering the same wait_until each time.
+        await client.execute_service(self_restart_service, {})
+        try:
+            await asyncio.wait_for(retry_complete.wait(), timeout=10.0)
+        except TimeoutError:
+            pytest.fail(f"Self-restart retry did not finish. Attempts: {attempts}")
+        assert attempts == [1, 2, 3, 4, 5], attempts
+
+        # Scenario 2: the first waiter through the gate stops the script while
+        # the other waiters are still queued in the same wait_until; both the
+        # not-yet-checked waiters (2, 3) and the already-checked still-waiting
+        # blocker (0) must be cancelled, not fired.
+        await client.execute_service(stop_service, {})
+        try:
+            await asyncio.wait_for(stop_complete.wait(), timeout=10.0)
+        except TimeoutError:
+            pytest.fail(f"Stop-during-wait did not finish. Gate passed: {gate_passed}")
+        assert gate_passed == [1], gate_passed
+
+        # If the cancelled blocker had been kept, its 1s wait_until timeout
+        # would still fire - give it the chance and check it stays silent.
+        await asyncio.sleep(1.5)
+        assert gate_passed == [1], gate_passed