[script] Disable queued script polling while the queue is empty (#18121)

esphome/components/script/script.h

@@ -142,6 +142,9 @@ template<typename... Ts> class QueueingScript : public Script<Ts...>, public Com
       // Use std::make_unique to replace the unique_ptr
       this->var_queue_[write_pos] = std::make_unique<std::tuple<Ts...>>(x...);
       this->num_queued_++;
+      // Enable loop now that there is something to dequeue - don't call loop()
+      // synchronously! Let the event loop call it to avoid reentrancy issues
+      this->enable_loop();
       return;
     }
 
@@ -168,6 +171,15 @@ template<typename... Ts> class QueueingScript : public Script<Ts...>, public Com
       this->queue_front_ = (this->queue_front_ + 1) % queue_capacity;
       this->trigger_tuple_(*tuple_ptr, std::make_index_sequence<sizeof...(Ts)>{});
     }
+    if (this->num_queued_ == 0 && !this->is_idle()) {
+      // Queue is now empty - disable loop until the next execute() queues an
+      // instance. The inline is_idle() check skips the out-of-line call when
+      // the loop is already disabled (execute() calls loop() synchronously).
+      // This can run before this component's setup() (execute() from on_boot),
+      // which leaves the state machine in LOOP_DONE and skips call_setup();
+      // this class therefore must not rely on a setup() override.
+      this->disable_loop();
+    }
   }
 
   void set_max_runs(int max_runs) { max_runs_ = max_runs; }

tests/integration/fixtures/script_queued.yaml

@@ -1,5 +1,17 @@
 esphome:
   name: test-script-queued
+  on_boot:
+    # Default priority (600.0) runs before the script component is set up
+    # This tests that an instance queued during boot still gets dequeued
+    # once the main loop starts (the idle-loop disabling must not eat it)
+    then:
+      - logger.log: "=== BOOT: Executing queued script twice ==="
+      - script.execute:
+          id: boot_script
+          tag: 1
+      - script.execute:
+          id: boot_script
+          tag: 2
 
 host:
 api:
@@ -98,6 +110,15 @@ api:
         - script.execute: no_params_script
         - script.execute: no_params_script
 
+    # Test 6: Re-execute after stop() cleared the queue
+    # (the idle loop must re-enable on demand)
+    - action: test_after_stop
+      then:
+        - logger.log: "=== TEST 6: Re-execute after stop ==="
+        - script.execute:
+            id: stop_script
+            num: 9
+
 logger:
   level: DEBUG
 
@@ -168,3 +189,18 @@ script:
       - logger.log: "No params: START"
       - delay: 50ms
       - logger.log: "No params: END"
+
+  # Boot script: executed twice from on_boot before setup()
+  - id: boot_script
+    mode: queued
+    max_runs: 3
+    parameters:
+      tag: int
+    then:
+      - logger.log:
+          format: "Boot queued: START %d"
+          args: ['tag']
+      - delay: 50ms
+      - logger.log:
+          format: "Boot queued: END %d"
+          args: ['tag']

tests/integration/fixtures/script_queued_idle_loop.yaml

@@ -0,0 +1,25 @@
+esphome:
+  name: test-script-queued-idle
+
+host:
+api:
+  actions:
+    # Execute twice: the first runs immediately, the second gets queued,
+    # which must re-enable the loop; draining must disable it again
+    - action: run_twice
+      then:
+        - script.execute: idle_script
+        - script.execute: idle_script
+
+# VERY_VERBOSE exposes the component framework's "loop disabled" and
+# "loop enabled" messages that this test asserts on
+logger:
+  level: VERY_VERBOSE
+
+script:
+  - id: idle_script
+    mode: queued
+    then:
+      - logger.log: "idle_script: START"
+      - delay: 50ms
+      - logger.log: "idle_script: END"

tests/integration/test_script_queued.py

@@ -26,6 +26,7 @@ async def test_script_queued(
         "stop": {"processed": [], "stop_logged": False},
         "rejection": {"processed": [], "rejections": 0},
         "no_params": {"executions": 0},
+        "boot": {"ended": []},
     }
 
     # Patterns for Test 1: Queue depth
