Merge remote-tracking branch 'origin/api-stringref-user-services' into integration

esphome/automation.py

@@ -57,8 +57,23 @@ 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,
+    *,
+    synchronous: bool = False,
+):
+    """Register an action type.
+
+    Actions default to ``synchronous=False`` (safe default), meaning string
+    arguments use owning std::string to prevent dangling references.
+
+    Set ``synchronous=True`` only for actions that complete synchronously
+    and never store trigger arguments for later execution.  This allows
+    the code generator to use non-owning StringRef for zero-copy access.
+    """
+    return ACTION_REGISTRY.register(name, action_type, schema, synchronous=synchronous)
 
 
 def register_condition(name: str, condition_type: MockObjClass, schema: cv.Schema):
@@ -335,7 +350,9 @@ 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),
 )
 async def delay_action_to_code(
     config: ConfigType,
@@ -366,6 +383,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),
     ),
+    synchronous=True,
 )
 async def if_action_to_code(
     config: ConfigType,
@@ -394,6 +412,7 @@ async def if_action_to_code(
             cv.Required(CONF_THEN): validate_action_list,
         }
     ),
+    synchronous=True,
 )
 async def while_action_to_code(
     config: ConfigType,
@@ -417,6 +436,7 @@ async def while_action_to_code(
             cv.Required(CONF_THEN): validate_action_list,
         }
     ),
+    synchronous=True,
 )
 async def repeat_action_to_code(
     config: ConfigType,
@@ -461,7 +481,12 @@ async def wait_until_action_to_code(
     return var
 
 
-@register_action("lambda", LambdaAction, cv.lambda_)
+# Lambda executes user C++ inline and returns — synchronous by execution model.
+# User code could theoretically store the StringRef for deferred use, but StringRef
+# is a view type and storing views beyond their scope is always unsafe regardless
+# of this optimization.  Marking non-synchronous would disable StringRef for nearly
+# all user services since most use lambda.
+@register_action("lambda", LambdaAction, cv.lambda_, synchronous=True)
 async def lambda_action_to_code(
     config: ConfigType,
     action_id: ID,
@@ -480,6 +505,7 @@ async def lambda_action_to_code(
             cv.Required(CONF_ID): cv.use_id(cg.PollingComponent),
         }
     ),
+    synchronous=True,
 )
 async def component_update_action_to_code(
     config: ConfigType,
@@ -499,6 +525,7 @@ async def component_update_action_to_code(
             cv.Required(CONF_ID): cv.use_id(cg.PollingComponent),
         }
     ),
+    synchronous=True,
 )
 async def component_suspend_action_to_code(
     config: ConfigType,
@@ -521,6 +548,7 @@ async def component_suspend_action_to_code(
             ),
         }
     ),
