preen

esphome/__main__.py

@@ -222,8 +222,13 @@ def choose_upload_log_host(
             else:
                 resolved.append(device)
         if not resolved:
+            if CORE.dashboard:
+                hint = "If you know the IP, set 'use_address' in your network config."
+            else:
+                hint = "If you know the IP, try --device <IP>"
             raise EsphomeError(
-                f"All specified devices {defaults} could not be resolved. Is the device connected to the network?"
+                f"All specified devices {defaults} could not be resolved. "
+                f"Is the device connected to the network? {hint}"
             )
         return resolved
 

esphome/automation.py

@@ -57,8 +57,14 @@ def maybe_conf(conf, *validators):
     return validate
 
 
-def register_action(name: str, action_type: MockObjClass, schema: cv.Schema):
-    return ACTION_REGISTRY.register(name, action_type, schema)
+def register_action(
+    name: str,
+    action_type: MockObjClass,
+    schema: cv.Schema,
+    *,
+    is_sync: bool | None = None,
+):
+    return ACTION_REGISTRY.register(name, action_type, schema, is_sync=is_sync)
 
 
 def register_condition(name: str, condition_type: MockObjClass, schema: cv.Schema):
@@ -83,6 +89,42 @@ def validate_potentially_and_condition(value):
     return validate_condition(value)
 
 
+def automation_is_synchronous(action_configs: list[ConfigType]) -> bool:
+    """Check if an automation action list runs fully synchronously.
+
+    Returns False if any action in the chain (including nested actions)
+    is async or unknown (e.g., delay, wait_until, unregistered actions).
+    Used to determine if StringRef arguments are safe (synchronous) or
+    need copying to std::string (async/unknown).
+
+    Actions must be explicitly marked with is_sync=True to be considered
+    synchronous. Unknown actions (is_sync=None) are treated as potentially
+    async for safety.
+    """
+    if not action_configs:
+        return True
+
+    for action_config in action_configs:
+        # Extract action type (first key in the config dict)
+        action_type = next(iter(action_config.keys()))
+
+        # Check if this action is explicitly synchronous
+        # Unknown actions (not in registry or is_sync=None) are treated as async
+        registry_entry = ACTION_REGISTRY.get(action_type)
+        if registry_entry is None or registry_entry.is_sync is not True:
+            return False
+
+        # Recursively check nested action lists
+        config = action_config[action_type]
+        if isinstance(config, dict):
+            for nested_key in (CONF_THEN, CONF_ELSE):
+                nested_actions = config.get(nested_key)
+                if nested_actions and not automation_is_synchronous(nested_actions):
+                    return False
+
+    return True
+
+
 def validate_potentially_or_condition(value):
     if isinstance(value, list):
         with cv.remove_prepend_path(["or"]):
@@ -335,7 +377,10 @@ async def component_is_idle_condition_to_code(
 
 
 @register_action(
-    "delay", DelayAction, cv.templatable(cv.positive_time_period_milliseconds)
+    "delay",
+    DelayAction,
+    cv.templatable(cv.positive_time_period_milliseconds),
+    is_sync=False,
 )
 async def delay_action_to_code(
     config: ConfigType,
@@ -355,6 +400,7 @@ async def delay_action_to_code(
     IfAction,
     cv.All(
         {
+            # Control flow - recursively checks nested actions
             cv.Exclusive(
                 CONF_CONDITION, CONF_CONDITION
             ): validate_potentially_and_condition,
@@ -366,6 +412,7 @@ async def delay_action_to_code(
         cv.has_at_least_one_key(CONF_THEN, CONF_ELSE),
         cv.has_at_least_one_key(CONF_CONDITION, CONF_ANY, CONF_ALL),
     ),
+    is_sync=True,  # Sync depends on nested actions, checked recursively
 )
 async def if_action_to_code(
     config: ConfigType,
@@ -394,6 +441,7 @@ async def if_action_to_code(
             cv.Required(CONF_THEN): validate_action_list,
         }
     ),
+    is_sync=True,  # Sync depends on nested actions, checked recursively
 )
 async def while_action_to_code(
     config: ConfigType,
@@ -417,6 +465,7 @@ async def while_action_to_code(
             cv.Required(CONF_THEN): validate_action_list,
         }
     ),
+    is_sync=True,  # Sync depends on nested actions, checked recursively
 )
 async def repeat_action_to_code(
     config: ConfigType,
@@ -445,7 +494,7 @@ _validate_wait_until = cv.maybe_simple_value(
 )
 
 
-@register_action("wait_until", WaitUntilAction, _validate_wait_until)
+@register_action("wait_until", WaitUntilAction, _validate_wait_until, is_sync=False)
 async def wait_until_action_to_code(
     config: ConfigType,
     action_id: ID,
@@ -461,7 +510,7 @@ async def wait_until_action_to_code(
     return var
 
 
-@register_action("lambda", LambdaAction, cv.lambda_)
+@register_action("lambda", LambdaAction, cv.lambda_, is_sync=True)
 async def lambda_action_to_code(
     config: ConfigType,
     action_id: ID,
@@ -480,6 +529,7 @@ async def lambda_action_to_code(
             cv.Required(CONF_ID): cv.use_id(cg.PollingComponent),
         }
     ),
+    is_sync=True,
 )
 async def component_update_action_to_code(
     config: ConfigType,
@@ -499,6 +549,7 @@ async def component_update_action_to_code(
             cv.Required(CONF_ID): cv.use_id(cg.PollingComponent),
         }
     ),
+    is_sync=True,
 )
 async def component_suspend_action_to_code(
     config: ConfigType,
@@ -521,6 +572,7 @@ async def component_suspend_action_to_code(
             ),
         }
     ),
+    is_sync=True,
 )
 async def component_resume_action_to_code(
     config: ConfigType,

esphome/codegen.py

@@ -69,6 +69,7 @@ from esphome.cpp_types import (  # noqa: F401
     JsonObjectConst,
     Parented,
     PollingComponent,
+    StringRef,
     arduino_json_ns,
     bool_,
     const_char_ptr,

esphome/components/api/__init__.py

@@ -381,8 +381,15 @@ async def to_code(config: ConfigType) -> None:
                     func_args.append((cg.bool_, "return_response"))
 
             service_arg_names: list[str] = []
+            # Check if automation is synchronous to enable zero-copy StringRef for strings
+            is_sync = automation.automation_is_synchronous(conf[CONF_THEN])
             for name, var_ in conf[CONF_VARIABLES].items():
-                native = SERVICE_ARG_NATIVE_TYPES[var_]
+                # For string args in synchronous automations, use StringRef (zero-copy)
+                # For async automations, use std::string (safe copy)
+                if var_ == "string" and is_sync:
+                    native = cg.StringRef.operator("const").operator("ref")
+                else:
+                    native = SERVICE_ARG_NATIVE_TYPES[var_]
                 service_template_args.append(native)
                 func_args.append((native, name))
                 service_arg_names.append(name)

esphome/components/api/user_services.cpp

@@ -12,6 +12,11 @@ template<> int32_t get_execute_arg_value<int32_t>(const ExecuteServiceArgument &
 template<> float get_execute_arg_value<float>(const ExecuteServiceArgument &arg) { return arg.float_; }
 template<> std::string get_execute_arg_value<std::string>(const ExecuteServiceArgument &arg) { return arg.string_; }
 
+// Zero-copy version for synchronous automations
+template<> const StringRef &get_execute_arg_value<const StringRef &>(const ExecuteServiceArgument &arg) {
+  return arg.string_;
+}
+
 // Legacy std::vector versions for external components using custom_api_device.h - optimized with reserve
 template<> std::vector<bool> get_execute_arg_value<std::vector<bool>>(const ExecuteServiceArgument &arg) {
   std::vector<bool> result;
@@ -62,6 +67,9 @@ template<> enums::ServiceArgType to_service_arg_type<int32_t>() { return enums::
 template<> enums::ServiceArgType to_service_arg_type<float>() { return enums::SERVICE_ARG_TYPE_FLOAT; }
 template<> enums::ServiceArgType to_service_arg_type<std::string>() { return enums::SERVICE_ARG_TYPE_STRING; }
 
+// Zero-copy version for synchronous automations
+template<> enums::ServiceArgType to_service_arg_type<const StringRef &>() { return enums::SERVICE_ARG_TYPE_STRING; }
+
 // Legacy std::vector versions for external components using custom_api_device.h
 template<> enums::ServiceArgType to_service_arg_type<std::vector<bool>>() { return enums::SERVICE_ARG_TYPE_BOOL_ARRAY; }
 template<> enums::ServiceArgType to_service_arg_type<std::vector<int32_t>>() {

esphome/components/api/user_services.h

@@ -32,6 +32,9 @@ class UserServiceDescriptor {
 
 template<typename T> T get_execute_arg_value(const ExecuteServiceArgument &arg);
 
+// Specialization declarations for explicit template instantiation
+template<> const StringRef &get_execute_arg_value<const StringRef &>(const ExecuteServiceArgument &arg);
+
 template<typename T> enums::ServiceArgType to_service_arg_type();
 
 // Base class for YAML-defined services (most common case)

esphome/components/logger/__init__.py

@@ -476,7 +476,9 @@ LOGGER_LOG_ACTION_SCHEMA = cv.All(
 )
 
 
-@automation.register_action(CONF_LOGGER_LOG, LambdaAction, LOGGER_LOG_ACTION_SCHEMA)
+@automation.register_action(
+    CONF_LOGGER_LOG, LambdaAction, LOGGER_LOG_ACTION_SCHEMA, is_sync=True
+)
 async def logger_log_action_to_code(config, action_id, template_arg, args):
     esp_log = LOG_LEVEL_TO_ESP_LOG[config[CONF_LEVEL]]
     args_ = [cg.RawExpression(str(x)) for x in config[CONF_ARGS]]
@@ -500,6 +502,7 @@ async def logger_log_action_to_code(config, action_id, template_arg, args):
         },
         key=CONF_LEVEL,
     ),
+    is_sync=True,
 )
 async def logger_set_level_to_code(config, action_id, template_arg, args):
     level = LOG_LEVELS[config[CONF_LEVEL]]

esphome/components/remote_base/aeha_protocol.cpp

@@ -85,8 +85,8 @@ optional<AEHAData> AEHAProtocol::decode(RemoteReceiveData src) {
 std::string AEHAProtocol::format_data_(const std::vector<uint8_t> &data) {
   std::string out;
   for (uint8_t byte : data) {
-    char buf[8];  // "0x%02X," = 5 chars + null + margin
-    snprintf(buf, sizeof(buf), "0x%02X,", byte);
+    char buf[6];
+    sprintf(buf, "0x%02X,", byte);
     out += buf;
   }
   out.pop_back();

esphome/components/remote_base/raw_protocol.cpp

@@ -1,5 +1,4 @@
 #include "raw_protocol.h"
-#include "esphome/core/helpers.h"
 #include "esphome/core/log.h"
 
 namespace esphome {
@@ -9,30 +8,36 @@ static const char *const TAG = "remote.raw";
 
 bool RawDumper::dump(RemoteReceiveData src) {
   char buffer[256];
-  size_t pos = buf_append_printf(buffer, sizeof(buffer), 0, "Received Raw: ");
+  uint32_t buffer_offset = 0;
+  buffer_offset += sprintf(buffer, "Received Raw: ");
 
   for (int32_t i = 0; i < src.size() - 1; i++) {
     const int32_t value = src[i];
-    size_t prev_pos = pos;
+    const uint32_t remaining_length = sizeof(buffer) - buffer_offset;
+    int written;
 
     if (i + 1 < src.size() - 1) {
-      pos = buf_append_printf(buffer, sizeof(buffer), pos, "%" PRId32 ", ", value);
+      written = snprintf(buffer + buffer_offset, remaining_length, "%" PRId32 ", ", value);
     } else {
-      pos = buf_append_printf(buffer, sizeof(buffer), pos, "%" PRId32, value);
+      written = snprintf(buffer + buffer_offset, remaining_length, "%" PRId32, value);
     }
 
-    if (pos >= sizeof(buffer) - 1) {
-      // buffer full, flush and continue
-      buffer[prev_pos] = '\0';
+    if (written < 0 || written >= int(remaining_length)) {
+      // write failed, flush...
+      buffer[buffer_offset] = '\0';
       ESP_LOGI(TAG, "%s", buffer);
+      buffer_offset = 0;
+      written = sprintf(buffer, "  ");
       if (i + 1 < src.size() - 1) {
-        pos = buf_append_printf(buffer, sizeof(buffer), 0, "  %" PRId32 ", ", value);
+        written += sprintf(buffer + written, "%" PRId32 ", ", value);
       } else {
-        pos = buf_append_printf(buffer, sizeof(buffer), 0, "  %" PRId32, value);
+        written += sprintf(buffer + written, "%" PRId32, value);
       }
     }
+
+    buffer_offset += written;
   }
-  if (pos != 0) {
+  if (buffer_offset != 0) {
     ESP_LOGI(TAG, "%s", buffer);
   }
   return true;

esphome/components/remote_base/remote_base.cpp

@@ -1,5 +1,4 @@
 #include "remote_base.h"
-#include "esphome/core/helpers.h"
 #include "esphome/core/log.h"
 
 #include <cinttypes>
@@ -170,31 +169,36 @@ void RemoteTransmitterBase::send_(uint32_t send_times, uint32_t send_wait) {
 #ifdef ESPHOME_LOG_HAS_VERY_VERBOSE
   const auto &vec = this->temp_.get_data();
   char buffer[256];
-  size_t pos = buf_append_printf(buffer, sizeof(buffer), 0,
-                                 "Sending times=%" PRIu32 " wait=%" PRIu32 "ms: ", send_times, send_wait);
+  uint32_t buffer_offset = 0;
+  buffer_offset += sprintf(buffer, "Sending times=%" PRIu32 " wait=%" PRIu32 "ms: ", send_times, send_wait);
 
   for (size_t i = 0; i < vec.size(); i++) {
     const int32_t value = vec[i];
-    size_t prev_pos = pos;
+    const uint32_t remaining_length = sizeof(buffer) - buffer_offset;
+    int written;
 
     if (i + 1 < vec.size()) {
-      pos = buf_append_printf(buffer, sizeof(buffer), pos, "%" PRId32 ", ", value);
+      written = snprintf(buffer + buffer_offset, remaining_length, "%" PRId32 ", ", value);
     } else {
-      pos = buf_append_printf(buffer, sizeof(buffer), pos, "%" PRId32, value);
+      written = snprintf(buffer + buffer_offset, remaining_length, "%" PRId32, value);
     }
 
-    if (pos >= sizeof(buffer) - 1) {
-      // buffer full, flush and continue
-      buffer[prev_pos] = '\0';
+    if (written < 0 || written >= int(remaining_length)) {
+      // write failed, flush...
+      buffer[buffer_offset] = '\0';
       ESP_LOGVV(TAG, "%s", buffer);
+      buffer_offset = 0;
+      written = sprintf(buffer, "  ");
       if (i + 1 < vec.size()) {
-        pos = buf_append_printf(buffer, sizeof(buffer), 0, "  %" PRId32 ", ", value);
+        written += sprintf(buffer + written, "%" PRId32 ", ", value);
       } else {
-        pos = buf_append_printf(buffer, sizeof(buffer), 0, "  %" PRId32, value);
+        written += sprintf(buffer + written, "%" PRId32, value);
       }
     }
+
+    buffer_offset += written;
   }
-  if (pos != 0) {
+  if (buffer_offset != 0) {
     ESP_LOGVV(TAG, "%s", buffer);
   }
 #endif

esphome/core/string_ref.h

@@ -80,6 +80,21 @@ class StringRef {
 
   operator std::string() const { return str(); }
 
+  // compare() methods for std::string API compatibility
+  int compare(const StringRef &other) const {
+    size_t min_len = len_ < other.len_ ? len_ : other.len_;
+    int result = std::memcmp(base_, other.base_, min_len);
+    if (result != 0)
+      return result;
+    if (len_ < other.len_)
+      return -1;
+    if (len_ > other.len_)
+      return 1;
+    return 0;
+  }
+  int compare(const std::string &other) const { return compare(StringRef(other)); }
+  int compare(const char *other) const { return compare(StringRef(other)); }
+
  private:
   const char *base_;
   size_type len_;

esphome/cpp_types.py

@@ -23,6 +23,7 @@ size_t = global_ns.namespace("size_t")
 const_char_ptr = global_ns.namespace("const char *")
 NAN = global_ns.namespace("NAN")
 esphome_ns = global_ns  # using namespace esphome;
+StringRef = esphome_ns.class_("StringRef")
 FixedVector = esphome_ns.class_("FixedVector")
 App = esphome_ns.App
 EntityBase = esphome_ns.class_("EntityBase")

esphome/espota2.py

@@ -400,6 +400,8 @@ def run_ota_impl_(
             "Error resolving IP address of %s. Is it connected to WiFi?",
             remote_host,
         )
+        if not CORE.dashboard:
+            _LOGGER.error("(If you know the IP, try --device <IP>)")
         _LOGGER.error(
             "(If this error persists, please set a static IP address: "
             "https://esphome.io/components/wifi/#manual-ips)"

esphome/util.py

@@ -24,11 +24,17 @@ class RegistryEntry:
         fun: Callable[..., Any],
         type_id: "MockObjClass",
         schema: "Schema",
+        *,
+        is_sync: bool | None = None,
     ):
         self.name = name
         self.fun = fun
         self.type_id = type_id
         self.raw_schema = schema
+        # None = unknown (treated as potentially async for safety)
+        # True = explicitly synchronous (safe for zero-copy)
+        # False = explicitly async (needs copy)
+        self.is_sync = is_sync
 
     @property
     def coroutine_fun(self):
@@ -49,9 +55,16 @@ class Registry(dict[str, RegistryEntry]):
         self.base_schema = base_schema or {}
         self.type_id_key = type_id_key
 
-    def register(self, name: str, type_id: "MockObjClass", schema: "Schema"):
+    def register(
+        self,
+        name: str,
+        type_id: "MockObjClass",
+        schema: "Schema",
+        *,
+        is_sync: bool | None = None,
+    ):
         def decorator(fun: Callable[..., Any]):
-            self[name] = RegistryEntry(name, fun, type_id, schema)
+            self[name] = RegistryEntry(name, fun, type_id, schema, is_sync=is_sync)
             return fun
 
         return decorator

tests/integration/fixtures/api_string_sync_async.yaml

@@ -0,0 +1,67 @@
+esphome:
+  name: api-string-sync-async-test
+host:
+
+api:
+  actions:
+    # ==========================================================================
+    # Synchronous service - uses zero-copy StringRef
+    # Tests that .compare() and .c_str() work on StringRef
+    # ==========================================================================
+    - action: sync_string_service
+      variables:
+        command: string
+      supports_response: none
+      then:
+        - lambda: !lambda |-
+            if (command.compare("start") == 0) {
+              ESP_LOGI("sync_test", "Sync command: start");
+            } else if (command.compare("stop") == 0) {
+              ESP_LOGI("sync_test", "Sync command: stop");
+            } else if (command.compare("restart") == 0) {
+              ESP_LOGI("sync_test", "Sync command: restart");
+            } else {
+              ESP_LOGW("sync_test", "Sync unknown command: %s", command.c_str());
+            }
+
+    # ==========================================================================
+    # Asynchronous service - uses std::string copy
+    # The delay makes this async, requiring string data to be copied
+    # Tests that async services still work correctly with std::string
+    # ==========================================================================
+    - action: async_string_service
+      variables:
+        message: string
+      supports_response: none
+      then:
+        - logger.log:
+            format: "Async before delay: %s"
+            args: [message.c_str()]
+        - delay: 100ms
+        - lambda: !lambda |-
+            // After delay, the original receive buffer would be freed
+            // so we need std::string copy for this to work safely
+            if (message.compare("test") == 0) {
+              ESP_LOGI("async_test", "Async after delay: test matched");
+            } else {
+              ESP_LOGI("async_test", "Async after delay: %s", message.c_str());
+            }
+
+    # ==========================================================================
+    # Synchronous service with multiple strings
+    # Tests that multiple StringRef args work
+    # ==========================================================================
+    - action: sync_multi_string
+      variables:
+        first: string
+        second: string
+      supports_response: none
+      then:
+        - lambda: !lambda |-
+            bool first_match = first.compare("hello") == 0;
+            bool second_match = second.compare("world") == 0;
+            ESP_LOGI("multi_test", "Multi string: first=%s (%d), second=%s (%d)",
+                     first.c_str(), first_match, second.c_str(), second_match);
+
+logger:
+  level: DEBUG

tests/integration/test_api_string_sync_async.py

@@ -0,0 +1,129 @@
+"""Integration test for API string arguments in sync/async services.
+
+Tests that:
+1. Synchronous services use zero-copy StringRef and .compare() works
+2. Asynchronous services (with delay) use std::string copy safely
+3. Multiple string arguments work correctly
+"""
+
+from __future__ import annotations
+
+import asyncio
+import re
+
+from aioesphomeapi import UserService, UserServiceArgType
+import pytest
+
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_api_string_sync_async(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Test that string arguments work in both sync and async services."""
+    loop = asyncio.get_running_loop()
+
+    # Track log messages
+    sync_start_future = loop.create_future()
+    sync_stop_future = loop.create_future()
+    sync_unknown_future = loop.create_future()
+    async_before_future = loop.create_future()
+    async_after_future = loop.create_future()
+    multi_string_future = loop.create_future()
+
+    # Patterns to match in logs
+    sync_start_pattern = re.compile(r"Sync command: start")
+    sync_stop_pattern = re.compile(r"Sync command: stop")
+    sync_unknown_pattern = re.compile(r"Sync unknown command: custom_cmd")
+    async_before_pattern = re.compile(r"Async before delay: test_message")
+    async_after_pattern = re.compile(r"Async after delay: test_message")
+    # The output uses %d for bool which prints 0/1
+    multi_string_pattern = re.compile(r"Multi string: first=hello.*second=world")
+
+    def check_output(line: str) -> None:
+        """Check log output for expected messages."""
+        if not sync_start_future.done() and sync_start_pattern.search(line):
+            sync_start_future.set_result(True)
+        elif not sync_stop_future.done() and sync_stop_pattern.search(line):
+            sync_stop_future.set_result(True)
+        elif not sync_unknown_future.done() and sync_unknown_pattern.search(line):
+            sync_unknown_future.set_result(True)
+        elif not async_before_future.done() and async_before_pattern.search(line):
+            async_before_future.set_result(True)
+        elif not async_after_future.done() and async_after_pattern.search(line):
+            async_after_future.set_result(True)
+        elif not multi_string_future.done() and multi_string_pattern.search(line):
+            multi_string_future.set_result(True)
+
+    async with (
+        run_compiled(yaml_config, line_callback=check_output),
+        api_client_connected() as client,
+    ):
+        # Verify device info
+        device_info = await client.device_info()
+        assert device_info is not None
+        assert device_info.name == "api-string-sync-async-test"
+
+        # List services
+        _, services = await client.list_entities_services()
+
+        # Should have 3 services
+        assert len(services) == 3, f"Expected 3 services, found {len(services)}"
+
+        # Find our services
+        sync_service: UserService | None = None
+        async_service: UserService | None = None
+        multi_service: UserService | None = None
+
+        for service in services:
+            if service.name == "sync_string_service":
+                sync_service = service
+            elif service.name == "async_string_service":
+                async_service = service
+            elif service.name == "sync_multi_string":
+                multi_service = service
+
+        assert sync_service is not None, "sync_string_service not found"
+        assert async_service is not None, "async_string_service not found"
+        assert multi_service is not None, "sync_multi_string not found"
+
+        # Verify service arguments
+        assert len(sync_service.args) == 1
+        assert sync_service.args[0].name == "command"
+        assert sync_service.args[0].type == UserServiceArgType.STRING
+
+        assert len(async_service.args) == 1
+        assert async_service.args[0].name == "message"
+        assert async_service.args[0].type == UserServiceArgType.STRING
+
+        assert len(multi_service.args) == 2
+        multi_arg_types = {arg.name: arg.type for arg in multi_service.args}
+        assert multi_arg_types["first"] == UserServiceArgType.STRING
+        assert multi_arg_types["second"] == UserServiceArgType.STRING
+
+        # Test sync service with "start" command (tests .compare())
+        await client.execute_service(sync_service, {"command": "start"})
+        await asyncio.wait_for(sync_start_future, timeout=5.0)
+
+        # Test sync service with "stop" command
+        await client.execute_service(sync_service, {"command": "stop"})
+        await asyncio.wait_for(sync_stop_future, timeout=5.0)
+
+        # Test sync service with unknown command (tests .c_str())
+        await client.execute_service(sync_service, {"command": "custom_cmd"})
+        await asyncio.wait_for(sync_unknown_future, timeout=5.0)
+
+        # Test async service - this has a delay so needs std::string copy
+        await client.execute_service(async_service, {"message": "test_message"})
+        await asyncio.wait_for(async_before_future, timeout=5.0)
+        # Wait for the delayed log (100ms delay + some margin)
+        await asyncio.wait_for(async_after_future, timeout=5.0)
+
+        # Test multi-string service
+        await client.execute_service(
+            multi_service, {"first": "hello", "second": "world"}
+        )
+        await asyncio.wait_for(multi_string_future, timeout=5.0)

tests/unit_tests/test_automation.py

@@ -0,0 +1,484 @@
+"""Unit tests for esphome.automation module."""
+
+from esphome.automation import automation_is_synchronous
+
+# Import logger to register its actions with is_sync=True
+import esphome.components.logger  # noqa: F401  # pylint: disable=unused-import
+from esphome.const import CONF_ELSE, CONF_THEN
+
+
+def test_automation_is_synchronous_empty_list() -> None:
+    """Test that empty action list is considered synchronous."""
+    assert automation_is_synchronous([]) is True
+
+
+def test_automation_is_synchronous_single_sync_action() -> None:
+    """Test that a single synchronous action returns True."""
+    actions = [{"logger.log": {"message": "Hello"}}]
+    assert automation_is_synchronous(actions) is True
+
+
+def test_automation_is_synchronous_multiple_sync_actions() -> None:
+    """Test that multiple synchronous actions return True."""
+    actions = [
+        {"logger.log": {"message": "First"}},
+        {"lambda": "id(my_sensor).publish_state(42);"},
+        {"logger.log": {"message": "Second"}},
+    ]
+    assert automation_is_synchronous(actions) is True
+
+
+def test_automation_is_synchronous_delay_action() -> None:
+    """Test that delay action makes automation async."""
+    actions = [{"delay": "1s"}]
+    assert automation_is_synchronous(actions) is False
+
+
+def test_automation_is_synchronous_delay_in_middle() -> None:
+    """Test that delay in the middle of actions makes automation async."""
+    actions = [
+        {"logger.log": {"message": "Before"}},
+        {"delay": "100ms"},
+        {"logger.log": {"message": "After"}},
+    ]
+    assert automation_is_synchronous(actions) is False
+
+
+def test_automation_is_synchronous_wait_until_action() -> None:
+    """Test that wait_until action makes automation async."""
+    actions = [{"wait_until": {"condition": {"lambda": "return true;"}}}]
+    assert automation_is_synchronous(actions) is False
+
+
+def test_automation_is_synchronous_wait_until_with_timeout() -> None:
+    """Test that wait_until with timeout makes automation async."""
+    actions = [
+        {
+            "wait_until": {
+                "condition": {"lambda": "return true;"},
+                "timeout": "5s",
+            }
+        }
+    ]
+    assert automation_is_synchronous(actions) is False
+
+
+# Nested if tests
+def test_automation_is_synchronous_if_sync_then() -> None:
+    """Test that if with sync then block is synchronous."""
+    actions = [
+        {
+            "if": {
+                "condition": {"lambda": "return true;"},
+                CONF_THEN: [{"logger.log": {"message": "Then"}}],
+            }
+        }
+    ]
+    assert automation_is_synchronous(actions) is True
+
+
+def test_automation_is_synchronous_if_async_then() -> None:
+    """Test that if with async then block is asynchronous."""
+    actions = [
+        {
+            "if": {
+                "condition": {"lambda": "return true;"},
+                CONF_THEN: [{"delay": "1s"}],
+            }
+        }
+    ]
+    assert automation_is_synchronous(actions) is False
+
+
+def test_automation_is_synchronous_if_sync_else() -> None:
+    """Test that if with sync else block is synchronous."""
+    actions = [
+        {
+            "if": {
+                "condition": {"lambda": "return true;"},
+                CONF_THEN: [{"logger.log": {"message": "Then"}}],
+                CONF_ELSE: [{"logger.log": {"message": "Else"}}],
+            }
+        }
+    ]
+    assert automation_is_synchronous(actions) is True
+
+
+def test_automation_is_synchronous_if_async_else() -> None:
+    """Test that if with async else block is asynchronous."""
+    actions = [
+        {
+            "if": {
+                "condition": {"lambda": "return true;"},
+                CONF_THEN: [{"logger.log": {"message": "Then"}}],
+                CONF_ELSE: [{"delay": "1s"}],
+            }
+        }
+    ]
+    assert automation_is_synchronous(actions) is False
+
+
+def test_automation_is_synchronous_if_sync_then_async_else() -> None:
+    """Test that if with sync then but async else is asynchronous."""
+    actions = [
+        {
+            "if": {
+                "condition": {"lambda": "return true;"},
+                CONF_THEN: [{"logger.log": {"message": "Then"}}],
+                CONF_ELSE: [{"wait_until": {"condition": {"lambda": "return true;"}}}],
+            }
+        }
+    ]
+    assert automation_is_synchronous(actions) is False
+
+
+def test_automation_is_synchronous_if_async_then_sync_else() -> None:
+    """Test that if with async then but sync else is asynchronous."""
+    actions = [
+        {
+            "if": {
+                "condition": {"lambda": "return true;"},
+                CONF_THEN: [{"delay": "500ms"}],
+                CONF_ELSE: [{"logger.log": {"message": "Else"}}],
+            }
+        }
+    ]
+    assert automation_is_synchronous(actions) is False
+
+
+# Nested while tests
+def test_automation_is_synchronous_while_sync_body() -> None:
+    """Test that while with sync body is synchronous."""
+    actions = [
+        {
+            "while": {
+                "condition": {"lambda": "return false;"},
+                CONF_THEN: [{"logger.log": {"message": "Loop"}}],
+            }
+        }
+    ]
+    assert automation_is_synchronous(actions) is True
+
+
+def test_automation_is_synchronous_while_async_body() -> None:
+    """Test that while with async body is asynchronous."""
+    actions = [
+        {
+            "while": {
+                "condition": {"lambda": "return true;"},
+                CONF_THEN: [{"delay": "100ms"}],
+            }
+        }
+    ]
+    assert automation_is_synchronous(actions) is False
+
+
+# Nested repeat tests
+def test_automation_is_synchronous_repeat_sync_body() -> None:
+    """Test that repeat with sync body is synchronous."""
+    actions = [
+        {
+            "repeat": {
+                "count": 5,
+                CONF_THEN: [{"logger.log": {"message": "Iteration"}}],
+            }
+        }
+    ]
+    assert automation_is_synchronous(actions) is True
+
+
+def test_automation_is_synchronous_repeat_async_body() -> None:
+    """Test that repeat with async body is asynchronous."""
+    actions = [
+        {
+            "repeat": {
+                "count": 3,
+                CONF_THEN: [{"delay": "50ms"}],
+            }
+        }
+    ]
+    assert automation_is_synchronous(actions) is False
+
+
+# Deeply nested tests
+def test_automation_is_synchronous_deeply_nested_sync() -> None:
+    """Test that deeply nested sync actions are synchronous."""
+    actions = [
+        {
+            "if": {
+                "condition": {"lambda": "return true;"},
+                CONF_THEN: [
+                    {
+                        "while": {
+                            "condition": {"lambda": "return false;"},
+                            CONF_THEN: [
+                                {
+                                    "repeat": {
+                                        "count": 3,
+                                        CONF_THEN: [
+                                            {"logger.log": {"message": "Deep"}}
+                                        ],
+                                    }
+                                }
+                            ],
+                        }
+                    }
+                ],
+            }
+        }
+    ]
+    assert automation_is_synchronous(actions) is True
+
+
+def test_automation_is_synchronous_deeply_nested_async_in_repeat() -> None:
+    """Test that deeply nested delay in repeat makes automation async."""
+    actions = [
+        {
+            "if": {
+                "condition": {"lambda": "return true;"},
+                CONF_THEN: [
+                    {
+                        "while": {
+                            "condition": {"lambda": "return false;"},
+                            CONF_THEN: [
+                                {
+                                    "repeat": {
+                                        "count": 3,
+                                        CONF_THEN: [{"delay": "10ms"}],
+                                    }
+                                }
+                            ],
+                        }
+                    }
+                ],
+            }
+        }
+    ]
+    assert automation_is_synchronous(actions) is False
+
+
+def test_automation_is_synchronous_deeply_nested_async_in_while() -> None:
+    """Test that deeply nested wait_until in while makes automation async."""
+    actions = [
+        {
+            "if": {
+                "condition": {"lambda": "return true;"},
+                CONF_THEN: [
+                    {
+                        "while": {
+                            "condition": {"lambda": "return true;"},
+                            CONF_THEN: [
+                                {
+                                    "wait_until": {
+                                        "condition": {"lambda": "return true;"}
+                                    }
+                                }
+                            ],
+                        }
+                    }
+                ],
+            }
+        }
+    ]
+    assert automation_is_synchronous(actions) is False
+
+
+def test_automation_is_synchronous_deeply_nested_async_in_else() -> None:
+    """Test that deeply nested delay in else branch makes automation async."""
+    actions = [
+        {
+            "if": {
+                "condition": {"lambda": "return true;"},
+                CONF_THEN: [{"logger.log": {"message": "Then"}}],
+                CONF_ELSE: [
+                    {
+                        "if": {
+                            "condition": {"lambda": "return false;"},
+                            CONF_THEN: [{"logger.log": {"message": "Inner then"}}],
+                            CONF_ELSE: [{"delay": "1s"}],
+                        }
+                    }
+                ],
+            }
+        }
+    ]
+    assert automation_is_synchronous(actions) is False
+
+
+def test_automation_is_synchronous_multiple_nested_ifs_all_sync() -> None:
+    """Test multiple nested if statements that are all synchronous."""
+    actions = [
+        {
+            "if": {
+                "condition": {"lambda": "return true;"},
+                CONF_THEN: [
+                    {
+                        "if": {
+                            "condition": {"lambda": "return false;"},
+                            CONF_THEN: [{"logger.log": {"message": "A"}}],
+                            CONF_ELSE: [{"logger.log": {"message": "B"}}],
+                        }
+                    }
+                ],
+                CONF_ELSE: [
+                    {
+                        "if": {
+                            "condition": {"lambda": "return true;"},
+                            CONF_THEN: [{"logger.log": {"message": "C"}}],
+                            CONF_ELSE: [{"logger.log": {"message": "D"}}],
+                        }
+                    }
+                ],
+            }
+        }
+    ]
+    assert automation_is_synchronous(actions) is True
+
+
+def test_automation_is_synchronous_mixed_sync_async_sequence() -> None:
+    """Test sequence with sync action followed by async action."""
+    actions = [
+        {"logger.log": {"message": "First"}},
+        {"lambda": "id(sensor).publish_state(1);"},
+        {"delay": "1s"},  # This makes it async
+        {"logger.log": {"message": "Last"}},
+    ]
+    assert automation_is_synchronous(actions) is False
+
+
+def test_automation_is_synchronous_unknown_action_treated_as_async() -> None:
+    """Test that unknown actions (not in registry) are treated as async for safety."""
+    # This simulates an external component's action that isn't registered.
+    # Unknown actions default to is_sync=None, which is treated as potentially
+    # async for safety (safe-by-default approach).
+    actions = [{"my_custom_component.do_something": {"param": "value"}}]
+    assert automation_is_synchronous(actions) is False
+
+
+def test_automation_is_synchronous_empty_if_branches() -> None:
+    """Test if with missing then/else branches."""
+    # Just condition, no then or else
+    actions = [
+        {
+            "if": {
+                "condition": {"lambda": "return true;"},
+            }
+        }
+    ]
+    assert automation_is_synchronous(actions) is True
+
+
+def test_automation_is_synchronous_repeat_inside_if_else() -> None:
+    """Test repeat with delay inside if else branch."""
+    actions = [
+        {
+            "if": {
+                "condition": {"lambda": "return true;"},
+                CONF_THEN: [{"logger.log": {"message": "Then"}}],
+                CONF_ELSE: [
+                    {
+                        "repeat": {
+                            "count": 5,
+                            CONF_THEN: [
+                                {"logger.log": {"message": "Repeat"}},
+                                {"delay": "100ms"},
+                            ],
+                        }
+                    }
+                ],
+            }
+        }
+    ]
+    assert automation_is_synchronous(actions) is False
+
+
+def test_automation_is_synchronous_while_inside_repeat() -> None:
+    """Test while with wait_until inside repeat."""
+    actions = [
+        {
+            "repeat": {
+                "count": 3,
+                CONF_THEN: [
+                    {
+                        "while": {
+                            "condition": {"lambda": "return true;"},
+                            CONF_THEN: [
+                                {
+                                    "wait_until": {
+                                        "condition": {"lambda": "return false;"}
+                                    }
+                                }
+                            ],
+                        }
+                    }
+                ],
+            }
+        }
+    ]
+    assert automation_is_synchronous(actions) is False
+
+
+def test_automation_is_synchronous_script_wait() -> None:
+    """Test that script.wait (unknown action) is treated as async for safety."""
+    actions = [{"script.wait": {"id": "my_script"}}]
+    assert automation_is_synchronous(actions) is False
+
+
+def test_automation_is_synchronous_ble_client_connect() -> None:
+    """Test that ble_client.connect (unknown action) is treated as async for safety."""
+    actions = [{"ble_client.connect": {"id": "my_ble_client"}}]
+    assert automation_is_synchronous(actions) is False
+
+
+def test_automation_is_synchronous_ble_client_disconnect() -> None:
+    """Test that ble_client.disconnect (unknown action) is treated as async for safety."""
+    actions = [{"ble_client.disconnect": {"id": "my_ble_client"}}]
+    assert automation_is_synchronous(actions) is False
+
+
+def test_automation_is_synchronous_ble_client_write() -> None:
+    """Test that ble_client.ble_write (unknown action) is treated as async for safety."""
+    actions = [
+        {
+            "ble_client.ble_write": {
+                "id": "my_ble_client",
+                "service_uuid": "1234",
+                "characteristic_uuid": "5678",
+                "value": [0x01, 0x02],
+            }
+        }
+    ]
+    assert automation_is_synchronous(actions) is False
+
+
+def test_automation_is_synchronous_espnow_send() -> None:
+    """Test that espnow.send (unknown action) is treated as async for safety."""
+    actions = [
+        {
+            "espnow.send": {
+                "address": "AA:BB:CC:DD:EE:FF",
+                "data": [0x01, 0x02, 0x03],
+            }
+        }
+    ]
+    assert automation_is_synchronous(actions) is False
+
+
+def test_automation_is_synchronous_espnow_broadcast() -> None:
+    """Test that espnow.broadcast (unknown action) is treated as async for safety."""
+    actions = [{"espnow.broadcast": {"data": [0x01, 0x02, 0x03]}}]
+    assert automation_is_synchronous(actions) is False
+
+
+def test_automation_is_synchronous_nested_script_wait() -> None:
+    """Test that nested unknown action in if block makes automation async."""
+    actions = [
+        {
+            "if": {
+                "condition": {"lambda": "return true;"},
+                CONF_THEN: [{"logger.log": {"message": "Then"}}],
+                CONF_ELSE: [{"script.wait": {"id": "my_script"}}],
+            }
+        }
+    ]
+    assert automation_is_synchronous(actions) is False