[core] Prefetch PlatformIO packages in parallel (#18769)

esphome/espidf/framework.py

@@ -2,7 +2,6 @@
 
 from collections.abc import Callable
 from ctypes.util import find_library
-from functools import partial
 import json
 import logging
 import os
@@ -24,16 +23,17 @@ from esphome.framework_helpers import (
     create_venv,
     download_and_extract,
     download_from_mirrors,
-    download_with_resume,
     failure_reason,
     get_python_env_executable_path,
     get_system_python_path,
+    resume_fetch_job,
     rmdir,
     run_batch_downloads,
     run_command,
     run_command_ok,
     str_to_lst_of_str,
     tool_version_runs,
+    warn_prefetch_failures,
 )
 from esphome.helpers import write_file_if_changed
 
@@ -686,18 +686,6 @@ def _patch_tools_json_demote_unused_tools(framework_path: Path) -> None:
     )
 
 
-def _download_tool(
-    dist_path: Path, entry: dict, tracker: Callable[[int], None]
-) -> None:
-    download_with_resume(
-        entry["url"],
-        dist_path / entry["dest"],
-        sha256=entry["sha256"],
-        size=entry["size"],
-        progress=tracker,
-    )
-
-
 def _prefetch_idf_tool_archives(
     framework_path: Path,
     targets_str: str,
@@ -775,15 +763,17 @@ def _prefetch_idf_tool_archives(
                 (
                     entry["name"],
                     entry["size"],
-                    partial(_download_tool, dist_path, entry),
+                    resume_fetch_job(
+                        entry["url"],
+                        dist_path / entry["dest"],
+                        sha256=entry["sha256"],
+                        size=entry["size"],
+                    ),
                 )
                 for entry in entries
             ],
         )
-        for name, e in failures:
-            # failure_reason: a message-less exception must not log blank
-            _LOGGER.warning("Could not prefetch %s: %s", name, failure_reason(e))
-            _LOGGER.debug("Prefetch failure detail", exc_info=e)
+        warn_prefetch_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/framework_helpers.py