+    synchronous=True,
 )
 async def component_resume_action_to_code(
     config: ConfigType,
@@ -578,6 +606,27 @@ async def build_condition_list(
     return conditions
 
 
+def has_non_synchronous_actions(actions: ConfigType) -> bool:
+    """Check if a validated action list contains any non-synchronous actions.
+
+    Non-synchronous actions (delay, wait_until, script.wait, etc.) store
+    trigger args for later execution, making non-owning types like StringRef
+    unsafe.  Actions that haven't been audited default to non-synchronous.
+    """
+    if isinstance(actions, list):
+        return any(has_non_synchronous_actions(item) for item in actions)
+    if isinstance(actions, dict):
+        for key in actions:
+            if key in ACTION_REGISTRY and not ACTION_REGISTRY[key].synchronous:
+                return True
+        return any(
+            has_non_synchronous_actions(v)
+            for v in actions.values()
+            if isinstance(v, (list, dict))
+        )
+    return False
+
+
 async def build_automation(
     trigger: MockObj, args: TemplateArgsType, config: ConfigType
 ) -> MockObj:

esphome/components/api/__init__.py

@@ -76,7 +76,7 @@ SERVICE_ARG_NATIVE_TYPES: dict[str, MockObj] = {
     "bool": cg.bool_,
     "int": cg.int32,
     "float": cg.float_,
-    "string": cg.std_string,
+    "string": cg.StringRef,
     "bool[]": cg.FixedVector.template(cg.bool_).operator("const").operator("ref"),
     "int[]": cg.FixedVector.template(cg.int32).operator("const").operator("ref"),
     "float[]": cg.FixedVector.template(cg.float_).operator("const").operator("ref"),
@@ -380,9 +380,18 @@ async def to_code(config: ConfigType) -> None:
                 if is_optional:
                     func_args.append((cg.bool_, "return_response"))
 
+            # Check if action chain has non-synchronous actions that would make
+            # non-owning StringRef dangle (rx_buf_ reused after delay)
+            has_non_synchronous = automation.has_non_synchronous_actions(
+                conf.get(CONF_THEN, [])
+            )
+
             service_arg_names: list[str] = []
             for name, var_ in conf[CONF_VARIABLES].items():
                 native = SERVICE_ARG_NATIVE_TYPES[var_]
+                # Fall back to std::string for string args if non-synchronous actions exist
+                if has_non_synchronous and native is cg.StringRef:
+                    native = cg.std_string
                 service_template_args.append(native)
                 func_args.append((native, name))
                 service_arg_names.append(name)
@@ -509,11 +518,13 @@ HOMEASSISTANT_ACTION_ACTION_SCHEMA = cv.All(
     "homeassistant.action",
     HomeAssistantServiceCallAction,
     HOMEASSISTANT_ACTION_ACTION_SCHEMA,
+    synchronous=True,
 )
 @automation.register_action(
     "homeassistant.service",
     HomeAssistantServiceCallAction,
     HOMEASSISTANT_ACTION_ACTION_SCHEMA,
+    synchronous=True,
 )
 async def homeassistant_service_to_code(
     config: ConfigType,
@@ -604,6 +615,7 @@ HOMEASSISTANT_EVENT_ACTION_SCHEMA = cv.Schema(
     "homeassistant.event",
     HomeAssistantServiceCallAction,
     HOMEASSISTANT_EVENT_ACTION_SCHEMA,
+    synchronous=True,
 )
 async def homeassistant_event_to_code(config, action_id, template_arg, args):
     cg.add_define("USE_API_HOMEASSISTANT_SERVICES")
@@ -644,6 +656,7 @@ HOMEASSISTANT_TAG_SCANNED_ACTION_SCHEMA = cv.maybe_simple_value(
     "homeassistant.tag_scanned",
     HomeAssistantServiceCallAction,
     HOMEASSISTANT_TAG_SCANNED_ACTION_SCHEMA,
+    synchronous=True,
 )
 async def homeassistant_tag_scanned_to_code(config, action_id, template_arg, args):
     cg.add_define("USE_API_HOMEASSISTANT_SERVICES")

esphome/components/api/api_connection.cpp

@@ -1702,37 +1702,42 @@ void APIConnection::on_home_assistant_state_response(const HomeAssistantStateRes
     return;
   }
 
+  // Null-terminate state in-place for safe c_str() usage (e.g., parse_number in callbacks).
+  // Safe: decode is complete, byte after string data was already consumed during parse,
+  // and frame helpers reserve RX_BUF_NULL_TERMINATOR extra byte in rx_buf_.
+  // const_cast is safe: msg references mutable rx_buf_ data; the const& handler
+  // signature is a generated protobuf pattern, not a true immutability contract.
+  if (!msg.state.empty()) {
+    const_cast<char *>(msg.state.c_str())[msg.state.size()] = '\0';
+  }
+
   for (auto &it : this->parent_->get_state_subs()) {
-    // Compare entity_id: check length matches and content matches
-    size_t entity_id_len = strlen(it.entity_id);
-    if (entity_id_len != msg.entity_id.size() ||
-        memcmp(it.entity_id, msg.entity_id.c_str(), msg.entity_id.size()) != 0) {
+    if (msg.entity_id != it.entity_id) {
       continue;
     }
 
-    // Compare attribute: either both have matching attribute, or both have none
-    size_t sub_attr_len = it.attribute != nullptr ? strlen(it.attribute) : 0;
-    if (sub_attr_len != msg.attribute.size() ||
-        (sub_attr_len > 0 && memcmp(it.attribute, msg.attribute.c_str(), sub_attr_len) != 0)) {
+    // If subscriber has attribute filter (non-null), message attribute must match it;
+    // if subscriber has no filter (nullptr), message must have no attribute.
+    if (it.attribute != nullptr ? msg.attribute != it.attribute : !msg.attribute.empty()) {
       continue;
     }
 
-    // Create null-terminated state for callback (parse_number needs null-termination)
-    // HA state max length is 255 characters, but attributes can be much longer
-    // Use stack buffer for common case (states), heap fallback for large attributes
-    size_t state_len = msg.state.size();
-    SmallBufferWithHeapFallback<MAX_STATE_LEN + 1> state_buf_alloc(state_len + 1);
-    char *state_buf = reinterpret_cast<char *>(state_buf_alloc.get());
-    if (state_len > 0) {
-      memcpy(state_buf, msg.state.c_str(), state_len);
-    }
-    state_buf[state_len] = '\0';
-    it.callback(StringRef(state_buf, state_len));
+    it.callback(msg.state);
   }
 }
 #endif
 #ifdef USE_API_USER_DEFINED_ACTIONS
 void APIConnection::on_execute_service_request(const ExecuteServiceRequest &msg) {
+  // Null-terminate string args in-place for safe c_str() usage in YAML service triggers.
+  // Safe: full ExecuteServiceRequest decode is complete, all bytes in rx_buf_ consumed,
+  // and frame helpers reserve RX_BUF_NULL_TERMINATOR extra byte for the last field.
+  // const_cast is safe: msg references mutable rx_buf_ data; the const& handler
+  // signature is a generated protobuf pattern, not a true immutability contract.
+  for (auto &arg : const_cast<ExecuteServiceRequest &>(msg).args) {
+    if (!arg.string_.empty()) {
+      const_cast<char *>(arg.string_.c_str())[arg.string_.size()] = '\0';
+    }
+  }
   bool found = false;
 #ifdef USE_API_USER_DEFINED_ACTION_RESPONSES
   // Register the call and get a unique server-generated action_call_id

esphome/components/api/api_frame_helper.h

@@ -29,6 +29,10 @@ static constexpr uint16_t MAX_MESSAGE_SIZE = 8192;  // 8 KiB for ESP8266
 static constexpr uint16_t MAX_MESSAGE_SIZE = 32768;  // 32 KiB for ESP32 and other platforms
 #endif
 
+// Extra byte reserved in rx_buf_ beyond the message size so protobuf
+// StringRef fields can be null-terminated in-place after decode.
+static constexpr uint16_t RX_BUF_NULL_TERMINATOR = 1;
+
 // Maximum number of messages to batch in a single write operation
 // Must be >= MAX_INITIAL_PER_BATCH in api_connection.h (enforced by static_assert there)
 static constexpr size_t MAX_MESSAGES_PER_BATCH = 34;

esphome/components/api/api_frame_helper_noise.cpp

@@ -194,16 +194,21 @@ APIError APINoiseFrameHelper::try_read_frame_() {
   uint16_t msg_size = (((uint16_t) rx_header_buf_[1]) << 8) | rx_header_buf_[2];
 
   // Check against size limits to prevent OOM: MAX_HANDSHAKE_SIZE for handshake, MAX_MESSAGE_SIZE for data
-  uint16_t limit = (state_ == State::DATA) ? MAX_MESSAGE_SIZE : MAX_HANDSHAKE_SIZE;
+  bool is_data = (state_ == State::DATA);
+  uint16_t limit = is_data ? MAX_MESSAGE_SIZE : MAX_HANDSHAKE_SIZE;
   if (msg_size > limit) {
     state_ = State::FAILED;
     HELPER_LOG("Bad packet: message size %u exceeds maximum %u", msg_size, limit);
-    return (state_ == State::DATA) ? APIError::BAD_DATA_PACKET : APIError::BAD_HANDSHAKE_PACKET_LEN;
+    return is_data ? APIError::BAD_DATA_PACKET : APIError::BAD_HANDSHAKE_PACKET_LEN;
   }
 
-  // Reserve space for body
-  if (this->rx_buf_.size() != msg_size) {
-    this->rx_buf_.resize(msg_size);
+  // Reserve space for body (+ null terminator in DATA state so protobuf
+  // StringRef fields can be safely null-terminated in-place after decode.
+  // During handshake, rx_buf_.size() is used in prologue construction, so
+  // the buffer must be exactly msg_size to avoid prologue mismatch.)
+  uint16_t alloc_size = msg_size + (is_data ? RX_BUF_NULL_TERMINATOR : 0);
+  if (this->rx_buf_.size() != alloc_size) {
+    this->rx_buf_.resize(alloc_size);
   }
 
   if (rx_buf_len_ < msg_size) {
@@ -407,7 +412,18 @@ APIError APINoiseFrameHelper::read_packet(ReadPacketBuffer *buffer) {
 
   NoiseBuffer mbuf;
   noise_buffer_init(mbuf);
-  noise_buffer_set_inout(mbuf, this->rx_buf_.data(), this->rx_buf_.size(), this->rx_buf_.size());
+  // read_packet() must only be called in DATA state; the extra
+  // RX_BUF_NULL_TERMINATOR byte is only allocated in DATA state
+  // (see try_read_frame_), so calling this during handshake would
+  // underflow the size calculation below.
+#ifdef ESPHOME_DEBUG_API
+  assert(this->state_ == State::DATA);
+#endif
+  // rx_buf_ has RX_BUF_NULL_TERMINATOR extra byte for null termination
+  // (only added in DATA state — see try_read_frame_), so subtract it
+  // to get the actual encrypted data size for decryption.
+  size_t encrypted_size = this->rx_buf_.size() - RX_BUF_NULL_TERMINATOR;
+  noise_buffer_set_inout(mbuf, this->rx_buf_.data(), encrypted_size, encrypted_size);
   int err = noise_cipherstate_decrypt(this->recv_cipher_, &mbuf);
   APIError decrypt_err =
       handle_noise_error_(err, LOG_STR("noise_cipherstate_decrypt"), APIError::CIPHERSTATE_DECRYPT_FAILED);
@@ -574,7 +590,9 @@ APIError APINoiseFrameHelper::init_handshake_() {
 }
 
 APIError APINoiseFrameHelper::check_handshake_finished_() {
+#ifdef ESPHOME_DEBUG_API
   assert(state_ == State::HANDSHAKE);
+#endif
 
   int action = noise_handshakestate_get_action(handshake_);
   if (action == NOISE_ACTION_READ_MESSAGE || action == NOISE_ACTION_WRITE_MESSAGE)

esphome/components/api/api_frame_helper_plaintext.cpp

@@ -163,9 +163,10 @@ APIError APIPlaintextFrameHelper::try_read_frame_() {
   }
   // header reading done
 
-  // Reserve space for body
-  if (this->rx_buf_.size() != this->rx_header_parsed_len_) {
-    this->rx_buf_.resize(this->rx_header_parsed_len_);
+  // Reserve space for body (+ null terminator so protobuf StringRef fields
+  // can be safely null-terminated in-place after decode)
+  if (this->rx_buf_.size() != this->rx_header_parsed_len_ + RX_BUF_NULL_TERMINATOR) {
+    this->rx_buf_.resize(this->rx_header_parsed_len_ + RX_BUF_NULL_TERMINATOR);
   }
 
   if (rx_buf_len_ < rx_header_parsed_len_) {

esphome/components/api/user_services.cpp

@@ -1,5 +1,6 @@
 #include "user_services.h"
 #include "esphome/core/log.h"
+#include "esphome/core/string_ref.h"
 
 namespace esphome::api {
 
@@ -11,6 +12,8 @@ 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 StringRef version for YAML-generated services (string_ is null-terminated after decode)
+template<> StringRef get_execute_arg_value<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) {
@@ -61,6 +64,8 @@ template<> enums::ServiceArgType to_service_arg_type<bool>() { return enums::SER
 template<> enums::ServiceArgType to_service_arg_type<int32_t>() { return enums::SERVICE_ARG_TYPE_INT; }
 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 StringRef version for YAML-generated services
+template<> enums::ServiceArgType to_service_arg_type<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; }

esphome/components/button/__init__.py

@@ -123,7 +123,9 @@ BUTTON_PRESS_SCHEMA = maybe_simple_id(
 )
 
 
-@automation.register_action("button.press", PressAction, BUTTON_PRESS_SCHEMA)
+@automation.register_action(
+    "button.press", PressAction, BUTTON_PRESS_SCHEMA, synchronous=True
+)
 async def button_press_to_code(config, action_id, template_arg, args):
     paren = await cg.get_variable(config[CONF_ID])
     return cg.new_Pvariable(action_id, template_arg, paren)

esphome/components/cover/__init__.py

@@ -248,25 +248,33 @@ COVER_ACTION_SCHEMA = maybe_simple_id(
 )
 
 
-@automation.register_action("cover.open", OpenAction, COVER_ACTION_SCHEMA)
+@automation.register_action(
+    "cover.open", OpenAction, COVER_ACTION_SCHEMA, synchronous=True
+)
 async def cover_open_to_code(config, action_id, template_arg, args):
     paren = await cg.get_variable(config[CONF_ID])
     return cg.new_Pvariable(action_id, template_arg, paren)
 
 
-@automation.register_action("cover.close", CloseAction, COVER_ACTION_SCHEMA)
+@automation.register_action(
+    "cover.close", CloseAction, COVER_ACTION_SCHEMA, synchronous=True
+)
 async def cover_close_to_code(config, action_id, template_arg, args):
     paren = await cg.get_variable(config[CONF_ID])
     return cg.new_Pvariable(action_id, template_arg, paren)
 
 
-@automation.register_action("cover.stop", StopAction, COVER_ACTION_SCHEMA)
+@automation.register_action(
+    "cover.stop", StopAction, COVER_ACTION_SCHEMA, synchronous=True
+)
 async def cover_stop_to_code(config, action_id, template_arg, args):
     paren = await cg.get_variable(config[CONF_ID])
     return cg.new_Pvariable(action_id, template_arg, paren)
 
 
-@automation.register_action("cover.toggle", ToggleAction, COVER_ACTION_SCHEMA)
+@automation.register_action(
+    "cover.toggle", ToggleAction, COVER_ACTION_SCHEMA, synchronous=True
+)
 async def cover_toggle_to_code(config, action_id, template_arg, args):
     paren = await cg.get_variable(config[CONF_ID])
     return cg.new_Pvariable(action_id, template_arg, paren)
@@ -283,7 +291,9 @@ COVER_CONTROL_ACTION_SCHEMA = cv.Schema(
 )
 
 
-@automation.register_action("cover.control", ControlAction, COVER_CONTROL_ACTION_SCHEMA)
+@automation.register_action(
+    "cover.control", ControlAction, COVER_CONTROL_ACTION_SCHEMA, synchronous=True
+)
 async def cover_control_to_code(config, action_id, template_arg, args):
     paren = await cg.get_variable(config[CONF_ID])
     var = cg.new_Pvariable(action_id, template_arg, paren)

esphome/components/fan/__init__.py

@@ -311,13 +311,17 @@ FAN_ACTION_SCHEMA = maybe_simple_id(
 )
 
 
-@automation.register_action("fan.toggle", ToggleAction, FAN_ACTION_SCHEMA)
+@automation.register_action(
+    "fan.toggle", ToggleAction, FAN_ACTION_SCHEMA, synchronous=True
+)
 async def fan_toggle_to_code(config, action_id, template_arg, args):
     paren = await cg.get_variable(config[CONF_ID])
     return cg.new_Pvariable(action_id, template_arg, paren)
 
 
-@automation.register_action("fan.turn_off", TurnOffAction, FAN_ACTION_SCHEMA)
+@automation.register_action(
+    "fan.turn_off", TurnOffAction, FAN_ACTION_SCHEMA, synchronous=True
+)
 async def fan_turn_off_to_code(config, action_id, template_arg, args):
     paren = await cg.get_variable(config[CONF_ID])
     return cg.new_Pvariable(action_id, template_arg, paren)
@@ -336,6 +340,7 @@ async def fan_turn_off_to_code(config, action_id, template_arg, args):
             ),
         }
     ),
+    synchronous=True,
 )
 async def fan_turn_on_to_code(config, action_id, template_arg, args):
     paren = await cg.get_variable(config[CONF_ID])

esphome/components/globals/__init__.py

@@ -102,6 +102,7 @@ async def to_code(config):
             cv.Required(CONF_VALUE): cv.templatable(cv.string_strict),
         }
     ),
+    synchronous=True,
 )
 async def globals_set_to_code(config, action_id, template_arg, args):
     full_id, paren = await cg.get_variable_with_full_id(config[CONF_ID])

esphome/components/light/automation.py

@@ -51,6 +51,7 @@ from .types import (
             ),
         }
     ),
+    synchronous=True,
 )
 async def light_toggle_to_code(config, action_id, template_arg, args):
     paren = await cg.get_variable(config[CONF_ID])
@@ -111,13 +112,13 @@ LIGHT_TURN_ON_ACTION_SCHEMA = automation.maybe_simple_id(
 
 
 @automation.register_action(
-    "light.turn_off", LightControlAction, LIGHT_TURN_OFF_ACTION_SCHEMA
+    "light.turn_off", LightControlAction, LIGHT_TURN_OFF_ACTION_SCHEMA, synchronous=True
 )
 @automation.register_action(
-    "light.turn_on", LightControlAction, LIGHT_TURN_ON_ACTION_SCHEMA
+    "light.turn_on", LightControlAction, LIGHT_TURN_ON_ACTION_SCHEMA, synchronous=True
 )
 @automation.register_action(
-    "light.control", LightControlAction, LIGHT_CONTROL_ACTION_SCHEMA
+    "light.control", LightControlAction, LIGHT_CONTROL_ACTION_SCHEMA, synchronous=True
 )
 async def light_control_to_code(config, action_id, template_arg, args):
     paren = await cg.get_variable(config[CONF_ID])
@@ -193,7 +194,10 @@ LIGHT_DIM_RELATIVE_ACTION_SCHEMA = cv.Schema(
 
 
 @automation.register_action(
-    "light.dim_relative", DimRelativeAction, LIGHT_DIM_RELATIVE_ACTION_SCHEMA
+    "light.dim_relative",
+    DimRelativeAction,
+    LIGHT_DIM_RELATIVE_ACTION_SCHEMA,
+    synchronous=True,
 )
 async def light_dim_relative_to_code(config, action_id, template_arg, args):
     paren = await cg.get_variable(config[CONF_ID])

esphome/components/logger/__init__.py

@@ -518,7 +518,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, synchronous=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]]

esphome/components/mqtt/__init__.py

@@ -492,7 +492,7 @@ MQTT_PUBLISH_ACTION_SCHEMA = cv.Schema(
 
 
 @automation.register_action(
-    "mqtt.publish", MQTTPublishAction, MQTT_PUBLISH_ACTION_SCHEMA
+    "mqtt.publish", MQTTPublishAction, MQTT_PUBLISH_ACTION_SCHEMA, synchronous=True
 )
 async def mqtt_publish_action_to_code(config, action_id, template_arg, args):
     paren = await cg.get_variable(config[CONF_ID])
@@ -521,7 +521,10 @@ MQTT_PUBLISH_JSON_ACTION_SCHEMA = cv.Schema(
 
 
 @automation.register_action(
-    "mqtt.publish_json", MQTTPublishJsonAction, MQTT_PUBLISH_JSON_ACTION_SCHEMA
+    "mqtt.publish_json",
+    MQTTPublishJsonAction,
+    MQTT_PUBLISH_JSON_ACTION_SCHEMA,
+    synchronous=True,
 )
 async def mqtt_publish_json_action_to_code(config, action_id, template_arg, args):
     paren = await cg.get_variable(config[CONF_ID])

esphome/components/number/__init__.py

@@ -347,6 +347,7 @@ OPERATION_BASE_SCHEMA = cv.Schema(
             cv.Required(CONF_VALUE): cv.templatable(cv.float_),
         }
     ),
+    synchronous=True,
 )
 async def number_set_to_code(config, action_id, template_arg, args):
     paren = await cg.get_variable(config[CONF_ID])
@@ -369,6 +370,7 @@ async def number_set_to_code(config, action_id, template_arg, args):
             }
         )
     ),
+    synchronous=True,
 )
 @automation.register_action(
     "number.decrement",
@@ -383,6 +385,7 @@ async def number_set_to_code(config, action_id, template_arg, args):
             }
         )
     ),
