[core] Install prefetched PlatformIO packages with parallel extraction (#18775)

2026-08-28 16:53:28 +00:00 · 2026-08-27 03:02:02 +00:00
parent e55a8aeabe
commit b99e7f5ae2
6 changed files with 1288 additions and 178 deletions
@@ -55,6 +55,7 @@ from esphome.helpers import (
    cpp_string_escape,
    fnv1a_32bit_hash,
    get_str_env,
+    get_usable_cpu_count,
    walk_files,
 )
 from esphome.types import ConfigType
@@ -205,16 +206,6 @@ def valid_project_name(value: str):
    return value


-def get_usable_cpu_count() -> int:
-    """Return the number of CPUs that can be used for processes.
-    On Python 3.13+ this is the number of CPUs that can be used for processes.
-    On older Python versions this is the number of CPUs.
-    """
-    return (
-        os.process_cpu_count() if hasattr(os, "process_cpu_count") else os.cpu_count()
-    )
-
-
 if "ESPHOME_DEFAULT_COMPILE_PROCESS_LIMIT" in os.environ:
    _compile_process_limit_default = min(
        int(os.environ["ESPHOME_DEFAULT_COMPILE_PROCESS_LIMIT"]), get_usable_cpu_count()
@@ -402,6 +402,15 @@ def sort_ip_addresses(address_list: list[str]) -> list[str]:
    return [socket.getnameinfo(r[4], socket.NI_NUMERICHOST)[0] for r in res]


+def get_usable_cpu_count() -> int:
+    """Return the number of CPUs usable by this process (affinity-aware
+    on Python 3.13+); 1 when the count is undeterminable."""
+    count = (
+        os.process_cpu_count() if hasattr(os, "process_cpu_count") else os.cpu_count()
+    )
+    return count or 1
+
+
 def get_bool_env(var, default=False):
    """Read a boolean env var: the ``cv.boolean`` spellings plus ``1``/``0``;
    anything else falls through to ``bool(value)``."""
@@ -1,29 +1,35 @@
-"""Parallel prefetch of the packages a PlatformIO run would install.
+"""Parallel prefetch and install of the packages a PlatformIO run needs.

 Downloads the archives concurrently into PlatformIO's own download cache
-(identical ``compute_download_path`` keys) so the serial installer finds
-them already cached. Runs in a subprocess like all PlatformIO execution:
-loading a platform executes its code (pioarduino's penv setup rewrites
-``sys.path``). A sentinel in the build dir lets warm builds skip the
-spawn. Best-effort: any failure logs and PlatformIO downloads as before.
-Across processes sharing a core dir every download destination is
-serialized by a file lock; checksum-less URL downloads additionally
-stage under a stable name and promote with an atomic rename.
+(identical ``compute_download_path`` keys), then installs them through
+PlatformIO's own ``_install`` with one worker per usable core, so
+extraction (the serial, single-core half of a cold install) parallelizes
+too and ``pio run`` finds every package already installed. Runs in a
+subprocess like all PlatformIO execution: loading a platform executes
+its code (pioarduino's penv setup rewrites ``sys.path``). A sentinel in
+the build dir lets warm builds skip the spawn. Best-effort: any failure
+logs and PlatformIO downloads and installs as before. Across processes
+sharing a core dir every download destination is serialized by a file
+lock; checksum-less URL downloads additionally stage under a stable
+name and promote with an atomic rename.
 """

 from __future__ import annotations

 from concurrent.futures import ThreadPoolExecutor
+from contextlib import suppress
 import hashlib
 import json
 import logging
 import os
 from pathlib import Path
+from queue import SimpleQueue
+import signal
 import subprocess
 import sys
 import threading
 import time
-from typing import Any
+from typing import Any, NamedTuple

 from esphome.framework_helpers import (
    content_length,
@@ -33,7 +39,7 @@ from esphome.framework_helpers import (
    run_batch_downloads,
    warn_prefetch_failures,
 )
-from esphome.helpers import get_bool_env
+from esphome.helpers import get_bool_env, get_usable_cpu_count, rmtree

 _LOGGER = logging.getLogger(__name__)

@@ -78,6 +84,18 @@ def _sweep_stale_sidecars(download_dir: Path, expire_seconds: int) -> None:
        _LOGGER.debug("Could not sweep %s", download_dir, exc_info=True)


+class _Resolved(NamedTuple):
+    """A registry spec resolved to its archive; ``cached`` skips the download."""
+
+    spec: Any
+    name: str
+    size: int
+    url: str
+    dl_path: Path
+    checksum: str
+    cached: bool
+
+
 # Child records a no-work run; the parent skips the next spawn while valid
 _SENTINEL_NAME = ".esphome_prefetch.json"
 _SENTINEL_SCHEMA = 1
@@ -151,24 +169,75 @@ def prefetch_platformio_packages() -> None:
        CORE.name,
    ]
    try:
-        proc = subprocess.run(cmd, env=env, check=False, timeout=_PREFETCH_TIMEOUT)
-    except subprocess.TimeoutExpired:
-        _LOGGER.warning("PlatformIO package prefetch timed out; continuing without it")
-        return
+        # Not a with-block: the lifetime spans the wait/terminate arms
+        proc = subprocess.Popen(cmd, env=env)  # pylint: disable=consider-using-with
    except Exception as err:  # noqa: BLE001  # pylint: disable=broad-exception-caught
        # The prefetch must never become a new way for the build to fail
        _LOGGER.warning("PlatformIO package prefetch skipped: %s", failure_reason(err))
        _LOGGER.debug("Prefetch failure detail", exc_info=True)
        return
-    if proc.returncode == _EXIT_HANDLED:
+    try:
+        returncode = proc.wait(timeout=_PREFETCH_TIMEOUT)
+    except subprocess.TimeoutExpired:
+        _stop_child(proc)
+        _LOGGER.warning("PlatformIO package prefetch timed out; continuing without it")
+        return
+    except BaseException as err:
+        # SIGKILL (subprocess.run's choice on interrupt) could land inside
+        # a package-directory copy pio run would then trust; ask first
+        _stop_child(proc)
+        if isinstance(err, Exception):
+            # An unexpected wait() failure must degrade, not fail the build
+            _LOGGER.warning(
+                "PlatformIO package prefetch skipped: %s", failure_reason(err)
+            )
+            return
+        raise
+    if returncode == _EXIT_HANDLED:
        # The child already warned with the reason; a second line is noise
        _LOGGER.debug("Prefetch child reported a handled failure")
-    elif proc.returncode != 0:
+    elif returncode != 0:
        # Exit 1 stays here: the interpreter exits 1 for import/module
        # failures before main() ever runs, a wiring break worth a warning
-        _LOGGER.warning(
-            "PlatformIO package prefetch skipped (exit %d)", proc.returncode
-        )
+        _LOGGER.warning("PlatformIO package prefetch skipped (exit %d)", returncode)
+
+
+def _stop_child(proc: subprocess.Popen) -> None:
+    """Stop the child without cutting an in-flight package install short.
+
+    Wait first (a terminal interrupt already unwinds the child), then
+    SIGTERM for the clean unwind main() installs, then kill. On Windows
+    terminate() cannot reach the handler, so its arm is a plain wait.
+    """
+    if proc.poll() is None:
+        _LOGGER.info("Waiting for the prefetch child to finish its current install")
+    try:
+        with suppress(subprocess.TimeoutExpired):
+            proc.wait(timeout=5)
+            return
+        if sys.platform != "win32":
+            proc.terminate()
+        with suppress(subprocess.TimeoutExpired):
+            proc.wait(timeout=30)
+            return
+        proc.kill()
+        proc.wait(timeout=5)
+        # The kill can land mid-copy; the uncertainty must be visible
+        _LOGGER.warning("Prefetch child killed; a package install may be incomplete")
+    except KeyboardInterrupt:
+        # Kill so an interrupted stop cannot orphan a still-writing child
+        # (BaseException: a further interrupt must not skip the kill),
+        # then re-raise so the build aborts
+        with suppress(BaseException):
+            proc.kill()
+            proc.wait(timeout=5)
+        if proc.poll() is None:
+            _LOGGER.warning("The prefetch child could not be confirmed stopped")
+        raise
+    except Exception:  # noqa: BLE001  # pylint: disable=broad-exception-caught
+        # A surviving child may still be writing packages pio run trusts
+        _LOGGER.warning("The prefetch child could not be confirmed stopped")
+        _LOGGER.debug("Stop detail", exc_info=True)


 def _project_platform_and_config(ini: Path, env: str) -> tuple[str | None, Any]:
@@ -183,25 +252,36 @@ def _project_platform_and_config(ini: Path, env: str) -> tuple[str | None, Any]:
    return config.get(f"env:{env}", "platform", None), config


+def _sibling_manager(manager: Any) -> Any:
+    """A same-store manager equivalent to the shared one."""
+    # Hard read: a renamed attribute must fail loudly, not silently drop
+    # the qualifiers wave-1 installs resolve with; is-not-None so a falsy
+    # PackageCompatibility still propagates
+    if (compatibility := manager.compatibility) is not None:
+        return manager.__class__(manager.package_dir, compatibility=compatibility)
+    return manager.__class__(manager.package_dir)
+
+
 def _registry_jobs(
-    manager, specs, seen: set[str]
-) -> tuple[list[tuple[str, int, Any]], int]:
+    manager: Any, specs: list[Any], seen: set[str]
+) -> tuple[list[tuple[str, int, Any]], int, list[tuple[str, Any]]]:
    """Resolve registry specs to ``(name, size, fetch)`` batch jobs.

    Mirrors PlatformIO's install path: best version, systype file, first
    mirror, and the same sha1(url + checksum) download-cache key. Also
-    returns how many resolutions errored (a clean skip is not an error).
+    returns how many resolutions errored (a clean skip is not an error)
+    and the ``(name, spec)`` pairs whose archives will be installable.
    """
    from platformio.registry.mirror import RegistryFileMirrorIterator

    local = threading.local()
    errors: list[str] = []

-    def _resolve(spec) -> tuple[str, int, str, Path, str] | object | None:
+    def _resolve(spec) -> _Resolved | object | None:
        # One manager (and registry HTTP session) per worker thread;
        # installed-state was already checked on the shared manager
        if (mgr := getattr(local, "mgr", None)) is None:
-            mgr = local.mgr = manager.__class__()
+            mgr = local.mgr = _sibling_manager(manager)
        try:
            packages = mgr.search_registry_packages(spec)
            if not packages:
@@ -218,13 +298,13 @@ def _registry_jobs(
            url, checksum = next(RegistryFileMirrorIterator(pkgfile["download_url"]))
            checksum = checksum or pkgfile["checksum"]["sha256"]
            dl_path = Path(mgr.compute_download_path(url, checksum))
-            if dl_path.is_file():
-                return None  # cached from an earlier run
+            cached = dl_path.is_file()  # fetched by an earlier run
            size = pkgfile.get("size")
-            if not size:
+            if not cached and not size:
                _LOGGER.debug("%s has no size; PlatformIO fetches it", spec)
                return None  # no size, no bar share
-            return f"{package['name']}@{version['name']}", size, url, dl_path, checksum
+            name = f"{package['name']}@{version['name']}"
+            return _Resolved(spec, name, size or 0, url, dl_path, checksum, cached)
        except Exception as err:  # noqa: BLE001  # pylint: disable=broad-exception-caught
            # One flaky spec must not discard the rest of the batch
            _LOGGER.debug("Could not resolve %s", spec, exc_info=True)
@@ -239,20 +319,27 @@ def _registry_jobs(
            unique.setdefault((s.owner, s.name, str(s.requirements)), s)
    pending = list(unique.values())
    if not pending:
-        return [], 0
+        return [], 0, []
    # Serial resolutions (registry GET + mirror HEAD each) dominate
    with ThreadPoolExecutor(max_workers=min(_RESOLVE_WORKERS, len(pending))) as ex:
        results = list(ex.map(_resolve, pending))
    jobs: list[tuple[str, int, Any]] = []
+    installable: list[tuple[str, Any]] = []
    for res in results:
        if res is None or res is _RESOLVE_FAILED:
            continue
-        name, size, url, dl_path, checksum = res
-        if str(dl_path) in seen:
-            continue  # duplicate spec; two workers must not share a .part
-        seen.add(str(dl_path))
+        installable.append((res.name, res.spec))
+        if res.cached or str(res.dl_path) in seen:
+            continue  # already fetched, or a duplicate must not share a .part
+        seen.add(str(res.dl_path))
        jobs.append(
-            (name, size, _registry_fetch_job(manager, url, dl_path, checksum, size))
+            (
+                res.name,
+                res.size,
+                _registry_fetch_job(
+                    manager, res.url, res.dl_path, res.checksum, res.size
+                ),
+            )
        )
    if failed := len(errors):
        # Visible once per build, naming a cause so an API break does not
@@ -264,18 +351,22 @@ def _registry_jobs(
            len(pending),
            errors[0],
        )
-    return jobs, failed
+    return jobs, failed, installable


-def _uri_jobs(manager, specs, seen: set[str]) -> tuple[list[tuple[str, int, Any]], int]:
+def _uri_jobs(
+    manager: Any, specs: list[Any], seen: set[str]
+) -> tuple[list[tuple[str, int, Any]], int, list[tuple[str, Any]]]:
    """Jobs for direct-URL specs; a HEAD sizes each for the combined bar.

    Also returns how many HEAD probes errored (an absent length is not an
-    error).
+    error) and the ``(name, spec)`` pairs whose archives will be
+    installable.
    """
    from esphome.net_retry import fetch_with_retry, http_request

-    candidates: list[tuple[str, str, Path]] = []
+    candidates: list[tuple[str, str, Path, Any]] = []
+    installable: list[tuple[str, Any]] = []
    for spec in specs:
        url = spec.uri
        if not url or not url.startswith(("http://", "https://")):
@@ -284,12 +375,21 @@ def _uri_jobs(manager, specs, seen: set[str]) -> tuple[list[tuple[str, int, Any]
            continue  # bare-URL VCS spec; PlatformIO clones it
        if manager.get_package(spec):
            continue
+        name = spec.name or url.rsplit("/", 1)[-1]
        # PlatformIO downloads URL specs with no checksum
        dl_path = Path(manager.compute_download_path(url, ""))
-        if dl_path.is_file() or str(dl_path) in seen:
-            continue  # cached, or another spec already claimed this .part
+        if dl_path.is_file():
+            if spec.has_custom_name():
+                # Only a custom name (Foo=https://...) is the destination
+                # dir; a URI-derived name's destination comes from the
+                # archive manifest, so its dedupe key could collide with
+                # another name and race one directory. pio run installs it.
+                installable.append((name, spec))  # fetched by an earlier run
+            continue
+        if str(dl_path) in seen:
+            continue  # another spec already claimed this .part
        seen.add(str(dl_path))
-        candidates.append((spec.name, url, dl_path))
+        candidates.append((spec.name, url, dl_path, spec))

    errors: list[str] = []

@@ -314,16 +414,19 @@ def _uri_jobs(manager, specs, seen: set[str]) -> tuple[list[tuple[str, int, Any]
        return content_length(resp)

    if not candidates:
-        return [], 0
+        return [], 0, installable
    with ThreadPoolExecutor(max_workers=min(_RESOLVE_WORKERS, len(candidates))) as ex:
-        sizes = list(ex.map(_head_size, [url for _, url, _ in candidates]))
+        sizes = list(ex.map(_head_size, [url for _, url, _, _ in candidates]))
    jobs: list[tuple[str, int, Any]] = []
    failed = 0
-    for (name, url, dl_path), size in zip(candidates, sizes, strict=True):
+    for (name, url, dl_path, spec), size in zip(candidates, sizes, strict=True):
        if size < 0:
            failed += 1
        elif size:
            jobs.append((name, size, _uri_fetch_job(manager, url, dl_path, size)))
+            if spec.has_custom_name():
+                # See above: derived-name specs stay with pio run's installer
+                installable.append((name, spec))
        else:
            # Missing or unusable Content-Length; visible under -v
            _LOGGER.debug("%s reports no usable length; PlatformIO fetches it", url)
@@ -335,7 +438,7 @@ def _uri_jobs(manager, specs, seen: set[str]) -> tuple[list[tuple[str, int, Any]
            len(candidates),
            errors[0],
        )
-    return jobs, failed
+    return jobs, failed, installable


 def _serialized_fetch_job(
@@ -460,11 +563,233 @@ def _uri_fetch_job(manager: Any, url: str, dl_path: Path, size: int) -> Any:
    return run


+# (name, spec) from wave 1, (name, spec, compatibility) from dep waves
+_Entry = tuple[str, Any] | tuple[str, Any, Any]
+
+
+def _dependency_entries(
+    manager: Any, entries: list[_Entry], seen_names: set[str]
+) -> list[_Entry]:
+    """Registry dependencies of the installed entries, one per new name.
+
+    Mostly local manifest reads; the builtin probe walks installed
+    platforms (each may run platform code). Name-only platform libs stay
+    with pio run.
+    """
+
+    # Hard read: losing this filter would pre-install incompatible
+    # packages pio run then trusts
+    compatibility = manager.compatibility
+    # Tool managers have no builtin table; the contract test pins the name
+    is_builtin = getattr(manager, "is_builtin_lib", None)
+    deps: dict[str, Any] = {}
+    skipped = 0
+    for name, spec, *_ in entries:
+        try:
+            deps_of = _entry_dependencies(manager, spec, compatibility, is_builtin)
+        except Exception:  # noqa: BLE001  # pylint: disable=broad-exception-caught
+            # One unreadable manifest must not drop the group's whole wave
+            _LOGGER.debug("Skipping dependencies of %s", name, exc_info=True)
+            skipped += 1
+            continue
+        for key, entry in deps_of:
+            if key not in seen_names:
+                deps.setdefault(key, entry)
+    if skipped:
+        # Visible at default verbosity: a dropped subtree silently
+        # degrades the wave; per-entry detail stays at debug
+        _LOGGER.warning(
+            "Could not read dependencies of %d of %d package(s)",
+            skipped,
+            len(entries),
+        )
+    return list(deps.values())
+
+
+def _entry_dependencies(
+    manager: Any, spec: Any, compatibility: Any, is_builtin: Any
+) -> list[tuple[str, _Entry]]:
+    from platformio.package.meta import PackageCompatibility
+
+    out: list[tuple[str, _Entry]] = []
+    if (pkg := manager.get_package(spec)) is None:
+        # Only successful installs are walked, so this is a real anomaly
+        # (stale memcache, name/dir mismatch, a pio API change); raising
+        # folds it into the caller's aggregate dropped-subtree warning
+        raise RuntimeError(f"just-installed {spec} is not resolvable")
+    for dep in manager.get_pkg_dependencies(pkg) or []:
+        if not (dep.get("owner") or dep.get("version")):
+            continue
+        if compatibility and not PackageCompatibility.from_dependency(
+            dep
+        ).is_compatible(compatibility):
+            continue  # pio's install_dependency would skip it too
+        dspec = manager.dependency_to_spec(dep)
+        if (
+            is_builtin
+            and not dspec.owner
+            and not dspec.external
+            and is_builtin(dspec.name)
+        ):
+            # pio's LibraryPackageManager.install_dependency skips
+            # builtins; a registry copy would shadow the bundled one
+            continue
+        if not (key := (dspec.name or "").lower()):
+            _LOGGER.debug("Dependency %r of %s has no name; left to pio run", dep, spec)
+            continue
+        if manager.get_package(dspec) is not None:
+            continue  # already installed
+        # Carry the dep's compatibility so _install searches the
+        # registry qualified, exactly like pio's install_dependency
+        out.append(
+            (key, (dspec.name, dspec, PackageCompatibility.from_dependency(dep)))
+        )
+    return out
+
+
+def _clean_failed_install(mgr: Any, name: str, spec: Any) -> None:
+    # A post-copy failure leaves a package pio run would trust; remove it
+    # so pio run genuinely reinstalls it
+    try:
+        mgr.memcache_reset()
+        if (pkg := mgr.get_package(spec)) is not None:
+            # Dropping the metadata is the invariant: pio's own install
+            # overwrites a metadata-less dir, so a stuck tree cannot be
+            # trusted. The rmtree is best-effort tidiness.
+            (Path(pkg.path) / ".piopm").unlink(missing_ok=True)
+            with suppress(OSError):
+                rmtree(pkg.path)
+        else:
+            # Nothing was moved into place; the common failure shape
+            _LOGGER.debug("No on-disk install of %s to remove", name)
+    except Exception as cleanup_err:  # noqa: BLE001  # pylint: disable=broad-exception-caught
+        _LOGGER.warning(
+            "Could not remove the failed install of %s: %s",
+            name,
+            failure_reason(cleanup_err),
+        )
+
+
+def _preinstall(
+    manager: Any, entries: list[_Entry], seen_names: set[str] | None = None
+) -> None:
+    """Install downloaded packages in parallel via pio's own ``_install``.
+
+    ``entries`` are ``_Entry`` tuples, one per destination directory.
+    The lock is held around each wave's pool, safe only because pio's
+    private ``_install`` never re-acquires it (a same-process re-lock
+    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))
+    # One manager per worker (_install mutates instance state); built
+    # serially because construction rewires the shared manager logger
+    managers: SimpleQueue = SimpleQueue()
+    for _ in range(workers):
+        managers.put(_sibling_manager(manager))
+    local = threading.local()
+
+    def _install_one(entry) -> bool:
+        # Wave-1 entries are (name, spec); dependency waves add compatibility
+        name, spec, *rest = entry
+        compat = rest[0] if rest else None
+        if (mgr := getattr(local, "mgr", None)) is None:
+            # at most `workers` pool threads, one dequeue each
+            mgr = local.mgr = managers.get_nowait()
+        try:
+            mgr._install(  # pylint: disable=protected-access  # noqa: SLF001
+                spec, skip_dependencies=True, compatibility=compat
+            )
+            return True
+        except Exception as err:  # noqa: BLE001  # pylint: disable=broad-exception-caught
+            _LOGGER.warning("Could not pre-install %s: %s", name, failure_reason(err))
+            _LOGGER.debug("Pre-install failure detail", exc_info=True)
+            _clean_failed_install(mgr, name, spec)
+            return False
+        except BaseException:
+            # A SystemExit from a postinstall must not skip the cleanup
+            # and leave a torn dir pio run trusts
+            _clean_failed_install(mgr, name, spec)
+            raise
+
+    _LOGGER.info(
+        "Installing %d PlatformIO package(s) with %d extraction worker(s): %s",
+        len(entries),
+        workers,
+        ", ".join(name for name, *_ in entries),
+    )
+    # Postinstall scripts chdir process-globally; the cwd is restored
+    # after the pool. Concurrent postinstalls can still race pio's
+    # non-reentrant fs.cd mid-pool; that install fails, warns, and is
+    # redone serially by pio run. Suppress interleaved progress bars.
+    os.environ.setdefault("PLATFORMIO_DISABLE_PROGRESSBAR", "true")
+    # get_tmp_dir/get_download_dir create without exist_ok; racing workers
+    # would FileExistsError, so create them serially first. Concurrent
+    # usage.db updates can drop download bookkeeping; never a bad build.
+    manager.get_tmp_dir()
+    manager.get_download_dir()
+    cwd = Path.cwd()
+    manager.lock()
+    try:
+        with ThreadPoolExecutor(max_workers=workers) as ex:
+            try:
+                results = list(ex.map(_install_one, entries))
+            except BaseException:
+                # Drop queued installs; in-flight ones finish so no
+                # package directory is left half copied
+                ex.shutdown(wait=True, cancel_futures=True)
+                raise
+    finally:
+        # Cleanup must not mask an in-flight exception or skip a step
+        # Each step runs even if an earlier one fails, and none may
+        # displace the in-flight exception (SIGTERM's SystemExit
+        # included) with a downgradeable one
+        wave_ok = True
+        for step, label in (
+            (manager.memcache_reset, "reset the storage cache"),
+            (manager.unlock, "release the manager lock"),
+            (lambda: os.chdir(cwd), "restore the working dir"),
+        ):
+            try:
+                step()
+            except Exception:  # noqa: BLE001,PERF203  # pylint: disable=broad-exception-caught
+                wave_ok = False
+                _LOGGER.warning("Could not %s", label)
+                _LOGGER.debug("Teardown detail", exc_info=True)
+    if len(entries) > 1 and not any(results):
+        # A systematic fault, not one bad archive; pio run installs serially
+        _LOGGER.warning(
+            "Could not pre-install any of %d PlatformIO package(s)", len(entries)
+        )
+
+    seen = seen_names if seen_names is not None else set()
+    # All entries join seen (failures must not be re-queued); only
+    # successful installs feed the dependency walk
+    seen.update(name.split("@", 1)[0].lower() for name, *_ in entries)
+    installed = [e for e, ok in zip(entries, results, strict=True) if ok]
+    if not wave_ok:
+        # A stale cache, an unknown lock state, or a lost cwd would
+        # poison the next wave; pio run installs the rest cleanly
+        _LOGGER.warning("Skipping the dependency wave")
+        return
+    # The builtin probe may construct platforms whose setup rewrites
+    # sys.path (see _prefetch); restore it for later imports
+    saved_sys_path = list(sys.path)
+    try:
+        next_entries = _dependency_entries(manager, installed, seen)
+    finally:
+        sys.path[:] = saved_sys_path
+    if next_entries:
+        # Terminates without a cap: every wave admits only never-seen
+        # names, so a cycle yields an empty next wave
+        _preinstall(manager, next_entries, seen)
+
+
 def _prefetch(build_dir: Path, env: str) -> None:
    from platformio.dependencies import get_core_dependencies
    from platformio.package.manager.library import LibraryPackageManager
    from platformio.package.manager.platform import PlatformPackageManager
-    from platformio.package.meta import PackageSpec
+    from platformio.package.meta import PackageCompatibility, PackageSpec
    from platformio.platform.factory import PlatformFactory

    platform_spec, config = _project_platform_and_config(
@@ -504,10 +829,16 @@ def _prefetch(build_dir: Path, env: str) -> None:
            )
        )
    lib_deps = config.get(f"env:{env}", "lib_deps", [])
