Revert "[api] Match log format types for the clamped version fields"

This reverts commit c212a46757.

esphome/components/api/api_connection.cpp

@@ -1752,10 +1752,12 @@ void APIConnection::complete_authentication_() {
 bool APIConnection::send_hello_response_(const HelloRequest &msg) {
   // Copy client name with truncation if needed (set_client_name handles truncation)
   this->helper_->set_client_name(msg.client_info.c_str(), msg.client_info.size());
-  this->client_api_version_major_ = msg.api_version_major;
-  this->client_api_version_minor_ = msg.api_version_minor;
+  this->client_api_version_major_ =
+      static_cast<uint8_t>(std::min<uint32_t>(msg.api_version_major, std::numeric_limits<uint8_t>::max()));
+  this->client_api_version_minor_ =
+      static_cast<uint8_t>(std::min<uint32_t>(msg.api_version_minor, std::numeric_limits<uint8_t>::max()));
   char peername[socket::SOCKADDR_STR_LEN];
-  ESP_LOGV(TAG, "Hello from client: '%s' | %s | API Version %" PRIu16 ".%" PRIu16, this->helper_->get_client_name(),
+  ESP_LOGV(TAG, "Hello from client: '%s' | %s | API Version %u.%u", this->helper_->get_client_name(),
            this->helper_->get_peername_to(peername), this->client_api_version_major_, this->client_api_version_minor_);
 
   HelloResponse resp;
@@ -2184,7 +2186,7 @@ bool APIConnection::try_to_clear_buffer_slow_(bool log_out_of_space) {
   }
   return false;
 }
-bool APIConnection::send_message_(uint32_t payload_size, uint8_t message_type, MessageEncodeFn encode_fn,
+bool APIConnection::send_message_(uint32_t payload_size, uint16_t message_type, MessageEncodeFn encode_fn,
                                   const void *msg) {
 #ifdef HAS_PROTO_MESSAGE_DUMP
   // Skip dump for log messages (recursive logging risk) and camera frames (high-frequency noise)
@@ -2213,7 +2215,7 @@ uint16_t APIConnection::encode_to_buffer_slow(uint32_t calculated_size, MessageE
                                               APIConnection *conn, uint32_t remaining_size) {
   return encode_to_buffer(calculated_size, encode_fn, msg, conn, remaining_size);
 }
-bool APIConnection::send_buffer(ProtoWriteBuffer buffer, uint8_t message_type) {
+bool APIConnection::send_buffer(ProtoWriteBuffer buffer, uint16_t message_type) {
   const bool is_log_message = (message_type == SubscribeLogsResponse::MESSAGE_TYPE);
 
   if (!this->try_to_clear_buffer(!is_log_message)) {
@@ -2243,12 +2245,12 @@ void APIConnection::on_fatal_error() {
   this->flags_.remove = true;
 }
 
-bool APIConnection::schedule_message_front_(EntityBase *entity, uint8_t message_type, uint8_t estimated_size) {
+bool APIConnection::schedule_message_front_(EntityBase *entity, uint16_t message_type, uint8_t estimated_size) {
   this->deferred_batch_.add_item_front(entity, message_type, estimated_size);
   return this->schedule_batch_();
 }
 
-bool APIConnection::send_message_smart_(EntityBase *entity, uint8_t message_type, uint8_t estimated_size,
+bool APIConnection::send_message_smart_(EntityBase *entity, uint16_t message_type, uint8_t estimated_size,
                                         uint8_t aux_data_index) {
   if (this->should_send_immediately_(message_type) && this->helper_->can_write_without_blocking()) {
     auto &shared_buf = this->parent_->get_shared_buffer_ref();

esphome/components/api/api_connection.h

@@ -326,8 +326,10 @@ class APIConnection final : public APIServerConnectionBase {
   bool is_marked_for_removal() const { return this->flags_.remove; }
   uint8_t get_log_subscription_level() const { return this->flags_.log_subscription; }
 
-  // Get client API version for feature detection
-  bool client_supports_api_version(uint16_t major, uint16_t minor) const {
+  // Get client API version for feature detection.
+  // Stored versions saturate at 255 (see send_hello_response_), so requesting
+  // a minimum above that can never match.
+  bool client_supports_api_version(uint8_t major, uint8_t minor) const {
     return this->client_api_version_major_ > major ||
            (this->client_api_version_major_ == major && this->client_api_version_minor_ >= minor);
   }
@@ -374,7 +376,7 @@ class APIConnection final : public APIServerConnectionBase {
       return true;
     return this->try_to_clear_buffer_slow_(log_out_of_space);
   }
-  bool send_buffer(ProtoWriteBuffer buffer, uint8_t message_type);
+  bool send_buffer(ProtoWriteBuffer buffer, uint16_t message_type);
 
   const char *get_name() const { return this->helper_->get_client_name(); }
   /// Get peer name (IP address) into caller-provided buffer, returns buf for convenience
@@ -423,7 +425,7 @@ class APIConnection final : public APIServerConnectionBase {
   }
 
   // Non-template buffer management for send_message
-  bool send_message_(uint32_t payload_size, uint8_t message_type, MessageEncodeFn encode_fn, const void *msg);
+  bool send_message_(uint32_t payload_size, uint16_t message_type, MessageEncodeFn encode_fn, const void *msg);
 
   // Core batch encoding logic. ALWAYS_INLINE so encode_fn devirtualizes at hot call sites.
   // Defined in api_connection_buffer.h (needs APIServer complete).
@@ -664,10 +666,9 @@ class APIConnection final : public APIServerConnectionBase {
 
     struct BatchItem {
       EntityBase *entity;                       // 4 bytes - Entity pointer
-      uint8_t message_type;                     // 1 byte - Message type for protocol and dispatch
+      uint16_t message_type;                    // 2 bytes - Message type for protocol and dispatch
       uint8_t estimated_size;                   // 1 byte - Estimated message size (max 255 bytes)
       uint8_t aux_data_index{AUX_DATA_UNUSED};  // 1 byte - For events: index into entity's event_types
-      // 1 byte padding
     };
 
     std::vector<BatchItem> items;
@@ -677,7 +678,7 @@ class APIConnection final : public APIServerConnectionBase {
     // connections that do, buffers are released after initial sync anyway
 
     // Add item to the batch (with deduplication)
-    void add_item(EntityBase *entity, uint8_t message_type, uint8_t estimated_size,
+    void add_item(EntityBase *entity, uint16_t message_type, uint8_t estimated_size,
                   uint8_t aux_data_index = AUX_DATA_UNUSED) {
       // Dedup: O(n) scan but optimized for RAM over performance
       // Skip deduplication for events - they are edge-triggered, every occurrence matters
@@ -693,7 +694,7 @@ class APIConnection final : public APIServerConnectionBase {
       this->items.push_back({entity, message_type, estimated_size, aux_data_index});
     }
     // Add item to the front of the batch (for high priority messages like ping)
-    void add_item_front(EntityBase *entity, uint8_t message_type, uint8_t estimated_size) {
+    void add_item_front(EntityBase *entity, uint16_t message_type, uint8_t estimated_size) {
       // Swap to front avoids expensive vector::insert which shifts all elements
       this->items.push_back({entity, message_type, estimated_size, AUX_DATA_UNUSED});
       if (this->items.size() > 1) {
@@ -758,13 +759,15 @@ class APIConnection final : public APIServerConnectionBase {
 #endif
   } flags_{};  // 2 bytes total
 
-  // 2-byte types immediately after flags_ (no padding between them)
-  uint16_t client_api_version_major_{0};
-  uint16_t client_api_version_minor_{0};
+  // 2-byte type immediately after flags_ (no padding between them)
+  uint16_t batch_message_type_{0};  // Current message type during batch encoding
   // 1-byte types to fill remaining space before next 4-byte boundary
+  // Client API versions are clamped to 255 on receive (see send_hello_response_)
+  uint8_t client_api_version_major_{0};
+  uint8_t client_api_version_minor_{0};
   ActiveIterator active_iterator_{ActiveIterator::NONE};
-  uint8_t batch_message_type_{0};  // Current message type during batch encoding
-  // Total: 2 (flags) + 2 + 2 + 1 + 1 = 8 bytes, aligned to 4-byte boundary
+  // Total: 2 (flags) + 2 + 1 + 1 + 1 + 1 (batch_header_size_ below) = 8 bytes,
+  // aligned to 4-byte boundary
 
   // Actual header size used by encode_to_buffer for the current message.
   // Read by process_batch_multi_ to pass into MessageInfo.
@@ -813,7 +816,7 @@ class APIConnection final : public APIServerConnectionBase {
   // 2. It's an EventResponse (events are edge-triggered - every occurrence matters)
   // 3. OR: User has opted into immediate sending (should_try_send_immediately = true
   //    AND batch_delay = 0)
-  inline bool should_send_immediately_(uint8_t message_type) const {
+  inline bool should_send_immediately_(uint16_t message_type) const {
     return (
 #ifdef USE_UPDATE
         message_type == UpdateStateResponse::MESSAGE_TYPE ||
@@ -827,11 +830,11 @@ class APIConnection final : public APIServerConnectionBase {
   // Helper method to send a message either immediately or via batching
   // Tries immediate send if should_send_immediately_() returns true and buffer has space
   // Falls back to batching if immediate send fails or isn't applicable
-  bool send_message_smart_(EntityBase *entity, uint8_t message_type, uint8_t estimated_size,
+  bool send_message_smart_(EntityBase *entity, uint16_t message_type, uint8_t estimated_size,
                            uint8_t aux_data_index = DeferredBatch::AUX_DATA_UNUSED);
 
   // Helper function to schedule a deferred message with known message type
-  bool schedule_message_(EntityBase *entity, uint8_t message_type, uint8_t estimated_size,
+  bool schedule_message_(EntityBase *entity, uint16_t message_type, uint8_t estimated_size,
                          uint8_t aux_data_index = DeferredBatch::AUX_DATA_UNUSED) {
     this->deferred_batch_.add_item(entity, message_type, estimated_size, aux_data_index);
     return this->schedule_batch_();
@@ -839,7 +842,7 @@ class APIConnection final : public APIServerConnectionBase {
 
   // Helper function to schedule a high priority message at the front of the batch
   // Out-of-line: callers (on_shutdown, check_keepalive_) are cold paths
-  bool schedule_message_front_(EntityBase *entity, uint8_t message_type, uint8_t estimated_size);
+  bool schedule_message_front_(EntityBase *entity, uint16_t message_type, uint8_t estimated_size);
 
   // Helper function to log client messages with name and peername
   void log_client_(int level, const LogString *message);

esphome/components/api/api_frame_helper.h

@@ -49,16 +49,16 @@ struct ReadPacketBuffer {
 };
 
 // Packed message info structure to minimize memory usage
-// Note: message_type is uint8_t — all current protobuf message types fit in 8 bits.
-// The noise wire format encodes types as 16-bit, but the high byte is always 0.
-// If message types ever exceed 255, this and encrypt_noise_message_ must be updated.
+// message_type matches the wire formats: noise carries a fixed 16-bit type
+// field, plaintext a type varint. The proto codegen caps message IDs at 16383
+// so the plaintext type varint fits the 2 bytes budgeted in HEADER_PADDING.
 struct MessageInfo {
   uint16_t offset;        // Offset in buffer where message starts
   uint16_t payload_size;  // Size of the message payload
-  uint8_t message_type;   // Message type (0-255)
+  uint16_t message_type;  // Message type (0-16383)
   uint8_t header_size;    // Actual header size used (avoids recomputation in write path)
 
-  MessageInfo(uint8_t type, uint16_t off, uint16_t size, uint8_t hdr)
+  MessageInfo(uint16_t type, uint16_t off, uint16_t size, uint8_t hdr)
       : offset(off), payload_size(size), message_type(type), header_size(hdr) {}
 };
 
@@ -173,7 +173,7 @@ class APIFrameHelper {
   }
   // Write a single protobuf message - the hot path (87-100% of all writes).
   // Caller must ensure state is DATA before calling.
-  virtual APIError write_protobuf_packet(uint8_t type, ProtoWriteBuffer buffer) = 0;
+  virtual APIError write_protobuf_packet(uint16_t type, ProtoWriteBuffer buffer) = 0;
   // Write multiple protobuf messages in a single batched operation.
   // Caller must ensure state is DATA and messages is not empty.
   // messages contains (message_type, offset, length) for each message in the buffer.
@@ -187,15 +187,15 @@ class APIFrameHelper {
   // Distinguishes protocols via frame_footer_size_ (noise always has a non-zero MAC
   // footer, plaintext has footer=0). If a protocol with a plaintext footer is ever
   // added, this should become a virtual method.
-  uint8_t frame_header_size(uint16_t payload_size, uint8_t message_type) const {
+  uint8_t frame_header_size(uint16_t payload_size, uint16_t message_type) const {
 #if defined(USE_API_NOISE) && defined(USE_API_PLAINTEXT)
     return this->frame_footer_size_
                ? this->frame_header_padding_
-               : static_cast<uint8_t>(1 + ProtoSize::varint16(payload_size) + ProtoSize::varint8(message_type));
+               : static_cast<uint8_t>(1 + ProtoSize::varint16(payload_size) + ProtoSize::varint16(message_type));
 #elif defined(USE_API_NOISE)
     return this->frame_header_padding_;
 #else  // USE_API_PLAINTEXT only
-    return static_cast<uint8_t>(1 + ProtoSize::varint16(payload_size) + ProtoSize::varint8(message_type));
+    return static_cast<uint8_t>(1 + ProtoSize::varint16(payload_size) + ProtoSize::varint16(message_type));
 #endif
   }
   // Get the frame footer size required by this protocol

esphome/components/api/api_frame_helper_noise.cpp

@@ -490,7 +490,7 @@ APIError APINoiseFrameHelper::read_packet(ReadPacketBuffer *buffer) {
 }
 // Encrypt a single noise message in place and return the encrypted frame length.
 // Returns APIError::OK on success.
-APIError APINoiseFrameHelper::encrypt_noise_message_(uint8_t *buf_start, uint16_t payload_size, uint8_t message_type,
+APIError APINoiseFrameHelper::encrypt_noise_message_(uint8_t *buf_start, uint16_t payload_size, uint16_t message_type,
                                                      uint16_t &encrypted_len_out) {
   // Write noise header
   buf_start[0] = 0x01;  // indicator
@@ -523,7 +523,7 @@ APIError APINoiseFrameHelper::encrypt_noise_message_(uint8_t *buf_start, uint16_
   return APIError::OK;
 }
 
-APIError APINoiseFrameHelper::write_protobuf_packet(uint8_t type, ProtoWriteBuffer buffer) {
+APIError APINoiseFrameHelper::write_protobuf_packet(uint16_t type, ProtoWriteBuffer buffer) {
 #ifdef ESPHOME_DEBUG_API
   assert(this->state_ == State::DATA);
 #endif

esphome/components/api/api_frame_helper_noise.h

@@ -31,7 +31,7 @@ class APINoiseFrameHelper final : public APIFrameHelper {
 #endif
   APIError loop() override;
   APIError read_packet(ReadPacketBuffer *buffer) override;
-  APIError write_protobuf_packet(uint8_t type, ProtoWriteBuffer buffer) override;
+  APIError write_protobuf_packet(uint16_t type, ProtoWriteBuffer buffer) override;
   APIError write_protobuf_messages(ProtoWriteBuffer buffer, std::span<const MessageInfo> messages) override;
 
  protected:
@@ -44,7 +44,7 @@ class APINoiseFrameHelper final : public APIFrameHelper {
   APIError state_action_handshake_write_();
   APIError try_read_frame_();
   APIError write_frame_(const uint8_t *data, uint16_t len);
-  APIError encrypt_noise_message_(uint8_t *buf_start, uint16_t payload_size, uint8_t message_type,
+  APIError encrypt_noise_message_(uint8_t *buf_start, uint16_t payload_size, uint16_t message_type,
                                   uint16_t &encrypted_len_out);
   APIError init_handshake_();
   APIError check_handshake_finished_();

esphome/components/api/api_frame_helper_plaintext.cpp

@@ -5,6 +5,7 @@
 #include "esphome/core/hal.h"
 #include "esphome/core/helpers.h"
 #include "esphome/core/log.h"
+#include "api_pb2.h"
 #include "proto.h"
 #include <cstring>
 #include <cinttypes>
@@ -252,24 +253,21 @@ ESPHOME_ALWAYS_INLINE static inline void encode_varint_16(uint16_t value, uint8_
   *p = static_cast<uint8_t>(value);
 }
 
-// Encode an 8-bit varint (1-2 bytes) using pre-computed length.
-ESPHOME_ALWAYS_INLINE static inline void encode_varint_8(uint8_t value, uint8_t varint_len, uint8_t *p) {
-  if (varint_len == 2) {
-    *p++ = static_cast<uint8_t>(value | 0x80);
-    *p = static_cast<uint8_t>(value >> 7);
-  } else {
-    *p = value;
-  }
-}
+// The generator rejects message IDs above MAX_MESSAGE_TYPE, so the type varint
+// can never outgrow the 2 bytes HEADER_PADDING budgets for it. Without this
+// bound, write_plaintext_header's header_offset would underflow for the first
+// message in a batch and the header write would land outside the buffer.
+static_assert(1 + 3 + ProtoSize::varint16(MAX_MESSAGE_TYPE) <= APIPlaintextFrameHelper::HEADER_PADDING,
+              "HEADER_PADDING cannot fit the type varint of the largest message ID");
 
 // Write plaintext header into pre-allocated padding before payload.
 // padding_size: bytes reserved before payload (HEADER_PADDING for first/single msg,
 //               actual header size for contiguous batch messages).
 // Returns the total header length (indicator + varints).
 ESPHOME_ALWAYS_INLINE static inline uint8_t write_plaintext_header(uint8_t *buf_start, uint16_t payload_size,
-                                                                   uint8_t message_type, uint8_t padding_size) {
+                                                                   uint16_t message_type, uint8_t padding_size) {
   uint8_t size_varint_len = ProtoSize::varint16(payload_size);
-  uint8_t type_varint_len = ProtoSize::varint8(message_type);
+  uint8_t type_varint_len = ProtoSize::varint16(message_type);
   uint8_t total_header_len = 1 + size_varint_len + type_varint_len;
 
   // The header is right-justified within the padding so it sits immediately before payload.
@@ -292,12 +290,12 @@ ESPHOME_ALWAYS_INLINE static inline uint8_t write_plaintext_header(uint8_t *buf_
 
   // Encode varints directly into buffer using pre-computed lengths
   encode_varint_16(payload_size, size_varint_len, buf_start + header_offset + 1);
-  encode_varint_8(message_type, type_varint_len, buf_start + header_offset + 1 + size_varint_len);
+  encode_varint_16(message_type, type_varint_len, buf_start + header_offset + 1 + size_varint_len);
 
   return total_header_len;
 }
 
-APIError APIPlaintextFrameHelper::write_protobuf_packet(uint8_t type, ProtoWriteBuffer buffer) {
+APIError APIPlaintextFrameHelper::write_protobuf_packet(uint16_t type, ProtoWriteBuffer buffer) {
 #ifdef ESPHOME_DEBUG_API
   assert(this->state_ == State::DATA);
 #endif

esphome/components/api/api_frame_helper_plaintext.h

@@ -10,7 +10,8 @@ class APIPlaintextFrameHelper final : public APIFrameHelper {
   // Plaintext header structure (worst case):
   // Pos 0: indicator (0x00)
   // Pos 1-3: payload size varint (up to 3 bytes)
-  // Pos 4-5: message type varint (up to 2 bytes)
+  // Pos 4-5: message type varint (up to 2 bytes; covers message IDs up to
+  //          16383, enforced by the proto codegen)
   // Pos 6+: actual payload data
   static constexpr uint8_t HEADER_PADDING = 1 + 3 + 2;  // indicator + size varint + type varint
 
@@ -21,7 +22,7 @@ class APIPlaintextFrameHelper final : public APIFrameHelper {
   APIError init() override;
   APIError loop() override;
   APIError read_packet(ReadPacketBuffer *buffer) override;
-  APIError write_protobuf_packet(uint8_t type, ProtoWriteBuffer buffer) override;
+  APIError write_protobuf_packet(uint16_t type, ProtoWriteBuffer buffer) override;
   APIError write_protobuf_messages(ProtoWriteBuffer buffer, std::span<const MessageInfo> messages) override;
 #ifdef USE_API_NOISE
   // After try_read_frame_ returned PROTOCOL_SWITCH_TO_NOISE: copy out the

esphome/components/api/proto.h

@@ -684,11 +684,6 @@ class ProtoSize {
     return value < VARINT_THRESHOLD_1_BYTE ? 1 : (value < VARINT_THRESHOLD_2_BYTE ? 2 : 3);
   }
 
-  // Varint encoded length for an 8-bit value (1 or 2 bytes).
-  static constexpr inline uint8_t ESPHOME_ALWAYS_INLINE varint8(uint8_t value) {
-    return value < VARINT_THRESHOLD_1_BYTE ? 1 : 2;
-  }
-
   /**
    * @brief Calculates the size in bytes needed to encode a uint32_t value as a varint
    *

esphome/espidf/framework.py

@@ -1,7 +1,6 @@
 """ESP-IDF framework tools for ESPHome."""
 
 from collections.abc import Callable
-from concurrent.futures import ThreadPoolExecutor
 from ctypes.util import find_library
 import json
 import logging
@@ -16,7 +15,6 @@ import platformdirs
 
 from esphome.core import CORE, Version
 from esphome.framework_helpers import (
-    BatchDownloadProgress,
     PathType,
     archive_extract_all,
     create_venv,
@@ -692,12 +690,6 @@ def _patch_tools_json_demote_unused_tools(framework_path: Path) -> None:
     )
 
 
-# Tool archives are large (tens to hundreds of MB) and served by GitHub /
-# dl.espressif.com; a few streams at once saturate most links without
-# hammering the host. Smaller than external_files' 8: those are tiny files.
-_PREFETCH_WORKERS = 4
-
-
 def _prefetch_idf_tool_archives(
     framework_path: Path,
     targets_str: str,
@@ -710,10 +702,10 @@ def _prefetch_idf_tool_archives(
     which makes large archives effectively impossible to fetch on unstable
     connections (#17703). This asks the framework's idf_tools (via
     ``get_tool_downloads.py``) which archives the coming install needs, then
-    downloads them into ``<IDF_TOOLS_PATH>/dist`` with
-    ``download_with_resume``, a few at a time under one combined progress
-    bar. The installer then finds the verified archives already in place
-    ("file ... is already downloaded") and never touches the network.
+    downloads each into ``<IDF_TOOLS_PATH>/dist`` with
+    ``download_with_resume``. The installer then finds the verified archives
+    already in place ("file ... is already downloaded") and never touches the
+    network.
 
     Strictly best-effort: any failure here just logs and returns, leaving
     ``idf_tools.py install`` to download whatever is missing exactly as
@@ -740,51 +732,21 @@ def _prefetch_idf_tool_archives(
             for entry in json.loads(stdout)
             if not (dist_path / entry["dest"]).is_file()
         ]
-        if not entries:
-            return
-        _LOGGER.info(
-            "Downloading %d ESP-IDF tool archive(s): %s",
-            len(entries),
-            ", ".join(entry["name"] for entry in entries),
-        )
-        # tools.json always carries sizes; should one be missing the combined
-        # bar could not be trusted, so show no bar at all (per-file bars from
-        # several threads would interleave) rather than a wrong one.
-        sizes = [entry["size"] for entry in entries]
-        progress = BatchDownloadProgress(
-            "Downloading ESP-IDF tools", sum(sizes) if all(sizes) else 0
-        )
-        # Reported after the bar is done so the warnings do not land on
-        # its row; list.append is atomic under the GIL.
-        failures: list[tuple[str, Exception]] = []
-
-        def _download(entry: dict) -> None:
-            tracker = progress.tracker()
+        for index, entry in enumerate(entries, start=1):
+            _LOGGER.info(
+                "Downloading %s (%d/%d) ...", entry["name"], index, len(entries)
+            )
             try:
                 download_with_resume(
                     entry["url"],
                     dist_path / entry["dest"],
                     sha256=entry["sha256"],
                     size=entry["size"],
-                    progress=tracker,
                 )
             except Exception as e:  # noqa: BLE001  # pylint: disable=broad-exception-caught
                 # Keep prefetching the remaining archives; the installer
                 # will retry this one itself (without resume).
-                tracker(0)
-                failures.append((entry["name"], e))
-
-        ex = ThreadPoolExecutor(max_workers=min(_PREFETCH_WORKERS, len(entries)))
-        try:
-            for future in [ex.submit(_download, entry) for entry in entries]:
-                future.result()
-        finally:
-            # On Ctrl-C drop the queued archives instead of downloading them
-            # all before the process can exit; in-flight ones still finish.
-            ex.shutdown(wait=True, cancel_futures=True)
-            progress.done()
-        for name, e in failures:
-            _LOGGER.warning("Could not prefetch %s: %s", name, e)
+                _LOGGER.warning("Could not prefetch %s: %s", entry["name"], e)
     except Exception as e:  # noqa: BLE001  # pylint: disable=broad-exception-caught
         # The installer downloads anything missing itself; never let the
         # prefetch become a new way for the install to fail.

esphome/framework_helpers.py

@@ -1,6 +1,6 @@
 """Generic toolchain installation helpers shared across framework implementations."""
 
-from collections.abc import Callable, Iterable
+from collections.abc import Iterable
 from contextlib import ExitStack
 import hashlib
 import io
@@ -10,7 +10,6 @@ import os
 from pathlib import Path
 import subprocess
 import sys
-import threading
 import time
 from typing import IO, TYPE_CHECKING
 
@@ -698,11 +697,7 @@ def _response_validator(resp: "requests.Response") -> str | None:
 
 
 def _stream_response_to_file(
-    resp: "requests.Response",
-    f: IO[bytes],
-    offset: int,
-    size: int | None = None,
-    progress: Callable[[int], None] | None = None,
+    resp: "requests.Response", f: IO[bytes], offset: int, size: int | None = None
 ) -> None:
     """Stream an open ``_open_ranged`` response body into ``f`` at ``offset``.
 
@@ -710,72 +705,21 @@ def _stream_response_to_file(
     (effective offset 0) discards the stale bytes. ``offset`` also seeds the
     progress bar so a resumed download shows overall progress. ``size`` is
     the known full file size; when None it is derived from the response's
-    content-length, and without either there is no progress bar. With
-    ``progress`` set, no bar is drawn here; the callback gets the absolute
-    byte count, seeded with ``offset`` and then after each chunk.
+    content-length, and without either there is no progress bar.
     """
     f.seek(offset)
     f.truncate(offset)
     total_size = size or offset + _content_length(resp)
     downloaded = offset
-    own_bar: ProgressBar | None = None
-    if progress is None:
-        own_bar = ProgressBar("Downloading") if total_size > 0 else None
-        progress = (
-            (lambda done: own_bar.update(done / total_size))
-            if own_bar
-            else (lambda _: None)
-        )
-    progress(downloaded)
+    progress = ProgressBar("Downloading") if total_size > 0 else None
     for chunk in resp.iter_content(chunk_size=256 * 1024):
         if chunk:
             f.write(chunk)
             downloaded += len(chunk)
-        progress(downloaded)
-    if own_bar is not None:
-        own_bar.update(1)
-
-
-class BatchDownloadProgress:
-    """One progress bar across several concurrent ``download_with_resume`` calls.
-
-    Each ``tracker()`` is a ``progress`` callback for one download; it reports
-    that file's absolute byte count and the bar shows the sum over ``total``.
-    The lock also serialises the bar's stderr writes, so worker threads never
-    interleave frames. With an unknown ``total`` (0) nothing is drawn. Call
-    ``done()`` once every download has finished (or failed) so a bar that
-    never reached 100% still ends its line before the next log message.
-    """
-
-    def __init__(self, header: str, total: int) -> None:
-        self._bar = ProgressBar(header) if total > 0 else None
-        self._total = total
-        self._sum = 0
-        self._lock = threading.Lock()
-
-    def tracker(self) -> Callable[[int], None]:
-        last = 0
-
-        def update(done: int) -> None:
-            nonlocal last
-            if self._bar is None:
-                return
-            with self._lock:
-                self._sum += done - last
-                last = done
-                self._bar.update(min(self._sum / self._total, 1))
-
-        return update
-
-    def done(self) -> None:
-        # Nothing to end unless a frame was drawn and it was not the final
-        # one (update(1) already emitted its own newline).
-        if (
-            self._bar is not None
-            and self._bar.last_progress is not None
-            and self._bar.last_progress != 100
-        ):
-            self._bar.done()
+        if progress is not None:
+            progress.update(downloaded / total_size)
+    if progress is not None:
+        progress.update(1)
 
 
 def download_with_resume(
@@ -788,7 +732,6 @@ def download_with_resume(
     attempts: int = 5,
     timeout: int = 30,
     retry_connect_errors: bool = True,
-    progress: Callable[[int], None] | None = None,
 ) -> None:
     """Download ``url`` to ``dest``, resuming partial downloads.
 
@@ -811,12 +754,6 @@ def download_with_resume(
     of consuming attempts — for callers with their own fallback, like
     ``download_from_mirrors``.
 
-    ``progress``, when given, replaces the built-in progress bar: it is called
-    with the absolute number of bytes of ``dest`` obtained so far (including
-    a resumed prefix, and the final size once the file is verified), so a
-    caller running several downloads at once can draw one combined bar (see
-    ``BatchDownloadProgress``).
-
     Raises EsphomeError when all attempts are exhausted.
     """
     # Imported lazily: requests is a heavy import (~85ms) and is only needed
@@ -840,8 +777,6 @@ def download_with_resume(
     if dest.is_file() and (sha256 is not None or size is not None):
         try:
             _verify_file(dest, sha256, size)
-            if progress is not None:
-                progress(size if size is not None else dest.stat().st_size)
             return
         except EsphomeError:
             dest.unlink()
@@ -887,7 +822,7 @@ def download_with_resume(
                             # Recorded so a later run can prove an If-Range
                             # resume of this part file safe.
                             _write_download_meta(meta, url, validator, expected_total)
-                        _stream_response_to_file(resp, f, offset, size, progress)
+                        _stream_response_to_file(resp, f, offset, size)
             # else: a previous run already wrote every byte (or more) but
             # was killed before the rename below. Skip the network entirely
             # — a Range request past EOF would draw HTTP 416 — and let
@@ -896,10 +831,6 @@ def download_with_resume(
 
             expected_size = size if size is not None else expected_total
             _verify_file(part, sha256, expected_size or None)
-            if progress is not None:
-                # Also credits a part file an earlier run completed without
-                # streaming anything this time.
-                progress(expected_size or part.stat().st_size)
             if not expected_size and sha256 is None:
                 # No sha, no size, and the server sent no usable
                 # content-length: nothing can prove the download complete

script/api_protobuf/api_protobuf.py

@@ -475,6 +475,19 @@ TYPE_INFO: dict[int, TypeInfo] = {}
 # TYPE_DOUBLE = 1, TYPE_FIXED64 = 6, TYPE_SFIXED64 = 16, TYPE_SINT64 = 18
 UNSUPPORTED_TYPES = {1: "double", 6: "fixed64", 16: "sfixed64", 18: "sint64"}
 
+# The plaintext frame header budgets 2 varint bytes for the message type
+# (APIPlaintextFrameHelper::HEADER_PADDING), which caps message IDs at 16383.
+MAX_MESSAGE_ID = 16383
+
+
+def validate_message_id(message_id: int, message_name: str) -> None:
+    """Reject message IDs whose plaintext type varint would not fit in 2 bytes."""
+    if message_id > MAX_MESSAGE_ID:
+        raise ValueError(
+            f"Message ID {message_id} for {message_name} exceeds the plaintext "
+            f"2-byte type varint maximum ({MAX_MESSAGE_ID})"
+        )
+
 
 def validate_field_type(field_type: int, field_name: str = "") -> None:
     """Validate that the field type is supported by ESPHome API.
@@ -2511,14 +2524,10 @@ def build_message_type(
 
     # Add MESSAGE_TYPE method if this is a service message
     if message_id is not None:
-        # Validate that message_id fits in uint8_t
-        if message_id > 255:
-            raise ValueError(
-                f"Message ID {message_id} for {desc.name} exceeds uint8_t maximum (255)"
-            )
+        validate_message_id(message_id, desc.name)
 
         # Add static constexpr for message type
-        public_content.append(f"static constexpr uint8_t MESSAGE_TYPE = {message_id};")
+        public_content.append(f"static constexpr uint16_t MESSAGE_TYPE = {message_id};")
 
         # Add estimated size constant
         estimated_size = calculate_message_estimated_size(desc)
@@ -3174,8 +3183,12 @@ def main() -> None:
 #include "api_pb2_includes.h"
 """
 
-    content += """
-namespace esphome::api {
+    content += f"""
+namespace esphome::api {{
+
+// Upper bound on message IDs, enforced by the code generator: the plaintext
+// frame header budgets 2 varint bytes for the type (HEADER_PADDING).
+static constexpr uint16_t MAX_MESSAGE_TYPE = {MAX_MESSAGE_ID};
 
 """
 

tests/integration/conftest.py

@@ -60,7 +60,11 @@ def _get_platformio_env(cache_dir: Path) -> dict[str, str]:
     env = os.environ.copy()
     env["PLATFORMIO_CORE_DIR"] = str(cache_dir)
     env["PLATFORMIO_CACHE_DIR"] = str(cache_dir / ".cache")
-    env["PLATFORMIO_LIBDEPS_DIR"] = str(cache_dir / "libdeps")
+    # libdeps is keyed only by env name (the device name), and fixtures share
+    # names; two xdist workers first-compiling the same name race pio pkg
+    # install in the same directory. Keep libdeps per worker.
+    worker = os.environ.get("PYTEST_XDIST_WORKER", "master")
+    env["PLATFORMIO_LIBDEPS_DIR"] = str(cache_dir / "libdeps" / worker)
     # Prevent cache cleaning during integration tests
     env["ESPHOME_SKIP_CLEAN_BUILD"] = "1"
     # Compile with THIS tree's esphome sources, not wherever the venv's editable

tests/unit_tests/components/api/test_api_protobuf_generator.py

@@ -15,7 +15,12 @@ import pytest
 
 sys.path.insert(0, str(Path(__file__).parents[4] / "script" / "api_protobuf"))
 
-from api_protobuf import _make_ifdef_line, get_varint64_ifdef  # noqa: E402
+from api_protobuf import (  # noqa: E402
+    MAX_MESSAGE_ID,
+    _make_ifdef_line,
+    get_varint64_ifdef,
+    validate_message_id,
+)
 from google.protobuf import descriptor_pb2  # noqa: E402
 
 
@@ -91,3 +96,14 @@ def test_make_ifdef_line_conjunction_and_negation() -> None:
     assert (
         _make_ifdef_line("USE_X && !USE_Y") == "#if defined(USE_X) && !defined(USE_Y)"
     )
+
+
+def test_message_id_at_maximum_is_accepted() -> None:
+    # 16383 is the largest ID whose plaintext type varint fits the 2 bytes
+    # budgeted in HEADER_PADDING.
+    validate_message_id(MAX_MESSAGE_ID, "MaxMessage")
+
+
+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")

tests/unit_tests/test_espidf_framework.py

@@ -2,7 +2,6 @@
 
 # pylint: disable=protected-access
 
-from concurrent.futures import ThreadPoolExecutor
 from contextlib import contextmanager
 import importlib.util
 import io
@@ -15,7 +14,7 @@ import subprocess
 import sys
 import tarfile
 from types import SimpleNamespace
-from unittest.mock import MagicMock, patch
+from unittest.mock import patch
 
 import pytest
 
@@ -896,72 +895,16 @@ def test_prefetch_downloads_each_archive_with_resume(tmp_path: Path) -> None:
         ),
         patch("esphome.espidf.framework.download_with_resume") as download,
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
-        patch("esphome.espidf.framework.BatchDownloadProgress") as progress_cls,
     ):
         _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
 
     dist = get_idf_tools_path() / "dist"
-    # Archives download concurrently, so the call order is not fixed.
-    calls = {call[0]: call[1] for call in download.call_args_list}
-    assert set(calls) == {
-        ("https://example.com/cmake.tar.gz", dist / "cmake-3.30.2.tar.gz"),
-        ("https://example.com/ninja.zip", dist / "ninja.zip"),
-    }
-    kwargs = calls[("https://example.com/cmake.tar.gz", dist / "cmake-3.30.2.tar.gz")]
-    assert kwargs["sha256"] == "ab" * 32
-    assert kwargs["size"] == 123
-    # every archive reports into the one combined progress bar
-    progress_cls.assert_called_once_with("Downloading ESP-IDF tools", 123 + 45)
-    tracker = progress_cls.return_value.tracker.return_value
-    assert all(kw["progress"] is tracker for kw in calls.values())
-
-
-def test_prefetch_downloads_archives_concurrently(tmp_path: Path) -> None:
-    """More than one archive fans out over a bounded thread pool."""
-    entries = [
-        {
-            "name": f"tool{i}@1",
-            "url": f"https://example.com/tool{i}.tar.gz",
-            "size": 10,
-            "sha256": "ab" * 32,
-            "dest": f"tool{i}.tar.gz",
-        }
-        for i in range(6)
-    ]
-    with (
-        patch(
-            "esphome.espidf.framework.run_command",
-            return_value=(True, json.dumps(entries), ""),
-        ),
-        patch("esphome.espidf.framework.download_with_resume") as download,
-        patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
-        patch(
-            "esphome.espidf.framework.ThreadPoolExecutor", wraps=ThreadPoolExecutor
-        ) as pool,
-    ):
-        _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
-
-    pool.assert_called_once_with(max_workers=4)
-    assert download.call_count == 6
-
-
-def test_prefetch_single_archive_uses_one_worker(tmp_path: Path) -> None:
-    entries = json.loads(_PREFETCH_JSON)[:1]
-    with (
-        patch(
-            "esphome.espidf.framework.run_command",
-            return_value=(True, json.dumps(entries), ""),
-        ),
-        patch("esphome.espidf.framework.download_with_resume") as download,
-        patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
-        patch(
-            "esphome.espidf.framework.ThreadPoolExecutor", wraps=ThreadPoolExecutor
-        ) as pool,
-    ):
-        _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
-
-    pool.assert_called_once_with(max_workers=1)
-    assert download.call_count == 1
+    assert download.call_count == 2
+    assert download.call_args_list[0][0] == (
+        "https://example.com/cmake.tar.gz",
+        dist / "cmake-3.30.2.tar.gz",
+    )
+    assert download.call_args_list[0][1] == {"sha256": "ab" * 32, "size": 123}
 
 
 def test_prefetch_skips_already_downloaded_archives(tmp_path: Path) -> None:
@@ -1021,11 +964,6 @@ def test_prefetch_one_failed_archive_does_not_stop_the_rest(
 ) -> None:
     """A single archive failing its download must not abort the prefetch of
     the remaining archives."""
-
-    def _fail_cmake_download(url: str, *args, **kwargs) -> None:
-        if "cmake" in url:
-            raise OSError("network down")
-
     with (
         patch(
             "esphome.espidf.framework.run_command",
@@ -1033,7 +971,7 @@ def test_prefetch_one_failed_archive_does_not_stop_the_rest(
         ),
         patch(
             "esphome.espidf.framework.download_with_resume",
-            side_effect=_fail_cmake_download,
+            side_effect=[OSError("network down"), None],
         ) as download,
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
     ):
@@ -1043,29 +981,6 @@ def test_prefetch_one_failed_archive_does_not_stop_the_rest(
     assert "Could not prefetch cmake@3.30.2" in caplog.text
 
 
-def test_prefetch_finishes_progress_bar_and_cancels_queue(tmp_path: Path) -> None:
-    """The batch bar is closed out after the pool, and the pool is shut down
-    with cancel_futures so Ctrl-C does not drain every queued archive."""
-    with (
-        patch(
-            "esphome.espidf.framework.run_command",
-            return_value=(True, _PREFETCH_JSON, ""),
-        ),
-        patch("esphome.espidf.framework.download_with_resume"),
-        patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
-        patch("esphome.espidf.framework.BatchDownloadProgress") as progress_cls,
-        patch(
-            "esphome.espidf.framework.ThreadPoolExecutor", wraps=ThreadPoolExecutor
-        ) as pool_cls,
-    ):
-        pool = MagicMock(wraps=ThreadPoolExecutor(max_workers=2))
-        pool_cls.return_value = pool
-        _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
-
-    pool.shutdown.assert_called_once_with(wait=True, cancel_futures=True)
-    progress_cls.return_value.done.assert_called_once_with()
-
-
 def test_prefetch_passes_targets_and_tools_to_script(tmp_path: Path) -> None:
     with (
         patch(

tests/unit_tests/test_framework_helpers.py

@@ -21,7 +21,6 @@ import requests as req
 from esphome import framework_helpers
 from esphome.core import EsphomeError
 from esphome.framework_helpers import (
-    BatchDownloadProgress,
     _7z_extract_all,
     _detect_archive_root,
     _is_transient_download_error,
@@ -1113,108 +1112,6 @@ class TestDownloadWithResume:
         assert mock_get.call_args[1]["headers"] == {}
         assert dest.read_bytes() == b"data"
 
-    def test_progress_callback_reports_absolute_bytes(self, tmp_path: Path) -> None:
-        """With a callback no bar is drawn; the callback sees the running
-        byte count of this file, then its final verified size."""
-        dest = tmp_path / "tool.tar.gz"
-        resp = _mock_response(b"")
-        resp.headers = {"content-length": "7"}
-        resp.iter_content.return_value = [b"1234", b"567"]
-        seen: list[int] = []
-        with (
-            patch("requests.get", return_value=resp),
-            patch("esphome.framework_helpers.ProgressBar") as bar,
-        ):
-            download_with_resume(
-                "https://example.com/t", dest, size=7, progress=seen.append
-            )
-        assert seen == [0, 4, 7, 7]
-        bar.assert_not_called()
-
-    def test_progress_callback_seeds_with_resume_offset(self, tmp_path: Path) -> None:
-        dest = tmp_path / "tool.tar.gz"
-        (tmp_path / "tool.tar.gz.part").write_bytes(b"12345")
-        good = hashlib.sha256(b"12345678").hexdigest()
-        seen: list[int] = []
-        with patch("requests.get", return_value=_resumed_response(b"678")):
-            download_with_resume(
-                "https://example.com/t", dest, sha256=good, size=8, progress=seen.append
-            )
-        assert seen[0] == 5
-        assert seen[-1] == 8
-
-    def test_progress_callback_credits_already_complete_download(
-        self, tmp_path: Path
-    ) -> None:
-        """A verified dest from an earlier run still counts toward the batch."""
-        dest = tmp_path / "tool.tar.gz"
-        dest.write_bytes(b"12345678")
-        seen: list[int] = []
-        with patch("requests.get") as mock_get:
-            download_with_resume(
-                "https://example.com/t", dest, size=8, progress=seen.append
-            )
-        mock_get.assert_not_called()
-        assert seen == [8]
-
-
-class TestBatchDownloadProgress:
-    def test_sums_trackers_into_one_bar(self) -> None:
-        with patch("esphome.framework_helpers.ProgressBar") as bar_cls:
-            progress = BatchDownloadProgress("Downloading", 100)
-            a = progress.tracker()
-            b = progress.tracker()
-            a(10)
-            b(20)
-            a(30)
-            a(0)  # a restart from zero takes that file's bytes back out
-        bar_cls.assert_called_once_with("Downloading")
-        updates = [c[0][0] for c in bar_cls.return_value.update.call_args_list]
-        assert updates == [0.1, 0.3, 0.5, 0.2]
-
-    def test_clamps_at_one(self) -> None:
-        """Sizes are advisory; an over-delivering server never pushes past 100%."""
-        with patch("esphome.framework_helpers.ProgressBar") as bar_cls:
-            progress = BatchDownloadProgress("Downloading", 10)
-            progress.tracker()(25)
-        assert bar_cls.return_value.update.call_args[0][0] == 1
-
-    def test_unknown_total_draws_nothing(self) -> None:
-        with patch("esphome.framework_helpers.ProgressBar") as bar_cls:
-            progress = BatchDownloadProgress("Downloading", 0)
-            progress.tracker()(5)
-            progress.done()
-        bar_cls.assert_not_called()
-
-    def test_done_ends_an_unfinished_bar(self) -> None:
-        """A batch that stops short of 100% (a failed archive) still ends its
-        line so the next log message starts on a fresh row."""
-        stream = io.StringIO()
-        stream.isatty = lambda: True  # type: ignore[method-assign]
-        with patch("esphome.helpers.sys.stderr", stream):
-            progress = BatchDownloadProgress("Downloading", 10)
-            progress.tracker()(5)
-            progress.done()
-        assert stream.getvalue().endswith("50% \n")
-
-    def test_done_before_any_frame_writes_nothing(self) -> None:
-        """A batch aborted before any tracker fired must not emit a stray
-        newline for a bar that was never drawn."""
-        stream = io.StringIO()
-        stream.isatty = lambda: True  # type: ignore[method-assign]
-        with patch("esphome.helpers.sys.stderr", stream):
-            BatchDownloadProgress("Downloading", 10).done()
-        assert stream.getvalue() == ""
-
-    def test_done_after_full_bar_adds_nothing(self) -> None:
-        stream = io.StringIO()
-        stream.isatty = lambda: True  # type: ignore[method-assign]
-        with patch("esphome.helpers.sys.stderr", stream):
-            progress = BatchDownloadProgress("Downloading", 10)
-            progress.tracker()(10)
-            progress.done()
-        assert stream.getvalue().endswith("100% Done...\r\n")
-
 
 class TestDownloadFromMirrors:
     def test_success_returns_url_and_writes_content(self, tmp_path: Path) -> None: