diff --git a/esphome/espidf/framework.py b/esphome/espidf/framework.py
index 39c036e6d7..c65b104280 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
@@ -15,7 +16,6 @@ import platformdirs
 
 from esphome.core import CORE, Version
 from esphome.framework_helpers import (
-    BatchDownloadProgress,
     PathType,
     archive_extract_all,
     create_venv,
@@ -692,6 +692,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,
@@ -734,22 +746,23 @@ def _prefetch_idf_tool_archives(
         for entry in json.loads(stdout):
             if (dist_path / entry["dest"]).is_file():
                 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"])
-            # tools.json always carries sha256 and size; an entry missing
-            # either must not be downloaded unverified here, so leave it to
-            # the installer (which fails loudly on a bad archive).
-            if entry.get("sha256") and entry.get("size"):
-                entries.append(entry)
-            else:
+            # 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"],
                 )
+                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(
@@ -758,29 +771,24 @@ def _prefetch_idf_tool_archives(
             ", ".join(entry["name"] for entry in entries),
         )
 
-        # Every entry carries a size (checked above), so the combined bar can
-        # be trusted. Unlike the library prefetch there is no sequential
-        # fallback: per-file bars from several threads would interleave, and
-        # skipping the prefetch would lose the resume workaround for #17703.
-        def _download(entry: dict):
-            return lambda tracker: download_with_resume(
-                entry["url"],
-                dist_path / entry["dest"],
-                sha256=entry["sha256"],
-                size=entry["size"],
-                progress=tracker,
-            )
-
+        # 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(
-            BatchDownloadProgress(
-                "Downloading ESP-IDF tools", sum(entry["size"] for entry in entries)
-            ),
-            [(entry["name"], _download(entry)) for entry in entries],
+            "Downloading ESP-IDF tools",
+            [
+                (
+                    entry["name"],
+                    entry["size"],
+                    partial(_download_tool, dist_path, entry),
+                )
+                for entry in entries
+            ],
         )
         for name, e in failures:
             _LOGGER.warning("Could not prefetch %s: %s", name, e)
+            _LOGGER.debug("Prefetch failure detail", exc_info=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.
diff --git a/esphome/framework_helpers.py b/esphome/framework_helpers.py
index af8854af06..c84b6ce503 100644
--- a/esphome/framework_helpers.py
+++ b/esphome/framework_helpers.py
@@ -712,9 +712,8 @@ 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 bar. With ``progress``
+    set no bar is drawn here; the callback gets the absolute byte count.
     """
     f.seek(offset)
     f.truncate(offset)
@@ -744,25 +743,25 @@ BATCH_DOWNLOAD_WORKERS = 4
 
 
 def run_batch_downloads(
-    progress: "BatchDownloadProgress",
-    jobs: list[tuple[str, Callable[[Callable[[int], None]], None]]],
+    header: str,
+    jobs: list[tuple[str, int, Callable[[Callable[[int], None]], None]]],
     max_workers: int = BATCH_DOWNLOAD_WORKERS,
 ) -> list[tuple[str, Exception]]:
-    """Run download jobs concurrently, reporting into one combined bar.
+    """Run ``(name, size, fetch)`` download jobs concurrently under one bar.
 
-    ``jobs`` holds ``(name, fetch)`` pairs where ``fetch(tracker)`` performs
-    one download reporting absolute byte counts to ``tracker``. Failures are
-    collected (list.append is atomic under the GIL) and returned after the
-    bar is done, so the caller's warnings never land on the bar's row; a
-    failed job credits its tracker 0 so the bar can still complete. Ctrl-C
-    drops queued jobs and aborts in-flight ones at their next progress
-    tick; resumable ``.part`` files keep the bytes already fetched.
+    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 tick; resumable destinations
+    (``download_with_resume``) keep their fetched ``.part`` bytes.
     ``jobs`` must be non-empty.
     """
-    failures: list[tuple[str, Exception]] = []
+    progress = _BatchDownloadProgress(header, sum(size for _, size, _ in jobs))
     cancelled = threading.Event()
 
-    def _run(name: str, fetch: Callable[[Callable[[int], None]], None]) -> None:
+    def _run(
+        name: str, fetch: Callable[[Callable[[int], None]], None]
+    ) -> tuple[str, Exception] | None:
         tracker = progress.tracker()
 
         def checked(done: int) -> None:
@@ -775,13 +774,14 @@ def run_batch_downloads(
         except _BatchDownloadCancelled:
             tracker(0)
         except Exception as err:  # noqa: BLE001  # pylint: disable=broad-exception-caught
-            failures.append((name, err))
             tracker(0)
+            return (name, err)
+        return None
 
-    ex = ThreadPoolExecutor(max_workers=min(max_workers, len(jobs)))
+    ex = ThreadPoolExecutor(max_workers=max_workers)
     try:
-        for future in [ex.submit(_run, name, fetch) for name, fetch in jobs]:
-            future.result()
+        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
@@ -790,22 +790,18 @@ def run_batch_downloads(
     finally:
         ex.shutdown(wait=True, cancel_futures=True)
         progress.done()
-    return failures
 
 
 class _BatchDownloadCancelled(Exception):
     """Raised inside a download job to abandon it after Ctrl-C."""
 
 
-class BatchDownloadProgress:
-    """One progress bar across several concurrent ``download_with_resume`` calls.
+class _BatchDownloadProgress:
+    """One bar across several concurrent downloads, summing tracker bytes.
 
-    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.
+    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:
@@ -829,13 +825,7 @@ class BatchDownloadProgress:
         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
-        ):
+        if self._bar is not None:
             self._bar.done()
 
 
@@ -872,11 +862,8 @@ 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``).
+    ``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.
     """
diff --git a/esphome/helpers.py b/esphome/helpers.py
index 7aa1a9a88c..26e98a561d 100644
--- a/esphome/helpers.py
+++ b/esphome/helpers.py
@@ -729,7 +729,8 @@ 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()
diff --git a/esphome/platformio/extra_script.py b/esphome/platformio/extra_script.py
index 363c40561f..f29cf32951 100644
--- a/esphome/platformio/extra_script.py
+++ b/esphome/platformio/extra_script.py
@@ -30,10 +30,7 @@ def apply_extra_script(
     pio_platform: str,
 ) -> None:
     """Run a library's ``extraScript`` and fold its captured env vars into
-    ``component.data["build"]["flags"]``.
-
-    ``board_mcu`` is a callable so its lookup runs only when a script will.
-    """
+    ``build.flags``; ``board_mcu`` is a callable so it resolves lazily."""
     extra_script = component.data.get("build", {}).get("extraScript")
     if not extra_script:
         return
@@ -177,12 +174,10 @@ def run_extra_script(
     board_mcu: str,
     pio_platform: str,
 ) -> ExtraScriptResult:
-    """Execute ``script_path`` with a fake SCons env and return captured vars.
+    """Execute ``script_path`` with a fake SCons env, ``library_dir`` as CWD.
 
-    Runs with ``library_dir`` as CWD so relative lookups resolve against
-    the library tree. A crashed script warns and returns an empty result,
-    never a partial capture.
-    """
+    A crashed script warns and returns an empty result, never a partial
+    capture."""
     env = _FakeSConsEnv(
         board_mcu=board_mcu,
         pio_env=f"esphome_{board_mcu}",
@@ -248,11 +243,8 @@ def run_extra_script(
 def captured_as_build_flags(
     result: ExtraScriptResult, *, library_dir: Path
 ) -> list[str]:
-    """Translate captured env vars into -L/-l/-D/raw build flags.
-
-    ``LIBPATH`` entries are made relative to ``library_dir`` so the
-    generated build files stay portable.
-    """
+    """Translate captured env vars into -L/-l/-D/raw build flags; path
+    entries anchor to ``library_dir`` so the build files stay portable."""
     flags: list[str] = []
 
     def _strs(bucket: list, kind: str) -> list[str]:
diff --git a/esphome/platformio/library.py b/esphome/platformio/library.py
index 9a1dcda6ea..49753cfc7f 100644
--- a/esphome/platformio/library.py
+++ b/esphome/platformio/library.py
@@ -14,8 +14,8 @@ regardless of which toolchain consumes the result.
 
 from collections import deque
 from collections.abc import Callable, Iterable
-from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass, field
+from functools import partial
 import glob
 import hashlib
 import itertools
@@ -32,8 +32,6 @@ from urllib.request import url2pathname
 from esphome import git
 from esphome.core import CORE, EsphomeError, Library
 from esphome.framework_helpers import (
-    BATCH_DOWNLOAD_WORKERS,
-    BatchDownloadProgress,
     archive_extract_all,
     download_from_mirrors,
     rmdir,
@@ -55,12 +53,9 @@ DEFAULT_BUILD_SRC_FILTER = (
 DEFAULT_BUILD_SRC_DIRS = "src"
 DEFAULT_BUILD_INCLUDE_DIR = "include"
 DEFAULT_BUILD_FLAGS = []
-# Suffix -> compiler kind (PlatformIO's CSUFFIXES/CXXSUFFIXES/ASSUFFIXES).
-# "asm" merges SCons's AS and ASPP sets: all compile as assembler-with-cpp.
-# The kind values drive the ESP8266 native ninja rules (later in this
-# chain); existing backends consume only the keys. Note .C/.C++ join the
-# suffix set here per CXXSUFFIXES; SCons demotes .C to C on
-# case-insensitive filesystems, we always treat it as C++.
+# Suffix -> compiler kind (PlatformIO's CSUFFIXES/CXXSUFFIXES/ASSUFFIXES);
+# "asm" merges SCons's AS and ASPP sets. Per CXXSUFFIXES .C/.C++ are C++
+# here, even where SCons demotes .C on case-insensitive filesystems.
 SOURCE_KIND_FOR_SUFFIX: dict[str, str] = {
     ".c": "c",
     ".cpp": "cxx",
@@ -87,12 +82,7 @@ ESPHOME_DATA_EXTRA_CMAKE_KEY = "EXTRA_CMAKE"
 
 class Source:
     def download(
-        self,
-        dir_suffix: str,
-        force: bool = False,
-        salt: str = "",
-        namespace: str = "",
-        progress: Callable[[int], None] | None = None,
+        self, dir_suffix: str, force: bool = False, salt: str = "", namespace: str = ""
     ) -> Path:
         raise NotImplementedError
 
@@ -106,8 +96,11 @@ 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 _cache_dir(self, dir_suffix: str, salt: str, namespace: str) -> Path:
         # Namespace the cache per backend (e.g. pio_components/idf, .../zephyr) so
@@ -169,12 +162,7 @@ class GitSource(Source):
         self.ref = ref
 
     def download(
-        self,
-        dir_suffix: str,
-        force: bool = False,
-        salt: str = "",
-        namespace: str = "",
-        progress: Callable[[int], None] | None = None,
+        self, dir_suffix: str, force: bool = False, salt: str = "", namespace: str = ""
     ) -> Path:
         domain = DOMAIN
         if namespace:
@@ -209,12 +197,7 @@ class LocalSource(Source):
         self.local_path = path
 
     def download(
-        self,
-        dir_suffix: str,
-        force: bool = False,
-        salt: str = "",
-        namespace: str = "",
-        progress: Callable[[int], None] | None = None,
+        self, dir_suffix: str, force: bool = False, salt: str = "", namespace: str = ""
     ) -> Path:
         src = Path(self.local_path)
         if not src.is_dir():
@@ -252,11 +235,8 @@ class InvalidLibrary(Exception):
 
 
 class IncompatiblePlatform(InvalidLibrary):
-    """The manifest's platform filter rejected the target platform.
-
-    A distinct type so callers can treat the routine cross-platform skip
-    differently from other manifest problems without matching message text.
-    """
+    """The routine cross-platform skip, typed so callers need not match
+    message text."""
 
 
 class ConvertedLibrary:
@@ -307,13 +287,7 @@ class ConvertedLibrary:
     def get_require_name(self):
         return self.get_sanitized_name().replace("/", "__")
 
-    def download(
-        self,
-        force: bool = False,
-        salt: str = "",
-        namespace: str = "",
-        progress: Callable[[int], None] | None = None,
-    ):
+    def download(self, force: bool = False, salt: str = "", namespace: str = ""):
         """Fetch the library into the shared cache and record its ``path``.
 
         The cache directory is named after the sanitized library name; backends
@@ -322,11 +296,7 @@ class ConvertedLibrary:
         ``get_require_name``). ``namespace`` keeps each backend's cache separate.
         """
         self.path = self.source.download(
-            self.get_sanitized_name(),
-            force=force,
-            salt=salt,
-            namespace=namespace,
-            progress=progress,
+            self.get_sanitized_name(), force=force, salt=salt, namespace=namespace
         )
         self.source_path = self.source.source_root(self.path)
 
@@ -584,9 +554,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
@@ -616,7 +587,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]:
@@ -635,9 +606,8 @@ def split_flag_entry(entry: Any, owner: str) -> list[str]:
 def lex_build_flags(entries: str | list[str], owner: str) -> list[str]:
     """Shell-lex ``build.flags`` entries the way PlatformIO's ParseFlags
     does; bare -I/-L/-l/-D tokens re-glue to their argument."""
-    # Join per entry, as SCons's ParseFlags lexes each string independently:
-    # a dangling -I ending one entry must warn, not absorb the next entry's
-    # first token.
+    # Lex per entry as ParseFlags does: a dangling -I must warn, not absorb
+    # the next entry's first token
     return [
         token
         for entry in ensure_list(entries)
@@ -650,13 +620,9 @@ BARE_ARG_FLAGS = frozenset({"-I", "-L", "-l", "-D"})
 
 
 def join_flag_args(tokens: Iterable[str], owner: str) -> list[str]:
-    """Join a bare ``-I``/``-L``/``-l``/``-D`` with its following token,
-    the way PlatformIO's ParseFlags lexes them.
-
-    A trailing or empty argument (``-D ""``) is warned and dropped: the
-    bare flag would make gcc eat the next flag as its argument (or add
-    the CWD for ``-L``); always a typo.
-    """
+    """Join a bare ``-I``/``-L``/``-l``/``-D`` with its following token, as
+    PlatformIO's ParseFlags does. A trailing or empty argument is warned and
+    dropped: the bare flag would make gcc eat the next flag."""
     out: list[str] = []
     it = iter(tokens)
     for tok in it:
@@ -676,11 +642,8 @@ def join_flag_args(tokens: Iterable[str], owner: str) -> list[str]:
 
 
 def warn_properties_depends(name: str, data: object) -> None:
-    """Warn when a manifest declares dependencies only as ``depends=``.
-
-    The dependency walk reads the JSON ``dependencies`` key; the raw
-    ``library.properties`` spelling would otherwise drop silently.
-    """
+    """Warn for ``depends=``-only manifests; the walk reads only the JSON
+    ``dependencies`` key, so they would otherwise drop silently."""
     if isinstance(data, dict) and not data.get("dependencies") and data.get("depends"):
         # INFO: common and unactionable for transitive libraries; a WARNING
         # on every build would train users to ignore the stream
@@ -711,13 +674,9 @@ def dependency_is_usable(
 
 
 def _valid_dependency_entry(entry: dict, manifest_name: str) -> bool:
-    """Whether a normalized entry carries a usable name and version.
-
-    The name must be a non-empty string (every consumer indexes or joins
-    it); a present version must be a string (a container would raise from
-    ``set.add()``, an int fails opaquely inside the registry resolution).
-    Invalid entries warn naming the manifest.
-    """
+    """Whether a normalized entry carries a usable name (non-empty string)
+    and version (string, if present); invalid entries warn naming the
+    manifest."""
     name = entry.get("name")
     if (
         isinstance(name, str)
@@ -914,29 +873,17 @@ def is_lib_ignored(name: str | None, lib_ignore: set[str]) -> bool:
     )
 
 
-def _content_lengths(urls: list[str]) -> list[int | None]:
-    """Content-Length per URL via HEAD requests; None when unknown."""
-    import requests
-
-    from esphome.happy_eyeballs import ensure_happy_eyeballs
-
-    # Same convention as every other network call: without it a broken-IPv6
-    # network burns the full timeout per HEAD before falling back
-    ensure_happy_eyeballs()
-
-    def head(url: str) -> int | None:
-        try:
-            resp = requests.head(url, timeout=10, allow_redirects=True)
-            if not resp.ok:
-                _LOGGER.debug("HEAD %s returned %s", url, resp.status_code)
-                return None
-            return int(resp.headers.get("content-length", 0)) or None
-        except (requests.RequestException, ValueError) as err:
-            _LOGGER.debug("HEAD %s failed: %s", url, err)
-            return None
-
-    with ThreadPoolExecutor(max_workers=min(BATCH_DOWNLOAD_WORKERS, len(urls))) as ex:
-        return list(ex.map(head, urls))
+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(
@@ -944,68 +891,58 @@ def _prefetch_wave(
 ) -> None:
     """Best-effort parallel download of a wave's registry archives.
 
-    The walk's own ``download()`` call stays authoritative (it surfaces real
-    failures, with resume); bars are suppressed since parallel bars would
-    interleave. Duplicate URLs prefetch once so two threads never extract
-    into the same cache directory.
+    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.
     """
-    components: list[ConvertedLibrary] = []
-    seen: set[str] = set()
-    for _key, component in wave:
-        if not isinstance(component.source, URLSource):
-            continue
-        if component.source.url in seen:
-            continue
-        seen.add(component.source.url)
-        try:
-            cached = component.source.is_cached(
-                component.get_sanitized_name(), salt=salt, namespace=namespace
-            )
-        except Exception as err:  # noqa: BLE001  # pylint: disable=broad-exception-caught
-            # Best-effort: a failing probe prefetches (and re-downloads)
-            _LOGGER.debug("Cache probe for %s failed: %s", component.name, err)
-            cached = False
-        if cached:
-            # A completed extraction downloads nothing; a warm build must
-            # stay silent
-            continue
-        components.append(component)
-    if len(components) < 2:
-        return
-    # One combined bar over the batch, sized by HEAD requests. An unknown
-    # size would mean a silent multi-MB download; fall back to sequential
-    # downloads with their per-file bars instead.
-    sizes = _content_lengths([c.source.url for c in components])
-    if not all(sizes):
-        # Announced before the sequential per-file downloads take over, so
-        # the fallback is distinguishable from a hang
+    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 Exception as err:  # noqa: BLE001  # pylint: disable=broad-exception-caught
+                # Best-effort: a failing probe prefetches (and re-downloads)
+                _LOGGER.debug("Cache probe for %s failed: %s", component.name, err)
+                cached = False
+            if cached:
+                # A warm build must stay silent
+                continue
+            components.append(component)
+        if len(components) < 2:
+            return
         _LOGGER.info(
-            "No Content-Length for %s; downloading sequentially",
-            ", ".join(
-                c.source.url
-                for c, size in zip(components, sizes, strict=True)
-                if not size
-            ),
+            "Downloading %d libraries: %s",
+            len(components),
+            ", ".join(c.name for c in components),
         )
-        return
-    _LOGGER.info(
-        "Downloading %d libraries: %s",
-        len(components),
-        ", ".join(c.name for c in components),
-    )
-
-    def _fetch(component: ConvertedLibrary):
-        return lambda tracker: component.download(
-            salt=salt, namespace=namespace, progress=tracker
+        failures = run_batch_downloads(
+            "Downloading libraries",
+            [
+                (c.name, c.source.size, partial(_fetch_source, c, salt, namespace))
+                for c in components
+            ],
         )
-
-    failures = run_batch_downloads(
-        BatchDownloadProgress("Downloading libraries", sum(sizes)),
-        [(component.name, _fetch(component)) for component 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, err)
+        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, 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", err)
+        _LOGGER.debug("Prefetch failure detail", exc_info=True)
 
 
 def convert_libraries(
@@ -1105,8 +1042,7 @@ def convert_libraries(
     worklist = deque(dict.fromkeys(top_level))
     while worklist:
         # Drain the frontier sequentially (spec resolution mutates shared
-        # node state), then prefetch the wave's registry archives in
-        # parallel; the per-component download() below stays authoritative.
+        # state), then prefetch the wave in parallel
         wave: list[tuple[str, ConvertedLibrary]] = []
         while worklist:
             key = worklist.popleft()
@@ -1124,20 +1060,19 @@ def convert_libraries(
             elif node.is_local:
                 component = ConvertedLibrary(key, "*", LocalSource(node.local_path))
             else:
-                owner, name, version, url = _resolve_registry_version(
+                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)
+                    _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]:
-                # An earlier wave entry grew this node's requirements after
-                # the drain resolved it; skip parsing and walking a manifest
-                # the next wave will replace (its archive is already fetched)
+                # 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)
diff --git a/tests/unit_tests/test_espidf_component.py b/tests/unit_tests/test_espidf_component.py
index c5f889a37a..a504cae8dd 100644
--- a/tests/unit_tests/test_espidf_component.py
+++ b/tests/unit_tests/test_espidf_component.py
@@ -614,7 +614,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")
@@ -623,7 +623,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"
 
 
@@ -673,7 +675,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
@@ -732,7 +734,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
@@ -788,6 +790,7 @@ def test_generate_idf_components_handles_dependency_cycle(
             pkgname,
             "1.0.0",
             f"http://x/{pkgname}.tar.gz",
+            None,
         ),
     )
 
@@ -845,6 +848,7 @@ def test_generate_idf_components_git_overrides_registry_warns(
             pkgname,
             "1.0.0",
             f"http://x/{pkgname}.tar.gz",
+            None,
         ),
     )
 
@@ -881,6 +885,7 @@ def test_generate_idf_components_missing_manifest_raises(
             pkgname,
             "1.0.0",
             f"http://x/{pkgname}.tar.gz",
+            None,
         ),
     )
 
@@ -925,6 +930,7 @@ def test_generate_idf_components_warns_on_noncanonical_duplicate(
             pkgname,
             "1.0.0",
             f"http://x/{pkgname}.tar.gz",
+            None,
         ),
     )
 
@@ -958,6 +964,7 @@ def test_generate_idf_components_incompatible_top_level_raises(
             pkgname,
             "1.0.0",
             f"http://x/{pkgname}.tar.gz",
+            None,
         ),
     )
 
@@ -994,6 +1001,7 @@ def test_generate_idf_components_incompatible_dependency_skipped(
             pkgname,
             "1.0.0",
             f"http://x/{pkgname}.tar.gz",
+            None,
         ),
     )
 
@@ -1066,7 +1074,7 @@ def test_idf_component_download_passes_salt() -> None:
     c.download(force=True, salt="abcd1234", namespace="idf")
 
     source.download.assert_called_once_with(
-        "owner/name", force=True, salt="abcd1234", namespace="idf", progress=None
+        "owner/name", force=True, salt="abcd1234", namespace="idf"
     )
     assert c.path == Path("/converted/owner/name")
 
diff --git a/tests/unit_tests/test_espidf_framework.py b/tests/unit_tests/test_espidf_framework.py
index 21073b1d8e..5c91f6e4e9 100644
--- a/tests/unit_tests/test_espidf_framework.py
+++ b/tests/unit_tests/test_espidf_framework.py
@@ -912,7 +912,7 @@ def test_prefetch_leaves_unverifiable_entries_to_the_installer(
         ),
         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,
+        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] == [
@@ -953,7 +953,7 @@ def test_prefetch_dedupes_entries_by_dest(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"),
+        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]
@@ -968,7 +968,7 @@ 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,
+        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
@@ -1022,25 +1022,6 @@ def test_prefetch_downloads_archives_concurrently(tmp_path: Path) -> None:
     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.framework_helpers.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
-
-
 def test_prefetch_skips_already_downloaded_archives(tmp_path: Path) -> None:
     dist = get_idf_tools_path() / "dist"
     dist.mkdir(parents=True)
@@ -1130,10 +1111,8 @@ def test_prefetch_finishes_progress_bar_and_cancels_queue(tmp_path: Path) -> Non
         ),
         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.framework_helpers.ThreadPoolExecutor", wraps=ThreadPoolExecutor
-        ) as pool_cls,
+        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
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 5bbc3f3444..772257ce14 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
 
@@ -21,8 +23,8 @@ import requests as req
 from esphome import framework_helpers
 from esphome.core import EsphomeError
 from esphome.framework_helpers import (
-    BatchDownloadProgress,
     _7z_extract_all,
+    _BatchDownloadProgress,
     _detect_archive_root,
     _is_transient_download_error,
     _rename_with_retry,
@@ -38,6 +40,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,
@@ -1161,11 +1164,6 @@ class TestDownloadWithResume:
 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."""
-    import threading
-    import time
-
-    from esphome.framework_helpers import BatchDownloadProgress, run_batch_downloads
-
     started = threading.Event()
     ticks: list[int] = []
 
@@ -1183,8 +1181,8 @@ def test_run_batch_downloads_ctrl_c_aborts_in_flight_jobs() -> None:
     t0 = time.monotonic()
     with pytest.raises(KeyboardInterrupt):
         run_batch_downloads(
-            BatchDownloadProgress("Downloading", 0),
-            [("boom", interrupter), ("slow", slow_download)],
+            "Downloading",
+            [("boom", 0, interrupter), ("slow", 0, slow_download)],
             max_workers=2,
         )
     # Uncancelled, slow_download alone takes ~5s
@@ -1192,10 +1190,10 @@ def test_run_batch_downloads_ctrl_c_aborts_in_flight_jobs() -> None:
     assert len(ticks) < 500
 
 
-class TestBatchDownloadProgress:
+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)
+            progress = _BatchDownloadProgress("Downloading", 100)
             a = progress.tracker()
             b = progress.tracker()
             a(10)
@@ -1209,13 +1207,13 @@ class TestBatchDownloadProgress:
     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 = _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 = _BatchDownloadProgress("Downloading", 0)
             progress.tracker()(5)
             progress.done()
         bar_cls.assert_not_called()
@@ -1226,7 +1224,7 @@ class TestBatchDownloadProgress:
         stream = io.StringIO()
         stream.isatty = lambda: True  # type: ignore[method-assign]
         with patch("esphome.helpers.sys.stderr", stream):
-            progress = BatchDownloadProgress("Downloading", 10)
+            progress = _BatchDownloadProgress("Downloading", 10)
             progress.tracker()(5)
             progress.done()
         assert stream.getvalue().endswith("50% \n")
@@ -1237,14 +1235,14 @@ class TestBatchDownloadProgress:
         stream = io.StringIO()
         stream.isatty = lambda: True  # type: ignore[method-assign]
         with patch("esphome.helpers.sys.stderr", stream):
-            BatchDownloadProgress("Downloading", 10).done()
+            _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 = _BatchDownloadProgress("Downloading", 10)
             progress.tracker()(10)
             progress.done()
         assert stream.getvalue().endswith("100% Done...\r\n")
@@ -1261,6 +1259,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",
diff --git a/tests/unit_tests/test_platformio_library.py b/tests/unit_tests/test_platformio_library.py
index b2374a74d7..269031bb1e 100644
--- a/tests/unit_tests/test_platformio_library.py
+++ b/tests/unit_tests/test_platformio_library.py
@@ -8,7 +8,6 @@ from contextlib import contextmanager
 import json
 import logging
 from pathlib import Path
-from types import SimpleNamespace
 
 import pytest
 
@@ -236,6 +235,7 @@ def _patch_registry_resolve(monkeypatch: pytest.MonkeyPatch) -> None:
             pkgname,
             "1.0.0",
             f"http://x/{pkgname}.tar.gz",
+            None,
         ),
     )
 
@@ -276,8 +276,7 @@ def test_wave_requirement_growth_defers_the_superseded_download(tmp_path, monkey
         (self.path / "library.json").write_text(json.dumps(manifests[self.name]))
 
     monkeypatch.setattr(ConvertedLibrary, "download", fake_download)
-    # Hermetic: unknown sizes take the sequential path instead of real HEADs
-    monkeypatch.setattr(lib, "_content_lengths", lambda urls: [None] * len(urls))
+    # 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)],
@@ -640,22 +639,23 @@ def test_prefetch_wave_downloads_registry_archives_in_parallel(
     git/local sources and failures are left to the sequential call."""
     calls: list[str] = []
 
-    def fake_download(self, force=False, salt="", namespace="", progress=None):
-        calls.append(self.source.url)
+    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.source.url:
+        if "boom" in self.url:
             raise RuntimeError("boom")
 
-    monkeypatch.setattr(ConvertedLibrary, "download", fake_download)
-    monkeypatch.setattr(lib, "_content_lengths", lambda urls: [1] * len(urls))
+    monkeypatch.setattr(URLSource, "download", fake_download)
     wave = [
-        ("a", ConvertedLibrary("a", "1.0", URLSource("https://x/a.tar.gz"))),
-        ("b", ConvertedLibrary("b", "1.0", URLSource("https://x/b.tar.gz"))),
+        ("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"))),
-        ("c", ConvertedLibrary("c", "1.0", URLSource("https://x/boom.tar.gz"))),
+        ("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")
@@ -668,25 +668,27 @@ def test_prefetch_wave_downloads_registry_archives_in_parallel(
     assert "Prefetch of c failed (retrying sequentially)" in caplog.text
 
 
-def test_prefetch_wave_unknown_size_falls_back_to_sequential(
-    setup_core, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+def test_prefetch_wave_unknown_size_left_to_sequential(
+    setup_core, monkeypatch: pytest.MonkeyPatch
 ) -> None:
-    """Any unknown HEAD size skips the parallel prefetch entirely so the
-    sequential downloads keep their per-file bars."""
-
-    def fail_download(self, force=False, salt="", namespace="", progress=None):
-        raise AssertionError("prefetched despite unknown size")
-
-    monkeypatch.setattr(ConvertedLibrary, "download", fail_download)
-    monkeypatch.setattr(lib, "_content_lengths", lambda urls: [1, 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"))),
-        ("b", ConvertedLibrary("b", "1.0", URLSource("https://x/b.tar.gz"))),
+        ("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"))),
     ]
-    caplog.set_level("DEBUG")
     lib._prefetch_wave(wave, "", "idf")
-    # The culprit URL is named so the fallback is traceable
-    assert "No Content-Length for https://x/b.tar.gz" in caplog.text
+    assert sorted(calls) == ["https://x/a.tar.gz", "https://x/b.tar.gz"]
 
 
 def test_join_flag_args_empty_argument_warns_and_drops(
@@ -697,54 +699,24 @@ def test_join_flag_args_empty_argument_warns_and_drops(
     assert "Ignoring '-D' with empty argument in build_flags" in caplog.text
 
 
-def test_content_lengths_head_requests(monkeypatch: pytest.MonkeyPatch) -> None:
-    """Sizes come from HEAD Content-Length; a failing HEAD reads as 0 so
-    the combined bar is skipped rather than wrong."""
-    import requests
-
-    def fake_head(url, timeout, allow_redirects):
-        if "bad" in url:
-            raise requests.ConnectionError("down")
-        if "gone" in url:
-            return SimpleNamespace(ok=False, status_code=404, headers={})
-        if "garbage" in url:
-            # A proxy/CDN doubling the header ("123, 123") or emitting junk
-            # must degrade to unknown, not ValueError the build
-            return SimpleNamespace(ok=True, headers={"content-length": "123, 123"})
-        return SimpleNamespace(ok=True, headers={"content-length": "123"})
-
-    monkeypatch.setattr(requests, "head", fake_head)
-    # None marks an unknown size (probe failure or non-2xx), distinct
-    # from a genuine zero
-    assert lib._content_lengths(
-        ["https://x/a", "https://x/bad", "https://x/gone", "https://x/garbage"]
-    ) == [
-        123,
-        None,
-        None,
-        None,
-    ]
-
-
 def test_prefetch_wave_cache_probe_failure_still_prefetches(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     """The cache probe is best-effort; a failing probe prefetches anyway."""
     calls: list[str] = []
     monkeypatch.setattr(
-        ConvertedLibrary,
+        URLSource,
         "download",
-        lambda self, **kw: calls.append(self.source.url),
+        lambda self, dir_suffix, **kw: calls.append(self.url),
     )
     monkeypatch.setattr(
         URLSource,
         "is_cached",
         lambda self, *a, **kw: (_ for _ in ()).throw(RuntimeError("no core")),
     )
-    monkeypatch.setattr(lib, "_content_lengths", lambda urls: [1] * len(urls))
     wave = [
-        ("a", ConvertedLibrary("a", "1.0", URLSource("https://x/a.tar.gz"))),
-        ("b", ConvertedLibrary("b", "1.0", URLSource("https://x/b.tar.gz"))),
+        ("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"]
@@ -756,13 +728,15 @@ def test_prefetch_wave_warm_cache_is_silent(
     """Already-extracted archives download nothing; a warm build must not
     print a Downloading line or draw a bar."""
     monkeypatch.setattr(
-        ConvertedLibrary,
+        URLSource,
         "download",
-        lambda self, **kw: (_ for _ in ()).throw(AssertionError("downloaded")),
+        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"))
+        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()
@@ -777,12 +751,14 @@ def test_prefetch_wave_single_archive_skips_the_pool(
     """One archive gains nothing from a pool; the sequential call keeps its
     progress bar."""
     monkeypatch.setattr(
-        ConvertedLibrary,
+        URLSource,
         "download",
-        lambda self, **kw: (_ for _ in ()).throw(AssertionError("prefetched")),
+        lambda self, dir_suffix, **kw: (_ for _ in ()).throw(
+            AssertionError("prefetched")
+        ),
     )
     lib._prefetch_wave(
-        [("a", ConvertedLibrary("a", "1.0", URLSource("https://x/a.tar.gz")))],
+        [("a", ConvertedLibrary("a", "1.0", URLSource("https://x/a.tar.gz", 1)))],
         "",
         "idf",
     )