+    synchronous=True,
 )
 @automation.register_action(
     "number.to_min",
@@ -396,6 +399,7 @@ async def number_set_to_code(config, action_id, template_arg, args):
             }
         )
     ),
+    synchronous=True,
 )
 @automation.register_action(
     "number.to_max",
@@ -409,6 +413,7 @@ async def number_set_to_code(config, action_id, template_arg, args):
             }
         )
     ),
+    synchronous=True,
 )
 @automation.register_action(
     "number.operation",
@@ -421,6 +426,7 @@ async def number_set_to_code(config, action_id, template_arg, args):
             cv.Optional(CONF_CYCLE, default=True): cv.templatable(cv.boolean),
         }
     ),
+    synchronous=True,
 )
 async def number_to_to_code(config, action_id, template_arg, args):
     paren = await cg.get_variable(config[CONF_ID])

esphome/components/output/__init__.py

@@ -74,14 +74,16 @@ BINARY_OUTPUT_ACTION_SCHEMA = maybe_simple_id(
 )
 
 
-@automation.register_action("output.turn_on", TurnOnAction, BINARY_OUTPUT_ACTION_SCHEMA)
+@automation.register_action(
+    "output.turn_on", TurnOnAction, BINARY_OUTPUT_ACTION_SCHEMA, synchronous=True
+)
 async def output_turn_on_to_code(config, action_id, template_arg, args):
     paren = await cg.get_variable(config[CONF_ID])
     return cg.new_Pvariable(action_id, template_arg, paren)
 
 
 @automation.register_action(
-    "output.turn_off", TurnOffAction, BINARY_OUTPUT_ACTION_SCHEMA
+    "output.turn_off", TurnOffAction, BINARY_OUTPUT_ACTION_SCHEMA, synchronous=True
 )
 async def output_turn_off_to_code(config, action_id, template_arg, args):
     paren = await cg.get_variable(config[CONF_ID])
