[esp32] Extract prefetched ESP-IDF tool archives in parallel (#18841)

esphome/espidf/_tool_resolution.py

@@ -0,0 +1,67 @@
+"""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], None],
+) -> 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)`` is told
+    about a failing installed binary, and the tool counts 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:
+            on_broken(name, e)
+        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

@@ -25,6 +25,7 @@ from esphome.framework_helpers import (
     create_venv,
     download_and_extract,
     download_from_mirrors,
+    extract_workers,
     failure_reason,
     get_python_env_executable_path,
     get_system_python_path,
@@ -308,11 +309,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(),
@@ -320,11 +323,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,
     )
 
 
@@ -758,7 +770,7 @@ def _prefetch_idf_tool_archives(
     targets_str: str,
     tools: list[str],
     env: dict[str, str] | None,
-) -> None:
+) -> bool:
     """Pre-download the tool archives ``idf_tools.py install`` would fetch.
 
     ``idf_tools.py``'s own downloader restarts from byte zero on every retry,
@@ -774,6 +786,9 @@ def _prefetch_idf_tool_archives(
     ``idf_tools.py install`` to download whatever is missing exactly as
     before. Leftover ``.part`` files live in ``dist/`` and are removed by the
     post-install cache prune.
+
+    Returns whether every archive in the list verified here, which is what
+    lets the pre-extraction trust ``dist/``.
     """
     try:
         success, stdout, stderr = _run_idf_tools_script(
@@ -788,13 +803,13 @@ def _prefetch_idf_tool_archives(
                 "Could not determine ESP-IDF tool downloads: %s",
                 (stderr or "").strip(),
             )
-            return
+            return False
         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
@@ -813,10 +828,12 @@ def _prefetch_idf_tool_archives(
             seen_dests.add(entry["dest"])
             entries.append(entry)
         if not entries:
-            return
+            return False
+        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),
         )
 
@@ -848,11 +865,47 @@ def _prefetch_idf_tool_archives(
                 "Every ESP-IDF tool prefetch failed; the installer will "
                 "download without resume"
             )
+        # Any failed entry may have left an archive at its final name that
+        # nothing verified: a cached file that could not be re-hashed, or
+        # one whose removal failed. The pre-extraction trusts what it finds
+        return not failures
     except Exception as e:  # noqa: BLE001  # pylint: disable=broad-exception-caught
         # The installer downloads anything missing itself; never let the
         # prefetch become a new way for the install to fail.
         _LOGGER.warning("ESP-IDF tool prefetch failed: %s", failure_reason(e))
         _LOGGER.debug("Prefetch failure detail", exc_info=True)
+    return False
+
+
+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 = _run_idf_tools_script(
+            framework_path,
+            "install_tool_archives.py",
+            "ESP-IDF tool archive extraction",
+            args=[
+                targets_str,
+                str(extract_workers()),
+                *tools,
+            ],
+            env=env,
+            stream_output=True,
+        )[0]
+        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
+        _LOGGER.warning("ESP-IDF tool pre-extraction failed: %s", failure_reason(e))
+        _LOGGER.debug("Pre-extraction failure detail", exc_info=True)
 
 
 def _check_esphome_idf_framework_install(
@@ -1006,7 +1059,10 @@ def _check_esphome_idf_framework_install(
     if install:
         _LOGGER.info("Installing ESP-IDF %s framework ...", version)
         targets_str = ",".join(targets)
-        _prefetch_idf_tool_archives(framework_path, targets_str, tools, env)
+        if _prefetch_idf_tool_archives(framework_path, targets_str, tools, env):
+            # Only a prefetch that ran proves the archives in dist/ were
+            # verified this run
+            _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,35 @@ 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) -> None:
+        # 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)
 
-    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,141 @@
+"""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 typing import Any
+
+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]
+) -> dict[tuple[str, str], Any]:
+    """The {(name, version): tool} jobs whose verified archive is on disk."""
+    dist_path = Path(g.idf_tools_path) / "dist"
+
+    broken: set[str] = set()
+
+    def on_broken(name: str, e: ToolBinaryError) -> None:
+        # Repairing a broken installed binary is the installer's job, but
+        # the prefetch listed the tool, so it must still claim its filename
+        print(f"leaving broken {name} to the installer: {e}", file=sys.stderr)
+        broken.add(name)
+
+    pending: dict[tuple[str, str], Any] = {}
+    claimed: dict[str, str] = {}
+    for tool, name, version, download in iter_tool_downloads(
+        targets_csv, tool_specs, on_broken
+    ):
+        # Mirror the prefetch: an entry it could not verify is never trusted
+        if not (download.sha256 and download.size):
+            continue
+        dist_name = archive_name(download)
+        # Mirror the prefetch's dest dedupe: a name claimed by a different
+        # sha256 was verified for that entry only, not this one
+        if claimed.setdefault(dist_name, download.sha256) != download.sha256:
+            print(
+                f"leaving {name}@{version} to the installer: {dist_name} "
+                "holds a different archive",
+                file=sys.stderr,
+            )
+            continue
+        if name in broken:
+            continue
+        # Trusted as-is: the caller only runs this pass after a prefetch
+        # that verified every archive at its final name
+        if (dist_path / dist_name).is_file():
+            pending[(name, version)] = tool
+    return pending
+
+
+def install_one(tool: Any, name: str, version: str) -> bool | str:
+    """True when extracted, False when failed and cleaned, the dest path
+    when a torn dir survived cleanup; failures go back to the installer."""
+    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 dest
+        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 = collect_pending(targets_csv, tool_specs)
+    if len(pending) < 2:
+        # Nothing to parallelize; the installer takes them
+        print(
+            f"{len(pending)} prefetched tool archive(s); 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,
+    )
+    ex = ThreadPoolExecutor(max_workers=workers)
+    try:
+        futures = [
+            ex.submit(install_one, tool, name, version)
+            for (name, version), tool in pending.items()
+        ]
+        results = [future.result() for future in futures]
+    finally:
+        # Ctrl-C drops queued extractions; in-flight ones finish whole
+        ex.shutdown(wait=True, cancel_futures=True)
+    if failed := sum(result is not True for result in results):
+        # Nonzero exit makes the caller warn; the installer redoes these
+        print(f"{failed} of {len(results)} pre-extractions failed", file=sys.stderr)
+        # Repeated last so it is not lost in the streamed install output
+        for dest in (r for r in results if isinstance(r, str)):
+            print(
+                f"partial tool dir survived: {dest}; delete it manually if "
+                "the build fails",
+                file=sys.stderr,
+            )
+        sys.exit(1)
+
+
+main()

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,41 @@ _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 = ""
+    if os.environ.get("TEST_BROKEN_COLLIDE"):
+        # Test hook: a broken tool claims shared.tar.gz first; a later good
+        # tool shares the basename under a different sha256
+        first = _Tool(
+            {"6.0": _Version(_Download("https://gh.test/shared.tar.gz", 66, "f1"))},
+            "6.0",
+            broken=True,
+        )
+        first.name = "broken-first-tool"
+        shadowed = _Tool(
+            {"7.0": _Version(_Download("https://other.test/shared.tar.gz", 77, "f2"))},
+            "7.0",
+        )
+        shadowed.name = "shadowed-tool"
+        return {**_TOOLS, "broken-first-tool": first, "shadowed-tool": shadowed}
+    if os.environ.get("TEST_COLLIDE"):
+        # Test hook: a second tool whose archive shares cmake's basename
+        # but carries a different sha256
+        collide = _Tool(
+            {"5.0": _Version(_Download("https://other.test/cmake.tar.gz", 55, "ee"))},
+            "5.0",
+        )
+        collide.name = "collide-tool"
+        return {**_TOOLS, "collide-tool": collide}
     return _TOOLS
 
 