@@ -49,12 +50,21 @@ async def test_script_queued(
     # Patterns for Test 5: No params
     no_params_end = re.compile(r"No params: END")
 
+    # Patterns for boot script (executed twice from on_boot before setup)
+    boot_end = re.compile(r"Boot queued: END (\d+)")
+
+    # Patterns for Test 6: Re-execute after stop
+    after_stop_end = re.compile(r"Stop test: END (\d+)")
+
     # Test completion futures
+    boot_complete = loop.create_future()
     test1_complete = loop.create_future()
     test2_complete = loop.create_future()
     test3_complete = loop.create_future()
     test4_complete = loop.create_future()
     test5_complete = loop.create_future()
+    test5_again_complete = loop.create_future()
+    test6_complete = loop.create_future()
 
     def check_output(line: str) -> None:
         """Check log output for all test messages."""
@@ -122,11 +132,24 @@ async def test_script_queued(
         # Test 5: No params
         if no_params_end.search(line):
             test_results["no_params"]["executions"] += 1
-            if (
-                test_results["no_params"]["executions"] == 3
-                and not test5_complete.done()
-            ):
-                test5_complete.set_result(True)
+            executions = test_results["no_params"]["executions"]
+            for count, future in ((3, test5_complete), (6, test5_again_complete)):
+                if executions == count and not future.done():
+                    future.set_result(True)
+
+        # Boot script (queued from on_boot before setup)
+        if match := boot_end.search(line):
+            test_results["boot"]["ended"].append(int(match.group(1)))
+            if len(test_results["boot"]["ended"]) == 2 and not boot_complete.done():
+                boot_complete.set_result(True)
+
+        # Test 6: Re-execute after stop
+        if (
+            (match := after_stop_end.search(line))
+            and int(match.group(1)) == 9
+            and not test6_complete.done()
+        ):
+            test6_complete.set_result(True)
 
     async with (
         run_compiled(yaml_config, line_callback=check_output),
@@ -135,6 +158,13 @@ async def test_script_queued(
         # Get services
         _, services = await client.list_entities_services()
 
+        # Boot: both executions from on_boot must complete, including the one
+        # that was queued before QueueingScript::setup() ran
+        await asyncio.wait_for(boot_complete, timeout=2.0)
+        assert sorted(test_results["boot"]["ended"]) == [1, 2], (
+            f"Boot: Expected both on_boot executions to complete, got {sorted(test_results['boot']['ended'])}"
+        )
+
         # Test 1: Queue depth limit
         test_service = next((s for s in services if s.name == "test_queue_depth"), None)
         assert test_service is not None, "test_queue_depth service not found"
@@ -203,3 +233,20 @@ async def test_script_queued(
         assert test_results["no_params"]["executions"] == 3, (
             f"Test 5: Expected 3 executions, got {test_results['no_params']['executions']}"
         )
+
+        # Test 5 again: after the queue fully drained (loop disabled while
+        # idle), executing again must still work
+        test_service = next((s for s in services if s.name == "test_no_params"), None)
+        assert test_service is not None, "test_no_params service not found"
+        await client.execute_service(test_service, {})
+        await asyncio.wait_for(test5_again_complete, timeout=2.0)
+        assert test_results["no_params"]["executions"] == 6, (
+            f"Test 5 again: Expected 6 executions total, got {test_results['no_params']['executions']}"
+        )
+
+        # Test 6: a stopped script (queue cleared, loop disabled) must run
+        # again on the next execute; the future resolves only on "END 9"
+        test_service = next((s for s in services if s.name == "test_after_stop"), None)
+        assert test_service is not None, "test_after_stop service not found"
+        await client.execute_service(test_service, {})
+        await asyncio.wait_for(test6_complete, timeout=2.0)

tests/integration/test_script_queued_idle_loop.py

@@ -0,0 +1,85 @@
+"""Test that an idle queued script disables its loop and re-enables on demand."""
+
+from __future__ import annotations
+
+import asyncio
+import re
+
+import pytest
+
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_script_queued_idle_loop(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Assert the loop state transitions of a queued script via VV logs.
+
+    Expected sequence: the idle script disables its loop on the first
+    iteration after boot, re-enables it when an instance gets queued,
+    and disables it again once the queue drains.
+    """
+    loop = asyncio.get_running_loop()
+
+    loop_state = re.compile(r"\bscript loop (disabled|enabled)\b")
+    script_end = re.compile(r"idle_script: END")
+
+    transitions: list[str] = []
+    end_count = 0
+
+    boot_disabled = loop.create_future()
+    enabled_after_queue = loop.create_future()
+    disabled_after_drain = loop.create_future()
+    runs_complete = loop.create_future()
+
+    def check_output(line: str) -> None:
+        nonlocal end_count
+        if match := loop_state.search(line):
+            transitions.append(match.group(1))
+            if transitions == ["disabled"] and not boot_disabled.done():
+                boot_disabled.set_result(True)
+            elif (
+                transitions == ["disabled", "enabled"]
+                and not enabled_after_queue.done()
+            ):
+                enabled_after_queue.set_result(True)
+            elif (
+                transitions
+                == [
+                    "disabled",
+                    "enabled",
+                    "disabled",
+                ]
+                and not disabled_after_drain.done()
+            ):
+                disabled_after_drain.set_result(True)
+
+        if script_end.search(line):
+            end_count += 1
+            if end_count == 2 and not runs_complete.done():
+                runs_complete.set_result(True)
+
+    async with (
+        run_compiled(yaml_config, line_callback=check_output),
+        api_client_connected() as client,
+    ):
+        # The idle script must disable its loop on the first iteration
+        await asyncio.wait_for(boot_disabled, timeout=5.0)
+
+        _, services = await client.list_entities_services()
+        run_twice = next((s for s in services if s.name == "run_twice"), None)
+        assert run_twice is not None, "run_twice service not found"
+        await client.execute_service(run_twice, {})
+
+        # Queueing the second instance must re-enable the loop
+        await asyncio.wait_for(enabled_after_queue, timeout=2.0)
+        # Both runs must complete and the drained queue must disable it again
+        await asyncio.wait_for(runs_complete, timeout=2.0)
+        await asyncio.wait_for(disabled_after_drain, timeout=2.0)
+
+        assert transitions == ["disabled", "enabled", "disabled"], (
+            f"Unexpected loop state sequence: {transitions}"
+        )