@@ -97,6 +99,7 @@ async def output_turn_off_to_code(config, action_id, template_arg, args):
             cv.Required(CONF_LEVEL): cv.templatable(cv.percentage),
         }
     ),
+    synchronous=True,
 )
 async def output_set_level_to_code(config, action_id, template_arg, args):
     paren = await cg.get_variable(config[CONF_ID])

esphome/components/script/__init__.py

@@ -160,6 +160,7 @@ async def to_code(config):
             cv.Optional(validate_parameter_name): cv.templatable(cv.valid),
         },
     ),
+    synchronous=True,
 )
 async def script_execute_action_to_code(config, action_id, template_arg, args):
     def convert(type: str):
@@ -208,6 +209,7 @@ async def script_execute_action_to_code(config, action_id, template_arg, args):
     "script.stop",
     ScriptStopAction,
     maybe_simple_id({cv.Required(CONF_ID): cv.use_id(Script)}),
+    synchronous=True,
 )
 async def script_stop_action_to_code(config, action_id, template_arg, args):
     full_id, paren = await cg.get_variable_with_full_id(config[CONF_ID])

esphome/components/select/__init__.py

@@ -145,6 +145,7 @@ OPERATION_BASE_SCHEMA = cv.Schema(
             cv.Required(CONF_OPTION): cv.templatable(cv.string_strict),
         }
     ),