-    # pio run's storage dir for this env: installed libraries skip by
-    # disk lookup
+    # pio run's storage dir for this env, with its compatibility
+    # qualifiers: an unqualified library install could land a different
+    # owner's package pio run would then trust
+    qualifiers: dict[str, Any] = {"platforms": [p.name]}
+    if framework := config.get(f"env:{env}", "framework", None):
+        qualifiers["frameworks"] = framework
    libdeps_dir = Path(config.get("platformio", "libdeps_dir")) / env
-    lm = LibraryPackageManager(str(libdeps_dir))
+    lm = LibraryPackageManager(
+        str(libdeps_dir), compatibility=PackageCompatibility(**qualifiers)
+    )
    # A bare name is usually a framework built-in (WiFi, SPI); with no
    # lib builders here to tell built-in from registry, skip it. The only
    # cost is that an owner-less user library is not prefetched
@@ -520,35 +851,71 @@ def _prefetch(build_dir: Path, env: str) -> None:

    seen: set[str] = set()
    jobs: list[tuple[str, int, Any]] = []
+    groups: list[tuple[Any, list[tuple[str, Any]]]] = []
    unresolved = 0
    for mgr, batch in ((p.pm, specs), (lm, lib_specs)):
+        entries: list[tuple[str, Any]] = []
        for build_jobs in (_registry_jobs, _uri_jobs):
