Merge branch 'esp8266-native-parallel-extract' into espidf-parallel-tool-extract

esphome/arduino8266/framework.py

@@ -23,7 +23,7 @@ from esphome.build_helpers.pch import ccache_pch_env
 from esphome.build_helpers.tools_cache import ARDUINO8266_TOOLS_CACHE, tools_cache_path
 from esphome.core import EsphomeError, Version
 from esphome.framework_helpers import str_to_lst_of_str
-from esphome.platformio.registry import install_package, prefetch_packages
+from esphome.platformio.registry import install_packages, prefetch_packages
 
 FRAMEWORK_PACKAGE = "framework-arduinoespressif8266"
 TOOLCHAIN_PACKAGE = "toolchain-xtensa"
@@ -116,10 +116,9 @@ def check_and_install(framework_version: Version) -> InstalledPaths:
             ("bin", "xtensa-lx106-elf"),
         ),
     )
-    # Fetch both archives at once; the installs below verify and extract
+    # Fetch both archives at once; the install verifies and extracts them
     prefetch_packages([spec[:4] for spec in specs], downloads_dir)
-    for name, version, dest, mirrors, expect in specs:
-        install_package(name, version, dest, mirrors, downloads_dir, expect=expect)
+    install_packages(specs, downloads_dir)
     return InstalledPaths(
         framework=framework_path, toolchain=toolchain_path, ninja=ninja_path
     )

esphome/espidf/_tool_resolution.py

@@ -0,0 +1,68 @@
+"""Shared tool resolution for the sibling idf_tools-backed scripts.
+
+Importable because ``python <script>`` puts this directory first on
+sys.path; ``idf_tools`` itself comes from PYTHONPATH.
+"""
+
+# pylint: disable=import-error  # idf_tools is on PYTHONPATH at runtime only
+
+from collections.abc import Callable, Iterator
+import os
+from pathlib import Path
+
+from idf_tools import (
+    CURRENT_PLATFORM,
+    TOOLS_FILE,
+    IDFEnv,
+    ToolBinaryError,
+    add_and_check_targets,
+    expand_tools_arg,
+    g,
+    load_tools_info,
+)
+
+
+def init_idf_tools(idf_framework_root: str) -> None:
+    """Point idf_tools' globals at the framework and IDF_TOOLS_PATH."""
+    g.idf_path = idf_framework_root
+    g.idf_tools_path = os.environ.get("IDF_TOOLS_PATH")
+    g.tools_json = str(Path(g.idf_path) / TOOLS_FILE)
+
+
+def archive_name(download: object) -> str:
+    """The dist/ filename idf_tools downloads and installs this from."""
+    return download.rename_dist or Path(download.url).name
+
+
+def iter_tool_downloads(
+    targets_csv: str,
+    tool_specs: list[str],
+    on_broken: Callable[[str, ToolBinaryError], bool],
+) -> Iterator[tuple[object, str, str, object]]:
+    """Yield (tool, name, version, download) per uninstalled tool, mirroring
+    ``idf_tools.py install``'s expansion; ``on_broken(name, err)`` returns
+    True to treat a tool with a failing installed binary as not installed."""
+    targets = add_and_check_targets(IDFEnv.get_idf_env(), targets_csv)
+    tools_info = load_tools_info()
+    for name in expand_tools_arg(tool_specs, tools_info, targets):
+        if "@" in name:
+            name, version = name.split("@", 1)
+        else:
+            version = None
+        tool = tools_info.get(name)
+        if tool is None or not tool.compatible_with_platform():
+            continue
+        version = version or tool.get_recommended_version()
+        if version is None:
+            continue
+        try:
+            tool.find_installed_versions()
+        except ToolBinaryError as e:
+            if not on_broken(name, e):
+                continue
+        if version in tool.versions_installed or version not in tool.versions:
+            continue
+        download = tool.versions[version].get_download_for_platform(CURRENT_PLATFORM)
+        if download is None:
+            continue
+        yield tool, name, version, download

esphome/espidf/framework.py