+    synchronous=True,
 )
 async def select_set_to_code(config, action_id, template_arg, args):
     paren = await cg.get_variable(config[CONF_ID])
@@ -162,6 +163,7 @@ async def select_set_to_code(config, action_id, template_arg, args):
             cv.Required(CONF_INDEX): cv.templatable(cv.positive_int),
         }
     ),
+    synchronous=True,
 )
 async def select_set_index_to_code(config, action_id, template_arg, args):
     paren = await cg.get_variable(config[CONF_ID])
@@ -217,6 +219,7 @@ async def select_is_to_code(config, condition_id, template_arg, args):
             cv.Optional(CONF_CYCLE, default=True): cv.templatable(cv.boolean),
         }
     ),
+    synchronous=True,
 )
 @automation.register_action(
     "select.next",
@@ -229,6 +232,7 @@ async def select_is_to_code(config, condition_id, template_arg, args):
             }
         )
     ),
+    synchronous=True,
 )
 @automation.register_action(
     "select.previous",
@@ -243,6 +247,7 @@ async def select_is_to_code(config, condition_id, template_arg, args):
             }
         )
     ),
+    synchronous=True,
 )
 @automation.register_action(
     "select.first",
@@ -254,6 +259,7 @@ async def select_is_to_code(config, condition_id, template_arg, args):
             }
         )
     ),
+    synchronous=True,
 )
 @automation.register_action(
     "select.last",
@@ -265,6 +271,7 @@ async def select_is_to_code(config, condition_id, template_arg, args):
             }
         )
     ),
+    synchronous=True,
 )
 async def select_operation_to_code(config, action_id, template_arg, args):
     paren = await cg.get_variable(config[CONF_ID])

esphome/components/switch/__init__.py

