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

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,
@@ -36,7 +38,7 @@ from esphome.framework_helpers import (
     tool_version_runs,
     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),
         )
 
@@ -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()

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_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
+    )