preen

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/mipi_dsi/models/guition.py

@@ -101,225 +101,4 @@ DriverChip(
         (0xFF, 0x77, 0x01, 0x00, 0x00, 0x00),
     ]
 )
-
-# jc8012P4A1 Driver Configuration (jd9365)
-# Using parameters from esp_lcd_jd9365.h and the working full init sequence
-# ----------------------------------------------------------------------------------------------------------------------
-# * Resolution: 800x1280
-# * PCLK Frequency: 60 MHz
-# * DSI Lane Bit Rate: 1 Gbps (using 2-Lane DSI configuration)
-# * Horizontal Timing (hsync_pulse_width=20, hsync_back_porch=20, hsync_front_porch=40)
-# * Vertical Timing (vsync_pulse_width=4, vsync_back_porch=8, vsync_front_porch=20)
-# ----------------------------------------------------------------------------------------------------------------------
-DriverChip(
-    "JC8012P4A1",
-    width=800,
-    height=1280,
-    hsync_back_porch=20,
-    hsync_pulse_width=20,
-    hsync_front_porch=40,
-    vsync_back_porch=8,
-    vsync_pulse_width=4,
-    vsync_front_porch=20,
-    pclk_frequency="60MHz",
-    lane_bit_rate="1Gbps",
-    swap_xy=cv.UNDEFINED,
-    color_order="RGB",
-    reset_pin=27,
-    initsequence=[
-        (0xE0, 0x00),
-        (0xE1, 0x93),
-        (0xE2, 0x65),
-        (0xE3, 0xF8),
-        (0x80, 0x01),
-        (0xE0, 0x01),
-        (0x00, 0x00),
-        (0x01, 0x39),
-        (0x03, 0x10),
-        (0x04, 0x41),
-        (0x0C, 0x74),
-        (0x17, 0x00),
-        (0x18, 0xD7),
-        (0x19, 0x00),
-        (0x1A, 0x00),
-        (0x1B, 0xD7),
-        (0x1C, 0x00),
-        (0x24, 0xFE),
-        (0x35, 0x26),
-        (0x37, 0x69),
-        (0x38, 0x05),
-        (0x39, 0x06),
-        (0x3A, 0x08),
-        (0x3C, 0x78),
-        (0x3D, 0xFF),
-        (0x3E, 0xFF),
-        (0x3F, 0xFF),
-        (0x40, 0x06),
-        (0x41, 0xA0),
-        (0x43, 0x14),
-        (0x44, 0x0B),
-        (0x45, 0x30),
-        (0x4B, 0x04),
-        (0x55, 0x02),
-        (0x57, 0x89),
-        (0x59, 0x0A),
-        (0x5A, 0x28),
-        (0x5B, 0x15),
-        (0x5D, 0x50),
-        (0x5E, 0x37),
-        (0x5F, 0x29),
-        (0x60, 0x1E),
-        (0x61, 0x1D),
-        (0x62, 0x12),
-        (0x63, 0x1A),
-        (0x64, 0x08),
-        (0x65, 0x25),
-        (0x66, 0x26),
-        (0x67, 0x28),
-        (0x68, 0x49),
-        (0x69, 0x3A),
-        (0x6A, 0x43),
-        (0x6B, 0x3A),
-        (0x6C, 0x3B),
-        (0x6D, 0x32),
-        (0x6E, 0x1F),
-        (0x6F, 0x0E),
-        (0x70, 0x50),
-        (0x71, 0x37),
-        (0x72, 0x29),
-        (0x73, 0x1E),
-        (0x74, 0x1D),
-        (0x75, 0x12),
-        (0x76, 0x1A),
-        (0x77, 0x08),
-        (0x78, 0x25),
-        (0x79, 0x26),
-        (0x7A, 0x28),
-        (0x7B, 0x49),
-        (0x7C, 0x3A),
-        (0x7D, 0x43),
-        (0x7E, 0x3A),
-        (0x7F, 0x3B),
-        (0x80, 0x32),
-        (0x81, 0x1F),
-        (0x82, 0x0E),
-        (0xE0, 0x02),
-        (0x00, 0x1F),
-        (0x01, 0x1F),
-        (0x02, 0x52),
-        (0x03, 0x51),
-        (0x04, 0x50),
-        (0x05, 0x4B),
-        (0x06, 0x4A),
-        (0x07, 0x49),
-        (0x08, 0x48),
-        (0x09, 0x47),
-        (0x0A, 0x46),
-        (0x0B, 0x45),
-        (0x0C, 0x44),
-        (0x0D, 0x40),
-        (0x0E, 0x41),
-        (0x0F, 0x1F),
-        (0x10, 0x1F),
-        (0x11, 0x1F),
-        (0x12, 0x1F),
-        (0x13, 0x1F),
-        (0x14, 0x1F),
-        (0x15, 0x1F),
-        (0x16, 0x1F),
-        (0x17, 0x1F),
-        (0x18, 0x52),
-        (0x19, 0x51),
-        (0x1A, 0x50),
-        (0x1B, 0x4B),
-        (0x1C, 0x4A),
-        (0x1D, 0x49),
-        (0x1E, 0x48),
-        (0x1F, 0x47),
-        (0x20, 0x46),
-        (0x21, 0x45),
-        (0x22, 0x44),
-        (0x23, 0x40),
-        (0x24, 0x41),
-        (0x25, 0x1F),
-        (0x26, 0x1F),
-        (0x27, 0x1F),
-        (0x28, 0x1F),
-        (0x29, 0x1F),
-        (0x2A, 0x1F),
-        (0x2B, 0x1F),
-        (0x2C, 0x1F),
-        (0x2D, 0x1F),
-        (0x2E, 0x52),
-        (0x2F, 0x40),
-        (0x30, 0x41),
-        (0x31, 0x48),
-        (0x32, 0x49),
-        (0x33, 0x4A),
-        (0x34, 0x4B),
-        (0x35, 0x44),
-        (0x36, 0x45),
-        (0x37, 0x46),
-        (0x38, 0x47),
-        (0x39, 0x51),
-        (0x3A, 0x50),
-        (0x3B, 0x1F),
-        (0x3C, 0x1F),
-        (0x3D, 0x1F),
-        (0x3E, 0x1F),
-        (0x3F, 0x1F),
-        (0x40, 0x1F),
-        (0x41, 0x1F),
-        (0x42, 0x1F),
-        (0x43, 0x1F),
-        (0x44, 0x52),
-        (0x45, 0x40),
-        (0x46, 0x41),
-        (0x47, 0x48),
-        (0x48, 0x49),
-        (0x49, 0x4A),
-        (0x4A, 0x4B),
-        (0x4B, 0x44),
-        (0x4C, 0x45),
-        (0x4D, 0x46),
-        (0x4E, 0x47),
-        (0x4F, 0x51),
-        (0x50, 0x50),
-        (0x51, 0x1F),
-        (0x52, 0x1F),
-        (0x53, 0x1F),
-        (0x54, 0x1F),
-        (0x55, 0x1F),
-        (0x56, 0x1F),
-        (0x57, 0x1F),
-        (0x58, 0x40),
-        (0x59, 0x00),
-        (0x5A, 0x00),
-        (0x5B, 0x10),
-        (0x5C, 0x05),
-        (0x5D, 0x50),
-        (0x5E, 0x01),
-        (0x5F, 0x02),
-        (0x60, 0x50),
-        (0x61, 0x06),
-        (0x62, 0x04),
-        (0x63, 0x03),
-        (0x64, 0x64),
-        (0x65, 0x65),
-        (0x66, 0x0B),
-        (0x67, 0x73),
-        (0x68, 0x07),
-        (0x69, 0x06),
-        (0x6A, 0x64),
-        (0x6B, 0x08),
-        (0x6C, 0x00),
-        (0x6D, 0x32),
-        (0x6E, 0x08),
-        (0xE0, 0x04),
-        (0x2C, 0x6B),
-        (0x35, 0x08),
-        (0x37, 0x00),
-        (0xE0, 0x00),
-    ]
-)
 # fmt: on

