[api] Assert the switch frame count instead of reading for it separately

Ping, device info, list entities done and disconnect all travel through
the ProtoMessage static entry points now that the no-op thunk is gone.

esphome/components/api/api_connection.h

@@ -345,11 +345,7 @@ class APIConnection final : public APIServerConnectionBase {
   /// Returns false as soon as the TCP buffer is full. Marked nodiscard so we
   /// have no silent failures: every caller must handle (or log) a refusal.
   template<typename T> [[nodiscard]] bool send_message(const T &msg) {
-    if constexpr (T::ESTIMATED_SIZE == 0) {
-      return this->send_message_(0, T::MESSAGE_TYPE, &encode_msg_noop, &msg);
-    } else {
-      return this->send_message_(msg.calculate_size(), T::MESSAGE_TYPE, &proto_encode_msg<T>, &msg);
-    }
+    return this->send_message_(T::calc_size_msg(&msg), T::MESSAGE_TYPE, &T::encode_msg, &msg);
   }
 
   /// Clear the shared write buffer and reserve space for the first message.
@@ -405,16 +401,6 @@ class APIConnection final : public APIServerConnectionBase {
   void process_state_subscriptions_();
 #endif
 
-  // Size thunk — converts void* back to concrete type for direct calculate_size() call
-  template<typename T> static uint32_t calc_size(const void *msg) {
-    return static_cast<const T *>(msg)->calculate_size();
-  }
-
-  // Shared no-op encode thunk for empty messages (ESTIMATED_SIZE == 0)
-  static uint8_t *encode_msg_noop(const void *, ProtoWriteBuffer &buf PROTO_ENCODE_DEBUG_PARAM) {
-    return buf.get_pos();
-  }
-
   // Non-template buffer management for send_message
   bool send_message_(uint32_t payload_size, uint16_t message_type, MessageEncodeFn encode_fn, const void *msg);
 
@@ -433,11 +419,7 @@ class APIConnection final : public APIServerConnectionBase {
   // Hot paths (state/info) go through fill_and_encode_entity_state/info instead.
   // batch_message_type_ is already set by dispatch_message_ before reaching here.
   template<typename T> static uint16_t encode_message_to_buffer(T &msg, APIConnection *conn, uint32_t remaining_size) {
-    if constexpr (T::ESTIMATED_SIZE == 0) {
-      return encode_to_buffer_slow(0, &encode_msg_noop, &msg, conn, remaining_size);
-    } else {
-      return encode_to_buffer_slow(msg.calculate_size(), &proto_encode_msg<T>, &msg, conn, remaining_size);
-    }
+    return encode_to_buffer_slow(T::calc_size_msg(&msg), &T::encode_msg, &msg, conn, remaining_size);
   }
 
   // Non-template core — fills state fields and encodes
@@ -449,7 +431,7 @@ class APIConnection final : public APIServerConnectionBase {
   template<typename T>
   static uint16_t fill_and_encode_entity_state(EntityBase *entity, T &msg, APIConnection *conn,
                                                uint32_t remaining_size) {
-    return fill_and_encode_entity_state(entity, msg, &calc_size<T>, &proto_encode_msg<T>, conn, remaining_size);
+    return fill_and_encode_entity_state(entity, msg, &T::calc_size_msg, &T::encode_msg, conn, remaining_size);
   }
 
   // Non-template core — fills info fields, allocates buffers, and encodes
@@ -461,7 +443,7 @@ class APIConnection final : public APIServerConnectionBase {
   template<typename T>
   static uint16_t fill_and_encode_entity_info(EntityBase *entity, T &msg, APIConnection *conn,
                                               uint32_t remaining_size) {
-    return fill_and_encode_entity_info(entity, msg, &calc_size<T>, &proto_encode_msg<T>, conn, remaining_size);
+    return fill_and_encode_entity_info(entity, msg, &T::calc_size_msg, &T::encode_msg, conn, remaining_size);
   }
 
   // Non-template core — fills device_class, then delegates to fill_and_encode_entity_info
@@ -475,8 +457,8 @@ class APIConnection final : public APIServerConnectionBase {
   static uint16_t fill_and_encode_entity_info_with_device_class(EntityBase *entity, T &msg,
                                                                 StringRef &device_class_field, APIConnection *conn,
                                                                 uint32_t remaining_size) {
-    return fill_and_encode_entity_info_with_device_class(entity, msg, device_class_field, &calc_size<T>,
-                                                         &proto_encode_msg<T>, conn, remaining_size);
+    return fill_and_encode_entity_info_with_device_class(entity, msg, device_class_field, &T::calc_size_msg,
+                                                         &T::encode_msg, conn, remaining_size);
   }
 
 #ifdef USE_VOICE_ASSISTANT

esphome/components/api/proto.h

@@ -287,19 +287,31 @@ class ProtoWriteBuffer {
   uint8_t *pos_;
 };
 
+// A four byte unaligned store is a memcpy call on ESP-IDF (-fno-builtin-memcpy) and on ARM cores without
+// unaligned access (Cortex-M0+, ARM9), so those targets share one outlined byte store helper per fixed32
+// field. Elsewhere the write inlines to a single store, or on ESP8266 to a few stores that measured
+// faster than a call, so it stays inline.
+#if defined(USE_ESP32) || (defined(__arm__) && !defined(__ARM_FEATURE_UNALIGNED))
+#define PROTO_OUTLINE_FOR_SIZE __attribute__((noinline))
+#define PROTO_FIXED32_BYTE_STORES true
+#else
+#define PROTO_OUTLINE_FOR_SIZE inline
+#define PROTO_FIXED32_BYTE_STORES false
+#endif
+
 // Varint encoding thresholds — used by both proto_encode_* free functions and ProtoSize.
 constexpr uint32_t VARINT_MAX_1_BYTE = 1 << 7;   // 128
 constexpr uint32_t VARINT_MAX_2_BYTE = 1 << 14;  // 16384
 
-/// Static encode helpers for generated encode() functions.
-/// Generated code hoists buffer.pos_ into a local uint8_t *__restrict__ pos,
-/// then calls these methods which take pos by reference. No struct, no overhead.
-/// For sub-messages, pos is synced back to buffer before the call and reloaded after.
+/// Static encode helpers for the generated encode bodies. Each takes the write cursor by value and
+/// returns it advanced, so outlined calls at -Os chain through the return register instead of a
+/// stack slot. Helpers without a _force suffix skip fields holding the proto3 default.
 class ProtoEncode {
  public:
   /// Write a multi-byte varint directly through a pos pointer.
   template<typename T>
-  static inline void encode_varint_raw_loop(uint8_t *__restrict__ &pos PROTO_ENCODE_DEBUG_PARAM, T value) {
+  [[nodiscard]] static inline uint8_t *encode_varint_raw_loop(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM,
+                                                              T value) {
     do {
       PROTO_ENCODE_CHECK_BOUNDS(pos, 1);
       *pos++ = static_cast<uint8_t>(value | 0x80);
@@ -307,48 +319,49 @@ class ProtoEncode {
     } while (value > 0x7F);
     PROTO_ENCODE_CHECK_BOUNDS(pos, 1);
     *pos++ = static_cast<uint8_t>(value);
+    return pos;
   }
-  static inline void ESPHOME_ALWAYS_INLINE encode_varint_raw(uint8_t *__restrict__ &pos PROTO_ENCODE_DEBUG_PARAM,
-                                                             uint32_t value) {
+  [[nodiscard]] static inline uint8_t *ESPHOME_ALWAYS_INLINE
+  encode_varint_raw(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM, uint32_t value) {
     if (value < VARINT_MAX_1_BYTE) [[likely]] {
       PROTO_ENCODE_CHECK_BOUNDS(pos, 1);
       *pos++ = static_cast<uint8_t>(value);
-      return;
+      return pos;
     }
-    encode_varint_raw_loop(pos PROTO_ENCODE_DEBUG_ARG, value);
+    return encode_varint_raw_loop(pos PROTO_ENCODE_DEBUG_ARG, value);
   }
   /// Encode a varint that is expected to be 1-2 bytes (e.g. zigzag RSSI, small lengths).
-  static inline void ESPHOME_ALWAYS_INLINE encode_varint_raw_short(uint8_t *__restrict__ &pos PROTO_ENCODE_DEBUG_PARAM,
-                                                                   uint32_t value) {
+  [[nodiscard]] static inline uint8_t *ESPHOME_ALWAYS_INLINE
+  encode_varint_raw_short(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM, uint32_t value) {
     if (value < VARINT_MAX_1_BYTE) [[likely]] {
       PROTO_ENCODE_CHECK_BOUNDS(pos, 1);
       *pos++ = static_cast<uint8_t>(value);
-      return;
+      return pos;
     }
     if (value < VARINT_MAX_2_BYTE) [[likely]] {
       PROTO_ENCODE_CHECK_BOUNDS(pos, 2);
       *pos++ = static_cast<uint8_t>(value | 0x80);
       *pos++ = static_cast<uint8_t>(value >> 7);
-      return;
+      return pos;
     }
-    encode_varint_raw_loop(pos PROTO_ENCODE_DEBUG_ARG, value);
+    return encode_varint_raw_loop(pos PROTO_ENCODE_DEBUG_ARG, value);
   }
-  static inline void ESPHOME_ALWAYS_INLINE encode_varint_raw_64(uint8_t *__restrict__ &pos PROTO_ENCODE_DEBUG_PARAM,
-                                                                uint64_t value) {
+  [[nodiscard]] static inline uint8_t *ESPHOME_ALWAYS_INLINE
+  encode_varint_raw_64(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM, uint64_t value) {
     if (value < VARINT_MAX_1_BYTE) [[likely]] {
       PROTO_ENCODE_CHECK_BOUNDS(pos, 1);
       *pos++ = static_cast<uint8_t>(value);
-      return;
+      return pos;
     }
-    encode_varint_raw_loop(pos PROTO_ENCODE_DEBUG_ARG, value);
+    return encode_varint_raw_loop(pos PROTO_ENCODE_DEBUG_ARG, value);
   }
   /// Encode a 48-bit MAC address (stored in a uint64) as varint.
   /// Real MAC addresses occupy the full 48 bits (OUI in upper 24), so the
   /// fast path -- any non-zero bit in the top 6 of 48 -- emits exactly 7 bytes
   /// with no per-byte branch. Falls back to the general loop otherwise.
   /// Caller must guarantee value fits in 48 bits (checked in debug builds).
-  static inline void ESPHOME_ALWAYS_INLINE encode_varint_raw_48bit(uint8_t *__restrict__ &pos PROTO_ENCODE_DEBUG_PARAM,
-                                                                   uint64_t value) {
+  [[nodiscard]] static inline uint8_t *ESPHOME_ALWAYS_INLINE
+  encode_varint_raw_48bit(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM, uint64_t value) {
 #ifdef ESPHOME_DEBUG_API
     assert(value < (1ULL << (MAC_ADDRESS_SIZE * 8)) && "encode_varint_raw_48bit: value exceeds 48 bits");
 #endif
@@ -363,38 +376,39 @@ class ProtoEncode {
       pos[4] = static_cast<uint8_t>((value >> 28) | 0x80);
       pos[5] = static_cast<uint8_t>((value >> 35) | 0x80);
       pos[6] = static_cast<uint8_t>(value >> 42);
-      pos += 7;
-      return;
+      return pos + 7;
     }
-    encode_varint_raw_64(pos PROTO_ENCODE_DEBUG_ARG, value);
+    return encode_varint_raw_64(pos PROTO_ENCODE_DEBUG_ARG, value);
   }
-  static inline void ESPHOME_ALWAYS_INLINE encode_field_raw(uint8_t *__restrict__ &pos PROTO_ENCODE_DEBUG_PARAM,
-                                                            uint32_t field_id, uint32_t type) {
-    encode_varint_raw(pos PROTO_ENCODE_DEBUG_ARG, (field_id << 3) | type);
+  [[nodiscard]] static inline uint8_t *ESPHOME_ALWAYS_INLINE
+  encode_field_raw(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM, uint32_t field_id, uint32_t type) {
+    return encode_varint_raw(pos PROTO_ENCODE_DEBUG_ARG, (field_id << 3) | type);
   }
   /// Write a single precomputed tag byte. Tag must be < 128.
-  static inline void ESPHOME_ALWAYS_INLINE write_raw_byte(uint8_t *__restrict__ &pos PROTO_ENCODE_DEBUG_PARAM,
-                                                          uint8_t b) {
+  [[nodiscard]] static inline uint8_t *ESPHOME_ALWAYS_INLINE
+  write_raw_byte(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM, uint8_t b) {
     PROTO_ENCODE_CHECK_BOUNDS(pos, 1);
     *pos++ = b;
+    return pos;
   }
   /// Reserve one byte for later backpatch (e.g., sub-message length).
   /// Advances pos past the reserved byte without writing a value.
-  static inline void ESPHOME_ALWAYS_INLINE reserve_byte(uint8_t *__restrict__ &pos PROTO_ENCODE_DEBUG_PARAM) {
+  [[nodiscard]] static inline uint8_t *ESPHOME_ALWAYS_INLINE
+  reserve_byte(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM) {
     PROTO_ENCODE_CHECK_BOUNDS(pos, 1);
-    pos++;
+    return pos + 1;
   }
   /// Write raw bytes to the buffer (no tag, no length prefix).
-  static inline void ESPHOME_ALWAYS_INLINE encode_raw(uint8_t *__restrict__ &pos PROTO_ENCODE_DEBUG_PARAM,
-                                                      const void *data, size_t len) {
+  [[nodiscard]] static inline uint8_t *ESPHOME_ALWAYS_INLINE
+  encode_raw(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM, const void *data, size_t len) {
     PROTO_ENCODE_CHECK_BOUNDS(pos, len);
     std::memcpy(pos, data, len);
-    pos += len;
+    return pos + len;
   }
   /// Encode tag + 1-byte length + raw string data. For strings with max_data_length < 128.
   /// Tag must be a single-byte varint (< 128). Always encodes (no zero check).
-  static inline void encode_short_string_force(uint8_t *__restrict__ &pos PROTO_ENCODE_DEBUG_PARAM, uint8_t tag,
-                                               const StringRef &ref) {
+  [[nodiscard]] static inline uint8_t *encode_short_string_force(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM,
+                                                                 uint8_t tag, const StringRef &ref) {
 #ifdef ESPHOME_DEBUG_API
     assert(ref.size() < 128 && "encode_short_string_force: string exceeds max_data_length < 128");
 #endif
@@ -402,137 +416,187 @@ class ProtoEncode {
     pos[0] = tag;
     pos[1] = static_cast<uint8_t>(ref.size());
     std::memcpy(pos + 2, ref.c_str(), ref.size());
-    pos += 2 + ref.size();
+    return pos + 2 + ref.size();
   }
-  /// Write a precomputed tag byte + 32-bit value in one operation.
-  static inline void ESPHOME_ALWAYS_INLINE write_tag_and_fixed32(uint8_t *__restrict__ &pos PROTO_ENCODE_DEBUG_PARAM,
-                                                                 uint8_t tag, uint32_t value) {
+  /// Unaligned little-endian store; __builtin_memcpy stays inline even under -fno-builtin-memcpy.
+  static inline void ESPHOME_ALWAYS_INLINE write_fixed32_le(uint8_t *__restrict__ pos, uint32_t value) {
+    if constexpr (PROTO_FIXED32_BYTE_STORES) {
+      // Spelled out so the outlined helper does not itself become a memcpy call
+      pos[0] = static_cast<uint8_t>(value);
+      pos[1] = static_cast<uint8_t>(value >> 8);
+      pos[2] = static_cast<uint8_t>(value >> 16);
+      pos[3] = static_cast<uint8_t>(value >> 24);
+    } else {
+      const uint32_t le = convert_little_endian(value);
+      __builtin_memcpy(pos, &le, 4);
+    }
+  }
+  /// Write a precomputed tag byte + 32-bit value. Outlined on embedded: one copy beats inline stores per field.
+  [[nodiscard]] static PROTO_OUTLINE_FOR_SIZE uint8_t *write_tag_and_fixed32(
+      uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM, uint8_t tag, uint32_t value) {
     PROTO_ENCODE_CHECK_BOUNDS(pos, 5);
     pos[0] = tag;
-#if __BYTE_ORDER__ == __ORDER_LITTLE_ENDIAN__
-    std::memcpy(pos + 1, &value, 4);
-#else
-    pos[1] = static_cast<uint8_t>(value & 0xFF);
-    pos[2] = static_cast<uint8_t>((value >> 8) & 0xFF);
-    pos[3] = static_cast<uint8_t>((value >> 16) & 0xFF);
-    pos[4] = static_cast<uint8_t>((value >> 24) & 0xFF);
-#endif
-    pos += 5;
+    write_fixed32_le(pos + 1, value);
+    return pos + 5;
   }
-  static inline void encode_string(uint8_t *__restrict__ &pos PROTO_ENCODE_DEBUG_PARAM, uint32_t field_id,
-                                   const char *string, size_t len, bool force = false) {
-    if (len == 0 && !force)
-      return;
-    encode_field_raw(pos PROTO_ENCODE_DEBUG_ARG, field_id, 2);  // type 2: Length-delimited string
+  [[nodiscard]] static inline uint8_t *encode_string_force(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM,
+                                                           uint32_t field_id, const char *string, size_t len) {
+    pos = encode_field_raw(pos PROTO_ENCODE_DEBUG_ARG, field_id, 2);  // type 2: Length-delimited string
     // NOLINTNEXTLINE(readability-inconsistent-ifelse-braces) -- false positive on [[likely]] attribute
     if (len < VARINT_MAX_1_BYTE) [[likely]] {
       PROTO_ENCODE_CHECK_BOUNDS(pos, 1 + len);
       *pos++ = static_cast<uint8_t>(len);
     } else {
-      encode_varint_raw_loop(pos PROTO_ENCODE_DEBUG_ARG, len);
+      pos = encode_varint_raw_loop(pos PROTO_ENCODE_DEBUG_ARG, len);
       PROTO_ENCODE_CHECK_BOUNDS(pos, len);
     }
     std::memcpy(pos, string, len);
-    pos += len;
+    return pos + len;
   }
-  static inline void encode_string(uint8_t *__restrict__ &pos PROTO_ENCODE_DEBUG_PARAM, uint32_t field_id,
-                                   const std::string &value, bool force = false) {
-    encode_string(pos PROTO_ENCODE_DEBUG_ARG, field_id, value.data(), value.size(), force);
+  [[nodiscard]] static inline uint8_t *encode_string(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM,
+                                                     uint32_t field_id, const char *string, size_t len) {
+    if (len == 0)
+      return pos;
+    return encode_string_force(pos PROTO_ENCODE_DEBUG_ARG, field_id, string, len);
   }
-  static inline void encode_string(uint8_t *__restrict__ &pos PROTO_ENCODE_DEBUG_PARAM, uint32_t field_id,
-                                   const StringRef &ref, bool force = false) {
-    encode_string(pos PROTO_ENCODE_DEBUG_ARG, field_id, ref.c_str(), ref.size(), force);
+  [[nodiscard]] static inline uint8_t *encode_string_force(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM,
+                                                           uint32_t field_id, const std::string &value) {
+    return encode_string_force(pos PROTO_ENCODE_DEBUG_ARG, field_id, value.data(), value.size());
   }
-  static inline void encode_bytes(uint8_t *__restrict__ &pos PROTO_ENCODE_DEBUG_PARAM, uint32_t field_id,
-                                  const uint8_t *data, size_t len, bool force = false) {
-    encode_string(pos PROTO_ENCODE_DEBUG_ARG, field_id, reinterpret_cast<const char *>(data), len, force);
+  [[nodiscard]] static inline uint8_t *encode_string(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM,
+                                                     uint32_t field_id, const StringRef &ref) {
+    return encode_string(pos PROTO_ENCODE_DEBUG_ARG, field_id, ref.c_str(), ref.size());
   }
-  static inline void encode_uint32(uint8_t *__restrict__ &pos PROTO_ENCODE_DEBUG_PARAM, uint32_t field_id,
-                                   uint32_t value, bool force = false) {
-    if (value == 0 && !force)
-      return;
-    encode_field_raw(pos PROTO_ENCODE_DEBUG_ARG, field_id, 0);
-    encode_varint_raw(pos PROTO_ENCODE_DEBUG_ARG, value);
+  [[nodiscard]] static inline uint8_t *encode_string_force(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM,
+                                                           uint32_t field_id, const StringRef &ref) {
+    return encode_string_force(pos PROTO_ENCODE_DEBUG_ARG, field_id, ref.c_str(), ref.size());
   }
-  static inline void encode_uint64(uint8_t *__restrict__ &pos PROTO_ENCODE_DEBUG_PARAM, uint32_t field_id,
-                                   uint64_t value, bool force = false) {
-    if (value == 0 && !force)
-      return;
-    encode_field_raw(pos PROTO_ENCODE_DEBUG_ARG, field_id, 0);
-    encode_varint_raw_64(pos PROTO_ENCODE_DEBUG_ARG, value);
+  [[nodiscard]] static inline uint8_t *encode_bytes(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM,
+                                                    uint32_t field_id, const uint8_t *data, size_t len) {
+    return encode_string(pos PROTO_ENCODE_DEBUG_ARG, field_id, reinterpret_cast<const char *>(data), len);
   }
-  static inline void encode_bool(uint8_t *__restrict__ &pos PROTO_ENCODE_DEBUG_PARAM, uint32_t field_id, bool value,
-                                 bool force = false) {
-    if (!value && !force)
-      return;
-    encode_field_raw(pos PROTO_ENCODE_DEBUG_ARG, field_id, 0);
+  [[nodiscard]] static inline uint8_t *encode_bytes_force(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM,
+                                                          uint32_t field_id, const uint8_t *data, size_t len) {
+    return encode_string_force(pos PROTO_ENCODE_DEBUG_ARG, field_id, reinterpret_cast<const char *>(data), len);
+  }
+  [[nodiscard]] static inline uint8_t *encode_uint32_force(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM,
+                                                           uint32_t field_id, uint32_t value) {
+    pos = encode_field_raw(pos PROTO_ENCODE_DEBUG_ARG, field_id, 0);
+    return encode_varint_raw(pos PROTO_ENCODE_DEBUG_ARG, value);
+  }
+  [[nodiscard]] static inline uint8_t *encode_uint32(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM,
+                                                     uint32_t field_id, uint32_t value) {
+    if (value == 0)
+      return pos;
+    return encode_uint32_force(pos PROTO_ENCODE_DEBUG_ARG, field_id, value);
+  }
+  [[nodiscard]] static inline uint8_t *encode_uint64_force(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM,
+                                                           uint32_t field_id, uint64_t value) {
+    pos = encode_field_raw(pos PROTO_ENCODE_DEBUG_ARG, field_id, 0);
+    return encode_varint_raw_64(pos PROTO_ENCODE_DEBUG_ARG, value);
+  }
+  [[nodiscard]] static inline uint8_t *encode_uint64(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM,
+                                                     uint32_t field_id, uint64_t value) {
+    if (value == 0)
+      return pos;
+    return encode_uint64_force(pos PROTO_ENCODE_DEBUG_ARG, field_id, value);
+  }
+  [[nodiscard]] static inline uint8_t *encode_bool_force(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM,
+                                                         uint32_t field_id, bool value) {
+    pos = encode_field_raw(pos PROTO_ENCODE_DEBUG_ARG, field_id, 0);
     PROTO_ENCODE_CHECK_BOUNDS(pos, 1);
     *pos++ = value ? 0x01 : 0x00;
+    return pos;
   }
-  static inline void encode_fixed32(uint8_t *__restrict__ &pos PROTO_ENCODE_DEBUG_PARAM, uint32_t field_id,
-                                    uint32_t value, bool force = false) {
-    if (value == 0 && !force)
-      return;
-    encode_field_raw(pos PROTO_ENCODE_DEBUG_ARG, field_id, 5);
+  [[nodiscard]] static inline uint8_t *encode_bool(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM,
+                                                   uint32_t field_id, bool value) {
+    if (!value)
+      return pos;
+    return encode_bool_force(pos PROTO_ENCODE_DEBUG_ARG, field_id, value);
+  }
+  /// Tag + fixed32 for multi-byte tags; single-byte tags use write_tag_and_fixed32.
+  [[nodiscard]] static PROTO_OUTLINE_FOR_SIZE uint8_t *encode_fixed32_force(
+      uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM, uint32_t field_id, uint32_t value) {
+    pos = encode_field_raw(pos PROTO_ENCODE_DEBUG_ARG, field_id, 5);
     PROTO_ENCODE_CHECK_BOUNDS(pos, 4);
-#if __BYTE_ORDER__ == __ORDER_LITTLE_ENDIAN__
-    std::memcpy(pos, &value, 4);
-    pos += 4;
-#else
-    *pos++ = (value >> 0) & 0xFF;
-    *pos++ = (value >> 8) & 0xFF;
-    *pos++ = (value >> 16) & 0xFF;
-    *pos++ = (value >> 24) & 0xFF;
-#endif
+    write_fixed32_le(pos, value);
+    return pos + 4;
+  }
+  [[nodiscard]] static inline uint8_t *encode_fixed32(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM,
+                                                      uint32_t field_id, uint32_t value) {
+    if (value == 0)
+      return pos;
+    return encode_fixed32_force(pos PROTO_ENCODE_DEBUG_ARG, field_id, value);
   }
   // NOTE: Wire type 1 (64-bit fixed: double, fixed64, sfixed64) is intentionally
   // not supported to reduce overhead on embedded systems. All ESPHome devices are
   // 32-bit microcontrollers where 64-bit operations are expensive. If 64-bit support
   // is needed in the future, the necessary encoding/decoding functions must be added.
-  static inline void encode_float(uint8_t *__restrict__ &pos PROTO_ENCODE_DEBUG_PARAM, uint32_t field_id, float value,
-                                  bool force = false) {
-    uint32_t raw = float_to_raw(value);
-    if (raw == 0 && !force)
-      return;
-    encode_fixed32(pos PROTO_ENCODE_DEBUG_ARG, field_id, raw);
+  [[nodiscard]] static inline uint8_t *encode_float(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM,
+                                                    uint32_t field_id, float value) {
+    return encode_fixed32(pos PROTO_ENCODE_DEBUG_ARG, field_id, float_to_raw(value));
   }
-  static inline void encode_int32(uint8_t *__restrict__ &pos PROTO_ENCODE_DEBUG_PARAM, uint32_t field_id, int32_t value,
-                                  bool force = false) {
+  [[nodiscard]] static inline uint8_t *encode_float_force(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM,
+                                                          uint32_t field_id, float value) {
+    return encode_fixed32_force(pos PROTO_ENCODE_DEBUG_ARG, field_id, float_to_raw(value));
+  }
+  [[nodiscard]] static inline uint8_t *encode_int32_force(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM,
+                                                          uint32_t field_id, int32_t value) {
     if (value < 0) {
       // negative int32 is always 10 byte long
-      encode_uint64(pos PROTO_ENCODE_DEBUG_ARG, field_id, static_cast<uint64_t>(value), force);
-      return;
+      return encode_uint64_force(pos PROTO_ENCODE_DEBUG_ARG, field_id, static_cast<uint64_t>(value));
     }
-    encode_uint32(pos PROTO_ENCODE_DEBUG_ARG, field_id, static_cast<uint32_t>(value), force);
+    return encode_uint32_force(pos PROTO_ENCODE_DEBUG_ARG, field_id, static_cast<uint32_t>(value));
   }
-  static inline void encode_int64(uint8_t *__restrict__ &pos PROTO_ENCODE_DEBUG_PARAM, uint32_t field_id, int64_t value,
-                                  bool force = false) {
-    encode_uint64(pos PROTO_ENCODE_DEBUG_ARG, field_id, static_cast<uint64_t>(value), force);
+  [[nodiscard]] static inline uint8_t *encode_int32(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM,
+                                                    uint32_t field_id, int32_t value) {
+    if (value == 0)
+      return pos;
+    return encode_int32_force(pos PROTO_ENCODE_DEBUG_ARG, field_id, value);
   }
-  static inline void encode_sint32(uint8_t *__restrict__ &pos PROTO_ENCODE_DEBUG_PARAM, uint32_t field_id,
-                                   int32_t value, bool force = false) {
-    encode_uint32(pos PROTO_ENCODE_DEBUG_ARG, field_id, encode_zigzag32(value), force);
+  [[nodiscard]] static inline uint8_t *encode_int64(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM,
+                                                    uint32_t field_id, int64_t value) {
+    return encode_uint64(pos PROTO_ENCODE_DEBUG_ARG, field_id, static_cast<uint64_t>(value));
   }
-  static inline void encode_sint64(uint8_t *__restrict__ &pos PROTO_ENCODE_DEBUG_PARAM, uint32_t field_id,
-                                   int64_t value, bool force = false) {
-    encode_uint64(pos PROTO_ENCODE_DEBUG_ARG, field_id, encode_zigzag64(value), force);
+  [[nodiscard]] static inline uint8_t *encode_int64_force(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM,
+                                                          uint32_t field_id, int64_t value) {
+    return encode_uint64_force(pos PROTO_ENCODE_DEBUG_ARG, field_id, static_cast<uint64_t>(value));
   }
-  /// Sub-message encoding: sync pos to buffer, delegate, get pos from return value.
+  [[nodiscard]] static inline uint8_t *encode_sint32(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM,
+                                                     uint32_t field_id, int32_t value) {
+    return encode_uint32(pos PROTO_ENCODE_DEBUG_ARG, field_id, encode_zigzag32(value));
+  }
+  [[nodiscard]] static inline uint8_t *encode_sint32_force(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM,
+                                                           uint32_t field_id, int32_t value) {
+    return encode_uint32_force(pos PROTO_ENCODE_DEBUG_ARG, field_id, encode_zigzag32(value));
+  }
+  [[nodiscard]] static inline uint8_t *encode_sint64(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM,
+                                                     uint32_t field_id, int64_t value) {
+    return encode_uint64(pos PROTO_ENCODE_DEBUG_ARG, field_id, encode_zigzag64(value));
+  }
+  [[nodiscard]] static inline uint8_t *encode_sint64_force(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM,
+                                                           uint32_t field_id, int64_t value) {
+    return encode_uint64_force(pos PROTO_ENCODE_DEBUG_ARG, field_id, encode_zigzag64(value));
+  }
+  /// Sub-message encoding: sync pos to buffer, delegate, read the cursor back.
   template<typename T>
-  static inline void encode_sub_message(uint8_t *__restrict__ &pos PROTO_ENCODE_DEBUG_PARAM, ProtoWriteBuffer &buffer,
-                                        uint32_t field_id, const T &value) {
+  [[nodiscard]] static inline uint8_t *encode_sub_message(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM,
+                                                          ProtoWriteBuffer &buffer, uint32_t field_id, const T &value) {
     buffer.set_pos(pos);
     buffer.encode_sub_message(field_id, value);
-    pos = buffer.get_pos();
+    return buffer.get_pos();
   }
   template<typename T>
-  static inline void encode_optional_sub_message(uint8_t *__restrict__ &pos PROTO_ENCODE_DEBUG_PARAM,
-                                                 ProtoWriteBuffer &buffer, uint32_t field_id, const T &value) {
+  [[nodiscard]] static inline uint8_t *encode_optional_sub_message(uint8_t *__restrict__ pos PROTO_ENCODE_DEBUG_PARAM,
+                                                                   ProtoWriteBuffer &buffer, uint32_t field_id,
+                                                                   const T &value) {
     buffer.set_pos(pos);
     buffer.encode_optional_sub_message(field_id, value);
-    pos = buffer.get_pos();
+    return buffer.get_pos();
   }
 };
+#undef PROTO_OUTLINE_FOR_SIZE
+#undef PROTO_FIXED32_BYTE_STORES
 
 #ifdef HAS_PROTO_MESSAGE_DUMP
 /**
@@ -624,11 +688,12 @@ class DumpBuffer {
 
 class ProtoMessage {
  public:
-  // Non-virtual defaults for messages with no fields.
-  // Concrete message classes hide these with their own implementations.
-  // All call sites use templates to preserve the concrete type, so virtual
-  // dispatch is not needed. This eliminates per-message vtable entries for
-  // encode/calculate_size, saving ~1.3 KB of flash across all message types.
+  // Non-virtual defaults for messages with no fields; generated classes hide all four. The
+  // static encode_msg/calc_size_msg take const void * so &T::encode_msg needs no thunk.
+  static uint8_t *encode_msg(const void *self, ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_PARAM) {
+    return buffer.get_pos();
+  }
+  static uint32_t calc_size_msg(const void *self) { return 0; }
   uint8_t *encode(ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_PARAM) const { return buffer.get_pos(); }
   uint32_t calculate_size() const { return 0; }
 #ifdef HAS_PROTO_MESSAGE_DUMP
@@ -876,19 +941,14 @@ class ProtoSize {
 
 // Implementation of methods that depend on ProtoSize being fully defined
 
-// Encode thunk — converts void* back to concrete type for direct encode() call
-template<typename T> uint8_t *proto_encode_msg(const void *msg, ProtoWriteBuffer &buf PROTO_ENCODE_DEBUG_PARAM) {
-  return static_cast<const T *>(msg)->encode(buf PROTO_ENCODE_DEBUG_ARG);
-}
-
 // Thin template wrapper; delegates to non-template core in proto.cpp.
 template<typename T> inline void ProtoWriteBuffer::encode_sub_message(uint32_t field_id, const T &value) {
-  this->encode_sub_message(field_id, &value, &proto_encode_msg<T>);
+  this->encode_sub_message(field_id, &value, &T::encode_msg);
 }
 
 // Thin template wrapper; delegates to non-template core.
 template<typename T> inline void ProtoWriteBuffer::encode_optional_sub_message(uint32_t field_id, const T &value) {
-  this->encode_optional_sub_message(field_id, value.calculate_size(), &value, &proto_encode_msg<T>);
+  this->encode_optional_sub_message(field_id, T::calc_size_msg(&value), &value, &T::encode_msg);
 }
 
 // Template decode_to_message - preserves concrete type so decode() resolves statically

esphome/components/esp32_hosted/__init__.py

@@ -37,6 +37,25 @@ CONF_HANDSHAKE_PIN = "handshake_pin"
 CONF_SDIO_FREQUENCY = "sdio_frequency"
 CONF_SPI_MODE = "spi_mode"
 
+# ESP-NOW-over-hosted shim (esp_now_hosted.cpp). esp-hosted proxies esp_wifi.h
+# but not esp_now.h (espressif/esp-hosted-mcu#19), and esp_wifi_remote injects
+# the esp_now.h header on the ESP32-P4 host with no implementation, leaving the
+# esp_now_* symbols undefined at link. On a P4 host, esp_now_hosted.cpp DEFINES
+# those symbols and forwards each call to the co-processor over esp-hosted's
+# CustomRpc "peer data transfer" channel, so ESPHome's `espnow` component links
+# and runs unchanged (proven on a Tab5, 2026-07-20). The .cpp is guarded to
+# CONFIG_IDF_TARGET_ESP32P4 so it compiles to nothing on hosts with a native
+# ESP-NOW stack. CustomRpc needs these two host-side Kconfig options. Host
+# registers 3 handlers (RESP, RECV, SEND); the coprocessor registers 1 (REQ);
+# we ask for 8 to leave room for other CustomRpc extensions alongside.
+#
+# The coprocessor must run the matching custom firmware (a parallel effort in
+# esphome/esp-hosted-firmware). esp_now_hosted_rpc.h here is the canonical copy
+# of the wire contract and MUST stay byte-identical to the copy that coprocessor
+# firmware uses — the packed structs are the on-wire layout, so any divergence
+# silently corrupts every ESP-NOW frame.
+_MAX_CUSTOM_MSG_HANDLERS = 8
+
 # Shared fields for both transport modes
 BASE_SCHEMA = cv.Schema(
     {
@@ -262,6 +281,23 @@ async def to_code(config: ConfigType) -> None:
     else:
         _configure_spi(config)
 
+    # ESP-NOW-over-hosted shim: only the radio-less ESP32-P4 host needs it (see
+    # the note by _MAX_CUSTOM_MSG_HANDLERS). Enabled for every P4 host, not
+    # gated on the `espnow` component being present: the shim is tiny and the
+    # esp_now_* symbols/CustomRpc calls it defines require these Kconfig options
+    # to link whenever esp_now_hosted.cpp compiles (which is on any P4 host), so
+    # coupling the two keeps the build consistent. When `espnow` is absent the
+    # symbols are simply unused and never register a callback at runtime.
+    if esp32.get_esp32_variant() == esp32.VARIANT_ESP32P4:
+        add_define("USE_ESP_NOW_HOSTED")
+        # esp-hosted's CustomRpc ("peer data transfer") path — off by default.
+        esp32.add_idf_sdkconfig_option(
+            "CONFIG_ESP_HOSTED_ENABLE_PEER_DATA_TRANSFER", True
+        )
+        esp32.add_idf_sdkconfig_option(
+            "CONFIG_ESP_HOSTED_MAX_CUSTOM_MSG_HANDLERS", _MAX_CUSTOM_MSG_HANDLERS
+        )
+
     # Place the transport mempool in PSRAM. Required on memory-tight host
     # configurations (e.g. P4 with a large LVGL UI) where the internal-RAM
     # mempool allocation fails at boot with `sdio_mempool_create` assert.

esphome/components/esp32_hosted/esp_now_hosted.cpp

@@ -0,0 +1,467 @@
+/*
+ * esp_now_hosted — host-side shim implementing <esp_now.h> over esp-hosted
+ * CustomRpc, so ESPHome's `espnow` component can run on a radio-less host
+ * (e.g. the ESP32-P4) whose radio lives on an esp-hosted co-processor.
+ *
+ * A radio-less host has no native ESP-NOW. esp_wifi_remote INJECTS the full
+ * esp_now.h header (types + declarations) but ships NO implementation, so every
+ * esp_now_* symbol is an undefined reference at link time. This translation
+ * unit provides those definitions; each forwards to the co-processor over
+ * CustomRpc (see esphome/esp-hosted-firmware for the matching coprocessor
+ * handlers). No esp-hosted or esp_wifi_remote source is patched, and there is no
+ * duplicate-symbol clash because nothing else defines these symbols here.
+ *
+ * See esp_now_hosted_rpc.h for the wire protocol.
+ */
+
+#include "sdkconfig.h"
+
+// Only build the shim on the radio-less host. On chips with a native ESP-NOW
+// stack (S3, C6, …) the real symbols exist and this file must stay empty to
+// avoid duplicate definitions.
+#if defined(CONFIG_IDF_TARGET_ESP32P4)
+
+#include <cstring>
+
+#include "freertos/FreeRTOS.h"
+#include "freertos/semphr.h"
+
+#include "esp_idf_version.h"
+#include "esp_log.h"
+#include "esp_timer.h"
+
+#include <esp_now.h>         // injected declarations we are now DEFINING
+#include <esp_wifi_types.h>  // wifi_pkt_rx_ctrl_t, wifi_tx_info_t
+
+// esp_hosted_misc.h (host) ships WITHOUT an extern "C" guard, so including it
+// from C++ would give its declarations C++ linkage and the real C symbols in
+// libesp_hosted would go unresolved at link. Wrap it. (Verified vs
+// esp_hosted 2.12.9.)
+extern "C" {
+#include "esp_hosted_misc.h"  // esp_hosted_{send_custom_data,register_custom_callback}
+}
+
+#include "esp_now_hosted_rpc.h"
+
+namespace {
+
+const char *const TAG = "esp_now_hosted";
+
+// One outstanding request at a time. ESPHome drives esp_now_* from the main
+// loop; the matching response and the async RECV/SEND events all arrive on the
+// single esp-hosted RPC RX thread. Serializing requests keeps the shared
+// response slot race-free; a sequence number stops a late/stale response from
+// being mistaken for ours.
+SemaphoreHandle_t g_req_mutex = nullptr;
+SemaphoreHandle_t g_resp_sem = nullptr;  // given when the matching RESP lands
+bool g_setup_done = false;               // set only after setup fully succeeds
+uint8_t g_seq = 0;
+volatile uint8_t g_expect_seq = 0;
+volatile int32_t g_resp_status = 0;
+uint8_t g_resp_ret[16];
+volatile uint16_t g_resp_ret_len = 0;
+
+// Written from the main loop (register/unregister/deinit), read from the
+// esp-hosted RX thread (on_recv/on_send). volatile for the same reason the
+// g_resp_* globals are: force the RX thread to observe an updated pointer
+// (e.g. a nulling by esp_now_deinit) rather than a cached one.
+volatile esp_now_recv_cb_t g_recv_cb = nullptr;
+volatile esp_now_send_cb_t g_send_cb = nullptr;
+
+// Local mirror of the co-processor's peer table. ESPHome's espnow component
+// calls esp_now_is_peer_exist() on the main loop for every received frame
+// (twice) and every send; forwarding each as a blocking RPC round-trip stalls
+// the loop. The shim is the only path that mutates the co-processor peer table
+// (add/del/deinit all go through here), so this mirror is authoritative and
+// esp_now_is_peer_exist() can answer from it with no round-trip.
+//
+// esp_now_* are public C symbols: any component or user lambda may call them,
+// and although ESPHome's espnow touches peers only from the main loop today
+// (its RX/TX callbacks merely enqueue), the shim cannot rely on that. A short
+// spinlock keeps the mirror consistent from any task/core, matching native
+// esp_now_*'s own internal thread-safety. The critical sections are a bounded
+// (<=20-entry) scan, so they stay tiny. ESP_NOW_MAX_TOTAL_PEER_NUM is 20.
+constexpr size_t ESP_NOW_HOSTED_MAX_PEERS = 20;
+uint8_t g_peer_cache[ESP_NOW_HOSTED_MAX_PEERS][6];
+size_t g_peer_count = 0;
+portMUX_TYPE g_peer_lock = portMUX_INITIALIZER_UNLOCKED;
+
+// Caller must hold g_peer_lock.
+int peer_cache_find_locked(const uint8_t *mac) {
+  for (size_t i = 0; i < g_peer_count; i++) {
+    if (memcmp(g_peer_cache[i], mac, 6) == 0)
+      return static_cast<int>(i);
+  }
+  return -1;
+}
+
+bool peer_cache_contains(const uint8_t *mac) {
+  portENTER_CRITICAL(&g_peer_lock);
+  const bool found = peer_cache_find_locked(mac) >= 0;
+  portEXIT_CRITICAL(&g_peer_lock);
+  return found;
+}
+
+void peer_cache_add(const uint8_t *mac) {
+  portENTER_CRITICAL(&g_peer_lock);
+  if (peer_cache_find_locked(mac) < 0 && g_peer_count < ESP_NOW_HOSTED_MAX_PEERS)
+    memcpy(g_peer_cache[g_peer_count++], mac, 6);
+  portEXIT_CRITICAL(&g_peer_lock);
+}
+
+void peer_cache_remove(const uint8_t *mac) {
+  portENTER_CRITICAL(&g_peer_lock);
+  const int idx = peer_cache_find_locked(mac);
+  if (idx >= 0) {
+    g_peer_count--;
+    if (static_cast<size_t>(idx) != g_peer_count)  // move the last entry into the gap
+      memcpy(g_peer_cache[idx], g_peer_cache[g_peer_count], 6);
+  }
+  portEXIT_CRITICAL(&g_peer_lock);
+}
+
+void peer_cache_clear() {
+  portENTER_CRITICAL(&g_peer_lock);
+  g_peer_count = 0;
+  portEXIT_CRITICAL(&g_peer_lock);
+}
+
+// ── CustomRpc event handlers (run on the esp-hosted RPC RX thread) ──────────
+// Keep them short and non-blocking. In particular they MUST NOT call back into
+// any esp_now_* shim function: that would try to take g_req_mutex / wait on the
+// RX thread that delivers the response, and deadlock.
+
+void on_resp(uint32_t /*msg_id*/, const uint8_t *data, size_t len, void * /*ctx*/) {
+  if (len < sizeof(esp_now_hosted_resp_t)) {
+    ESP_LOGW(TAG, "RESP too short: %u bytes", static_cast<unsigned>(len));
+    return;
+  }
+  const auto *r = reinterpret_cast<const esp_now_hosted_resp_t *>(data);
+  if (r->seq != g_expect_seq) {  // late response from a timed-out request (expected)
+    ESP_LOGV(TAG, "dropping stale RESP seq %u (want %u)", r->seq, g_expect_seq);
+    return;
+  }
+  g_resp_status = r->status;
+  uint16_t rl = r->ret_len;
+  if (rl > sizeof(g_resp_ret)) {
+    // Larger than any real opcode return — a likely wire-format drift signal.
+    ESP_LOGW(TAG, "RESP ret_len %u exceeds buffer, clamping (wire drift?)", rl);
+    rl = sizeof(g_resp_ret);
+  }
+  if (len >= sizeof(esp_now_hosted_resp_t) + rl) {
+    memcpy(g_resp_ret, r->ret, rl);
+  } else {
+    // Truncated frame: fail closed. Never hand the caller stale bytes left in
+    // g_resp_ret by a previous response, and don't let request() report a
+    // zeroed payload as success — override the status to an error.
+    ESP_LOGW(TAG, "RESP truncated: claims %u ret bytes, frame too short", rl);
+    rl = 0;
+    g_resp_status = ESP_ERR_INVALID_RESPONSE;
+  }
+  g_resp_ret_len = rl;
+  xSemaphoreGive(g_resp_sem);
+}
+
+void on_recv(uint32_t /*msg_id*/, const uint8_t *data, size_t len, void * /*ctx*/) {
+  // Read the volatile pointer once: esp_now_unregister_recv_cb()/deinit() (via
+  // the espnow component's disable()) can null it on the main loop between the
+  // guard and the call, which would otherwise turn the call into a null-deref.
+  const esp_now_recv_cb_t cb = g_recv_cb;
+  if (cb == nullptr)
+    return;
+  if (len < sizeof(esp_now_hosted_recv_evt_t)) {
+    ESP_LOGW(TAG, "RECV too short: %u bytes", static_cast<unsigned>(len));
+    return;
+  }
+  const auto *e = reinterpret_cast<const esp_now_hosted_recv_evt_t *>(data);
+  if (len < sizeof(esp_now_hosted_recv_evt_t) + e->data_len) {
+    ESP_LOGW(TAG, "RECV data_len %u exceeds frame", e->data_len);
+    return;
+  }
+
+  // ESPHome dereferences info->rx_ctrl->{rssi,timestamp}; give it a real one.
+  wifi_pkt_rx_ctrl_t rx_ctrl;
+  memset(&rx_ctrl, 0, sizeof(rx_ctrl));
+  rx_ctrl.rssi = e->rssi;
+  rx_ctrl.channel = e->channel;
+  rx_ctrl.timestamp = static_cast<uint32_t>(esp_timer_get_time());
+
+  esp_now_recv_info_t info;
+  info.src_addr = const_cast<uint8_t *>(e->src_addr);
+  info.des_addr = const_cast<uint8_t *>(e->des_addr);
+  info.rx_ctrl = &rx_ctrl;
+  cb(&info, e->data, static_cast<int>(e->data_len));
+}
+
+void on_send(uint32_t /*msg_id*/, const uint8_t *data, size_t len, void * /*ctx*/) {
+  // Read the volatile pointer once (see on_recv): disable()/deinit() can null it
+  // on the main loop concurrently with this RX-thread callback.
+  const esp_now_send_cb_t cb = g_send_cb;
+  if (cb == nullptr)
+    return;
+  if (len < sizeof(esp_now_hosted_send_evt_t)) {
+    ESP_LOGW(TAG, "SEND evt too short: %u bytes", static_cast<unsigned>(len));
+    return;
+  }
+  const auto *e = reinterpret_cast<const esp_now_hosted_send_evt_t *>(data);
+#if ESP_IDF_VERSION >= ESP_IDF_VERSION_VAL(5, 5, 0)
+  // IDF >= 5.5: esp_now_send_cb_t takes esp_now_send_info_t (== wifi_tx_info_t),
+  // whose des_addr is a POINTER (not an inline array). Point it at the event's
+  // MAC (valid for this callback) — do NOT memcpy into it (that writes NULL and
+  // faults). ESPHome reads only info->des_addr.
+  esp_now_send_info_t si;
+  memset(&si, 0, sizeof(si));
+  si.des_addr = const_cast<uint8_t *>(e->des_addr);
+  cb(&si, static_cast<esp_now_send_status_t>(e->status));
+#else
+  cb(e->des_addr, static_cast<esp_now_send_status_t>(e->status));
+#endif
+}
+
+esp_err_t ensure_setup() {
+  // Gate on g_setup_done, not on g_req_mutex: a failure part-way through (a
+  // semaphore that did not allocate, a callback that did not register) must not
+  // leave a later call thinking setup completed. Semaphore creation is guarded
+  // so a retry after a partial failure does not leak the earlier handles.
+  if (g_setup_done)
+    return ESP_OK;
+  if (g_req_mutex == nullptr)
+    g_req_mutex = xSemaphoreCreateMutex();
+  if (g_resp_sem == nullptr)
+    g_resp_sem = xSemaphoreCreateBinary();
+  if (g_req_mutex == nullptr || g_resp_sem == nullptr)
+    return ESP_ERR_NO_MEM;
+  esp_err_t err;
+  if ((err = esp_hosted_register_custom_callback(ESP_NOW_HOSTED_MSG_RESP, on_resp, nullptr)) != ESP_OK)
+    return err;
+  if ((err = esp_hosted_register_custom_callback(ESP_NOW_HOSTED_MSG_RECV, on_recv, nullptr)) != ESP_OK)
+    return err;
+  if ((err = esp_hosted_register_custom_callback(ESP_NOW_HOSTED_MSG_SEND, on_send, nullptr)) != ESP_OK)
+    return err;
+  g_setup_done = true;
+  return ESP_OK;
+}
+
+// Send one request envelope. With wait=true (default) block until the matching
+// response (or timeout); with wait=false return as soon as the frame is handed
+// to the transport (fire-and-forget, used by esp_now_send).
+//
+// `tail` is an optional second chunk written straight after `payload`. Callers
+// with a fixed header plus a bulk body (esp_now_send) pass the two separately
+// so they never need a build buffer of their own: both chunks are laid into the
+// request buffer here, under g_req_mutex, which keeps concurrent callers from
+// racing and saves a full copy of the body on every transmit.
+esp_err_t request(uint8_t opcode, const void *payload, uint16_t plen, void *ret, uint16_t ret_cap, uint16_t *ret_len,
+                  bool wait = true, const void *tail = nullptr, uint16_t tail_len = 0) {
+  esp_err_t err = ensure_setup();
+  if (err != ESP_OK)
+    return err;
+  if (plen > ESP_NOW_HOSTED_MAX_PAYLOAD || tail_len > ESP_NOW_HOSTED_MAX_PAYLOAD - plen)
+    return ESP_ERR_INVALID_SIZE;
+  const uint16_t total_len = static_cast<uint16_t>(plen + tail_len);
+
+  if (xSemaphoreTake(g_req_mutex, portMAX_DELAY) != pdTRUE)
+    return ESP_FAIL;
+
+  static uint8_t buf[sizeof(esp_now_hosted_req_t) + ESP_NOW_HOSTED_MAX_PAYLOAD];  // guarded by g_req_mutex
+  auto *req = reinterpret_cast<esp_now_hosted_req_t *>(buf);
+  req->opcode = opcode;
+  req->seq = ++g_seq;
+  req->payload_len = total_len;
+  if (plen != 0)
+    memcpy(req->payload, payload, plen);
+  if (tail_len != 0)
+    memcpy(req->payload + plen, tail, tail_len);
+  g_expect_seq = req->seq;
+
+  xSemaphoreTake(g_resp_sem, 0);  // drain any stale signal before sending
+  err = esp_hosted_send_custom_data(ESP_NOW_HOSTED_MSG_REQ, buf, sizeof(esp_now_hosted_req_t) + total_len);
+  if (err != ESP_OK) {
+    xSemaphoreGive(g_req_mutex);
+    return err;
+  }
+  if (!wait) {
+    // Fire-and-forget (esp_now_send): the co-processor enqueues the frame and
+    // reports the real TX result later via the async SEND event, exactly like
+    // native esp_now_send. Returning here keeps the main loop off the ~100 ms+
+    // RPC round-trip. The matching RESP is ignored (seq won't match the next
+    // waited request, so on_resp drops it).
+    xSemaphoreGive(g_req_mutex);
+    return ESP_OK;
+  }
+  if (xSemaphoreTake(g_resp_sem, pdMS_TO_TICKS(ESP_NOW_HOSTED_TIMEOUT_MS)) != pdTRUE) {
+    ESP_LOGW(TAG, "opcode %u timed out", opcode);
+    xSemaphoreGive(g_req_mutex);
+    return ESP_ERR_TIMEOUT;
+  }
+
+  const int32_t status = g_resp_status;
+  if (ret != nullptr && ret_cap != 0) {
+    uint16_t n = g_resp_ret_len < ret_cap ? g_resp_ret_len : ret_cap;
+    memcpy(ret, const_cast<const uint8_t *>(g_resp_ret), n);
+    if (ret_len != nullptr)
+      *ret_len = n;
+  }
+  xSemaphoreGive(g_req_mutex);
+  return static_cast<esp_err_t>(status);
+}
+
+}  // namespace
+
+// ── The <esp_now.h> surface, defined for the radio-less host ────────────────
+extern "C" {
+
+esp_err_t esp_now_init(void) { return request(ESP_NOW_HOSTED_OP_INIT, nullptr, 0, nullptr, 0, nullptr); }
+
+esp_err_t esp_now_deinit(void) {
+  g_recv_cb = nullptr;
+  g_send_cb = nullptr;
+  peer_cache_clear();  // the co-processor drops all peers on deinit
+  return request(ESP_NOW_HOSTED_OP_DEINIT, nullptr, 0, nullptr, 0, nullptr);
+}
+
+esp_err_t esp_now_get_version(uint32_t *version) {
+  uint32_t v = 0;
+  uint16_t rl = 0;
+  esp_err_t err = request(ESP_NOW_HOSTED_OP_GET_VERSION, nullptr, 0, &v, sizeof(v), &rl);
+  if (version != nullptr)
+    *version = v;
+  return err;
+}
+
+esp_err_t esp_now_register_recv_cb(esp_now_recv_cb_t cb) {
+  // Only arm the callback once the CustomRpc handlers are actually registered,
+  // so a failed setup leaves g_recv_cb null rather than falsely "registered".
+  esp_err_t err = ensure_setup();
+  if (err != ESP_OK)
+    return err;
+  g_recv_cb = cb;
+  return ESP_OK;
+}
+esp_err_t esp_now_unregister_recv_cb(void) {
+  g_recv_cb = nullptr;
+  return ESP_OK;
+}
+esp_err_t esp_now_register_send_cb(esp_now_send_cb_t cb) {
+  esp_err_t err = ensure_setup();
+  if (err != ESP_OK)
+    return err;
+  g_send_cb = cb;
+  return ESP_OK;
+}
+esp_err_t esp_now_unregister_send_cb(void) {
+  g_send_cb = nullptr;
+  return ESP_OK;
+}
+
+static esp_err_t add_or_mod_peer(uint8_t opcode, const esp_now_peer_info_t *peer, bool wait) {
+  if (peer == nullptr)
+    return ESP_ERR_ESPNOW_ARG;
+  esp_now_hosted_peer_t p;
+  memset(&p, 0, sizeof(p));
+  memcpy(p.peer_addr, peer->peer_addr, 6);
+  memcpy(p.lmk, peer->lmk, 16);
+  p.channel = peer->channel;
+  p.ifidx = static_cast<uint8_t>(peer->ifidx);
+  p.encrypt = peer->encrypt ? 1 : 0;
+  return request(opcode, &p, sizeof(p), nullptr, 0, nullptr, wait);
+}
+esp_err_t esp_now_add_peer(const esp_now_peer_info_t *peer) {
+  // Fire-and-forget (wait=false): adding a peer is a blocking RPC round-trip,
+  // and ESPHome's espnow calls it on the main loop when a device joins the mesh
+  // — under co-processor load that stalls the UI (peer-churn stutter). Issue it
+  // without waiting and mirror it locally. Safe against a following
+  // esp_now_send to the same peer: both ride the same in-order CustomRpc
+  // channel (mutex-serialized on the host) and the co-processor processes REQs
+  // FIFO, so ADD_PEER is applied before the SEND. Trade-off: a co-processor-side
+  // failure (e.g. peer table full) is no longer reported synchronously — the
+  // same limitation as esp_now_send — but ESPHome only adds peers it validated.
+  esp_err_t err = add_or_mod_peer(ESP_NOW_HOSTED_OP_ADD_PEER, peer, /*wait=*/false);
+  if (err == ESP_OK)
+    peer_cache_add(peer->peer_addr);  // keep the local mirror in sync
+  return err;
+}
+esp_err_t esp_now_mod_peer(const esp_now_peer_info_t *peer) {
+  // mod_peer changes a peer's parameters, not its existence, so the cache is
+  // unaffected. Kept synchronous — it is not on any hot path (espnow never
+  // calls it), so the extra round-trip does not matter and the status is useful.
+  return add_or_mod_peer(ESP_NOW_HOSTED_OP_MOD_PEER, peer, /*wait=*/true);
+}
+
+esp_err_t esp_now_del_peer(const uint8_t *peer_addr) {
+  if (peer_addr == nullptr)
+    return ESP_ERR_ESPNOW_ARG;
+  // Fire-and-forget for the same reason as add_peer (peer churn on the main
+  // loop). Removal is order-independent, so this is strictly safe.
+  esp_err_t err = request(ESP_NOW_HOSTED_OP_DEL_PEER, peer_addr, 6, nullptr, 0, nullptr, /*wait=*/false);
+  if (err == ESP_OK)
+    peer_cache_remove(peer_addr);  // keep the local mirror in sync
+  return err;
+}
+
+bool esp_now_is_peer_exist(const uint8_t *peer_addr) {
+  if (peer_addr == nullptr)
+    return false;
+  // Answered from the local mirror — no RPC round-trip. ESPHome's espnow calls
+  // this on the main loop for every received frame and every send, so a
+  // blocking round-trip here would stall rendering under mesh traffic.
+  return peer_cache_contains(peer_addr);
+}
+
+esp_err_t esp_now_send(const uint8_t *peer_addr, const uint8_t *data, size_t len) {
+  if (len > ESP_NOW_HOSTED_MAX_FRAME)
+    return ESP_ERR_ESPNOW_ARG;
+  if (data == nullptr && len != 0)  // native esp_now_send treats this as an arg error
+    return ESP_ERR_ESPNOW_ARG;
+  // Only the small fixed header is built here; the caller's frame goes over as
+  // the request tail, so request() lays both into its own buffer under
+  // g_req_mutex. esp_now_send is a public C symbol and may be called from any
+  // task, and a shared build buffer here would let two callers corrupt each
+  // other's frame. Passing the body through also drops a full-frame copy per
+  // transmit, on the path this shim exists to keep quick.
+  uint8_t hdr[sizeof(esp_now_hosted_send_req_t)];
+  auto *s = reinterpret_cast<esp_now_hosted_send_req_t *>(hdr);
+  s->has_addr = peer_addr != nullptr ? 1 : 0;
+  if (peer_addr != nullptr)
+    memcpy(s->peer_addr, peer_addr, 6);
+  else
+    memset(s->peer_addr, 0, 6);
+  s->data_len = static_cast<uint16_t>(len);
+  // Fire-and-forget (wait=false): native esp_now_send returns once the frame is
+  // queued, with the real TX result delivered later through the send callback.
+  // The co-processor mirrors that — it acks enqueue immediately and reports the
+  // outcome via the async SEND event (on_send -> on_send_report). Waiting for
+  // the RPC RESP here would block the main loop for the full round-trip on
+  // every transmit.
+  return request(ESP_NOW_HOSTED_OP_SEND, hdr, sizeof(hdr), nullptr, 0, nullptr, /*wait=*/false, data,
+                 static_cast<uint16_t>(len));
+}
+
+esp_err_t esp_now_set_pmk(const uint8_t *pmk) {
+  if (pmk == nullptr)
+    return ESP_ERR_ESPNOW_ARG;
+  return request(ESP_NOW_HOSTED_OP_SET_PMK, pmk, 16, nullptr, 0, nullptr);
+}
+
+// Remainder of the <esp_now.h> surface. Not used by ESPHome's espnow component
+// today; provided so the whole header links and future callers get a defined
+// (if unimplemented) symbol rather than a link error. Wire them through
+// CustomRpc if a use case appears.
+esp_err_t esp_now_get_peer(const uint8_t * /*peer_addr*/, esp_now_peer_info_t * /*peer*/) {
+  return ESP_ERR_NOT_SUPPORTED;
+}
+esp_err_t esp_now_fetch_peer(bool /*from_head*/, esp_now_peer_info_t * /*peer*/) { return ESP_ERR_NOT_SUPPORTED; }
+esp_err_t esp_now_get_peer_num(esp_now_peer_num_t * /*num*/) { return ESP_ERR_NOT_SUPPORTED; }
+esp_err_t esp_now_set_wake_window(uint16_t /*window*/) {
+  return ESP_ERR_NOT_SUPPORTED;  // power-save wake window is not forwarded; don't claim success
+}
+esp_err_t esp_now_set_peer_rate_config(const uint8_t * /*peer_addr*/, esp_now_rate_config_t * /*cfg*/) {
+  return ESP_ERR_NOT_SUPPORTED;
+}
+esp_err_t esp_wifi_config_espnow_rate(wifi_interface_t /*ifx*/, wifi_phy_rate_t /*rate*/) {
+  return ESP_ERR_NOT_SUPPORTED;
+}
+
+}  // extern "C"
+
+#endif  // CONFIG_IDF_TARGET_ESP32P4

esphome/components/esp32_hosted/esp_now_hosted_rpc.h

@@ -0,0 +1,128 @@
+/*
+ * esp_now_hosted — ESP-NOW-over-CustomRpc wire protocol.
+ *
+ * Shared, byte-for-byte-identical contract between:
+ *   - the host shim         (esphome/components/esp32_hosted/esp_now_hosted.cpp)
+ *   - the coprocessor firmware (esphome/esp-hosted-firmware)
+ *
+ * It rides esp-hosted's CustomRpc channel (RPC ID 388, "peer data transfer",
+ * available since esp-hosted v2.8.1), teaching the radio-less host <-> radio
+ * co-processor link to carry esp_now.h, which esp-hosted itself does not proxy
+ * (Espressif issue espressif/esp-hosted-mcu#19).
+ *
+ * KEEP THE TWO COPIES IN SYNC. The canonical copy lives here; the coprocessor
+ * firmware uses a verbatim copy. Both sides are little-endian, so these packed
+ * structs are wire-compatible with no byte-swapping.
+ */
+
+#ifndef ESP_NOW_HOSTED_RPC_H
+#define ESP_NOW_HOSTED_RPC_H
+
+#ifdef __cplusplus
+#include <cstdint>
+#else
+#include <stdint.h>
+#endif
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+/* ── CustomRpc message IDs (any uint32_t except 0xFFFFFFFF) ──────────────────
+ * One REQ handler slot on the device; three event handler slots on the host.
+ * The bytes spell "now" + index, a private range unlikely to clash with other
+ * CustomRpc users (e.g. the stock peer_data_transfer example's 1..6).         */
+#define ESP_NOW_HOSTED_MSG_REQ 0x6E6F7701u  /* host   -> device : request envelope */
+#define ESP_NOW_HOSTED_MSG_RESP 0x6E6F7702u /* device -> host   : reply to a REQ   */
+#define ESP_NOW_HOSTED_MSG_RECV 0x6E6F7703u /* device -> host   : async RX frame   */
+#define ESP_NOW_HOSTED_MSG_SEND 0x6E6F7704u /* device -> host   : async TX status  */
+
+/* ── Request opcodes ────────────────────────────────────────────────────── */
+enum {
+  ESP_NOW_HOSTED_OP_INIT = 1,          /* esp_now_init + register device recv/send cbs */
+  ESP_NOW_HOSTED_OP_DEINIT = 2,        /* unregister cbs + esp_now_deinit             */
+  ESP_NOW_HOSTED_OP_ADD_PEER = 3,      /* payload: esp_now_hosted_peer_t              */
+  ESP_NOW_HOSTED_OP_DEL_PEER = 4,      /* payload: 6-byte peer MAC                    */
+  ESP_NOW_HOSTED_OP_IS_PEER_EXIST = 5, /* payload: 6-byte MAC; ret: 1 byte bool       */
+  ESP_NOW_HOSTED_OP_SEND = 6,          /* payload: esp_now_hosted_send_req_t          */
+  ESP_NOW_HOSTED_OP_GET_VERSION = 7,   /* ret: uint32 version                        */
+  ESP_NOW_HOSTED_OP_SET_PMK = 8,       /* payload: 16-byte PMK                        */
+  ESP_NOW_HOSTED_OP_MOD_PEER = 9,      /* payload: esp_now_hosted_peer_t              */
+};
+
+/* Largest ESP-NOW payload we forward. ESP-NOW v2 (IDF >= 5.4) is 1470 B; well
+ * under esp-hosted's 8166 B CustomRpc cap, so the shim never truncates.        */
+#define ESP_NOW_HOSTED_MAX_FRAME 1470u
+/* Envelope slack for the largest opcode payload (a SEND req wrapping a frame). */
+#define ESP_NOW_HOSTED_MAX_PAYLOAD (ESP_NOW_HOSTED_MAX_FRAME + 16u)
+/* Host request/response round-trip timeout over the transport. Generous:
+ * normal RTT is sub-millisecond, but Wi-Fi/BLE contention on the co-processor
+ * can stall the RX thread.                                                     */
+#define ESP_NOW_HOSTED_TIMEOUT_MS 2000
+
+/* ── Envelopes ──────────────────────────────────────────────────────────── */
+
+/* These payloads are shared verbatim with the C co-processor firmware, so they
+ * use C's `typedef struct {...} name;` idiom rather than C++ `using` aliases,
+ * which would not compile there. Silence clang-tidy's modernize-use-using for
+ * the shared struct block. */
+// NOLINTBEGIN(modernize-use-using)
+typedef struct {
+  uint8_t opcode;       /* one of ESP_NOW_HOSTED_OP_*                        */
+  uint8_t seq;          /* wraps 0..255; echoed in the response for matching */
+  uint16_t payload_len; /* bytes of opcode-specific payload that follow      */
+  uint8_t payload[];    /* flexible                                          */
+} __attribute__((packed)) esp_now_hosted_req_t;
+
+typedef struct {
+  uint8_t opcode;   /* echoes the request opcode                         */
+  uint8_t seq;      /* echoes the request seq                            */
+  int32_t status;   /* esp_err_t from the native call on the co-processor */
+  uint16_t ret_len; /* bytes of return payload that follow               */
+  uint8_t ret[];    /* flexible (e.g. version u32, is_peer_exist bool)   */
+} __attribute__((packed)) esp_now_hosted_resp_t;
+
+/* ── Opcode payloads ────────────────────────────────────────────────────── */
+
+/* esp_now_peer_info_t minus the host-only `priv` pointer, which is meaningless
+ * across the transport and never set by ESPHome's espnow component.           */
+typedef struct {
+  uint8_t peer_addr[6];
+  uint8_t lmk[16];
+  uint8_t channel; /* 0 = current channel                              */
+  uint8_t ifidx;   /* wifi_interface_t (0=STA, 1=AP)                    */
+  uint8_t encrypt; /* bool                                             */
+} __attribute__((packed)) esp_now_hosted_peer_t;
+
+typedef struct {
+  uint8_t has_addr; /* 0 => peer_addr is NULL (broadcast to all peers)   */
+  uint8_t peer_addr[6];
+  uint16_t data_len;
+  uint8_t data[]; /* flexible, up to ESP_NOW_HOSTED_MAX_FRAME          */
+} __attribute__((packed)) esp_now_hosted_send_req_t;
+
+/* ── Async events (device -> host) ──────────────────────────────────────── */
+
+/* Reconstructed on the host into an esp_now_recv_info_t + a minimal
+ * wifi_pkt_rx_ctrl_t. ESPHome's espnow reads info->src_addr, info->des_addr,
+ * info->rx_ctrl->rssi and info->rx_ctrl->timestamp.                           */
+typedef struct {
+  uint8_t src_addr[6];
+  uint8_t des_addr[6];
+  int8_t rssi;
+  uint8_t channel;
+  uint16_t data_len;
+  uint8_t data[]; /* flexible                                          */
+} __attribute__((packed)) esp_now_hosted_recv_evt_t;
+
+typedef struct {
+  uint8_t des_addr[6];
+  uint8_t status; /* esp_now_send_status_t (0 = success)               */
+} __attribute__((packed)) esp_now_hosted_send_evt_t;
+// NOLINTEND(modernize-use-using)
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* ESP_NOW_HOSTED_RPC_H */

esphome/components/espnow/__init__.py

@@ -3,6 +3,7 @@ from typing import Any
 from esphome import automation, core
 import esphome.codegen as cg
 from esphome.components import wifi
+from esphome.components.esp32 import VARIANT_ESP32P4, get_esp32_variant
 from esphome.components.udp import CONF_ON_RECEIVE
 import esphome.config_validation as cv
 from esphome.const import (
@@ -17,6 +18,7 @@ from esphome.const import (
 )
 from esphome.core import CORE, HexInt
 from esphome.cpp_generator import MockObj, TemplateArgsType
+import esphome.final_validate as fv
 from esphome.types import ConfigType
 
 CODEOWNERS = ["@jesserockz"]
@@ -132,6 +134,24 @@ CONFIG_SCHEMA = cv.All(
 )
 
 
+def _validate_variant(config: ConfigType) -> ConfigType:
+    # ESP-NOW rides the Wi-Fi PHY. Radio-less esp32 variants have no native
+    # ESP-NOW; only the ESP32-P4 has a path, via the esp32_hosted shim that
+    # supplies the esp_now_* symbols. Fail here with a clear message instead of
+    # letting the build reach an "undefined reference to esp_now_*" link error.
+    variant = get_esp32_variant()
+    if wifi.variant_has_wifi(variant):
+        return config
+    if variant != VARIANT_ESP32P4:
+        raise cv.Invalid(f"ESP-NOW is not supported on {variant} (no Wi-Fi radio)")
+    if "esp32_hosted" not in fv.full_config.get():
+        raise cv.Invalid(f"ESP-NOW on {variant} requires the esp32_hosted component")
+    return config
+
+
+FINAL_VALIDATE_SCHEMA = _validate_variant
+
+
 async def _trigger_to_code(config: ConfigType) -> MockObj:
     if address := config.get(CONF_ADDRESS):
         address = address.parts

esphome/components/mdns/__init__.py

@@ -192,6 +192,8 @@ async def to_code(config: ConfigType) -> None:
     if CORE.using_arduino:
         if CORE.is_esp8266:
             cg.add_library("ESP8266mDNS", None)
+            # No MDNS global in the build; mdns_esp8266.cpp owns a guarded MDNSResponder
+            cg.add_build_flag("-DNO_GLOBAL_MDNS")
         elif CORE.is_rp2:
             cg.add_library("LEAmDNS", None)
 

esphome/components/mdns/mdns_esp8266.cpp

@@ -13,8 +13,47 @@
 
 namespace esphome::mdns {
 
+// Main-loop calls into LEAmDNS that send (update() and close(); begin(), addService() and
+// the scheduled restart never reach a send) can yield inside UdpContext::sendTimeout(); a
+// packet arriving then re-enters LEAmDNS from lwIP on the same UdpContext and both sides
+// free the same tx pbufs (#18760). Received packets stay queued during such a call and are
+// processed from the main loop afterwards.
+class GuardedMDNSResponder : public ::esp8266::MDNSImplementation::MDNSResponder {
+ public:
+  void update_guarded() { this->run_guarded_(&GuardedMDNSResponder::update); }
+  void close_guarded() { this->run_guarded_(&GuardedMDNSResponder::close); }
+
+ private:
+  void run_guarded_(bool (GuardedMDNSResponder::*fn)()) {
+    UdpContext *ctx = this->m_pUDPContext;
+    if (ctx == nullptr) {
+      (this->*fn)();
+      return;
+    }
+    // Set every time: a restart replaces the context together with its stock handler. Only
+    // begin() and the scheduled netif callback restart, never update() or close(), so the
+    // context cannot change underneath this call.
+    ctx->onRx([this]() {
+      if (!this->in_loop_call_) {
+        this->_callProcess();
+      }
+    });
+    this->in_loop_call_ = true;
+    (this->*fn)();
+    // close() releases the context; a yield in here queues further packets for this loop too
+    while (this->m_pUDPContext != nullptr && this->m_pUDPContext->next()) {
+      this->_parseMessage();
+    }
+    this->in_loop_call_ = false;
+  }
+
+  volatile bool in_loop_call_{false};
+};
+
+static GuardedMDNSResponder mdns_responder;  // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
+
 static void register_esp8266(MDNSComponent *, StaticVector<MDNSService, MDNS_SERVICE_COUNT> &services) {
-  MDNS.begin(App.get_name().c_str());
+  mdns_responder.begin(App.get_name().c_str());
 
   for (const auto &service : services) {
     // Strip the leading underscore from the proto and service_type. While it is
@@ -30,10 +69,10 @@ static void register_esp8266(MDNSComponent *, StaticVector<MDNSService, MDNS_SER
       service_type++;
     }
     uint16_t port = service.port.value();
-    MDNS.addService(FPSTR(service_type), FPSTR(proto), port);
+    mdns_responder.addService(FPSTR(service_type), FPSTR(proto), port);
     for (const auto &record : service.txt_records) {
-      MDNS.addServiceTxt(FPSTR(service_type), FPSTR(proto), FPSTR(MDNS_STR_ARG(record.key)),
-                         FPSTR(MDNS_STR_ARG(record.value)));
+      mdns_responder.addServiceTxt(FPSTR(service_type), FPSTR(proto), FPSTR(MDNS_STR_ARG(record.key)),
+                                   FPSTR(MDNS_STR_ARG(record.value)));
     }
   }
 }
@@ -52,7 +91,7 @@ void MDNSComponent::start_polling_window_() {
     if (wifi->is_roaming() || (!wifi->is_connected() && !wifi->is_ap_active()))
       return;
 #endif
-    MDNS.update();
+    mdns_responder.update_guarded();
   });
   this->set_timeout(MDNS_POLL_STOP_ID, MDNS_POLL_WINDOW_MS, [this]() { this->cancel_interval(MDNS_POLL_ID); });
 }
@@ -81,7 +120,7 @@ void MDNSComponent::on_ip_state(const network::IPAddresses &ips, const network::
 #endif
 
 void MDNSComponent::on_shutdown() {
-  MDNS.close();
+  mdns_responder.close_guarded();
   delay(10);
 }
 

esphome/components/noise/__init__.py

@@ -88,12 +88,12 @@ def encryption_schema(config: ConfigType | None) -> ConfigType:
 
 async def to_code(config: ConfigType) -> None:
     cg.add_define("USE_NOISE")
-    cg.add_library("esphome/noise-c", "0.1.21")
+    cg.add_library("esphome/noise-c", "0.1.24")
     # noise-c depends on libsodium, but declaring it here too lets the
     # library manager see the full set up front instead of discovering
     # libsodium only after noise-c has downloaded, so the two can download
     # in parallel. The version must match noise-c's library.json.
-    cg.add_library("esphome/libsodium", "1.10021.4")
+    cg.add_library("esphome/libsodium", "1.10021.6")
     # Enable optimized memzero/memcmp in libsodium instead of volatile byte loops
     cg.add_build_flag("-DHAVE_WEAK_SYMBOLS=1")
     cg.add_build_flag("-DHAVE_INLINE_ASM=1")

esphome/core/defines.h

@@ -71,6 +71,7 @@
 #define USE_ESP32_HOSTED
 #define USE_ESP32_HOSTED_HTTP_UPDATE
 #define USE_ESP32_IMPROV_STATE_CALLBACK
+#define USE_ESP_NOW_HOSTED
 #define USE_EVENT
 #define USE_FAN
 #define USE_GPIO_BINARY_SENSOR_INTERRUPT

esphome/framework_helpers.py

@@ -23,6 +23,7 @@ from esphome.net_retry import (
 )
 
 if TYPE_CHECKING:
+    from filelock import FileLock
     import requests
 
 PathType = str | os.PathLike
@@ -909,6 +910,61 @@ def _part_path(dest: Path) -> Path:
     return dest.with_name(dest.name + ".part")
 
 
+def downloaded_bytes(dest: Path, size: int | None = None) -> int:
+    """Bytes of ``dest`` on disk (its ``.part`` while streaming), capped at ``size``."""
+    done = 0
+    for candidate in (_part_path(dest), dest):
+        try:
+            done = candidate.stat().st_size
+            break
+        except FileNotFoundError:
+            continue
+    return done if size is None else min(done, size)
+
+
+# Short lock-acquire slices so a waiting worker still observes Ctrl-C
+_DOWNLOAD_LOCK_POLL = 1
+
+# Waiting on another process's download; past this the caller leaves the
+# file to its holder (the later sequential install waits on the same lock)
+DOWNLOAD_LOCK_TIMEOUT = 60
+
+
+class DownloadLockUnavailable(OSError):
+    """The lock file cannot be used at all (a lock-less filesystem)."""
+
+
+def wait_for_download_lock(
+    lock: "FileLock",
+    tracker: Callable[[int], None],
+    on_disk: Callable[[], int],
+    name: str,
+) -> None:
+    """Acquire ``lock``, reporting ``on_disk()`` to ``tracker`` each poll so the
+    bar follows the holder's download. Raises filelock's ``Timeout`` once
+    ``DOWNLOAD_LOCK_TIMEOUT`` seconds pass."""
+    from filelock import Timeout
+
+    deadline = time.monotonic() + DOWNLOAD_LOCK_TIMEOUT
+    waiting = False
+    while True:
+        try:
+            lock.acquire(timeout=_DOWNLOAD_LOCK_POLL)
+            return
+        except Timeout:
+            pass
+        except OSError as err:
+            # Distinct from an OSError out of on_disk(), which must not
+            # read as "locks unsupported"
+            raise DownloadLockUnavailable(*err.args) from err
+        if not waiting:
+            waiting = True
+            _LOGGER.info("Waiting for another process downloading %s", name)
+        tracker(on_disk())  # raises when the batch is cancelled
+        if time.monotonic() >= deadline:
+            raise Timeout(lock.lock_file)
+
+
 def discard_partial_download(dest: Path) -> None:
     """Remove ``dest`` and the resume sidecars of an abandoned download."""
     part = _part_path(dest)
@@ -1319,10 +1375,7 @@ def download_from_mirrors(
             )
             # Tick with the bytes already on disk so a combined bar holds
             # steady during the backoff instead of rewinding to zero
-            done = 0
-            if progress is not None:
-                part = _part_path(path_target)
-                done = part.stat().st_size if part.is_file() else 0
+            done = downloaded_bytes(path_target) if progress is not None else 0
             _cancellable_sleep(delay, progress, done)
 
     # 3. Report every attempted URL if all mirrors failed. failures spans

esphome/platformio/prefetch.py

@@ -33,11 +33,14 @@ import time
 from typing import Any, NamedTuple
 
 from esphome.framework_helpers import (
+    DownloadLockUnavailable,
     content_length,
     discard_partial_download,
+    downloaded_bytes,
     failure_reason,
     resume_fetch_job,
     run_batch_downloads,
+    wait_for_download_lock,
     warn_prefetch_failures,
 )
 from esphome.helpers import get_bool_env, get_usable_cpu_count, rmtree
@@ -61,16 +64,10 @@ _RESOLVE_WORKERS = 8
 # A hung child must not block the build; downloads resume on the next run
 _PREFETCH_TIMEOUT = 20 * 60
 
-# Waiting on another process's URL download; past this, leave it to pio
-_DOWNLOAD_LOCK_TIMEOUT = 60
-
 # Child exit for a handled, already-warned failure; 1 would collide with
 # the interpreter's own import-failure exit
 _EXIT_HANDLED = 3
 
-# Short lock-acquire slices so a waiting worker still observes Ctrl-C
-_URI_LOCK_POLL = 1
-
 # Resolution errored (vs a clean skip); suppresses the warm sentinel
 _RESOLVE_FAILED = object()
 
@@ -462,51 +459,54 @@ def _uri_jobs(
 
 
 def _serialized_fetch_job(
-    dl_path: Path, lock_path: str, body: Any, unlocked_ok: bool = True
+    dl_path: Path,
+    lock_path: str,
+    body: Any,
+    size: int,
+    stream_dest: Path | None = None,
+    unlocked_ok: bool = True,
 ) -> Any:
-    """Wrap ``body`` so the shared destination is single-writer.
-
-    Interleaved writers truncate each other's ``.part`` bytes (see
-    registry.py). The bounded poll observes Ctrl-C via the tracker; a
-    blown deadline is a clean skip (the holder's copy is what the build
-    needs). On a lock-less filesystem a sha256-verified body runs
-    unlocked with one warning; a checksum-less one
-    (``unlocked_ok=False``) is a counted failure instead.
+    """Wrap ``body`` so the shared destination is single-writer (interleaved
+    writers truncate each other's ``.part``, see registry.py). A blown deadline
+    is a clean skip. On a lock-less filesystem a sha256-verified body runs
+    unlocked with one warning; a checksum-less one (``unlocked_ok=False``) fails.
     """
 
+    def on_disk() -> int:
+        # A URL job's holder streams beside the staging path until it
+        # promotes; after that only dl_path is left
+        done = downloaded_bytes(dl_path, size)
+        if not done and stream_dest is not None:
+            done = downloaded_bytes(stream_dest, size)
+        return done
+
     def run(tracker: Any) -> None:
         from filelock import FileLock, Timeout
 
         # fallback_to_soft would leave a stale marker on lock-less
         # filesystems that blocks every later build (see git.py)
         lock = FileLock(lock_path, fallback_to_soft=False)
-        deadline = time.monotonic() + _DOWNLOAD_LOCK_TIMEOUT
-        while True:
-            try:
-                lock.acquire(timeout=_URI_LOCK_POLL)
-                break
-            except Timeout:
-                tracker(0)  # raises when the batch is cancelled
-                if time.monotonic() >= deadline:
-                    # Another process is fetching this same file; its copy
-                    # is what the build needs (a large framework archive
-                    # can hold the lock far longer than this deadline)
-                    _LOGGER.debug("Leaving %s to its current downloader", dl_path.name)
-                    return
-            except OSError as err:
-                if not unlocked_ok:
-                    # A body with no checksum to catch interleaved corruption
-                    raise
-                lock = None
-                _LOGGER.warning(
-                    "Could not lock %s (%s); downloading unlocked",
-                    dl_path.name,
-                    err,
-                )
-                break
+        try:
+            wait_for_download_lock(lock, tracker, on_disk, dl_path.name)
+        except Timeout:
+            # The holder's copy is what the build needs (a large
+            # framework archive can outlast this deadline)
+            _LOGGER.debug("Leaving %s to its current downloader", dl_path.name)
+            return
+        except DownloadLockUnavailable as err:
+            if not unlocked_ok:
+                # A body with no checksum to catch interleaved corruption
+                raise
+            lock = None
+            _LOGGER.warning(
+                "Could not lock %s (%s); downloading unlocked",
+                dl_path.name,
+                err,
+            )
         try:
             if dl_path.is_file():
-                return  # another process finished it while we waited
+                tracker(size)  # another process finished it while we waited
+                return
             body(tracker)
         finally:
             if lock is not None:
@@ -540,6 +540,7 @@ def _registry_fetch_job(
         dl_path,
         f"{dl_path}.esphome.lock",
         resume_fetch_job(url, dl_path, sha256=checksum, size=size),
+        size,
     )
 
     def run(tracker: Any) -> None:
@@ -571,9 +572,9 @@ def _uri_fetch_job(manager: Any, url: str, dl_path: Path, size: int) -> Any:
         tmp.replace(dl_path)
 
     def run(tracker: Any) -> None:
-        _serialized_fetch_job(dl_path, f"{tmp}.lock", promote, unlocked_ok=False)(
-            tracker
-        )
+        _serialized_fetch_job(
+            dl_path, f"{tmp}.lock", promote, size, tmp, unlocked_ok=False
+        )(tracker)
         if dl_path.is_file():
             # Won or lost, the race is over; staging files left behind
             # are dead weight PlatformIO's cache never prunes

esphome/platformio/registry.py

@@ -17,8 +17,10 @@ from esphome.framework_helpers import (
     archive_extract_all,
     download_from_mirrors,
     download_with_resume,
+    downloaded_bytes,
     rmdir,
     run_batch_downloads,
+    wait_for_download_lock,
 )
 from esphome.net_retry import fetch_with_retry, http_request
 
@@ -164,11 +166,17 @@ class _PendingArchive(NamedTuple):
     name: str
     version: str
     dest: Path
+    archive: Path
     url: str
     sha256: str
     size: int
 
 
+def _archive_path(downloads_dir: Path, name: str, version: str) -> Path:
+    """The one archive path the prefetch and the sequential install share."""
+    return downloads_dir / f"{name}-{version}"
+
+
 def _already_installed(dest: Path) -> bool:
     """Whether ``dest`` holds a completed install (extraction marker)."""
     return (dest / ".esphome_extracted").is_file()
@@ -187,18 +195,18 @@ def prefetch_packages(
     lock as ``install_package``: the archive's ``.part`` file is shared, and
     two concurrent writers would truncate each other's bytes.
     """
-    from filelock import FileLock
+    from filelock import FileLock, Timeout
 
     pending: list[_PendingArchive] = []
-    seen: set[str] = set()
+    seen: set[Path] = set()
     for name, version, dest, mirrors in packages:
         if mirrors or (dest / ".esphome_extracted").is_file():
             continue
-        archive_name = f"{name}-{version}"
-        if archive_name in seen:
+        archive = _archive_path(downloads_dir, name, version)
+        if archive in seen:
             # A duplicate entry would race itself between two workers
             continue
-        seen.add(archive_name)
+        seen.add(archive)
         try:
             url, sha256, size = registry_download(name, version)
         except EsphomeError as err:
@@ -207,10 +215,9 @@ def prefetch_packages(
             continue
         if not size:
             continue
-        archive = downloads_dir / archive_name
         if archive.is_file() and archive.stat().st_size == size:
             continue
-        pending.append(_PendingArchive(name, version, dest, url, sha256, size))
+        pending.append(_PendingArchive(name, version, dest, archive, url, sha256, size))
     if len(pending) < 2:
         return
     downloads_dir.mkdir(parents=True, exist_ok=True)
@@ -222,20 +229,36 @@ def prefetch_packages(
 
     def _fetch(entry: _PendingArchive, tracker: Callable[[int], None]) -> None:
         entry.dest.parent.mkdir(parents=True, exist_ok=True)
-        with FileLock(f"{entry.dest}.lock", fallback_to_soft=False):
-            # Marker re-check: a concurrent build may have installed (and
-            # deleted the archive of) this package while we waited;
-            # re-downloading would orphan a fresh copy in downloads_dir
-            # no branch: the thread tracer misses the skip edge; both
-            # arms of _already_installed are pinned directly
-            if not _already_installed(entry.dest):  # pragma: no branch
-                download_with_resume(
-                    entry.url,
-                    downloads_dir / f"{entry.name}-{entry.version}",
-                    sha256=entry.sha256,
-                    size=entry.size,
-                    progress=tracker,
-                )
+
+        def on_disk() -> int:
+            if done := downloaded_bytes(entry.archive, entry.size):
+                return done
+            # The holder deletes the archive once it has installed it
+            return entry.size if _already_installed(entry.dest) else 0
+
+        lock = FileLock(f"{entry.dest}.lock", fallback_to_soft=False)
+        try:
+            wait_for_download_lock(lock, tracker, on_disk, entry.name)
+        except Timeout:
+            # install_package waits on this same lock and verifies the
+            # holder's copy
+            _LOGGER.debug("Leaving %s to its current downloader", entry.name)
+            return
+        try:
+            if _already_installed(entry.dest):
+                # A concurrent build installed it while we waited; a
+                # re-download would orphan a fresh copy in downloads_dir
+                tracker(entry.size)
+                return
+            download_with_resume(
+                entry.url,
+                entry.archive,
+                sha256=entry.sha256,
+                size=entry.size,
+                progress=tracker,
+            )
+        finally:
+            lock.release()
 
     failures = run_batch_downloads(
         "Downloading packages",
@@ -288,7 +311,7 @@ def install_package(
         rmdir(dest, msg=f"Clean up incomplete {name} install")
         # Persistent location so an interrupted download resumes across runs.
         downloads_dir.mkdir(parents=True, exist_ok=True)
-        archive = downloads_dir / f"{name}-{version}"
+        archive = _archive_path(downloads_dir, name, version)
         _LOGGER.info("Downloading %s %s ...", name, version)
         if mirrors:
             _LOGGER.warning(

platformio.ini

@@ -45,7 +45,7 @@ lib_deps_base =
 lib_deps =
     ${common.lib_deps_base}
     https://github.com/dudanov/MideaUART.git#eeea6c3e9b4474f067054592b435be1c4e466815 ; midea
-    esphome/noise-c@0.1.21                 ; noise (api, ota)
+    esphome/noise-c@0.1.24                 ; noise (api, ota)
     improv/Improv@1.2.7                    ; improv_serial / esp32_improv
     kikuchan98/pngle@1.1.0                 ; online_image
     ; Using the repository directly, otherwise ESP-IDF can't use the library
@@ -244,7 +244,7 @@ lib_deps =
     ${common:idf-component-libs.lib_deps}
     ESP32Async/ESPAsyncWebServer@3.9.6    ; web_server_base
     droscy/esp_wireguard@0.4.5            ; wireguard
-    esphome/noise-c@0.1.21                ; noise (api, ota)
+    esphome/noise-c@0.1.24                ; noise (api, ota)
     ESP32Async/AsyncTCP@3.4.5             ; async_tcp
     DNSServer                             ; captive_portal
     heman/AsyncMqttClient-esphome@2.0.0   ; mqtt
@@ -641,7 +641,7 @@ build_unflags =
 extends = common
 platform = platformio/native
 lib_deps =
-    esphome/noise-c@0.1.21  ; used by noise (api, ota)
+    esphome/noise-c@0.1.24  ; used by noise (api, ota)
     lvgl/lvgl@9.5.0         ; lvgl
 build_flags =
     ${common.build_flags}

script/api_protobuf/api_protobuf.py

@@ -131,6 +131,12 @@ def force_str(force: bool) -> str:
     return str(force).lower()
 
 
+def _encode_call(func: str, *args: str, force: bool = False) -> str:
+    """Emit one ProtoEncode call; every helper takes the cursor and returns it advanced."""
+    suffix = "_force" if force else ""
+    return f"pos = ProtoEncode::{func}{suffix}({', '.join(('pos', *args))});"
+
+
 class TypeInfo(ABC):
     """Base class for all type information."""
 
@@ -264,14 +270,16 @@ class TypeInfo(ABC):
     # write_raw_byte(tag) + raw encode instead of the full encode_* method,
     # eliminating the zero-check branch and encode_field_raw indirection.
     # {value} is replaced with the actual field expression.
-    RAW_ENCODE_MAP: dict[str, str] = {
-        "encode_uint32": "ProtoEncode::encode_varint_raw(pos, {value});",
-        "encode_uint64": "ProtoEncode::encode_varint_raw_64(pos, {value});",
-        "encode_sint32": "ProtoEncode::encode_varint_raw_short(pos, encode_zigzag32({value}));",
-        "encode_sint64": "ProtoEncode::encode_varint_raw_64(pos, encode_zigzag64({value}));",
-        "encode_int64": "ProtoEncode::encode_varint_raw_64(pos, static_cast<uint64_t>({value}));",
-        "encode_bool": "ProtoEncode::write_raw_byte(pos, {value} ? 0x01 : 0x00);",
+    RAW_ENCODE_MAP: dict[str, tuple[str, str]] = {
+        "encode_uint32": ("encode_varint_raw", "{value}"),
+        "encode_uint64": ("encode_varint_raw_64", "{value}"),
+        "encode_sint32": ("encode_varint_raw_short", "encode_zigzag32({value})"),
+        "encode_sint64": ("encode_varint_raw_64", "encode_zigzag64({value})"),
+        "encode_int64": ("encode_varint_raw_64", "static_cast<uint64_t>({value})"),
+        "encode_bool": ("write_raw_byte", "{value} ? 0x01 : 0x00"),
     }
+    # Fixed32 value expression for the shared tag+fixed32 writer; None for other wire types
+    fixed32_value_template: str | None = None
 
     def _encode_with_precomputed_tag(self, value_expr: str) -> str | None:
         """Try to emit a precomputed-tag encode for a field.
@@ -288,12 +296,17 @@ class TypeInfo(ABC):
             return None
         max_val = self.max_value
         # Only use RAW_ENCODE_MAP for forced fields or fields with max_value
-        raw_expr = None
+        raw = None
         if self.force or max_val is not None:
-            raw_expr = self.RAW_ENCODE_MAP.get(self.encode_func)
-        if raw_expr is None:
+            raw = self.RAW_ENCODE_MAP.get(self.encode_func)
+        if raw is None:
             return None
-        body = f"ProtoEncode::write_raw_byte(pos, {tag});\n{raw_expr.format(value=value_expr)}"
+        func, arg = raw
+        body = (
+            _encode_call("write_raw_byte", str(tag))
+            + "\n"
+            + _encode_call(func, arg.format(value=value_expr))
+        )
         if self.force:
             return body
         # Non-forced with max_value: inline zero-check + raw encode
@@ -314,23 +327,43 @@ class TypeInfo(ABC):
             return None
         # When max_len < 128, length varint is always 1 byte
         len_encode = (
-            f"ProtoEncode::write_raw_byte(pos, static_cast<uint8_t>({len_expr}));"
+            _encode_call("write_raw_byte", f"static_cast<uint8_t>({len_expr})")
             if max_len is not None and max_len < 128
-            else f"ProtoEncode::encode_varint_raw(pos, {len_expr});"
+            else _encode_call("encode_varint_raw", len_expr)
         )
+        return "\n".join(
+            (
+                _encode_call("write_raw_byte", str(tag)),
+                len_encode,
+                _encode_call("encode_raw", data_expr, len_expr),
+            )
+        )
+
+    def _encode_fixed32_with_precomputed_tag(self, value_expr: str) -> str | None:
+        """Single-byte tag fixed32 write, or None for multi-byte tags."""
+        tag = self.calculate_tag()
+        if tag >= 128:
+            return None
+        if self.force:
+            return _encode_call("write_tag_and_fixed32", str(tag), value_expr)
         return (
-            f"ProtoEncode::write_raw_byte(pos, {tag});\n"
-            f"{len_encode}\n"
-            f"ProtoEncode::encode_raw(pos, {data_expr}, {len_expr});"
+            f"if (uint32_t raw = {value_expr}; raw != 0) [[likely]] {{\n"
+            f"  {_encode_call('write_tag_and_fixed32', str(tag), 'raw')}\n"
+            "}"
         )
 
     @property
     def encode_content(self) -> str:
-        if result := self._encode_with_precomputed_tag(f"this->{self.field_name}"):
+        value = f"this->{self.field_name}"
+        if result := self._encode_with_precomputed_tag(value):
             return result
-        if self.force:
-            return f"ProtoEncode::{self.encode_func}(pos, {self.number}, this->{self.field_name}, true);"
-        return f"ProtoEncode::{self.encode_func}(pos, {self.number}, this->{self.field_name});"
+        if self.fixed32_value_template is not None and (
+            result := self._encode_fixed32_with_precomputed_tag(
+                self.fixed32_value_template.format(value=value)
+            )
+        ):
+            return result
+        return _encode_call(self.encode_func, str(self.number), value, force=self.force)
 
     encode_func = None
 
@@ -635,6 +668,8 @@ class FloatType(FixedSizeTypeMixin, TypeInfo):
     encode_func = "encode_float"
     wire_type = WireType.FIXED32  # Uses wire type 5
 
+    fixed32_value_template = "float_to_raw({value})"
+
     def dump(self, name: str) -> str:
         o = f'snprintf(buffer, sizeof(buffer), "%g", {name});\n'
         o += "out.append(buffer);"
@@ -697,11 +732,11 @@ class UInt64Type(VarintTypeMixin, TypeInfo):
         return self._get_simple_size_calculation(name, force, "uint64")
 
     @property
-    def RAW_ENCODE_MAP(self) -> dict[str, str]:  # noqa: N802
+    def RAW_ENCODE_MAP(self) -> dict[str, tuple[str, str]]:  # noqa: N802
         if self.mac_address:
             return {
                 **TypeInfo.RAW_ENCODE_MAP,
-                "encode_uint64": "ProtoEncode::encode_varint_raw_48bit(pos, {value});",
+                "encode_uint64": ("encode_varint_raw_48bit", "{value}"),
             }
         return TypeInfo.RAW_ENCODE_MAP
 
@@ -769,15 +804,7 @@ class Fixed32Type(FixedSizeTypeMixin, TypeInfo):
         o += "out.append(buffer);"
         return o
 
-    @property
-    def encode_content(self) -> str:
-        tag = self.calculate_tag()
-        if self.force and tag < 128:
-            # Emit combined tag+value write: precomputed tag + direct memcpy
-            return f"ProtoEncode::write_tag_and_fixed32(pos, {tag}, this->{self.field_name});"
-        if self.force:
-            return f"ProtoEncode::{self.encode_func}(pos, {self.number}, this->{self.field_name}, true);"
-        return f"ProtoEncode::{self.encode_func}(pos, {self.number}, this->{self.field_name});"
+    fixed32_value_template = "{value}"
 
     def get_size_calculation(self, name: str, force: bool = False) -> str:
         field_id_size = self.calculate_field_id_size()
@@ -851,9 +878,12 @@ class StringType(TypeInfo):
             f"this->{self.field_name}_ref_.size()",
         ):
             return result
-        if self.force:
-            return f"ProtoEncode::encode_string(pos, {self.number}, this->{self.field_name}_ref_, true);"
-        return f"ProtoEncode::encode_string(pos, {self.number}, this->{self.field_name}_ref_);"
+        return _encode_call(
+            "encode_string",
+            str(self.number),
+            f"this->{self.field_name}_ref_",
+            force=self.force,
+        )
 
     def dump(self, name):
         # If name is 'it', this is a repeated field element - always use string
@@ -951,7 +981,9 @@ class MessageType(TypeInfo):
     @property
     def encode_content(self) -> str:
         # Sub-message encoding needs buffer for backpatch/sync
-        return f"ProtoEncode::{self.encode_func}(pos, buffer, {self.number}, this->{self.field_name});"
+        return _encode_call(
+            self.encode_func, "buffer", str(self.number), f"this->{self.field_name}"
+        )
 
     @property
     def decode_length(self) -> str:
@@ -1058,9 +1090,13 @@ class BytesType(TypeInfo):
             f"this->{self.field_name}_ptr_", f"this->{self.field_name}_len_"
         ):
             return result
-        if self.force:
-            return f"ProtoEncode::encode_bytes(pos, {self.number}, this->{self.field_name}_ptr_, this->{self.field_name}_len_, true);"
-        return f"ProtoEncode::encode_bytes(pos, {self.number}, this->{self.field_name}_ptr_, this->{self.field_name}_len_);"
+        return _encode_call(
+            "encode_bytes",
+            str(self.number),
+            f"this->{self.field_name}_ptr_",
+            f"this->{self.field_name}_len_",
+            force=self.force,
+        )
 
     def dump(self, name: str) -> str:
         ptr_dump = f"format_hex_pretty(this->{self.field_name}_ptr_, this->{self.field_name}_len_)"
@@ -1170,9 +1206,13 @@ class PointerToBytesBufferType(PointerToBufferTypeBase):
             f"this->{self.field_name}", f"this->{self.field_name}_len"
         ):
             return result
-        if self.force:
-            return f"ProtoEncode::encode_bytes(pos, {self.number}, this->{self.field_name}, this->{self.field_name}_len, true);"
-        return f"ProtoEncode::encode_bytes(pos, {self.number}, this->{self.field_name}, this->{self.field_name}_len);"
+        return _encode_call(
+            "encode_bytes",
+            str(self.number),
+            f"this->{self.field_name}",
+            f"this->{self.field_name}_len",
+            force=self.force,
+        )
 
     @property
     def decode_length_content(self) -> str | None:
@@ -1224,16 +1264,19 @@ class PointerToStringBufferType(PointerToBufferTypeBase):
         if max_len is not None and max_len < 128 and self.force:
             tag = self.calculate_tag()
             if tag < 128:
-                return f"ProtoEncode::encode_short_string_force(pos, {tag}, this->{self.field_name});"
+                return _encode_call(
+                    "encode_short_string_force", str(tag), f"this->{self.field_name}"
+                )
         if result := self._encode_bytes_with_precomputed_tag(
             f"this->{self.field_name}.c_str()",
             f"this->{self.field_name}.size()",
         ):
             return result
-        if self.force:
-            return f"ProtoEncode::encode_string(pos, {self.number}, this->{self.field_name}, true);"
-        return (
-            f"ProtoEncode::encode_string(pos, {self.number}, this->{self.field_name});"
+        return _encode_call(
+            "encode_string",
+            str(self.number),
+            f"this->{self.field_name}",
+            force=self.force,
         )
 
     @property
@@ -1421,9 +1464,13 @@ class FixedArrayBytesType(TypeInfo):
             f"this->{self.field_name}", f"this->{self.field_name}_len", max_len=max_len
         ):
             return result
-        if self.force:
-            return f"ProtoEncode::encode_bytes(pos, {self.number}, this->{self.field_name}, this->{self.field_name}_len, true);"
-        return f"ProtoEncode::encode_bytes(pos, {self.number}, this->{self.field_name}, this->{self.field_name}_len);"
+        return _encode_call(
+            "encode_bytes",
+            str(self.number),
+            f"this->{self.field_name}",
+            f"this->{self.field_name}_len",
+            force=self.force,
+        )
 
     def dump(self, name: str) -> str:
         return f"out.append(format_hex_pretty({name}, {name}_len));"
@@ -1520,9 +1567,9 @@ class EnumType(VarintTypeMixin, TypeInfo):
     @property
     def encode_content(self) -> str:
         value_expr = f"static_cast<uint32_t>(this->{self.field_name})"
-        if self.force:
-            return f"ProtoEncode::{self.encode_func}(pos, {self.number}, {value_expr}, true);"
-        return f"ProtoEncode::{self.encode_func}(pos, {self.number}, {value_expr});"
+        return _encode_call(
+            self.encode_func, str(self.number), value_expr, force=self.force
+        )
 
     def dump(self, name: str) -> str:
         return f"out.append_p(proto_enum_to_string<{self.cpp_type}>({name}));"
@@ -1701,9 +1748,9 @@ def _generate_inline_encode_block(
 
     lines = []
     lines.append(f"auto &sub_msg = {element};")
-    lines.append(f"ProtoEncode::write_raw_byte(pos, {tag});")
+    lines.append(_encode_call("write_raw_byte", str(tag)))
     lines.append("uint8_t *len_pos = pos;")
-    lines.append("ProtoEncode::reserve_byte(pos);")
+    lines.append(_encode_call("reserve_byte"))
 
     # Generate inline field encoding for each sub-message field
     for field in sub_desc.field:
@@ -1775,17 +1822,22 @@ class FixedArrayRepeatedType(TypeInfo):
     def _encode_element(self, element: str) -> str:
         """Helper to generate encode statement for a single element."""
         if isinstance(self._ti, EnumType):
-            return f"ProtoEncode::{self._ti.encode_func}(pos, {self.number}, static_cast<uint32_t>({element}), true);"
+            return _encode_call(
+                self._ti.encode_func,
+                str(self.number),
+                f"static_cast<uint32_t>({element})",
+                force=True,
+            )
         # Repeated message elements use encode_sub_message (force=true is default)
         if isinstance(self._ti, MessageType):
             if _is_inline_encode(self._ti.cpp_type):
                 return _generate_inline_encode_block(
                     self.number, self._ti.cpp_type, element
                 )
-            return f"ProtoEncode::encode_sub_message(pos, buffer, {self.number}, {element});"
-        return (
-            f"ProtoEncode::{self._ti.encode_func}(pos, {self.number}, {element}, true);"
-        )
+            return _encode_call(
+                "encode_sub_message", "buffer", str(self.number), element
+            )
+        return _encode_call(self._ti.encode_func, str(self.number), element, force=True)
 
     @property
     def cpp_type(self) -> str:
@@ -2137,13 +2189,18 @@ class RepeatedTypeInfo(TypeInfo):
     def _encode_element_call(self, element: str) -> str:
         """Helper to generate encode call for a single element."""
         if isinstance(self._ti, EnumType):
-            return f"ProtoEncode::{self._ti.encode_func}(pos, {self.number}, static_cast<uint32_t>({element}), true);"
+            return _encode_call(
+                self._ti.encode_func,
+                str(self.number),
+                f"static_cast<uint32_t>({element})",
+                force=True,
+            )
         # Repeated message elements use encode_sub_message (force=true is default)
         if isinstance(self._ti, MessageType):
-            return f"ProtoEncode::encode_sub_message(pos, buffer, {self.number}, {element});"
-        return (
-            f"ProtoEncode::{self._ti.encode_func}(pos, {self.number}, {element}, true);"
-        )
+            return _encode_call(
+                "encode_sub_message", "buffer", str(self.number), element
+            )
+        return _encode_call(self._ti.encode_func, str(self.number), element, force=True)
 
     @property
     def encode_content(self) -> str:
@@ -2152,7 +2209,7 @@ class RepeatedTypeInfo(TypeInfo):
             # Special handling for const char* elements (when container_no_template contains "const char")
             if "const char" in self._container_no_template:
                 o = f"for (const char *it : *this->{self.field_name}) {{\n"
-                o += f"  ProtoEncode::{self._ti.encode_func}(pos, {self.number}, it, strlen(it), true);\n"
+                o += f"  {_encode_call(self._ti.encode_func, str(self.number), 'it', 'strlen(it)', force=True)}\n"
             else:
                 o = f"for (const auto &it : *this->{self.field_name}) {{\n"
                 o += f"  {self._encode_element_call('it')}\n"
@@ -2784,28 +2841,36 @@ def build_message_type(
             )
             for line in encode
         ]
-        o = f"{speed_attr}uint8_t *{desc.name}::encode(ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_PARAM) const {{\n"
+        o = f"{speed_attr}uint8_t *{desc.name}::encode_msg(const void *self, ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_PARAM) {{\n"
+        o += f"  const auto &msg = *static_cast<const {desc.name} *>(self);\n"
         o += "  uint8_t *__restrict__ pos = buffer.get_pos();\n"
-        o += indent("\n".join(encode_debug)) + "\n"
+        o += indent("\n".join(encode_debug)).replace("this->", "msg.") + "\n"
         o += "  return pos;\n"
         o += "}\n"
         cpp += o
-        prot = (
-            "uint8_t *encode(ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_PARAM) const;"
+        public_content.append(
+            "static uint8_t *encode_msg(const void *self, ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_PARAM);"
+        )
+        public_content.append(
+            "uint8_t *encode(ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_PARAM) const {\n"
+            "  return encode_msg(this, buffer PROTO_ENCODE_DEBUG_ARG);\n"
+            "}"
         )
-        public_content.append(prot)
     # If no fields to encode or message doesn't need encoding, the default implementation in ProtoMessage will be used
 
     # Add calculate_size method only if this message needs encoding and has fields
     if needs_encode and size_calc and not is_inline_only:
-        o = f"{speed_attr}uint32_t {desc.name}::calculate_size() const {{\n"
+        o = f"{speed_attr}uint32_t {desc.name}::calc_size_msg(const void *self) {{\n"
+        o += f"  const auto &msg = *static_cast<const {desc.name} *>(self);\n"
         o += "  uint32_t size = 0;\n"
-        o += indent("\n".join(size_calc)) + "\n"
+        o += indent("\n".join(size_calc)).replace("this->", "msg.") + "\n"
         o += "  return size;\n"
         o += "}\n"
         cpp += o
-        prot = "uint32_t calculate_size() const;"
-        public_content.append(prot)
+        public_content.append("static uint32_t calc_size_msg(const void *self);")
+        public_content.append(
+            "uint32_t calculate_size() const { return calc_size_msg(this); }"
+        )
     # If no fields to calculate size for or message doesn't need encoding, the default implementation in ProtoMessage will be used
 
     # dump_to method declaration in header

script/ci-custom.py

@@ -294,6 +294,9 @@ def highlight(s):
         "esphome/components/socket/headers.h",
         "esphome/core/defines.h",
         "esphome/components/http_request/httplib.h",
+        # Shared C wire header (byte-identical with the co-processor firmware);
+        # these are protocol constants and constexpr is C++-only.
+        "esphome/components/esp32_hosted/esp_now_hosted_rpc.h",
     ],
 )
 def lint_no_defines(fname, match):
@@ -816,6 +819,10 @@ def lint_relative_py_import(fname: Path, line, col, content):
         "esphome/components/host/helpers.cpp",
         "esphome/components/zephyr/helpers.cpp",
         "esphome/components/http_request/httplib.h",
+        # Global extern "C" esp_now_* linker symbols + shared C wire header;
+        # neither can live in a C++ namespace.
+        "esphome/components/esp32_hosted/esp_now_hosted.cpp",
+        "esphome/components/esp32_hosted/esp_now_hosted_rpc.h",
     ],
 )
 def lint_namespace(fname: Path, content: str) -> str | None:
@@ -841,7 +848,15 @@ def lint_esphome_h(fname, line, col, content):
     )
 
 
-@lint_content_check(include=["*.h"], exclude=["esphome/core/entity_types.h"])
+@lint_content_check(
+    include=["*.h"],
+    exclude=[
+        "esphome/core/entity_types.h",
+        # Shared C wire header; uses a classic #ifndef guard for portability
+        # across the co-processor firmware repo it stays byte-identical with.
+        "esphome/components/esp32_hosted/esp_now_hosted_rpc.h",
+    ],
+)
 def lint_pragma_once(fname, content):
     if "#pragma once" not in content:
         return (

tests/components/api/test_proto_mac_varint.cpp

@@ -59,7 +59,7 @@ static void verify_mac(uint64_t mac, size_t expected_bytes) {
 #ifdef ESPHOME_DEBUG_API
   uint8_t *proto_debug_end_ = api_buf.data() + api_buf.size();
 #endif
-  ProtoEncode::encode_varint_raw_48bit(pos PROTO_ENCODE_DEBUG_ARG, mac);
+  pos = ProtoEncode::encode_varint_raw_48bit(pos PROTO_ENCODE_DEBUG_ARG, mac);
   size_t new_len = pos - api_buf.data();
 
   EXPECT_EQ(new_len, expected_bytes) << "mac=0x" << std::hex << mac << std::dec;

tests/components/esp32_hosted/test-espnow.esp32-p4-idf.yaml

@@ -0,0 +1,5 @@
+# Exercises the ESP-NOW-over-hosted shim: on the ESP32-P4 host, esp32_hosted
+# supplies the esp_now_* symbols that the espnow component links against.
+packages:
+  esp32_hosted: !include common.yaml
+  espnow: !include ../espnow/common.yaml

tests/integration/fixtures/api_empty_message_roundtrip.yaml

@@ -0,0 +1,11 @@
+esphome:
+  name: api-empty-message-test
+host:
+api:
+logger:
+  level: DEBUG
+
+switch:
+  - platform: template
+    name: "Empty Message Switch"
+    optimistic: true

tests/integration/fixtures/api_encode_boundaries.yaml

@@ -0,0 +1,58 @@
+esphome:
+  name: api-encode-boundaries-test
+  # Top-level area fills DeviceInfoResponse.suggested_area (field 16, a two-byte tag)
+  area:
+    id: kitchen_area
+    name: Kitchen
+  on_boot:
+    - sensor.template.publish:
+        id: zero_then_value
+        state: 0.0
+
+host:
+api:
+logger:
+  level: DEBUG
+
+sensor:
+  - platform: template
+    name: "Zero Then Value"
+    id: zero_then_value
+    # Negative int32 takes the ten byte varint path
+    accuracy_decimals: -2
+    update_interval: never
+
+text_sensor:
+  - platform: template
+    name: "Long Text"
+    id: long_text
+    update_interval: never
+
+number:
+  - platform: template
+    name: "Negative Number"
+    optimistic: true
+    min_value: -1000
+    max_value: 1000
+    step: 0.5
+    initial_value: -123.5
+
+select:
+  - platform: template
+    name: "Long Option Select"
+    optimistic: true
+    options:
+      - short
+      - "option-with-a-name-long-enough-that-its-length-prefix-needs-two-varint-bytes-when-the-list-entities-response-is-encoded-xxxxxxxxxx"
+    initial_option: short
+
+button:
+  - platform: template
+    name: "Publish Values"
+    on_press:
+      - sensor.template.publish:
+          id: zero_then_value
+          state: 12.5
+      - text_sensor.template.publish:
+          id: long_text
+          state: !lambda return std::string(200, 'y');

tests/integration/state_utils.py

@@ -57,6 +57,43 @@ async def wait_for_state(
     return await asyncio.wait_for(future, timeout=timeout)
 
 
+class StateWaiter:
+    """Route one state subscription to any number of predicate waits."""
+
+    def __init__(self) -> None:
+        self._waiters: list[
+            tuple[Callable[[EntityState], bool], asyncio.Future[EntityState]]
+        ] = []
+
+    def on_state(self, state: EntityState) -> None:
+        for predicate, future in self._waiters:
+            if future.done():
+                continue
+            try:
+                matched = predicate(state)
+            except Exception as exc:  # noqa: BLE001  the wait re-raises it, the callback must not die
+                future.set_exception(exc)
+                continue
+            if matched:
+                future.set_result(state)
+
+    async def expect(
+        self, predicate: Callable[[EntityState], bool], timeout: float = 5.0
+    ) -> EntityState:
+        """Wait for the next state matching ``predicate``; states seen before this call do not count."""
+        entry = (predicate, asyncio.get_running_loop().create_future())
+        self._waiters.append(entry)
+        try:
+            async with asyncio.timeout(timeout):
+                return await entry[1]
+        except TimeoutError:
+            raise TimeoutError(
+                f"no state matched {predicate} within {timeout}s"
+            ) from None
+        finally:
+            self._waiters.remove(entry)
+
+
 def find_entity[T: EntityInfo](
     entities: list[EntityInfo],
     object_id_substring: str,

tests/integration/test_api_empty_message_roundtrip.py

@@ -0,0 +1,37 @@
+"""Messages without fields go through the shared ProtoMessage entry points on both directions."""
+
+from __future__ import annotations
+
+from aioesphomeapi import api_pb2
+import pytest
+
+from .raw_api_client import MESSAGE_TYPE_OF, RawApiClient
+from .types import RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_api_empty_message_roundtrip(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    unused_tcp_port: int,
+) -> None:
+    async with run_compiled(yaml_config), RawApiClient(unused_tcp_port) as client:
+        await client.connect()
+
+        # Field free request and reply on the plain send path
+        await client.send_message(api_pb2.PingRequest())
+        await client.read_until_frame(MESSAGE_TYPE_OF[api_pb2.PingResponse])
+
+        # Field free request answered by a message with fields, and a list that ends with
+        # the field free ListEntitiesDoneResponse through the batching path
+        await client.send_message(api_pb2.DeviceInfoRequest())
+        await client.read_until_frame(MESSAGE_TYPE_OF[api_pb2.DeviceInfoResponse])
+        await client.send_message(api_pb2.ListEntitiesRequest())
+        await client.read_until_frame(MESSAGE_TYPE_OF[api_pb2.ListEntitiesDoneResponse])
+        assert (
+            client.frame_counts[MESSAGE_TYPE_OF[api_pb2.ListEntitiesSwitchResponse]]
+            == 1
+        )
+
+        await client.send_message(api_pb2.DisconnectRequest())
+        await client.read_until_frame(MESSAGE_TYPE_OF[api_pb2.DisconnectResponse])

tests/integration/test_api_encode_boundaries.py

@@ -0,0 +1,76 @@
+"""Encode paths at their branch boundaries: zero skipped float, fixed32 state, negative int32,
+length prefixes of two varint bytes and two byte field tags."""
+
+from __future__ import annotations
+
+import asyncio
+
+from aioesphomeapi import (
+    NumberState,
+    SelectInfo,
+    SensorInfo,
+    SensorState,
+    TextSensorState,
+)
+import pytest
+
+from .state_utils import InitialStateHelper, StateWaiter, require_entity
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+LONG_OPTION = (
+    "option-with-a-name-long-enough-that-its-length-prefix-needs-two-varint-bytes-"
+    "when-the-list-entities-response-is-encoded-xxxxxxxxxx"
+)
+
+
+@pytest.mark.asyncio
+async def test_api_encode_boundaries(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    async with run_compiled(yaml_config), api_client_connected() as client:
+        device_info, (entities, _) = await asyncio.gather(
+            client.device_info(), client.list_entities_services()
+        )
+        assert device_info.suggested_area == "Kitchen"
+
+        sensor = require_entity(entities, "zero_then_value", SensorInfo)
+        assert sensor.accuracy_decimals == -2
+        select = require_entity(entities, "long_option_select", SelectInfo)
+        assert len(LONG_OPTION) >= 128
+        assert select.options == ["short", LONG_OPTION]
+        text = require_entity(entities, "long_text")
+        number = require_entity(entities, "negative_number")
+        button = require_entity(entities, "publish_values")
+
+        initial = InitialStateHelper(entities)
+        waiter = StateWaiter()
+        client.subscribe_states(initial.on_state_wrapper(waiter.on_state))
+        await initial.wait_for_initial_states()
+
+        # A float of exactly zero is skipped on the wire and must still read as 0.0, not missing
+        first = initial.initial_states[sensor.key]
+        assert isinstance(first, SensorState)
+        assert first.state == 0.0 and not first.missing_state
+        first_number = initial.initial_states[number.key]
+        assert isinstance(first_number, NumberState)
+        assert first_number.state == -123.5
+
+        client.button_command(button.key)
+        await asyncio.gather(
+            waiter.expect(
+                lambda s: (
+                    isinstance(s, SensorState)
+                    and s.key == sensor.key
+                    and s.state == 12.5
+                )
+            ),
+            waiter.expect(
+                lambda s: (
+                    isinstance(s, TextSensorState)
+                    and s.key == text.key
+                    and s.state == "y" * 200
+                )
+            ),
+        )

tests/script/test_platformio_install_deps.py

@@ -35,8 +35,8 @@ def _load_script():
 def test_spec_key_collapses_destinations() -> None:
     """Two specs delivering one package share a directory and one key."""
     mod = _load_script()
-    assert mod.spec_key("esphome/noise-c @ 0.1.21") == "noise-c"
-    assert mod.spec_key("esphome/noise-c@0.1.21") == "noise-c"
+    assert mod.spec_key("esphome/noise-c @ 0.1.24") == "noise-c"
+    assert mod.spec_key("esphome/noise-c@0.1.24") == "noise-c"
     assert mod.spec_key("ESP32Async/AsyncTCP @ ^3.4.10") == mod.spec_key(
         "esp32async/asynctcp @ 3.5.0"
     )
@@ -54,23 +54,23 @@ def test_parse_specs_and_cli_args(tmp_path: Path) -> None:
         "[env:a]\n"
         "platform = fake/platform@1\n"
         "lib_deps =\n"
-        "    esphome/noise-c @ 0.1.21\n"
+        "    esphome/noise-c @ 0.1.24\n"
         "    ${common.lib_deps}\n"
         "    internal_lib\n"
         "[env:b]\n"
         "lib_deps =\n"
-        "    esphome/noise-c @ 0.1.21\n"
+        "    esphome/noise-c @ 0.1.24\n"
     )
     mod = _load_script()
     args = Namespace(libraries=True, platforms=True, tools=False)
     libs, platforms, tools = mod.parse_specs(str(ini), args)
     # exact-string duplicates collapse; distinct version pins survive
-    assert libs == ["esphome/noise-c @ 0.1.21"]
+    assert libs == ["esphome/noise-c @ 0.1.24"]
     assert platforms == ["fake/platform@1"]
     assert tools == []
     assert mod.build_cli_args(libs, platforms, tools) == [
         "-l",
-        "esphome/noise-c @ 0.1.21",
+        "esphome/noise-c @ 0.1.24",
         "-p",
         "fake/platform@1",
     ]
@@ -162,13 +162,13 @@ def test_parallel_install_behavior(tmp_path: Path) -> None:
     mod.parallel_install(
         cls,
         [
-            "esphome/noise-c @ 0.1.21",
-            "esphome/noise-c @ 0.1.21",
+            "esphome/noise-c @ 0.1.24",
+            "esphome/noise-c @ 0.1.24",
             "esphome/already @ 1.0",
             "https://x/framework.tar.xz",
         ],
     )
-    assert cls.calls == ["esphome/noise-c @ 0.1.21"]
+    assert cls.calls == ["esphome/noise-c @ 0.1.24"]
     assert cls.lock_events == ["lock", "unlock"]
 
 
@@ -205,7 +205,7 @@ def test_parallel_install_runs_dependency_waves(tmp_path: Path) -> None:
     mod = _load_script()
     cls = _reset_fake(str(tmp_path))
     cls.deps = {
-        "esphome/noise-c @ 0.1.21": [
+        "esphome/noise-c @ 0.1.24": [
             {"owner": "esphome", "name": "libsodium", "version": "^1.0"},
             {"name": "SPI"},
         ],
@@ -213,12 +213,12 @@ def test_parallel_install_runs_dependency_waves(tmp_path: Path) -> None:
             {"owner": "esphome", "name": "libsodium", "version": "^1.0"},
         ],
     }
-    mod.parallel_install(cls, ["esphome/noise-c @ 0.1.21", "esphome/wg @ 1.0"])
+    mod.parallel_install(cls, ["esphome/noise-c @ 0.1.24", "esphome/wg @ 1.0"])
     assert len(cls.calls) == 3  # the shared dep installs exactly once
     assert {mod.spec_key(c) for c in cls.calls} == {"noise-c", "wg", "libsodium"}
     # Wave-1 strings carry no compatibility; the dependency wave does
     compats = dict(cls.compat_calls)
-    assert compats["esphome/noise-c @ 0.1.21"] is None
+    assert compats["esphome/noise-c @ 0.1.24"] is None
     dep_compat = next(v for k, v in cls.compat_calls if "libsodium" in k)
     assert dep_compat is not None  # mirrors pio's install_dependency
 
@@ -229,11 +229,11 @@ def test_dependency_wave_excludes_url_specs(tmp_path: Path) -> None:
     mod = _load_script()
     cls = _reset_fake(str(tmp_path))
     cls.deps = {
-        "esphome/noise-c @ 0.1.21": [
+        "esphome/noise-c @ 0.1.24": [
             {"name": "vendored", "version": "https://github.com/x/y.git"},
         ],
     }
-    mod.parallel_install(cls, ["esphome/noise-c @ 0.1.21"])
+    mod.parallel_install(cls, ["esphome/noise-c @ 0.1.24"])
     assert {mod.spec_key(c) for c in cls.calls} == {"noise-c"}
 
 
@@ -348,13 +348,13 @@ def test_warm_store_still_walks_dependencies(tmp_path: Path) -> None:
     """Already-installed top-level packages still feed the dependency
     wave; a warm store can be missing a transitive dep."""
     mod = _load_script()
-    cls = _reset_fake(str(tmp_path), installed={"esphome/noise-c @ 0.1.21"})
+    cls = _reset_fake(str(tmp_path), installed={"esphome/noise-c @ 0.1.24"})
     cls.deps = {
-        "esphome/noise-c @ 0.1.21": [
+        "esphome/noise-c @ 0.1.24": [
             {"owner": "esphome", "name": "libsodium", "version": "^1.0"},
         ],
     }
-    mod.parallel_install(cls, ["esphome/noise-c @ 0.1.21"])
+    mod.parallel_install(cls, ["esphome/noise-c @ 0.1.24"])
     assert [mod.spec_key(c) for c in cls.calls] == ["libsodium"]
 
 

tests/unit_tests/components/api/test_api_proto.py

@@ -194,17 +194,17 @@ def test_superseded_device_info_fields_still_declared_in_header() -> None:
 
 def test_superseded_device_info_fields_still_encoded_and_sized() -> None:
     """Each superseded field must still be touched by DeviceInfoResponse's
-    generated encode() and calculate_size(), i.e. it is still put on the wire.
+    generated encode_msg() and calc_size_msg(), i.e. it is still put on the wire.
     """
-    encode_body = _extract_function_body(CPP_TEXT, "DeviceInfoResponse::encode")
-    size_body = _extract_function_body(CPP_TEXT, "DeviceInfoResponse::calculate_size")
+    encode_body = _extract_function_body(CPP_TEXT, "DeviceInfoResponse::encode_msg")
+    size_body = _extract_function_body(CPP_TEXT, "DeviceInfoResponse::calc_size_msg")
     for field_name in SUPERSEDED_FIELDS:
-        assert f"this->{field_name}" in encode_body, (
-            f"DeviceInfoResponse::encode() no longer references {field_name}. "
+        assert f"msg.{field_name}" in encode_body, (
+            f"DeviceInfoResponse::encode_msg() no longer references {field_name}. "
             f"{DEPRECATED_FIELD_TRAP}"
         )
-        assert f"this->{field_name}" in size_body, (
-            f"DeviceInfoResponse::calculate_size() no longer references "
+        assert f"msg.{field_name}" in size_body, (
+            f"DeviceInfoResponse::calc_size_msg() no longer references "
             f"{field_name}. {DEPRECATED_FIELD_TRAP}"
         )
 
@@ -380,3 +380,13 @@ def test_api_version_minor_is_at_least_15() -> None:
         "clients to see api_version >= 1.15 in HelloResponse before they will "
         "ever request it."
     )
+
+
+def test_generated_encode_calls_keep_the_cursor() -> None:
+    """No generated ProtoEncode call may drop the returned cursor."""
+    dropped = [
+        line
+        for line in CPP_TEXT.splitlines()
+        if line.lstrip().startswith("ProtoEncode::")
+    ]
+    assert not dropped, dropped[:5]

tests/unit_tests/components/api/test_api_protobuf_generator.py

@@ -15,9 +15,11 @@ import pytest
 
 sys.path.insert(0, str(Path(__file__).parents[4] / "script" / "api_protobuf"))
 
+import aioesphomeapi.api_options_pb2 as pb  # noqa: E402
 from api_protobuf import (  # noqa: E402
     MAX_MESSAGE_ID,
     _make_ifdef_line,
+    create_field_type_info,
     get_varint64_ifdef,
     validate_message_id,
 )
@@ -43,7 +45,14 @@ UINT64 = descriptor_pb2.FieldDescriptorProto.TYPE_UINT64
 INT64 = descriptor_pb2.FieldDescriptorProto.TYPE_INT64
 SINT64 = descriptor_pb2.FieldDescriptorProto.TYPE_SINT64
 UINT32 = descriptor_pb2.FieldDescriptorProto.TYPE_UINT32
+INT32 = descriptor_pb2.FieldDescriptorProto.TYPE_INT32
+SINT32 = descriptor_pb2.FieldDescriptorProto.TYPE_SINT32
 FIXED64 = descriptor_pb2.FieldDescriptorProto.TYPE_FIXED64
+FIXED32 = descriptor_pb2.FieldDescriptorProto.TYPE_FIXED32
+FLOAT = descriptor_pb2.FieldDescriptorProto.TYPE_FLOAT
+BOOL = descriptor_pb2.FieldDescriptorProto.TYPE_BOOL
+STRING = descriptor_pb2.FieldDescriptorProto.TYPE_STRING
+BYTES = descriptor_pb2.FieldDescriptorProto.TYPE_BYTES
 
 
 def test_no_varint64_fields() -> None:
@@ -107,3 +116,69 @@ def test_message_id_at_maximum_is_accepted() -> None:
 def test_message_id_above_maximum_is_rejected() -> None:
     with pytest.raises(ValueError, match="exceeds the plaintext"):
         validate_message_id(MAX_MESSAGE_ID + 1, "TooBigMessage")
+
+
+def _field(
+    field_type: int, number: int = 1, *, force: bool = False, repeated: bool = False
+) -> descriptor_pb2.FieldDescriptorProto:
+    field = descriptor_pb2.FieldDescriptorProto(
+        name="value", number=number, type=field_type
+    )
+    if repeated:
+        field.label = descriptor_pb2.FieldDescriptorProto.LABEL_REPEATED
+    if force:
+        field.options.Extensions[pb.force] = True
+    return field
+
+
+def _encode_field(
+    field_type: int, number: int = 1, force: bool = False, repeated: bool = False
+) -> str:
+    """Return the encode statement the generator emits for one encode-only field."""
+    field = _field(field_type, number, force=force, repeated=repeated)
+    return create_field_type_info(
+        field, needs_decode=False, needs_encode=True
+    ).encode_content
+
+
+SCALAR_TYPES = [
+    BOOL,
+    UINT32,
+    INT32,
+    UINT64,
+    INT64,
+    SINT32,
+    FLOAT,
+    FIXED32,
+    STRING,
+    BYTES,
+]
+
+
+@pytest.mark.parametrize("field_type", SCALAR_TYPES)
+def test_forced_fields_use_the_force_overload_or_raw_writes(field_type: int) -> None:
+    content = _encode_field(field_type, force=True)
+    assert (
+        "_force(" in content
+        or "write_raw_byte(" in content
+        or "write_tag_and_fixed32(" in content
+    ), content
+
+
+@pytest.mark.parametrize("field_type", [FLOAT, FIXED32])
+def test_single_byte_tag_fixed32_shares_the_outlined_writer(field_type: int) -> None:
+    unconditional = _encode_field(field_type, force=True)
+    assert unconditional.count("write_tag_and_fixed32(pos, 13,") == 1, unconditional
+    guarded = _encode_field(field_type, force=False)
+    assert guarded.startswith("if ("), guarded
+    assert "[[likely]]" in guarded
+    assert "write_tag_and_fixed32(pos, 13," in guarded
+
+
+@pytest.mark.parametrize("field_type", [FLOAT, FIXED32])
+def test_multi_byte_tag_fixed32_falls_back_to_the_generic_helper(
+    field_type: int,
+) -> None:
+    content = _encode_field(field_type, number=16)
+    assert "write_tag_and_fixed32" not in content, content
+    assert content.startswith("pos = ProtoEncode::encode_"), content

tests/unit_tests/components/test_espnow.py

@@ -0,0 +1,48 @@
+"""Tests for the espnow component's final validation."""
+
+import pytest
+
+from esphome.components.esp32.const import (
+    VARIANT_ESP32C3,
+    VARIANT_ESP32H2,
+    VARIANT_ESP32P4,
+)
+from esphome.components.espnow import _validate_variant
+import esphome.config_validation as cv
+import esphome.final_validate as fv
+from esphome.types import ConfigType
+
+
+def _run(
+    monkeypatch, variant: str, full_config: dict, config: ConfigType
+) -> ConfigType:
+    monkeypatch.setattr("esphome.components.espnow.get_esp32_variant", lambda: variant)
+    token = fv.full_config.set(full_config)
+    try:
+        return _validate_variant(config)
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_variant_with_native_wifi_passes(monkeypatch) -> None:
+    """A variant with a native Wi-Fi PHY needs no shim; config passes through."""
+    config = {"id": "espnow"}
+    assert _run(monkeypatch, VARIANT_ESP32C3, {}, config) is config
+
+
+def test_radioless_non_p4_variant_rejected(monkeypatch) -> None:
+    """Radio-less variants without any ESP-NOW path are rejected outright."""
+    with pytest.raises(cv.Invalid, match="not supported"):
+        _run(monkeypatch, VARIANT_ESP32H2, {}, {})
+
+
+def test_p4_without_esp32_hosted_rejected(monkeypatch) -> None:
+    """The P4 needs the esp32_hosted shim to supply the esp_now_* symbols."""
+    with pytest.raises(cv.Invalid, match="esp32_hosted"):
+        _run(monkeypatch, VARIANT_ESP32P4, {}, {})
+
+
+def test_p4_with_esp32_hosted_passes(monkeypatch) -> None:
+    """The P4 with esp32_hosted present validates; config passes through."""
+    config = {"id": "espnow"}
+    assert _run(monkeypatch, VARIANT_ESP32P4, {"esp32_hosted": {}}, config) is config

tests/unit_tests/conftest.py

@@ -9,7 +9,7 @@ not be part of a unit test suite.
 
 """
 
-from collections.abc import Generator
+from collections.abc import Callable, Generator
 import os
 from pathlib import Path
 import sys
@@ -137,3 +137,40 @@ def mock_get_component() -> Generator[Mock, None, None]:
     """Mock get_component for config module."""
     with patch("esphome.config.get_component") as mock:
         yield mock
+
+
+@pytest.fixture
+def held_lock() -> Callable[..., Callable[..., None]]:
+    """Factory for a ``FileLock.acquire`` fake held by another downloader.
+
+    Each poll writes the next chunk to ``part`` (or runs it, for a callable)
+    and raises ``Timeout``; when the chunks run out the part is removed,
+    ``land()`` runs, and the acquire succeeds (also for any later job, so
+    ``land`` must be idempotent).
+    """
+    from filelock import Timeout
+
+    def make(
+        part: Path,
+        chunks: list[bytes | Callable[[], None]],
+        land: Callable[[], None],
+    ) -> Callable[..., None]:
+        polls = iter(chunks)
+
+        def acquire(*args, **kwargs) -> None:
+            try:
+                chunk = next(polls)
+            except StopIteration:
+                part.unlink(missing_ok=True)
+                land()
+                return
+            if callable(chunk):
+                chunk()
+            else:
+                part.parent.mkdir(parents=True, exist_ok=True)
+                part.write_bytes(chunk)
+            raise Timeout("held")
+
+        return acquire
+
+    return make

tests/unit_tests/test_framework_helpers.py

@@ -2353,3 +2353,20 @@ def test_discard_partial_download_logs_undeletable(
     ):
         framework_helpers.discard_partial_download(dest)
     assert "Could not remove" in caplog.text
+
+
+def test_downloaded_bytes_reports_what_is_on_disk(tmp_path: Path) -> None:
+    """Part file first, then the landed file, both capped at size; else 0."""
+    dest = tmp_path / "archive"
+    assert framework_helpers.downloaded_bytes(dest, 4) == 0
+    part = tmp_path / "archive.part"
+    part.write_bytes(b"ab")
+    assert framework_helpers.downloaded_bytes(dest, 4) == 2
+    part.write_bytes(b"abcdef")
+    assert framework_helpers.downloaded_bytes(dest, 4) == 4
+    part.unlink()
+    dest.write_bytes(b"abc")
+    assert framework_helpers.downloaded_bytes(dest, 4) == 3
+    assert framework_helpers.downloaded_bytes(dest) == 3
+    dest.write_bytes(b"abcdef")
+    assert framework_helpers.downloaded_bytes(dest, 4) == 4

tests/unit_tests/test_platformio_prefetch.py

@@ -454,23 +454,96 @@ def test_uri_fetch_job_waits_out_a_briefly_held_lock(tmp_path: Path) -> None:
     assert dl_path.read_bytes() == b"data"
 
 
-def test_lock_deadline_leaves_download_to_the_holder(tmp_path: Path) -> None:
-    """A lock held past the deadline means another process is fetching the
-    same file; skipping cleanly beats a misleading failure warning. The
-    tracker is still polled so a parked worker observes cancellation."""
+@pytest.mark.parametrize("staged", [b"", b"ab"])
+def test_lock_deadline_leaves_download_to_the_holder(
+    tmp_path: Path, staged: bytes
+) -> None:
+    """A lock held past the deadline is another process's download; skip
+    cleanly, polling the tracker with what the holder has staged so far."""
     dl_path = tmp_path / "archive"
+    (tmp_path / "archive.prefetch.part").write_bytes(staged)
     ticks: list[int] = []
     with (
         patch("esphome.framework_helpers.download_with_resume") as mock_download,
         patch("filelock.FileLock.acquire", side_effect=Timeout("held")),
-        patch.object(pf, "_DOWNLOAD_LOCK_TIMEOUT", 0),
+        patch("esphome.framework_helpers.DOWNLOAD_LOCK_TIMEOUT", 0),
     ):
         pf._uri_fetch_job(MagicMock(), "https://x/a.zip", dl_path, 4)(ticks.append)
     mock_download.assert_not_called()
-    assert ticks == [0]
+    assert ticks == [len(staged)]
     assert not dl_path.exists()
 
 
+@pytest.mark.parametrize(
+    ("job", "part_name", "chunks", "expected"),
+    [
+        (
+            lambda dl_path: pf._registry_fetch_job(
+                MagicMock(), "https://x/a.tar.gz", dl_path, "ab" * 32, 4
+            ),
+            "archive.part",
+            [b"a", b"abc"],
+            [1, 3, 4],
+        ),
+        (
+            lambda dl_path: pf._uri_fetch_job(
+                MagicMock(), "https://x/a.zip", dl_path, 4
+            ),
+            "archive.prefetch.part",
+            [b"ab"],
+            [2, 4],
+        ),
+    ],
+    ids=["registry", "uri"],
+)
+def test_lock_wait_reports_the_holders_progress(
+    tmp_path: Path,
+    caplog: pytest.LogCaptureFixture,
+    held_lock,
+    job,
+    part_name: str,
+    chunks: list[bytes],
+    expected: list[int],
+) -> None:
+    """A waiting job reports the holder's part file (the staging one for a
+    URL job), then the full size once the holder lands the archive."""
+    dl_path = tmp_path / "archive"
+    ticks: list[int] = []
+    acquire = held_lock(
+        tmp_path / part_name, chunks, lambda: dl_path.write_bytes(b"abcd")
+    )
+    with (
+        patch("esphome.framework_helpers.download_with_resume") as mock_download,
+        patch("filelock.FileLock.acquire", side_effect=acquire),
+        patch("filelock.FileLock.release"),
+        caplog.at_level(logging.INFO),
+    ):
+        job(dl_path)(ticks.append)
+    mock_download.assert_not_called()
+    assert ticks == expected
+    assert caplog.text.count("Waiting for another process downloading archive") == 1
+
+
+def test_uri_lock_wait_prefers_the_landed_archive(tmp_path: Path, held_lock) -> None:
+    """Between the holder's promotion rename and its release the staging
+    part is gone; the landed cache file is credited instead of 0."""
+    dl_path = tmp_path / "archive"
+    ticks: list[int] = []
+    acquire = held_lock(
+        tmp_path / "archive.prefetch.part",
+        [b"ab", lambda: dl_path.write_bytes(b"abcd")],
+        lambda: None,
+    )
+    with (
+        patch("esphome.framework_helpers.download_with_resume") as mock_download,
+        patch("filelock.FileLock.acquire", side_effect=acquire),
+        patch("filelock.FileLock.release"),
+    ):
+        pf._uri_fetch_job(MagicMock(), "https://x/a.zip", dl_path, 4)(ticks.append)
+    mock_download.assert_not_called()
+    assert ticks == [2, 4, 4]
+
+
 def test_registry_lock_deadline_skips_registration(tmp_path: Path) -> None:
     """A registry job that lost the download race to another process
     must not stamp a nonexistent archive into pio's usage.db."""
@@ -479,7 +552,7 @@ def test_registry_lock_deadline_skips_registration(tmp_path: Path) -> None:
     with (
         patch("esphome.framework_helpers.download_with_resume") as mock_download,
         patch("filelock.FileLock.acquire", side_effect=Timeout("held")),
-        patch.object(pf, "_DOWNLOAD_LOCK_TIMEOUT", 0),
+        patch("esphome.framework_helpers.DOWNLOAD_LOCK_TIMEOUT", 0),
     ):
         pf._registry_fetch_job(manager, "https://x/a.tar.gz", dl_path, "ab" * 32, 4)(
             lambda done: None
@@ -1576,7 +1649,7 @@ def test_preinstall_runs_dependency_waves(tmp_path: Path) -> None:
         {"name": "SPI"},
     ]
     m.dependency_to_spec.side_effect = lambda dep: _FakeSpec(name=dep["name"])
-    pf._preinstall(m, [("noise-c@0.1.21", _FakeSpec(name="noise-c"))])
+    pf._preinstall(m, [("noise-c@0.1.24", _FakeSpec(name="noise-c"))])
     assert installed == ["noise-c", "libsodium"]  # dep deduped, SPI left out
     # The dep wave carries its compatibility so _install searches qualified
     dep_call = m._install.call_args_list[-1]
@@ -1596,7 +1669,7 @@ def test_preinstall_dependency_wave_skips_seen_names(tmp_path: Path) -> None:
     m._install.side_effect = lambda spec, skip_dependencies, compatibility=None: (
         installed.append(getattr(spec, "name", str(spec)))
     )
-    pf._preinstall(m, [("noise-c@0.1.21", _FakeSpec(name="noise-c"))])
+    pf._preinstall(m, [("noise-c@0.1.24", _FakeSpec(name="noise-c"))])
     assert installed == ["noise-c"]
 
 

tests/unit_tests/test_platformio_registry.py

@@ -8,6 +8,7 @@ import os
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
+from filelock import Timeout
 import pytest
 
 from esphome.core import EsphomeError
@@ -540,16 +541,13 @@ def test_prefetch_packages_skips_freshly_installed_dest(tmp_path: Path) -> None:
     dest = tmp_path / "a"
     dest.mkdir()
 
-    from contextlib import contextmanager
-
-    @contextmanager
-    def marker_appears_under_lock(path, **kwargs):
+    def marker_appears_under_lock(*args, **kwargs):
         # Simulates the concurrent build finishing while we waited
         (dest / ".esphome_extracted").touch()
-        yield
 
     with (
-        patch("filelock.FileLock", side_effect=marker_appears_under_lock),
+        patch("filelock.FileLock.acquire", side_effect=marker_appears_under_lock),
+        patch("filelock.FileLock.release"),
         patch.object(registry, "download_with_resume") as mock_download,
         patch.object(
             registry, "registry_download", side_effect=_resolve_for({"a": 10})
@@ -559,6 +557,69 @@ def test_prefetch_packages_skips_freshly_installed_dest(tmp_path: Path) -> None:
     mock_download.assert_not_called()
 
 
+def test_prefetch_packages_waits_with_the_holders_progress(
+    tmp_path: Path, held_lock
+) -> None:
+    """A worker parked on another build's lock reports that build's part
+    file, then the full size once the marker appears."""
+    dest = tmp_path / "a"
+    dest.mkdir()
+    ticks: list[int] = []
+    part = tmp_path / "dl" / "a-1.0.part"
+
+    def installed_and_pruned() -> None:
+        # install_package touches the marker, then unlinks the archive
+        (dest / ".esphome_extracted").touch()
+        part.unlink()
+
+    acquire = held_lock(
+        part,
+        [lambda: None, b"abc", installed_and_pruned],
+        (dest / ".esphome_extracted").touch,
+    )
+
+    def fake_batch(header, jobs):
+        for _name, _size, fetch in jobs:
+            fetch(ticks.append)
+        return []
+
+    with (
+        patch("filelock.FileLock.acquire", side_effect=acquire),
+        patch("filelock.FileLock.release"),
+        patch.object(registry, "run_batch_downloads", side_effect=fake_batch),
+        patch.object(registry, "download_with_resume") as mock_download,
+        patch.object(
+            registry, "registry_download", side_effect=_resolve_for({"a": 10, "b": 5})
+        ),
+    ):
+        registry.prefetch_packages(
+            [("a", "1.0", dest, []), ("b", "2.0", tmp_path / "b", [])],
+            tmp_path / "dl",
+        )
+    assert ticks == [0, 3, 10, 10]
+    mock_download.assert_called_once()
+
+
+def test_prefetch_packages_leaves_a_long_held_lock_to_its_holder(
+    tmp_path: Path,
+) -> None:
+    """Past the deadline the worker skips; install_package waits on the same
+    lock later and verifies whatever the holder produced."""
+    with (
+        patch("filelock.FileLock.acquire", side_effect=Timeout("held")),
+        patch("esphome.framework_helpers.DOWNLOAD_LOCK_TIMEOUT", 0),
+        patch.object(registry, "download_with_resume") as mock_download,
+        patch.object(
+            registry, "registry_download", side_effect=_resolve_for({"a": 10, "b": 5})
+        ),
+    ):
+        registry.prefetch_packages(
+            [("a", "1.0", tmp_path / "a", []), ("b", "2.0", tmp_path / "b", [])],
+            tmp_path / "dl",
+        )
+    mock_download.assert_not_called()
+
+
 def test_already_installed_probe(tmp_path: Path) -> None:
     """Both arms of the marker probe the prefetch worker keys on."""
     dest = tmp_path / "pkg"