tests/unit_tests/test_espidf_framework.py

@@ -36,6 +36,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),
@@ -965,7 +967,7 @@ def test_prefetch_all_entries_unverifiable_is_a_noop(tmp_path: Path) -> None:
         patch("esphome.framework_helpers.download_with_resume") as download,
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
     ):
-        _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
+        assert not _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
     download.assert_not_called()
 
 
@@ -1001,7 +1003,7 @@ def test_prefetch_downloads_each_archive_with_resume(tmp_path: Path) -> None:
     ):
         # Materialize the lazy mock before threads race its first creation
         tracker = progress_cls.return_value.tracker.return_value
-        _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
+        assert _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
 
     dist = get_idf_tools_path() / "dist"
     # Archives download concurrently, so the call order is not fixed.
@@ -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(
@@ -1098,9 +1108,12 @@ def test_prefetch_failures_never_raise(
         ),
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
     ):
-        _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
+        ran = _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
 
     assert expected_log in caplog.text
+    # Nothing verified the archives end to end, so the pre-extraction must
+    # not trust dist/
+    assert not ran
 
 
 def test_prefetch_total_failure_logs_error(
@@ -1124,6 +1137,31 @@ def test_prefetch_total_failure_logs_error(
     assert "Every ESP-IDF tool prefetch failed" in caplog.text
 
 
+def test_prefetch_partial_failure_does_not_claim_dist_is_verified(
+    tmp_path: Path,
+) -> None:
+    """A failed entry may leave a cached archive nobody could re-hash or
+    remove behind; the pre-extraction would extract it unchecked."""
+
+    def _fail_ninja(url: str, *args, **kwargs) -> None:
+        if "ninja" in url:
+            raise OSError("still there, unverified")
+
+    with (
+        patch(
+            "esphome.espidf.framework.run_command",
+            return_value=(True, _PREFETCH_JSON, ""),
+        ),
+        patch(
+            "esphome.framework_helpers.download_with_resume",
+            side_effect=_fail_ninja,
+        ),
+        patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
+        patch("esphome.framework_helpers._BatchDownloadProgress"),
+    ):
+        assert not _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
+
+
 def test_prefetch_one_failed_archive_does_not_stop_the_rest(
     tmp_path: Path, caplog: pytest.LogCaptureFixture
 ) -> None:
@@ -1188,22 +1226,33 @@ 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"),
+            side_effect=lambda *a, **k: calls.append("prefetch") or True,
+        ),
+        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: (
@@ -1211,7 +1260,24 @@ def test_framework_install_prefetches_before_installer(
         )
         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")
+
+
+def test_framework_install_skips_preextraction_when_prefetch_did_not_run(
+    espidf_mocks: SimpleNamespace,
+) -> None:
+    """Without a prefetch nothing verified the archives in dist/, so the
+    installer alone decides what to trust."""
+    with (
+        patch(
+            "esphome.espidf.framework._prefetch_idf_tool_archives",
+            return_value=False,
+        ),
+        patch("esphome.espidf.framework._preinstall_idf_tool_archives") as preinstall,
+    ):
+        check_esp_idf_install(_IDF_VERSION, force=True)
+
+    preinstall.assert_not_called()
 
 
 # ---------------------------------------------------------------------------
@@ -1220,13 +1286,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 +1303,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 +1320,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 +1336,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 +1349,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 +1366,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)
 
 
@@ -2068,3 +2157,298 @@ 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.framework_helpers.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.framework_helpers.get_usable_cpu_count", return_value=64),
+    ):
+        _preinstall_idf_tool_archives(tmp_path, "esp32", ["required"], None)
+    from esphome.framework_helpers import BATCH_EXTRACT_WORKERS
+
+    assert run_script.call_args.kwargs["args"][1] == str(BATCH_EXTRACT_WORKERS)
+
+
+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.framework_helpers.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 DEBUG like the prefetch sibling."""
+    with (
+        caplog.at_level(logging.DEBUG),
+        patch(
+            "esphome.espidf.framework._run_idf_tools_script",
+            side_effect=TypeError("bad call"),
+        ),
+        patch("esphome.framework_helpers.get_usable_cpu_count", return_value=1),
+    ):
+        _preinstall_idf_tool_archives(tmp_path, "esp32", ["required"], None)
+    assert any("pre-extraction failed" in r.message for r in caplog.records)
+    detail = next(
+        r for r in caplog.records if r.message == "Pre-extraction failure detail"
+    )
+    assert detail.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_skips_colliding_dest(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """A second tool sharing a dist filename with a different sha256 was
+    never verified by the prefetch; it stays with the installer."""
+    _make_dist(tmp_path, "cmake.tar.gz", "ninja-v1.zip")
+    monkeypatch.setenv("TEST_COLLIDE", "1")
+    _run_espidf_script_inprocess(
+        tmp_path, monkeypatch, "install_tool_archives.py", "esp32", "8", "required"
+    )
+    tools = tmp_path / "tp" / "tools"
+    assert (tools / "cmake" / "3.30.2" / ".installed").is_file()
+    assert not (tools / "collide-tool").exists()
+    assert (
+        "leaving collide-tool@5.0 to the installer: cmake.tar.gz "
+        "holds a different archive" in capsys.readouterr().err
+    )
+
+
+def test_install_tool_archives_broken_tool_still_claims_its_dest(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """The prefetch verified the broken tool's archive, so a later tool
+    sharing its filename under another sha256 must not extract from it."""
+    _make_dist(tmp_path, "cmake.tar.gz", "ninja-v1.zip", "shared.tar.gz")
+    monkeypatch.setenv("TEST_BROKEN_COLLIDE", "1")
+    _run_espidf_script_inprocess(
+        tmp_path, monkeypatch, "install_tool_archives.py", "esp32", "8", "required"
+    )
+    tools = tmp_path / "tp" / "tools"
+    assert not (tools / "broken-first-tool").exists()
+    assert not (tools / "shadowed-tool").exists()
+    assert (
+        "leaving shadowed-tool@7.0 to the installer: shared.tar.gz "
+        "holds a different archive" in capsys.readouterr().err
+    )
+
+
+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 prefetched tool archive(s)" 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 removes the torn dest dir so the installer cannot
+    trust it, and the other tools still install; the nonzero exit lets the
+    caller point at the streamed detail."""
+    _make_dist(tmp_path, "cmake.tar.gz", "ninja-v1.zip")
+    monkeypatch.setenv("TEST_FAIL_INSTALL", "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
+    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
+    # The surviving dir is named again after the summary, where the caller's
+    # warning points
+    assert err.rstrip().endswith("delete it manually if the build fails")
+    assert "partial tool dir survived" 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 prefetched tool archive(s)" in (capsys.readouterr().out)