-            batch_jobs, failed = build_jobs(mgr, batch, seen)
+            batch_jobs, failed, installable = build_jobs(mgr, batch, seen)
            jobs += batch_jobs
            unresolved += failed
+            entries += installable
+        if entries:
+            groups.append((mgr, entries))

    sentinel = build_dir / _SENTINEL_NAME
-    if not jobs:
-        if not unresolved:
-            # Record the no-work run so the parent skips the next spawn.
-            # A failed resolution is not "no work": a registry outage must
-            # not be cached as warm.
-            dirs = [config.get("platformio", "packages_dir")]
-            if lib_specs:
-                dirs.append(str(libdeps_dir))
-            sentinel.write_text(
-                json.dumps({**_sentinel_state(build_dir), "dirs": dirs}),
-                encoding="utf-8",
-            )
-        return
-    sentinel.unlink(missing_ok=True)
-    _LOGGER.info(
-        "Prefetching %d PlatformIO package(s): %s",
-        len(jobs),
-        ", ".join(name for name, _, _ in jobs),
-    )
-    # PlatformIO retries failed packages itself, without resume
-    warn_prefetch_failures(run_batch_downloads("Downloading PlatformIO packages", jobs))
+    if jobs or groups:
+        # Real work invalidates any previous no-work record
+        sentinel.unlink(missing_ok=True)
+    failed_names: set[str] = set()
+    if jobs:
+        _LOGGER.info(
+            "Prefetching %d PlatformIO package(s): %s",
+            len(jobs),
+            ", ".join(name for name, _, _ in jobs),
+        )
+        # PlatformIO retries failed packages itself, without resume
+        failures = run_batch_downloads("Downloading PlatformIO packages", jobs)
+        warn_prefetch_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.
+        # A failed resolution is not "no work": a registry outage must
+        # not be cached as warm.
+        dirs = [config.get("platformio", "packages_dir")]
+        if lib_specs:
+            dirs.append(str(libdeps_dir))
+        sentinel.write_text(
+            json.dumps({**_sentinel_state(build_dir), "dirs": dirs}),
+            encoding="utf-8",
+        )
+
+    for mgr, entries in groups:
+        # One install per destination: pio derives the directory from
+        # the package name, so key on the name part
+        to_install = {
+            name.split("@", 1)[0].lower(): (name, spec)
+            for name, spec in entries
+            if name not in failed_names
+        }
+        if to_install:
+            try:
+                _preinstall(mgr, list(to_install.values()))
+            except Exception as err:  # noqa: BLE001  # pylint: disable=broad-exception-caught
+                # Each group degrades independently; pio run installs
+                # whatever this one did not
+                _LOGGER.warning(
+                    "Pre-install failed for the %s group: %s",
+                    mgr.__class__.__name__,
+                    failure_reason(err),
+                )
+                _LOGGER.debug("Pre-install group failure detail", exc_info=True)
+
+
+def _sigterm(_signum, _frame) -> None:
+    # Raised in the main thread: the pool's BaseException arm cancels
+    # queued installs while in-flight copies finish, then finally runs
+    raise SystemExit(143)


 def main(argv: list[str]) -> int:
