Merge branch 'nrf52-pch' into esp8266-native-parallel-extract

.github/workflows/ci-docker.yml

@@ -210,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 (nrf52).
+        # 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
@@ -156,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.
@@ -186,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/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/espidf/framework.py

@@ -34,7 +34,7 @@ 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
 
@@ -774,7 +774,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

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)

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/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/test_arduino8266_framework.py

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

tests/unit_tests/test_espidf_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_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