@@ -198,7 +198,7 @@ SWITCH_CONTROL_ACTION_SCHEMA = automation.maybe_simple_id(
 
 
 @automation.register_action(
-    "switch.control", ControlAction, SWITCH_CONTROL_ACTION_SCHEMA
+    "switch.control", ControlAction, SWITCH_CONTROL_ACTION_SCHEMA, synchronous=True
 )
 async def switch_control_to_code(config, action_id, template_arg, args):
     paren = await cg.get_variable(config[CONF_ID])
@@ -208,9 +208,15 @@ async def switch_control_to_code(config, action_id, template_arg, args):
     return var
 
 
-@automation.register_action("switch.toggle", ToggleAction, SWITCH_ACTION_SCHEMA)
-@automation.register_action("switch.turn_off", TurnOffAction, SWITCH_ACTION_SCHEMA)
-@automation.register_action("switch.turn_on", TurnOnAction, SWITCH_ACTION_SCHEMA)
+@automation.register_action(
+    "switch.toggle", ToggleAction, SWITCH_ACTION_SCHEMA, synchronous=True
+)
+@automation.register_action(
+    "switch.turn_off", TurnOffAction, SWITCH_ACTION_SCHEMA, synchronous=True
+)
+@automation.register_action(
+    "switch.turn_on", TurnOnAction, SWITCH_ACTION_SCHEMA, synchronous=True
+)
 async def switch_toggle_to_code(config, action_id, template_arg, args):
     paren = await cg.get_variable(config[CONF_ID])
     return cg.new_Pvariable(action_id, template_arg, paren)

esphome/components/text/__init__.py

@@ -164,6 +164,7 @@ OPERATION_BASE_SCHEMA = cv.Schema(
             cv.Required(CONF_VALUE): cv.templatable(cv.string_strict),
         }
     ),
+    synchronous=True,
 )
 async def text_set_to_code(config, action_id, template_arg, args):
     paren = await cg.get_variable(config[CONF_ID])

esphome/core/string_ref.h

@@ -81,6 +81,20 @@ class StringRef {
 
   operator std::string() const { return str(); }
 
+  /// Compare (compatible with std::string::compare)
+  int compare(const StringRef &other) const {
+    int result = std::memcmp(base_, other.base_, std::min(len_, other.len_));
+    if (result != 0)
+      return result;
+    if (len_ < other.len_)
+      return -1;
+    if (len_ > other.len_)
+      return 1;
+    return 0;
+  }
+  int compare(const char *s) const { return compare(StringRef(s)); }
+  int compare(const std::string &s) const { return compare(StringRef(s)); }
+
   /// Find first occurrence of substring, returns std::string::npos if not found.
   /// Note: Requires the underlying string to be null-terminated.
   size_type find(const char *s, size_type pos = 0) const {

esphome/platformio_api.py

@@ -5,6 +5,7 @@ import os
 from pathlib import Path
 import re
 import subprocess
+import time
 from typing import Any
 
 from esphome.const import CONF_COMPILE_PROCESS_LIMIT, CONF_ESPHOME, KEY_CORE
@@ -44,31 +45,61 @@ def patch_structhash():
 
 
 def patch_file_downloader():
-    """Patch PlatformIO's FileDownloader to retry on PackageException errors."""
+    """Patch PlatformIO's FileDownloader to retry on PackageException errors.
+
+    PlatformIO's FileDownloader uses HTTPSession which lacks built-in retry
+    for 502/503 errors. We add retries with exponential backoff and close the
+    session between attempts to force a fresh TCP connection, which may route
+    to a different CDN edge node.
+    """
     from platformio.package.download import FileDownloader
     from platformio.package.exception import PackageException
 
+    if getattr(FileDownloader.__init__, "_esphome_patched", False):
+        return
+
     original_init = FileDownloader.__init__
 
     def patched_init(self, *args: Any, **kwargs: Any) -> None:
-        max_retries = 3
+        max_retries = 5
 
         for attempt in range(max_retries):
             try:
-                return original_init(self, *args, **kwargs)
+                original_init(self, *args, **kwargs)
+                return
             except PackageException as e:
                 if attempt < max_retries - 1:
+                    # Exponential backoff: 2, 4, 8, 16 seconds
+                    delay = 2 ** (attempt + 1)
                     _LOGGER.warning(
-                        "Package download failed: %s. Retrying... (attempt %d/%d)",
+                        "Package download failed: %s. "
+                        "Retrying in %d seconds... (attempt %d/%d)",
                         str(e),
+                        delay,
                         attempt + 1,
                         max_retries,
                     )
+                    # Close the response and session to free resources
+                    # and force a new TCP connection on retry, which may
+                    # route to a different CDN edge node
+                    # pylint: disable=protected-access,broad-except
+                    try:
+                        if (
+                            hasattr(self, "_http_response")
+                            and self._http_response is not None
+                        ):
+                            self._http_response.close()
+                        if hasattr(self, "_http_session"):
+                            self._http_session.close()
+                    except Exception:
+                        pass
+                    # pylint: enable=protected-access,broad-except
+                    time.sleep(delay)
                 else:
                     # Final attempt - re-raise
                     raise
-        return None
 
+    patched_init._esphome_patched = True  # type: ignore[attr-defined]  # pylint: disable=protected-access
     FileDownloader.__init__ = patched_init
 
 

esphome/util.py

@@ -24,11 +24,14 @@ class RegistryEntry:
         fun: Callable[..., Any],
         type_id: "MockObjClass",
         schema: "Schema",
+        *,
+        synchronous: bool = False,
     ):
         self.name = name
         self.fun = fun
         self.type_id = type_id
         self.raw_schema = schema
+        self.synchronous = synchronous
 
     @property
     def coroutine_fun(self):
@@ -49,9 +52,18 @@ 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",
+        *,
+        synchronous: bool = False,
+    ):
         def decorator(fun: Callable[..., Any]):
-            self[name] = RegistryEntry(name, fun, type_id, schema)
+            self[name] = RegistryEntry(
+                name, fun, type_id, schema, synchronous=synchronous
+            )
             return fun
 
         return decorator

tests/unit_tests/test_automation.py