@@ -701,7 +701,7 @@ def _write_download_meta(
         _LOGGER.debug("Could not update download metadata %s: %s", meta, e)
 
 
-def _content_length(resp: "requests.Response") -> int:
+def content_length(resp: "requests.Response") -> int:
     """Return the response's Content-Length, or 0 when absent or malformed.
 
     0 means "unknown", which downstream disables the progress bar and the
@@ -744,7 +744,7 @@ def _stream_response_to_file(
     """
     f.seek(offset)
     f.truncate(offset)
-    total_size = size or offset + _content_length(resp)
+    total_size = size or offset + content_length(resp)
     downloaded = offset
     own_bar: ProgressBar | None = None
     if progress is None:
@@ -909,6 +909,19 @@ def _part_path(dest: Path) -> Path:
     return dest.with_name(dest.name + ".part")
 
 
+def discard_partial_download(dest: Path) -> None:
+    """Remove ``dest`` and the resume sidecars of an abandoned download."""
+    part = _part_path(dest)
+    for stale in (dest, part, part.with_name(part.name + ".meta")):
+        try:
+            stale.unlink()
+        except FileNotFoundError:
+            continue
+        except OSError as err:
+            # The caller's cache is never pruned; leave a trace
+            _LOGGER.debug("Could not remove %s: %s", stale, err)
+
+
 def _cancellable_sleep(
     delay: float, progress: Callable[[int], None] | None, done: int
 ) -> None:
@@ -922,6 +935,31 @@ def _cancellable_sleep(
         time.sleep(min(0.5, remaining))
 
 
+def resume_fetch_job(
+    url: str, dest: PathType, **kwargs
+) -> Callable[[Callable[[int], None]], None]:
+    """A ``run_batch_downloads`` job callable wrapping ``download_with_resume``.
+
+    Forwards the runner's positional tracker as the ``progress`` keyword.
+    """
+
+    def fetch(tracker: Callable[[int], None]) -> None:
+        download_with_resume(url, dest, progress=tracker, **kwargs)
+
+    return fetch
+
+
+def warn_prefetch_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."""
+    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)
+
+
 def download_with_resume(
     url: str,
     dest: PathType,
@@ -1022,7 +1060,7 @@ def download_with_resume(
                         streamed = True
                         if offset == 0:
                             validator = _response_validator(resp)
-                            expected_total = _content_length(resp)
+                            expected_total = content_length(resp)
                             # Recorded so a later run can prove an If-Range
                             # resume of this part file safe.
                             _write_download_meta(meta, url, validator, expected_total)

esphome/platformio/library.py

@@ -35,6 +35,7 @@ from esphome.framework_helpers import (
     failure_reason,
     rmdir,
     run_batch_downloads,
+    warn_prefetch_failures,
 )
 
 _LOGGER = logging.getLogger(__name__)
@@ -977,14 +978,10 @@ def _prefetch_wave(
                 for c in components
             ],
         )
-        for name, err in failures:
-            # The sequential call below retries and raises the real error
-            _LOGGER.warning(
-                "Prefetch of %s failed (retrying sequentially): %s",
-                name,
-                failure_reason(err),
-            )
-            _LOGGER.debug("Prefetch failure detail", exc_info=err)
+        # The sequential call below retries and raises the real error
+        warn_prefetch_failures(
+            failures, "Prefetch of %s failed (retrying sequentially): %s"
+        )
     except Exception as err:  # noqa: BLE001  # pylint: disable=broad-exception-caught
         # Same policy as the ESP-IDF twin: the prefetch must never become a
         # new way for the build to fail

esphome/platformio/prefetch.py

@@ -0,0 +1,600 @@
+"""Parallel prefetch of the packages a PlatformIO run would install.
+
+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.
+"""
+
+from __future__ import annotations
+
+from concurrent.futures import ThreadPoolExecutor
+import hashlib
+import json
+import logging
+import os
+from pathlib import Path
+import subprocess
+import sys
+import threading
+import time
+from typing import Any
+
+from esphome.framework_helpers import (
+    content_length,
+    discard_partial_download,
+    failure_reason,
+    resume_fetch_job,
+    run_batch_downloads,
+    warn_prefetch_failures,
+)
+from esphome.helpers import get_bool_env
+
+_LOGGER = logging.getLogger(__name__)
+
+# Concurrent registry resolutions / HEAD probes (each is network-bound)
+_RESOLVE_WORKERS = 8
+
+# A hung child must not block the build; downloads resume on the next run
+_PREFETCH_TIMEOUT = 20 * 60
+
+# Waiting on another process's URL download; past this, leave it to pio
+_DOWNLOAD_LOCK_TIMEOUT = 60
+
+# Child exit for a handled, already-warned failure; 1 would collide with
+# the interpreter's own import-failure exit
+_EXIT_HANDLED = 3
+
+# Short lock-acquire slices so a waiting worker still observes Ctrl-C
+_URI_LOCK_POLL = 1
+
+# Resolution errored (vs a clean skip); suppresses the warm sentinel
+_RESOLVE_FAILED = object()
+
+
+def _sweep_stale_sidecars(download_dir: Path, expire_seconds: int) -> None:
+    """Prune resume sidecars pio's usage.db pruner cannot see.
+
+    A version bump strands an aborted archive's sidecars forever. Lock
+    files stay: a held lock can carry an ancient mtime (O_TRUNC keeps
+    it), and unlinking one reopens the single-writer hole it guards.
+    """
+    cutoff = time.time() - expire_seconds
+    try:
+        for f in download_dir.iterdir():
+            if f.suffix not in (".part", ".meta", ".prefetch"):
+                continue
+            try:
+                if f.stat().st_mtime < cutoff:
+                    f.unlink()
+            except OSError as err:
+                _LOGGER.debug("Could not remove %s: %s", f, err)
+    except OSError:
+        _LOGGER.debug("Could not sweep %s", download_dir, exc_info=True)
+
+
+# Child records a no-work run; the parent skips the next spawn while valid
+_SENTINEL_NAME = ".esphome_prefetch.json"
+_SENTINEL_SCHEMA = 1
+
+
+def _ini_sha256(build_dir: Path) -> str:
+    return hashlib.sha256((build_dir / "platformio.ini").read_bytes()).hexdigest()
+
+
+def _sentinel_state(build_dir: Path) -> dict[str, Any]:
+    """The environment fingerprint a sentinel must match to stay valid."""
+    # Same fingerprint as the heal stamp: the sentinel's dirs die with its wipe
+    from esphome.platformio.toolchain import current_python_minor
+
+    return {
+        "schema": _SENTINEL_SCHEMA,
+        "ini_sha256": _ini_sha256(build_dir),
+        "python": current_python_minor(),
+        "core_dir_env": os.environ.get("PLATFORMIO_CORE_DIR", ""),
+    }
+
+
+def _prefetch_is_warm(build_dir: Path) -> bool:
+    """Whether the last prefetch found nothing to do and nothing changed since."""
+    try:
+        data = json.loads((build_dir / _SENTINEL_NAME).read_text(encoding="utf-8"))
+        dirs = data.pop("dirs")
+        return (
+            data == _sentinel_state(build_dir)
+            and bool(dirs)
+            and all(Path(d).is_dir() for d in dirs)
+        )
+    except FileNotFoundError:
+        return False
+    except (OSError, ValueError, KeyError, AttributeError, TypeError):
+        _LOGGER.debug("Ignoring invalid prefetch sentinel", exc_info=True)
+        return False
+
+
+def prefetch_platformio_packages() -> None:
+    """Warm PlatformIO's download cache for the current project, in parallel."""
+    from esphome.core import CORE
+    from esphome.platformio.toolchain import (
+        default_libdeps_dir,
+        heal_platformio_python_env,
+    )
+
+    # Heal first: its Python-version wipe would discard freshly warmed
+    # caches and the sentinel's dirs (the later heal call is a no-op)
+    heal_platformio_python_env()
+    build_dir = Path(CORE.build_path)
+    if _prefetch_is_warm(build_dir):
+        return
+    # The child is esphome itself: PYTHONPATH stays so it imports this
+    # tree's esphome (tests/integration pins the source tree through it)
+    env = dict(os.environ)
+    # Must match run_platformio_cli's default or warm builds re-resolve
+    # every library
+    env.setdefault("PLATFORMIO_LIBDEPS_DIR", default_libdeps_dir())
+    # -v/-vv must reach the child's debug logging or the swallowed
+    # failure detail is undiagnosable in the field
+    env["ESPHOME_PREFETCH_LOG_LEVEL"] = str(logging.getLogger().getEffectiveLevel())
+    if CORE.dashboard:
+        # The child's progress bar and log escaping key off CORE.dashboard
+        env["ESPHOME_PREFETCH_DASHBOARD"] = "1"
+    cmd = [
+        sys.executable,
+        "-m",
+        "esphome.platformio.prefetch",
+        str(build_dir),
+        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
+    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:
+        # The child already warned with the reason; a second line is noise
+        _LOGGER.debug("Prefetch child reported a handled failure")
+    elif proc.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
+        )
+
+
+def _project_platform_and_config(ini: Path, env: str) -> tuple[str | None, Any]:
+    """The env's platform spec and the ProjectConfig for the given ini."""
+    from platformio import app
+    from platformio.project.config import ProjectConfig
+
+    # PlatformBase.config reads the default ProjectConfig; it must see
+    # this ini's env options
+    app.set_session_var("custom_project_conf", str(ini))
+    config = ProjectConfig.get_instance(str(ini))
+    return config.get(f"env:{env}", "platform", None), config
+
+
+def _registry_jobs(
+    manager, specs, seen: set[str]
+) -> tuple[list[tuple[str, int, Any]], int]:
+    """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).
+    """
+    from platformio.registry.mirror import RegistryFileMirrorIterator
+
+    local = threading.local()
+    errors: list[str] = []
+
+    def _resolve(spec) -> tuple[str, int, str, Path, str] | 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__()
+        try:
+            packages = mgr.search_registry_packages(spec)
+            if not packages:
+                _LOGGER.debug("%s is unknown to the registry", spec)
+                return None  # let PlatformIO report it
+            package, version = mgr.find_best_registry_version(packages, spec)
+            if not package or not version:
+                _LOGGER.debug("%s has no matching registry version", spec)
+                return None
+            pkgfile = mgr.pick_compatible_pkg_file(version["files"])
+            if not pkgfile:
+                _LOGGER.debug("%s has no file for this systype", spec)
+                return None
+            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
+            size = pkgfile.get("size")
+            if 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
+        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)
+            errors.append(failure_reason(err))
+            return _RESOLVE_FAILED
+
+    # Serial disk lookups on the shared manager: a fully warm build
+    # resolves nothing, and duplicate specs resolve once
+    unique: dict[tuple[str | None, str, str], Any] = {}
+    for s in specs:
+        if not s.uri and not manager.get_package(s):
+            unique.setdefault((s.owner, s.name, str(s.requirements)), s)
+    pending = list(unique.values())
+    if not pending:
+        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]] = []
+    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))
+        jobs.append(
+            (name, size, _registry_fetch_job(manager, url, dl_path, checksum, size))
+        )
+    if failed := len(errors):
+        # Visible once per build, naming a cause so an API break does not
+        # read as an outage; per-spec detail stays at debug
+        _LOGGER.warning(
+            "Could not resolve %d of %d PlatformIO package(s) (%s); "
+            "PlatformIO will download them serially",
+            failed,
+            len(pending),
+            errors[0],
+        )
+    return jobs, failed
+
+
+def _uri_jobs(manager, specs, seen: set[str]) -> tuple[list[tuple[str, int, Any]], int]:
+    """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).
+    """
+    from esphome.net_retry import fetch_with_retry, http_request
+
+    candidates: list[tuple[str, str, Path]] = []
+    for spec in specs:
+        url = spec.uri
+        if not url or not url.startswith(("http://", "https://")):
+            continue  # git+/file specs are cloned/copied, not downloaded
+        if url.split("#", 1)[0].endswith(".git"):
+            continue  # bare-URL VCS spec; PlatformIO clones it
+        if manager.get_package(spec):
+            continue
+        # 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
+        seen.add(str(dl_path))
+        candidates.append((spec.name, url, dl_path))
+
+    errors: list[str] = []
+
+    def _head_size(url: str) -> int:
+        try:
+            resp = fetch_with_retry(url, lambda: http_request("HEAD", url, timeout=30))
+        except Exception as err:  # noqa: BLE001  # pylint: disable=broad-exception-caught
+            _LOGGER.debug("HEAD %s failed", url, exc_info=True)
+            errors.append(failure_reason(err))
+            return -1
+        if not resp.ok:
+            # An error page's Content-Length is not a download size
+            _LOGGER.debug("HEAD %s returned %s", url, resp.status_code)
+            if resp.status_code in (401, 403, 408, 429) or resp.status_code >= 500:
+                # 401/403 included: registries rate-limit with them
+                errors.append(f"HTTP {resp.status_code}")
+                return -1  # transient; must not be cached as warm
+            # Permanent (405/501 HEAD-unsupported, 401/403/404): a clean
+            # skip so the warm sentinel is not disabled forever; pio run
+            # surfaces a genuinely broken URL when it downloads
+            return 0
+        return content_length(resp)
+
+    if not candidates:
+        return [], 0
+    with ThreadPoolExecutor(max_workers=min(_RESOLVE_WORKERS, len(candidates))) as ex:
+        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):
+        if size < 0:
+            failed += 1
+        elif size:
+            jobs.append((name, size, _uri_fetch_job(manager, url, dl_path, size)))
+        else:
+            # Missing or unusable Content-Length; visible under -v
+            _LOGGER.debug("%s reports no usable length; PlatformIO fetches it", url)
+    if failed:
+        _LOGGER.warning(
+            "Could not size %d of %d PlatformIO package URL(s) (%s); "
+            "PlatformIO will download them serially",
+            failed,
+            len(candidates),
+            errors[0],
+        )
+    return jobs, failed
+
+
+def _serialized_fetch_job(
+    dl_path: Path, lock_path: str, body: Any, unlocked_ok: bool = True
+) -> Any:
+    """Wrap ``body`` so the shared destination is single-writer.
+
+    Interleaved writers truncate each other's ``.part`` bytes (see
+    registry.py). The bounded poll observes Ctrl-C via the tracker; a
+    blown deadline is a clean skip (the holder's copy is what the build
+    needs). On a lock-less filesystem a sha256-verified body runs
+    unlocked with one warning; a checksum-less one
+    (``unlocked_ok=False``) is a counted failure instead.
+    """
+
+    def run(tracker: Any) -> None:
+        from filelock import FileLock, Timeout
+
+        # fallback_to_soft would leave a stale marker on lock-less
+        # filesystems that blocks every later build (see git.py)
+        lock = FileLock(lock_path, fallback_to_soft=False)
+        deadline = time.monotonic() + _DOWNLOAD_LOCK_TIMEOUT
+        while True:
+            try:
+                lock.acquire(timeout=_URI_LOCK_POLL)
+                break
+            except Timeout:
+                tracker(0)  # raises when the batch is cancelled
+                if time.monotonic() >= deadline:
+                    # Another process is fetching this same file; its copy
+                    # is what the build needs (a large framework archive
+                    # can hold the lock far longer than this deadline)
+                    _LOGGER.debug("Leaving %s to its current downloader", dl_path.name)
+                    return
+            except OSError as err:
+                if not unlocked_ok:
+                    # A body with no checksum to catch interleaved corruption
+                    raise
+                lock = None
+                _LOGGER.warning(
+                    "Could not lock %s (%s); downloading unlocked",
+                    dl_path.name,
+                    err,
+                )
+                break
+        try:
+            if dl_path.is_file():
+                return  # another process finished it while we waited
+            body(tracker)
+        finally:
+            if lock is not None:
+                lock.release()
+
+    return run
+
+
+# usage.db is a whole-file rewrite behind pio's self-unlinking LockFile;
+# concurrent writers could reset every recorded entry
+_REGISTER_LOCK = threading.Lock()
+
+
+def _register_download(manager: Any, dl_path: Path) -> None:
+    """Hand the archive to pio's usage.db pruner; an unregistered one is
+    never expired (disk garbage, never a bad build)."""
+    try:
+        with _REGISTER_LOCK:
+            manager.set_download_utime(str(dl_path))
+    except Exception as err:  # noqa: BLE001  # pylint: disable=broad-exception-caught
+        _LOGGER.debug("Could not register %s with pio's cache: %s", dl_path, err)
+
+
+def _registry_fetch_job(
+    manager: Any, url: str, dl_path: Path, checksum: str, size: int
+) -> Any:
+    """A locked fetch straight to the cache path; sha256 verifies it."""
+    # .esphome.lock: pio's own LockFile(dl_path) owns <dl_path>.lock and
+    # deletes it on release, which would unlink a held filelock
+    fetch = _serialized_fetch_job(
+        dl_path,
+        f"{dl_path}.esphome.lock",
+        resume_fetch_job(url, dl_path, sha256=checksum, size=size),
+    )
+
+    def run(tracker: Any) -> None:
+        fetch(tracker)
+        if dl_path.is_file():
+            # The deadline skip can end with no archive landed
+            _register_download(manager, dl_path)
+
+    return run
+
+
+def _uri_fetch_job(manager: Any, url: str, dl_path: Path, size: int) -> Any:
+    """Fetch to a locked staging path, then rename into the cache.
+
+    The stable staging name keeps resume working across interrupted
+    runs; the rename makes the promotion atomic.
+    """
+    tmp = dl_path.with_name(f"{dl_path.name}.prefetch")
+    # attempts=2: the size is only a HEAD probe's word, and a HEAD/GET
+    # disagreement would otherwise re-download the archive five times
+    fetch = resume_fetch_job(url, tmp, size=size, attempts=2)
+
+    def promote(tracker: Any) -> None:
+        fetch(tracker)
+        if (actual := tmp.stat().st_size) != size:
+            # A wrong-length checksum-less body must never be published
+            discard_partial_download(tmp)
+            raise ValueError(f"expected {size} bytes, fetched {actual}")
+        tmp.replace(dl_path)
+
+    def run(tracker: Any) -> None:
+        _serialized_fetch_job(dl_path, f"{tmp}.lock", promote, unlocked_ok=False)(
+            tracker
+        )
+        if dl_path.is_file():
+            # Won or lost, the race is over; staging files left behind
+            # are dead weight PlatformIO's cache never prunes
+            discard_partial_download(tmp)
+            _register_download(manager, dl_path)
+
+    return run
+
+
+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.platform.factory import PlatformFactory
+
+    platform_spec, config = _project_platform_and_config(
+        build_dir / "platformio.ini", env
+    )
+    if not platform_spec:
+        # An env mismatch must not disable the feature with no trace
+        _LOGGER.debug(
+            "No platform for env %s in %s; nothing to prefetch", env, build_dir
+        )
+        return
+
+    # The platform (manifest plus build scripts) installs first and
+    # resolves the rest. Its setup may rewrite sys.path (pioarduino's penv
+    # setup does); restore it so later imports here still resolve.
+    saved_sys_path = list(sys.path)
+    pm = PlatformPackageManager()
+    _sweep_stale_sidecars(Path(pm.get_download_dir()), pm.DOWNLOAD_CACHE_EXPIRE)
+    pkg = pm.install(platform_spec, skip_dependencies=True)
+    p = PlatformFactory.new(pkg)
+    p.configure_project_packages(env, ["run"])
+    sys.path[:] = saved_sys_path
+
+    specs = [
+        p.get_package_spec(name)
+        for name, opts in p.packages.items()
+        if not opts.get("optional")
+    ]
+    # PIO's build engine installs outside the platform package list;
+    # skipped when the platform lists it itself
+    if not any(s.name == "tool-scons" for s in specs):
+        specs.append(
+            PackageSpec(
+                owner="platformio",
+                name="tool-scons",
+                requirements=get_core_dependencies()["tool-scons"],
+            )
+        )
+    lib_deps = config.get(f"env:{env}", "lib_deps", [])
+    # pio run's storage dir for this env: installed libraries skip by
+    # disk lookup
+    libdeps_dir = Path(config.get("platformio", "libdeps_dir")) / env
+    lm = LibraryPackageManager(str(libdeps_dir))
+    # 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
+    lib_specs = [
+        spec
+        for dep in lib_deps
+        if dep and not dep.startswith("$")
+        if (spec := PackageSpec(dep)).external or spec.owner
+    ]
+
+    seen: set[str] = set()
+    jobs: list[tuple[str, int, Any]] = []
+    unresolved = 0
+    for mgr, batch in ((p.pm, specs), (lm, lib_specs)):
+        for build_jobs in (_registry_jobs, _uri_jobs):
+            batch_jobs, failed = build_jobs(mgr, batch, seen)
+            jobs += batch_jobs
+            unresolved += failed
+
+    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))
+
+
+def main(argv: list[str]) -> int:
+    """Subprocess entry point: ``prefetch <build_dir> <env_name>``."""
+    from esphome.core import CORE
+    from esphome.log import setup_log
+
+    raw_level = os.environ.get("ESPHOME_PREFETCH_LOG_LEVEL")
+    try:
+        level = int(raw_level) if raw_level is not None else logging.INFO
+    except ValueError:
+        level = logging.INFO
+    # Mirror the parent's log setup: warnings keep their level prefix and
+    # color, and the download bar still draws under the dashboard
+    CORE.dashboard = get_bool_env("ESPHOME_PREFETCH_DASHBOARD")
+    setup_log(level)
+    # pio's managers attach their own handler and still propagate; without
+    # this every manager line also prints through the root handler. Their
+    # construction re-pins the logger to INFO, so a logger-level filter
+    # (which survives pio's handler reset) enforces a quiet level instead.
+    for cls_name in (
+        "ToolPackageManager",
+        "LibraryPackageManager",
+        "PlatformPackageManager",
+    ):
+        manager_logger = logging.getLogger(cls_name.replace("Package", " "))
+        manager_logger.propagate = False
+        manager_logger.addFilter(lambda record: record.levelno >= level)
+    if len(argv) != 2:
+        # A wiring bug, not a network failure; make it distinguishable
+        _LOGGER.warning("prefetch usage: <build_dir> <env_name>")
+        return 2
+    build_dir, env = argv
+    try:
+        _prefetch(Path(build_dir), env)
+    except KeyboardInterrupt:
+        # Shared process group: exit quietly, no traceback on the terminal
+        _LOGGER.debug("Prefetch interrupted", exc_info=True)
+        return 130
+    except Exception as err:  # noqa: BLE001  # pylint: disable=broad-exception-caught
+        # The parent treats any exit as warn-and-continue, never a failure
+        _LOGGER.warning("PlatformIO package prefetch skipped: %s", failure_reason(err))
+        _LOGGER.debug("Prefetch failure detail", exc_info=True)
+        return _EXIT_HANDLED
+    return 0
+
+
+if __name__ == "__main__":  # pragma: no cover
+    sys.exit(main(sys.argv[1:]))

