From 19866c60ee7b88cfbc5417ebea453029e7326422 Mon Sep 17 00:00:00 2001
From: "J. Nick Koston" <nick@koston.org>
Date: Mon, 24 Aug 2026 15:59:00 -0500
Subject: [PATCH] [core] Parallelize registry and tool downloads (#18662)

---
 esphome/espidf/framework.py                 | 100 +++--
 esphome/framework_helpers.py                | 237 +++++++++++-
 esphome/helpers.py                          |  13 +-
 esphome/platformio/library.py               | 393 +++++++++++++-------
 tests/unit_tests/test_espidf_component.py   |  16 +-
 tests/unit_tests/test_espidf_framework.py   | 186 ++++++++-
 tests/unit_tests/test_espota2.py            |   8 +-
 tests/unit_tests/test_framework_helpers.py  | 275 ++++++++++++++
 tests/unit_tests/test_helpers.py            |  14 +
 tests/unit_tests/test_platformio_library.py | 195 +++++++++-
 10 files changed, 1242 insertions(+), 195 deletions(-)

diff --git a/esphome/espidf/framework.py b/esphome/espidf/framework.py
index 0f6ef873b8..179346e072 100644
--- a/esphome/espidf/framework.py
+++ b/esphome/espidf/framework.py
@@ -2,6 +2,7 @@
 
 from collections.abc import Callable
 from ctypes.util import find_library
+from functools import partial
 import json
 import logging
 import os
@@ -20,9 +21,11 @@ from esphome.framework_helpers import (
     create_venv,
     download_from_mirrors,
     download_with_resume,
+    failure_reason,
     get_python_env_executable_path,
     get_system_python_path,
     rmdir,
+    run_batch_downloads,
     run_command,
     run_command_ok,
     str_to_lst_of_str,
@@ -690,6 +693,18 @@ def _patch_tools_json_demote_unused_tools(framework_path: Path) -> None:
     )
 
 
+def _download_tool(
+    dist_path: Path, entry: dict, tracker: Callable[[int], None]
+) -> None:
+    download_with_resume(
+        entry["url"],
+        dist_path / entry["dest"],
+        sha256=entry["sha256"],
+        size=entry["size"],
+        progress=tracker,
+    )
+
+
 def _prefetch_idf_tool_archives(
     framework_path: Path,
     targets_str: str,
@@ -702,10 +717,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 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.
+    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.
 
     Strictly best-effort: any failure here just logs and returns, leaving
     ``idf_tools.py install`` to download whatever is missing exactly as
@@ -727,30 +742,67 @@ def _prefetch_idf_tool_archives(
             )
             return
         dist_path = get_idf_tools_path() / "dist"
-        entries = [
-            entry
-            for entry in json.loads(stdout)
-            if not (dist_path / entry["dest"]).is_file()
-        ]
-        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"],
+        entries = []
+        seen_dests: set[str] = set()
+        for entry in json.loads(stdout):
+            if (dist_path / entry["dest"]).is_file():
+                continue
+            # Never download unverified: an entry without sha256/size is
+            # left to the installer, which fails loudly on a bad archive.
+            # Checked before the dedupe so it cannot shadow a verifiable
+            # duplicate of the same dest.
+            if not (entry.get("sha256") and entry.get("size")):
+                _LOGGER.warning(
+                    "Tool %s has no sha256/size in the download list; "
+                    "leaving it to the installer",
+                    entry["name"],
                 )
-            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).
-                _LOGGER.warning("Could not prefetch %s: %s", entry["name"], e)
+                continue
+            if entry["dest"] in seen_dests:
+                # Two workers on one .part file would interleave
+                # seek/truncate writes; mirror the library prefetch's dedupe
+                continue
+            seen_dests.add(entry["dest"])
+            entries.append(entry)
+        if not entries:
+            return
+        _LOGGER.info(
+            "Downloading %d ESP-IDF tool archive(s): %s",
+            len(entries),
+            ", ".join(entry["name"] for entry in entries),
+        )
+
+        # No sequential fallback here: skipping the prefetch would lose the
+        # resume workaround for #17703, and every entry has a size (above).
+        # A failed archive is retried by the installer itself (without
+        # resume); keep prefetching the rest.
+        failures = run_batch_downloads(
+            "Downloading ESP-IDF tools",
+            [
+                (
+                    entry["name"],
+                    entry["size"],
+                    partial(_download_tool, dist_path, entry),
+                )
+                for entry in entries
+            ],
+        )
+        for name, e in failures:
+            # failure_reason: a message-less exception must not log blank
+            _LOGGER.warning("Could not prefetch %s: %s", name, failure_reason(e))
+            _LOGGER.debug("Prefetch failure detail", exc_info=e)
+        if len(failures) == len(entries):
+            # A systematic fault, not one flaky mirror: the resume
+            # workaround (#17703) is off for this whole install
+            _LOGGER.error(
+                "Every ESP-IDF tool prefetch failed; the installer will "
+                "download without resume"
+            )
     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.
-        _LOGGER.warning("ESP-IDF tool prefetch failed: %s", e)
+        _LOGGER.warning("ESP-IDF tool prefetch failed: %s", failure_reason(e))
+        _LOGGER.debug("Prefetch failure detail", exc_info=True)
 
 
 def _check_esphome_idf_framework_install(
diff --git a/esphome/framework_helpers.py b/esphome/framework_helpers.py
index 105791c518..2a2ce6dacf 100644
--- a/esphome/framework_helpers.py
+++ b/esphome/framework_helpers.py
@@ -1,7 +1,8 @@
 """Generic toolchain installation helpers shared across framework implementations."""
 
-from collections.abc import Iterable
-from contextlib import ExitStack
+from collections.abc import Callable, Iterable, Iterator
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import ExitStack, contextmanager, suppress
 import hashlib
 import io
 import json
@@ -10,6 +11,7 @@ import os
 from pathlib import Path
 import subprocess
 import sys
+import threading
 import time
 from typing import IO, TYPE_CHECKING
 
@@ -24,6 +26,7 @@ PathType = str | os.PathLike
 
 _LOGGER = logging.getLogger(__name__)
 
+
 # Attempts per mirror URL before falling through to the next mirror; only
 # mid-stream drops retry (resuming when the server gave a validator),
 # connect errors move on to the next mirror immediately.
@@ -699,7 +702,11 @@ 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
+    resp: "requests.Response",
+    f: IO[bytes],
+    offset: int,
+    size: int | None = None,
+    progress: Callable[[int], None] | None = None,
 ) -> None:
     """Stream an open ``_open_ranged`` response body into ``f`` at ``offset``.
 
@@ -707,21 +714,187 @@ 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.
+    content-length, and without either there is no bar. With ``progress``
+    set no bar is drawn here; the callback gets the absolute byte count.
     """
     f.seek(offset)
     f.truncate(offset)
     total_size = size or offset + _content_length(resp)
     downloaded = offset
-    progress = ProgressBar("Downloading") if total_size > 0 else None
+    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)
     for chunk in resp.iter_content(chunk_size=256 * 1024):
         if chunk:
             f.write(chunk)
             downloaded += len(chunk)