@@ -0,0 +1,177 @@
+"""Tests for esphome.automation module."""
+
+from collections.abc import Generator
+from unittest.mock import patch
+
+import pytest
+
+from esphome.automation import has_non_synchronous_actions
+from esphome.util import RegistryEntry
+
+
+def _make_registry(non_synchronous_actions: set[str]) -> dict[str, RegistryEntry]:
+    """Create a mock ACTION_REGISTRY with specified non-synchronous actions.
+
+    Uses the default synchronous=False, matching the real registry behavior.
+    """
+    registry: dict[str, RegistryEntry] = {}
+    for name in non_synchronous_actions:
+        registry[name] = RegistryEntry(name, lambda: None, None, None)
+    return registry
+
+
+@pytest.fixture
+def mock_registry() -> Generator[dict[str, RegistryEntry]]:
+    """Fixture that patches ACTION_REGISTRY with delay, wait_until, script.wait as non-synchronous."""
+    registry: dict[str, RegistryEntry] = _make_registry(
+        {"delay", "wait_until", "script.wait"}
+    )
+    registry["logger.log"] = RegistryEntry(
+        "logger.log", lambda: None, None, None, synchronous=True
+    )
+    with patch("esphome.automation.ACTION_REGISTRY", registry):
+        yield registry
+
+
+def test_has_non_synchronous_actions_empty_list(
+    mock_registry: dict[str, RegistryEntry],
+) -> None:
+    assert has_non_synchronous_actions([]) is False
+
+
+def test_has_non_synchronous_actions_empty_dict(
+    mock_registry: dict[str, RegistryEntry],
+) -> None:
+    assert has_non_synchronous_actions({}) is False
+
+
+def test_has_non_synchronous_actions_non_dict_non_list(
+    mock_registry: dict[str, RegistryEntry],
+) -> None:
+    assert has_non_synchronous_actions("string") is False
+    assert has_non_synchronous_actions(42) is False
+    assert has_non_synchronous_actions(None) is False
+
+
+def test_has_non_synchronous_actions_delay(
+    mock_registry: dict[str, RegistryEntry],
+) -> None:
+    assert has_non_synchronous_actions([{"delay": "1s"}]) is True
+
+
+def test_has_non_synchronous_actions_wait_until(
+    mock_registry: dict[str, RegistryEntry],
+) -> None:
+    assert has_non_synchronous_actions([{"wait_until": {"condition": {}}}]) is True
+
+
+def test_has_non_synchronous_actions_script_wait(
+    mock_registry: dict[str, RegistryEntry],
+) -> None:
+    assert has_non_synchronous_actions([{"script.wait": "script_id"}]) is True
+
+
+def test_has_non_synchronous_actions_synchronous(
+    mock_registry: dict[str, RegistryEntry],
+) -> None:
+    assert has_non_synchronous_actions([{"logger.log": "hello"}]) is False
+
+
+def test_has_non_synchronous_actions_unknown_not_in_registry(
+    mock_registry: dict[str, RegistryEntry],
+) -> None:
+    """Unknown actions not in registry are not flagged (only registered actions count)."""
+    assert has_non_synchronous_actions([{"unknown.action": "value"}]) is False
+
+
+def test_has_non_synchronous_actions_default_non_synchronous(
+    mock_registry: dict[str, RegistryEntry],
+) -> None:
+    """Actions registered without explicit synchronous=True default to non-synchronous."""
+    mock_registry["some.action"] = RegistryEntry(
+        "some.action", lambda: None, None, None
+    )
+    assert has_non_synchronous_actions([{"some.action": "value"}]) is True
+
+
+def test_has_non_synchronous_actions_nested_in_then(
+    mock_registry: dict[str, RegistryEntry],
+) -> None:
+    """Non-synchronous action nested inside a synchronous action's then block."""
+    actions: list[dict[str, object]] = [
+        {
+            "logger.log": "first",
+            "then": [{"delay": "1s"}],
+        }
+    ]
+    assert has_non_synchronous_actions(actions) is True
+
+
+def test_has_non_synchronous_actions_deeply_nested(
+    mock_registry: dict[str, RegistryEntry],
+) -> None:
+    """Non-synchronous action deeply nested in action structure."""
+    actions: list[dict[str, object]] = [
+        {
+            "if": {
+                "then": [
+                    {"logger.log": "hello"},
+                    {"delay": "500ms"},
+                ]
+            }
+        }
+    ]
+    assert has_non_synchronous_actions(actions) is True
+
+
+def test_has_non_synchronous_actions_none_in_nested(
+    mock_registry: dict[str, RegistryEntry],
+) -> None:
+    """No non-synchronous actions even with nesting."""
+    actions: list[dict[str, object]] = [
+        {
+            "if": {
+                "then": [
+                    {"logger.log": "hello"},
+                ]
+            }
+        }
+    ]
+    assert has_non_synchronous_actions(actions) is False
+
+
+def test_has_non_synchronous_actions_multiple_one_non_synchronous(
+    mock_registry: dict[str, RegistryEntry],
+) -> None:
+    assert (
+        has_non_synchronous_actions(
+            [
+                {"logger.log": "first"},
+                {"delay": "1s"},
+                {"logger.log": "second"},
+            ]
+        )
+        is True
+    )
+
+
+def test_has_non_synchronous_actions_multiple_all_synchronous(
+    mock_registry: dict[str, RegistryEntry],
+) -> None:
+    assert (
+        has_non_synchronous_actions(
+            [
+                {"logger.log": "first"},
+                {"logger.log": "second"},
+            ]
+        )
+        is False
+    )
+
+
+def test_has_non_synchronous_actions_dict_input(
+    mock_registry: dict[str, RegistryEntry],
+) -> None:
+    """Direct dict input (single action)."""
+    assert has_non_synchronous_actions({"delay": "1s"}) is True
+    assert has_non_synchronous_actions({"logger.log": "hello"}) is False

tests/unit_tests/test_platformio_api.py

@@ -6,7 +6,7 @@ import os
 from pathlib import Path
 import shutil
 from types import SimpleNamespace
-from unittest.mock import MagicMock, Mock, patch
+from unittest.mock import MagicMock, Mock, call, patch
 
 import pytest
 
