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

.github/workflows/ci-docker.yml

@@ -21,6 +21,7 @@ on:
       - "esphome/core/**"
       - "esphome/writer.py"
       - "esphome/build_gen/**"
+      - "esphome/build_helpers/**"
       - "esphome/espidf/**"
       - "esphome/platformio/**"
       - "esphome/components/bk72xx/**"
@@ -209,8 +210,7 @@ jobs:
           - nrf52
           - host
         # Strict by default so a new matrix id cannot silently join in the
-        # degrade-quietly mode the knob exists to catch; the knob is inert
-        # where no pch code runs (esp32-*-platformio, nrf52, host).
+        # degrade-quietly mode the knob exists to catch.
         # Opt-outs: libretiny GCC rejects its own pch until a toolchain bump.
         include:
           - id: bk72xx-arduino

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/build_gen/espidf.py

@@ -291,35 +291,10 @@ target_link_options(${{COMPONENT_LIB}} PUBLIC
 
 
 def _pch_cmake() -> str:
-    """The src component's precompiled-header block (C++ TUs only).
-
-    The -include stays relative (resolved from the compiler cwd, the build
-    dir); an absolute path would poison ccache keys.
-    """
-    if not pch_enabled():
-        return ""
-    # Strict inverts: a per-process consumer rejection reds the build.
-    # Baked at generation: a knob flip takes effect when the CMakeLists is
-    # rewritten (every esphome compile); a hand-run idf.py keeps the old one
-    escalation = pch.pch_consumer_escalation()
-    return f"""
-# ESPHome precompiled header (see esphome/build_helpers/pch.py).
-# OBJECT_DEPENDS is on the header, not the .gch: pch-baked headers drop
-# out of TU depfiles, and prepare_pch() touches the header on rebuild.
-target_compile_options(${{COMPONENT_LIB}} PRIVATE
-    "$<$<COMPILE_LANGUAGE:CXX>:-Winvalid-pch>"
-    "$<$<COMPILE_LANGUAGE:CXX>:{escalation}>"
-    "$<$<COMPILE_LANGUAGE:CXX>:-include>"
-    "$<$<COMPILE_LANGUAGE:CXX>:{PCH_HEADER_NAME}>"
-)
-set_source_files_properties(${{app_sources}} PROPERTIES
-    OBJECT_DEPENDS "${{CMAKE_BINARY_DIR}}/{PCH_HEADER_NAME}")
-"""
-
-
-def discard_pch() -> None:
-    """Drop the pch sidecars in the IDF build dir."""
-    pch.discard_pch(CORE.relative_build_path("build"))
+    """Consumer block for the component CMakeLists. Baked at generation:
+    a strict-knob flip takes effect on the next esphome compile; a
+    hand-run idf.py keeps the old one."""
+    return pch.pch_cmake_consumer("${COMPONENT_LIB}", "${app_sources}")
 
 
 def prepare_pch() -> None:

esphome/build_helpers/pch.py

@@ -7,7 +7,7 @@ it too; Arduino.h visibility there is intended (esphome#8693).
 
 from __future__ import annotations
 
-from collections.abc import Iterable
+from collections.abc import Callable, Iterable
 from contextlib import suppress
 from dataclasses import dataclass
 import hashlib
@@ -65,6 +65,10 @@ PCH_ARTIFACT_NAMES = (
 # The core defines header every backend anchors its prefix on.
 PCH_CORE_HEADER = "esphome/core/defines.h"
 
+# Guarded curated-prefix wrapper for PlatformIO backends without framework
+# force-includes (host, esp32); folded by the pch script via build_src_flags.
+PCH_PREFIX_HEADER = "esphome/core/pch_prefix.h"
+
 # Prefix-header contents for backends that inject a curated set (rather
 # than mirroring the TUs' own force-includes), defines.h first so USE_*
 # macros exist for the rest. Deliberately hard-coded: frequency-derived
@@ -152,6 +156,32 @@ def pch_consumer_escalation() -> str:
     return "-Werror=invalid-pch" if pch_strict() else "-Wno-error=invalid-pch"
 
 
+def pch_cmake_consumer(target: str, sources_var: str) -> str:
+    """Emit the CMake block making ``target``'s C++ sources consume the
+    pch; empty when disabled. OBJECT_DEPENDS is on the header, not the
+    .gch (pch-baked headers drop out of TU depfiles); the -include stays
+    relative — an absolute path would poison ccache keys."""
+    if not pch_enabled():
+        return ""
+    escalation = pch_consumer_escalation()
+    return f"""
+# ESPHome precompiled header (see esphome/build_helpers/pch.py).
+# The touch keeps OBJECT_DEPENDS satisfiable when the build system itself
+# wiped the build dir after the header was written (west --pristine)
+if(NOT EXISTS "${{CMAKE_BINARY_DIR}}/{PCH_HEADER_NAME}")
+  file(TOUCH "${{CMAKE_BINARY_DIR}}/{PCH_HEADER_NAME}")
+endif()
+target_compile_options({target} PRIVATE
+    "$<$<COMPILE_LANGUAGE:CXX>:-Winvalid-pch>"
+    "$<$<COMPILE_LANGUAGE:CXX>:{escalation}>"
+    "$<$<COMPILE_LANGUAGE:CXX>:-include>"
+    "$<$<COMPILE_LANGUAGE:CXX>:{PCH_HEADER_NAME}>"
+)
+set_source_files_properties({sources_var} PROPERTIES
+    OBJECT_DEPENDS "${{CMAKE_BINARY_DIR}}/{PCH_HEADER_NAME}")
+"""
+
+
 def ccache_pch_env() -> dict[str, str]:
     """Settings ccache needs to cache compiles that consume the .gch;
     empty unless this build actually emitted one. User-set values win.
@@ -182,6 +212,30 @@ def ccache_pch_env() -> dict[str, str]:
     return env
 
 
+def guarded_prepare(build_dir: Path, prepare: Callable[[], None]) -> None:
+    """Run a backend's pch preparation; an optional speedup must never
+    abort the build. Strict is read first so its own knob error cannot
+    mask the real failure; discard_pch raises if a stale .gch survives;
+    the header is ensured so OBJECT_DEPENDS stays satisfiable."""
+    try:
+        prepare()
+    except Exception:  # noqa: BLE001  # pylint: disable=broad-exception-caught
+        strict = pch_strict()
+        discard_pch(build_dir)
+        if strict:
+            raise
+        header = build_dir / PCH_HEADER_NAME
+        if not header.exists():
+            try:
+                header.touch()
+            except OSError as err:
+                # The coming OBJECT_DEPENDS error would hide the real cause
+                _LOGGER.warning("Could not create the pch placeholder: %s", err)
+        _LOGGER.warning(
+            "Precompiled header setup failed; compiling without it", exc_info=True
+        )
+
+
 def pch_extra_scripts() -> list[str]:
     """The extra_scripts entries a PlatformIO platform registers for the
     pch; empty when disabled (the script itself has no enable check)."""

esphome/components/esp32/__init__.py

@@ -10,6 +10,7 @@ import subprocess
 from typing import Any
 
 from esphome import yaml_util
+from esphome.build_helpers.pch import PCH_PREFIX_HEADER, pch_enabled, pch_extra_scripts
 import esphome.codegen as cg
 from esphome.components.const import CONF_ENABLE_OTA_DOWNGRADE_PROTECTION
 from esphome.config_helpers import filter_source_files_from_defines
@@ -57,6 +58,7 @@ from esphome.coroutine import CoroPriority, coroutine_with_priority
 from esphome.espidf.component import generate_idf_components
 import esphome.final_validate as fv
 from esphome.helpers import copy_file_if_changed, rmtree, write_file_if_changed
+from esphome.platformio.toolchain import copy_pch_script
 from esphome.schema_extractors import SCHEMA_EXTRACT, schema_extractor
 from esphome.types import ConfigType
 from esphome.writer import clean_build, clean_cmake_cache
@@ -2455,6 +2457,11 @@ async def to_code(config):
 
         cg.add_platformio_option("lib_ldf_mode", "off")
         cg.add_platformio_option("lib_compat_mode", "strict")
+        # CI-speed only: this toolchain is being dropped, so the pch gets
+        # the same curated prefix as host with no further investment
+        cg.add_platformio_option("extra_scripts", pch_extra_scripts())
+        if pch_enabled():
+            cg.add_platformio_option("build_src_flags", f"-include {PCH_PREFIX_HEADER}")
         cg.add_platformio_option("platform", conf[CONF_PLATFORM_VERSION])
         cg.add_platformio_option("board", config[CONF_BOARD])
         cg.add_platformio_option("board_upload.flash_size", config[CONF_FLASH_SIZE])
@@ -3372,6 +3379,8 @@ def _write_idf_component_yml():
 def copy_files():
     _write_sdkconfig()
     _write_idf_component_yml()
+    if not CORE.using_toolchain_esp_idf:
+        copy_pch_script()
 
     if "partitions.csv" not in CORE.data[KEY_ESP32][KEY_EXTRA_BUILD_FILES]:
         flash_size = CORE.data[KEY_ESP32][KEY_FLASH_SIZE]

esphome/components/host/__init__.py

@@ -1,3 +1,4 @@
+from esphome.build_helpers.pch import PCH_PREFIX_HEADER, pch_enabled, pch_extra_scripts
 import esphome.codegen as cg
 import esphome.config_validation as cv
 from esphome.const import (
@@ -10,7 +11,7 @@ from esphome.const import (
     ThreadModel,
 )
 from esphome.core import CORE
-from esphome.platformio.toolchain import copy_ccache_script
+from esphome.platformio.toolchain import copy_ccache_script, copy_pch_script
 from esphome.types import ConfigType
 
 from .const import KEY_HOST
@@ -55,9 +56,18 @@ async def to_code(config: ConfigType) -> None:
     cg.add_platformio_option("platform", "platformio/native")
     cg.add_platformio_option("lib_ldf_mode", "off")
     cg.add_platformio_option("lib_compat_mode", "strict")
-    cg.add_platformio_option("extra_scripts", ["pre:ccache.py"])
+    cg.add_platformio_option("extra_scripts", ["pre:ccache.py", *pch_extra_scripts()])
+    if pch_enabled():
+        # Curated prefix for the pch (the script folds it plus defines.h):
+        # host has no framework force-includes, and the per-TU cost is the
+        # STL closure behind the core headers. Measured -43% compile CPU.
+        # Gated so ESPHOME_PCH_ENABLE=0 restores the strict view. When the
+        # .gch fails to build or load, the force-include stays and every TU
+        # parses the closure as text: correct, but slower than no pch.
+        cg.add_platformio_option("build_src_flags", f"-include {PCH_PREFIX_HEADER}")
 
 
 # Called by writer.py
 def copy_files() -> None:
     copy_ccache_script()
+    copy_pch_script()

esphome/components/nrf52/__init__.py

@@ -1,13 +1,24 @@
 from __future__ import annotations
 
 import asyncio
+from functools import partial
 import logging
 from pathlib import Path
 import re
 import shutil
+import stat
 import subprocess
 
 from esphome import pins
+from esphome.build_helpers import pch
+from esphome.build_helpers.pch import (
+    PCH_DEFAULT_HEADERS,
+    PCH_HEADER_NAME,
+    mark_pch_emitted,
+    pch_cmake_consumer,
+    pch_enabled,
+    pch_header_text,
+)
 import esphome.codegen as cg
 from esphome.components.zephyr import (
     add_extra_script,
@@ -805,6 +816,9 @@ def _generate_cmake_lists() -> bool:
             ")",
         ]
 
+    if consumer := pch_cmake_consumer("app", "${APP_SOURCES}"):
+        lines += consumer.splitlines()
+
     if link_flags:
         lines += [
             "",
@@ -819,6 +833,66 @@ def _generate_cmake_lists() -> bool:
     )
 
 
+def _app_build_dir(build_dir: Path) -> Path:
+    """The CMake binary dir of the app image: sysbuild nests it in a
+    domain dir named after the app source dir. Probed on disk (the
+    non-sysbuild zephyr/ output dir has no CMakeCache.txt) so it stays
+    truthful mid-build, unlike an SDK-version check."""
+    sysbuild_app = build_dir / "zephyr"
+    try:
+        cache = (sysbuild_app / "CMakeCache.txt").stat()
+    except (FileNotFoundError, NotADirectoryError):
+        return build_dir
+    # Other stat errors propagate; is_file() would silently mislocate the pch
+    return sysbuild_app if stat.S_ISREG(cache.st_mode) else build_dir
+
+
+def _prepare_pch(app_dir: Path) -> None:
+    """Build the .gch between the cmake and compile phases of west."""
+    if not pch_enabled():
+        pch.discard_pch(app_dir)
+        pch.pch_disabled_degraded()
+        return
+    # First, so OBJECT_DEPENDS is satisfied even when the pch degrades
+    app_dir.mkdir(parents=True, exist_ok=True)
+    write_file_if_changed(
+        app_dir / PCH_HEADER_NAME, pch_header_text(PCH_DEFAULT_HEADERS)
+    )
+    # New layout first (Zephyr >= 3.4 nests under zephyr/); fixed candidates
+    # keep the .sum identity deterministic and skip walking generated/
+    generated = app_dir / "zephyr" / "include" / "generated"
+    autoconf = None
+    for candidate in (generated / "zephyr" / "autoconf.h", generated / "autoconf.h"):
+        if candidate.exists():
+            autoconf = candidate
+            break
+    if autoconf is None:
+        # Fail closed: autoconf.h is the .sum's Kconfig identity
+        _LOGGER.warning("No autoconf.h found; compiling without the pch")
+        pch.discard_pch(app_dir)
+        pch.pch_degraded("autoconf.h missing")
+        return
+    try:
+        autoconf_text = autoconf.read_text(encoding="utf-8")
+    except OSError as err:
+        _LOGGER.warning(
+            "Could not read %s; compiling without the pch: %s", autoconf, err
+        )
+        pch.discard_pch(app_dir)
+        pch.pch_degraded(f"autoconf unreadable: {err}")
+        return
+    pch.prepare_pch(
+        app_dir,
+        PCH_DEFAULT_HEADERS,
+        (
+            str(CORE.data[KEY_CORE][KEY_FRAMEWORK_VERSION]),
+            zephyr_data()[KEY_BOARD],
+            autoconf_text,
+            *get_project_compile_flags(),
+        ),
+    )
+
+
 def _copy_if_exists(src: Path, dst: Path) -> None:
     if src.is_file():
         shutil.copy2(src, dst)
@@ -871,6 +945,58 @@ def run_compile(args, config: ConfigType) -> bool:
         str(source_dir),
     ]
 
+    if pch_enabled():
+        # Consumers carry the -include; gate the ccache relaxation on it
+        # (Zephyr auto-enables ccache as the compiler launcher when found)
+        mark_pch_emitted()
+        env.update(pch.ccache_pch_env())
+
+        # Configure first so the .gch compiles from settled compile DB
+        # flags. Only when the app DB is missing: input changes wipe the
+        # build dir, so an existing DB is settled and --cmake-only would
+        # reconfigure for nothing.
+        prepare = True
+        app_dir = _app_build_dir(build_dir)
+        if not (app_dir / "compile_commands.json").is_file():
+            if not run_command_ok(
+                west_cmd + ["--cmake-only", "--", "-DCMAKE_EXPORT_COMPILE_COMMANDS=ON"],
+                env=env,
+                stream_output=True,
+                cwd=str(paths["framework_path"]),
+            ):
+                raise EsphomeError("nRF52 native build configure failed")
+            # The configure phase creates the sysbuild domain dir: re-resolve
+            app_dir = _app_build_dir(build_dir)
+            # kernel.h needs the build-time syscall headers; under sysbuild
+            # the target exists only in the app domain's ninja
+            if not run_command_ok(
+                [
+                    "cmake",
+                    "--build",
+                    str(app_dir),
+                    "--target",
+                    "zephyr_generated_headers",
+                ],
+                env=env,
+                stream_output=True,
+                cwd=str(paths["framework_path"]),
+            ):
+                # A pch-only prerequisite: degrade, let the real build report.
+                # Also skip the .gch compile: it would fail on the missing
+                # headers and latch .gch.failed until an identity change
+                _LOGGER.warning(
+                    "Zephyr header generation failed; compiling without the pch"
+                )
+                pch.discard_pch(app_dir)
+                pch.pch_degraded("zephyr_generated_headers failed")
+                prepare = False
+    else:
+        prepare = True
+        app_dir = _app_build_dir(build_dir)
+
+    if prepare:
+        pch.guarded_prepare(app_dir, partial(_prepare_pch, app_dir))
+
     if not run_command_ok(
         west_cmd,
         env=env,

esphome/core/__init__.py

@@ -1143,13 +1143,14 @@ class EsphomeCore:
     def add_platformio_option(
         self, key: str, value: str | list[str], *, replace: bool = False
     ) -> None:
-        """Set a platformio.ini option; list values append to an existing list
-        unless ``replace`` is True, which overwrites any existing value."""
+        """Set a platformio.ini option; values append to an existing list
+        (a string as one element) unless ``replace`` is True, which
+        overwrites any existing value."""
         new_val = value
         old_val = self.platformio_options.get(key)
         if not replace and isinstance(old_val, list):
-            assert isinstance(value, list)
-            new_val = old_val + value
+            # A user platformio_options string must merge, not assert
+            new_val = old_val + ([value] if isinstance(value, str) else value)
         self.platformio_options[key] = new_val
 
     def _get_variable_generator(self, id):

esphome/core/pch_prefix.h

@@ -0,0 +1,8 @@
+#pragma once
+// Curated precompiled-header prefix for backends that force-include it via
+// build_src_flags (the pch script folds it into the .gch). Guarded because
+// build_src_flags also reaches C and assembly src edges.
+#ifdef __cplusplus
+#include "esphome/core/application.h"
+#include "esphome/core/automation.h"
+#endif

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/espidf/toolchain.py

@@ -528,24 +528,11 @@ def run_compile(config, verbose: bool) -> int:
             return result.returncode
         _patch_memory_segments()
 
-    # After every reconfigure so compile_commands and sdkconfig are settled.
-    # An optional speedup must never abort the build
-    from esphome.build_gen.espidf import discard_pch, prepare_pch
+    # After every reconfigure so compile_commands and sdkconfig are settled
+    from esphome.build_gen.espidf import prepare_pch
+    from esphome.build_helpers.pch import guarded_prepare
 
-    try:
-        prepare_pch()
-    except Exception:  # noqa: BLE001  # pylint: disable=broad-exception-caught
-        from esphome.build_helpers.pch import pch_strict
-
-        # Strict first: its own knob error must not mask the real failure
-        strict = pch_strict()
-        # Raises itself if a stale .gch survives (silently wrong output)
-        discard_pch()
-        if strict:
-            raise
-        _LOGGER.warning(
-            "Precompiled header setup failed; compiling without it", exc_info=True
-        )
+    guarded_prepare(CORE.relative_build_path("build"), prepare_pch)
 
     # Build
     args = []

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)

esphome/writer.py

@@ -7,7 +7,7 @@ import re
 import time
 
 from esphome import loader
-from esphome.build_helpers.pch import PCH_ARTIFACT_NAMES
+from esphome.build_helpers.pch import PCH_ARTIFACT_NAMES, PCH_PREFIX_HEADER
 from esphome.compiled_config import save_compiled_config
 from esphome.config import iter_component_configs, iter_components
 from esphome.const import (
@@ -247,6 +247,7 @@ def copy_src_tree():
         Path(
             "esphome/core/ring_buffer.h"
         ),  # moved to components/ring_buffer/, removed in 2026.11.0
+        Path(PCH_PREFIX_HEADER),  # build machinery, not user API
     }
     include_l = []
     for target, _ in source_files_l:

script/determine-jobs.py

@@ -543,6 +543,11 @@ _SMOKE_HARNESS_TRIGGER_FILES = frozenset(
 
 ESP32_PLATFORMIO_TRIGGER_FILES = _SMOKE_HARNESS_TRIGGER_FILES | {
     "esphome/build_gen/platformio.py",
+    # The pch machinery the strict smoke job polices, and the modules it
+    # imports; the rest of build_helpers/ does not affect PlatformIO builds
+    "esphome/build_helpers/pch.py",
+    "esphome/build_helpers/ccache.py",
+    "esphome/build_helpers/idedata.py",
 }
 
 

tests/script/test_determine_jobs.py

@@ -1013,6 +1013,10 @@ _ESP32_PLATFORMIO_FULL_LIST_FILES = [
     # PlatformIO subsystem (path-prefix trigger) + build generator
     ["esphome/platformio/runner.py"],
     ["esphome/platformio/toolchain.py"],
+    # The pch modules are standalone triggers, not the whole build_helpers/
+    ["esphome/build_helpers/pch.py"],
+    ["esphome/build_helpers/ccache.py"],
+    ["esphome/build_helpers/idedata.py"],
     ["esphome/build_gen/platformio.py"],
     # Workflow / harness files
     ["script/test_build_components.py"],
@@ -1071,6 +1075,9 @@ def test_esp32_platformio_components_to_test_returns_full_list_on_infrastructure
         # Non-PlatformIO files in esphome/build_gen/ do NOT trigger the
         # full list -- only esphome/build_gen/platformio.py is a trigger.
         (["esphome/build_gen/espidf.py"], [], []),
+        # build_helpers modules the pch does not import are not triggers.
+        (["esphome/build_helpers/size_summary.py"], [], []),
+        (["esphome/build_helpers/ninja.py"], [], []),
         # Docs / unrelated files -> empty.
         (["README.md"], [], []),
         ([], [], []),

tests/unit_tests/build_helpers/test_pch.py

@@ -232,6 +232,51 @@ def test_pch_strict(
     assert pch.pch_strict() is expected
 
 
+def test_pch_cmake_consumer_substitutes_target_and_sources(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.delenv("ESPHOME_PCH_ENABLE", raising=False)
+    monkeypatch.delenv("ESPHOME_PCH_STRICT", raising=False)
+    block = pch.pch_cmake_consumer("app", "${APP_SOURCES}")
+    assert "target_compile_options(app PRIVATE" in block
+    assert '"$<$<COMPILE_LANGUAGE:CXX>:-Winvalid-pch>"' in block
+    assert "-Wno-error=invalid-pch" in block
+    assert '"$<$<COMPILE_LANGUAGE:CXX>:esphome_pch.h>"' in block
+    assert "set_source_files_properties(${APP_SOURCES} PROPERTIES" in block
+    assert 'OBJECT_DEPENDS "${CMAKE_BINARY_DIR}/esphome_pch.h"' in block
+    # Placeholder guard: survives a build-system-side pristine wipe
+    assert 'file(TOUCH "${CMAKE_BINARY_DIR}/esphome_pch.h")' in block
+
+
+def test_pch_cmake_consumer_strict_escalates(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.delenv("ESPHOME_PCH_ENABLE", raising=False)
+    monkeypatch.setenv("ESPHOME_PCH_STRICT", "1")
+    assert "-Werror=invalid-pch" in pch.pch_cmake_consumer("app", "${APP_SOURCES}")
+
+
+def test_pch_cmake_consumer_empty_when_disabled(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("ESPHOME_PCH_ENABLE", "0")
+    assert pch.pch_cmake_consumer("app", "${APP_SOURCES}") == ""
+
+
+def test_guarded_prepare_logs_placeholder_failure(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A failed placeholder touch must be traceable, not silent."""
+    monkeypatch.delenv("ESPHOME_PCH_STRICT", raising=False)
+
+    def boom() -> None:
+        raise RuntimeError("boom")
+
+    # Missing build dir: the touch raises and only warns
+    pch.guarded_prepare(tmp_path / "missing", boom)
+    assert "Could not create the pch placeholder" in caplog.text
+
+
 def test_pch_degraded_raises_only_in_strict(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:

tests/unit_tests/components/esp32/test_pch_wiring.py

@@ -0,0 +1,43 @@
+"""The pch script must reach PlatformIO builds only; the native ESP-IDF
+toolchain has its own pch flow in build_gen/espidf.py."""
+
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from esphome.components import esp32
+from esphome.const import Toolchain
+from esphome.core import CORE
+
+
+@pytest.mark.parametrize(
+    ("toolchain", "copied"),
+    [(Toolchain.PLATFORMIO, True), (Toolchain.ESP_IDF, False)],
+)
+def test_copy_files_gates_pch_script_on_toolchain(
+    toolchain: Toolchain, copied: bool, tmp_path: Path
+) -> None:
+    CORE.toolchain = toolchain
+    CORE.build_path = tmp_path
+    from esphome.components.esp32 import (
+        KEY_ESP32,
+        KEY_EXTRA_BUILD_FILES,
+        KEY_FLASH_SIZE,
+    )
+
+    with (
+        patch.object(esp32, "_write_sdkconfig") as write_sdkconfig,
+        patch.object(esp32, "_write_idf_component_yml"),
+        patch.object(esp32, "copy_pch_script") as copy_script,
+        patch.object(esp32, "write_file_if_changed"),
+        patch.object(esp32, "get_partition_csv"),
+        patch.dict(
+            CORE.data,
+            {KEY_ESP32: {KEY_EXTRA_BUILD_FILES: {}, KEY_FLASH_SIZE: "4MB"}},
+        ),
+    ):
+        esp32.copy_files()
+    # Proves execution reached (and passed) the gate on both rows
+    assert write_sdkconfig.called
+    assert copy_script.called is copied

tests/unit_tests/components/nrf52/test_pch.py

@@ -0,0 +1,322 @@
+"""nrf52 sdk-nrf pch wiring: the CMake consumer block, the prepare wrapper,
+and the two-phase west split in run_compile."""
+
+from collections.abc import Generator
+from pathlib import Path
+from unittest.mock import Mock, patch
+
+import pytest
+
+from esphome.components import nrf52
+from esphome.components.zephyr.const import KEY_BOARD
+from esphome.const import KEY_CORE, KEY_FRAMEWORK_VERSION, Toolchain
+from esphome.core import CORE, EsphomeError
+
+
+@pytest.fixture(autouse=True)
+def pch_env(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("ESPHOME_PCH_ENABLE", raising=False)
+    monkeypatch.delenv("ESPHOME_PCH_STRICT", raising=False)
+
+
+@pytest.fixture
+def build_dir(tmp_path: Path) -> Path:
+    d = tmp_path / "build" / ".pioenvs" / "livingroom"
+    d.mkdir(parents=True)
+    return d
+
+
+_AUTOCONF_TEXT = "#define CONFIG_GPIO 1\n"
+
+
+def _write_autoconf(build_dir: Path) -> Path:
+    autoconf = build_dir / "zephyr" / "include" / "generated" / "zephyr" / "autoconf.h"
+    autoconf.parent.mkdir(parents=True)
+    autoconf.write_text(_AUTOCONF_TEXT)
+    return autoconf
+
+
+def test_prepare_pch_disabled_discards_and_degrades(
+    monkeypatch: pytest.MonkeyPatch, build_dir: Path
+) -> None:
+    monkeypatch.setenv("ESPHOME_PCH_ENABLE", "0")
+    gch = build_dir / "esphome_pch.h.gch"
+    gch.write_bytes(b"x")
+    nrf52._prepare_pch(build_dir)
+    assert not gch.exists()
+
+
+def test_prepare_pch_disabled_strict_raises(
+    monkeypatch: pytest.MonkeyPatch, build_dir: Path
+) -> None:
+    monkeypatch.setenv("ESPHOME_PCH_ENABLE", "0")
+    monkeypatch.setenv("ESPHOME_PCH_STRICT", "1")
+    with pytest.raises(EsphomeError, match="ESPHOME_PCH_STRICT"):
+        nrf52._prepare_pch(build_dir)
+
+
+def test_prepare_pch_missing_autoconf_degrades(
+    build_dir: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    with patch.object(nrf52.pch, "prepare_pch") as prepare:
+        nrf52._prepare_pch(build_dir)
+    assert not prepare.called
+    assert "No autoconf.h found" in caplog.text
+    # The header is written first so OBJECT_DEPENDS stays satisfied
+    assert (build_dir / "esphome_pch.h").is_file()
+
+
+def test_prepare_pch_missing_autoconf_strict_raises(
+    monkeypatch: pytest.MonkeyPatch, build_dir: Path
+) -> None:
+    monkeypatch.setenv("ESPHOME_PCH_STRICT", "1")
+    with pytest.raises(EsphomeError, match="autoconf.h missing"):
+        nrf52._prepare_pch(build_dir)
+
+
+def test_prepare_pch_unreadable_autoconf_fails_closed(
+    build_dir: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    # A directory named autoconf.h: read_text raises OSError
+    autoconf = build_dir / "zephyr" / "include" / "generated" / "autoconf.h"
+    autoconf.mkdir(parents=True)
+    with patch.object(nrf52.pch, "prepare_pch") as prepare:
+        nrf52._prepare_pch(build_dir)
+    assert not prepare.called
+    assert "Could not read" in caplog.text
+
+
+def test_app_build_dir_sysbuild_layout(build_dir: Path) -> None:
+    app = build_dir / "zephyr"
+    app.mkdir()
+    (app / "CMakeCache.txt").write_text("")
+    assert nrf52._app_build_dir(build_dir) == app
+
+
+def test_app_build_dir_top_level_layout(build_dir: Path) -> None:
+    # Non-sysbuild: build_dir/zephyr is the Zephyr output dir, no cache
+    (build_dir / "zephyr").mkdir()
+    assert nrf52._app_build_dir(build_dir) == build_dir
+
+
+def test_app_build_dir_ignores_cache_directory(build_dir: Path) -> None:
+    (build_dir / "zephyr" / "CMakeCache.txt").mkdir(parents=True)
+    assert nrf52._app_build_dir(build_dir) == build_dir
+
+
+def test_app_build_dir_propagates_stat_errors(build_dir: Path) -> None:
+    # is_file() would swallow this and mislocate the pch
+    with (
+        patch.object(Path, "stat", side_effect=PermissionError("denied")),
+        pytest.raises(PermissionError),
+    ):
+        nrf52._app_build_dir(build_dir)
+
+
+def test_prepare_pch_extras_carry_build_identity(build_dir: Path) -> None:
+    _write_autoconf(build_dir)
+    with (
+        patch.dict(CORE.data, {KEY_CORE: {KEY_FRAMEWORK_VERSION: "2.9.2"}}),
+        patch.object(
+            nrf52, "zephyr_data", return_value={KEY_BOARD: "adafruit_feather"}
+        ),
+        patch.object(nrf52, "get_project_compile_flags", return_value=["-Os"]),
+        patch.object(nrf52.pch, "prepare_pch") as prepare,
+    ):
+        nrf52._prepare_pch(build_dir)
+    assert (build_dir / "esphome_pch.h").is_file()
+    (passed_dir, headers, extras) = prepare.call_args.args
+    assert passed_dir == build_dir
+    assert headers == nrf52.PCH_DEFAULT_HEADERS
+    assert list(extras) == ["2.9.2", "adafruit_feather", _AUTOCONF_TEXT, "-Os"]
+
+
+def _generate_cmake(tmp_path: Path) -> str:
+    CORE.config_path = tmp_path / "test.yaml"
+    CORE.build_path = tmp_path / "build"
+    CORE.name = "livingroom"
+    with (
+        patch(
+            "esphome.components.zephyr.library.generate_zephyr_modules",
+            return_value=[],
+        ),
+        patch.object(nrf52, "get_project_compile_flags", return_value=["-Os"]),
+        patch.object(nrf52, "get_project_link_flags", return_value=[]),
+    ):
+        nrf52._generate_cmake_lists()
+    return (tmp_path / "build" / "zephyr" / "CMakeLists.txt").read_text()
+
+
+def test_cmake_lists_include_pch_consumer_block(tmp_path: Path) -> None:
+    # Content contract is pinned by the shared pch_cmake_consumer tests;
+    # here only that the block reaches the generated CMakeLists
+    text = _generate_cmake(tmp_path)
+    assert "target_compile_options(app PRIVATE" in text
+    assert 'OBJECT_DEPENDS "${CMAKE_BINARY_DIR}/esphome_pch.h"' in text
+
+
+def test_cmake_lists_pch_block_disabled(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    monkeypatch.setenv("ESPHOME_PCH_ENABLE", "0")
+    assert "esphome_pch.h" not in _generate_cmake(tmp_path)
+
+
+CompileCtx = tuple[Mock, Mock, Path]
+
+
+class TestRunCompilePhases:
+    """The pch pre-build block in run_compile: header write, conditional
+    cmake-only phase, and the never-abort-the-build exception contract."""
+
+    @pytest.fixture
+    def compile_ctx(self, tmp_path: Path) -> Generator[CompileCtx, None, None]:
+        CORE.config_path = tmp_path / "test.yaml"
+        CORE.build_path = tmp_path / "build"
+        CORE.name = "livingroom"
+        CORE.toolchain = Toolchain.SDK_NRF
+        with (
+            patch.object(nrf52, "check_and_install"),
+            patch.object(nrf52, "_generate_cmake_lists", return_value=False),
+            patch.object(
+                nrf52,
+                "get_build_paths",
+                return_value={
+                    "python_executable": "python3",
+                    "framework_path": tmp_path,
+                },
+            ),
+            patch.object(nrf52, "get_build_env", return_value={}),
+            patch.object(nrf52, "zephyr_data", return_value={KEY_BOARD: "board"}),
+            patch.object(nrf52, "run_command_ok") as run_cmd,
+            patch.object(nrf52, "_prepare_pch") as prepare,
+        ):
+            yield run_cmd, prepare, CORE.relative_pioenvs_path(CORE.name)
+
+    def _run(self) -> None:
+        nrf52.run_compile(None, {})
+
+    def test_missing_db_runs_cmake_phase(self, compile_ctx: CompileCtx) -> None:
+        run_cmd, prepare, build_dir = compile_ctx
+        # cmake-only ok, generated headers ok, final build fails
+        results = iter([True, True, False])
+
+        def west(cmd, **kwargs):
+            # Phase 1 configures the sysbuild app domain
+            app = build_dir / "zephyr"
+            app.mkdir(parents=True, exist_ok=True)
+            (app / "CMakeCache.txt").write_text("")
+            return next(results)
+
+        run_cmd.side_effect = west
+        with pytest.raises(EsphomeError, match="nRF52 native build failed"):
+            self._run()
+        assert "--cmake-only" in run_cmd.call_args_list[0].args[0]
+        # Generated syscall headers are built in the app domain pre-pch
+        headers_cmd = run_cmd.call_args_list[1].args[0]
+        assert headers_cmd[:2] == ["cmake", "--build"]
+        assert str(build_dir / "zephyr") in headers_cmd
+        assert "zephyr_generated_headers" in headers_cmd
+        assert "--cmake-only" not in run_cmd.call_args_list[2].args[0]
+        # The pch is prepared in the app domain dir, not the sysbuild root
+        assert prepare.call_args.args[0] == build_dir / "zephyr"
+
+    def test_generated_headers_failure_degrades(
+        self, compile_ctx, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        run_cmd, prepare, _ = compile_ctx
+        # headers target fails, the real build still runs (and fails here)
+        run_cmd.side_effect = [True, False, False]
+        with pytest.raises(EsphomeError, match="nRF52 native build failed"):
+            self._run()
+        assert "Zephyr header generation failed" in caplog.text
+        # The doomed .gch compile is skipped: it would latch .gch.failed
+        assert not prepare.called
+
+    def test_generated_headers_failure_strict_raises(
+        self, monkeypatch: pytest.MonkeyPatch, compile_ctx: CompileCtx
+    ) -> None:
+        monkeypatch.setenv("ESPHOME_PCH_STRICT", "1")
+        run_cmd, prepare, _ = compile_ctx
+        run_cmd.side_effect = [True, False]
+        with pytest.raises(EsphomeError, match="ESPHOME_PCH_STRICT"):
+            self._run()
+        assert not prepare.called
+
+    def test_cmake_phase_failure_raises(self, compile_ctx: CompileCtx) -> None:
+        run_cmd, prepare, _ = compile_ctx
+        run_cmd.side_effect = [False]
+        with pytest.raises(EsphomeError, match="configure failed"):
+            self._run()
+        assert not prepare.called
+
+    def test_ccache_pch_env_reaches_west(self, compile_ctx: CompileCtx) -> None:
+        run_cmd, _, _ = compile_ctx
+        run_cmd.side_effect = [False]
+        # clear=True also drops ambient CCACHE_*/ESPHOME_PCH_* overrides
+        with (
+            patch.dict("os.environ", {}, clear=True),
+            pytest.raises(EsphomeError, match="configure failed"),
+        ):
+            self._run()
+        env = run_cmd.call_args.kwargs["env"]
+        assert env["CCACHE_PCH_EXTSUM"] == "true"
+        assert env["CCACHE_SLOPPINESS"] == "pch_defines,time_macros"
+
+    @pytest.mark.parametrize("sysbuild", [False, True])
+    def test_settled_db_skips_cmake_phase(
+        self, sysbuild: bool, compile_ctx: CompileCtx
+    ) -> None:
+        run_cmd, prepare, build_dir = compile_ctx
+        app = build_dir / "zephyr" if sysbuild else build_dir
+        app.mkdir(parents=True)
+        # A present top-level cache keeps the pristine wipe from dropping
+        # the DB; the app-dir cache is the sysbuild layout marker
+        (build_dir / "CMakeCache.txt").write_text("")
+        (app / "CMakeCache.txt").write_text("")
+        (app / "compile_commands.json").write_text("[]")
+        run_cmd.side_effect = [False]
+        with pytest.raises(EsphomeError, match="nRF52 native build failed"):
+            self._run()
+        assert run_cmd.call_count == 1
+        assert "--cmake-only" not in run_cmd.call_args.args[0]
+        assert prepare.call_args.args[0] == app
+
+    def test_disabled_skips_header_and_cmake_phase(
+        self, monkeypatch: pytest.MonkeyPatch, compile_ctx: CompileCtx
+    ) -> None:
+        monkeypatch.setenv("ESPHOME_PCH_ENABLE", "0")
+        run_cmd, prepare, build_dir = compile_ctx
+        run_cmd.side_effect = [False]
+        with pytest.raises(EsphomeError, match="nRF52 native build failed"):
+            self._run()
+        assert run_cmd.call_count == 1
+        assert not (build_dir / "esphome_pch.h").exists()
+        # The wrapper still runs: it discards stale sidecars and feeds strict
+        assert prepare.called
+
+    def test_prepare_failure_never_aborts_the_build(
+        self, compile_ctx, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        run_cmd, prepare, build_dir = compile_ctx
+        build_dir.mkdir(parents=True)
+        # Keep the pristine wipe from dropping the dir the fallback touches
+        (build_dir / "CMakeCache.txt").write_text("")
+        prepare.side_effect = RuntimeError("boom")
+        run_cmd.side_effect = [True, True, False]
+        with pytest.raises(EsphomeError, match="nRF52 native build failed"):
+            self._run()
+        assert run_cmd.call_count == 3
+        assert "Precompiled header setup failed" in caplog.text
+        # The fallback still satisfies OBJECT_DEPENDS
+        assert (build_dir / "esphome_pch.h").is_file()
+
+    def test_prepare_failure_strict_raises(
+        self, monkeypatch: pytest.MonkeyPatch, compile_ctx: CompileCtx
+    ) -> None:
+        monkeypatch.setenv("ESPHOME_PCH_STRICT", "1")
+        run_cmd, prepare, _ = compile_ctx
+        prepare.side_effect = RuntimeError("boom")
+        run_cmd.side_effect = [True, True]
+        with pytest.raises(RuntimeError, match="boom"):
+            self._run()

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_core.py

@@ -568,6 +568,16 @@ class TestEsphomeCore:
         target.config_path = Path("foo/config")
         return target
 
+    def test_add_platformio_option_merges_string_into_list(self, target) -> None:
+        """A user platformio_options string lands after a component's list
+        (FINAL priority) and must merge as one element, not assert."""
+        target.add_platformio_option("extra_scripts", ["post:pch.py"])
+        target.add_platformio_option("extra_scripts", "pre:mine.py")
+        assert target.platformio_options["extra_scripts"] == [
+            "post:pch.py",
+            "pre:mine.py",
+        ]
+
     def test_reset(self, target):
         """Call reset on target and compare to new instance"""
         other = core.EsphomeCore().__dict__

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_espidf_toolchain.py

@@ -684,7 +684,7 @@ def test_run_compile_aborts_when_stale_pch_survives_discard(
         patch.object(toolchain, "print_summary"),
         patch("esphome.build_gen.espidf.prepare_pch", side_effect=RuntimeError("boom")),
         patch(
-            "esphome.build_gen.espidf.discard_pch",
+            "esphome.build_helpers.pch.discard_pch",
             side_effect=EsphomeError("Could not discard the stale precompiled header"),
         ),
         pytest.raises(EsphomeError, match="Could not discard"),

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_pch_prefix.py

@@ -0,0 +1,21 @@
+"""The pch prefix must keep resolving; a rename would silently
+collapse the precompiled set to defines.h with strict CI still green."""
+
+from pathlib import Path
+import re
+
+from esphome.build_helpers.pch import PCH_PREFIX_HEADER
+
+REPO = Path(__file__).parents[2]
+
+
+def test_pch_prefix_resolves() -> None:
+    prefix = REPO / PCH_PREFIX_HEADER
+    assert prefix.is_file()
+    body = prefix.read_text()
+    includes = re.findall(r'#include "([^"]+)"', body)
+    assert includes, "prefix wrapper folds nothing"
+    for name in includes:
+        assert (REPO / name).is_file(), f"{name} does not resolve"
+    # The C guard is what keeps build_src_flags safe on C/assembly edges
+    assert "#ifdef __cplusplus" in body

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