-        if progress is not None:
-            progress.update(downloaded / total_size)
-    if progress is not None:
-        progress.update(1)
+        progress(downloaded)
+    if own_bar is not None:
+        own_bar.update(1)
+
+
+# Concurrent downloads per batch; enough to hide latency without
+# hammering the host or the mirrors.
+BATCH_DOWNLOAD_WORKERS = 4
+
+
+def run_batch_downloads(
+    header: str,
+    jobs: list[tuple[str, int, Callable[[Callable[[int], None]], None]]],
+    max_workers: int = BATCH_DOWNLOAD_WORKERS,
+) -> list[tuple[str, BaseException]]:
+    """Run ``(name, size, fetch)`` download jobs concurrently under one bar.
+
+    Each ``fetch(tracker)`` reports absolute byte counts; the bar total is
+    the sum of the sizes. Failures are returned after the bar is done so
+    warnings never land on its row. Ctrl-C drops queued jobs and aborts
+    in-flight ones at their next progress tick or backoff boundary (a
+    parked socket read defers that by its timeout, and an in-progress
+    archive extraction runs to completion); resumable destinations
+    (``download_with_resume``) keep their fetched ``.part`` bytes.
+    ``jobs`` must be non-empty.
+    """
+    progress = _BatchDownloadProgress(header, sum(size for _, size, _ in jobs))
+    cancelled = threading.Event()
+
+    def _run(
+        name: str, fetch: Callable[[Callable[[int], None]], None]
+    ) -> tuple[str, BaseException] | None:
+        tracker = progress.tracker()
+
+        def checked(done: int) -> None:
+            if cancelled.is_set():
+                raise _BatchDownloadCancelled
+            tracker(done)
+
+        try:
+            fetch(checked)
+        except (_BatchDownloadCancelled, Exception) as err:  # noqa: BLE001  # pylint: disable=broad-exception-caught
+            # The cancelled arm exists for the tracker rollback below; the
+            # batch re-raises the interrupt, so the list is never returned
+            # after Ctrl-C. A bar-frame write failure must not displace the
+            # download error.
+            with suppress(Exception):
+                tracker(0)
+            failure = (name, err)
+        else:
+            failure = None
+        return failure
+
+    ex = ThreadPoolExecutor(max_workers=max_workers)
+    try:
+        with progress.logging_guard():
+            futures = [ex.submit(_run, name, fetch) for name, _, fetch in jobs]
+            return [failure for f in futures if (failure := f.result()) is not None]
+    except BaseException:
+        # Without this the non-daemon workers download to completion before
+        # the interpreter can exit, making Ctrl-C ineffective for minutes
+        cancelled.set()
+        raise
+    finally:
+        ex.shutdown(wait=True, cancel_futures=True)
+        progress.done()
+
+
+class _BatchDownloadCancelled(BaseException):
+    """Raised inside a download job to abandon it after Ctrl-C.
+
+    BaseException, like KeyboardInterrupt: a broad ``except Exception`` in
+    the download layers must not convert an abort into a retry.
+    """
+
+
+class _BatchDownloadProgress:
+    """One bar across several concurrent downloads, summing tracker bytes.
+
+    The lock also serialises stderr writes so workers never interleave
+    frames; a ``total`` of 0 draws nothing. Call ``done()`` at the end so a
+    bar short of 100% still ends its line.
+    """
+
+    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]:
+        if self._bar is None:
+            return lambda _: None
+        last = 0
+
+        def update(done: int) -> None:
+            nonlocal last
+            with self._lock:
+                self._sum += done - last
+                last = done
+                # A bar-write failure (broken stderr pipe) must not surface
+                # as a download failure and cost the .part file
+                with suppress(Exception):
+                    self._bar.update(min(self._sum / self._total, 1))
+
+        return update
+
+    def done(self) -> None:
+        if self._bar is not None:
+            self._bar.done()
+
+    @contextmanager
+    def logging_guard(self) -> Iterator[None]:
+        r"""End a partial bar row before any log record while active.
+
+        Worker warnings (mirror retries) share stderr with the bar's \r
+        frames; without this the record lands mid-row and the next frame
+        overwrites it. A handler-level filter runs just before emit, so
+        only a tiny window remains for a concurrent frame.
+        """
+        the_bar = self._bar
+        if the_bar is None:
+            yield
+            return
+        lock = self._lock
+
+        class _EndRow(logging.Filter):
+            def filter(self, record: logging.LogRecord) -> bool:
+                # Handler.handle() runs filters outside handleError's try; a
+                # stderr write failure must not escape through the log call
+                with lock, suppress(Exception):
+                    the_bar.interrupt()
+                return True
+
+        end_row = _EndRow()
+        handlers = logging.getLogger().handlers
+        for handler in handlers:
+            handler.addFilter(end_row)
+        try:
+            yield
+        finally:
+            for handler in handlers:
+                handler.removeFilter(end_row)
+
+
+def _part_path(dest: Path) -> Path:
+    """The in-progress sidecar ``download_with_resume`` streams into."""
+    return dest.with_name(dest.name + ".part")
+
+
+def _cancellable_sleep(
+    delay: float, progress: Callable[[int], None] | None, done: int
+) -> None:
+    """Backoff sleep that still observes a batch cancellation tick."""
+    if progress is None:
+        time.sleep(delay)
+        return
+    end = time.monotonic() + delay
+    while (remaining := end - time.monotonic()) > 0:
+        progress(done)  # raises when the batch was cancelled
+        time.sleep(min(0.5, remaining))
 
 
 def download_with_resume(
@@ -734,6 +907,7 @@ 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.
 
@@ -756,6 +930,9 @@ def download_with_resume(
     of consuming attempts — for callers with their own fallback, like
     ``download_from_mirrors``.
 
+    ``progress`` replaces the built-in bar: it receives the absolute bytes of
+    ``dest`` obtained so far (see ``_BatchDownloadProgress``).
+
     Raises EsphomeError when all attempts are exhausted.
     """
     # Imported lazily: requests is a heavy import (~85ms) and is only needed
@@ -767,7 +944,7 @@ def download_with_resume(
     ensure_happy_eyeballs()
 
     dest = Path(dest)
-    part = dest.with_name(dest.name + ".part")
+    part = _part_path(dest)
     meta = part.with_name(part.name + ".meta")
     dest.parent.mkdir(parents=True, exist_ok=True)
     last_error: Exception | None = None
@@ -779,6 +956,8 @@ 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()
@@ -824,7 +1003,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)
+                        _stream_response_to_file(resp, f, offset, size, progress)
             # 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
@@ -833,6 +1012,10 @@ 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
@@ -880,11 +1063,11 @@ def download_with_resume(
 
     raise EsphomeError(
         f"Failed to download {url} after {attempts} attempts: "
-        f"{_failure_reason(last_error)}"
+        f"{failure_reason(last_error)}"
     ) from last_error
 
 
-def _failure_reason(e: Exception) -> str:
+def failure_reason(e: BaseException) -> str:
     """Format a download exception for the aggregated error message.
 
     ``requests`` appends " for url: <url>" to HTTP errors; the URL is already
@@ -900,7 +1083,7 @@ def _spent_attempts_error(e: Exception, attempts: int) -> Exception:
     the sweep classifies it as permanent."""
     from esphome.core import EsphomeError
 
-    err = EsphomeError(f"failed after {attempts} attempts: {_failure_reason(e)}")
+    err = EsphomeError(f"failed after {attempts} attempts: {failure_reason(e)}")
     err.__cause__ = e
     return err
 
@@ -911,6 +1094,7 @@ def _try_mirrors_once(
     f: IO[bytes] | None,
     timeout: int,
     failures: list[tuple[str, Exception]],
+    progress: Callable[[int], None] | None = None,
 ) -> str | None:
     """Single pass over the resolved mirror ``urls``, one try per URL.
 
@@ -939,6 +1123,7 @@ def _try_mirrors_once(
                     # next mirror immediately; only mid-stream drops
                     # retry-with-resume on the same URL.
                     retry_connect_errors=False,
+                    progress=progress,
                 )
                 return url
             except (requests.RequestException, OSError, EsphomeError) as e:
@@ -980,7 +1165,7 @@ def _try_mirrors_once(
                         if offset == 0:
                             validator = _response_validator(resp)
                             expected_total = _content_length(resp)
-                        _stream_response_to_file(resp, f, offset)
+                        _stream_response_to_file(resp, f, offset, progress=progress)
 
                 if expected_total and f.tell() != expected_total:
                     raise EsphomeError(
@@ -1029,6 +1214,7 @@ def download_from_mirrors(
     substitutions: dict[str, str],
     target: io.RawIOBase | IO[bytes] | PathType,
     timeout: int = 30,
+    progress: Callable[[int], None] | None = None,
 ) -> str:
     """
     Download file from multiple mirrors with substitution support.
@@ -1038,6 +1224,8 @@ def download_from_mirrors(
         substitutions: Dictionary of substitutions to apply to URLs
         target: Target file path or file-like object
         timeout: Download timeout in seconds
+        progress: Passed through to the download (see ``download_with_resume``);
+            replaces the built-in per-file bar
 
     Returns:
         The source URL.
@@ -1102,7 +1290,9 @@ def download_from_mirrors(
     for sweep in range(1, _MIRROR_SWEEP_ATTEMPTS + 1):
         sweep_failures: list[tuple[str, Exception]] = []
         if (
-            url := _try_mirrors_once(urls, path_target, f, timeout, sweep_failures)
+            url := _try_mirrors_once(
+                urls, path_target, f, timeout, sweep_failures, progress
+            )
         ) is not None:
             return url
         failures.extend(sweep_failures)
@@ -1119,12 +1309,21 @@ def download_from_mirrors(
             _LOGGER.warning(
                 "Download of %s failed (%s); retrying in %d seconds (attempt %d/%d)",
                 transient[0],
-                _failure_reason(transient[1]),
+                failure_reason(transient[1]),
                 delay,
                 sweep + 1,
                 _MIRROR_SWEEP_ATTEMPTS,
             )
-            time.sleep(delay)
+            # 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:
+                if f is not None:
+                    done = f.tell()
+                else:
+                    part = _part_path(path_target)
+                    done = part.stat().st_size if part.is_file() else 0
+            _cancellable_sleep(delay, progress, done)
 
     # 4. Report every attempted URL if all mirrors failed. failures spans
     # all sweeps (deduplicated by URL and reason), so neither an early
@@ -1133,7 +1332,7 @@ def download_from_mirrors(
         seen: set[tuple[str, str]] = set()
         attempts = ""
         for url, e in failures:
-            reason = _failure_reason(e)
+            reason = failure_reason(e)
             if (url, reason) not in seen:
                 seen.add((url, reason))
                 attempts += f"\n  {url}\n    {reason}"
diff --git a/esphome/helpers.py b/esphome/helpers.py
index b3102ca277..d30e9b16a2 100644
--- a/esphome/helpers.py
+++ b/esphome/helpers.py
@@ -738,11 +738,22 @@ class ProgressBar:
         sys.stderr.flush()
 
     def done(self) -> None:
-        if not self.enabled:
+        # No frame drawn, or the 100% frame already ended its own line
+        if not self.enabled or self.last_progress is None or self.last_progress == 100:
             return
         sys.stderr.write("\n")
         sys.stderr.flush()
 
+    def interrupt(self) -> None:
+        """End a mid-row frame so the next write starts on its own row.
+
+        The next ``update()`` redraws the bar; a finished bar stays done.
+        """
+        if self.last_progress == 100:
+            return
+        self.done()
+        self.last_progress = None
+
 
 def docs_url(path: str) -> str:
     """Return the URL to the documentation for a given path."""
diff --git a/esphome/platformio/library.py b/esphome/platformio/library.py
index a3899fa860..bf9c323b84 100644
--- a/esphome/platformio/library.py
+++ b/esphome/platformio/library.py
@@ -15,6 +15,7 @@ regardless of which toolchain consumes the result.
 from collections import deque
 from collections.abc import Callable, Iterable
 from dataclasses import dataclass, field
+from functools import partial
 import glob
 import hashlib
 import itertools
@@ -30,7 +31,13 @@ from urllib.request import url2pathname
 
 from esphome import git
 from esphome.core import CORE, EsphomeError, Library
-from esphome.framework_helpers import archive_extract_all, download_from_mirrors, rmdir
+from esphome.framework_helpers import (
+    archive_extract_all,
+    download_from_mirrors,
+    failure_reason,
+    rmdir,
+    run_batch_downloads,
+)
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -70,6 +77,10 @@ SRC_FILE_EXTENSIONS = list(SOURCE_KIND_FOR_SUFFIX)
 
 DOMAIN = "pio_components"
 
+# Marks a cache dir whose archive finished extracting; a missing marker
+# means a torn extraction that must be redone
+_EXTRACTED_MARKER = ".esphome_extracted"
+
 ESPHOME_DATA_KEY = "ESPHOME"
 ESPHOME_DATA_EXTRA_CMAKE_KEY = "EXTRA_CMAKE"
 # Captured extra-script LINKFLAGS; kept apart from build.flags so they reach
@@ -93,12 +104,13 @@ class Source:
 
 
 class URLSource(Source):
-    def __init__(self, url: str):
+    def __init__(self, url: str, size: int | None = None):
         self.url = url
+        # Archive size as reported by the registry, when known; sizes the
+        # combined prefetch bar without any extra network probe
+        self.size = size
 
-    def download(
-        self, dir_suffix: str, force: bool = False, salt: str = "", namespace: str = ""
-    ) -> Path:
+    def _cache_dir(self, dir_suffix: str, salt: str, namespace: str) -> Path:
         # Namespace the cache per backend (e.g. pio_components/idf, .../zephyr) so
         # the build files each backend writes into the library dir can't collide.
         base_dir = Path(CORE.data_dir) / DOMAIN
@@ -108,22 +120,40 @@ class URLSource(Source):
         h.update(self.url.encode())
         if salt:
             h.update(salt.encode())
-        path = base_dir / h.hexdigest()[:8] / dir_suffix
+        return base_dir / h.hexdigest()[:8] / dir_suffix
+
+    def is_cached(self, dir_suffix: str, salt: str = "", namespace: str = "") -> bool:
+        """Whether a completed extraction already exists for this source."""
+        return (
+            self._cache_dir(dir_suffix, salt, namespace) / _EXTRACTED_MARKER
+        ).is_file()
+
+    def download(
+        self,
+        dir_suffix: str,
+        force: bool = False,
+        salt: str = "",
+        namespace: str = "",
+        progress: Callable[[int], None] | None = None,
+    ) -> Path:
+        path = self._cache_dir(dir_suffix, salt, namespace)
         # Marker file written last to signal a complete extraction. Using a
         # marker (instead of just `path.is_dir()`) means an interrupted
         # extraction is correctly detected and re-run on the next invocation,
         # and lets us extract directly into ``path`` — avoiding a
         # post-extraction rename that races with antivirus on Windows.
-        extracted_marker = path / ".esphome_extracted"
+        extracted_marker = path / _EXTRACTED_MARKER
         if not extracted_marker.is_file() or force:
             rmdir(path, msg=f"Clean up library directory {path}")
 
             # Download in temporary file
             with tempfile.NamedTemporaryFile() as tmp:
-                _LOGGER.info("Downloading %s ...", self.url)
+                if progress is None:
+                    # A batch caller draws one combined bar and logs the list
+                    _LOGGER.info("Downloading %s ...", self.url)
                 _LOGGER.debug("Location: %s", path)
 
-                download_from_mirrors([self.url], {}, tmp.file)
+                download_from_mirrors([self.url], {}, tmp.file, progress=progress)
 
                 _LOGGER.debug("Extracting archive to %s ...", path)
                 archive_extract_all(tmp.file, path)
@@ -415,6 +445,27 @@ def split_list_by_condition(
     return matched, non_matched
 
 
+def _valid_manifest_shape(data: Any) -> bool:
+    """Whether the manifest has the dict shapes every backend dereferences.
+
+    A bare json.load imposes no shape; validating once here means a
+    malformed third-party manifest fails by library name instead of a raw
+    TypeError/AttributeError in a backend.
+    """
+    if not isinstance(data, dict):
+        return False
+    build = data.get("build", {})
+    esphome_data = data.get(ESPHOME_DATA_KEY, {})
+    return (
+        isinstance(build, dict)
+        and isinstance(esphome_data, dict)
+        and isinstance(esphome_data.get(ESPHOME_DATA_LINK_FLAGS_KEY, []), list)
+        and isinstance(build.get("srcDir", ""), str)
+        and isinstance(build.get("includeDir", ""), str)
+        and isinstance(build.get("srcFilter", ""), (str, list))
+    )
+
+
 def check_library_data(data: dict, platform: str | None, framework: str):
     """
     Check whether a library manifest is compatible with the target toolchain.
@@ -537,9 +588,10 @@ def _make_registry_client() -> Any:
 
 def _resolve_registry_version(
     owner: str | None, pkgname: str, requirements: set[str]
-) -> tuple[str, str, str, str]:
+) -> tuple[str, str, str, str, int | None]:
     """Resolve a registry package to the single highest version satisfying ALL
-    the given requirements; return ``(owner, name, version, download_url)``.
+    the given requirements; return ``(owner, name, version, download_url,
+    size)`` (``size`` is None when the registry omits it).
 
     Intersecting every requirement (rather than resolving each consumer in
     isolation) makes the result independent of processing order and guarantees
@@ -569,7 +621,7 @@ def _resolve_registry_version(
     pkgfile = registry.pick_compatible_pkg_file(best["files"])
     if not pkgfile:
         raise RuntimeError(f"No package file for {owner}/{name}@{best['name']}")
-    return owner, name, best["name"], pkgfile["download_url"]
+    return owner, name, best["name"], pkgfile["download_url"], pkgfile.get("size")
 
 
 def split_flag_entry(entry: Any, owner: str) -> list[str]:
@@ -859,6 +911,85 @@ def is_lib_ignored(name: str | None, lib_ignore: set[str]) -> bool:
     )
 
 
+def _fetch_source(
+    component: ConvertedLibrary,
+    salt: str,
+    namespace: str,
+    tracker: Callable[[int], None],
+) -> None:
+    # Straight to URLSource: only it takes progress, and mutating the
+    # shared component from a worker is the authoritative loop's job
+    component.source.download(
+        component.get_sanitized_name(), salt=salt, namespace=namespace, progress=tracker
+    )
+
+
+def _prefetch_wave(
+    wave: list[tuple[str, ConvertedLibrary]], salt: str, namespace: str
+) -> None:
+    """Best-effort parallel download of a wave's registry archives.
+
+    The walk's own ``download()`` stays authoritative; duplicate URLs
+    prefetch once so two threads never share a cache directory. Archives
+    whose size the registry did not report are left to the sequential
+    loop, whose per-file bars don't interleave. A node a sibling in the
+    same wave supersedes has its archive fetched in vain (knowing better
+    would need the manifests being downloaded).
+    """
+    try:
+        components: list[ConvertedLibrary] = []
+        seen: set[str] = set()
+        for _key, component in wave:
+            source = component.source
+            if not isinstance(source, URLSource) or not source.size:
+                continue
+            if source.url in seen:
+                continue
+            seen.add(source.url)
+            try:
+                cached = source.is_cached(
+                    component.get_sanitized_name(), salt=salt, namespace=namespace
+                )
+            except OSError as err:
+                # Best-effort, but visibly: a systematic probe failure makes
+                # every warm build re-download every archive
+                _LOGGER.warning("Cache probe for %s failed: %s", component.name, err)
+                cached = False
+            if cached:
+                # A warm build must stay silent
+                continue
+            components.append(component)
+        if not components:
+            return
+        # Single-item waves (a dependency chain discovers one archive per
+        # wave) go through the same runner: one download method, one bar
+        _LOGGER.info(
+            "Downloading %d library archive(s): %s",
+            len(components),
+            ", ".join(c.name for c in components),
+        )
+        failures = run_batch_downloads(
+            "Downloading libraries",
+            [
+                (c.name, c.source.size, partial(_fetch_source, c, salt, namespace))
+                for c in components
+            ],
+        )
+        for name, err in failures:
+            # The sequential call below retries and raises the real error
+            _LOGGER.warning(
+                "Prefetch of %s failed (retrying sequentially): %s",
+                name,
+                failure_reason(err),
+            )
+            _LOGGER.debug("Prefetch failure detail", exc_info=err)
+    except Exception as err:  # noqa: BLE001  # pylint: disable=broad-exception-caught
+        # Same policy as the ESP-IDF twin: the prefetch must never become a
+        # new way for the build to fail
+        _LOGGER.warning("Library prefetch failed: %s", failure_reason(err))
+        _LOGGER.debug("Prefetch failure detail", exc_info=True)
+
+
 def convert_libraries(
     libraries: list[Library], backend: LibraryBackend
 ) -> list[ConvertedLibrary]:
@@ -955,136 +1086,134 @@ def convert_libraries(
     top_level_keys = set(top_level)
     worklist = deque(dict.fromkeys(top_level))
     while worklist:
-        key = worklist.popleft()
-        node = nodes[key]
+        # Drain the frontier sequentially (spec resolution mutates shared
+        # state), then prefetch the wave in parallel
+        wave: list[tuple[str, ConvertedLibrary]] = []
+        while worklist:
+            key = worklist.popleft()
+            node = nodes[key]
 
-        # Re-resolve only when the requirement set grew; requirements
-        # only ever grow, so the fixpoint converges and cycles terminate
-        requirements = frozenset(node.requirements)
-        if resolved_requirements.get(key) == requirements:
-            continue
-        resolved_requirements[key] = requirements
+            # Re-resolve only when the requirement set grew; requirements
+            # only ever grow, so the fixpoint converges and cycles terminate
+            requirements = frozenset(node.requirements)
+            if resolved_requirements.get(key) == requirements:
+                continue
+            resolved_requirements[key] = requirements
 
-        if node.is_git:
-            component = ConvertedLibrary(key, "*", GitSource(node.url, node.ref))
-        elif node.is_local:
-            component = ConvertedLibrary(key, "*", LocalSource(node.local_path))
-        else:
-            owner, name, version, url = _resolve_registry_version(
-                node.owner, node.pkgname, node.requirements
-            )
-            component = ConvertedLibrary(
-                _owner_pkgname_to_name(owner, name), version, URLSource(url)
-            )
-        component.download(salt=salt, namespace=backend.cache_key)
+            if node.is_git:
+                component = ConvertedLibrary(key, "*", GitSource(node.url, node.ref))
+            elif node.is_local:
+                component = ConvertedLibrary(key, "*", LocalSource(node.local_path))
+            else:
+                owner, name, version, url, size = _resolve_registry_version(
+                    node.owner, node.pkgname, node.requirements
+                )
+                component = ConvertedLibrary(
+                    _owner_pkgname_to_name(owner, name), version, URLSource(url, size)
+                )
+            wave.append((key, component))
+        _prefetch_wave(wave, salt, backend.cache_key)
+        for key, component in wave:
+            node = nodes[key]
+            if frozenset(node.requirements) != resolved_requirements[key]:
+                # Requirements grew mid-wave: skip parsing a manifest the
+                # next wave will re-resolve and replace
+                worklist.append(key)
+                continue
+            component.download(salt=salt, namespace=backend.cache_key)
 
-        source_dir = component.source_dir
-        library_json_path = source_dir / "library.json"
-        library_properties_path = source_dir / "library.properties"
-        has_json = library_json_path.is_file()
-        has_properties = library_properties_path.is_file()
-        if not has_json and not has_properties and not node.is_local:
-            # An interrupted clone/extraction self-heals with one forced
-            # re-download; a local source has nothing to re-download
-            _LOGGER.warning(
-                "Library %s at %s is missing library.json and library.properties; "
-                "re-downloading",
-                key,
-                source_dir,
-            )
-            component.download(force=True, salt=salt, namespace=backend.cache_key)
+            source_dir = component.source_dir
+            library_json_path = source_dir / "library.json"
+            library_properties_path = source_dir / "library.properties"
             has_json = library_json_path.is_file()
             has_properties = library_properties_path.is_file()
-        if has_json:
-            component.data = parse_library_json(library_json_path)
-        elif has_properties:
-            component.data = parse_library_properties(library_properties_path)
-        else:
-            # Local sources are user input (EsphomeError); a registry/git
-            # miss means a corrupt cache (RuntimeError)
-            error_cls = EsphomeError if node.is_local else RuntimeError
-            raise error_cls(
-                f"Invalid PIO library {key}: missing library.json and "
-                f"library.properties in {source_dir}"
-            )
-
-        # A bare json.load imposes no shape; every backend dereferences
-        # these fields, so validate once here and name the library
-        malformed = not isinstance(component.data, dict)
-        if not malformed:
-            build = component.data.get("build", {})
-            esphome_data = component.data.get(ESPHOME_DATA_KEY, {})
-            malformed = (
-                not isinstance(build, dict)
-                or not isinstance(esphome_data, dict)
-                or not isinstance(
-                    esphome_data.get(ESPHOME_DATA_LINK_FLAGS_KEY, []), list
+            if not has_json and not has_properties and not node.is_local:
+                # An interrupted clone/extraction self-heals with one forced
+                # re-download; a local source has nothing to re-download
+                _LOGGER.warning(
+                    "Library %s at %s is missing library.json and library.properties; "
+                    "re-downloading",
+                    key,
+                    source_dir,
                 )
-                or not isinstance(build.get("srcDir", ""), str)
-                or not isinstance(build.get("includeDir", ""), str)
-                or not isinstance(build.get("srcFilter", ""), (str, list))
-            )
-        if malformed:
-            # Fail fast only for a library the user asked for; a defect in
-            # an unrequested corner of the graph must not block the build
-            if key in top_level_keys:
-                raise EsphomeError(f"Library {key} has a malformed manifest")
-            _LOGGER.warning("Skipping dependency %s: malformed manifest", key)
-            continue
-        warn_properties_depends(component.name, component.data)
-
-        try:
-            check_library_data(component.data, backend.platform, backend.framework)
-        except InvalidLibrary as e:
-            # An explicitly requested library fails fast; the routine
-            # cross-platform skip stays at debug, other causes warn
-            if key in top_level_keys:
-                reason = (
-                    f"is not compatible with {backend.framework}"
-                    if isinstance(e, IncompatiblePlatform)
-                    else "has a malformed manifest"
-                )
-                raise RuntimeError(f"Requested library {key} {reason}: {e}") from e
-            if isinstance(e, IncompatiblePlatform):
-                _LOGGER.debug("Skip incompatible dependency %s: %s", key, str(e))
+                component.download(force=True, salt=salt, namespace=backend.cache_key)
+                has_json = library_json_path.is_file()
+                has_properties = library_properties_path.is_file()
+            if has_json:
+                component.data = parse_library_json(library_json_path)
+            elif has_properties:
+                component.data = parse_library_properties(library_properties_path)
             else:
-                _LOGGER.warning("Skipping dependency %s: %s", key, str(e))
-            continue
-        components[key] = component
-
-        # Requirements changed (we got past the short-circuit above), so
-        # (re)walk this component's dependencies.
-        node.edges = set()
-        for dependency in normalize_dependencies(
-            component.data.get("dependencies"), component.name
-        ):
-            if "version" not in dependency:
-                # Cannot resolve from the registry; common for bundled
-                # names (Wire, SPI) -- unactionable noise above debug
-                _LOGGER.debug(
-                    "Skip version-less dependency %r of %s",
-                    dependency.get("name"),
-                    component.name,
+                # Local sources are user input (EsphomeError); a registry/git
+                # miss means a corrupt cache (RuntimeError)
+                error_cls = EsphomeError if node.is_local else RuntimeError
+                raise error_cls(
+                    f"Invalid PIO library {key}: missing library.json and "
+                    f"library.properties in {source_dir}"
                 )
+
+            if not _valid_manifest_shape(component.data):
+                # Fail fast only for a library the user asked for; a defect
+                # in an unrequested corner of the graph must not block the
+                # build
+                if key in top_level_keys:
+                    raise EsphomeError(f"Library {key} has a malformed manifest")
+                _LOGGER.warning("Skipping dependency %s: malformed manifest", key)
                 continue
-            if not dependency_is_usable(
-                dependency, backend.platform, backend.framework, component.name
+            warn_properties_depends(component.name, component.data)
+
+            try:
+                check_library_data(component.data, backend.platform, backend.framework)
+            except InvalidLibrary as e:
+                # An explicitly requested library fails fast; the routine
+                # cross-platform skip stays at debug, other causes warn
+                if key in top_level_keys:
+                    reason = (
+                        f"is not compatible with {backend.framework}"
+                        if isinstance(e, IncompatiblePlatform)
+                        else "has a malformed manifest"
+                    )
+                    raise RuntimeError(f"Requested library {key} {reason}: {e}") from e
+                if isinstance(e, IncompatiblePlatform):
+                    _LOGGER.debug("Skip incompatible dependency %s: %s", key, str(e))
+                else:
+                    _LOGGER.warning("Skipping dependency %s: %s", key, str(e))
+                continue
+            components[key] = component
+
+            # Requirements changed (we got past the short-circuit above), so
+            # (re)walk this component's dependencies.
+            node.edges = set()
+            for dependency in normalize_dependencies(
+                component.data.get("dependencies"), component.name
             ):
-                continue
-            dep_name = _owner_pkgname_to_name(
-                dependency.get("owner"), dependency.get("name")
-            )
-            if is_lib_ignored(dep_name, lib_ignore):
-                _LOGGER.debug("Skip ignored dependency %s", dep_name)
-                continue
-            # The version field may actually be a URL (git/archive dependency).
-            dep_version = dependency["version"]
-            dep_url = _url_or_none(dep_version)
-            if dep_url is not None:
-                dep_version = None
-            dep_key = add_spec(dep_name, dep_version, dep_url)
-            node.edges.add(dep_key)
-            worklist.append(dep_key)
+                if "version" not in dependency:
+                    # Cannot resolve from the registry; common for bundled
+                    # names (Wire, SPI) -- unactionable noise above debug
+                    _LOGGER.debug(
+                        "Skip version-less dependency %r of %s",
+                        dependency.get("name"),
+                        component.name,
+                    )
+                    continue
+                if not dependency_is_usable(
+                    dependency, backend.platform, backend.framework, component.name
+                ):
+                    continue
+                dep_name = _owner_pkgname_to_name(
+                    dependency.get("owner"), dependency.get("name")
+                )
+                if is_lib_ignored(dep_name, lib_ignore):
+                    _LOGGER.debug("Skip ignored dependency %s", dep_name)
+                    continue
+                # The version field may actually be a URL (git/archive dependency).
+                dep_version = dependency["version"]
+                dep_url = _url_or_none(dep_version)
+                if dep_url is not None:
+                    dep_version = None
+                dep_key = add_spec(dep_name, dep_version, dep_url)
+                node.edges.add(dep_key)
+                worklist.append(dep_key)
 
     # A git or local source wins over the same component requested from the
     # registry. That's intentional, but warn so the dropped registry spec isn't
diff --git a/tests/unit_tests/test_espidf_component.py b/tests/unit_tests/test_espidf_component.py
index 7b4f848979..3789eefc64 100644
--- a/tests/unit_tests/test_espidf_component.py
+++ b/tests/unit_tests/test_espidf_component.py
@@ -660,7 +660,7 @@ def _patch_registry(monkeypatch, versions):
 
 def test_resolve_registry_version_intersects_constraints(monkeypatch):
     _patch_registry(monkeypatch, ["1.10018.1", "1.10021.0", "1.10021.1"])
-    owner, name, version, url = _resolve_registry_version(
+    owner, name, version, url, _size = _resolve_registry_version(
         "esphome", "libsodium", {"==1.10021.0", "^1.10018.1"}
     )
     assert (owner, name, version) == ("esphome", "libsodium", "1.10021.0")
@@ -669,7 +669,9 @@ def test_resolve_registry_version_intersects_constraints(monkeypatch):
 
 def test_resolve_registry_version_picks_highest_satisfying(monkeypatch):
     _patch_registry(monkeypatch, ["1.0.0", "1.5.0", "2.0.0"])
-    _owner, _name, version, _url = _resolve_registry_version("o", "p", {"^1.0.0"})
+    _owner, _name, version, _url, _size = _resolve_registry_version(
+        "o", "p", {"^1.0.0"}
+    )
     assert version == "1.5.0"
 
 
@@ -719,7 +721,7 @@ def test_generate_idf_components_dedupes_shared_dependency(
         resolve_calls.append(pkgname)
         captured[f"{owner}/{pkgname}"] = set(requirements)
         version = "1.10021.0" if pkgname == "C" else "1.0.0"
-        return owner, pkgname, version, f"http://x/{pkgname}.tar.gz"
+        return owner, pkgname, version, f"http://x/{pkgname}.tar.gz", None
 
     monkeypatch.setattr(
         esphome.platformio.library, "_resolve_registry_version", fake_resolve
@@ -778,7 +780,7 @@ def test_generate_idf_components_lib_ignore_filters_top_level_and_dependencies(
 
     def fake_resolve(owner, pkgname, requirements):
         resolve_calls.append(pkgname)
-        return owner, pkgname, "1.0.0", f"http://x/{pkgname}.tar.gz"
+        return owner, pkgname, "1.0.0", f"http://x/{pkgname}.tar.gz", None
 
     monkeypatch.setattr(
         esphome.platformio.library, "_resolve_registry_version", fake_resolve
@@ -834,6 +836,7 @@ def test_generate_idf_components_handles_dependency_cycle(
             pkgname,
             "1.0.0",
             f"http://x/{pkgname}.tar.gz",
+            None,
         ),
     )
 
@@ -891,6 +894,7 @@ def test_generate_idf_components_git_overrides_registry_warns(
             pkgname,
             "1.0.0",
             f"http://x/{pkgname}.tar.gz",
+            None,
         ),
     )
 
@@ -927,6 +931,7 @@ def test_generate_idf_components_missing_manifest_raises(
             pkgname,
             "1.0.0",
             f"http://x/{pkgname}.tar.gz",
+            None,
         ),
     )
 
@@ -971,6 +976,7 @@ def test_generate_idf_components_warns_on_noncanonical_duplicate(
             pkgname,
             "1.0.0",
             f"http://x/{pkgname}.tar.gz",
+            None,
         ),
     )
 
@@ -1004,6 +1010,7 @@ def test_generate_idf_components_incompatible_top_level_raises(
             pkgname,
             "1.0.0",
             f"http://x/{pkgname}.tar.gz",
+            None,
         ),
     )
 
@@ -1040,6 +1047,7 @@ def test_generate_idf_components_incompatible_dependency_skipped(
             pkgname,
             "1.0.0",
             f"http://x/{pkgname}.tar.gz",
+            None,
         ),
     )
 
diff --git a/tests/unit_tests/test_espidf_framework.py b/tests/unit_tests/test_espidf_framework.py
index d8e7738569..6288933a6a 100644
--- a/tests/unit_tests/test_espidf_framework.py
+++ b/tests/unit_tests/test_espidf_framework.py
@@ -2,6 +2,7 @@
 
 # pylint: disable=protected-access
 
+from concurrent.futures import ThreadPoolExecutor
 from contextlib import contextmanager
 import importlib.util
 import io
@@ -14,7 +15,7 @@ import subprocess
 import sys
 import tarfile
 from types import SimpleNamespace
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
 import pytest
 
@@ -887,6 +888,78 @@ _PREFETCH_JSON = json.dumps(
 )
 
 
+def test_prefetch_leaves_unverifiable_entries_to_the_installer(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """An entry missing sha256 or size must not download unverified; the
+    installer handles it and fails loudly on a bad archive."""
+    entries = json.loads(_PREFETCH_JSON)
+    del entries[0]["sha256"]
+    del entries[1]["size"]
+    entries.append(
+        {
+            "name": "gcc@14.2.0",
+            "url": "https://example.com/gcc.tar.gz",
+            "size": 67,
+            "sha256": "ef" * 32,
+            "dest": "gcc.tar.gz",
+        }
+    )
+    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.framework_helpers._BatchDownloadProgress") as progress_cls,
+    ):
+        _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
+    assert [call[0][0] for call in download.call_args_list] == [
+        "https://example.com/gcc.tar.gz"
+    ]
+    assert download.call_args[1]["sha256"] == "ef" * 32
+    progress_cls.assert_called_once_with("Downloading ESP-IDF tools", 67)
+    assert "cmake@3.30.2 has no sha256/size" in caplog.text
+    assert "ninja@1.12.1 has no sha256/size" in caplog.text
+
+
+def test_prefetch_all_entries_unverifiable_is_a_noop(tmp_path: Path) -> None:
+    entries = json.loads(_PREFETCH_JSON)
+    for entry in entries:
+        del entry["sha256"]
+    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"),
+    ):
+        _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
+    download.assert_not_called()
+
+
+def test_prefetch_dedupes_entries_by_dest(tmp_path: Path) -> None:
+    """Two entries resolving to one dest would interleave writes into the
+    same .part file; only the first downloads."""
+    entries = json.loads(_PREFETCH_JSON)
+    dup = dict(entries[0]) | {"name": "cmake-alias@3.30.2"}
+    entries.append(dup)
+    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.framework_helpers._BatchDownloadProgress"),
+    ):
+        _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
+    dests = [call[0][1].name for call in download.call_args_list]
+    assert dests.count("cmake-3.30.2.tar.gz") == 1
+
+
 def test_prefetch_downloads_each_archive_with_resume(tmp_path: Path) -> None:
     with (
         patch(
@@ -895,16 +968,58 @@ 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.framework_helpers._BatchDownloadProgress") as progress_cls,
     ):
+        # Materialize the lazy mock before threads race its first creation
+        tracker = progress_cls.return_value.tracker.return_value
         _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
 
     dist = get_idf_tools_path() / "dist"
-    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}
+    # 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 via the
+    # cancellation-checked wrapper; verify it delegates to the tracker
+    progress_cls.assert_called_once_with("Downloading ESP-IDF tools", 123 + 45)
+    before = tracker.call_count
+    for kw in calls.values():
+        kw["progress"](7)
+    assert tracker.call_count == before + len(calls)
+
+
+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.framework_helpers.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_skips_already_downloaded_archives(tmp_path: Path) -> None:
@@ -959,11 +1074,12 @@ def test_prefetch_failures_never_raise(
     assert expected_log in caplog.text
 
 
-def test_prefetch_one_failed_archive_does_not_stop_the_rest(
+def test_prefetch_total_failure_logs_error(
     tmp_path: Path, caplog: pytest.LogCaptureFixture
 ) -> None:
-    """A single archive failing its download must not abort the prefetch of
-    the remaining archives."""
+    """Every archive failing is a systematic fault (proxy, bad kwarg), not
+    a flaky mirror; it must be distinguishable at ERROR because the resume
+    workaround is off for the whole install."""
     with (
         patch(
             "esphome.espidf.framework.run_command",
@@ -971,7 +1087,32 @@ def test_prefetch_one_failed_archive_does_not_stop_the_rest(
         ),
         patch(
             "esphome.espidf.framework.download_with_resume",
-            side_effect=[OSError("network down"), None],
+            side_effect=OSError("proxy refuses everything"),
+        ),
+        patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
+    ):
+        _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
+    assert "Every ESP-IDF tool prefetch failed" in caplog.text
+
+
+def test_prefetch_one_failed_archive_does_not_stop_the_rest(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> 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",
+            return_value=(True, _PREFETCH_JSON, ""),
+        ),
+        patch(
+            "esphome.espidf.framework.download_with_resume",
+            side_effect=_fail_cmake_download,
         ) as download,
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
     ):
@@ -979,6 +1120,29 @@ def test_prefetch_one_failed_archive_does_not_stop_the_rest(
 
     assert download.call_count == 2
     assert "Could not prefetch cmake@3.30.2" in caplog.text
+    # One flaky archive is routine, never the systematic-fault ERROR
+    assert "Every ESP-IDF tool prefetch failed" not 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.framework_helpers._BatchDownloadProgress") as progress_cls,
+        patch("esphome.framework_helpers.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:
diff --git a/tests/unit_tests/test_espota2.py b/tests/unit_tests/test_espota2.py
index e0e9185e1c..8867e2c215 100644
--- a/tests/unit_tests/test_espota2.py
+++ b/tests/unit_tests/test_espota2.py
@@ -998,10 +998,10 @@ def test_progress_bar(capsys: CaptureFixture[str]) -> None:
     assert "100%" in captured.err
     assert "Done" in captured.err
 
-    # Test done method
+    # done() after the 100% frame adds nothing; that frame ended its line
     progress.done()
     captured = capsys.readouterr()
-    assert captured.err == "\n"
+    assert captured.err == ""
 
     # Test same progress doesn't update
     progress.update(0.5)
@@ -1010,6 +1010,10 @@ def test_progress_bar(capsys: CaptureFixture[str]) -> None:
     # Should only see one update (second call shouldn't write)
     assert captured.err.count("50%") == 1
 
+    # done() after a mid-way frame ends the line
+    progress.done()
+    assert capsys.readouterr().err == "\n"
+
 
 # Tests for SHA256 authentication
 @pytest.mark.usefixtures("mock_time")
diff --git a/tests/unit_tests/test_framework_helpers.py b/tests/unit_tests/test_framework_helpers.py
index 500705ef67..5916a2fd60 100644
--- a/tests/unit_tests/test_framework_helpers.py
+++ b/tests/unit_tests/test_framework_helpers.py
@@ -12,6 +12,8 @@ from pathlib import Path
 import subprocess
 import sys
 import tarfile
+import threading
+import time
 from unittest.mock import MagicMock, Mock, call, patch
 import zipfile
 
@@ -22,6 +24,7 @@ from esphome import framework_helpers
 from esphome.core import EsphomeError
 from esphome.framework_helpers import (
     _7z_extract_all,
+    _BatchDownloadProgress,
     _detect_archive_root,
     _rename_with_retry,
     _tar_extract_all,
@@ -36,6 +39,7 @@ from esphome.framework_helpers import (
     get_python_env_executable_path,
     get_system_python_path,
     rmdir,
+    run_batch_downloads,
     run_command,
     run_command_ok,
     str_to_lst_of_str,
@@ -1111,6 +1115,218 @@ 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_cls,
+        ):
+            download_with_resume(
+                "https://example.com/t", dest, size=7, progress=seen.append
+            )
+        assert seen == [0, 4, 7, 7]
+        bar_cls.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]
+
+
+def test_run_batch_downloads_ctrl_c_aborts_in_flight_jobs() -> None:
+    """Ctrl-C cancels in-flight downloads at their next tick instead of
+    letting non-daemon workers download to completion."""
+    started = threading.Event()
+    ticks: list[int] = []
+
+    def interrupter(tracker) -> None:
+        started.wait(5)
+        raise KeyboardInterrupt
+
+    def slow_download(tracker) -> None:
+        started.set()
+        for i in range(500):
+            tracker(i)
+            ticks.append(i)
+            time.sleep(0.01)
+
+    t0 = time.monotonic()
+    with pytest.raises(KeyboardInterrupt):
+        run_batch_downloads(
+            "Downloading",
+            [("boom", 0, interrupter), ("slow", 0, slow_download)],
+            max_workers=2,
+        )
+    # Uncancelled, slow_download alone takes ~5s
+    assert time.monotonic() - t0 < 3
+    assert len(ticks) < 500
+
+
+def test_cancellation_escapes_broad_except_in_fetch() -> None:
+    """A fetch that wraps its work in except Exception cannot swallow the
+    Ctrl-C sentinel (it is a BaseException)."""
+    from esphome.framework_helpers import _BatchDownloadCancelled
+
+    started = threading.Event()
+    swallowed = []
+
+    def interrupter(tracker) -> None:
+        started.wait(5)
+        raise KeyboardInterrupt
+
+    def greedy_fetch(tracker) -> None:
+        started.set()
+        try:
+            for i in range(500):
+                tracker(i)
+                time.sleep(0.01)
+        except Exception as err:  # noqa: BLE001  # pylint: disable=broad-exception-caught
+            swallowed.append(err)
+
+    t0 = time.monotonic()
+    with pytest.raises(KeyboardInterrupt):
+        run_batch_downloads(
+            "Downloading",
+            [("boom", 0, interrupter), ("greedy", 0, greedy_fetch)],
+            max_workers=2,
+        )
+    assert time.monotonic() - t0 < 3
+    assert not swallowed
+    assert issubclass(_BatchDownloadCancelled, BaseException)
+    assert not issubclass(_BatchDownloadCancelled, Exception)
+
+
+def test_logging_guard_ends_the_bar_row_before_a_record() -> None:
+    r"""A worker warning gets its own line instead of the bar's \r 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)
+        with progress.logging_guard():
+            logging.getLogger("esphome.test").warning("mirror retry")
+        # The partial 50% frame ended its line before the record was emitted
+        assert stream.getvalue().endswith("50% \n")
+        # And the next tick redraws the frame on a fresh row
+        progress.tracker()(2)
+        assert stream.getvalue().endswith("70% ")
+
+
+def test_logging_guard_without_a_bar_is_a_no_op() -> None:
+    """An unknown total draws no bar; the guard passes records through."""
+    progress = _BatchDownloadProgress("Downloading", 0)
+    with progress.logging_guard():
+        logging.getLogger("esphome.test").warning("plain record")
+
+
+def test_cancellable_sleep_sleeps_between_ticks() -> None:
+    """An uncancelled backoff actually waits out its delay in slices."""
+    from esphome.framework_helpers import _cancellable_sleep
+
+    ticks: list[int] = []
+    t0 = time.monotonic()
+    _cancellable_sleep(0.05, ticks.append, 3)
+    assert time.monotonic() - t0 >= 0.05
+    assert ticks and all(t == 3 for t in ticks)
+
+
+def test_cancellable_sleep_aborts_at_the_tick() -> None:
+    """A backoff sleep observes the cancellation raise promptly."""
+    from esphome.framework_helpers import _BatchDownloadCancelled, _cancellable_sleep
+
+    def cancelled_tick(done: int) -> None:
+        raise _BatchDownloadCancelled
+
+    t0 = time.monotonic()
+    with pytest.raises(_BatchDownloadCancelled):
+        _cancellable_sleep(30, cancelled_tick, 0)
+    assert time.monotonic() - t0 < 1
+
+
+class Test_BatchDownloadProgress:
+    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:
@@ -1123,6 +1339,22 @@ class TestDownloadFromMirrors:
         assert url == "https://example.com/f"
         assert target.read_bytes() == b"filedata"
 
+    def test_file_object_target_reports_progress(self) -> None:
+        """The library prefetch's production path: a file-object target
+        streams through the mirror fallback and ticks the tracker."""
+        buf = io.BytesIO()
+        ticks: list[int] = []
+        with patch(
+            "requests.get",
+            return_value=_mock_response(b"filedata"),
+        ):
+            url = download_from_mirrors(
+                ["https://example.com/f"], {}, buf, progress=ticks.append
+            )
+        assert url == "https://example.com/f"
+        assert buf.getvalue() == b"filedata"
+        assert ticks and ticks[-1] == len(b"filedata")
+
     def test_substitutions_applied_to_url(self, tmp_path: Path) -> None:
         with patch(
             "requests.get",
@@ -1468,6 +1700,49 @@ class TestDownloadFromMirrors:
         assert mock_get.call_count == 2
         mock_sleep.assert_called_once_with(2)
 
+    def test_backoff_tick_reports_filelike_bytes(self) -> None:
+        """For a file-like target the backoff tick carries f.tell(), so the
+        combined bar holds steady through the sweep retry."""
+        target = io.BytesIO()
+        ticks: list[int] = []
+        with (
+            patch(
+                "requests.get",
+                side_effect=[
+                    req.ConnectionError("down"),
+                    _mock_response(b"data"),
+                ],
+            ),
+            patch("esphome.framework_helpers._cancellable_sleep") as mock_sleep,
+        ):
+            download_from_mirrors(
+                ["https://mirror1.com/f"], {}, target, progress=ticks.append
+            )
+        # No bytes had streamed at backoff time, so the tick carries 0
+        assert mock_sleep.call_args == call(2, ticks.append, 0)
+        assert target.getvalue() == b"data"
+
+    def test_backoff_tick_reports_partial_bytes(self, tmp_path: Path) -> None:
+        """The backoff tick carries the bytes already in the part file, so a
+        combined bar holds steady instead of rewinding to zero."""
+        dest = tmp_path / "out.bin"
+        (tmp_path / "out.bin.part").write_bytes(b"12345")
+        ticks: list[int] = []
+        with (
+            patch(
+                "requests.get",
+                side_effect=[
+                    req.ConnectionError("down"),
+                    _mock_response(b"data"),
+                ],
+            ),
+            patch("esphome.framework_helpers._cancellable_sleep") as mock_sleep,
+        ):
+            download_from_mirrors(
+                ["https://mirror1.com/f"], {}, dest, progress=ticks.append
+            )
+        assert mock_sleep.call_args == call(2, ticks.append, 5)
+
     def test_permanent_failure_does_not_retry_sweep(self, tmp_path: Path) -> None:
         """An HTTP 404 will not heal on its own; fail after a single pass."""
         with (
diff --git a/tests/unit_tests/test_helpers.py b/tests/unit_tests/test_helpers.py
index 3160469063..683fef22cf 100644
--- a/tests/unit_tests/test_helpers.py
+++ b/tests/unit_tests/test_helpers.py
@@ -1124,6 +1124,20 @@ def test_progressbar_enabled_on_pipe_with_dashboard(monkeypatch) -> None:
     assert bar.enabled is True
 
 
+def test_progressbar_interrupt_keeps_finished_bar_done(monkeypatch) -> None:
+    """interrupt() on a bar whose 100% frame already ended its own line
+    must not reset it, or the next tick would redraw a second Done row."""
+    stream = MagicMock(spec=io.TextIOWrapper)
+    stream.isatty.return_value = True
+    monkeypatch.setattr(CORE, "dashboard", False)
+
+    bar = ProgressBar("Uploading", stream=stream)
+    bar.update(1)
+    assert bar.last_progress == 100
+    bar.interrupt()
+    assert bar.last_progress == 100
+
+
 @pytest.mark.parametrize(
     ("seconds", "expected"),
     [
diff --git a/tests/unit_tests/test_platformio_library.py b/tests/unit_tests/test_platformio_library.py
index bf8340cea0..792d7dab61 100644
--- a/tests/unit_tests/test_platformio_library.py
+++ b/tests/unit_tests/test_platformio_library.py
@@ -153,13 +153,15 @@ def test_localsource_download_returns_empty_build_dir(setup_core: Path) -> None:
     assert plain != out
 
 
-def test_urlsource_download_extracts_then_reuses_marker(setup_core, monkeypatch):
+def test_urlsource_download_extracts_then_reuses_marker(
+    setup_core, monkeypatch, caplog
+):
     monkeypatch.setattr(lib, "rmdir", lambda path, msg="": None)
     dl_calls: list[list[str]] = []
     monkeypatch.setattr(
         lib,
         "download_from_mirrors",
-        lambda urls, headers, f: dl_calls.append(urls),
+        lambda urls, headers, f, progress=None: dl_calls.append(urls),
     )
 
     def fake_extract(fileobj, path):
@@ -178,6 +180,12 @@ def test_urlsource_download_extracts_then_reuses_marker(setup_core, monkeypatch)
     assert out2 == out
     assert len(dl_calls) == 1
 
+    # A batch caller passes a tracker and owns the messaging; no per-file INFO
+    caplog.set_level("INFO")
+    src.download("mylib-batch", progress=lambda done: None)
+    assert len(dl_calls) == 2
+    assert "Downloading" not in caplog.text
+
 
 def test_resolve_registry_version_raises_without_pkg_file(monkeypatch):
     registry = lib._make_registry_client()
@@ -211,6 +219,7 @@ def _patch_registry_resolve(monkeypatch: pytest.MonkeyPatch) -> None:
             pkgname,
             "1.0.0",
             f"http://x/{pkgname}.tar.gz",
+            None,
         ),
     )
 
@@ -230,6 +239,38 @@ def _patch_download_with_manifests(monkeypatch, tmp_path, manifests, *, properti
     _patch_registry_resolve(monkeypatch)
 
 
+def test_wave_requirement_growth_defers_the_superseded_download(tmp_path, monkeypatch):
+    """A's manifest constrains B while B sits in the same wave: B's
+    drain-time resolution is superseded, so its download defers to the
+    next wave instead of fetching a version that is immediately replaced."""
+    download_names: list[str] = []
+    manifests = {
+        "esphome/A": {
+            "name": "A",
+            "build": {},
+            "dependencies": {"esphome/B": ">=1.0"},
+        },
+        "esphome/B": {"name": "B", "build": {}},
+    }
+
+    def fake_download(self, force=False, salt="", namespace="", progress=None):
+        download_names.append(self.name)
+        self.path = tmp_path / self.get_require_name()
+        self.path.mkdir(parents=True, exist_ok=True)
+        (self.path / "library.json").write_text(json.dumps(manifests[self.name]))
+
+    monkeypatch.setattr(ConvertedLibrary, "download", fake_download)
+    # Hermetic: the stubbed registry reports no size, so no batch prefetch
+    _patch_registry_resolve(monkeypatch)
+    top = convert_libraries(
+        [Library("esphome/A", "1.0.0", None), Library("esphome/B", None, None)],
+        _backend(),
+    )
+    assert sorted(c.name for c in top) == ["esphome/A", "esphome/B"]
+    # B downloads exactly once, after its requirement set stabilized
+    assert download_names.count("esphome/B") == 1
+
+
 def test_convert_libraries_parses_library_properties(tmp_path, monkeypatch):
     # A manifest provided as library.properties (Arduino style) instead of
     # library.json must still be parsed and converted.
@@ -574,6 +615,65 @@ def test_lex_build_flags_dangling_flag_does_not_cross_entries(
     assert "Ignoring trailing '-I'" in caplog.text
 
 
+def test_prefetch_wave_downloads_registry_archives_in_parallel(
+    setup_core, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Registry archives in one wave download concurrently, deduped by URL;
+    git/local sources and failures are left to the sequential call."""
+    calls: list[str] = []
+
+    def fake_download(
+        self, dir_suffix, force=False, salt="", namespace="", progress=None
+    ):
+        calls.append(self.url)
+        if progress is not None:
+            progress(0)
+        if "boom" in self.url:
+            raise RuntimeError("boom")
+
+    monkeypatch.setattr(URLSource, "download", fake_download)
+    wave = [
+        ("a", ConvertedLibrary("a", "1.0", URLSource("https://x/a.tar.gz", 1))),
+        ("b", ConvertedLibrary("b", "1.0", URLSource("https://x/b.tar.gz", 1))),
+        # Duplicate URL must prefetch once (two threads must never extract
+        # into the same cache directory)
+        ("b2", ConvertedLibrary("b2", "1.0", URLSource("https://x/b.tar.gz", 1))),
+        ("c", ConvertedLibrary("c", "1.0", URLSource("https://x/boom.tar.gz", 1))),
+        ("g", ConvertedLibrary("g", "*", lib.GitSource("https://x/g.git", None))),
+    ]
+    lib._prefetch_wave(wave, "", "idf")
+    assert sorted(calls) == [
+        "https://x/a.tar.gz",
+        "https://x/b.tar.gz",
+        "https://x/boom.tar.gz",
+    ]
+    # The failure surfaces at default verbosity, after the bar
+    assert "Prefetch of c failed (retrying sequentially)" in caplog.text
+
+
+def test_prefetch_wave_unknown_size_left_to_sequential(
+    setup_core, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Archives without a registry-reported size skip the batch (their
+    sequential per-file bars don't interleave); the known subset still
+    prefetches."""
+    calls: list[str] = []
+    monkeypatch.setattr(
+        URLSource,
+        "download",
+        lambda self, dir_suffix, force=False, salt="", namespace="", progress=None: (
+            calls.append(self.url)
+        ),
+    )
+    wave = [
+        ("a", ConvertedLibrary("a", "1.0", URLSource("https://x/a.tar.gz", 1))),
+        ("b", ConvertedLibrary("b", "1.0", URLSource("https://x/b.tar.gz", 1))),
+        ("u", ConvertedLibrary("u", "1.0", URLSource("https://x/u.tar.gz"))),
+    ]
+    lib._prefetch_wave(wave, "", "idf")
+    assert sorted(calls) == ["https://x/a.tar.gz", "https://x/b.tar.gz"]
+
+
 def test_join_flag_args_empty_argument_warns_and_drops(
     caplog: pytest.LogCaptureFixture,
 ) -> None:
@@ -582,6 +682,97 @@ def test_join_flag_args_empty_argument_warns_and_drops(
     assert "Ignoring '-D' with empty argument in build_flags" in caplog.text
 
 
+def test_prefetch_wave_cache_probe_failure_still_prefetches(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A filesystem probe failure warns (a systematic one re-downloads
+    everything) but still prefetches; a programming error is NOT swallowed
+    here, it reaches the outer blanket guard."""
+    calls: list[str] = []
+    monkeypatch.setattr(
+        URLSource,
+        "download",
+        lambda self, dir_suffix, **kw: calls.append(self.url),
+    )
+    monkeypatch.setattr(
+        URLSource,
+        "is_cached",
+        lambda self, *a, **kw: (_ for _ in ()).throw(OSError("cache root denied")),
+    )
+    wave = [
+        ("a", ConvertedLibrary("a", "1.0", URLSource("https://x/a.tar.gz", 1))),
+        ("b", ConvertedLibrary("b", "1.0", URLSource("https://x/b.tar.gz", 1))),
+    ]
+    lib._prefetch_wave(wave, "", "idf")
+    assert sorted(calls) == ["https://x/a.tar.gz", "https://x/b.tar.gz"]
+    assert "Cache probe for a failed: cache root denied" in caplog.text
+
+
+def test_prefetch_wave_internal_error_never_fails_the_build(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """The blanket guard keeps a prefetch bug from failing the walk."""
+    monkeypatch.setattr(URLSource, "is_cached", lambda self, *a, **kw: False)
+    monkeypatch.setattr(
+        lib,
+        "run_batch_downloads",
+        lambda *a, **kw: (_ for _ in ()).throw(RuntimeError("bug")),
+    )
+    wave = [
+        ("a", ConvertedLibrary("a", "1.0", URLSource("https://x/a.tar.gz", 1))),
+        ("b", ConvertedLibrary("b", "1.0", URLSource("https://x/b.tar.gz", 1))),
+    ]
+    lib._prefetch_wave(wave, "", "idf")
+    assert "Library prefetch failed: bug" in caplog.text
+
+
+def test_prefetch_wave_warm_cache_is_silent(
+    setup_core, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Already-extracted archives download nothing; a warm build must not
+    print a Downloading line or draw a bar."""
+    monkeypatch.setattr(
+        URLSource,
+        "download",
+        lambda self, dir_suffix, **kw: (_ for _ in ()).throw(
+            AssertionError("downloaded")
+        ),
+    )
+    wave = []
+    for name in ("a", "b", "c"):
+        comp = ConvertedLibrary(name, "1.0", URLSource(f"https://x/{name}.tar.gz", 1))
+        marker_dir = comp.source._cache_dir(comp.get_sanitized_name(), "", "idf")
+        marker_dir.mkdir(parents=True)
+        (marker_dir / ".esphome_extracted").touch()
+        wave.append((name, comp))
+    lib._prefetch_wave(wave, "", "idf")
+    assert "Downloading" not in caplog.text
+
+
+def test_prefetch_wave_single_archive_uses_the_batch(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A dependency chain discovers one archive per wave; it downloads
+    through the same runner so there is one download method and one bar."""
+    caplog.set_level("INFO")
+    calls: list[str] = []
+    monkeypatch.setattr(URLSource, "is_cached", lambda self, *a, **kw: False)
+    monkeypatch.setattr(
+        URLSource,
+        "download",
+        lambda self, dir_suffix, force=False, salt="", namespace="", progress=None: (
+            calls.append(self.url)
+        ),
+    )
+    lib._prefetch_wave(
+        [("a", ConvertedLibrary("a", "1.0", URLSource("https://x/a.tar.gz", 1)))],
+        "",
+        "idf",
+    )
+    assert calls == ["https://x/a.tar.gz"]
+    assert "Downloading 1 library archive(s): a" in caplog.text
+
+
 def test_normalize_dependencies_forms(caplog) -> None:
     """Every PIO-legal spelling normalizes; unrecognizable entries warn."""
     from esphome.platformio.library import normalize_dependencies