@@ -20,6 +20,7 @@ from esphome.build_helpers.pch import ccache_pch_env
 from esphome.build_helpers.tools_cache import IDF_TOOLS_CACHE, tools_cache_path
 from esphome.core import Version
 from esphome.framework_helpers import (
+    BATCH_EXTRACT_WORKERS,
     PathType,
     create_venv,
     download_and_extract,
@@ -27,6 +28,7 @@ from esphome.framework_helpers import (
     failure_reason,
     get_python_env_executable_path,
     get_system_python_path,
+    is_expected_fetch_error,
     resume_fetch_job,
     rmdir,
     run_batch_downloads,
@@ -34,9 +36,9 @@ from esphome.framework_helpers import (
     run_command_ok,
     str_to_lst_of_str,
     tool_version_runs,
-    warn_prefetch_failures,
+    warn_batch_failures,
 )
-from esphome.helpers import write_file_if_changed
+from esphome.helpers import get_usable_cpu_count, write_file_if_changed
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -292,11 +294,13 @@ def _run_idf_tools_script(
     msg: str,
     args: list[str] | None = None,
     env: dict[str, str] | None = None,
+    stream_output: bool = False,
 ) -> tuple[bool, str | None, str | None]:
     """Run one of the sibling idf_tools-backed helper scripts.
 
-    The script is executed with the framework's ``tools`` directory on
-    PYTHONPATH so it imports the framework's own ``idf_tools`` module.
+    PYTHONPATH carries this directory (sibling imports like
+    ``_tool_resolution``), the esphome package root (``esphome.helpers``),
+    and the framework's ``tools`` dir (its own ``idf_tools`` module).
     """
     cmd = [
         get_system_python_path(),
@@ -304,11 +308,20 @@ def _run_idf_tools_script(
         str(idf_framework_root),
         *(args or []),
     ]
+    # Explicit paths: the scripts dir (sibling imports must survive
+    # PYTHONSAFEPATH), the esphome package root, and the framework's idf_tools
+    pythonpath = os.pathsep.join(
+        (
+            str(_SCRIPTS_DIR),
+            str(_SCRIPTS_DIR.parents[1]),
+            str(Path(idf_framework_root) / "tools"),
+        )
+    )
     return run_command(
         cmd,
         msg=msg,
-        env=(env or os.environ)
-        | {"PYTHONPATH": str(Path(idf_framework_root) / "tools")},
+        env=(env or os.environ) | {"PYTHONPATH": pythonpath},
+        stream_output=stream_output,
     )
 
 
@@ -726,9 +739,9 @@ def _prefetch_idf_tool_archives(
         dist_path = get_idf_tools_path() / "dist"
         entries = []
         seen_dests: set[str] = set()
+        # Pre-existing archives are not skipped: download_with_resume keeps
+        # them only on a sha256 match, so the pre-extraction can trust dist/
         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
@@ -748,9 +761,11 @@ def _prefetch_idf_tool_archives(
             entries.append(entry)
         if not entries:
             return
+        cached = sum((dist_path / entry["dest"]).is_file() for entry in entries)
         _LOGGER.info(
-            "Downloading %d ESP-IDF tool archive(s): %s",
+            "Downloading %d ESP-IDF tool archive(s)%s: %s",
             len(entries),
+            f" ({cached} cached, verifying)" if cached else "",
             ", ".join(entry["name"] for entry in entries),
         )
 
@@ -774,7 +789,7 @@ def _prefetch_idf_tool_archives(
                 for entry in entries
             ],
         )
-        warn_prefetch_failures(failures)
+        warn_batch_failures(failures)
         if len(failures) == len(entries):
             # A systematic fault, not one flaky mirror: the resume
             # workaround (#17703) is off for this whole install
@@ -789,6 +804,42 @@ def _prefetch_idf_tool_archives(
         _LOGGER.debug("Prefetch failure detail", exc_info=True)
 
 
+def _preinstall_idf_tool_archives(
+    framework_path: Path,
+    targets_str: str,
+    tools: list[str],
+    env: dict[str, str] | None,
+) -> None:
+    """Run install_tool_archives.py to extract the prefetched tool archives
+    in parallel. Strictly best-effort: the sequential installer remains the
+    authority (see that script's docstring)."""
+    try:
+        success, _stdout, _stderr = _run_idf_tools_script(
+            framework_path,
+            "install_tool_archives.py",
+            "ESP-IDF tool archive extraction",
+            args=[
+                targets_str,
+                str(min(get_usable_cpu_count(), BATCH_EXTRACT_WORKERS)),
+                *tools,
+            ],
+            env=env,
+            stream_output=True,
+        )
+        if not success:
+            # Detail already streamed to the terminal by the script; a
+            # surviving torn dir prints its own guidance there
+            _LOGGER.warning("ESP-IDF tool pre-extraction failed; see above")
+    except Exception as e:  # noqa: BLE001  # pylint: disable=broad-exception-caught
+        # A programming error keeps its traceback at WARNING
+        _LOGGER.warning(
+            "ESP-IDF tool pre-extraction failed: %s",
+            failure_reason(e),
+            exc_info=None if is_expected_fetch_error(e) else e,
+        )
+        _LOGGER.debug("Pre-extraction failure detail", exc_info=True)
+
+
 def _check_esphome_idf_framework_install(
     version: str,
     targets: list[str],
@@ -940,6 +991,7 @@ def _check_esphome_idf_framework_install(
         _LOGGER.info("Installing ESP-IDF %s framework ...", version)
         targets_str = ",".join(targets)
         _prefetch_idf_tool_archives(framework_path, targets_str, tools, env)
+        _preinstall_idf_tool_archives(framework_path, targets_str, tools, env)
         cmd = [
             get_system_python_path(),
             str(idf_tools_path),

esphome/espidf/get_tool_downloads.py

@@ -1,8 +1,9 @@
 """Print JSON download info for the ESP-IDF tools an install would fetch.
 
 Run via ``python <this file> <idf_framework_root> <targets-csv> <tool-spec>...``.
-PYTHONPATH must include ``<idf_framework_root>/tools`` so ``idf_tools`` is
-importable, and IDF_TOOLS_PATH must be set. Prints a JSON list of
+PYTHONPATH must include this directory (for ``_tool_resolution``) and
+``<idf_framework_root>/tools`` (for ``idf_tools``), and IDF_TOOLS_PATH must
+be set. Prints a JSON list of
 ``{name, url, size, sha256, dest}`` for every tool version that is not yet
 installed, where ``dest`` is the archive filename ``idf_tools.py install``
 expects to find in ``<IDF_TOOLS_PATH>/dist``. Tools with no download for the
@@ -18,67 +19,36 @@ or written — this script only reports what the install would download.
 
 from contextlib import redirect_stdout
 import json
-import os
-from pathlib import Path
 import sys
 
-from idf_tools import (
-    CURRENT_PLATFORM,
-    TOOLS_FILE,
-    IDFEnv,
-    ToolBinaryError,
-    add_and_check_targets,
-    expand_tools_arg,
-    g,
-    get_idf_download_url_apply_mirrors,
-    load_tools_info,
-)
+from _tool_resolution import archive_name, init_idf_tools, iter_tool_downloads
+from idf_tools import ToolBinaryError, get_idf_download_url_apply_mirrors
 
 
 def collect_downloads() -> list[dict]:
-    g.idf_path = sys.argv[1]
-    g.idf_tools_path = os.environ.get("IDF_TOOLS_PATH")
-    g.tools_json = str(Path(g.idf_path) / TOOLS_FILE)
+    init_idf_tools(sys.argv[1])
 
-    targets = add_and_check_targets(IDFEnv.get_idf_env(), sys.argv[2])
-    tools_info = load_tools_info()
-    downloads: list[dict] = []
+    def on_broken(name: str, e: ToolBinaryError) -> bool:
+        # A broken installed binary is idf_tools' problem to repair on
+        # install; note it and treat the version as not installed.
+        print(f"tool {name} failed its binary check: {e}", file=sys.stderr)
+        return True
 
-    for name in expand_tools_arg(sys.argv[3:], tools_info, targets):
-        if "@" in name:
-            name, version = name.split("@", 1)
-        else:
-            version = None
-        tool = tools_info.get(name)
-        if tool is None or not tool.compatible_with_platform():
-            continue
-        version = version or tool.get_recommended_version()
-        if version is None:
-            continue
-        try:
-            tool.find_installed_versions()
-        except ToolBinaryError as e:
-            # A broken installed binary is idf_tools' problem to repair on
-            # install; note it and treat the version as not installed.
-            print(f"tool {name} failed its binary check: {e}", file=sys.stderr)
-        if version in tool.versions_installed or version not in tool.versions:
-            continue
-        download = tool.versions[version].get_download_for_platform(CURRENT_PLATFORM)
-        if download is None:
-            continue
-        downloads.append(
-            {
-                "name": f"{name}@{version}",
-                # Apply the same IDF_MIRROR_PREFIX_MAP / IDF_GITHUB_ASSETS
-                # rewriting the installer's own downloader applies, so users
-                # behind a mirror prefetch from the mirror too.
-                "url": get_idf_download_url_apply_mirrors(None, download.url),
-                "size": download.size,
-                "sha256": download.sha256,
-                "dest": download.rename_dist or Path(download.url).name,
-            }
+    return [
+        {
+            "name": f"{name}@{version}",
+            # Apply the same IDF_MIRROR_PREFIX_MAP / IDF_GITHUB_ASSETS
+            # rewriting the installer's own downloader applies, so users
+            # behind a mirror prefetch from the mirror too.
+            "url": get_idf_download_url_apply_mirrors(None, download.url),
+            "size": download.size,
+            "sha256": download.sha256,
+            "dest": archive_name(download),
+        }
+        for _tool, name, version, download in iter_tool_downloads(
+            sys.argv[2], sys.argv[3:], on_broken
         )
-    return downloads
+    ]
 
 
 # idf_tools prints informational lines (e.g. mirror URL rewrites) to stdout;

esphome/espidf/install_tool_archives.py

@@ -0,0 +1,129 @@
+"""Extract prefetched ESP-IDF tool archives in parallel.
+
+Run via ``python <this file> <idf_framework_root> <targets-csv> <workers>
+<tool-spec>...`` with idf_tools and the esphome package root on PYTHONPATH
+and IDF_TOOLS_PATH set.
+Drives idf_tools' own ``IDFTool.install()`` so extraction semantics match
+the sequential installer, which still runs afterwards as the authority and
+redoes anything this best-effort pass failed on. Archives are trusted from
+the prefetch's sha256 verification, not re-hashed here.
+"""
+
+# pylint: disable=import-error  # idf_tools is on PYTHONPATH at runtime only
+
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+import sys
+
+from _tool_resolution import archive_name, init_idf_tools, iter_tool_downloads
+from idf_tools import ToolBinaryError, g
+
+from esphome.helpers import rmtree
+
+
+def collect_pending(
+    targets_csv: str, tool_specs: list[str]
+) -> tuple[dict[tuple[str, str], object], int]:
+    """The {(name, version): tool} jobs whose verified archive is on disk,
+    and how many distinct uninstalled tools were resolved overall."""
+    dist_path = Path(g.idf_tools_path) / "dist"
+
+    def on_broken(name: str, e: ToolBinaryError) -> bool:
+        # Repairing a broken installed binary is the installer's job
+        print(f"leaving broken {name} to the installer: {e}", file=sys.stderr)
+        return False
+
+    pending: dict[tuple[str, str], object] = {}
+    resolved: set[tuple[str, str]] = set()
+    for tool, name, version, download in iter_tool_downloads(
+        targets_csv, tool_specs, on_broken
+    ):
+        resolved.add((name, version))
+        # Mirror the prefetch: an entry it could not verify is never trusted
+        if not (download.sha256 and download.size):
+            continue
+        # Trusted as-is: the prefetch verifies archives at their final name,
+        # and the installer redoes anything this pass fails on
+        if (name, version) in pending or not (
+            dist_path / archive_name(download)
+        ).is_file():
+            continue
+        pending[(name, version)] = tool
+    return pending, len(resolved)
+
+
+def install_one(tool: object, name: str, version: str) -> bool | None:
+    """True on success, False on a cleaned-up failure, None when the torn
+    dest dir survived and could fool the installer's binary probe."""
+    try:
+        tool.install(version)
+    # check_binary_valid exits via SystemExit; the installer redoes failures
+    except (Exception, SystemExit) as e:  # noqa: BLE001  # pylint: disable=broad-exception-caught
+        # Name the type: idf_tools' fatal() raises SystemExit(1), which
+        # would render as a bare "1"
+        print(
+            f"pre-extracting {name}@{version} failed, leaving it to the "
+            f"installer: {type(e).__name__}: {e}",
+            file=sys.stderr,
+        )
+        # A torn dest dir must not look installed to the installer
+        dest = tool.get_path_for_version(version)
+        try:
+            rmtree(dest)
+        except FileNotFoundError:  # pragma: no cover  # failed before mkdir
+            pass
+        except OSError as cleanup_err:
+            print(
+                f"could not remove {dest}: {cleanup_err}; the installer may "
+                "trust the partial tool dir, delete it manually if the build "
+                "fails",
+                file=sys.stderr,
+            )
+            return None
+        return False
+    # Per-tool completion keeps the multi-minute unpack phase visibly alive
+    print(f"extracted {name}@{version}", flush=True)
+    return True
+
+
+def main() -> None:
+    _script, idf_framework_root, targets_csv, workers_str, *tool_specs = sys.argv
+    init_idf_tools(idf_framework_root)
+    pending, resolved = collect_pending(targets_csv, tool_specs)
+    if len(pending) < 2:
+        # Nothing to parallelize; the count makes a naming/resolution drift
+        # that would silently disable this pass observable
+        print(
+            f"{len(pending)} of {resolved} uninstalled tool(s) have a "
+            "prefetched archive; leaving them to the installer",
+            flush=True,
+        )
+        return
+    workers = min(int(workers_str), len(pending))
+    print(
+        f"Extracting {len(pending)} ESP-IDF tool archive(s) with "
+        f"{workers} worker(s): "
+        + ", ".join(f"{name}@{version}" for name, version in pending),
+        flush=True,
+    )
+    with ThreadPoolExecutor(max_workers=workers) as ex:
+        futures = [
+            ex.submit(install_one, tool, name, version)
+            for (name, version), tool in pending.items()
+        ]
+        try:
+            results = [future.result() for future in futures]
+        except BaseException:  # pragma: no cover
+            # Ctrl-C: drop queued extractions; in-flight ones finish whole
+            ex.shutdown(wait=True, cancel_futures=True)
+            raise
+    # A survivor could fool the installer; every job failing is systematic.
+    # Either way a nonzero exit makes the caller warn
+    failed = sum(result is not True for result in results)
+    if failed:
+        print(f"{failed} of {len(results)} pre-extractions failed", file=sys.stderr)
+        if None in results or failed == len(results):
+            sys.exit(1)
+
+
+main()

esphome/framework_helpers.py

@@ -287,10 +287,25 @@ def _detect_archive_root(names: Iterable[str]) -> str | None:
     return root if has_descendant else None
 
 
+def _resolve_progress(
+    progress: Callable[[float], None] | None,
+    progress_header: str | None,
+    has_work: bool,
+) -> Callable[[float], None] | None:
+    """Fraction reporter for an extractor: the caller's callback wins over a
+    private ``progress_header`` bar."""
+    if progress is not None:
+        return progress
+    if progress_header and has_work:
+        return ProgressBar(progress_header).update
+    return None
+
+
 def _tar_extract_all(
     data: io.BufferedIOBase,
     extract_dir: PathType = ".",
     progress_header: str | None = None,
+    progress: Callable[[float], None] | None = None,
 ):
     """
     Extract a TAR archive to the specified directory.
@@ -305,6 +320,7 @@ def _tar_extract_all(
         data: File-like object containing the TAR archive
         extract_dir: Directory to extract contents to
         progress_header: If set, show a progress bar with this header
+        progress: fraction callback (0..1, ends at 1.0); overrides progress_header
     """
     import tarfile
 
@@ -363,21 +379,20 @@ def _tar_extract_all(
             safe_members.append(member)
 
         total = len(safe_members)
-        progress = (
-            ProgressBar(progress_header) if progress_header and total > 0 else None
-        )
+        report = _resolve_progress(progress, progress_header, total > 0)
         for i, member in enumerate(safe_members, 1):
             tar_ref.extract(member, abs_dest)
-            if progress is not None:
-                progress.update(i / total)
-        if progress is not None:
-            progress.update(1)
+            if report is not None:
+                report(i / total)
+        if report is not None:
+            report(1)
 
 
 def _zip_extract_all(
     data: io.BufferedIOBase,
     extract_dir: PathType = ".",
     progress_header: str | None = None,
+    progress: Callable[[float], None] | None = None,
 ):
     """
     Extract a ZIP archive to the specified directory.
@@ -386,6 +401,7 @@ def _zip_extract_all(
         data: File-like object containing the ZIP archive
         extract_dir: Directory to extract contents to
         progress_header: If set, show a progress bar with this header
+        progress: fraction callback (0..1, ends at 1.0); overrides progress_header
     """
     import zipfile
 
@@ -402,9 +418,7 @@ def _zip_extract_all(
         strip_prefix = f"{strip_root}/" if strip_root is not None else None
 
         total = len(all_members)
-        progress = (
-            ProgressBar(progress_header) if progress_header and total > 0 else None
-        )
+        report = _resolve_progress(progress, progress_header, total > 0)
 
         for i, member in enumerate(all_members, 1):
             # 1. Normalize name
@@ -437,10 +451,10 @@ def _zip_extract_all(
             # 6. Extract
             zip_ref.extract(member, extract_dir)
 
-            if progress is not None:
-                progress.update(i / total)
-        if progress is not None:
-            progress.update(1)
+            if report is not None:
+                report(i / total)
+        if report is not None:
+            report(1)
 
 
 def _rename_with_retry(
@@ -471,6 +485,7 @@ def _7z_extract_all(
     data: io.BufferedIOBase,
     extract_dir: PathType = ".",
     progress_header: str | None = None,
+    progress: Callable[[float], None] | None = None,
 ):
     """
     Extract a 7z archive to the specified directory.
@@ -485,6 +500,7 @@ def _7z_extract_all(
         data: File-like object containing the 7z archive (must be seekable)
         extract_dir: Directory to extract contents to
         progress_header: If set, show a progress bar with this header
+        progress: called with 1.0 on completion; overrides progress_header
     """
     import py7zr
 
@@ -523,19 +539,15 @@ def _7z_extract_all(
                     continue
                 safe_targets.append(raw)
 
-            progress = (
-                ProgressBar(progress_header)
-                if progress_header and safe_targets
-                else None
-            )
+            report = _resolve_progress(progress, progress_header, bool(safe_targets))
 
             if len(safe_targets) == len(all_names):
                 z.extractall(path=staging)
             else:
                 z.extract(path=staging, targets=safe_targets)
 
-            if progress is not None:
-                progress.update(1)
+            if report is not None:
+                report(1)
 
         src_root = staging / strip_root if strip_root else staging
         for item in src_root.iterdir():
@@ -566,6 +578,7 @@ def archive_extract_all(
     archive: PathType | io.RawIOBase | IO[bytes],
     extract_dir: PathType = ".",
     progress_header: str | None = None,
+    progress: Callable[[float], None] | None = None,
 ):
     """
     Extract an archive file to the specified directory.
@@ -574,6 +587,7 @@ def archive_extract_all(
         archive: Path to archive file or file-like object
         extract_dir: Directory to extract contents to
         progress_header: If set, show a progress bar with this header
+        progress: fraction callback (0..1, ends at 1.0); overrides progress_header
 
     Raises:
         TypeError: If archive is not a valid type
@@ -604,7 +618,9 @@ def archive_extract_all(
                 break
         if matched_fct is None:
             raise ValueError("Unsupported archive format")
-        matched_fct(archive_ref, extract_dir, progress_header=progress_header)
+        matched_fct(
+            archive_ref, extract_dir, progress_header=progress_header, progress=progress
+        )
 
 
 def _open_ranged(
@@ -768,13 +784,17 @@ def _stream_response_to_file(
 # hammering the host or the mirrors.
 BATCH_DOWNLOAD_WORKERS = 4
 
+# Concurrent archive extractions per batch; unpacking stops scaling well
+# before high core counts since the workers share one disk.
+BATCH_EXTRACT_WORKERS = 10
+
 
 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.
+    """Run ``(name, size, fetch)`` 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
@@ -949,15 +969,26 @@ def resume_fetch_job(
     return fetch
 
 
-def warn_prefetch_failures(
+def is_expected_fetch_error(err: BaseException) -> bool:
+    """Download failures the callers degrade on, vs programming errors."""
+    from esphome.core import EsphomeError  # local import avoids circular dependency
+
+    return isinstance(err, (EsphomeError, OSError))
+
+
+def warn_batch_failures(
     failures: list[tuple[str, BaseException]],
     message: str = "Could not prefetch %s: %s",
 ) -> None:
-    """Warn per failed batch-prefetch job; the caller's installer retries them."""
+    """Warn per failed batch job, keeping the traceback of unexpected errors."""
     for name, err in failures:
         # failure_reason: a message-less exception must not log blank
-        _LOGGER.warning(message, name, failure_reason(err))
-        _LOGGER.debug("Prefetch failure detail", exc_info=err)
+        if is_expected_fetch_error(err):
+            _LOGGER.warning(message, name, failure_reason(err))
+            _LOGGER.debug("Failure detail", exc_info=err)
+        else:
+            # A programming error must not be reduced to a bare message
+            _LOGGER.warning(message, name, failure_reason(err), exc_info=err)
 
 
 def download_with_resume(

esphome/platformio/library.py

@@ -35,7 +35,7 @@ from esphome.framework_helpers import (
     failure_reason,
     rmdir,
     run_batch_downloads,
-    warn_prefetch_failures,
+    warn_batch_failures,
 )
 
 _LOGGER = logging.getLogger(__name__)
@@ -1091,7 +1091,7 @@ def _prefetch_wave(
             + [(c.name, 0, partial(_clone_source, c, salt, namespace)) for c in clones],
         )
         # The sequential call below retries and raises the real error
-        warn_prefetch_failures(
+        warn_batch_failures(
             failures, "Prefetch of %s failed (retrying sequentially): %s"
         )
     except Exception as err:  # noqa: BLE001  # pylint: disable=broad-exception-caught

esphome/platformio/prefetch.py

@@ -33,12 +33,13 @@ import time
 from typing import Any, NamedTuple
 
 from esphome.framework_helpers import (
+    BATCH_EXTRACT_WORKERS,
     content_length,
     discard_partial_download,
     failure_reason,
     resume_fetch_job,
     run_batch_downloads,
-    warn_prefetch_failures,
+    warn_batch_failures,
 )
 from esphome.helpers import get_bool_env, get_usable_cpu_count, rmtree
 
@@ -701,7 +702,7 @@ def _preinstall(
     would hang, not fail). Waves skip dependencies; the installed
     manifests feed the next wave. Any failure falls back to pio run.
     """
-    workers = min(get_usable_cpu_count(), len(entries))
+    workers = min(get_usable_cpu_count(), len(entries), BATCH_EXTRACT_WORKERS)
     # One manager per worker (_install mutates instance state); built
     # serially because construction rewires the shared manager logger
     managers: SimpleQueue = SimpleQueue()
@@ -890,7 +891,7 @@ def _prefetch(build_dir: Path, env: str) -> None:
         )
         # PlatformIO retries failed packages itself, without resume
         failures = run_batch_downloads("Downloading PlatformIO packages", jobs)
-        warn_prefetch_failures(failures)
+        warn_batch_failures(failures)
         failed_names = {name for name, _ in failures}
     elif not groups and not unresolved:
         # Record the no-work run so the parent skips the next spawn.

esphome/platformio/registry.py

@@ -14,12 +14,16 @@ from typing import NamedTuple
 
 from esphome.core import EsphomeError
 from esphome.framework_helpers import (
+    BATCH_EXTRACT_WORKERS,
     archive_extract_all,
     download_from_mirrors,
     download_with_resume,
+    is_expected_fetch_error,
     rmdir,
     run_batch_downloads,
+    warn_batch_failures,
 )
+from esphome.helpers import get_usable_cpu_count
 from esphome.net_retry import fetch_with_retry, http_request
 
 _LOGGER = logging.getLogger(__name__)
@@ -160,6 +164,14 @@ def _check_layout(name: str, dest: Path, expect: Collection[str]) -> None:
             )
 
 
+# (name, version, dest, mirrors, expect) as accepted by install_packages
+PackageSpec = tuple[str, str, Path, list[str], Collection[str]]
+
+
+def _archive_path(downloads_dir: Path, name: str, version: str) -> Path:
+    return downloads_dir / f"{name}-{version}"
+
+
 class _PendingArchive(NamedTuple):
     name: str
     version: str
@@ -174,6 +186,26 @@ def _already_installed(dest: Path) -> bool:
     return (dest / ".esphome_extracted").is_file()
 
 
+def _batched_download_progress(
+    name: str, version: str, size: int | None, extract_progress: Callable[[float], None]
+) -> Callable[[int], None]:
+    """Download tracker for a batched install: no private bar, no bytes (the
+    shared bar must never run backwards); the zero tick keeps cancellation
+    observable and a real refetch is announced once."""
+    announced = False
+
+    def progress(done: int) -> None:
+        nonlocal announced
+        # size-less registry entries still announce: streaming starts at
+        # done=0, while a verify no-op credits the full file in one tick
+        if not announced and done < (size or 1):
+            _LOGGER.info("Re-downloading %s %s ...", name, version)
+            announced = True
+        extract_progress(0.0)
+
+    return progress
+
+
 def prefetch_packages(
     packages: list[tuple[str, str, Path, list[str]]], downloads_dir: Path
 ) -> None:
@@ -242,7 +274,7 @@ def prefetch_packages(
         [(entry.name, entry.size, partial(_fetch, entry)) for entry in pending],
     )
     for name, err in failures:
-        if isinstance(err, (EsphomeError, OSError)):
+        if is_expected_fetch_error(err):
             # Expected download failures: install_package retries this one
             # itself, with a visible bar
             _LOGGER.debug("Prefetch of %s failed: %s", name, err)
@@ -259,6 +291,7 @@ def install_package(
     mirrors: list[str],
     downloads_dir: Path,
     expect: Collection[str],
+    extract_progress: Callable[[float], None] | None = None,
 ) -> None:
     """Download, verify, and extract one package if not already installed.
 
@@ -266,6 +299,9 @@ def install_package(
     publishes; a mirror override (URL templates with ``{VERSION}``/``{SYSTEM}``
     substitution) is trusted as configured. ``downloads_dir`` holds the
     archive between runs so an interrupted download resumes.
+
+    ``extract_progress`` receives extraction fractions in [0, 1] instead of
+    the private per-file bars (see ``install_packages``).
     """
     if not expect:
         # Layout validation before marker.touch() is the only guard against
@@ -288,8 +324,15 @@ def install_package(
         rmdir(dest, msg=f"Clean up incomplete {name} install")
         # Persistent location so an interrupted download resumes across runs.
         downloads_dir.mkdir(parents=True, exist_ok=True)
-        archive = downloads_dir / f"{name}-{version}"
-        _LOGGER.info("Downloading %s %s ...", name, version)
+        archive = _archive_path(downloads_dir, name, version)
+        # Batch header names each package; keep INFO when an archive
+        # unexpectedly needs a real download (the shared bar won't move)
+        log = (
+            _LOGGER.debug
+            if extract_progress is not None and archive.is_file()
+            else _LOGGER.info
+        )
+        log("Downloading %s %s ...", name, version)
         if mirrors:
             _LOGGER.warning(
                 "Downloading %s from a mirror override; checksum verification "
@@ -301,11 +344,85 @@ def install_package(
             )
         else:
             url, sha256, size = registry_download(name, version)
-            download_with_resume(url, archive, sha256=sha256, size=size)
-        _LOGGER.info("Extracting %s ...", name)
-        archive_extract_all(archive, dest, progress_header="Extracting")
+            download_with_resume(
+                url,
+                archive,
+                sha256=sha256,
+                size=size,
+                progress=None
+                if extract_progress is None
+                else _batched_download_progress(name, version, size, extract_progress),
+            )
+        log("Extracting %s ...", name)
+        archive_extract_all(
+            archive, dest, progress_header="Extracting", progress=extract_progress
+        )
         # Validate the layout before recording success, so an unexpected
         # package is never cached as a working install.
         _check_layout(name, dest, expect)
         marker.touch()
         archive.unlink(missing_ok=True)
+
+
+def install_packages(specs: Collection[PackageSpec], downloads_dir: Path) -> None:
+    """Install several packages; prefetched archives extract in parallel under
+    one shared bar, the rest take the sequential ``install_package`` path.
+    The first failure is re-raised."""
+    pending: list[tuple[PackageSpec, int]] = []
+    rest: list[PackageSpec] = []
+    seen: set[str] = set()
+    for spec in specs:
+        name, version, dest, mirrors, _expect = spec
+        archive = _archive_path(downloads_dir, name, version)
+        # Duplicate entries share one archive and would race each other
+        # between two workers; mirror prefetch_packages' dedupe
+        if _already_installed(dest) or mirrors or archive.name in seen:
+            rest.append(spec)
+            continue
+        try:
+            # An archive at its final name already passed sha256/size
+            # verification
+            size = archive.stat().st_size
+        except FileNotFoundError:
+            rest.append(spec)
+            continue
+        seen.add(archive.name)
+        pending.append((spec, size))
+    if len(pending) < 2:
+        for name, version, dest, mirrors, expect in specs:
+            install_package(name, version, dest, mirrors, downloads_dir, expect=expect)
+        return
+    workers = min(get_usable_cpu_count(), len(pending), BATCH_EXTRACT_WORKERS)
+    _LOGGER.info(
+        "Extracting %d package archive(s) with %d worker(s): %s",
+        len(pending),
+        workers,
+        ", ".join(spec[0] for spec, _ in pending),
+    )
+
+    def _install(spec: PackageSpec, size: int, tracker: Callable[[int], None]) -> None:
+        name, version, dest, mirrors, expect = spec
+        install_package(
+            name,
+            version,
+            dest,
+            mirrors,
+            downloads_dir,
+            expect=expect,
+            extract_progress=lambda frac: tracker(int(frac * size)),
+        )
+
+    failures = run_batch_downloads(
+        "Extracting packages",
+        [(spec[0], size, partial(_install, spec, size)) for spec, size in pending],
+        max_workers=workers,
+    )
+    if failures:
+        # Warn on the first failure too: the raised exception's message may
+        # not name which package failed
+        warn_batch_failures(failures, "Could not install %s: %s")
+        raise failures[0][1]
+    # Sequential remainder after the batch, so a duplicate spec cannot
+    # unlink the archive its batched twin was sized from
+    for name, version, dest, mirrors, expect in rest:
+        install_package(name, version, dest, mirrors, downloads_dir, expect=expect)

tests/unit_tests/fixtures/idf_tools_stub/idf_tools.py

@@ -1,7 +1,8 @@
-"""Minimal idf_tools stand-in for get_tool_downloads.py tests."""
+"""Minimal idf_tools stand-in for the espidf helper-script tests."""
 
 from collections.abc import Iterable
 import os
+import pathlib
 
 CURRENT_PLATFORM = "linux-amd64"
 TOOLS_FILE = "tools/tools.json"
@@ -54,6 +55,7 @@ class _Tool:
         installed: Iterable[str] = (),
         broken: bool = False,
     ) -> None:
+        self.name = ""  # filled in from the _TOOLS key below
         self.versions = versions
         self._recommended = recommended
         self.versions_installed = list(installed)
@@ -69,6 +71,19 @@ class _Tool:
         if self._broken:
             raise ToolBinaryError("broken binary")
 
+    def get_path_for_version(self, version: str) -> str:
+        return str(pathlib.Path(g.idf_tools_path) / "tools" / self.name / version)
+
+    def install(self, version: str) -> None:
+        dest = pathlib.Path(self.get_path_for_version(version))
+        dest.mkdir(exist_ok=True, parents=True)
+        if self.name in os.environ.get("TEST_FAIL_INSTALL", "").split(","):
+            # Fail mid-install like a torn unpack: the partial dir is left
+            # behind and check_binary_valid's failure path exits the process
+            (dest / ".partial").write_text("torn", encoding="utf-8")
+            raise SystemExit(1)
+        (dest / ".installed").write_text("ok", encoding="utf-8")
+
 
 _TOOLS = {
     "cmake": _Tool(
@@ -97,8 +112,17 @@ _TOOLS = {
     "no-download-tool": _Tool({"4.0": _Version(None)}, "4.0"),
 }
 
+for _name, _tool in _TOOLS.items():
+    _tool.name = _name
+
 
 def load_tools_info() -> dict[str, _Tool]:
+    # Test hook: strip verification metadata from the named tools
+    for name in os.environ.get("TEST_NO_SHA", "").split(","):
+        if (tool := _TOOLS.get(name)) is not None:
+            for version in tool.versions.values():
+                if (download := version.get_download_for_platform("")) is not None:
+                    download.sha256 = ""
     return _TOOLS
 
 

tests/unit_tests/test_arduino8266_framework.py

@@ -57,7 +57,7 @@ def test_tools_path_default_and_prefix(tmp_path: Path) -> None:
 def test_check_and_install_returns_paths(tmp_path: Path) -> None:
     with (
         patch.dict(os.environ, {"ESPHOME_ARDUINO8266_PREFIX": str(tmp_path)}),
-        patch.object(framework, "install_package") as mock_install,
+        patch.object(framework, "install_packages") as mock_install,
         patch.object(framework, "prefetch_packages") as mock_prefetch,
         patch.object(framework, "find_ninja", return_value=tmp_path / "ninja"),
     ):
@@ -65,26 +65,27 @@ def test_check_and_install_returns_paths(tmp_path: Path) -> None:
     assert paths.framework == tmp_path / "frameworks" / "3.30102.0"
     assert paths.toolchain == tmp_path / "toolchains" / framework.TOOLCHAIN_VERSION
     assert paths.ninja == tmp_path / "ninja"
-    assert mock_install.call_count == 2
     # Full argument pinning: a copy-paste swap between the two near-identical
-    # calls (mirrors, destination) must not stay green
-    fw_call, tc_call = mock_install.call_args_list
-    assert fw_call.args == (
-        framework.FRAMEWORK_PACKAGE,
-        "3.30102.0",
-        tmp_path / "frameworks" / "3.30102.0",
-        framework.ESPHOME_ARDUINO8266_FRAMEWORK_MIRRORS,
+    # specs (mirrors, destination) must not stay green
+    assert mock_install.call_args.args == (
+        (
+            (
+                framework.FRAMEWORK_PACKAGE,
+                "3.30102.0",
+                tmp_path / "frameworks" / "3.30102.0",
+                framework.ESPHOME_ARDUINO8266_FRAMEWORK_MIRRORS,
+                ("cores/esp8266", "tools/sdk", "libraries"),
+            ),
+            (
+                framework.TOOLCHAIN_PACKAGE,
+                framework.TOOLCHAIN_VERSION,
+                tmp_path / "toolchains" / framework.TOOLCHAIN_VERSION,
+                framework.ESPHOME_ARDUINO8266_TOOLCHAIN_MIRRORS,
+                ("bin", "xtensa-lx106-elf"),
+            ),
+        ),
         tmp_path / "downloads",
     )
-    assert fw_call.kwargs["expect"] == ("cores/esp8266", "tools/sdk", "libraries")
-    assert tc_call.args == (
-        framework.TOOLCHAIN_PACKAGE,
-        framework.TOOLCHAIN_VERSION,
-        tmp_path / "toolchains" / framework.TOOLCHAIN_VERSION,
-        framework.ESPHOME_ARDUINO8266_TOOLCHAIN_MIRRORS,
-        tmp_path / "downloads",
-    )
-    assert tc_call.kwargs["expect"] == ("bin", "xtensa-lx106-elf")
     # The prefetch sees the same package specs as the installs
     assert mock_prefetch.call_args.args == (
         [

tests/unit_tests/test_espidf_framework.py

@@ -37,6 +37,7 @@ from esphome.espidf.framework import (
     _patch_tools_json_demote_unused_tools,
     _patch_tools_json_for_linux_arm64,
     _prefetch_idf_tool_archives,
+    _preinstall_idf_tool_archives,
     _read_stamp,
     _stamp_covers,
     _windows_long_paths_enabled,
@@ -403,6 +404,7 @@ def espidf_mocks(setup_core: Path):
         patch("esphome.espidf.framework._patch_tools_json_for_linux_arm64"),
         patch("esphome.espidf.framework._patch_tools_json_demote_unused_tools"),
         patch("esphome.espidf.framework._prefetch_idf_tool_archives"),
+        patch("esphome.espidf.framework._preinstall_idf_tool_archives"),
         patch("esphome.espidf.framework._write_stamp"),
         patch("esphome.espidf.framework._check_stamp", return_value=True),
         patch("esphome.espidf.framework._stamp_covers", return_value=True),
@@ -1051,11 +1053,16 @@ def test_prefetch_downloads_archives_concurrently(tmp_path: Path) -> None:
     assert download.call_count == 6
 
 
-def test_prefetch_skips_already_downloaded_archives(tmp_path: Path) -> None:
+def test_prefetch_reverifies_already_downloaded_archives(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A pre-existing archive is not skipped: download_with_resume keeps it
+    only when the sha256 matches, so the pre-extraction can trust it."""
     dist = get_idf_tools_path() / "dist"
     dist.mkdir(parents=True)
     (dist / "cmake-3.30.2.tar.gz").write_bytes(b"cached")
     with (
+        caplog.at_level(logging.INFO),
         patch(
             "esphome.espidf.framework.run_command",
             return_value=(True, _PREFETCH_JSON, ""),
@@ -1065,9 +1072,12 @@ def test_prefetch_skips_already_downloaded_archives(tmp_path: Path) -> None:
     ):
         _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
 
-    # only the missing archive is downloaded
-    assert download.call_count == 1
-    assert download.call_args[0][1] == dist / "ninja.zip"
+    assert sorted(call[0][1] for call in download.call_args_list) == [
+        dist / "cmake-3.30.2.tar.gz",
+        dist / "ninja.zip",
+    ]
+    # The log distinguishes verifying cached archives from real downloads
+    assert "Downloading 2 ESP-IDF tool archive(s) (1 cached, verifying)" in caplog.text
 
 
 @pytest.mark.parametrize(
@@ -1188,30 +1198,41 @@ def test_prefetch_passes_targets_and_tools_to_script(tmp_path: Path) -> None:
     cmd = run.call_args[0][0]
     assert cmd[-3:] == ["esp32,esp32c3", "required", "cmake"]
     assert cmd[1].endswith("get_tool_downloads.py")
-    # the script inherits the caller's env plus the framework tools PYTHONPATH
+    # the script inherits the caller's env plus an explicit PYTHONPATH:
+    # sibling scripts, the esphome package root, the framework's idf_tools
     env = run.call_args[1]["env"]
     assert env["IDF_TOOLS_PATH"] == "/x"
-    assert env["PYTHONPATH"] == str(tmp_path / "tools")
+    assert env["PYTHONPATH"] == os.pathsep.join(
+        (
+            str(_ESPIDF_SCRIPTS_DIR),
+            str(_ESPIDF_SCRIPTS_DIR.parents[1]),
+            str(tmp_path / "tools"),
+        )
+    )
 
 
 def test_framework_install_prefetches_before_installer(
     espidf_mocks: SimpleNamespace,
 ) -> None:
-    """The prefetch runs before idf_tools.py install so the installer finds
-    the archives already in dist/."""
+    """The prefetch downloads and the pre-extraction both run before
+    idf_tools.py install so the installer finds the tools in place."""
     calls: list[str] = []
     with (
         patch(
             "esphome.espidf.framework._prefetch_idf_tool_archives",
             side_effect=lambda *a, **k: calls.append("prefetch"),
         ),
+        patch(
+            "esphome.espidf.framework._preinstall_idf_tool_archives",
+            side_effect=lambda *a, **k: calls.append("preinstall"),
+        ),
     ):
         espidf_mocks.run_ok.side_effect = lambda *a, **k: (
             calls.append("install") or True
         )
         check_esp_idf_install(_IDF_VERSION, force=True)
 
-    assert calls.index("prefetch") < calls.index("install")
+    assert calls.index("prefetch") < calls.index("preinstall") < calls.index("install")
 
 
 # ---------------------------------------------------------------------------
@@ -1220,13 +1241,16 @@ def test_framework_install_prefetches_before_installer(
 
 
 _IDF_TOOLS_STUB_DIR = Path(__file__).parent / "fixtures" / "idf_tools_stub"
+_ESPIDF_SCRIPTS_DIR = Path(__file__).parents[2] / "esphome" / "espidf"
 
 
-def _run_downloads_script(
-    tmp_path: Path, *args: str, env_extra: dict[str, str] | None = None
+def _run_espidf_script(
+    tmp_path: Path,
+    script_name: str,
+    *args: str,
+    env_extra: dict[str, str] | None = None,
 ) -> subprocess.CompletedProcess[str]:
-    """Run the real get_tool_downloads.py against the stub idf_tools module."""
-    script = Path(__file__).parents[2] / "esphome" / "espidf" / "get_tool_downloads.py"
+    """Run a real espidf helper script against the stub idf_tools module."""
     env = os.environ | {
         "PYTHONPATH": str(_IDF_TOOLS_STUB_DIR),
         "IDF_TOOLS_PATH": str(tmp_path / "tp"),
@@ -1234,7 +1258,12 @@ def _run_downloads_script(
     if env_extra:
         env |= env_extra
     return subprocess.run(
-        [sys.executable, str(script), str(tmp_path / "fw"), *args],
+        [
+            sys.executable,
+            str(_ESPIDF_SCRIPTS_DIR / script_name),
+            str(tmp_path / "fw"),
+            *args,
+        ],
         capture_output=True,
         text=True,
         env=env,
@@ -1246,7 +1275,7 @@ def test_get_tool_downloads_lists_missing_tools(tmp_path: Path) -> None:
     """Installed versions are skipped, tools that fail their binary check are
     still listed, rename_dist decides the dist filename, and idf_tools' stdout
     chatter stays off the JSON channel."""
-    result = _run_downloads_script(tmp_path, "esp32", "required")
+    result = _run_espidf_script(tmp_path, "get_tool_downloads.py", "esp32", "required")
 
     assert result.returncode == 0, result.stderr
     downloads = {d["name"]: d for d in json.loads(result.stdout)}
@@ -1262,8 +1291,9 @@ def test_get_tool_downloads_lists_missing_tools(tmp_path: Path) -> None:
 
 
 def test_get_tool_downloads_applies_mirror_rewrite(tmp_path: Path) -> None:
-    result = _run_downloads_script(
+    result = _run_espidf_script(
         tmp_path,
+        "get_tool_downloads.py",
         "esp32",
         "required",
         env_extra={"TEST_MIRROR_PREFIX": "https://mirror.test/"},
@@ -1274,13 +1304,13 @@ def test_get_tool_downloads_applies_mirror_rewrite(tmp_path: Path) -> None:
     assert all(d["url"].startswith("https://mirror.test/") for d in downloads)
 
 
-def _run_downloads_inprocess(
+def _run_espidf_script_inprocess(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
-    capsys: pytest.CaptureFixture[str],
+    script_name: str,
     *args: str,
-) -> list[dict]:
-    """Execute get_tool_downloads.py in-process against the stub idf_tools.
+) -> None:
+    """Execute an espidf helper script in-process against the stub idf_tools.
 
     Unlike the subprocess variant this runs under coverage, exercising the
     script's own lines.
@@ -1291,10 +1321,24 @@ def _run_downloads_inprocess(
     stub = importlib.util.module_from_spec(spec)
     spec.loader.exec_module(stub)
     monkeypatch.setitem(sys.modules, "idf_tools", stub)
+    # _tool_resolution binds idf_tools objects at import; force a fresh
+    # import against this test's stub instance
+    monkeypatch.delitem(sys.modules, "_tool_resolution", raising=False)
+    # python <script> puts the script's directory on sys.path; runpy does not
+    monkeypatch.syspath_prepend(str(_ESPIDF_SCRIPTS_DIR))
     monkeypatch.setenv("IDF_TOOLS_PATH", str(tmp_path / "tp"))
-    script = Path(__file__).parents[2] / "esphome" / "espidf" / "get_tool_downloads.py"
+    script = _ESPIDF_SCRIPTS_DIR / script_name
     monkeypatch.setattr(sys, "argv", [str(script), str(tmp_path / "fw"), *args])
     runpy.run_path(str(script))
+
+
+def _run_downloads_inprocess(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+    *args: str,
+) -> list[dict]:
+    _run_espidf_script_inprocess(tmp_path, monkeypatch, "get_tool_downloads.py", *args)
     return json.loads(capsys.readouterr().out)
 
 
@@ -2008,3 +2052,245 @@ def test_check_windows_path_length_long_path_warns(
     assert "long path support" in message
     # The install is global now; the remedy is the prefix env, not moving the project.
     assert "ESPHOME_ESP_IDF_PREFIX" in message
+
+
+# ---------------------------------------------------------------------------
+# _preinstall_idf_tool_archives
+# ---------------------------------------------------------------------------
+
+
+def test_preinstall_streams_script_with_workers(tmp_path: Path) -> None:
+    """The pre-extraction streams install_tool_archives.py with the worker
+    count and the same targets/tools the installer will get."""
+    with (
+        patch(
+            "esphome.espidf.framework._run_idf_tools_script",
+            return_value=(True, None, None),
+        ) as run_script,
+        patch("esphome.espidf.framework.get_usable_cpu_count", return_value=3),
+    ):
+        _preinstall_idf_tool_archives(
+            tmp_path, "esp32,esp32c3", ["required", "cmake"], {"IDF_TOOLS_PATH": "x"}
+        )
+    run_script.assert_called_once_with(
+        tmp_path,
+        "install_tool_archives.py",
+        "ESP-IDF tool archive extraction",
+        args=["esp32,esp32c3", "3", "required", "cmake"],
+        env={"IDF_TOOLS_PATH": "x"},
+        stream_output=True,
+    )
+
+
+def test_preinstall_caps_workers(tmp_path: Path) -> None:
+    """A high core count is capped; the workers share one disk."""
+    with (
+        patch(
+            "esphome.espidf.framework._run_idf_tools_script",
+            return_value=(True, None, None),
+        ) as run_script,
+        patch("esphome.espidf.framework.get_usable_cpu_count", return_value=64),
+    ):
+        _preinstall_idf_tool_archives(tmp_path, "esp32", ["required"], None)
+    assert run_script.call_args.kwargs["args"][1] == "10"
+
+
+def test_preinstall_script_failure_only_warns(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A failed pre-extraction leaves the install to the sequential path."""
+    with (
+        patch(
+            "esphome.espidf.framework._run_idf_tools_script",
+            return_value=(False, None, None),
+        ),
+        patch("esphome.espidf.framework.get_usable_cpu_count", return_value=1),
+    ):
+        _preinstall_idf_tool_archives(tmp_path, "esp32", ["required"], None)
+    assert "pre-extraction failed" in caplog.text
+
+
+def test_preinstall_exception_only_warns(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """An unexpected error must not become a new way for the install to fail,
+    and keeps its traceback at WARNING."""
+    with (
+        patch(
+            "esphome.espidf.framework._run_idf_tools_script",
+            side_effect=TypeError("bad call"),
+        ),
+        patch("esphome.espidf.framework.get_usable_cpu_count", return_value=1),
+    ):
+        _preinstall_idf_tool_archives(tmp_path, "esp32", ["required"], None)
+    record = next(r for r in caplog.records if "pre-extraction failed" in r.message)
+    assert record.exc_info is not None
+
+
+# ---------------------------------------------------------------------------
+# install_tool_archives.py (against the stub idf_tools module in fixtures/)
+# ---------------------------------------------------------------------------
+
+
+def _make_dist(tmp_path: Path, *names: str) -> None:
+    dist = tmp_path / "tp" / "dist"
+    dist.mkdir(parents=True, exist_ok=True)
+    for name in names:
+        (dist / name).write_bytes(b"x")
+
+
+def test_install_tool_archives_extracts_pending_in_parallel(tmp_path: Path) -> None:
+    """Subprocess end-to-end: tools with a prefetched archive install
+    concurrently; installed and broken tools stay with the installer."""
+    _make_dist(tmp_path, "cmake.tar.gz", "ninja-v1.zip", "x.tar.gz", "y.tar.gz")
+    result = _run_espidf_script(
+        tmp_path, "install_tool_archives.py", "esp32", "8", "required"
+    )
+    assert result.returncode == 0, result.stderr
+    tools = tmp_path / "tp" / "tools"
+    assert (tools / "cmake" / "3.30.2" / ".installed").is_file()
+    assert (tools / "ninja" / "1.12.1" / ".installed").is_file()
+    assert not (tools / "installed-tool").exists()
+    assert not (tools / "broken-tool").exists()
+    # The worker count clamps to the pending count
+    assert (
+        "Extracting 2 ESP-IDF tool archive(s) with 2 worker(s): "
+        "cmake@3.30.2, ninja@1.12.1" in result.stdout
+    )
+    assert "leaving broken broken-tool to the installer" in result.stderr
+
+
+def test_install_tool_archives_single_pending_stays_sequential(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """One pending archive has nothing to parallelize; the installer keeps
+    its normal output."""
+    _make_dist(tmp_path, "cmake.tar.gz")
+    _run_espidf_script_inprocess(
+        tmp_path, monkeypatch, "install_tool_archives.py", "esp32", "4", "required"
+    )
+    assert not (tmp_path / "tp" / "tools").exists()
+    out = capsys.readouterr().out
+    assert "Extracting" not in out
+    # A resolution drift that empties pending stays observable
+    assert "1 of 2 uninstalled tool(s) have a prefetched archive" in out
+
+
+def test_install_tool_archives_failed_install_left_to_installer(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """A per-tool failure warns, removes the torn dest dir so the installer
+    cannot trust it, and moves on; the other tools still install."""
+    _make_dist(tmp_path, "cmake.tar.gz", "ninja-v1.zip")
+    monkeypatch.setenv("TEST_FAIL_INSTALL", "ninja")
+    _run_espidf_script_inprocess(
+        tmp_path, monkeypatch, "install_tool_archives.py", "esp32", "4", "required"
+    )
+    tools = tmp_path / "tp" / "tools"
+    assert (tools / "cmake" / "3.30.2" / ".installed").is_file()
+    assert not (tools / "ninja" / "1.12.1").exists()
+    err = capsys.readouterr().err
+    assert "pre-extracting ninja@1.12.1 failed" in err
+    assert "1 of 2 pre-extractions failed" in err
+
+
+def test_install_tool_archives_all_failed_exits_nonzero(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """Every job failing is a systematic fault; the nonzero exit lets the
+    caller log it."""
+    _make_dist(tmp_path, "cmake.tar.gz", "ninja-v1.zip")
+    monkeypatch.setenv("TEST_FAIL_INSTALL", "cmake,ninja")
+    with pytest.raises(SystemExit) as excinfo:
+        _run_espidf_script_inprocess(
+            tmp_path, monkeypatch, "install_tool_archives.py", "esp32", "4", "required"
+        )
+    assert excinfo.value.code == 1
+    assert "2 of 2 pre-extractions failed" in capsys.readouterr().err
+    tools = tmp_path / "tp" / "tools"
+    assert not (tools / "cmake" / "3.30.2").exists()
+    assert not (tools / "ninja" / "1.12.1").exists()
+
+
+def test_install_tool_archives_inprocess_dedupes_and_skips(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """In-process full flow: duplicate tool@version specs collapse to one
+    job, broken and installed tools are skipped, both pending tools install."""
+    _make_dist(tmp_path, "cmake.tar.gz", "ninja-v1.zip", "x.tar.gz", "y.tar.gz")
+    _run_espidf_script_inprocess(
+        tmp_path,
+        monkeypatch,
+        "install_tool_archives.py",
+        "esp32",
+        "8",
+        "cmake",
+        "ninja",
+        "cmake@3.30.2",
+        "installed-tool",
+        "broken-tool",
+    )
+    captured = capsys.readouterr()
+    assert (
+        "Extracting 2 ESP-IDF tool archive(s) with 2 worker(s): "
+        "cmake@3.30.2, ninja@1.12.1" in captured.out
+    )
+    assert "extracted cmake@3.30.2" in captured.out
+    assert "extracted ninja@1.12.1" in captured.out
+    assert "leaving broken broken-tool to the installer" in captured.err
+    tools = tmp_path / "tp" / "tools"
+    assert (tools / "cmake" / "3.30.2" / ".installed").is_file()
+    assert (tools / "ninja" / "1.12.1" / ".installed").is_file()
+    assert not (tools / "installed-tool").exists()
+    assert not (tools / "broken-tool").exists()
+
+
+def test_install_tool_archives_surviving_torn_dir_escalates(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """A torn dir that survives cleanup could fool the installer; the exit
+    is nonzero even though the other tool succeeded."""
+    import esphome.helpers
+
+    _make_dist(tmp_path, "cmake.tar.gz", "ninja-v1.zip")
+    monkeypatch.setenv("TEST_FAIL_INSTALL", "ninja")
+    monkeypatch.setattr(
+        esphome.helpers, "rmtree", MagicMock(side_effect=OSError("busy"))
+    )
+    with pytest.raises(SystemExit) as excinfo:
+        _run_espidf_script_inprocess(
+            tmp_path, monkeypatch, "install_tool_archives.py", "esp32", "4", "required"
+        )
+    assert excinfo.value.code == 1
+    err = capsys.readouterr().err
+    assert "could not remove" in err
+    assert "1 of 2 pre-extractions failed" in err
+    assert (tmp_path / "tp" / "tools" / "cmake" / "3.30.2" / ".installed").is_file()
+
+
+def test_install_tool_archives_skips_unverifiable_archives(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """An entry the prefetch could not verify is never extracted, even with
+    an archive on disk."""
+    _make_dist(tmp_path, "cmake.tar.gz", "ninja-v1.zip")
+    monkeypatch.setenv("TEST_NO_SHA", "cmake,ninja")
+    _run_espidf_script_inprocess(
+        tmp_path, monkeypatch, "install_tool_archives.py", "esp32", "4", "required"
+    )
+    assert not (tmp_path / "tp" / "tools").exists()
+    assert "0 of 2 uninstalled tool(s) have a prefetched archive" in (
+        capsys.readouterr().out
+    )

tests/unit_tests/test_framework_helpers.py

@@ -523,6 +523,17 @@ class TestArchiveExtractAll:
         archive_extract_all(archive, dest)
         assert (dest / "file.txt").read_text() == "hi"
 
+    def test_progress_callback_passed_through(self, tmp_path: Path) -> None:
+        """The progress kwarg reaches the dispatched extractor."""
+        archive = tmp_path / "test.tar.gz"
+        archive.write_bytes(_gzip_tar_bytes({"file.txt": b"hello"}))
+        dest = tmp_path / "out"
+        dest.mkdir()
+        fractions: list[float] = []
+        archive_extract_all(archive, dest, progress=fractions.append)
+        assert fractions[-1] == 1
+        assert (dest / "file.txt").read_bytes() == b"hello"
+
     def test_invalid_type_raises_type_error(self) -> None:
         with pytest.raises(TypeError, match="archive must be"):
             archive_extract_all(42, ".")  # type: ignore[arg-type]
@@ -1951,6 +1962,19 @@ class TestTarExtractAllBranches:
         mock_pb.assert_called_once_with("Extracting")
         mock_pb.return_value.update.assert_called()
 
+    def test_progress_callback_replaces_bar(self, tmp_path: Path) -> None:
+        """A progress callback wins over progress_header and ends at 1.0."""
+        buf = _make_tar([_reg("a.txt"), _reg("b.txt")], {"a.txt": b"x", "b.txt": b"y"})
+        fractions: list[float] = []
+        with patch("esphome.framework_helpers.ProgressBar") as mock_pb:
+            _tar_extract_all(
+                buf, tmp_path, progress_header="Extracting", progress=fractions.append
+            )
+        mock_pb.assert_not_called()
+        assert fractions == sorted(fractions)
+        assert fractions[-1] == 1
+        assert (tmp_path / "a.txt").is_file()
+
 
 # ---------------------------------------------------------------------------
 # _zip_extract_all — additional branch coverage
@@ -1980,6 +2004,19 @@ class TestZipExtractAllBranches:
         mock_pb.assert_called_once_with("Unzipping")
         mock_pb.return_value.update.assert_called()
 
+    def test_progress_callback_replaces_bar(self, tmp_path: Path) -> None:
+        """A progress callback wins over progress_header and ends at 1.0."""
+        buf = _make_zip([("a.txt", "aaa"), ("b.txt", "bbb")])
+        fractions: list[float] = []
+        with patch("esphome.framework_helpers.ProgressBar") as mock_pb:
+            _zip_extract_all(
+                buf, tmp_path, progress_header="Unzipping", progress=fractions.append
+            )
+        mock_pb.assert_not_called()
+        assert fractions == sorted(fractions)
+        assert fractions[-1] == 1
+        assert (tmp_path / "a.txt").is_file()
+
 
 # ---------------------------------------------------------------------------
 # _rename_with_retry
@@ -2137,6 +2174,20 @@ class TestSevenZipExtractAll:
         mock_pb.assert_called_once_with("Unpacking 7z")
         mock_pb.return_value.update.assert_called()
 
+    def test_progress_callback_replaces_bar(self, tmp_path: Path) -> None:
+        """A progress callback wins over progress_header; 7z reports 1.0 once."""
+        buf = self._make_7z({"file.txt": b"x"})
+        out = tmp_path / "out"
+        out.mkdir()
+        fractions: list[float] = []
+        with patch("esphome.framework_helpers.ProgressBar") as mock_pb:
+            _7z_extract_all(
+                buf, out, progress_header="Unpacking 7z", progress=fractions.append
+            )
+        mock_pb.assert_not_called()
+        assert fractions == [1]
+        assert (out / "file.txt").is_file()
+
     def test_absolute_path_in_names_skipped(self, tmp_path: Path) -> None:
         """Names that resolve as absolute are silently skipped."""
         import py7zr
@@ -2294,18 +2345,36 @@ def test_resume_fetch_job_threads_tracker(tmp_path: Path) -> None:
     )
 
 
-def test_warn_prefetch_failures_names_each_failure(
+def test_warn_batch_failures_names_each_failure(
     caplog: pytest.LogCaptureFixture,
 ) -> None:
     """The shared failure loop warns per job with the failure reason."""
-    from esphome.framework_helpers import warn_prefetch_failures
+    from esphome.framework_helpers import warn_batch_failures
 
-    warn_prefetch_failures([("toolchain-x@1", OSError("down"))])
+    warn_batch_failures([("toolchain-x@1", OSError("down"))])
     assert "Could not prefetch toolchain-x@1: down" in caplog.text
-    warn_prefetch_failures([("lib", OSError("gone"))], "Prefetch of %s failed: %s")
+    warn_batch_failures([("lib", OSError("gone"))], "Prefetch of %s failed: %s")
     assert "Prefetch of lib failed: gone" in caplog.text
 
 
+def test_warn_batch_failures_unexpected_error_keeps_traceback(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """An unexpected error type is not reduced to a bare message; expected
+    download failures stay message-only at WARNING."""
+    from esphome.framework_helpers import warn_batch_failures
+
+    with caplog.at_level(logging.DEBUG):
+        warn_batch_failures(
+            [("pkg", TypeError("bad call")), ("lib", OSError("down"))],
+            "Could not install %s: %s",
+        )
+    warnings = {r.getMessage(): r for r in caplog.records if r.levelname == "WARNING"}
+    assert warnings["Could not install pkg: bad call"].exc_info is not None
+    assert warnings["Could not install lib: down"].exc_info is None
+    assert "Failure detail" in caplog.text
+
+
 @pytest.mark.parametrize(
     ("platform", "input_path", "expected"),
     [

tests/unit_tests/test_platformio_prefetch.py

@@ -1803,3 +1803,20 @@ def test_platformio_private_api_contract() -> None:
     derived = PackageSpec("https://x/y/archive/master.zip")
     assert derived.name and not derived.has_custom_name()
     assert PackageSpec("Foo=https://x/y/archive/master.zip").has_custom_name()
+
+
+def test_preinstall_caps_workers(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A high core count is capped; the workers share one disk."""
+    with (
+        caplog.at_level(logging.INFO),
+        patch.object(pf, "get_usable_cpu_count", return_value=64),
+    ):
+        pf._preinstall(
+            _fake_manager(tmp_path),
+            [(f"p{i}@1", _FakeSpec(name=f"p{i}")) for i in range(11)],
+        )
+    assert "Installing 11 PlatformIO package(s) with 10 extraction worker(s)" in (
+        caplog.text
+    )

tests/unit_tests/test_platformio_registry.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 from contextlib import contextmanager
 import json
+import logging
 import os
 from pathlib import Path
 from unittest.mock import MagicMock, patch
@@ -307,7 +308,11 @@ def test_install_package_downloads_via_registry(tmp_path: Path) -> None:
             "pkg", "1.0.0", dest, [], tmp_path / "dl", expect=("payload",)
         )
     assert mock_download.call_args[0][0] == "http://x/pkg.tar.gz"
-    assert mock_download.call_args[1] == {"sha256": "abc123", "size": 42}
+    assert mock_download.call_args[1] == {
+        "sha256": "abc123",
+        "size": 42,
+        "progress": None,
+    }
 
 
 def test_install_package_validates_expected_layout(tmp_path: Path) -> None:
@@ -723,3 +728,255 @@ def test_prefetch_packages_unexpected_failure_warns(
             tmp_path / "dl",
         )
     assert "TypeError" in caplog.text
+
+
+def _spec(name: str, version: str, dest: Path, mirrors=None, expect=("payload",)):
+    return (name, version, dest, mirrors or [], expect)
+
+
+def test_install_packages_extracts_verified_archives_in_parallel(
+    tmp_path: Path,
+) -> None:
+    """Two prefetched archives install concurrently under one shared bar."""
+    dl = tmp_path / "dl"
+    dl.mkdir()
+    (dl / "a-1.0").write_bytes(b"x" * 10)
+    (dl / "b-2.0").write_bytes(b"y" * 20)
+    with patch.object(registry, "install_package") as mock_install:
+        registry.install_packages(
+            [_spec("a", "1.0", tmp_path / "a"), _spec("b", "2.0", tmp_path / "b")], dl
+        )
+    assert mock_install.call_count == 2
+    calls = sorted(mock_install.call_args_list, key=lambda c: c[0][0])
+    for c, (name, version) in zip(calls, [("a", "1.0"), ("b", "2.0")], strict=True):
+        assert c[0][:3] == (name, version, tmp_path / name)
+        assert c[1]["expect"] == ("payload",)
+        assert callable(c[1]["extract_progress"])
+        # Driving the tracker exercises the fraction-to-bytes scaling
+        c[1]["extract_progress"](0.5)
+        c[1]["extract_progress"](1.0)
+
+
+def test_install_packages_single_archive_stays_sequential(tmp_path: Path) -> None:
+    """One verified archive has nothing to parallelize; original order kept."""
+    dl = tmp_path / "dl"
+    dl.mkdir()
+    (dl / "a-1.0").write_bytes(b"x")
+    specs = [_spec("a", "1.0", tmp_path / "a"), _spec("b", "2.0", tmp_path / "b")]
+    with patch.object(registry, "install_package") as mock_install:
+        registry.install_packages(specs, dl)
+    assert [c[0][0] for c in mock_install.call_args_list] == ["a", "b"]
+    for c in mock_install.call_args_list:
+        assert "extract_progress" not in c[1]
+
+
+def test_install_packages_mirror_and_marker_stay_sequential(tmp_path: Path) -> None:
+    """Mirror overrides and marker hits never enter the parallel batch."""
+    dl = tmp_path / "dl"
+    dl.mkdir()
+    for name, ver in (("a", "1.0"), ("b", "2.0"), ("c", "3.0"), ("d", "4.0")):
+        (dl / f"{name}-{ver}").write_bytes(b"x")
+    marked = tmp_path / "c"
+    marked.mkdir()
+    (marked / ".esphome_extracted").touch()
+    specs = [
+        _spec("a", "1.0", tmp_path / "a"),
+        _spec("b", "2.0", tmp_path / "b", mirrors=["http://m"]),
+        _spec("c", "3.0", marked),
+        _spec("d", "4.0", tmp_path / "d"),
+    ]
+    with patch.object(registry, "install_package") as mock_install:
+        registry.install_packages(specs, dl)
+    sequential = [
+        c for c in mock_install.call_args_list if "extract_progress" not in c[1]
+    ]
+    batched = [c for c in mock_install.call_args_list if "extract_progress" in c[1]]
+    assert sorted(c[0][0] for c in sequential) == ["b", "c"]
+    assert sorted(c[0][0] for c in batched) == ["a", "d"]
+
+
+def test_install_packages_first_failure_reraised(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Installs are mandatory: the first failure propagates, extras are logged."""
+    dl = tmp_path / "dl"
+    dl.mkdir()
+    (dl / "a-1.0").write_bytes(b"x")
+    (dl / "b-2.0").write_bytes(b"y")
+    boom = EsphomeError("bad layout")
+
+    def _fail(name: str, *_a, **_kw) -> None:
+        raise boom if name == "a" else EsphomeError("also bad")
+
+    with (
+        patch.object(registry, "install_package", side_effect=_fail),
+        pytest.raises(EsphomeError),
+    ):
+        registry.install_packages(
+            [_spec("a", "1.0", tmp_path / "a"), _spec("b", "2.0", tmp_path / "b")], dl
+        )
+    # Every failure is named, including the re-raised one: its exception
+    # message may not identify the package
+    assert "Could not install a" in caplog.text
+    assert "Could not install b" in caplog.text
+
+
+@contextmanager
+def _batched_install(tmp_path: Path, extract_progress, prefill_archive: bool = True):
+    """Run a batched install_package of pkg@1.0.0; yields the download mock."""
+    dest = tmp_path / "pkg"
+    if prefill_archive:
+        (tmp_path / "dl").mkdir()
+        (tmp_path / "dl" / "pkg-1.0.0").write_bytes(b"x")
+    with (
+        patch.object(registry, "download_with_resume") as mock_download,
+        patch.object(registry, "archive_extract_all") as mock_extract,
+        patch.object(
+            registry,
+            "registry_download",
+            return_value=("http://x/pkg.tar.gz", "abc123", 42),
+        ),
+    ):
+        mock_extract.side_effect = lambda *_a, **_kw: (dest / "payload").mkdir(
+            parents=True
+        )
+        registry.install_package(
+            "pkg",
+            "1.0.0",
+            dest,
+            [],
+            tmp_path / "dl",
+            expect=("payload",),
+            extract_progress=extract_progress,
+        )
+        yield mock_download, mock_extract
+
+
+def test_install_package_extract_progress_suppresses_bars(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A batched install routes extraction fractions to the caller and keeps
+    both private bars and per-package INFO lines off the shared bar."""
+    fractions: list[float] = []
+    with (
+        caplog.at_level(logging.INFO),
+        _batched_install(tmp_path, fractions.append) as (mock_download, mock_extract),
+    ):
+        pass
+    assert mock_extract.call_args[1]["progress"] == fractions.append
+    # The download tracker reports zero bytes, keeping the shared bar honest
+    download_progress = mock_download.call_args[1]["progress"]
+    assert callable(download_progress)
+    download_progress(42)
+    assert fractions == [0.0]
+    assert "Downloading pkg" not in caplog.text
+    assert "Extracting pkg" not in caplog.text
+
+
+def test_install_package_batched_missing_archive_keeps_info_log(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A batched archive that unexpectedly needs a real download keeps the
+    INFO line; the shared bar shows no progress for it."""
+    with (
+        caplog.at_level(logging.INFO),
+        _batched_install(tmp_path, lambda _frac: None, prefill_archive=False),
+    ):
+        pass
+    assert "Downloading pkg 1.0.0" in caplog.text
+
+
+def test_install_packages_dedupes_duplicate_specs(tmp_path: Path) -> None:
+    """Duplicate (name, version) entries share one archive and would race
+    each other; the duplicate takes the sequential path."""
+    dl = tmp_path / "dl"
+    dl.mkdir()
+    (dl / "a-1.0").write_bytes(b"x")
+    (dl / "b-2.0").write_bytes(b"y")
+    specs = [
+        _spec("a", "1.0", tmp_path / "a"),
+        _spec("a", "1.0", tmp_path / "a2"),
+        _spec("b", "2.0", tmp_path / "b"),
+    ]
+    with patch.object(registry, "install_package") as mock_install:
+        registry.install_packages(specs, dl)
+    sequential = [
+        c for c in mock_install.call_args_list if "extract_progress" not in c[1]
+    ]
+    batched = [c for c in mock_install.call_args_list if "extract_progress" in c[1]]
+    assert [(c[0][0], c[0][2]) for c in sequential] == [("a", tmp_path / "a2")]
+    assert sorted(c[0][0] for c in batched) == ["a", "b"]
+    # The duplicate runs after the batch, which unlinks their shared archive
+    assert mock_install.call_args_list[-1] == sequential[0]
+
+
+def test_install_packages_caps_workers(tmp_path: Path) -> None:
+    """A high core count is capped; the workers share one disk."""
+    dl = tmp_path / "dl"
+    dl.mkdir()
+    specs = []
+    for i in range(12):
+        (dl / f"p{i}-1.0").write_bytes(b"x")
+        specs.append(_spec(f"p{i}", "1.0", tmp_path / f"p{i}"))
+    with (
+        patch.object(registry, "get_usable_cpu_count", return_value=64),
+        patch.object(registry, "run_batch_downloads", return_value=[]) as batch,
+        patch.object(registry, "install_package"),
+    ):
+        registry.install_packages(specs, dl)
+    assert batch.call_args.kwargs["max_workers"] == 10
+
+
+def test_install_package_batched_refetch_announced_once(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A batched archive that fails verification and refetches is announced;
+    a verify no-op (full size credited immediately) stays silent."""
+    with (
+        caplog.at_level(logging.INFO),
+        _batched_install(tmp_path, lambda _frac: None) as (mock_download, _),
+    ):
+        progress = mock_download.call_args[1]["progress"]
+        progress(42)
+        assert "Re-downloading pkg 1.0.0" not in caplog.text
+        progress(10)
+        progress(20)
+    assert caplog.text.count("Re-downloading pkg 1.0.0") == 1
+
+
+def test_install_package_batched_refetch_announced_without_size(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A size-less registry entry still announces its refetch on the first
+    streaming tick."""
+    with (
+        caplog.at_level(logging.INFO),
+        patch.object(registry, "download_with_resume") as mock_download,
+        patch.object(registry, "archive_extract_all") as mock_extract,
+        patch.object(
+            registry,
+            "registry_download",
+            return_value=("http://x/pkg.tar.gz", "abc123", None),
+        ),
+    ):
+        dest = tmp_path / "pkg"
+        (tmp_path / "dl").mkdir()
+        (tmp_path / "dl" / "pkg-1.0.0").write_bytes(b"x")
+        mock_extract.side_effect = lambda *_a, **_kw: (dest / "payload").mkdir(
+            parents=True
+        )
+        registry.install_package(
+            "pkg",
+            "1.0.0",
+            dest,
+            [],
+            tmp_path / "dl",
+            expect=("payload",),
+            extract_progress=lambda _frac: None,
+        )
+        progress = mock_download.call_args[1]["progress"]
+        # A verify no-op credits the whole (nonempty) file in one tick
+        progress(1)
+        assert "Re-downloading pkg 1.0.0" not in caplog.text
+        progress(0)
+    assert "Re-downloading pkg 1.0.0" in caplog.text