@@ -673,6 +673,200 @@ def test_process_stacktrace_bad_alloc(
     assert state is False
 
 
+def test_patch_file_downloader_succeeds_first_try() -> None:
+    """Test patch_file_downloader succeeds on first attempt."""
+    mock_exception_cls = type("PackageException", (Exception,), {})
+    original_init = MagicMock()
+
+    with patch.dict(
+        "sys.modules",
+        {
+            "platformio": MagicMock(),
+            "platformio.package": MagicMock(),
+            "platformio.package.download": SimpleNamespace(
+                FileDownloader=type("FileDownloader", (), {"__init__": original_init})
+            ),
+            "platformio.package.exception": SimpleNamespace(
+                PackageException=mock_exception_cls
+            ),
+        },
+    ):
+        platformio_api.patch_file_downloader()
+
+        from platformio.package.download import FileDownloader
+
+        instance = object.__new__(FileDownloader)
+        FileDownloader.__init__(instance, "http://example.com/file.zip")
+
+        original_init.assert_called_once()
+
+
+def test_patch_file_downloader_retries_on_failure() -> None:
+    """Test patch_file_downloader retries with backoff on PackageException."""
+    mock_exception_cls = type("PackageException", (Exception,), {})
+    call_count = 0
+
+    def failing_init(self, *args, **kwargs):
+        nonlocal call_count
+        call_count += 1
+        if call_count < 3:
+            raise mock_exception_cls(f"502 error attempt {call_count}")
+
+    with (
+        patch.dict(
+            "sys.modules",
+            {
+                "platformio": MagicMock(),
+                "platformio.package": MagicMock(),
+                "platformio.package.download": SimpleNamespace(
+                    FileDownloader=type(
+                        "FileDownloader", (), {"__init__": failing_init}
+                    )
+                ),
+                "platformio.package.exception": SimpleNamespace(
+                    PackageException=mock_exception_cls
+                ),
+            },
+        ),
+        patch("time.sleep") as mock_sleep,
+    ):
+        platformio_api.patch_file_downloader()
+
+        from platformio.package.download import FileDownloader
+
+        instance = object.__new__(FileDownloader)
+        FileDownloader.__init__(instance, "http://example.com/file.zip")
+
+        # Should have been called 3 times (2 failures + 1 success)
+        assert call_count == 3
+
+        # Should have slept with exponential backoff: 2s, 4s
+        assert mock_sleep.call_count == 2
+        mock_sleep.assert_any_call(2)
+        mock_sleep.assert_any_call(4)
+
+
+def test_patch_file_downloader_raises_after_max_retries() -> None:
+    """Test patch_file_downloader raises after exhausting all retries."""
+    mock_exception_cls = type("PackageException", (Exception,), {})
+
+    def always_failing_init(self, *args, **kwargs):
+        raise mock_exception_cls("502 error")
+
+    with (
+        patch.dict(
+            "sys.modules",
+            {
+                "platformio": MagicMock(),
+                "platformio.package": MagicMock(),
+                "platformio.package.download": SimpleNamespace(
+                    FileDownloader=type(
+                        "FileDownloader", (), {"__init__": always_failing_init}
+                    )
+                ),
+                "platformio.package.exception": SimpleNamespace(
+                    PackageException=mock_exception_cls
+                ),
+            },
+        ),
+        patch("time.sleep") as mock_sleep,
+    ):
+        platformio_api.patch_file_downloader()
+
+        from platformio.package.download import FileDownloader
+
+        instance = object.__new__(FileDownloader)
+        with pytest.raises(mock_exception_cls, match="502 error"):
+            FileDownloader.__init__(instance, "http://example.com/file.zip")
+
+        # Should have slept 4 times (before attempts 2-5), not on final attempt
+        assert mock_sleep.call_count == 4
+        mock_sleep.assert_has_calls([call(2), call(4), call(8), call(16)])
+
+
+def test_patch_file_downloader_closes_session_and_response_between_retries() -> None:
+    """Test patch_file_downloader closes HTTP session and response between retries."""
+    mock_exception_cls = type("PackageException", (Exception,), {})
+    mock_session = MagicMock()
+    mock_response = MagicMock()
+    call_count = 0
+
+    def failing_init_with_session(self, *args, **kwargs):
+        nonlocal call_count
+        call_count += 1
+        self._http_session = mock_session
+        self._http_response = mock_response
+        if call_count < 2:
+            raise mock_exception_cls("502 error")
+
+    with (
+        patch.dict(
+            "sys.modules",
+            {
+                "platformio": MagicMock(),
+                "platformio.package": MagicMock(),
+                "platformio.package.download": SimpleNamespace(
+                    FileDownloader=type(
+                        "FileDownloader",
+                        (),
+                        {"__init__": failing_init_with_session},
+                    )
+                ),
+                "platformio.package.exception": SimpleNamespace(
+                    PackageException=mock_exception_cls
+                ),
+            },
+        ),
+        patch("time.sleep"),
+    ):
+        platformio_api.patch_file_downloader()
+
+        from platformio.package.download import FileDownloader
+
+        instance = object.__new__(FileDownloader)
+        FileDownloader.__init__(instance, "http://example.com/file.zip")
+
+        # Both response and session should have been closed between retries
+        mock_response.close.assert_called_once()
+        mock_session.close.assert_called_once()
+
+
+def test_patch_file_downloader_idempotent() -> None:
+    """Test patch_file_downloader does not stack wrappers when called multiple times."""
+    mock_exception_cls = type("PackageException", (Exception,), {})
+    call_count = 0
+
+    def counting_init(self, *args, **kwargs):
+        nonlocal call_count
+        call_count += 1
+
+    with patch.dict(
+        "sys.modules",
+        {
+            "platformio": MagicMock(),
+            "platformio.package": MagicMock(),
+            "platformio.package.download": SimpleNamespace(
+                FileDownloader=type("FileDownloader", (), {"__init__": counting_init})
+            ),
+            "platformio.package.exception": SimpleNamespace(
+                PackageException=mock_exception_cls
+            ),
+        },
+    ):
+        # Patch multiple times
+        platformio_api.patch_file_downloader()
+        platformio_api.patch_file_downloader()
+        platformio_api.patch_file_downloader()
+
+        from platformio.package.download import FileDownloader
+
+        instance = object.__new__(FileDownloader)
+        FileDownloader.__init__(instance, "http://example.com/file.zip")
+
+        # Should only be called once, not 3 times from stacked wrappers
+        assert call_count == 1
+
+
 def test_platformio_log_filter_allows_non_platformio_messages() -> None:
     """Test that non-platformio logger messages are allowed through."""
     log_filter = platformio_api.PlatformioLogFilter()