esphome/platformio/toolchain.py

@@ -96,7 +96,7 @@ def _clean_platformio_python_env(config: "ProjectConfig", core_dir: Path) -> Non
         rmtree(penv)
 
 
-def _current_python_minor() -> str:
+def current_python_minor() -> str:
     """Return the running interpreter's ``major.minor`` (e.g. ``3.13``)."""
     return f"{sys.version_info.major}.{sys.version_info.minor}"
 
@@ -161,7 +161,7 @@ def heal_platformio_python_env() -> None:
 
 def _check_platformio_python_stamp(config: "ProjectConfig") -> None:
     """Compare the stamp to the running interpreter; wipe and restamp on mismatch."""
-    current = _current_python_minor()
+    current = current_python_minor()
     stamp_dir = _pio_stamp_dir(config)
     # Host the stamp/lock even before PlatformIO's first run creates the dir.
     stamp_dir.mkdir(parents=True, exist_ok=True)
@@ -289,6 +289,12 @@ def copy_ccache_script() -> None:
     )
 
 
+def default_libdeps_dir() -> str:
+    """The PLATFORMIO_LIBDEPS_DIR value a pio run defaults to; the package
+    prefetch must resolve installed libraries against the same dir."""
+    return str(CORE.relative_piolibdeps_path().absolute())
+
+
 def run_platformio_cli(*args, **kwargs) -> str | int:
     # Re-provision the PlatformIO cache if the interpreter's major.minor changed
     # since it was last built; a stale platform otherwise rejects the new Python
@@ -296,9 +302,7 @@ def run_platformio_cli(*args, **kwargs) -> str | int:
     heal_platformio_python_env()
     os.environ["PLATFORMIO_FORCE_COLOR"] = "true"
     os.environ["PLATFORMIO_BUILD_DIR"] = str(CORE.relative_pioenvs_path().absolute())
-    os.environ.setdefault(
-        "PLATFORMIO_LIBDEPS_DIR", str(CORE.relative_piolibdeps_path().absolute())
-    )
+    os.environ.setdefault("PLATFORMIO_LIBDEPS_DIR", default_libdeps_dir())
     # Suppress Python syntax warnings from third-party scripts during compilation
     os.environ.setdefault("PYTHONWARNINGS", "ignore::SyntaxWarning")
     # Increase uv retry count to handle transient network errors (default is 3)
@@ -346,6 +350,9 @@ def run_platformio_cli_run(config, verbose, *args, **kwargs) -> str | int:
 
 
 def run_compile(config, verbose):
+    from esphome.platformio.prefetch import prefetch_platformio_packages
+
+    prefetch_platformio_packages()
     args = []
     if CONF_COMPILE_PROCESS_LIMIT in config[CONF_ESPHOME]:
         args += [f"-j{config[CONF_ESPHOME][CONF_COMPILE_PROCESS_LIMIT]}"]

tests/unit_tests/test_espidf_framework.py

@@ -911,7 +911,7 @@ def test_prefetch_leaves_unverifiable_entries_to_the_installer(
             "esphome.espidf.framework.run_command",
             return_value=(True, json.dumps(entries), ""),
         ),
-        patch("esphome.espidf.framework.download_with_resume") as download,
+        patch("esphome.framework_helpers.download_with_resume") as download,
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
         patch("esphome.framework_helpers._BatchDownloadProgress") as progress_cls,
     ):
@@ -934,7 +934,7 @@ def test_prefetch_all_entries_unverifiable_is_a_noop(tmp_path: Path) -> None:
             "esphome.espidf.framework.run_command",
             return_value=(True, json.dumps(entries), ""),
         ),
-        patch("esphome.espidf.framework.download_with_resume") as download,
+        patch("esphome.framework_helpers.download_with_resume") as download,
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
     ):
         _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
@@ -952,7 +952,7 @@ def test_prefetch_dedupes_entries_by_dest(tmp_path: Path) -> None:
             "esphome.espidf.framework.run_command",
             return_value=(True, json.dumps(entries), ""),
         ),
-        patch("esphome.espidf.framework.download_with_resume") as download,
+        patch("esphome.framework_helpers.download_with_resume") as download,
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
         patch("esphome.framework_helpers._BatchDownloadProgress"),
     ):
@@ -967,7 +967,7 @@ def test_prefetch_downloads_each_archive_with_resume(tmp_path: Path) -> None:
             "esphome.espidf.framework.run_command",
             return_value=(True, _PREFETCH_JSON, ""),
         ),
-        patch("esphome.espidf.framework.download_with_resume") as download,
+        patch("esphome.framework_helpers.download_with_resume") as download,
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
         patch("esphome.framework_helpers._BatchDownloadProgress") as progress_cls,
     ):