esphome/core/helpers.cpp

@@ -487,26 +487,19 @@ static constexpr const char *BASE64_CHARS = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
                                             "abcdefghijklmnopqrstuvwxyz"
                                             "0123456789+/";
 
-// Helper function to find the index of a base64/base64url character in the lookup table.
+// Helper function to find the index of a base64 character in the lookup table.
 // Returns the character's position (0-63) if found, or 0 if not found.
-// Supports both standard base64 (+/) and base64url (-_) alphabets.
 // NOTE: This returns 0 for both 'A' (valid base64 char at index 0) and invalid characters.
 // This is safe because is_base64() is ALWAYS checked before calling this function,
 // preventing invalid characters from ever reaching here. The base64_decode function
 // stops processing at the first invalid character due to the is_base64() check in its
 // while loop condition, making this edge case harmless in practice.
 static inline uint8_t base64_find_char(char c) {
-  // Handle base64url variants: '-' maps to '+' (index 62), '_' maps to '/' (index 63)
-  if (c == '-')
-    return 62;
-  if (c == '_')
-    return 63;
   const char *pos = strchr(BASE64_CHARS, c);
   return pos ? (pos - BASE64_CHARS) : 0;
 }
 
-// Check if character is valid base64 or base64url
-static inline bool is_base64(char c) { return (isalnum(c) || (c == '+') || (c == '/') || (c == '-') || (c == '_')); }
+static inline bool is_base64(char c) { return (isalnum(c) || (c == '+') || (c == '/')); }
 
 std::string base64_encode(const std::vector<uint8_t> &buf) { return base64_encode(buf.data(), buf.size()); }
 

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/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