[core] Script improvements including choose (#20547)

esphome/automation.py

@@ -14,6 +14,7 @@ from esphome.const import (
     CONF_COUNT,
     CONF_ELSE,
     CONF_ID,
+    CONF_IF,
     CONF_THEN,
     CONF_TIME,
     CONF_TIMEOUT,
@@ -34,6 +35,9 @@ from esphome.schema_extractors import SCHEMA_EXTRACT, schema_extractor
 from esphome.types import ConfigType, SafeExpType
 from esphome.util import Registry
 
+CONF_CHOOSE = "choose"
+CONF_CHOICE_ID = "choice_id"
+
 
 def progmem_bytes(name: str, data: bytes | list[int]) -> MockObj:
     """Shared PROGMEM table for constant bytes; equal payloads share one, empty is nullptr."""
@@ -840,10 +844,39 @@ async def delay_action_to_code(
     return var
 
 
+def _wrap_bare_if_condition(value: Any) -> Any:
+    """Allow a single condition directly under `if:` in place of `condition: {...}`."""
+    if not isinstance(value, dict):
+        return value
+    # `any` and `all` are also registered conditions, but here they are the if action's own keys
+    bare = [
+        key
+        for key in value
+        if key in CONDITION_REGISTRY and key not in (CONF_ANY, CONF_ALL)
+    ]
+    if not bare:
+        return value
+    if len(bare) > 1:
+        raise cv.Invalid(
+            f"Only one condition may be given directly under 'if', found {', '.join(bare)}. "
+            f"Use '{CONF_CONDITION}', '{CONF_ALL}' or '{CONF_ANY}' to combine conditions"
+        )
+    if other := [k for k in (CONF_CONDITION, CONF_ANY, CONF_ALL) if k in value]:
+        raise cv.Invalid(
+            f"'{bare[0]}' given directly under 'if' cannot be used with '{other[0]}'. "
+            f"Move it into '{other[0]}'"
+        )
+    key = bare[0]
+    result = {k: v for k, v in value.items() if k != key}
+    result[CONF_CONDITION] = {key: value[key]}
+    return result
+
+
 @register_action(
     "if",
     IfAction,
     cv.All(
+        _wrap_bare_if_condition,
         {
             cv.Exclusive(
                 CONF_CONDITION, CONF_CONDITION
@@ -870,14 +903,104 @@ async def if_action_to_code(
     cond_conf = next(el for el in config if el in (CONF_ANY, CONF_ALL, CONF_CONDITION))
     condition = await build_condition(config[cond_conf], template_arg, args)
     var = cg.new_Pvariable(action_id, if_template_arg, condition)
+    then_actions = else_actions = None
     if CONF_THEN in config:
-        for action in await build_action_list(config[CONF_THEN], template_arg, args):
+        then_actions = await build_action_list(config[CONF_THEN], template_arg, args)
+    if has_else:
+        else_actions = await build_action_list(config[CONF_ELSE], template_arg, args)
+    _add_if_branches(var, then_actions, else_actions)
+    return var
+
+
+def _add_if_branches(
+    var: MockObj,
+    then_actions: list[MockObj] | None,
+    else_actions: list[MockObj] | None,
+) -> None:
+    """Add the branches of an IfAction one action at a time; None leaves a branch out."""
+    if then_actions is not None:
+        for action in then_actions:
             cg.add(var.add_then_action(action))
         cg.add(var.finish_then())
-    if has_else:
-        for action in await build_action_list(config[CONF_ELSE], template_arg, args):
+    if else_actions is not None:
+        for action in else_actions:
             cg.add(var.add_else_action(action))
         cg.add(var.finish_else())
+
+
+def _validate_choice(choice: ConfigType) -> ConfigType:
+    """A choice is either `if` with `then`, or `else` on its own."""
+    if CONF_ELSE in choice:
+        for key in (CONF_IF, CONF_THEN):
+            if key in choice:
+                raise cv.Invalid(f"'{key}' cannot be used with 'else'", path=[key])
+        return choice
+    for key in (CONF_IF, CONF_THEN):
+        if key not in choice:
+            raise cv.Invalid(f"'{key}' is required", path=[key])
+    return choice
+
+
+def _validate_choose_else(choices: list[ConfigType]) -> list[ConfigType]:
+    """`else` may only be the last choice, and not the only one."""
+    for index, choice in enumerate(choices[:-1]):
+        if CONF_ELSE in choice:
+            raise cv.Invalid(
+                "Only the last choice may be 'else'", path=[index, CONF_ELSE]
+            )
+    if len(choices) == 1 and CONF_ELSE in choices[0]:
+        raise cv.Invalid(
+            "At least one choice must have 'if' and 'then'", path=[0, CONF_ELSE]
+        )
+    return choices
+
+
+@register_action(
+    CONF_CHOOSE,
+    IfAction,
+    cv.All(
+        cv.ensure_list(
+            cv.All(
+                cv.Schema(
+                    {
+                        cv.GenerateID(CONF_CHOICE_ID): cv.declare_id(IfAction),
+                        cv.Optional(CONF_IF): validate_potentially_and_condition,
+                        cv.Optional(CONF_THEN): validate_action_list,
+                        cv.Optional(CONF_ELSE): validate_action_list,
+                    }
+                ),
+                _validate_choice,
+            )
+        ),
+        cv.Length(min=1),
+        _validate_choose_else,
+    ),
+    synchronous=True,
+)
+async def choose_action_to_code(
+    config: list[ConfigType],
+    action_id: ID,
+    template_arg: cg.TemplateArguments,
+    args: TemplateArgsType,
+) -> MockObj:
+    # Built as a chain of if/else actions: each choice's else branch holds the next choice,
+    # and the last choice's else branch holds the `else` actions, if any.
+    else_actions: list[MockObj] | None = None
+    if CONF_ELSE in config[-1]:
+        else_actions = await build_action_list(
+            config[-1][CONF_ELSE], template_arg, args
+        )
+        config = config[:-1]
+    var: MockObj | None = None
+    for index in reversed(range(len(config))):
+        choice = config[index]
+        if_template_arg = cg.TemplateArguments(else_actions is not None, *template_arg)
+        condition = await build_condition(choice[CONF_IF], template_arg, args)
+        choice_id = action_id if index == 0 else choice[CONF_CHOICE_ID]
+        var = cg.new_Pvariable(choice_id, if_template_arg, condition)
+        then_actions = await build_action_list(choice[CONF_THEN], template_arg, args)
+        _add_if_branches(var, then_actions, else_actions)
+        else_actions = [var]
     return var
 
 

esphome/components/script/__init__.py

@@ -1,9 +1,13 @@
+from typing import Any
+
 from esphome import automation
 from esphome.automation import maybe_simple_id
 import esphome.codegen as cg
+from esphome.components.const import CONF_ON_STOP
 import esphome.config_validation as cv
 from esphome.const import CONF_ID, CONF_MODE, CONF_PARAMETERS, CONF_RESTART
 from esphome.core import CORE, EsphomeError
+from esphome.types import ConfigType
 
 CODEOWNERS = ["@esphome/core"]
 DOMAIN = "script"
@@ -20,6 +24,7 @@ CONF_SINGLE = "single"
 CONF_QUEUED = "queued"
 CONF_PARALLEL = "parallel"
 CONF_MAX_RUNS = "max_runs"
+CONF_SCRIPT_STOP = "script.stop"
 
 SCRIPT_MODES = {
     CONF_SINGLE: SingleScript,
@@ -109,22 +114,72 @@ def validate_parameter_type(value):
     raise cv.Invalid("Parameter type contains invalid characters")
 
 
-CONFIG_SCHEMA = automation.validate_automation(
-    {
-        # Don't declare id as cv.declare_id yet, because the ID type
-        # depends on the mode. Will be checked later with assign_declare_id
-        cv.Required(CONF_ID): cv.string_strict,
-        cv.Optional(CONF_MODE, default=CONF_SINGLE): cv.one_of(
-            *SCRIPT_MODES, lower=True
-        ),
-        cv.Optional(CONF_MAX_RUNS): cv.int_range(min=0, max=100),
-        cv.Optional(CONF_PARAMETERS, default={}): cv.Schema(
-            {
-                validate_parameter_name: validate_parameter_type,
-            }
-        ),
-    },
-    extra_validators=cv.All(check_max_runs, assign_declare_id),
+def _fill_script_stop_ids(value: Any, script_id: str) -> Any:
+    """Give `script.stop` actions without an id in a raw action tree the id `script_id`.
+
+    Containers are copied only along changed paths so untouched YAML nodes keep their source locations.
+    """
+    if isinstance(value, list):
+        result = [
+            {CONF_SCRIPT_STOP: {CONF_ID: script_id}}
+            if item == CONF_SCRIPT_STOP
+            else _fill_script_stop_ids(item, script_id)
+            for item in value
+        ]
+        if all(new is old for new, old in zip(result, value, strict=True)):
+            return value
+        return result
+    if not isinstance(value, dict):
+        return value
+    result = value
+    for key, item in value.items():
+        if key == CONF_SCRIPT_STOP and (
+            item is None or (isinstance(item, dict) and CONF_ID not in item)
+        ):
+            new_item = {**(item or {}), CONF_ID: script_id}
+        else:
+            new_item = _fill_script_stop_ids(item, script_id)
+        if new_item is not item:
+            if result is value:
+                result = value.copy()
+            result[key] = new_item
+    return result
+
+
+def _fill_script_stop_ids_in_script(item: Any) -> Any:
+    if isinstance(item, dict) and isinstance(script_id := item.get(CONF_ID), str):
+        return _fill_script_stop_ids(item, script_id)
+    return item
+
+
+def default_script_stop_ids(value: Any) -> Any:
+    """Make `script.stop` with no id, used inside a script, stop that script."""
+    if isinstance(value, list):
+        return [_fill_script_stop_ids_in_script(item) for item in value]
+    # A single script may be given as a mapping instead of a list
+    return _fill_script_stop_ids_in_script(value)
+
+
+CONFIG_SCHEMA = cv.All(
+    default_script_stop_ids,
+    automation.validate_automation(
+        {
+            # Don't declare id as cv.declare_id yet, because the ID type
+            # depends on the mode. Will be checked later with assign_declare_id
+            cv.Required(CONF_ID): cv.string_strict,
+            cv.Optional(CONF_MODE, default=CONF_SINGLE): cv.one_of(
+                *SCRIPT_MODES, lower=True
+            ),
+            cv.Optional(CONF_MAX_RUNS): cv.int_range(min=0, max=100),
+            cv.Optional(CONF_PARAMETERS, default={}): cv.Schema(
+                {
+                    validate_parameter_name: validate_parameter_type,
+                }
+            ),
+            cv.Optional(CONF_ON_STOP): automation.validate_automation(single=True),
+        },
+        extra_validators=cv.All(check_max_runs, assign_declare_id),
+    ),
 )
 
 
@@ -145,8 +200,12 @@ async def to_code(config):
 
         triggers.append((trigger, func_args, conf))
 
+    if any(CONF_ON_STOP in conf for conf in config):
+        cg.add_define("USE_SCRIPT_ON_STOP")
     for trigger, func_args, conf in triggers:
         await automation.build_automation(trigger, func_args, conf)
+        if (stop_conf := conf.get(CONF_ON_STOP)) is not None:
+            await automation.build_automation(trigger.get_stop_trigger(), [], stop_conf)
 
 
 @automation.register_action(
@@ -205,9 +264,19 @@ async def script_execute_action_to_code(config, action_id, template_arg, args):
     return var
 
 
+def _require_script_stop_id(config: ConfigType) -> ConfigType:
+    # Inside a script the id has already been filled in by default_script_stop_ids
+    if CONF_ID not in config:
+        raise cv.Invalid(
+            "'script.stop' without an id can only be used inside a script",
+            path=[CONF_ID],
+        )
+    return config
+
+
 automation.register_apply_action(
-    "script.stop",
-    maybe_simple_id({cv.Required(CONF_ID): cv.use_id(Script)}),
+    CONF_SCRIPT_STOP,
+    maybe_simple_id({cv.Optional(CONF_ID): cv.use_id(Script)}, _require_script_stop_id),
     automation.ApplyCall("stop()"),
 )
 

esphome/components/script/script.h

@@ -33,8 +33,18 @@ template<typename... Ts> class Script : public ScriptLogger, public Trigger<Ts..
   virtual void execute(Ts...) = 0;
   /// Check if any instance of this script is currently running.
   virtual bool is_running() { return this->is_action_running(); }
+#ifdef USE_SCRIPT_ON_STOP
+  /// Stop all instances of this script. Fires the on_stop trigger once if any instance was running.
+  virtual void stop() {
+    if (this->is_action_running())
+      this->stop_running_();
+  }
+
+  Trigger<> *get_stop_trigger() { return &this->stop_trigger_; }
+#else
   /// Stop all instances of this script.
   virtual void stop() { this->stop_action(); }
+#endif
 
   // execute this script using a tuple that contains the arguments
   void execute_tuple(const std::tuple<Ts...> &tuple) {
@@ -57,7 +67,18 @@ template<typename... Ts> class Script : public ScriptLogger, public Trigger<Ts..
     this->trigger(x...);
   }
 
+  // Stop the running actions early and fire the on_stop trigger.
+  void stop_running_() {
+    this->stop_action();
+#ifdef USE_SCRIPT_ON_STOP
+    this->stop_trigger_.trigger();
+#endif
+  }
+
   const LogString *name_{nullptr};
+#ifdef USE_SCRIPT_ON_STOP
+  Trigger<> stop_trigger_;
+#endif
 };
 
 /** A script type for which only a single instance at a time is allowed.
@@ -88,7 +109,12 @@ template<typename... Ts> class RestartScript : public Script<Ts...> {
   void execute(Ts... x) override {
     if (this->is_action_running()) {
       this->esp_logd_(__LINE__, ESPHOME_LOG_FORMAT("Script '%s' restarting (mode: restart)"), LOG_STR_ARG(this->name_));
-      this->stop_action();
+      this->stop_running_();
+#ifdef USE_SCRIPT_ON_STOP
+      // on_stop may have executed this script again; only the newest run may continue
+      if (this->is_action_running())
+        this->stop_action();
+#endif
     }
 
     this->run_actions_(x...);
@@ -145,12 +171,21 @@ template<typename... Ts> class QueueingScript : public Script<Ts...>, public Com
   }
 
   void stop() override {
+#ifdef USE_SCRIPT_ON_STOP
+    // Queued runs are unfinished work too, even between two runs when nothing is running
+    const bool stopping = this->is_action_running() || this->num_queued_ != 0;
+#endif
     // Clear all queued items to free memory immediately
     // Resetting the array automatically destroys all unique_ptrs and their contents
     this->var_queue_.reset();
     this->num_queued_ = 0;
     this->queue_front_ = 0;
+#ifdef USE_SCRIPT_ON_STOP
+    if (stopping)
+      this->stop_running_();
+#else
     Script<Ts...>::stop();
+#endif
   }
 
   void loop() override {

esphome/core/defines.h

@@ -266,6 +266,7 @@
 #define USE_SAFE_MODE_BOOT_IS_GOOD_ON_SHUTDOWN
 #define USE_SAFE_MODE_CALLBACK
 #define ESPHOME_SAFE_MODE_CALLBACK_COUNT 1
+#define USE_SCRIPT_ON_STOP
 #define USE_SELECT
 #define USE_SELEC_METER_EM4M
 #define USE_SENSOR

tests/components/script/__init__.py

@@ -0,0 +1,9 @@
+import esphome.codegen as cg
+from tests.testing_helpers import ComponentManifestOverride
+
+
+def override_manifest(manifest: ComponentManifestOverride) -> None:
+    async def to_code_testing(config):
+        cg.add_define("USE_SCRIPT_ON_STOP")
+
+    manifest.to_code = to_code_testing

tests/components/script/common.yaml

@@ -22,6 +22,25 @@ esphome:
             - script.is_running: my_script
           then:
             - logger.log: my_script is running
+      - if:
+          any:
+            - script.is_running: my_script
+            - script.is_running: my_script_restart
+          then:
+            - logger.log: a script is running
+      - script.execute: my_script_stops_itself
+      - choose:
+          - if:
+              script.is_running: my_script
+            then:
+              - logger.log: first choice
+          - if:
+              - lambda: return true;
+              - script.is_running: my_script_restart
+            then:
+              - logger.log: second choice
+          - else:
+              - logger.log: default choice
 
 globals:
   - id: nested_total
@@ -44,8 +63,22 @@ script:
       - lambda: 'ESP_LOGD("main", "Hello World!");'
   - id: my_script_restart
     mode: restart
+    on_stop:
+      - logger.log: my_script_restart stopped
     then:
       - lambda: 'ESP_LOGD("main", "Hello World!");'
+  - id: my_script_stops_itself
+    on_stop:
+      then:
+        - logger.log: my_script_stops_itself stopped
+    then:
+      - delay: 1s
+      - if:
+          lambda: return true;
+          then:
+            - script.stop
+      - script.stop:
+      - logger.log: never reached
   - id: my_script_with_array_params
     parameters:
       ints: int[]

tests/components/script/test_script_on_stop.cpp

@@ -0,0 +1,77 @@
+#include <gtest/gtest.h>
+
+#include "esphome/components/script/script.h"
+
+namespace esphome::script::testing {
+
+class CountAction : public Action<> {
+ public:
+  explicit CountAction(int *count) : count_(count) {}
+
+ protected:
+  void play() override { (*this->count_)++; }
+  int *count_;
+};
+
+// Stays running until release() is called, like a delay that never expires on its own.
+class HoldAction : public Action<> {
+ public:
+  void play_complex() override {
+    this->num_running_++;
+    this->plays++;
+  }
+  void release() { this->play_next_(); }
+  int plays{0};
+
+ protected:
+  void play() override {}
+};
+
+class QueueingScriptOnStopTest : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    this->script_.set_max_runs(3);
+    this->automation_.add_action(&this->hold_);
+    this->stop_automation_.add_action(&this->count_);
+  }
+
+  QueueingScript<> script_;
+  Automation<> automation_{&this->script_};
+  HoldAction hold_;
+  int on_stop_calls_{0};
+  Automation<> stop_automation_{this->script_.get_stop_trigger()};
+  CountAction count_{&this->on_stop_calls_};
+};
+
+TEST_F(QueueingScriptOnStopTest, StopWhileRunningFiresOnStopOnceAndDropsQueuedRun) {
+  this->script_.execute();
+  this->script_.execute();
+  ASSERT_EQ(this->hold_.plays, 1);
+
+  this->script_.stop();
+  EXPECT_EQ(this->on_stop_calls_, 1);
+
+  this->script_.loop();
+  EXPECT_EQ(this->hold_.plays, 1);
+}
+
+// After one run ends, the next queued run only starts on the following loop() call
+TEST_F(QueueingScriptOnStopTest, StopBetweenQueuedRunsFiresOnStop) {
+  this->script_.execute();
+  this->script_.execute();
+  this->hold_.release();
+  ASSERT_FALSE(this->script_.is_running());
+
+  this->script_.stop();
+  EXPECT_EQ(this->on_stop_calls_, 1);
+
+  this->script_.loop();
+  EXPECT_EQ(this->hold_.plays, 1);
+}
+
+TEST_F(QueueingScriptOnStopTest, StopWhenIdleDoesNotFireOnStop) {
+  this->script_.stop();
+  EXPECT_EQ(this->on_stop_calls_, 0);
+}
+
+}  // namespace esphome::script::testing

tests/integration/fixtures/script_on_stop_and_choose.yaml

@@ -0,0 +1,145 @@
+esphome:
+  name: test-script-on-stop-choose
+
+host:
+
+api:
+  actions:
+    - action: run_choose
+      variables:
+        value: int
+      then:
+        - choose:
+            - if:
+                lambda: return value < 10;
+              then:
+                - logger.log:
+                    format: "CHOOSE small %d"
+                    args: [value]
+            - if:
+                lambda: return value < 100;
+              then:
+                - delay: 20ms
+                - logger.log:
+                    format: "CHOOSE medium %d"
+                    args: [value]
+            # Also true for smaller values, but only the first true choice runs
+            - if:
+                lambda: return value < 1000;
+              then:
+                - logger.log:
+                    format: "CHOOSE large %d"
+                    args: [value]
+            - else:
+                - logger.log:
+                    format: "CHOOSE default %d"
+                    args: [value]
+        - logger.log:
+            format: "CHOOSE done %d"
+            args: [value]
+    - action: run_self_stop
+      then:
+        - script.execute: self_stop
+    - action: run_external_stop
+      then:
+        - script.execute: long_script
+        - delay: 50ms
+        - script.stop: long_script
+    - action: run_idle_stop
+      then:
+        - script.stop: long_script
+        - logger.log: "IDLE stop done"
+    - action: run_restart
+      then:
+        - script.execute: restart_script
+        - script.execute: restart_script
+    - action: run_normal
+      then:
+        - script.execute: normal_script
+    - action: run_reenter
+      then:
+        - script.execute: reenter_script
+        - delay: 20ms
+        - script.execute: reenter_script
+    - action: run_parallel_stop
+      then:
+        - script.execute: parallel_script
+        - script.execute: parallel_script
+        - delay: 20ms
+        - script.stop: parallel_script
+        - logger.log: "PARALLEL stop done"
+    - action: run_queued_stop
+      then:
+        - script.execute: queued_script
+        - script.execute: queued_script
+        - delay: 20ms
+        - script.stop: queued_script
+        - delay: 300ms
+        - logger.log: "QUEUED stop done"
+
+logger:
+  level: DEBUG
+
+script:
+  - id: self_stop
+    on_stop:
+      - logger.log: "SELF on_stop"
+    then:
+      - logger.log: "SELF start"
+      - delay: 20ms
+      - if:
+          condition:
+            lambda: return true;
+          then:
+            - script.stop
+      - logger.log: "SELF not reached"
+
+  - id: long_script
+    on_stop:
+      - logger.log: "LONG on_stop"
+    then:
+      - delay: 10s
+      - logger.log: "LONG finished"
+
+  - id: restart_script
+    mode: restart
+    on_stop:
+      - logger.log: "RESTART on_stop"
+    then:
+      - logger.log: "RESTART start"
+      - delay: 50ms
+      - logger.log: "RESTART finished"
+
+  - id: normal_script
+    on_stop:
+      - logger.log: "NORMAL on_stop"
+    then:
+      - logger.log: "NORMAL finished"
+
+  # on_stop executes the same restart script again; only one run may continue
+  - id: reenter_script
+    mode: restart
+    on_stop:
+      - logger.log: "REENTER on_stop"
+      - script.execute: reenter_script
+    then:
+      - logger.log: "REENTER start"
+      - delay: 100ms
+      - logger.log: "REENTER finished"
+
+  - id: parallel_script
+    mode: parallel
+    on_stop:
+      - logger.log: "PARALLEL on_stop"
+    then:
+      - delay: 10s
+      - logger.log: "PARALLEL finished"
+
+  - id: queued_script
+    mode: queued
+    on_stop:
+      - logger.log: "QUEUED on_stop"
+    then:
+      - logger.log: "QUEUED start"
+      - delay: 100ms
+      - logger.log: "QUEUED finished"

tests/integration/test_script_on_stop_and_choose.py

@@ -0,0 +1,117 @@
+"""Integration test for the script on_stop trigger, script.stop without an id, and the choose action."""
+
+from __future__ import annotations
+
+import asyncio
+
+from aioesphomeapi import UserService
+import pytest
+
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_script_on_stop_and_choose(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test on_stop, self-stopping scripts and choose."""
+    lines: list[str] = []
+    waiters: list[tuple[str, asyncio.Future[None]]] = []
+
+    def check_output(line: str) -> None:
+        lines.append(line)
+        for text, future in waiters:
+            if text in line and not future.done():
+                future.set_result(None)
+
+    def expect(text: str) -> asyncio.Future[None]:
+        future: asyncio.Future[None] = asyncio.get_running_loop().create_future()
+        waiters.append((text, future))
+        return future
+
+    def count(text: str) -> int:
+        return sum(text in line for line in lines)
+
+    async with (
+        run_compiled(yaml_config, line_callback=check_output),
+        api_client_connected() as client,
+    ):
+        _, services = await client.list_entities_services()
+        service: dict[str, UserService] = {s.name: s for s in services}
+
+        # choose runs only the first choice whose condition is true, or the else
+        # choice if none is, then continues
+        for value, label in (
+            (5, "small"),
+            (50, "medium"),
+            (500, "large"),
+            (5000, "default"),
+        ):
+            done = expect(f"CHOOSE done {value}")
+            await client.execute_service(service["run_choose"], {"value": value})
+            await asyncio.wait_for(done, timeout=5.0)
+            assert count(f"CHOOSE {label} {value}") == 1
+        assert count("CHOOSE small") == 1
+        assert count("CHOOSE medium") == 1
+        assert count("CHOOSE large") == 1
+        assert count("CHOOSE default") == 1
+
+        # A bare script.stop inside a script stops that script and fires on_stop
+        stopped = expect("SELF on_stop")
+        await client.execute_service(service["run_self_stop"], {})
+        await asyncio.wait_for(stopped, timeout=5.0)
+
+        # Stopping a running script from outside fires on_stop
+        stopped = expect("LONG on_stop")
+        await client.execute_service(service["run_external_stop"], {})
+        await asyncio.wait_for(stopped, timeout=5.0)
+
+        # Stopping a script that is not running does not fire on_stop
+        idle_done = expect("IDLE stop done")
+        await client.execute_service(service["run_idle_stop"], {})
+        await asyncio.wait_for(idle_done, timeout=5.0)
+
+        # Restarting a running script stops the old run early
+        finished = expect("RESTART finished")
+        await client.execute_service(service["run_restart"], {})
+        await asyncio.wait_for(finished, timeout=5.0)
+
+        # A script that finishes normally does not fire on_stop
+        finished = expect("NORMAL finished")
+        await client.execute_service(service["run_normal"], {})
+        await asyncio.wait_for(finished, timeout=5.0)
+
+        # A restart script whose on_stop executes it again ends up with one run
+        finished = expect("REENTER finished")
+        await client.execute_service(service["run_reenter"], {})
+        await asyncio.wait_for(finished, timeout=5.0)
+
+        # Stopping several parallel runs fires on_stop once
+        done = expect("PARALLEL stop done")
+        await client.execute_service(service["run_parallel_stop"], {})
+        await asyncio.wait_for(done, timeout=5.0)
+
+        # Stopping a queued script fires on_stop once and drops the queued run
+        done = expect("QUEUED stop done")
+        await client.execute_service(service["run_queued_stop"], {})
+        await asyncio.wait_for(done, timeout=5.0)
+        await asyncio.sleep(0.2)
+
+        assert count("SELF not reached") == 0
+        assert count("SELF on_stop") == 1
+        assert count("LONG finished") == 0
+        assert count("LONG on_stop") == 1
+        assert count("RESTART start") == 2
+        assert count("RESTART on_stop") == 1
+        assert count("RESTART finished") == 1
+        assert count("NORMAL on_stop") == 0
+        assert count("REENTER on_stop") == 1
+        assert count("REENTER start") == 3
+        assert count("REENTER finished") == 1
+        assert count("PARALLEL on_stop") == 1
+        assert count("PARALLEL finished") == 0
+        assert count("QUEUED on_stop") == 1
+        assert count("QUEUED start") == 1
+        assert count("QUEUED finished") == 0

tests/unit_tests/components/test_script.py

@@ -0,0 +1,160 @@
+"""Tests for the script component's config validation."""
+
+from typing import Any
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from esphome import automation
+from esphome.components.const import CONF_ON_STOP
+from esphome.components.script import (
+    CONF_SCRIPT_STOP,
+    SingleScript,
+    default_script_stop_ids,
+    to_code,
+)
+import esphome.config_validation as cv
+from esphome.const import CONF_ID, CONF_MODE, CONF_PARAMETERS, CONF_THEN
+from esphome.core import ID
+from esphome.cpp_generator import MockObj
+
+
+def test_bare_script_stop_defaults_to_enclosing_script() -> None:
+    config = [
+        {
+            CONF_ID: "my_script",
+            CONF_THEN: [
+                CONF_SCRIPT_STOP,
+                {CONF_SCRIPT_STOP: None},
+                {CONF_SCRIPT_STOP: {}},
+                {CONF_SCRIPT_STOP: "other_script"},
+                {CONF_SCRIPT_STOP: {CONF_ID: "other_script"}},
+                {"if": {"condition": "x", CONF_THEN: [CONF_SCRIPT_STOP]}},
+            ],
+            "on_stop": [{CONF_SCRIPT_STOP: None}],
+        },
+        {CONF_ID: "second", CONF_THEN: [CONF_SCRIPT_STOP]},
+    ]
+
+    result = default_script_stop_ids(config)
+
+    assert result[0][CONF_THEN] == [
+        {CONF_SCRIPT_STOP: {CONF_ID: "my_script"}},
+        {CONF_SCRIPT_STOP: {CONF_ID: "my_script"}},
+        {CONF_SCRIPT_STOP: {CONF_ID: "my_script"}},
+        {CONF_SCRIPT_STOP: "other_script"},
+        {CONF_SCRIPT_STOP: {CONF_ID: "other_script"}},
+        {
+            "if": {
+                "condition": "x",
+                CONF_THEN: [{CONF_SCRIPT_STOP: {CONF_ID: "my_script"}}],
+            }
+        },
+    ]
+    assert result[0]["on_stop"] == [{CONF_SCRIPT_STOP: {CONF_ID: "my_script"}}]
+    assert result[1][CONF_THEN] == [{CONF_SCRIPT_STOP: {CONF_ID: "second"}}]
+    # The input is not modified
+    assert config[1][CONF_THEN] == [CONF_SCRIPT_STOP]
+
+
+def test_unchanged_script_config_is_not_copied() -> None:
+    then = [{"logger.log": "hello"}]
+    config = [{CONF_ID: "my_script", CONF_THEN: then}]
+
+    result = default_script_stop_ids(config)
+
+    assert result[0] is config[0]
+    assert result[0][CONF_THEN] is then
+
+
+def test_script_stop_without_id_outside_script_is_invalid() -> None:
+    with pytest.raises(cv.Invalid, match="can only be used inside a script"):
+        automation.validate_action(CONF_SCRIPT_STOP)
+
+
+def test_script_stop_with_id_is_valid() -> None:
+    result = automation.validate_action({CONF_SCRIPT_STOP: "my_script"})
+
+    assert result[CONF_SCRIPT_STOP][CONF_ID].id == "my_script"
+
+
+@pytest.mark.parametrize(
+    "value",
+    [
+        pytest.param("my_script", id="not_a_list_or_mapping"),
+        pytest.param({CONF_THEN: [CONF_SCRIPT_STOP]}, id="mapping_without_id"),
+        pytest.param(["my_script", {CONF_THEN: [CONF_SCRIPT_STOP]}], id="no_id"),
+    ],
+)
+def test_script_stop_ids_left_alone_without_script_id(value: Any) -> None:
+    assert default_script_stop_ids(value) == value
+
+
+def test_bare_script_stop_in_single_script_mapping() -> None:
+    config = {CONF_ID: "my_script", CONF_THEN: [CONF_SCRIPT_STOP]}
+
+    result = default_script_stop_ids(config)
+
+    assert result[CONF_THEN] == [{CONF_SCRIPT_STOP: {CONF_ID: "my_script"}}]
+
+
+@pytest.mark.asyncio
+async def test_on_stop_builds_stop_trigger_automation() -> None:
+    stop_conf = {CONF_THEN: []}
+    config = [
+        {
+            CONF_ID: ID("my_script", is_declaration=True, type=SingleScript),
+            CONF_MODE: "single",
+            CONF_PARAMETERS: {},
+            CONF_THEN: [],
+            CONF_ON_STOP: stop_conf,
+        },
+        {
+            CONF_ID: ID("other_script", is_declaration=True, type=SingleScript),
+            CONF_MODE: "single",
+            CONF_PARAMETERS: {},
+            CONF_THEN: [],
+        },
+    ]
+    with (
+        patch("esphome.codegen.new_Pvariable") as new_pvariable,
+        patch("esphome.codegen.add"),
+        patch("esphome.codegen.add_define") as add_define,
+        patch(
+            "esphome.automation.build_automation", new_callable=AsyncMock
+        ) as build_automation,
+    ):
+        new_pvariable.side_effect = lambda id_, *args: MockObj(id_.id, "->")
+        await to_code(config)
+
+    add_define.assert_called_once_with("USE_SCRIPT_ON_STOP")
+
+    assert [
+        (str(c.args[0]), *c.args[1:]) for c in build_automation.await_args_list
+    ] == [
+        ("my_script", [], config[0]),
+        ("my_script->get_stop_trigger()", [], stop_conf),
+        ("other_script", [], config[1]),
+    ]
+
+
+@pytest.mark.asyncio
+async def test_no_on_stop_define_without_on_stop() -> None:
+    config = [
+        {
+            CONF_ID: ID("my_script", is_declaration=True, type=SingleScript),
+            CONF_MODE: "single",
+            CONF_PARAMETERS: {},
+            CONF_THEN: [],
+        }
+    ]
+    with (
+        patch("esphome.codegen.new_Pvariable") as new_pvariable,
+        patch("esphome.codegen.add"),
+        patch("esphome.codegen.add_define") as add_define,
+        patch("esphome.automation.build_automation", new_callable=AsyncMock),
+    ):
+        new_pvariable.side_effect = lambda id_, *args: MockObj(id_.id, "->")
+        await to_code(config)
+
+    add_define.assert_not_called()

tests/unit_tests/test_automation.py

@@ -8,6 +8,8 @@ from unittest.mock import AsyncMock, MagicMock, call, patch
 import pytest
 
 from esphome.automation import (
+    CONF_CHOICE_ID,
+    CONF_CHOOSE,
     ApplyAction,
     ApplyCall,
     ApplyCondition,
@@ -25,7 +27,9 @@ from esphome.automation import (
     build_parent_callback_automation,
     build_trigger_automations,
     build_trigger_callback,
+    choose_action_to_code,
     has_non_synchronous_actions,
+    if_action_to_code,
     literal_with_length,
     maybe_simple_id,
     parent_ref,
@@ -40,14 +44,19 @@ from esphome.automation import (
     register_simple_condition,
     string_ref_literal,
     templatable_bytes,
+    validate_action,
     validate_no_deferring_actions,
     validate_synchronous_automation,
 )
 import esphome.codegen as cg
 import esphome.config_validation as cv
 from esphome.const import (
+    CONF_ALL,
+    CONF_ANY,
     CONF_AUTOMATION_ID,
+    CONF_CONDITION,
     CONF_COUNT,
+    CONF_ELSE,
     CONF_ID,
     CONF_THEN,
     CONF_TYPE_ID,
@@ -1557,6 +1566,219 @@ async def test_build_callback_automation_forwarder(mock_cg: MockCodegen) -> None
     )
 
 
+def test_if_accepts_bare_condition() -> None:
+    """A single condition directly under `if:` is the same as `condition: {...}`."""
+    bare = validate_action({"if": {"or": [], CONF_THEN: [{"delay": "1s"}]}})
+    wrapped = validate_action(
+        {"if": {CONF_CONDITION: {"or": []}, CONF_THEN: [{"delay": "1s"}]}}
+    )
+    assert bare["if"][CONF_CONDITION] == wrapped["if"][CONF_CONDITION]
+    assert "or" not in bare["if"]
+
+
+@pytest.mark.parametrize("key", [CONF_ANY, CONF_ALL])
+def test_if_any_all_are_not_bare_conditions(key: str) -> None:
+    """`any:` and `all:` under `if:` are the if action's own keys, not a bare condition."""
+    result = validate_action({"if": {key: [{"or": []}], CONF_THEN: [{"delay": "1s"}]}})
+    assert key in result["if"]
+    assert CONF_CONDITION not in result["if"]
+
+
+@pytest.mark.parametrize(
+    ("value", "match"),
+    [
+        pytest.param(
+            {"or": [], "and": [], CONF_THEN: []},
+            "Only one condition",
+            id="two_bare",
+        ),
+        pytest.param(
+            {"or": [], CONF_CONDITION: {"and": []}, CONF_THEN: []},
+            "cannot be used with 'condition'",
+            id="bare_and_wrapped",
+        ),
+        pytest.param(
+            {"or": [], CONF_ANY: [{"and": []}], CONF_THEN: []},
+            "cannot be used with 'any'",
+            id="bare_and_any",
+        ),
+    ],
+)
+def test_if_rejects_ambiguous_bare_condition(
+    value: dict[str, object], match: str
+) -> None:
+    with pytest.raises(cv.Invalid, match=match):
+        validate_action({"if": value})
+
+
+def test_if_with_non_dict_value_is_invalid() -> None:
+    with pytest.raises(cv.Invalid, match="expected a dictionary"):
+        validate_action({"if": "not a dict"})
+
+
+CHOICE = {"if": {"and": []}, CONF_THEN: [{"delay": "1s"}]}
+
+
+@pytest.mark.parametrize(
+    ("value", "match"),
+    [
+        pytest.param([], "at least 1", id="empty"),
+        pytest.param([{CONF_THEN: [{"delay": "1s"}]}], "'if' is required", id="no_if"),
+        pytest.param([{"if": {"and": []}}], "'then' is required", id="no_then"),
+        pytest.param(
+            [{CONF_ELSE: [{"delay": "1s"}]}],
+            "At least one choice must have 'if' and 'then'",
+            id="only_else",
+        ),
+        pytest.param(
+            [{CONF_ELSE: [{"delay": "1s"}]}, CHOICE],
+            "Only the last choice may be 'else'",
+            id="else_not_last",
+        ),
+        pytest.param(
+            [CHOICE, {"if": {"and": []}, CONF_ELSE: []}],
+            "'if' cannot be used with 'else'",
+            id="else_with_if",
+        ),
+        pytest.param(
+            [CHOICE, {CONF_THEN: [], CONF_ELSE: []}],
+            "'then' cannot be used with 'else'",
+            id="else_with_then",
+        ),
+    ],
+)
+def test_choose_invalid(value: list[ConfigType], match: str) -> None:
+    with pytest.raises(cv.Invalid, match=match):
+        validate_action({CONF_CHOOSE: value})
+
+
+def test_choose_valid() -> None:
+    result = validate_action(
+        {
+            CONF_CHOOSE: [
+                CHOICE,
+                {"if": [{"or": []}, {"and": []}], CONF_THEN: [{"delay": "2s"}]},
+                {CONF_ELSE: [{"delay": "3s"}]},
+            ]
+        }
+    )
+
+    choices = result[CONF_CHOOSE]
+    assert len(choices) == 3
+    assert "and" in choices[1]["if"]
+    assert CONF_ELSE in choices[2]
+    assert all(CONF_CHOICE_ID in choice for choice in choices)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("has_else", [True, False])
+async def test_choose_builds_if_else_chain(has_else: bool) -> None:
+    """Each choice is an IfAction whose else branch holds the next choice, then the else actions."""
+    config: list[ConfigType] = [
+        {"if": "cond_a", CONF_THEN: "then_a", CONF_CHOICE_ID: ID("choice_a")},
+        {"if": "cond_b", CONF_THEN: "then_b", CONF_CHOICE_ID: ID("choice_b")},
+    ]
+    if has_else:
+        config.append({CONF_ELSE: "else_c", CONF_CHOICE_ID: ID("choice_c")})
+    action_id = ID("action")
+    with (
+        patch(
+            "esphome.automation.build_condition",
+            new_callable=AsyncMock,
+            side_effect=lambda conf, *_: f"built_{conf}",
+        ),
+        patch(
+            "esphome.automation.build_action_list",
+            new_callable=AsyncMock,
+            side_effect=lambda conf, *_: [MockObj(f"built_{conf}")],
+        ),
+        patch("esphome.codegen.new_Pvariable") as new_pvariable,
+        patch("esphome.codegen.add") as add,
+    ):
+        new_pvariable.side_effect = lambda id_, *args: MockObj(id_.id, "->")
+        result = await choose_action_to_code(config, action_id, TEMPLATE_ARG, [])
+
+    # Built from the last choice back; the first choice takes the action's own id
+    created = [
+        (c.args[0].id, str(c.args[1]), c.args[2]) for c in new_pvariable.call_args_list
+    ]
+    assert created == [
+        ("choice_b", f"<{str(has_else).lower()}>", "built_cond_b"),
+        ("action", "<true>", "built_cond_a"),
+    ]
+    assert str(result) == "action"
+    statements = [str(c.args[0]) for c in add.call_args_list]
+    expected_b_else = (
+        ["choice_b->add_else_action(built_else_c)", "choice_b->finish_else()"]
+        if has_else
+        else []
+    )
+    assert statements == [
+        "choice_b->add_then_action(built_then_b)",
+        "choice_b->finish_then()",
+        *expected_b_else,
+        "action->add_then_action(built_then_a)",
+        "action->finish_then()",
+        "action->add_else_action(choice_b)",
+        "action->finish_else()",
+    ]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("config", "template_arg", "expected"),
+    [
+        pytest.param(
+            {CONF_CONDITION: "cond", CONF_THEN: "then", CONF_ELSE: "else"},
+            "<true>",
+            [
+                "action->add_then_action(built_then)",
+                "action->finish_then()",
+                "action->add_else_action(built_else)",
+                "action->finish_else()",
+            ],
+            id="then_and_else",
+        ),
+        pytest.param(
+            {CONF_ANY: "cond", CONF_ELSE: "else"},
+            "<true>",
+            ["action->add_else_action(built_else)", "action->finish_else()"],
+            id="only_else",
+        ),
+        pytest.param(
+            {CONF_ALL: "cond", CONF_THEN: "then"},
+            "<false>",
+            ["action->add_then_action(built_then)", "action->finish_then()"],
+            id="only_then",
+        ),
+    ],
+)
+async def test_if_adds_branches_one_action_at_a_time(
+    config: ConfigType, template_arg: str, expected: list[str]
+) -> None:
+    with (
+        patch(
+            "esphome.automation.build_condition",
+            new_callable=AsyncMock,
+            side_effect=lambda conf, *_: f"built_{conf}",
+        ),
+        patch(
+            "esphome.automation.build_action_list",
+            new_callable=AsyncMock,
+            side_effect=lambda conf, *_: [MockObj(f"built_{conf}")],
+        ),
+        patch("esphome.codegen.new_Pvariable") as new_pvariable,
+        patch("esphome.codegen.add") as add,
+    ):
+        new_pvariable.side_effect = lambda id_, *args: MockObj(id_.id, "->")
+        await if_action_to_code(config, ID("action"), TEMPLATE_ARG, [])
+
+    new_pvariable.assert_called_once()
+    assert str(new_pvariable.call_args.args[1]) == template_arg
+    assert new_pvariable.call_args.args[2] == "built_cond"
+    assert [str(c.args[0]) for c in add.call_args_list] == expected
+
+
 def test_shadow_iteration_leaves_other_names() -> None:
     args = [(cg.std_string, "prefix"), (cg.uint32, "iteration_1")]
     assert _shadow_iteration(args) == args