@@ -556,6 +923,7 @@ def main(argv: list[str]) -> int:
    from esphome.core import CORE
    from esphome.log import setup_log

+    signal.signal(signal.SIGTERM, _sigterm)
    raw_level = os.environ.get("ESPHOME_PREFETCH_LOG_LEVEL")
    try:
        level = int(raw_level) if raw_level is not None else logging.INFO
@@ -3,7 +3,6 @@
 from collections.abc import Callable
 import os
 from pathlib import Path
-import types
 from typing import Any
 from unittest.mock import MagicMock, Mock, patch

@@ -705,33 +704,6 @@ def test_include_file_with_c_header(
        assert '#include "c_library.h"' in mock_raw_statement.text


-def test_get_usable_cpu_count() -> None:
-    """Test get_usable_cpu_count returns CPU count."""
-    count = config.get_usable_cpu_count()
-    assert isinstance(count, int)
-    assert count > 0
-
-
-def test_get_usable_cpu_count_with_process_cpu_count() -> None:
-    """Test get_usable_cpu_count uses process_cpu_count when available."""
-    # Test with process_cpu_count (Python 3.13+)
-    # Create a mock os module with process_cpu_count
-
-    mock_os = types.SimpleNamespace(process_cpu_count=lambda: 8, cpu_count=lambda: 4)
-
-    with patch("esphome.core.config.os", mock_os):
-        # When process_cpu_count exists, it should be used
-        count = config.get_usable_cpu_count()
-        assert count == 8
-
-    # Test fallback to cpu_count when process_cpu_count not available
-    mock_os_no_process = types.SimpleNamespace(cpu_count=lambda: 4)
-
-    with patch("esphome.core.config.os", mock_os_no_process):
-        count = config.get_usable_cpu_count()
-        assert count == 4
-
-
 def test_list_target_platforms(tmp_path: Path) -> None:
    """Test _list_target_platforms returns available platforms."""
    # Create mock components directory structure
@@ -4,6 +4,7 @@ import os
 from pathlib import Path
 import socket
 import stat
+import types
 from unittest.mock import MagicMock, patch

 from aioesphomeapi.host_resolver import AddrInfo, IPv4Sockaddr, IPv6Sockaddr
@@ -1154,3 +1155,26 @@ def test_progressbar_interrupt_keeps_finished_bar_done(monkeypatch) -> None:
 def test_format_duration(seconds: float, expected: str) -> None:
    """Test that durations are rendered as short human-readable strings."""
    assert helpers.format_duration(seconds) == expected
+
+
+def test_get_usable_cpu_count() -> None:
+    """Returns a positive int on the real host."""
+    count = helpers.get_usable_cpu_count()
+    assert isinstance(count, int)
+    assert count > 0
+
+
+def test_get_usable_cpu_count_sources() -> None:
+    """Prefers process_cpu_count, falls back to cpu_count, degrades to 1."""
+    mock_os = types.SimpleNamespace(process_cpu_count=lambda: 8, cpu_count=lambda: 4)
+    with patch("esphome.helpers.os", mock_os):
+        assert helpers.get_usable_cpu_count() == 8
+
+    mock_os_no_process = types.SimpleNamespace(cpu_count=lambda: 4)
+    with patch("esphome.helpers.os", mock_os_no_process):
+        assert helpers.get_usable_cpu_count() == 4
+
+    # An undeterminable count degrades to one worker, never zero
+    mock_os_unknown = types.SimpleNamespace(cpu_count=lambda: None)
+    with patch("esphome.helpers.os", mock_os_unknown):
+        assert helpers.get_usable_cpu_count() == 1