@@ -1011,7 +1011,7 @@ def test_prefetch_downloads_archives_concurrently(tmp_path: Path) -> None:
             "esphome.espidf.framework.run_command",
             return_value=(True, json.dumps(entries), ""),
         ),
-        patch("esphome.espidf.framework.download_with_resume") as download,
+        patch("esphome.framework_helpers.download_with_resume") as download,
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
         patch(
             "esphome.framework_helpers.ThreadPoolExecutor", wraps=ThreadPoolExecutor
@@ -1032,7 +1032,7 @@ def test_prefetch_skips_already_downloaded_archives(tmp_path: Path) -> None:
             "esphome.espidf.framework.run_command",
             return_value=(True, _PREFETCH_JSON, ""),
         ),
-        patch("esphome.espidf.framework.download_with_resume") as download,
+        patch("esphome.framework_helpers.download_with_resume") as download,
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
     ):
         _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
@@ -1065,7 +1065,7 @@ def test_prefetch_failures_never_raise(
     with (
         patch("esphome.espidf.framework.run_command", return_value=run_result),
         patch(
-            "esphome.espidf.framework.download_with_resume",
+            "esphome.framework_helpers.download_with_resume",
             side_effect=download_error,
         ),
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
@@ -1087,7 +1087,7 @@ def test_prefetch_total_failure_logs_error(
             return_value=(True, _PREFETCH_JSON, ""),
         ),
         patch(
-            "esphome.espidf.framework.download_with_resume",
+            "esphome.framework_helpers.download_with_resume",
             side_effect=OSError("proxy refuses everything"),
         ),
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
@@ -1112,7 +1112,7 @@ def test_prefetch_one_failed_archive_does_not_stop_the_rest(
             return_value=(True, _PREFETCH_JSON, ""),
         ),
         patch(
-            "esphome.espidf.framework.download_with_resume",
+            "esphome.framework_helpers.download_with_resume",
             side_effect=_fail_cmake_download,
         ) as download,
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
@@ -1133,7 +1133,7 @@ def test_prefetch_finishes_progress_bar_and_cancels_queue(tmp_path: Path) -> Non
             "esphome.espidf.framework.run_command",
             return_value=(True, _PREFETCH_JSON, ""),
         ),
-        patch("esphome.espidf.framework.download_with_resume"),
+        patch("esphome.framework_helpers.download_with_resume"),
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
         patch("esphome.framework_helpers._BatchDownloadProgress") as progress_cls,
         patch("esphome.framework_helpers.ThreadPoolExecutor") as pool_cls,

tests/unit_tests/test_framework_helpers.py

@@ -2280,6 +2280,32 @@ class TestGetProjectCxxCompileFlags:
             assert get_project_cxx_compile_flags() == []
 
 
+def test_resume_fetch_job_threads_tracker(tmp_path: Path) -> None:
+    """The batch runner passes the tracker positionally; the shared adapter
+    must deliver it as download_with_resume's progress keyword."""
+    from esphome.framework_helpers import resume_fetch_job
+
+    with patch("esphome.framework_helpers.download_with_resume") as mock_download:
+        fetch = resume_fetch_job("https://x/a.zip", tmp_path / "a", sha256="ff", size=9)
+        tracker = lambda done: None  # noqa: E731
+        fetch(tracker)
+    mock_download.assert_called_once_with(
+        "https://x/a.zip", tmp_path / "a", progress=tracker, sha256="ff", size=9
+    )
+
+
+def test_warn_prefetch_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
+
+    warn_prefetch_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")
+    assert "Prefetch of lib failed: gone" in caplog.text
+
+
 @pytest.mark.parametrize(
     ("platform", "input_path", "expected"),
     [
@@ -2312,3 +2338,18 @@ def test_strip_win_long_path_prefix(
     r"""``\\?\`` and ``\\?\UNC\`` prefixes are stripped only on win32."""
     with patch("esphome.framework_helpers.sys.platform", platform):
         assert framework_helpers.strip_win_long_path_prefix(input_path) == expected
+
+
+def test_discard_partial_download_logs_undeletable(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """An unremovable staging file leaves a debug trace; the caller's
+    cache is never pruned, so silence would hide unbounded growth."""
+    dest = tmp_path / "archive"
+    dest.write_bytes(b"stale")
+    with (
+        patch.object(Path, "unlink", side_effect=OSError("busy")),
+        caplog.at_level(logging.DEBUG),
+    ):
+        framework_helpers.discard_partial_download(dest)
+    assert "Could not remove" in caplog.text

tests/unit_tests/test_platformio_prefetch.py

@@ -0,0 +1,985 @@
+"""Tests for the parallel PlatformIO package prefetch."""
+
+import errno
+import json
+import os
+from pathlib import Path
+import sys
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from filelock import Timeout
+import pytest
+
+from esphome.core import CORE
+import esphome.platformio.prefetch as pf
+
+
+@pytest.fixture(autouse=True)
+def _core(tmp_path: Path):
+    CORE.reset()
+    CORE.build_path = str(tmp_path)
+    CORE.name = "testenv"
+    pio_loggers = ("Tool Manager", "Library Manager", "Platform Manager")
+    saved_propagate = {n: pf.logging.getLogger(n).propagate for n in pio_loggers}
+    saved_filters = {n: list(pf.logging.getLogger(n).filters) for n in pio_loggers}
+    # The real setup_log would swap pytest's root-handler formatter
+    with patch("esphome.log.setup_log"):
+        yield
+    # main() flips these process-wide; keep the suite hermetic
+    for n, flag in saved_propagate.items():
+        pf.logging.getLogger(n).propagate = flag
+        pf.logging.getLogger(n).filters[:] = saved_filters[n]
+    CORE.reset()
+
+
+class _FakeSpec(SimpleNamespace):
+    """PackageSpec stand-in for the attributes the prefetch reads."""
+
+    def __init__(
+        self, *, owner=None, requirements=None, external=False, **kwargs
+    ) -> None:
+        super().__init__(
+            owner=owner, requirements=requirements, external=external, **kwargs
+        )
+
+
+def _fake_manager(tmp_path: Path) -> MagicMock:
+    m = MagicMock()
+    m.__class__ = lambda: m  # _resolve constructs a same-class instance
+    m.get_package.return_value = None
+    m.search_registry_packages.return_value = [{"any": 1}]
+    m.find_best_registry_version.return_value = (
+        {"name": "toolchain-xtensa"},
+        {
+            "name": "2.0.0",
+            "files": [
+                {
+                    "download_url": "https://dl.example/t.tar.gz",
+                    "checksum": {"sha256": "cafe"},
+                    "size": 1000,
+                }
+            ],
+        },
+    )
+    m.pick_compatible_pkg_file.side_effect = lambda files: files[0]
+    m.compute_download_path.side_effect = lambda url, checksum: str(
+        tmp_path / "dl" / f"{abs(hash((url, checksum)))}"
+    )
+    return m
+
+
+def _mirror_patch():
+    return patch.dict(
+        "sys.modules",
+        {
+            "platformio.registry.mirror": SimpleNamespace(
+                RegistryFileMirrorIterator=lambda url: iter(
+                    [("https://mirror.example/t.tar.gz", "beef")]
+                )
+            )
+        },
+    )
+
+
+def test_registry_jobs_resolves_like_platformio(tmp_path: Path) -> None:
+    """A registry spec resolves to a job keyed by mirror URL and checksum."""
+    m = _fake_manager(tmp_path)
+    with _mirror_patch():
+        jobs, failed = pf._registry_jobs(
+            m, [_FakeSpec(uri=None, name="toolchain-xtensa")], set()
+        )
+    assert failed == 0
+    assert len(jobs) == 1
+    name, size, fetch = jobs[0]
+    assert name == "toolchain-xtensa@2.0.0"
+    assert size == 1000
+    m.compute_download_path.assert_called_once_with(
+        "https://mirror.example/t.tar.gz", "beef"
+    )
+    assert callable(fetch)
+
+
+@pytest.mark.parametrize(
+    ("method", "attr", "value"),
+    [
+        ("get_package", "return_value", object()),  # already installed
+        ("search_registry_packages", "return_value", []),  # unknown package
+        ("find_best_registry_version", "return_value", (None, None)),  # no match
+        ("pick_compatible_pkg_file", "side_effect", lambda files: None),  # no file
+    ],
+)
+def test_registry_jobs_skips(tmp_path: Path, method, attr, value) -> None:
+    """Entries PlatformIO would not download produce no job."""
+    m = _fake_manager(tmp_path)
+    setattr(getattr(m, method), attr, value)
+    with _mirror_patch():
+        assert pf._registry_jobs(m, [_FakeSpec(uri=None, name="x")], set()) == ([], 0)
+
+
+def test_registry_jobs_skips_cached_and_sizeless(tmp_path: Path) -> None:
+    """Cached or sizeless files are left to PlatformIO."""
+    m = _fake_manager(tmp_path)
+    dl = Path(m.compute_download_path("https://mirror.example/t.tar.gz", "beef"))
+    dl.parent.mkdir(parents=True, exist_ok=True)
+    dl.touch()
+    with _mirror_patch():
+        assert pf._registry_jobs(m, [_FakeSpec(uri=None, name="x")], set()) == ([], 0)
+    dl.unlink()
+    m.find_best_registry_version.return_value[1]["files"][0]["size"] = 0
+    with _mirror_patch():
+        assert pf._registry_jobs(m, [_FakeSpec(uri=None, name="x")], set()) == ([], 0)
+
+
+def test_registry_jobs_dedupes_download_paths(tmp_path: Path) -> None:
+    """Duplicate specs resolve once and one archive yields one job (two
+    workers must never share a .part); nine specs against eight workers
+    also exercise the thread-local manager reuse."""
+    m = _fake_manager(tmp_path)
+    specs = [_FakeSpec(uri=None, name="dup"), _FakeSpec(uri=None, name="dup")]
+    specs += [_FakeSpec(uri=None, name=f"n{i}") for i in range(8)]
+    with _mirror_patch():
+        jobs, failed = pf._registry_jobs(m, specs, set())
+    # the fake resolves every spec to the same mirror URL and checksum
+    assert failed == 0
+    assert len(jobs) == 1
+    assert m.search_registry_packages.call_count == 9  # dup resolved once
+
+
+def test_registry_jobs_uri_specs_excluded(tmp_path: Path) -> None:
+    """URL specs never reach the registry resolution."""
+    m = _fake_manager(tmp_path)
+    assert pf._registry_jobs(
+        m, [_FakeSpec(uri="https://x/y.zip", name="y")], set()
+    ) == ([], 0)
+    m.search_registry_packages.assert_not_called()
+
+
+def test_registry_jobs_dedup_keeps_distinct_owners(tmp_path: Path) -> None:
+    """platformio/x and pioarduino/x are different packages."""
+    m = _fake_manager(tmp_path)
+    specs = [
+        _FakeSpec(uri=None, name="framework-x", owner="platformio"),
+        _FakeSpec(uri=None, name="framework-x", owner="pioarduino"),
+    ]
+    with _mirror_patch():
+        pf._registry_jobs(m, specs, set())
+    assert m.search_registry_packages.call_count == 2
+
+
+def test_registry_jobs_all_failed_warns_once(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A whole-batch failure is a systemic fault and must be visible."""
+    m = _fake_manager(tmp_path)
+    m.search_registry_packages.side_effect = RuntimeError("registry down")
+    with _mirror_patch():
+        jobs, failed = pf._registry_jobs(
+            m,
+            [_FakeSpec(uri=None, name="a"), _FakeSpec(uri=None, name="b")],
+            set(),
+        )
+    assert (jobs, failed) == ([], 2)
+    # The aggregate warning names a cause so an API break does not read
+    # as a registry outage
+    assert "Could not resolve 2 of 2" in caplog.text
+    assert "registry down" in caplog.text
+
+
+def test_uri_fetch_job_promotes_atomically(tmp_path: Path) -> None:
+    """Checksum-less URL archives land via a locked staging file and an
+    atomic rename (the stable name is what keeps .part resume working)."""
+    dl_path = tmp_path / "archive"
+
+    def fake_download(url, dest, progress=None, **kwargs):
+        Path(dest).write_bytes(b"data")
+
+    manager = MagicMock()
+    with patch(
+        "esphome.framework_helpers.download_with_resume", side_effect=fake_download
+    ):
+        pf._uri_fetch_job(manager, "https://x/a.zip", dl_path, 4)(lambda done: None)
+    assert dl_path.read_bytes() == b"data"
+    # The archive is handed to pio's usage.db pruner
+    manager.set_download_utime.assert_called_once_with(str(dl_path))
+    # no orphaned staging file; the lock file may or may not persist
+    # (filelock removes it on release on some platforms)
+    leftovers = {f.name for f in tmp_path.iterdir()}
+    assert leftovers - {f"{dl_path.name}.prefetch.lock"} == {dl_path.name}
+
+
+def test_registry_fetch_job_skips_when_cached(tmp_path: Path) -> None:
+    """A destination another process completed is not re-downloaded."""
+    dl_path = tmp_path / "archive"
+    dl_path.write_bytes(b"done")
+    with patch("esphome.framework_helpers.download_with_resume") as mock_download:
+        pf._registry_fetch_job(
+            MagicMock(), "https://x/a.tar.gz", dl_path, "ab" * 32, 4
+        )(lambda done: None)
+    mock_download.assert_not_called()
+    assert dl_path.read_bytes() == b"done"
+
+
+def test_registry_fetch_job_downloads_under_lock(tmp_path: Path) -> None:
+    """Registry downloads write the shared cache path under the same lock
+    the URL path uses; interleaved writers would corrupt the archive."""
+    dl_path = tmp_path / "archive"
+    order: list[str] = []
+    with (
+        patch(
+            "esphome.framework_helpers.download_with_resume",
+            side_effect=lambda url, dest, progress=None, **kw: (
+                order.append("fetch"),
+                Path(dest).write_bytes(b"data"),  # registration needs a real file
+            ),
+        ),
+        patch(
+            "filelock.FileLock.acquire",
+            side_effect=lambda *a, **k: order.append("lock"),
+        ),
+        patch(
+            "filelock.FileLock.release",
+            side_effect=lambda *a, **k: order.append("unlock"),
+        ),
+    ):
+        manager = MagicMock()
+        pf._registry_fetch_job(manager, "https://x/a.tar.gz", dl_path, "ab" * 32, 4)(
+            lambda done: None
+        )
+    # FileLock.__del__ may add a trailing release; the contract is the order
+    assert order[:2] == ["lock", "fetch"]
+    assert "unlock" in order[2:]
+    manager.set_download_utime.assert_called_once_with(str(dl_path))
+
+
+def test_lockless_filesystem_downloads_unlocked(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A filesystem without lock support (ENOSYS/EPERM) degrades to an
+    unlocked download with one warning, never a per-package failure."""
+    dl_path = tmp_path / "archive"
+    with (
+        patch("esphome.framework_helpers.download_with_resume") as mock_download,
+        patch(
+            "filelock.FileLock.acquire",
+            side_effect=OSError(errno.ENOSYS, "no locks"),
+        ),
+        patch("filelock.FileLock.release"),
+    ):
+        pf._registry_fetch_job(
+            MagicMock(), "https://x/a.tar.gz", dl_path, "ab" * 32, 4
+        )(lambda done: None)
+    mock_download.assert_called_once()
+    assert "downloading unlocked" in caplog.text
+
+
+def test_uri_fetch_job_failed_download_keeps_staging(tmp_path: Path) -> None:
+    """A failed fetch keeps the .part staging bytes for the next resume."""
+    dl_path = tmp_path / "archive"
+    part = tmp_path / "archive.prefetch.part"
+    part.write_bytes(b"partial")
+    with (
+        patch(
+            "esphome.framework_helpers.download_with_resume",
+            side_effect=OSError("network gone"),
+        ),
+        pytest.raises(OSError, match="network gone"),
+    ):
+        pf._uri_fetch_job(MagicMock(), "https://x/a.zip", dl_path, 4)(lambda done: None)
+    assert part.read_bytes() == b"partial"
+    assert not dl_path.exists()
+
+
+def test_uri_fetch_job_rejects_wrong_length(tmp_path: Path) -> None:
+    """A checksum-less body of the wrong length is never published under a
+    cache key pio would trust forever."""
+    dl_path = tmp_path / "archive"
+
+    def fake_download(url, dest, progress=None, **kwargs):
+        Path(dest).write_bytes(b"short")
+
+    with (
+        patch(
+            "esphome.framework_helpers.download_with_resume", side_effect=fake_download
+        ),
+        pytest.raises(ValueError, match="expected 9999 bytes"),
+    ):
+        pf._uri_fetch_job(MagicMock(), "https://x/a.zip", dl_path, 9999)(
+            lambda done: None
+        )
+    assert not dl_path.exists()
+    assert not (tmp_path / "archive.prefetch").exists()
+
+
+def test_sweep_stale_sidecars(tmp_path: Path) -> None:
+    """Sidecars past pio's own expiry are pruned; fresh and foreign files
+    stay."""
+    old_time = pf.time.time() - 110
+    stale = tmp_path / "a.tar.gz.part"
+    stale.write_bytes(b"x")
+    os.utime(stale, (old_time, old_time))
+    fresh = tmp_path / "b.tar.gz.part"
+    fresh.write_bytes(b"x")
+    keep = tmp_path / "c.tar.gz"
+    keep.write_bytes(b"x")
+    os.utime(keep, (old_time, old_time))
+    # A held lock can carry an ancient mtime (O_TRUNC keeps it); locks
+    # must never be swept or the single-writer guarantee reopens
+    held_lock = tmp_path / "d.tar.gz.esphome.lock"
+    held_lock.write_bytes(b"")
+    os.utime(held_lock, (old_time, old_time))
+    pf._sweep_stale_sidecars(tmp_path, 100)
+    assert not stale.exists()
+    assert fresh.exists()
+    assert keep.exists()
+    assert held_lock.exists()
+    pf._sweep_stale_sidecars(tmp_path / "missing", 100)  # tolerated
+
+
+def test_register_download_failure_leaves_a_trace(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A failed usage.db registration is traced; an unregistered archive
+    is never pruned, so silence would hide the leak coming back."""
+    manager = MagicMock()
+    manager.set_download_utime.side_effect = RuntimeError("db locked")
+    with caplog.at_level(pf.logging.DEBUG):
+        pf._register_download(manager, tmp_path / "a.tar.gz")
+    assert "Could not register" in caplog.text
+
+
+def test_sweep_logs_unprunable_files(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A sidecar that cannot be removed leaves a trace; a sweep that never
+    prunes must not look like a clean sweep."""
+    old_time = pf.time.time() - 110
+    stale = tmp_path / "a.tar.gz.part"
+    stale.write_bytes(b"x")
+    os.utime(stale, (old_time, old_time))
+    with (
+        patch.object(Path, "unlink", side_effect=OSError("busy")),
+        caplog.at_level(pf.logging.DEBUG),
+    ):
+        pf._sweep_stale_sidecars(tmp_path, 100)
+    assert "Could not remove" in caplog.text
+
+
+def test_uri_lock_failure_is_a_counted_failure(tmp_path: Path) -> None:
+    """The checksum-less URL path never degrades to an unlocked shared
+    write; interleaved right-length corruption would go undetected."""
+    dl_path = tmp_path / "archive"
+    with (
+        patch("esphome.framework_helpers.download_with_resume") as mock_download,
+        patch(
+            "filelock.FileLock.acquire",
+            side_effect=OSError(errno.ENOSYS, "no locks"),
+        ),
+        pytest.raises(OSError),
+    ):
+        pf._uri_fetch_job(MagicMock(), "https://x/a.zip", dl_path, 4)(lambda done: None)
+    mock_download.assert_not_called()
+
+
+def test_uri_fetch_job_no_discard_without_a_file(tmp_path: Path) -> None:
+    """When no archive landed (degraded serialized run), the staging bytes
+    stay for the next resume instead of being discarded."""
+    dl_path = tmp_path / "archive"
+    part = tmp_path / "archive.prefetch.part"
+    part.write_bytes(b"partial")
+    with patch.object(pf, "_serialized_fetch_job", return_value=lambda tracker: None):
+        pf._uri_fetch_job(MagicMock(), "https://x/a.zip", dl_path, 4)(lambda done: None)
+    assert part.read_bytes() == b"partial"
+
+
+def test_uri_fetch_job_waits_out_a_briefly_held_lock(tmp_path: Path) -> None:
+    """A lock freed within the deadline lets the job proceed normally."""
+    dl_path = tmp_path / "archive"
+
+    def fake_download(url, dest, progress=None, **kwargs):
+        Path(dest).write_bytes(b"data")
+
+    with (
+        patch(
+            "esphome.framework_helpers.download_with_resume", side_effect=fake_download
+        ),
+        patch("filelock.FileLock.acquire", side_effect=[Timeout("held"), None]),
+        patch("filelock.FileLock.release"),
+    ):
+        pf._uri_fetch_job(MagicMock(), "https://x/a.zip", dl_path, 4)(lambda done: None)
+    assert dl_path.read_bytes() == b"data"
+
+
+def test_lock_deadline_leaves_download_to_the_holder(tmp_path: Path) -> None:
+    """A lock held past the deadline means another process is fetching the
+    same file; skipping cleanly beats a misleading failure warning. The
+    tracker is still polled so a parked worker observes cancellation."""
+    dl_path = tmp_path / "archive"
+    ticks: list[int] = []
+    with (
+        patch("esphome.framework_helpers.download_with_resume") as mock_download,
+        patch("filelock.FileLock.acquire", side_effect=Timeout("held")),
+        patch.object(pf, "_DOWNLOAD_LOCK_TIMEOUT", 0),
+    ):
+        pf._uri_fetch_job(MagicMock(), "https://x/a.zip", dl_path, 4)(ticks.append)
+    mock_download.assert_not_called()
+    assert ticks == [0]
+    assert not dl_path.exists()
+
+
+def test_registry_lock_deadline_skips_registration(tmp_path: Path) -> None:
+    """A registry job that lost the download race to another process
+    must not stamp a nonexistent archive into pio's usage.db."""
+    manager = MagicMock()
+    dl_path = tmp_path / "archive"
+    with (
+        patch("esphome.framework_helpers.download_with_resume") as mock_download,
+        patch("filelock.FileLock.acquire", side_effect=Timeout("held")),
+        patch.object(pf, "_DOWNLOAD_LOCK_TIMEOUT", 0),
+    ):
+        pf._registry_fetch_job(manager, "https://x/a.tar.gz", dl_path, "ab" * 32, 4)(
+            lambda done: None
+        )
+    mock_download.assert_not_called()
+    manager.set_download_utime.assert_not_called()
+
+
+def test_main_interrupt_exits_quietly(tmp_path: Path) -> None:
+    """Ctrl-C reaches the child via the shared process group; it must exit
+    without a traceback."""
+    with (
+        patch("esphome.log.setup_log"),
+        patch.object(pf, "_prefetch", side_effect=KeyboardInterrupt),
+    ):
+        assert pf.main([str(tmp_path), "testenv"]) == 130
+
+
+def test_main_bad_log_level_falls_back(tmp_path: Path) -> None:
+    with (
+        patch.dict("os.environ", {"ESPHOME_PREFETCH_LOG_LEVEL": "verbose"}),
+        patch("esphome.log.setup_log") as mock_setup,
+        patch.object(pf, "_prefetch"),
+    ):
+        assert pf.main([str(tmp_path), "testenv"]) == 0
+    assert mock_setup.call_args[0][0] == pf.logging.INFO
+
+
+def test_main_silences_pio_manager_propagation(tmp_path: Path) -> None:
+    """The pio manager loggers carry their own handler; propagation to
+    the root handler would print every install line twice."""
+    with patch("esphome.log.setup_log"), patch.object(pf, "_prefetch"):
+        assert pf.main([str(tmp_path), "testenv"]) == 0
+    for name in ("Tool Manager", "Library Manager", "Platform Manager"):
+        assert pf.logging.getLogger(name).propagate is False
+
+
+def test_main_quiet_level_reaches_pio_manager_loggers(tmp_path: Path) -> None:
+    """Manager construction re-pins its logger to INFO, so a quiet run
+    needs the logger-level filter to keep per-package lines out."""
+    with (
+        patch.dict("os.environ", {"ESPHOME_PREFETCH_LOG_LEVEL": "30"}),
+        patch("esphome.log.setup_log"),
+        patch.object(pf, "_prefetch"),
+    ):
+        assert pf.main([str(tmp_path), "testenv"]) == 0
+    lib_logger = pf.logging.getLogger("Library Manager")
+    lib_logger.setLevel(pf.logging.INFO)  # what pio's _setup_logger does
+    info = pf.logging.LogRecord("Library Manager", 20, __file__, 1, "x", (), None)
+    warning = pf.logging.LogRecord("Library Manager", 30, __file__, 1, "x", (), None)
+    # Logger.filter returns falsy to drop, the record itself to pass
+    assert not lib_logger.filter(info)
+    assert lib_logger.filter(warning)
+
+
+def test_main_mirrors_parent_log_setup(tmp_path: Path) -> None:
+    """The child adopts the parent's dashboard flag and log formatter so
+    its warnings and progress bar match the parent's."""
+    with (
+        patch.dict(
+            "os.environ",
+            {"ESPHOME_PREFETCH_LOG_LEVEL": "30", "ESPHOME_PREFETCH_DASHBOARD": "1"},
+        ),
+        patch("esphome.log.setup_log") as mock_setup,
+        patch.object(pf, "_prefetch"),
+    ):
+        assert pf.main([str(tmp_path), "testenv"]) == 0
+    mock_setup.assert_called_once_with(30)
+    assert CORE.dashboard is True
+
+
+def test_uri_fetch_job_skips_when_another_process_won(tmp_path: Path) -> None:
+    """A lost race discards the staging files; the cache never prunes them."""
+    dl_path = tmp_path / "archive"
+    dl_path.write_bytes(b"done")
+    stale = [
+        tmp_path / "archive.prefetch",
+        tmp_path / "archive.prefetch.part",
+        tmp_path / "archive.prefetch.part.meta",
+    ]
+    for f in stale:
+        f.write_bytes(b"stale")
+    with patch("esphome.framework_helpers.download_with_resume") as mock_download:
+        pf._uri_fetch_job(MagicMock(), "https://x/a.zip", dl_path, 4)(lambda done: None)
+    mock_download.assert_not_called()
+    assert dl_path.read_bytes() == b"done"
+    assert not any(f.exists() for f in stale)
+
+
+def test_registry_jobs_one_bad_spec_keeps_the_rest(tmp_path: Path) -> None:
+    """A flaky resolution counts as failed without discarding the batch."""
+    m = _fake_manager(tmp_path)
+    m.search_registry_packages.side_effect = [
+        RuntimeError("registry 500"),
+        [{"any": 1}],
+    ]
+    with _mirror_patch():
+        jobs, failed = pf._registry_jobs(
+            m,
+            [_FakeSpec(uri=None, name="flaky"), _FakeSpec(uri=None, name="good")],
+            set(),
+        )
+    assert failed == 1
+    assert len(jobs) == 1
+
+
+def test_uri_jobs_head_sizes_the_bar(tmp_path: Path) -> None:
+    """HEAD sizes direct-URL specs; git and unreachable URLs are skipped."""
+    m = _fake_manager(tmp_path)
+    resp = MagicMock()
+    resp.headers = {"content-length": "2222"}
+    with patch("esphome.net_retry.http_request", return_value=resp):
+        jobs, failed = pf._uri_jobs(
+            m,
+            [
+                _FakeSpec(uri="https://x/big.zip", name="big"),
+                _FakeSpec(uri="git+https://x/repo.git", name="repo"),
+                _FakeSpec(uri="https://x/repo.git#v1", name="barevcs"),
+                _FakeSpec(uri=None, name="registry"),
+            ],
+            set(),
+        )
+    assert failed == 0
+    assert [(n, s) for n, s, _ in jobs] == [("big", 2222)]
+    # a successful HEAD with no Content-Length is a clean skip
+    resp.headers = {}
+    with patch("esphome.net_retry.http_request", return_value=resp):
+        assert pf._uri_jobs(
+            m, [_FakeSpec(uri="https://x/nolen.zip", name="nolen")], set()
+        ) == ([], 0)
+
+
+def test_uri_jobs_head_failure_counts_as_unresolved(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Network errors and transient statuses count as unresolved (and warn);
+    any permanent error status is a clean skip so the sentinel can still
+    be written (pio run names a broken URL when it downloads)."""
+    m = _fake_manager(tmp_path)
+    spec = [_FakeSpec(uri="https://x/a.zip", name="a")]
+    with patch("esphome.net_retry.http_request", side_effect=OSError("no route")):
+        assert pf._uri_jobs(m, spec, set()) == ([], 1)
+    resp = MagicMock(ok=False, status_code=503)
+    resp.headers = {"content-length": "999"}
+    with patch("esphome.net_retry.http_request", return_value=resp):
+        assert pf._uri_jobs(m, spec, set()) == ([], 1)
+    # 403 is how registries rate-limit; it must not be cached as warm
+    resp = MagicMock(ok=False, status_code=403)
+    resp.headers = {"content-length": "999"}
+    with patch("esphome.net_retry.http_request", return_value=resp):
+        assert pf._uri_jobs(m, spec, set()) == ([], 1)
+    resp = MagicMock(ok=False, status_code=405)
+    resp.headers = {"content-length": "999"}
+    with patch("esphome.net_retry.http_request", return_value=resp):
+        assert pf._uri_jobs(m, spec, set()) == ([], 0)
+    assert "HEAD https://x/a.zip" not in caplog.text
+    resp = MagicMock(ok=False, status_code=404)
+    resp.headers = {"content-length": "999"}
+    with patch("esphome.net_retry.http_request", return_value=resp):
+        assert pf._uri_jobs(m, spec, set()) == ([], 0)
+    assert "returned 404" not in caplog.text
+
+
+def test_uri_jobs_dedupes_duplicate_urls(tmp_path: Path) -> None:
+    """Two specs with one URL yield one HEAD and one job."""
+    m = _fake_manager(tmp_path)
+    resp = MagicMock()
+    resp.headers = {"content-length": "5"}
+    with patch("esphome.net_retry.http_request", return_value=resp) as mock_head:
+        jobs, failed = pf._uri_jobs(
+            m,
+            [
+                _FakeSpec(uri="https://x/a.zip", name="a"),
+                _FakeSpec(uri="https://x/a.zip", name="a"),
+            ],
+            set(),
+        )
+    assert failed == 0
+    assert len(jobs) == 1
+    mock_head.assert_called_once()
+
+
+def test_uri_jobs_skips_installed_cached_and_seen(tmp_path: Path) -> None:
+    m = _fake_manager(tmp_path)
+    m.get_package.return_value = object()
+    spec = [_FakeSpec(uri="https://x/a.zip", name="a")]
+    assert pf._uri_jobs(m, spec, set()) == ([], 0)
+    m.get_package.return_value = None
+    dl = Path(m.compute_download_path("https://x/a.zip", ""))
+    dl.parent.mkdir(parents=True, exist_ok=True)
+    dl.touch()
+    assert pf._uri_jobs(m, spec, set()) == ([], 0)
+    dl.unlink()
+    # a registry job already claimed this download path
+    assert pf._uri_jobs(m, spec, {str(dl)}) == ([], 0)
+
+
+def test_prefetch_spawns_isolated_subprocess(tmp_path: Path) -> None:
+    """Heal runs first, then the subprocess spawns with pio run's libdeps
+    dir and the parent's PYTHONPATH preserved (the child is esphome)."""
+    proc = MagicMock(returncode=0)
+    order = MagicMock()
+    order.run.return_value = proc
+    with (
+        patch(
+            "esphome.platformio.toolchain.heal_platformio_python_env",
+            order.heal,
+        ),
+        patch.object(pf.subprocess, "run", order.run) as mock_run,
+        patch.dict("os.environ", {"PYTHONPATH": "/leak"}),
+    ):
+        pf.prefetch_platformio_packages()
+    assert [c[0] for c in order.mock_calls[:2]] == ["heal", "run"]
+    (cmd,), kwargs = mock_run.call_args
+    assert cmd == [
+        sys.executable,
+        "-m",
+        "esphome.platformio.prefetch",
+        str(CORE.build_path),
+        "testenv",
+    ]
+    assert kwargs["env"]["PLATFORMIO_LIBDEPS_DIR"] == str(
+        CORE.relative_piolibdeps_path().absolute()
+    )
+    # The child is esphome itself; PYTHONPATH must survive so it imports
+    # the same tree (tests/integration pins the source tree through it)
+    assert kwargs["env"]["PYTHONPATH"] == "/leak"
+    assert "ESPHOME_PREFETCH_DASHBOARD" not in kwargs["env"]
+    assert kwargs["timeout"] == pf._PREFETCH_TIMEOUT
+
+
+def test_prefetch_passes_dashboard_flag(tmp_path: Path) -> None:
+    """The dashboard flag reaches the child so its bar still draws."""
+    CORE.dashboard = True
+    with (
+        patch("esphome.platformio.toolchain.heal_platformio_python_env"),
+        patch.object(
+            pf.subprocess, "run", return_value=MagicMock(returncode=0)
+        ) as mock_run,
+    ):
+        pf.prefetch_platformio_packages()
+    assert mock_run.call_args[1]["env"]["ESPHOME_PREFETCH_DASHBOARD"] == "1"
+
+
+@pytest.mark.parametrize(
+    ("run_effect", "expected"),
+    [
+        (
+            {"side_effect": pf.subprocess.TimeoutExpired("cmd", pf._PREFETCH_TIMEOUT)},
+            "prefetch timed out",
+        ),
+        ({"return_value": MagicMock(returncode=4)}, "prefetch skipped (exit 4)"),
+        # Exit 1 is the interpreter's own import-failure code, never quiet
+        ({"return_value": MagicMock(returncode=1)}, "prefetch skipped (exit 1)"),
+        ({"side_effect": OSError("no exec")}, "PlatformIO package prefetch skipped"),
+    ],
+)
+def test_prefetch_spawn_failures_warn_and_continue(
+    caplog: pytest.LogCaptureFixture, run_effect, expected
+) -> None:
+    """Timeouts, nonzero exits, and spawn failures each warn, never raise."""
+    with (
+        patch("esphome.platformio.toolchain.heal_platformio_python_env"),
+        patch.object(pf.subprocess, "run", **run_effect),
+    ):
+        pf.prefetch_platformio_packages()
+    assert expected in caplog.text
+
+
+def test_prefetch_child_handled_failure_is_quiet(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Exit _EXIT_HANDLED (3) means the child already warned with the
+    reason; the parent adds no second warning."""
+    with (
+        patch("esphome.platformio.toolchain.heal_platformio_python_env"),
+        patch.object(
+            pf.subprocess, "run", return_value=MagicMock(returncode=pf._EXIT_HANDLED)
+        ),
+    ):
+        pf.prefetch_platformio_packages()
+    assert "prefetch skipped" not in caplog.text
+
+
+def test_main_guards_and_exits_nonzero(caplog: pytest.LogCaptureFixture) -> None:
+    """A swallowed failure still reaches the parent as a nonzero exit; the
+    parent warns and continues, never failing the build."""
+    with patch.object(pf, "_prefetch", side_effect=RuntimeError("boom")):
+        assert pf.main(["/b", "testenv"]) == pf._EXIT_HANDLED
+    assert "PlatformIO package prefetch skipped" in caplog.text
+
+
+def test_main_runs_prefetch(tmp_path: Path) -> None:
+    with patch.object(pf, "_prefetch") as mock_prefetch:
+        assert pf.main([str(tmp_path), "testenv"]) == 0
+    mock_prefetch.assert_called_once_with(tmp_path, "testenv")
+
+
+def test_main_bad_argv_is_a_distinct_exit(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A parent/child wiring bug must not look like a network failure."""
+    with patch.object(pf, "_prefetch") as mock_prefetch:
+        assert pf.main(["only-one"]) == 2
+    mock_prefetch.assert_not_called()
+    assert "prefetch usage" in caplog.text
+
+
+def _write_ini(tmp_path: Path, body: str) -> None:
+    (tmp_path / "platformio.ini").write_text(body)
+
+
+def _write_valid_sentinel(tmp_path: Path, dirs: list[str]) -> None:
+    (tmp_path / pf._SENTINEL_NAME).write_text(
+        json.dumps({**pf._sentinel_state(tmp_path), "dirs": dirs}), encoding="utf-8"
+    )
+
+
+def test_prefetch_no_platform_returns(tmp_path: Path) -> None:
+    _write_ini(tmp_path, "[env:testenv]\n")
+    with patch.object(pf, "_registry_jobs") as mock_jobs:
+        pf._prefetch(tmp_path, "testenv")
+    mock_jobs.assert_not_called()
+
+
+def _pio_modules(tmp_path: Path, fake_platform, fake_pm, config, lib_captures=None):
+    # A bare MagicMock's get_download_dir would fspath to '' and point the
+    # sidecar sweep at the process cwd
+    fake_pm.get_download_dir.return_value = str(tmp_path / "downloads")
+    fake_pm.DOWNLOAD_CACHE_EXPIRE = 86400 * 30
+
+    def fake_lib_manager(storage_dir):
+        if lib_captures is not None:
+            lib_captures.append(storage_dir)
+        return _fake_manager(tmp_path)
+
+    modules = {
+        "platformio": MagicMock(),
+        "platformio.app": MagicMock(),
+        "platformio.project": MagicMock(),
+        "platformio.project.config": MagicMock(),
+        "platformio.dependencies": SimpleNamespace(
+            get_core_dependencies=lambda: {
+                "tool-scons": "~4.0",
+                "contrib-piohome": "~3",
+            }
+        ),
+        "platformio.package": MagicMock(),
+        "platformio.package.manager": MagicMock(),
+        "platformio.package.manager.library": SimpleNamespace(
+            LibraryPackageManager=fake_lib_manager
+        ),
+        "platformio.package.manager.platform": SimpleNamespace(
+            PlatformPackageManager=lambda: fake_pm
+        ),
+        "platformio.package.meta": SimpleNamespace(
+            PackageSpec=lambda *a, **kw: _FakeSpec(
+                uri=None,
+                name=kw.get("name") or (a[0] if a else None),
+                owner=kw.get("owner")
+                or (str(a[0]).split("/")[0] if a and "/" in str(a[0]) else None),
+                external=bool(a and "://" in str(a[0])),
+            )
+        ),
+        "platformio.platform": MagicMock(),
+        "platformio.platform.factory": SimpleNamespace(
+            PlatformFactory=SimpleNamespace(new=lambda pkg: fake_platform)
+        ),
+    }
+    modules[
+        "platformio.project.config"
+    ].ProjectConfig.get_instance.return_value = config
+    return modules
+
+
+def _fake_config(tmp_path: Path, env_options: dict):
+    config = MagicMock()
+    options = {
+        "libdeps_dir": str(tmp_path / "libdeps"),
+        "packages_dir": str(tmp_path / "packages"),
+        **env_options,
+    }
+    config.get.side_effect = lambda section, key, default=None: options.get(
+        key, default
+    )
+    return config
+
+
+def test_prefetch_all_cached_is_quiet_and_writes_sentinel(tmp_path: Path) -> None:
+    """A no-work run neither logs nor batches, and records the sentinel."""
+    _write_ini(tmp_path, "[env:testenv]\nplatform = fake/p@1\n")
+    (tmp_path / "packages").mkdir()
+    (tmp_path / "libdeps" / "testenv").mkdir(parents=True)
+    fake_platform = MagicMock()
+    fake_platform.packages = {}
+    config = _fake_config(
+        tmp_path, {"platform": "fake/p@1", "lib_deps": ["esphome/noise-c@1.0"]}
+    )
+    modules = _pio_modules(tmp_path, fake_platform, MagicMock(), config)
+    with (
+        patch.dict("sys.modules", modules),
+        patch.object(pf, "_registry_jobs", return_value=([], 0)),
+        patch.object(pf, "_uri_jobs", return_value=([], 0)),
+        patch.object(pf, "run_batch_downloads") as mock_batch,
+    ):
+        pf._prefetch(tmp_path, "testenv")
+    mock_batch.assert_not_called()
+    assert pf._prefetch_is_warm(tmp_path)
+
+
+def test_prefetch_failed_resolution_is_not_cached_as_warm(tmp_path: Path) -> None:
+    """A registry outage must not write the sentinel."""
+    _write_ini(tmp_path, "[env:testenv]\nplatform = fake/p@1\n")
+    (tmp_path / "packages").mkdir()
+    fake_platform = MagicMock()
+    fake_platform.packages = {}
+    config = _fake_config(tmp_path, {"platform": "fake/p@1"})
+    modules = _pio_modules(tmp_path, fake_platform, MagicMock(), config)
+    with (
+        patch.dict("sys.modules", modules),
+        patch.object(pf, "_registry_jobs", return_value=([], 1)),
+        patch.object(pf, "_uri_jobs", return_value=([], 0)),
+        patch.object(pf, "run_batch_downloads") as mock_batch,
+    ):
+        pf._prefetch(tmp_path, "testenv")
+    mock_batch.assert_not_called()
+    assert not (tmp_path / pf._SENTINEL_NAME).exists()
+
+
+def test_sentinel_invalidation(tmp_path: Path) -> None:
+    """Ini changes, missing dirs, and garbage sentinels all read as cold."""
+    _write_ini(tmp_path, "[env:testenv]\nplatform = fake/p@1\n")
+    pkg_dir = tmp_path / "packages"
+    pkg_dir.mkdir()
+    assert not pf._prefetch_is_warm(tmp_path)  # no sentinel yet
+    _write_valid_sentinel(tmp_path, [str(pkg_dir)])
+    assert pf._prefetch_is_warm(tmp_path)
+    _write_ini(tmp_path, "[env:testenv]\nplatform = fake/p@2\n")
+    assert not pf._prefetch_is_warm(tmp_path)  # ini changed
+    _write_ini(tmp_path, "[env:testenv]\nplatform = fake/p@1\n")
+    pkg_dir.rmdir()
+    assert not pf._prefetch_is_warm(tmp_path)  # recorded dir gone
+    (tmp_path / pf._SENTINEL_NAME).write_text("not json", encoding="utf-8")
+    assert not pf._prefetch_is_warm(tmp_path)
+
+
+def test_prefetch_warm_sentinel_skips_spawn(tmp_path: Path) -> None:
+    """A valid sentinel skips the subprocess entirely."""
+    _write_ini(tmp_path, "[env:testenv]\nplatform = fake/p@1\n")
+    pkg_dir = tmp_path / "packages"
+    pkg_dir.mkdir()
+    _write_valid_sentinel(tmp_path, [str(pkg_dir)])
+    with (
+        patch("esphome.platformio.toolchain.heal_platformio_python_env"),
+        patch.object(pf.subprocess, "run") as mock_run,
+    ):
+        pf.prefetch_platformio_packages()
+    mock_run.assert_not_called()
+
+
+def test_prefetch_end_to_end_wiring(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Platform installs dep-free, non-optional packages plus tool-scons
+    resolve, libraries use the env libdeps dir, a platform sys.path rewrite
+    is undone, and failures warn by name."""
+    _write_ini(tmp_path, "[env:testenv]\nplatform = fake/platform@1.0\n")
+    fake_platform = MagicMock()
+    fake_platform.packages = {
+        "toolchain-x": {"optional": False},
+        "framework-y": {"optional": True},
+    }
+    fake_platform.get_package_spec.side_effect = lambda name: _FakeSpec(
+        uri=None, name=name
+    )
+    # Platform setup code rewrites sys.path (pioarduino penv); _prefetch
+    # must restore it
+    bogus = str(tmp_path / "penv-site-packages")
+    fake_platform.configure_project_packages.side_effect = lambda env, targets: (
+        sys.path.insert(0, bogus)
+    )
+    fake_pm = MagicMock()
+    config = _fake_config(
+        tmp_path,
+        {
+            "platform": "fake/platform@1.0",
+            # the bare built-in name and the interpolation are skipped;
+            # only the owner-qualified library resolves
+            "lib_deps": ["esphome/noise-c@1.0", "WiFi", "${common.lib_deps}"],
+        },
+    )
+    lib_dirs: list[str] = []
+    modules = _pio_modules(tmp_path, fake_platform, fake_pm, config, lib_dirs)
+    (tmp_path / pf._SENTINEL_NAME).write_text("{}", encoding="utf-8")
+    captured: dict = {}
+
+    def fake_registry_jobs(manager, specs, seen):
+        captured.setdefault("spec_batches", []).append([s.name for s in specs])
+        return [("toolchain-x@1", 10, lambda t: None)], 0
+
+    with (
+        patch.dict("sys.modules", modules),
+        patch.object(pf, "_registry_jobs", side_effect=fake_registry_jobs),
+        patch.object(pf, "_uri_jobs", return_value=([], 0)),
+        patch.object(
+            pf,
+            "run_batch_downloads",
+            return_value=[("toolchain-x@1", OSError("down"))],
+        ) as mock_batch,
+    ):
+        pf._prefetch(tmp_path, "testenv")
+    fake_pm.install.assert_called_once_with("fake/platform@1.0", skip_dependencies=True)
+    assert not (tmp_path / pf._SENTINEL_NAME).exists()  # stale sentinel removed
+    fake_platform.configure_project_packages.assert_called_once_with("testenv", ["run"])
+    assert bogus not in sys.path
+    # non-optional platform package + tool-scons (never piohome), then libs
+    assert captured["spec_batches"][0] == ["toolchain-x", "tool-scons"]
+    assert captured["spec_batches"][1] == ["esphome/noise-c@1.0"]
+    assert lib_dirs == [str(Path(tmp_path / "libdeps") / "testenv")]
+    mock_batch.assert_called_once()
+    assert "Could not prefetch toolchain-x@1" in caplog.text
+
+
+def test_prefetch_skips_duplicate_tool_scons(tmp_path: Path) -> None:
+    """A platform that lists tool-scons itself does not get it appended."""
+    _write_ini(tmp_path, "[env:testenv]\nplatform = fake/p@1\n")
+    fake_platform = MagicMock()
+    fake_platform.packages = {"tool-scons": {"optional": False}}
+    fake_platform.get_package_spec.side_effect = lambda name: _FakeSpec(
+        uri=None, name=name
+    )
+    config = _fake_config(tmp_path, {"platform": "fake/p@1"})
+    modules = _pio_modules(tmp_path, fake_platform, MagicMock(), config)
+    batches: list[list[str]] = []
+    with (
+        patch.dict("sys.modules", modules),
+        patch.object(
+            pf,
+            "_registry_jobs",
+            side_effect=lambda mgr, specs, seen: (
+                batches.append([s.name for s in specs]) or ([], 0)
+            ),
+        ),
+        patch.object(pf, "_uri_jobs", return_value=([], 0)),
+    ):
+        pf._prefetch(tmp_path, "testenv")
+    assert batches[0] == ["tool-scons"]

tests/unit_tests/test_platformio_toolchain.py

@@ -932,8 +932,13 @@ def test_run_compile(setup_core: Path, mock_run_platformio_cli_run: Mock) -> Non
     config = {CONF_ESPHOME: {CONF_COMPILE_PROCESS_LIMIT: 4}}
     mock_run_platformio_cli_run.return_value = 0
 
-    toolchain.run_compile(config, verbose=True)
+    with patch(
+        "esphome.platformio.prefetch.prefetch_platformio_packages"
+    ) as mock_prefetch:
+        toolchain.run_compile(config, verbose=True)
 
+    # The only wiring of the prefetch into a build lives here
+    mock_prefetch.assert_called_once_with()
     mock_run_platformio_cli_run.assert_called_once_with(config, True, "-j4")
 
 
@@ -947,7 +952,8 @@ def test_run_compile_without_process_limit(
     config = {CONF_ESPHOME: {}}
     mock_run_platformio_cli_run.return_value = 0
 
-    toolchain.run_compile(config, verbose=False)
+    with patch("esphome.platformio.prefetch.prefetch_platformio_packages"):
+        toolchain.run_compile(config, verbose=False)
 
     mock_run_platformio_cli_run.assert_called_once_with(config, False)
 
@@ -1677,8 +1683,8 @@ def pio_core_dir(tmp_path: Path) -> Path:
 
 
 def test_current_python_minor_matches_running_interpreter() -> None:
-    """_current_python_minor returns major.minor of the running interpreter."""
-    assert toolchain._current_python_minor() == _CURRENT_MINOR
+    """current_python_minor returns major.minor of the running interpreter."""
+    assert toolchain.current_python_minor() == _CURRENT_MINOR
 
 
 def test_pio_stamp_round_trip(tmp_path: Path) -> None: