Merge branch 'esp8266-native-build-infra' into esp8266-native-framework-installer

# Conflicts:
#	esphome/writer.py

esphome/arduino8266/__init__.py

@@ -0,0 +1,9 @@
+"""Native (PlatformIO-free) build support for the ESP8266 Arduino core.
+
+This package downloads the Arduino ESP8266 core and the xtensa-lx106
+toolchain, generates a ninja build for them plus the ESPHome sources, and
+drives the build directly — the ESP8266 equivalent of ``esphome.espidf``.
+
+Deliberately importable without the esp8266 component to avoid circular
+imports; the component wires these modules in via lazy imports.
+"""

esphome/arduino8266/framework.py

@@ -0,0 +1,164 @@
+"""Download and install the Arduino ESP8266 core, toolchain, and ninja.
+
+Artifacts land in a machine-global cache (shared across projects, like the
+ESP-IDF install in ``esphome.espidf.framework``):
+
+    <cache>/arduino8266/frameworks/<version>/   framework-arduinoespressif8266
+    <cache>/arduino8266/toolchains/<version>/   toolchain-xtensa (gcc 10.3)
+
+Packages come from the PlatformIO registry (identical bits to the PlatformIO
+backend); ``ESPHOME_ARDUINO8266_*_MIRRORS`` overrides the URLs. ninja comes
+from PATH or the ninja PyPI wheel.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from typing import NamedTuple
+
+from esphome.build_helpers.ccache import ccache_defaults_env
+from esphome.build_helpers.ninja import find_ninja
+from esphome.build_helpers.tools_cache import ARDUINO8266_TOOLS_CACHE, tools_cache_path
+from esphome.core import EsphomeError, Version
+from esphome.framework_helpers import str_to_lst_of_str
+from esphome.platformio.registry import install_package, prefetch_packages
+
+FRAMEWORK_PACKAGE = "framework-arduinoespressif8266"
+TOOLCHAIN_PACKAGE = "toolchain-xtensa"
+# gcc 10.3, the toolchain Arduino core 3.x builds with; the build
+# generator's compile flags are tuned to it.
+TOOLCHAIN_VERSION = "2.100300.220621"
+
+ESPHOME_ARDUINO8266_FRAMEWORK_MIRRORS = str_to_lst_of_str(
+    os.environ.get("ESPHOME_ARDUINO8266_FRAMEWORK_MIRRORS", "")
+)
+ESPHOME_ARDUINO8266_TOOLCHAIN_MIRRORS = str_to_lst_of_str(
+    os.environ.get("ESPHOME_ARDUINO8266_TOOLCHAIN_MIRRORS", "")
+)
+
+
+def get_arduino8266_tools_path() -> Path:
+    # Machine-global so all projects share one install; see
+    # espidf.framework.get_idf_tools_path for the location rationale.
+    return tools_cache_path(*ARDUINO8266_TOOLS_CACHE)
+
+
+# 3.1.1 rather than 3.1.0: the registry has no package for 3.1.0, and the
+# encoder below cannot name 3.0.0/3.0.1 either (see its docstring)
+MIN_FRAMEWORK_VERSION = Version(3, 1, 1)
+
+
+def framework_package_version(ver: Version) -> str:
+    """Map an Arduino core version to its registry package version (3.1.2 ->
+    3.30102.0; the leading 3 is the package major).
+
+    Exact registry names only for cores > 2.6.2 and >= 3.0.2; callers floor
+    at MIN_FRAMEWORK_VERSION.
+    """
+    if ver.major > 3:
+        raise EsphomeError(
+            f"Arduino core {ver} is not supported yet; "
+            "the newest known core series is 3.x"
+        )
+    if ver <= Version(2, 6, 2):
+        # Cores <= 2.6.2 use the older 1.x/2.x package-major encodings (same
+        # boundary as _format_framework_arduino_version's era guard)
+        raise EsphomeError(
+            f"Arduino core {ver} uses an older package encoding than this "
+            "helper implements (newer than 2.6.2)"
+        )
+    return f"3.{ver.major}{ver.minor:02d}{ver.patch:02d}.0"
+
+
+def get_framework_path(package_version: str) -> Path:
+    return get_arduino8266_tools_path() / "frameworks" / package_version
+
+
+def get_toolchain_path() -> Path:
+    return get_arduino8266_tools_path() / "toolchains" / TOOLCHAIN_VERSION
+
+
+class InstalledPaths(NamedTuple):
+    """Locations of the installed framework, toolchain, and ninja binary."""
+
+    framework: Path
+    toolchain: Path
+    ninja: Path
+
+
+def check_and_install(framework_version: Version) -> InstalledPaths:
+    """Ensure framework, toolchain, and ninja are installed; return their paths."""
+    if framework_version < MIN_FRAMEWORK_VERSION:
+        # Config validation enforces this too; keep the module honest when
+        # called directly.
+        raise EsphomeError(
+            f"The native toolchain requires the Arduino core "
+            f">= {MIN_FRAMEWORK_VERSION}, got {framework_version}"
+        )
+    # Probe the cheap local dependency before ~110 MB of downloads
+    ninja_path = find_ninja()
+    package_version = framework_package_version(framework_version)
+    framework_path = get_framework_path(package_version)
+    downloads_dir = get_arduino8266_tools_path() / "downloads"
+    toolchain_path = get_toolchain_path()
+    # One spec per package: the prefetch and the installs must agree
+    specs = (
+        (
+            FRAMEWORK_PACKAGE,
+            package_version,
+            framework_path,
+            ESPHOME_ARDUINO8266_FRAMEWORK_MIRRORS,
+            ("cores/esp8266", "tools/sdk", "libraries"),
+        ),
+        (
+            TOOLCHAIN_PACKAGE,
+            TOOLCHAIN_VERSION,
+            toolchain_path,
+            ESPHOME_ARDUINO8266_TOOLCHAIN_MIRRORS,
+            # xtensa-lx106-elf pins the target: every gcc package has a bin/
+            ("bin", "xtensa-lx106-elf"),
+        ),
+    )
+    # Fetch both archives at once; the installs below verify and extract
+    prefetch_packages([spec[:4] for spec in specs], downloads_dir)
+    for name, version, dest, mirrors, expect in specs:
+        install_package(name, version, dest, mirrors, downloads_dir, expect=expect)
+    return InstalledPaths(
+        framework=framework_path, toolchain=toolchain_path, ninja=ninja_path
+    )
+
+
+def toolchain_tool(toolchain_path: Path, name: str) -> Path:
+    """Path to one toolchain tool (gcc, g++, ar, size, addr2line, ...).
+
+    The single owner of the ``bin/xtensa-lx106-elf-<name>`` layout and the
+    Windows suffix, so a toolchain package bump touches one spot.
+    """
+    suffix = ".exe" if os.name == "nt" else ""
+    return toolchain_path / "bin" / f"xtensa-lx106-elf-{name}{suffix}"
+
+
+def get_build_env(toolchain_path: Path, ccache: str | None) -> dict[str, str]:
+    env = os.environ.copy()
+    # Drop empty entries: a trailing separator from an absent PATH would
+    # make the shell search the current directory for tools
+    parts = [
+        str(toolchain_path / "bin"),
+        *filter(None, env.get("PATH", "").split(os.pathsep)),
+    ]
+    env["PATH"] = os.pathsep.join(parts)
+    env.update(ccache_env(ccache))
+    return env
+
+
+def ccache_env(ccache: str | None) -> dict[str, str]:
+    """Return ccache settings for the build subprocess (not os.environ).
+
+    ``ccache`` is the pre-resolved binary (resolve_ccache_path), or None
+    when disabled. Values the user already set in the environment are
+    respected.
+    """
+    if ccache is None:
+        return {}
+    return ccache_defaults_env(get_arduino8266_tools_path() / "ccache")

esphome/build_helpers/ccache.py

@@ -0,0 +1,93 @@
+"""Shared ccache policy for build backends: env-knob parsing, binary
+resolution, and default ``CCACHE_*`` values."""
+
+from __future__ import annotations
+
+import logging
+import os
+from pathlib import Path
+
+from esphome.framework_helpers import strip_win_long_path_prefix, tool_version_runs
+from esphome.helpers import FALSY_BOOL_STRINGS, TRUTHY_BOOL_STRINGS
+
+_LOGGER = logging.getLogger(__name__)
+
+# cv.boolean's spelling tables plus the 1/0 env convention
+TRUTHY_ENV_STRINGS = TRUTHY_BOOL_STRINGS | {"1"}
+FALSY_ENV_STRINGS = FALSY_BOOL_STRINGS | {"0"}
+
+
+def _ccache_runs(ccache: str) -> bool:
+    """Return True when the ``ccache`` found on PATH actually runs."""
+    return tool_version_runs(
+        ccache,
+        "Ignoring ccache at %s because it failed to run; compiling without ccache",
+    )
+
+
+def parse_enable_env(name: str) -> bool | None:
+    """Strictly parse an on/off environment knob; None when unset or invalid.
+
+    ``bool(str)`` truthiness would flip ``no``/``off`` to enabled, so only
+    1/true/yes/on and 0/false/no/off count; anything else warns and reads
+    as unset so the caller's default policy applies.
+    """
+    raw = os.environ.get(name)
+    if raw is None:
+        return None
+    lowered = raw.strip().lower()
+    if lowered in TRUTHY_ENV_STRINGS:
+        return True
+    if lowered in FALSY_ENV_STRINGS:
+        return False
+    _LOGGER.warning("Ignoring unrecognized %s=%r; use 1 or 0", name, raw)
+    return None
+
+
+def resolve_ccache_path() -> str | None:
+    """The ccache binary to wrap compiles with, or None when disabled.
+
+    An explicit ``ESPHOME_CCACHE_ENABLE=1`` skips the runnability probe; the
+    Windows extended-length prefix is stripped before probing (#18399).
+    """
+    import shutil
+
+    explicit = parse_enable_env("ESPHOME_CCACHE_ENABLE")
+    if explicit is False:
+        return None
+    ccache = shutil.which("ccache")
+    if ccache is None:
+        if explicit:
+            _LOGGER.warning(
+                "ESPHOME_CCACHE_ENABLE is set but no ccache binary is on PATH; "
+                "compiling without ccache"
+            )
+        return None
+    ccache = strip_win_long_path_prefix(ccache)
+    if not explicit and not _ccache_runs(ccache):
+        return None
+    return ccache
+
+
+def ccache_defaults_env(cache_dir: Path) -> dict[str, str]:
+    """Default ``CCACHE_*`` values for a build subprocess (not os.environ).
+
+    Values the user already set in the environment are respected. Depend
+    mode is on: both native backends emit depfiles (-MMD / CMake), which
+    keeps cache-miss overhead low.
+    """
+    from esphome.core import CORE
+
+    # An unset build_path means the env was built before preload; fail loudly
+    # rather than silently drop CCACHE_BASEDIR.
+    if CORE.build_path is None:
+        raise ValueError(
+            "CORE.build_path must be set before constructing the build environment"
+        )
+    defaults = {
+        "CCACHE_DIR": str(cache_dir),
+        "CCACHE_NOHASHDIR": "true",
+        "CCACHE_DEPEND": "1",
+        "CCACHE_BASEDIR": str(Path(CORE.build_path).resolve()),
+    }
+    return {k: v for k, v in defaults.items() if k not in os.environ}

esphome/build_helpers/ninja.py

@@ -0,0 +1,92 @@
+"""Platform-neutral helpers for ninja-driven native builds."""
+
+from __future__ import annotations
+
+import logging
+import os
+from pathlib import Path
+import re
+import shutil
+
+from esphome.core import EsphomeError
+from esphome.framework_helpers import strip_win_long_path_prefix, tool_version_runs
+
+_LOGGER = logging.getLogger(__name__)
+
+
+def _ninja_runs(binary: str) -> bool:
+    """Whether the ninja found on PATH actually runs (see tool_version_runs)."""
+    return tool_version_runs(
+        binary,
+        "Ignoring ninja at %s because it failed to run; "
+        "falling back to the bundled wheel",
+    )
+
+
+def find_ninja() -> Path:
+    """Locate the ninja binary: a runnable PATH hit first, else the ninja
+    PyPI wheel."""
+    if binary := shutil.which("ninja"):
+        binary = strip_win_long_path_prefix(binary)
+        if _ninja_runs(binary):
+            return Path(binary)
+    import_error: ImportError | None = None
+    try:
+        import ninja
+    except ImportError as err:
+        import_error = err
+        wheel_binary = None
+    else:
+        wheel_binary = Path(ninja.BIN_DIR) / (
+            "ninja.exe" if os.name == "nt" else "ninja"
+        )
+    if wheel_binary is None or not wheel_binary.is_file():
+        raise EsphomeError(
+            "ninja not found on PATH or in the ninja package; reinstall the "
+            "esphome Python environment"
+        ) from import_error
+    return wheel_binary
+
+
+def escape(value: Path | str) -> str:
+    """Escape a path or token for a ninja file."""
+    return str(value).replace("$", "$$").replace(":", "$:").replace(" ", "$ ")
+
+
+def quote_arg(tok: str) -> str:
+    """Quote with the CreateProcess argv rule (as ``subprocess.list2cmdline``):
+    backslash runs double only before a quote. Windows-only; ``$`` must
+    already be doubled for ninja.
+    """
+    quoted = re.sub(r'(\\*)"', lambda m: m.group(1) * 2 + '\\"', tok)
+    quoted = re.sub(r"(\\+)\Z", lambda m: m.group(1) * 2, quoted)
+    return f'"{quoted}"'
+
+
+# Force-quote any token containing a character outside the shlex.quote-style
+# safe set: ninja hands POSIX commands to /bin/sh -c, so bare (, ;, <, *, `
+# and friends would be re-parsed as shell syntax.
+_NEEDS_QUOTE = re.compile(r"[^\w@%+=:,./-]")
+
+
+def shell_token(tok: str, force: bool = False) -> str:
+    """Re-quote a lexed token for the platform shell; ``force`` always quotes.
+
+    Single quotes on POSIX (/bin/sh), the argv rule on Windows
+    (CreateProcess). ``$`` is doubled first because ninja expands it before
+    the command reaches the shell.
+    """
+    tok = tok.replace("$", "$$")  # ninja would expand a bare $ to nothing
+    if not (force or not tok or _NEEDS_QUOTE.search(tok)):
+        return tok
+    # An empty token must become '' / "" or it vanishes from the argv
+    if os.name == "nt":
+        return quote_arg(tok)
+    # shlex.quote's rule; inlined because the $-doubled token must not be
+    # re-examined for safe characters
+    return "'" + tok.replace("'", "'\"'\"'") + "'"
+
+
+def quote_path(value: Path | str) -> str:
+    """Force-quote a path for the ninja command line (shell/CreateProcess)."""
+    return shell_token(str(value), force=True)

esphome/build_helpers/tools_cache.py

@@ -0,0 +1,36 @@
+"""Machine-global tools cache location shared by the native backends."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+
+def tools_cache_path(env_var: str, subdir: str) -> Path:
+    """A backend's machine-global tools directory, with an env override.
+
+    A blank/whitespace override is treated as unset: ``Path("")`` resolves
+    to the CWD, which ``clean-all`` would then delete.
+    """
+    import platformdirs
+
+    from esphome.helpers import get_str_env
+
+    if prefix := get_str_env(env_var, "").strip():
+        # resolve(): symlinked prefixes otherwise trip idf.py's
+        # venv-mismatch warning on every build
+        return Path(prefix).expanduser().resolve()
+    # appauthor=False keeps the Windows path short (no vendor segment);
+    # deep IDF trees run into MAX_PATH otherwise
+    return (
+        Path(platformdirs.user_cache_dir("esphome", appauthor=False)) / subdir
+    ).resolve()
+
+
+# (env override, cache subdir) per native backend. writer.clean_all wipes
+# every entry via tools_cache_path, so listing a cache here is the single
+# step that registers it for removal; the backends' own path getters use
+# the same named pairs so the two cannot drift.
+IDF_TOOLS_CACHE = ("ESPHOME_ESP_IDF_PREFIX", "idf")
+SDK_NRF_TOOLS_CACHE = ("ESPHOME_SDK_NRF_PREFIX", "sdk-nrf")
+ARDUINO8266_TOOLS_CACHE = ("ESPHOME_ARDUINO8266_PREFIX", "arduino8266")
+TOOLS_CACHE_SPECS = (IDF_TOOLS_CACHE, SDK_NRF_TOOLS_CACHE, ARDUINO8266_TOOLS_CACHE)

esphome/components/esp8266/__init__.py

@@ -136,7 +136,16 @@ def _format_framework_arduino_version(ver: cv.Version) -> str:
         return f"~1.{ver.major}{ver.minor:02d}{ver.patch:02d}.0"
     if ver <= cv.Version(2, 6, 2):
         return f"~2.{ver.major}{ver.minor:02d}{ver.patch:02d}.0"
-    return f"~3.{ver.major}{ver.minor:02d}{ver.patch:02d}.0"
+    # Same encoding the native toolchain uses for its package download, so a
+    # version bump cannot drift between the two paths.
+    from esphome.arduino8266.framework import framework_package_version
+
+    try:
+        return f"~{framework_package_version(ver)}"
+    except EsphomeError as err:
+        # Anchor the 4.x rejection to the framework version line instead of
+        # aborting with a bare traceback-level error
+        raise cv.Invalid(str(err), path=[CONF_VERSION]) from err
 
 
 # NOTE: Keep this in mind when updating the recommended version:

esphome/components/nrf52/framework.py

@@ -7,8 +7,7 @@ import shutil
 import sys
 import tempfile
 
-import platformdirs
-
+from esphome.build_helpers.tools_cache import SDK_NRF_TOOLS_CACHE, tools_cache_path
 import esphome.config_validation as cv
 from esphome.const import KEY_CORE, KEY_FRAMEWORK_VERSION
 from esphome.core import CORE, EsphomeError
@@ -21,7 +20,6 @@ from esphome.framework_helpers import (
     run_command_ok,
     str_to_lst_of_str,
 )
-from esphome.helpers import get_str_env
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -51,15 +49,9 @@ SDK_NG_MINIMAL_MIRRORS = str_to_lst_of_str(
 
 
 def get_sdk_nrf_tools_path() -> Path:
-    # A blank ESPHOME_SDK_NRF_PREFIX must be treated as unset: Path("")
-    # resolves to the CWD, which clean-all would then delete.
-    if prefix := get_str_env("ESPHOME_SDK_NRF_PREFIX", "").strip():
-        path = Path(prefix).expanduser()
-    else:
-        # Machine-global (OS user cache dir) so all projects share one install;
-        # see espidf.framework.get_idf_tools_path for the location rationale.
-        path = Path(platformdirs.user_cache_dir("esphome", appauthor=False)) / "sdk-nrf"
-    return path.resolve()
+    # Machine-global (OS user cache dir) so all projects share one install;
+    # see espidf.framework.get_idf_tools_path for the location rationale.
+    return tools_cache_path(*SDK_NRF_TOOLS_CACHE)
 
 
 def _needs_venv_rebuild(

esphome/config_validation.py

@@ -93,7 +93,13 @@ from esphome.core import (
 )
 from esphome.enum import StrEnum
 from esphome.expression import SUBSTITUTION_VARIABLE_PROG as VARIABLE_PROG
-from esphome.helpers import add_class_to_obj, docs_url, list_starts_with
+from esphome.helpers import (
+    FALSY_BOOL_STRINGS,
+    TRUTHY_BOOL_STRINGS,
+    add_class_to_obj,
+    docs_url,
+    list_starts_with,
+)
 from esphome.schema_extractors import (
     SCHEMA_EXTRACT,
     schema_extractor,
@@ -581,9 +587,9 @@ def boolean(value):
         return value
     if isinstance(value, str):
         value = value.lower()
-        if value in ("true", "yes", "on", "enable"):
+        if value in TRUTHY_BOOL_STRINGS:
             return True
-        if value in ("false", "no", "off", "disable"):
+        if value in FALSY_BOOL_STRINGS:
             return False
     raise Invalid(
         f"Expected boolean value, but cannot convert {value} to a boolean. Please use 'true' or 'false'"

esphome/espidf/framework.py

@@ -11,9 +11,13 @@ import re
 import shutil
 from typing import Any, NoReturn
 
-import platformdirs
-
-from esphome.core import CORE, Version
+from esphome.build_helpers.ccache import (
+    ccache_defaults_env,
+    parse_enable_env,
+    resolve_ccache_path,
+)
+from esphome.build_helpers.tools_cache import IDF_TOOLS_CACHE, tools_cache_path
+from esphome.core import Version
 from esphome.framework_helpers import (
     BatchDownloadProgress,
     PathType,
@@ -29,7 +33,7 @@ from esphome.framework_helpers import (
     run_command_ok,
     str_to_lst_of_str,
 )
-from esphome.helpers import get_bool_env, get_str_env, write_file_if_changed
+from esphome.helpers import write_file_if_changed
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -90,22 +94,10 @@ def get_idf_tools_path() -> Path:
     Returns:
         Path object pointing to the ESP-IDF tools directory
     """
-    # Treat an empty/whitespace ESPHOME_ESP_IDF_PREFIX as unset: Path("")
-    # resolves to the CWD, which would install into (and let clean-all delete)
-    # the working directory by accident.
-    if prefix := get_str_env("ESPHOME_ESP_IDF_PREFIX", "").strip():
-        path = Path(prefix).expanduser()
-    else:
-        # Machine-global so all projects share the multi-GB install instead of
-        # a per-config-directory copy. The user cache dir (not ~/.esphome)
-        # avoids colliding with data_dir when configs live in the home dir.
-        # appauthor=False drops the redundant <author>\ segment on Windows
-        # (which otherwise repeats "esphome\esphome\") to keep the path short.
-        path = Path(platformdirs.user_cache_dir("esphome", appauthor=False)) / "idf"
-    # Resolve so an unnormalized config path (e.g. compiling ``../config/x.yaml``)
-    # doesn't leave ``..`` segments in the IDF_TOOLS_PATH handed to idf.py, which
-    # otherwise warns that the venv interpreter path doesn't match the install.
-    return path.resolve()
+    # Machine-global so all projects share the multi-GB install instead of
+    # a per-config-directory copy; see build_helpers.tools_cache.tools_cache_path
+    # for the env-override and normalization rules.
+    return tools_cache_path(*IDF_TOOLS_CACHE)
 
 
 # Windows' default MAX_PATH is 260 characters. ESP-IDF toolchains nest deeply
@@ -1179,8 +1171,10 @@ def check_esp_idf_install(
 def _ccache_env() -> dict[str, str]:
     """Return ccache settings for ESP-IDF compiles.
 
-    Enabled by default whenever the ``ccache`` binary is on PATH; set
-    ``IDF_CCACHE_ENABLE=0`` in the environment to opt out. The cache lives under
+    Enabled by default whenever a runnable ``ccache`` binary is on PATH.
+    ``IDF_CCACHE_ENABLE=0`` opts out and ``=1`` forces it on; when that knob
+    is unset the shared ``ESPHOME_CCACHE_ENABLE`` applies (same 0/1 forms,
+    unrecognized values warn and count as unset). The cache lives under
     the IDF tools path (the machine-global cache dir, or
     ``ESPHOME_ESP_IDF_PREFIX``), so it is shared across all projects and removed
     by ``esphome clean-all`` along with the framework.
@@ -1195,33 +1189,23 @@ def _ccache_env() -> dict[str, str]:
     Only values the user has not already set in the environment are returned, so
     a custom ``CCACHE_DIR`` / ``CCACHE_MAXSIZE`` / etc. is respected.
     """
-    # Honor an explicit choice already in the environment (opt-out or opt-in).
-    if "IDF_CCACHE_ENABLE" in os.environ:
-        if not get_bool_env("IDF_CCACHE_ENABLE"):
-            return {}
-    elif shutil.which("ccache") is None:
+    # IDF_CCACHE_ENABLE (the backend-native knob) wins over the shared
+    # ESPHOME_CCACHE_ENABLE.
+    idf_knob = parse_enable_env("IDF_CCACHE_ENABLE")
+    if idf_knob is False:
+        # The raw value (e.g. "disable") is still inherited by idf.py via
+        # os.environ, where a non-false-constant string reads as truthy;
+        # export the canonical off spelling instead
+        return {"IDF_CCACHE_ENABLE": "0"}
+    if idf_knob is None and resolve_ccache_path() is None:
         # ESP-IDF silently skips ccache without the binary; don't enable it.
         return {}
 
-    # ccache is enabled past here. build_path is set during preload for every
-    # config-loading command, so it being unset means a caller built the IDF env
-    # too early -- fail loudly rather than silently drop CCACHE_BASEDIR (which
-    # would quietly cost cross-device cache hits).
-    if CORE.build_path is None:
-        raise ValueError(
-            "CORE.build_path must be set before constructing the ESP-IDF build "
-            "environment"
-        )
-
-    defaults = {
-        "IDF_CCACHE_ENABLE": "1",
-        "CCACHE_DIR": str(get_idf_tools_path() / "ccache"),
-        "CCACHE_NOHASHDIR": "true",
-        "CCACHE_DEPEND": "1",
-        "CCACHE_BASEDIR": str(Path(CORE.build_path).resolve()),
-    }
-    # Don't override CCACHE_* values the user already set in their environment.
-    return {k: v for k, v in defaults.items() if k not in os.environ}
+    env = ccache_defaults_env(get_idf_tools_path() / "ccache")
+    if idf_knob is None:
+        # An unparsable IDF_CCACHE_ENABLE must not leak to idf.py as truthy
+        env["IDF_CCACHE_ENABLE"] = "1"
+    return env
 
 
 def get_framework_env(

esphome/framework_helpers.py

@@ -199,6 +199,30 @@ def run_command(
         return False, None, None
 
 
+def tool_version_runs(binary: str, warning: str) -> bool:
+    """Probe ``binary --version``; on failure warn with ``warning`` % binary.
+
+    ``shutil.which`` proves existence, not runnability (Windows .bat/.cmd
+    shims, stale package-manager shims).
+    """
+    try:
+        subprocess.run(
+            [binary, "--version"],
+            check=True,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+            timeout=15,
+            # Repo-wide convention (posix_spawn fast path)
+            close_fds=False,
+        )
+    except (OSError, subprocess.SubprocessError) as err:
+        # The cause (permission denied, missing DLL, timeout) is the one
+        # detail the user needs to fix it
+        _LOGGER.warning("%s (%s)", warning % binary, err)
+        return False
+    return True
+
+
 def run_command_ok(*args, **kwargs) -> bool:
     """
     Execute a command and return only the success status.
@@ -1287,3 +1311,37 @@ def download_from_mirrors(
             f"No mirror URL template matched the provided substitutions:{details}"
         )
     raise ValueError("download_from_mirrors called with an empty mirrors list")
+
+
+def strip_win_long_path_prefix(path: str) -> str:
+    r"""Strip the Windows extended-length path prefix from ``path``.
+
+    Handles both forms documented at
+    https://learn.microsoft.com/windows/win32/fileio/naming-a-file:
+
+    * ``\\?\C:\path\to\file`` -> ``C:\path\to\file``
+    * ``\\?\UNC\server\share\path`` -> ``\\server\share\path``
+
+    The NSIS-installed ``esphome.exe`` launcher on Windows starts Python with
+    ``sys.executable`` already prefixed with ``\\?\``. That prefix propagates
+    into PlatformIO's ``$PYTHONEXE`` (PlatformIO reads ``PYTHONEXEPATH`` from
+    the environment, falling back to ``os.path.normpath(sys.executable)``)
+    and ends up baked into SCons-emitted command lines for build steps such
+    as the esp8266 ``elf2bin`` invocation. ``cmd.exe`` does not understand
+    the ``\\?\`` prefix, so the build fails with
+    "The system cannot find the path specified." Stripping the prefix early
+    keeps the path shell-quotable.
+
+    Also applied to the ccache path exported by the ccache helpers, which
+    ``shutil.which`` can return with the same prefix.
+
+    No-op on non-Windows platforms.
+    """
+    if sys.platform != "win32":
+        return path
+    if path.startswith("\\\\?\\UNC\\"):
+        # \\?\UNC\server\share\... -> \\server\share\...
+        return "\\\\" + path[len("\\\\?\\UNC\\") :]
+    if path.startswith("\\\\?\\"):
+        return path[len("\\\\?\\") :]
+    return path

esphome/helpers.py

@@ -31,6 +31,12 @@ SockAddr = IPv4SockAddr | IPv6SockAddr
 
 _LOGGER = logging.getLogger(__name__)
 
+# cv.boolean's closed spelling tables, shared with the strict env-knob
+# parser (build_helpers.ccache.parse_enable_env). The legacy get_bool_env
+# below keeps its own laxer table for backward compatibility.
+TRUTHY_BOOL_STRINGS = frozenset({"true", "yes", "on", "enable"})
+FALSY_BOOL_STRINGS = frozenset({"false", "no", "off", "disable"})
+
 IS_MACOS = platform.system() == "Darwin"
 IS_WINDOWS = platform.system() == "Windows"
 IS_LINUX = platform.system() == "Linux"

esphome/platformio/registry.py

@@ -0,0 +1,280 @@
+"""Install packages from the PlatformIO registry without importing the
+platformio package (identical bits, esphome's own download machinery)."""
+
+from __future__ import annotations
+
+from collections.abc import Collection
+import io
+import json
+import logging
+import os
+from pathlib import Path
+import platform
+
+from esphome.core import EsphomeError
+from esphome.framework_helpers import (
+    BatchDownloadProgress,
+    archive_extract_all,
+    download_from_mirrors,
+    download_with_resume,
+    rmdir,
+    run_batch_downloads,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+_REGISTRY_URL = (
+    "https://api.registry.platformio.org/v3/packages/platformio/tool/{package}"
+)
+
+
+def get_systype() -> str:
+    """The registry system tag for the current host.
+
+    Transliterates ``platformio.util.get_systype()`` (same
+    ``PLATFORMIO_SYSTEM_TYPE`` override). Deviation: windows-arm64 maps to
+    ``windows_amd64`` (no arm64 toolchains; x86 emulation).
+    """
+    if systype := os.environ.get("PLATFORMIO_SYSTEM_TYPE"):
+        return systype
+    system = platform.system().lower()
+    arch = platform.machine().lower()
+    if system == "windows":
+        if not arch:  # same fallback as upstream (platformio issue #4353)
+            arch = "x86_" + platform.architecture()[0]
+        if "x86" in arch:
+            arch = "amd64" if "64" in arch else "x86"
+        elif arch == "arm64":
+            arch = "amd64"
+    if arch == "aarch64" and platform.architecture()[0] == "32bit":
+        # 64-bit kernel with a 32-bit userland (e.g. 32-bit Raspberry Pi OS)
+        arch = "armv7l"
+    return f"{system}_{arch}" if arch else system
+
+
+def registry_download(package: str, version: str) -> tuple[str, str, int | None]:
+    """Resolve a package's download URL, sha256, and size via the registry.
+
+    The metadata fetch goes through ``download_from_mirrors`` so it shares
+    the retry, backoff, and error reporting of every other download here.
+    """
+    buf = io.BytesIO()
+    download_from_mirrors([_REGISTRY_URL], {"package": package}, buf)
+    try:
+        data = json.loads(buf.getvalue())
+    except ValueError as err:
+        raise EsphomeError(
+            f"The package registry returned invalid JSON for {package}: {err}"
+        ) from err
+    if not isinstance(data, dict):
+        raise EsphomeError(
+            f"Unexpected package registry response for {package}: {str(data)[:200]}"
+        )
+    systype = get_systype()
+    versions = data.get("versions")
+    if not isinstance(versions, list):
+        # A schema change or an error/captive-portal payload must not be
+        # reported as "version not found"
+        raise EsphomeError(
+            f"Unexpected package registry response for {package}: {str(data)[:200]}"
+        )
+    for ver in versions:
+        if not isinstance(ver, dict):
+            raise EsphomeError(
+                f"Unexpected package registry response for {package}: {str(data)[:200]}"
+            )
+        if ver.get("name") != version:
+            continue
+        files = ver.get("files")
+        if not isinstance(files, list):
+            raise EsphomeError(
+                f"Unexpected package registry response for {package}: {str(ver)[:200]}"
+            )
+        for file in files:
+            if not isinstance(file, dict):
+                raise EsphomeError(
+                    f"Unexpected package registry response for {package}: "
+                    f"{str(ver)[:200]}"
+                )
+            # Only a missing key means "any system"; an empty list must not
+            # match, and a bare string would make ``in`` a substring test.
+            systems = file.get("system")
+            if systems is None:
+                systems = ["*"]
+            elif isinstance(systems, str):
+                systems = [systems]
+            elif not isinstance(systems, list):
+                # An int would make ``in`` a TypeError and a dict a key test
+                raise EsphomeError(
+                    f"Unexpected package registry response for {package}: "
+                    f"{str(file)[:200]}"
+                )
+            if "*" in systems or systype in systems:
+                sha256 = (file.get("checksum") or {}).get("sha256")
+                if not sha256:
+                    # Never extract an unverified archive; the registry
+                    # publishes a checksum for every package file.
+                    raise EsphomeError(
+                        f"The package registry returned no sha256 for "
+                        f"{package} {version}; refusing the unverified download"
+                    )
+                url = file.get("download_url")
+                if not url:
+                    raise EsphomeError(
+                        f"The package registry returned no download URL for "
+                        f"{package} {version}"
+                    )
+                return (url, sha256, file.get("size"))
+        raise EsphomeError(
+            f"No {package} {version} build for this platform ({systype})"
+        )
+    raise EsphomeError(f"{package} {version} not found in the package registry")
+
+
+def _check_layout(name: str, dest: Path, expect: Collection[str]) -> None:
+    """Raise when an install tree is missing an expected directory (runs on
+    fresh extracts and on marker hits)."""
+    for rel in expect:
+        if not (dest / rel).is_dir():
+            raise EsphomeError(
+                f"{name} at {dest} is missing the expected {rel} "
+                "directory; run 'esphome clean-all' and retry"
+            )
+
+
+def prefetch_packages(
+    packages: list[tuple[str, str, Path, list[str]]], downloads_dir: Path
+) -> None:
+    """Download pending package archives in parallel under one combined bar.
+
+    ``packages`` holds ``(name, version, dest, mirrors)`` per package. Purely
+    an optimization: ``install_package`` verifies every archive and
+    re-downloads anything this pass left unfinished. Mirror overrides and
+    registry entries without a size stay on the sequential path so its
+    per-file bars remain trustworthy. Each fetch holds the same per-dest
+    lock as ``install_package``: the archive's ``.part`` file is shared, and
+    two concurrent writers would truncate each other's bytes.
+    """
+    from filelock import FileLock
+
+    pending: list[tuple[str, str, Path, str, str, int]] = []
+    seen: set[str] = set()
+    for name, version, dest, mirrors in packages:
+        if mirrors or (dest / ".esphome_extracted").is_file():
+            continue
+        archive_name = f"{name}-{version}"
+        if archive_name in seen:
+            # A duplicate entry would race itself between two workers
+            continue
+        seen.add(archive_name)
+        try:
+            url, sha256, size = registry_download(name, version)
+        except EsphomeError as err:
+            # The sequential install reports the real failure with context
+            _LOGGER.debug("Prefetch resolve for %s failed: %s", name, err)
+            continue
+        if not size:
+            continue
+        archive = downloads_dir / archive_name
+        if archive.is_file() and archive.stat().st_size == size:
+            continue
+        pending.append((name, version, dest, url, sha256, size))
+    if len(pending) < 2:
+        return
+    downloads_dir.mkdir(parents=True, exist_ok=True)
+    _LOGGER.info(
+        "Downloading %d package archive(s): %s",
+        len(pending),
+        ", ".join(name for name, *_ in pending),
+    )
+
+    def _fetch(entry: tuple[str, str, Path, str, str, int]):
+        name, version, dest, url, sha256, size = entry
+
+        def fetch(tracker):
+            dest.parent.mkdir(parents=True, exist_ok=True)
+            with FileLock(f"{dest}.lock", fallback_to_soft=False):
+                download_with_resume(
+                    url,
+                    downloads_dir / f"{name}-{version}",
+                    sha256=sha256,
+                    size=size,
+                    progress=tracker,
+                )
+
+        return fetch
+
+    failures = run_batch_downloads(
+        BatchDownloadProgress(
+            "Downloading packages", sum(size for *_, size in pending)
+        ),
+        [(entry[0], _fetch(entry)) for entry in pending],
+    )
+    for name, err in failures:
+        if isinstance(err, (EsphomeError, OSError)):
+            # Expected download failures: install_package retries this one
+            # itself, with a visible bar
+            _LOGGER.debug("Prefetch of %s failed: %s", name, err)
+        else:
+            # Anything else is a programming error that would otherwise
+            # become a permanent silent no-op
+            _LOGGER.warning("Prefetch of %s failed: %r", name, err)
+
+
+def install_package(
+    name: str,
+    version: str,
+    dest: Path,
+    mirrors: list[str],
+    downloads_dir: Path,
+    expect: Collection[str],
+) -> None:
+    """Download, verify, and extract one package if not already installed.
+
+    The registry path is integrity-checked against the sha256 the registry
+    publishes; a mirror override (URL templates with ``{VERSION}``/``{SYSTEM}``
+    substitution) is trusted as configured. ``downloads_dir`` holds the
+    archive between runs so an interrupted download resumes.
+    """
+    if not expect:
+        # Layout validation before marker.touch() is the only guard against
+        # caching a truncated mirror archive as a good install
+        raise ValueError("install_package requires a non-empty expect")
+    marker = dest / ".esphome_extracted"
+    if marker.is_file():
+        _check_layout(name, dest, expect)
+        return
+    from filelock import FileLock
+
+    # Serialize concurrent cold builds (same filelock pattern as git.py).
+    dest.parent.mkdir(parents=True, exist_ok=True)
+    # A soft-lock fallback would turn a hard-killed run into a permanent
+    # hang (see git.py).
+    with FileLock(f"{dest}.lock", fallback_to_soft=False):
+        if marker.is_file():
+            # Another process finished the install while we waited
+            return
+        rmdir(dest, msg=f"Clean up incomplete {name} install")
+        # Persistent location so an interrupted download resumes across runs.
+        downloads_dir.mkdir(parents=True, exist_ok=True)
+        archive = downloads_dir / f"{name}-{version}"
+        _LOGGER.info("Downloading %s %s ...", name, version)
+        if mirrors:
+            _LOGGER.warning(
+                "Downloading %s from a mirror override; checksum verification "
+                "is skipped for mirrors",
+                name,
+            )
+            download_from_mirrors(
+                mirrors, {"VERSION": version, "SYSTEM": get_systype()}, archive
+            )
+        else:
+            url, sha256, size = registry_download(name, version)
+            download_with_resume(url, archive, sha256=sha256, size=size)
+        _LOGGER.info("Extracting %s ...", name)
+        archive_extract_all(archive, dest, progress_header="Extracting")
+        # Validate the layout before recording success, so an unexpected
+        # package is never cached as a working install.
+        _check_layout(name, dest, expect)
+        marker.touch()
+        archive.unlink(missing_ok=True)

esphome/platformio/toolchain.py

@@ -4,19 +4,18 @@ import logging
 import os
 from pathlib import Path
 import re
-import shutil
-import subprocess
 import sys
 from typing import TYPE_CHECKING, Any
 
 import platformdirs
 
+from esphome.build_helpers.ccache import resolve_ccache_path
 from esphome.const import CONF_COMPILE_PROCESS_LIMIT, CONF_ESPHOME, KEY_CORE
 from esphome.core import CORE, EsphomeError
+from esphome.framework_helpers import strip_win_long_path_prefix
 from esphome.helpers import (
     add_git_ceiling_directory,
     copy_file_if_changed,
-    get_bool_env,
     rmtree,
     write_file,
 )
@@ -41,40 +40,6 @@ _PIO_PYTHON_STAMP_LOCK = ".esphome.pio.stamp.lock"
 _PIO_PYTHON_STAMP_SCHEMA = "0"
 
 
-def _strip_win_long_path_prefix(path: str) -> str:
-    r"""Strip the Windows extended-length path prefix from ``path``.
-
-    Handles both forms documented at
-    https://learn.microsoft.com/windows/win32/fileio/naming-a-file:
-
-    * ``\\?\C:\path\to\file`` -> ``C:\path\to\file``
-    * ``\\?\UNC\server\share\path`` -> ``\\server\share\path``
-
-    The NSIS-installed ``esphome.exe`` launcher on Windows starts Python with
-    ``sys.executable`` already prefixed with ``\\?\``. That prefix propagates
-    into PlatformIO's ``$PYTHONEXE`` (PlatformIO reads ``PYTHONEXEPATH`` from
-    the environment, falling back to ``os.path.normpath(sys.executable)``)
-    and ends up baked into SCons-emitted command lines for build steps such
-    as the esp8266 ``elf2bin`` invocation. ``cmd.exe`` does not understand
-    the ``\\?\`` prefix, so the build fails with
-    "The system cannot find the path specified." Stripping the prefix early
-    keeps the path shell-quotable.
-
-    Also applied to the ccache path exported by ``_ccache_env()``, which
-    ``shutil.which`` can return with the same prefix.
-
-    No-op on non-Windows platforms.
-    """
-    if sys.platform != "win32":
-        return path
-    if path.startswith("\\\\?\\UNC\\"):
-        # \\?\UNC\server\share\... -> \\server\share\...
-        return "\\\\" + path[len("\\\\?\\UNC\\") :]
-    if path.startswith("\\\\?\\"):
-        return path[len("\\\\?\\") :]
-    return path
-
-
 def get_platformio_config() -> "ProjectConfig | None":
     """Return PlatformIO's ``ProjectConfig``, or None when PlatformIO is absent."""
     try:
@@ -238,35 +203,6 @@ def _check_platformio_python_stamp(config: "ProjectConfig") -> None:
         _write_pio_stamp_python(stamp_file, current)
 
 
-def _ccache_runs(ccache: str) -> bool:
-    """Return True when the ``ccache`` found on PATH actually runs.
-
-    ``shutil.which`` proves existence, not runnability: on Windows it also
-    matches ``.bat``/``.cmd`` wrappers and stale package-manager shims whose
-    target is gone. Wrapping compiles around such a find fails every compile
-    step with an opaque OS error, so probe once and fall back to compiling
-    without ccache when the probe fails.
-    """
-    try:
-        subprocess.run(
-            [ccache, "--version"],
-            check=True,
-            stdout=subprocess.DEVNULL,
-            stderr=subprocess.DEVNULL,
-            timeout=15,
-            # Repo-wide convention (posix_spawn fast path); see the
-            # close_fds=False call sites across esphome/ and script/helpers.py
-            close_fds=False,
-        )
-    except (OSError, subprocess.SubprocessError):
-        _LOGGER.warning(
-            "Ignoring ccache at %s because it failed to run; compiling without ccache",
-            ccache,
-        )
-        return False
-    return True
-
-
 def _ccache_env() -> dict[str, str]:
     r"""Return ccache settings for PlatformIO builds.
 
@@ -285,7 +221,7 @@ def _ccache_env() -> dict[str, str]:
     runs fine through ``CreateProcess``, which is how ESP-IDF invokes it,
     but SCons runs every compile through ``cmd.exe``, which fails on it with
     "The system cannot find the path specified." (#18399), so the prefix is
-    stripped here with ``_strip_win_long_path_prefix()`` before the
+    stripped here with ``strip_win_long_path_prefix()`` before the
     runnability probe, which therefore validates the exact string the build
     will execute.
     ``ESPHOME_CCACHE_PATH`` is an internal channel, not a user setting: the
@@ -311,22 +247,8 @@ def _ccache_env() -> dict[str, str]:
     build dir. The other ``CCACHE_*`` values the user already set in the
     environment are respected.
     """
-    explicit = "ESPHOME_CCACHE_ENABLE" in os.environ
-    if explicit and not get_bool_env("ESPHOME_CCACHE_ENABLE"):
-        return {"ESPHOME_CCACHE_ENABLE": "0"}
-    ccache_path = shutil.which("ccache")
+    ccache_path = resolve_ccache_path()
     if ccache_path is None:
-        if explicit:
-            _LOGGER.warning(
-                "ESPHOME_CCACHE_ENABLE is set but no ccache binary is on PATH; "
-                "compiling without ccache"
-            )
-        return {"ESPHOME_CCACHE_ENABLE": "0"}
-    # Strip before probing so the probe validates (and the failure warning
-    # names) the exact string the build will execute through cmd.exe.
-    ccache_path = _strip_win_long_path_prefix(ccache_path)
-    # An explicit opt-in skips the runnability probe.
-    if not explicit and not _ccache_runs(ccache_path):
         return {"ESPHOME_CCACHE_ENABLE": "0"}
     env = {
         "ESPHOME_CCACHE_ENABLE": "1",
@@ -388,7 +310,7 @@ def run_platformio_cli(*args, **kwargs) -> str | int:
     # Strip the Windows extended-length path prefix from sys.executable so it
     # doesn't propagate into PlatformIO's $PYTHONEXE and break SCons-emitted
     # command lines run through cmd.exe.
-    python_exe = _strip_win_long_path_prefix(sys.executable)
+    python_exe = strip_win_long_path_prefix(sys.executable)
     if python_exe != sys.executable:
         # Only override PYTHONEXEPATH when we actually stripped a prefix.
         # PlatformIO's get_pythonexe_path() reads this and falls back to

esphome/writer.py

@@ -690,14 +690,17 @@ def clean_all(configuration: list[str]):
     # the per-config loop above can't reach. Wipe the default cache root
     # (also catches leftovers from older install layouts), then the resolved
     # install paths for the ESPHOME_*_PREFIX overrides (docker/add-on/CI)
-    # that live outside it.
+    # that live outside it. Every backend's cache is listed in
+    # TOOLS_CACHE_SPECS, so registering one there is the only step.
     import platformdirs
 
-    from esphome.components.nrf52.framework import get_sdk_nrf_tools_path
-    from esphome.espidf.framework import get_idf_tools_path
+    from esphome.build_helpers.tools_cache import TOOLS_CACHE_SPECS, tools_cache_path
 
     cache_root = Path(platformdirs.user_cache_dir("esphome", appauthor=False)).resolve()
-    for install_path in (cache_root, get_idf_tools_path(), get_sdk_nrf_tools_path()):
+    install_paths = [cache_root] + [
+        tools_cache_path(*spec) for spec in TOOLS_CACHE_SPECS
+    ]
+    for install_path in install_paths:
         if install_path.is_dir():
             _LOGGER.info("Deleting %s", install_path)
             rmtree(install_path)

requirements.txt

@@ -28,6 +28,7 @@ smpclient==7.2.0
 requests==2.34.2
 py7zr==1.1.3
 platformdirs==4.11.3  # native esp-idf toolchain global cache dir
+ninja==1.13.0  # native esp8266 arduino toolchain build driver
 filelock==3.32.3  # inter-process locks (PlatformIO cache heal, git clone cache); >=3.32 for FileLock(fallback_to_soft=...), older versions silently drop the kwarg
 
 # esp-idf >= 5.0 requires this

tests/unit_tests/build_helpers/test_ccache.py

@@ -0,0 +1,119 @@
+"""Tests for the shared ccache policy in esphome.build_helpers.ccache."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+
+from esphome.build_helpers import ccache
+
+
+def test_resolve_opt_out() -> None:
+    with patch.dict(os.environ, {"ESPHOME_CCACHE_ENABLE": "0"}):
+        assert ccache.resolve_ccache_path() is None
+
+
+def test_resolve_no_binary(caplog: pytest.LogCaptureFixture) -> None:
+    with (
+        patch.dict(os.environ, {}, clear=True),
+        patch("shutil.which", return_value=None),
+    ):
+        assert ccache.resolve_ccache_path() is None
+    assert "no ccache binary" not in caplog.text
+
+
+def test_resolve_probe_failure() -> None:
+    with (
+        patch.dict(os.environ, {}, clear=True),
+        patch("shutil.which", return_value="/usr/bin/ccache"),
+        patch("esphome.framework_helpers.subprocess.run", side_effect=OSError("boom")),
+    ):
+        assert ccache.resolve_ccache_path() is None
+
+
+def test_resolve_explicit_skips_probe_and_warns_missing(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    with (
+        patch.dict(os.environ, {"ESPHOME_CCACHE_ENABLE": "1"}, clear=True),
+        patch("shutil.which", return_value="/usr/bin/ccache"),
+        patch.object(ccache, "_ccache_runs", side_effect=AssertionError),
+    ):
+        assert ccache.resolve_ccache_path() == "/usr/bin/ccache"
+    with (
+        patch.dict(os.environ, {"ESPHOME_CCACHE_ENABLE": "1"}, clear=True),
+        patch("shutil.which", return_value=None),
+    ):
+        assert ccache.resolve_ccache_path() is None
+    assert "no ccache binary is on PATH" in caplog.text
+
+
+def test_probe_spawns_with_close_fds_false() -> None:
+    with patch("esphome.framework_helpers.subprocess.run") as mock_run:
+        assert ccache._ccache_runs("/usr/bin/ccache") is True
+    assert mock_run.call_args.kwargs["close_fds"] is False
+
+
+def test_defaults_env(tmp_path: Path) -> None:
+    with (
+        patch("esphome.core.CORE", SimpleNamespace(build_path=tmp_path / "b")),
+        patch.dict(os.environ, {"CCACHE_NOHASHDIR": "false"}, clear=True),
+    ):
+        env = ccache.ccache_defaults_env(tmp_path / "cache")
+    assert env["CCACHE_DIR"] == str(tmp_path / "cache")
+    assert env["CCACHE_DEPEND"] == "1"
+    assert "CCACHE_NOHASHDIR" not in env  # user value respected
+
+
+def test_defaults_env_requires_build_path() -> None:
+    with (
+        patch("esphome.core.CORE", SimpleNamespace(build_path=None)),
+        pytest.raises(ValueError, match="build_path"),
+    ):
+        ccache.ccache_defaults_env(Path("/x"))
+
+
+@pytest.mark.parametrize("value", ["no", "off", "false", "0"])
+def test_resolve_opt_out_synonyms(value: str) -> None:
+    """Every recognized falsy spelling disables ccache."""
+    with patch.dict(os.environ, {"ESPHOME_CCACHE_ENABLE": value}):
+        assert ccache.resolve_ccache_path() is None
+
+
+def test_resolve_unrecognized_value_warns_and_probes(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """An unparsable ESPHOME_CCACHE_ENABLE is treated as unset: it must not
+    silently enable ccache or skip the runnability probe."""
+    with (
+        patch.dict(os.environ, {"ESPHOME_CCACHE_ENABLE": "enabled"}),
+        patch("shutil.which", return_value="/usr/bin/ccache"),
+        patch.object(ccache, "_ccache_runs", return_value=False) as mock_probe,
+    ):
+        assert ccache.resolve_ccache_path() is None
+    mock_probe.assert_called_once()
+    assert "unrecognized ESPHOME_CCACHE_ENABLE" in caplog.text
+
+
+@pytest.mark.parametrize(
+    ("raw", "expected"),
+    [
+        ("1", True),
+        ("enable", True),
+        ("ON", True),
+        ("0", False),
+        ("disable", False),
+        ("Off", False),
+        ("maybe", None),
+    ],
+)
+def test_parse_enable_env_spelling_tables(
+    monkeypatch: pytest.MonkeyPatch, raw: str, expected: bool | None
+) -> None:
+    """cv.boolean's spelling tables plus the 1/0 env convention."""
+    monkeypatch.setenv("ESPHOME_CCACHE_ENABLE", raw)
+    assert ccache.parse_enable_env("ESPHOME_CCACHE_ENABLE") is expected

tests/unit_tests/build_helpers/test_ninja.py

@@ -0,0 +1,143 @@
+"""Tests for esphome.build_helpers.ninja."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+import subprocess
+import sys
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from esphome.build_helpers import ninja as ninja_helper
+from esphome.core import EsphomeError
+
+
+def test_find_ninja_prefers_path(tmp_path: Path) -> None:
+    with (
+        patch("shutil.which", return_value=str(tmp_path / "ninja")),
+        patch.object(ninja_helper, "_ninja_runs", return_value=True),
+    ):
+        assert ninja_helper.find_ninja() == tmp_path / "ninja"
+
+
+def test_find_ninja_falls_back_to_wheel(tmp_path: Path) -> None:
+    """Without a PATH entry, the ninja PyPI wheel's binary is used."""
+    binary_name = "ninja.exe" if os.name == "nt" else "ninja"
+    (tmp_path / binary_name).touch()
+    wheel = MagicMock(BIN_DIR=str(tmp_path))
+    with (
+        patch("shutil.which", return_value=None),
+        patch.dict(sys.modules, {"ninja": wheel}),
+    ):
+        assert ninja_helper.find_ninja() == tmp_path / binary_name
+
+
+def test_find_ninja_package_not_installed() -> None:
+    """A missing ninja package raises the actionable message, not ImportError."""
+    with (
+        patch("shutil.which", return_value=None),
+        patch.dict(sys.modules, {"ninja": None}),
+        pytest.raises(EsphomeError, match="ninja not found"),
+    ):
+        ninja_helper.find_ninja()
+
+
+def test_find_ninja_missing_everywhere(tmp_path: Path) -> None:
+    wheel = MagicMock(BIN_DIR=str(tmp_path))
+    with (
+        patch("shutil.which", return_value=None),
+        patch.dict(sys.modules, {"ninja": wheel}),
+        pytest.raises(EsphomeError, match="ninja not found"),
+    ):
+        ninja_helper.find_ninja()
+
+
+def test_escape_ninja_specials() -> None:
+    assert ninja_helper.escape("a b:c$d") == "a$ b$:c$$d"
+
+
+def _q(tok: str) -> str:
+    """The platform's shell_token quote wrapper (argv rule on Windows)."""
+    return f'"{tok}"' if os.name == "nt" else f"'{tok}'"
+
+
+def test_quote_arg_windows_argv_rule() -> None:
+    # Backslash runs double only before a quote (subprocess.list2cmdline rule)
+    assert ninja_helper.quote_arg('-DX=a\\"b c') == '"-DX=a\\\\\\"b c"'
+    assert ninja_helper.quote_arg("a b\\") == '"a b\\\\"'
+
+
+def test_shell_token_quotes_only_when_needed() -> None:
+    assert ninja_helper.shell_token("-Os") == "-Os"
+    assert ninja_helper.shell_token("-DP=C:\\x y") == _q("-DP=C:\\x y")
+    assert ninja_helper.shell_token("plain", force=True) == _q("plain")
+
+
+def test_shell_token_quotes_shell_metacharacters() -> None:
+    """Tokens like -DMASK=(1<<3) must not reach /bin/sh -c bare."""
+    assert ninja_helper.shell_token("-DMASK=(1<<3)") == _q("-DMASK=(1<<3)")
+    assert ninja_helper.shell_token("-DX=a;b") == _q("-DX=a;b")
+    assert ninja_helper.shell_token("-DX=$HOME") == _q("-DX=$$HOME")
+
+
+def test_shell_token_posix_roundtrips_through_sh() -> None:
+    """Backslash runs, $, backticks, and quotes must reach the compiler
+    exactly as lexed once ninja un-doubles $$ and /bin/sh strips quotes."""
+
+    if sys.platform == "win32":
+        pytest.skip("POSIX sh quoting")
+    for tok in ("-DP=a\\\\b", "-DX=$VAR", "-DY=`date`", "-DZ=it's", '-DC="q"'):
+        quoted = ninja_helper.shell_token(tok).replace("$$", "$")
+        out = subprocess.run(
+            ["/bin/sh", "-c", f'printf "%s" {quoted}'],
+            capture_output=True,
+            text=True,
+            check=True,
+        )
+        assert out.stdout == tok
+
+
+def test_quote_path_force_quotes() -> None:
+    assert ninja_helper.quote_path(Path("a b")) == _q("a b")
+    assert ninja_helper.quote_path("simple") == _q("simple")
+
+
+def test_shell_token_empty_token_is_quoted() -> None:
+    """An empty argv element must survive as an explicit pair of quotes."""
+    assert ninja_helper.shell_token("") == _q("")
+
+
+def test_find_ninja_probes_path_hit(tmp_path: Path) -> None:
+    """A broken PATH shim falls back to the wheel instead of failing every
+    build later."""
+    binary_name = "ninja.exe" if os.name == "nt" else "ninja"
+    (tmp_path / binary_name).touch()
+    wheel = MagicMock(BIN_DIR=str(tmp_path))
+    with (
+        patch("shutil.which", return_value="/broken/ninja"),
+        patch.object(ninja_helper, "_ninja_runs", return_value=False),
+        patch.dict(sys.modules, {"ninja": wheel}),
+    ):
+        assert ninja_helper.find_ninja() == tmp_path / binary_name
+
+
+def test_ninja_probe_failure_warns(caplog: pytest.LogCaptureFixture) -> None:
+    with patch("esphome.framework_helpers.subprocess.run", side_effect=OSError("boom")):
+        assert ninja_helper._ninja_runs("/broken/ninja") is False
+    assert "failed to run" in caplog.text
+
+
+def test_ninja_probe_success() -> None:
+    with patch("esphome.framework_helpers.subprocess.run") as mock_run:
+        assert ninja_helper._ninja_runs("/usr/bin/ninja") is True
+    assert mock_run.call_args.kwargs["close_fds"] is False
+
+
+def test_shell_token_windows_branch_uses_argv_rule() -> None:
+    """The nt branch quotes with the CreateProcess argv rule (the ubuntu
+    coverage run never takes it naturally)."""
+    with patch.object(os, "name", "nt"):
+        assert ninja_helper.shell_token("a b") == '"a b"'
+        assert ninja_helper.shell_token("", force=True) == '""'

tests/unit_tests/test_arduino8266_framework.py

@@ -0,0 +1,170 @@
+"""Tests for esphome.arduino8266.framework (downloads and environment)."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from esphome.arduino8266 import framework
+import esphome.config_validation as cv
+from esphome.core import CORE, EsphomeError
+
+
+@pytest.fixture(autouse=True)
+def _build_path(tmp_path: Path) -> None:
+    CORE.build_path = tmp_path
+
+
+def test_framework_package_version() -> None:
+    assert framework.framework_package_version(cv.Version(3, 1, 2)) == "3.30102.0"
+    assert framework.framework_package_version(cv.Version(3, 2, 0)) == "3.30200.0"
+    # 2.6.3+ cores use the same package-major-3 encoding (PlatformIO path)
+    assert framework.framework_package_version(cv.Version(2, 7, 4)) == "3.20704.0"
+    # A future major bump needs its own encoding, not a doomed registry lookup
+    with pytest.raises(EsphomeError, match="not supported yet"):
+        framework.framework_package_version(cv.Version(4, 0, 0))
+    # The boundary matches the PlatformIO era guard; a 2.6.2 pre-release
+    # keeps this encoding
+    with pytest.raises(EsphomeError, match="older package encoding"):
+        framework.framework_package_version(cv.Version(2, 6, 2))
+    assert framework.framework_package_version(cv.Version(2, 6, 2, "b1")) == "3.20602.0"
+    assert framework.framework_package_version(cv.Version(2, 6, 3)) == "3.20603.0"
+
+
+def test_format_framework_arduino_version_pins_all_series() -> None:
+    """The esp8266 component's PIO source formatter across every encoding
+    era, including the 4.x rejection it now shares with the installer."""
+    from esphome.components.esp8266 import _format_framework_arduino_version as fmt
+
+    assert fmt(cv.Version(2, 4, 1)) == "~1.20401.0"
+    assert fmt(cv.Version(2, 6, 2)) == "~2.20602.0"
+    assert fmt(cv.Version(2, 7, 4)) == "~3.20704.0"
+    assert fmt(cv.Version(3, 1, 2)) == "~3.30102.0"
+    # Anchored to the framework version line, not a bare EsphomeError
+    with pytest.raises(cv.Invalid, match="not supported yet") as excinfo:
+        fmt(cv.Version(4, 0, 0))
+    assert excinfo.value.path == ["version"]
+
+
+def test_tools_path_default_and_prefix(tmp_path: Path) -> None:
+    with patch.dict(os.environ, {"ESPHOME_ARDUINO8266_PREFIX": str(tmp_path)}):
+        assert framework.get_arduino8266_tools_path() == tmp_path.resolve()
+    # A blank prefix must be treated as unset, not as the CWD
+    with patch.dict(os.environ, {"ESPHOME_ARDUINO8266_PREFIX": "  "}):
+        path = framework.get_arduino8266_tools_path()
+    assert path.name == "arduino8266"
+    assert path != Path.cwd()
+
+
+def test_check_and_install_returns_paths(tmp_path: Path) -> None:
+    with (
+        patch.dict(os.environ, {"ESPHOME_ARDUINO8266_PREFIX": str(tmp_path)}),
+        patch.object(framework, "install_package") as mock_install,
+        patch.object(framework, "prefetch_packages") as mock_prefetch,
+        patch.object(framework, "find_ninja", return_value=tmp_path / "ninja"),
+    ):
+        paths = framework.check_and_install(cv.Version(3, 1, 2))
+    assert paths.framework == tmp_path / "frameworks" / "3.30102.0"
+    assert paths.toolchain == tmp_path / "toolchains" / framework.TOOLCHAIN_VERSION
+    assert paths.ninja == tmp_path / "ninja"
+    assert mock_install.call_count == 2
+    # Full argument pinning: a copy-paste swap between the two near-identical
+    # calls (mirrors, destination) must not stay green
+    fw_call, tc_call = mock_install.call_args_list
+    assert fw_call.args == (
+        framework.FRAMEWORK_PACKAGE,
+        "3.30102.0",
+        tmp_path / "frameworks" / "3.30102.0",
+        framework.ESPHOME_ARDUINO8266_FRAMEWORK_MIRRORS,
+        tmp_path / "downloads",
+    )
+    assert fw_call.kwargs["expect"] == ("cores/esp8266", "tools/sdk", "libraries")
+    assert tc_call.args == (
+        framework.TOOLCHAIN_PACKAGE,
+        framework.TOOLCHAIN_VERSION,
+        tmp_path / "toolchains" / framework.TOOLCHAIN_VERSION,
+        framework.ESPHOME_ARDUINO8266_TOOLCHAIN_MIRRORS,
+        tmp_path / "downloads",
+    )
+    assert tc_call.kwargs["expect"] == ("bin", "xtensa-lx106-elf")
+    # The prefetch sees the same package specs as the installs
+    assert mock_prefetch.call_args.args == (
+        [
+            (
+                framework.FRAMEWORK_PACKAGE,
+                "3.30102.0",
+                tmp_path / "frameworks" / "3.30102.0",
+                framework.ESPHOME_ARDUINO8266_FRAMEWORK_MIRRORS,
+            ),
+            (
+                framework.TOOLCHAIN_PACKAGE,
+                framework.TOOLCHAIN_VERSION,
+                tmp_path / "toolchains" / framework.TOOLCHAIN_VERSION,
+                framework.ESPHOME_ARDUINO8266_TOOLCHAIN_MIRRORS,
+            ),
+        ],
+        tmp_path / "downloads",
+    )
+
+
+def test_get_build_env_prepends_toolchain_bin(tmp_path: Path) -> None:
+    with patch.object(framework, "ccache_env", return_value={"CCACHE_DIR": "x"}):
+        env = framework.get_build_env(tmp_path, None)
+    assert env["PATH"].startswith(str(tmp_path / "bin") + os.pathsep)
+    assert env["CCACHE_DIR"] == "x"
+
+
+def test_ccache_env(tmp_path: Path) -> None:
+    assert framework.ccache_env(None) == {}
+    with patch.dict(os.environ, {"CCACHE_NOHASHDIR": "false"}, clear=True):
+        env = framework.ccache_env("/usr/bin/ccache")
+    # User-set values are respected; the rest get defaults
+    assert "CCACHE_NOHASHDIR" not in env
+    assert env["CCACHE_DEPEND"] == "1"
+    assert env["CCACHE_BASEDIR"] == str(Path(CORE.build_path).resolve())
+    assert env["CCACHE_DIR"].endswith("ccache")
+
+
+def test_check_and_install_rejects_old_core(tmp_path: Path) -> None:
+    """Calling the installer below the floor fails before any download."""
+    with pytest.raises(EsphomeError, match=">= 3.1.1"):
+        framework.check_and_install(cv.Version(3, 0, 2))
+
+
+def test_get_build_env_without_path_has_no_empty_entry(tmp_path: Path) -> None:
+    """An absent PATH must not leave a trailing separator (an empty entry
+    means the current directory to the shell)."""
+    with (
+        patch.dict(os.environ, {}, clear=True),
+        patch.object(framework, "ccache_env", return_value={}),
+    ):
+        env = framework.get_build_env(tmp_path, None)
+    assert env["PATH"] == str(tmp_path / "bin")
+    with (
+        patch.dict(
+            os.environ, {"PATH": f"/usr/bin{os.pathsep}{os.pathsep}/bin"}, clear=True
+        ),
+        patch.object(framework, "ccache_env", return_value={}),
+    ):
+        env = framework.get_build_env(tmp_path, None)
+    assert env["PATH"].split(os.pathsep) == [str(tmp_path / "bin"), "/usr/bin", "/bin"]
+
+
+def test_ccache_env_accepts_a_preresolved_path() -> None:
+    """The caller resolves ccache once and threads it through; None means
+    resolved-and-disabled."""
+    with patch.dict(os.environ, {}, clear=True):
+        assert framework.ccache_env(None) == {}
+        env = framework.ccache_env("/usr/bin/ccache")
+    assert env["CCACHE_DIR"].endswith("ccache")
+
+
+def test_toolchain_tool_layout(tmp_path: Path) -> None:
+    """One owner for the bin/xtensa-lx106-elf-<name> layout."""
+    tool = framework.toolchain_tool(tmp_path, "addr2line")
+    assert tool.parent == tmp_path / "bin"
+    assert tool.name.startswith("xtensa-lx106-elf-addr2line")
+    assert (tool.suffix == ".exe") is (os.name == "nt")

tests/unit_tests/test_espidf_framework.py

@@ -1553,13 +1553,14 @@ def test_get_framework_env_without_python_env_uses_os_path(tmp_path: Path) -> No
 
 def _ccache_patches(tmp_path: Path, which: str | None, build_path: Path | None):
     return (
-        patch("esphome.espidf.framework.shutil.which", return_value=which),
+        patch("esphome.espidf.framework.resolve_ccache_path", return_value=which),
         patch(
             "esphome.espidf.framework.get_idf_tools_path",
             return_value=tmp_path / "tools",
         ),
+        # ccache_defaults_env (build_helpers.ccache) reads CORE at call time
         patch(
-            "esphome.espidf.framework.CORE",
+            "esphome.core.CORE",
             SimpleNamespace(build_path=build_path),
         ),
     )
@@ -1588,7 +1589,9 @@ def test_ccache_env_opt_out_via_env(tmp_path: Path) -> None:
     # short-circuits before build_path is needed.
     p1, p2, p3 = _ccache_patches(tmp_path, "/usr/bin/ccache", None)
     with patch.dict("os.environ", {"IDF_CCACHE_ENABLE": "0"}, clear=True), p1, p2, p3:
-        assert _ccache_env() == {}
+        # The canonical off spelling is exported: the raw value is inherited
+        # by idf.py, where a spelling like "disable" would read as truthy
+        assert _ccache_env() == {"IDF_CCACHE_ENABLE": "0"}
 
 
 def test_ccache_env_opt_in_without_binary(tmp_path: Path) -> None:
@@ -1602,6 +1605,48 @@ def test_ccache_env_opt_in_without_binary(tmp_path: Path) -> None:
     assert env["CCACHE_DEPEND"] == "1"
 
 
+def test_ccache_env_honors_shared_esphome_opt_out(tmp_path: Path) -> None:
+    """ESPHOME_CCACHE_ENABLE=0 disables ccache here too; the shared policy
+    must not apply to every backend except this one."""
+    _p1, p2, p3 = _ccache_patches(tmp_path, "/usr/bin/ccache", tmp_path / "build")
+    env_vars = {"ESPHOME_CCACHE_ENABLE": "0", "PATH": "/usr/bin"}
+    with patch.dict("os.environ", env_vars, clear=True), p2, p3:
+        # The real resolver runs so the opt-out parse is exercised
+        assert _ccache_env() == {}
+
+
+@pytest.mark.parametrize("value", ["off", "no"])
+def test_ccache_env_idf_knob_parses_strictly(tmp_path: Path, value: str) -> None:
+    """IDF_CCACHE_ENABLE uses the same strict table as the shared knob, so
+    "off" disables instead of reading as truthy."""
+    p1, p2, p3 = _ccache_patches(tmp_path, "/usr/bin/ccache", tmp_path / "build")
+    with patch.dict("os.environ", {"IDF_CCACHE_ENABLE": value}, clear=True), p1, p2, p3:
+        assert _ccache_env() == {"IDF_CCACHE_ENABLE": "0"}
+
+
+def test_ccache_env_idf_knob_unrecognized_warns_and_defers(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """An unparsable IDF_CCACHE_ENABLE warns, defers to the shared resolver,
+    and is not forwarded to idf.py as truthy."""
+    p1, p2, p3 = _ccache_patches(tmp_path, "/usr/bin/ccache", tmp_path / "build")
+    env_vars = {"IDF_CCACHE_ENABLE": "enabled"}
+    with patch.dict("os.environ", env_vars, clear=True), p1, p2, p3:
+        env = _ccache_env()
+    assert "unrecognized IDF_CCACHE_ENABLE" in caplog.text
+    assert env["IDF_CCACHE_ENABLE"] == "1"
+
+
+def test_ccache_env_idf_knob_wins_over_shared_opt_out(tmp_path: Path) -> None:
+    """IDF_CCACHE_ENABLE=1 takes precedence over ESPHOME_CCACHE_ENABLE=0."""
+    p1, p2, p3 = _ccache_patches(tmp_path, None, tmp_path / "build")
+    env_vars = {"IDF_CCACHE_ENABLE": "1", "ESPHOME_CCACHE_ENABLE": "0"}
+    with patch.dict("os.environ", env_vars, clear=True), p1, p2, p3:
+        env = _ccache_env()
+    assert env["CCACHE_DIR"] == str(tmp_path / "tools" / "ccache")
+    assert "IDF_CCACHE_ENABLE" not in env
+
+
 def test_ccache_env_preserves_user_overrides(tmp_path: Path) -> None:
     # User-set CCACHE_* values must not be clobbered; unset ones still default.
     p1, p2, p3 = _ccache_patches(tmp_path, "/usr/bin/ccache", tmp_path / "build")

tests/unit_tests/test_framework_helpers.py

@@ -2194,3 +2194,37 @@ class TestGetProjectCxxCompileFlags:
     def test_empty_flags(self) -> None:
         with patch("esphome.core.CORE", _make_core_cxx(set())):
             assert get_project_cxx_compile_flags() == []
+
+
+@pytest.mark.parametrize(
+    ("platform", "input_path", "expected"),
+    [
+        # win32: drive-letter extended-length prefix is stripped
+        (
+            "win32",
+            "\\\\?\\C:\\Users\\jesse\\AppData\\Local\\ESPHome Builder\\python\\python.exe",
+            "C:\\Users\\jesse\\AppData\\Local\\ESPHome Builder\\python\\python.exe",
+        ),
+        # win32: UNC extended-length prefix is translated to a regular UNC path
+        (
+            "win32",
+            "\\\\?\\UNC\\server\\share\\python.exe",
+            "\\\\server\\share\\python.exe",
+        ),
+        # win32: paths without the prefix are returned unchanged
+        (
+            "win32",
+            "C:\\Users\\jesse\\AppData\\Local\\ESPHome Builder\\python\\python.exe",
+            "C:\\Users\\jesse\\AppData\\Local\\ESPHome Builder\\python\\python.exe",
+        ),
+        # non-win32: prefix is left alone (no-op)
+        ("linux", "\\\\?\\C:\\python.exe", "\\\\?\\C:\\python.exe"),
+        ("darwin", "/usr/bin/python3", "/usr/bin/python3"),
+    ],
+)
+def test_strip_win_long_path_prefix(
+    platform: str, input_path: str, expected: str
+) -> None:
+    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

tests/unit_tests/test_platformio_registry.py

@@ -0,0 +1,650 @@
+"""Tests for esphome.platformio.registry (PIO-registry package installs)."""
+
+from __future__ import annotations
+
+from contextlib import contextmanager
+import json
+import os
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from esphome.core import EsphomeError
+from esphome.platformio import registry
+
+
+@pytest.mark.parametrize(
+    ("system", "machine", "expected"),
+    [
+        ("Darwin", "arm64", "darwin_arm64"),
+        ("Darwin", "x86_64", "darwin_x86_64"),
+        ("Windows", "AMD64", "windows_amd64"),
+        # Deviation from upstream: auto-mapped to the emulated-x86 packages
+        ("Windows", "ARM64", "windows_amd64"),
+        ("Windows", "x86", "windows_x86"),
+        ("Linux", "x86_64", "linux_x86_64"),
+        ("Linux", "aarch64", "linux_aarch64"),
+        ("Linux", "i686", "linux_i686"),
+        ("Linux", "armv7l", "linux_armv7l"),
+        # Unknown hosts pass through like upstream; the registry lookup
+        # then fails naming the tag
+        ("FreeBSD", "amd64", "freebsd_amd64"),
+    ],
+)
+def test_get_systype(system: str, machine: str, expected: str) -> None:
+    with (
+        patch("platform.system", return_value=system),
+        patch("platform.machine", return_value=machine),
+        patch("platform.architecture", return_value=("64bit", "")),
+    ):
+        assert registry.get_systype() == expected
+
+
+def test_get_systype_env_override() -> None:
+    """PLATFORMIO_SYSTEM_TYPE wins, exactly as in upstream get_systype()."""
+    with patch.dict(os.environ, {"PLATFORMIO_SYSTEM_TYPE": "windows_amd64"}):
+        assert registry.get_systype() == "windows_amd64"
+
+
+def test_get_systype_aarch64_32bit_userland() -> None:
+    """A 32-bit userland on a 64-bit arm kernel gets armv7l binaries."""
+    with (
+        patch("platform.system", return_value="Linux"),
+        patch("platform.machine", return_value="aarch64"),
+        patch("platform.architecture", return_value=("32bit", "")),
+    ):
+        assert registry.get_systype() == "linux_armv7l"
+
+
+def test_get_systype_windows_empty_machine() -> None:
+    """An empty machine string falls back to the architecture bits."""
+    with (
+        patch("platform.system", return_value="Windows"),
+        patch("platform.machine", return_value=""),
+        patch("platform.architecture", return_value=("64bit", "")),
+    ):
+        assert registry.get_systype() == "windows_amd64"
+
+
+def _registry_response(files: list[dict]):
+    """Patch the shared downloader to serve a canned registry response."""
+    payload = {"versions": [{"name": "1.0.0", "files": files}]}
+
+    def fake_download(mirrors: list[str], substitutions: dict, target) -> str:
+        target.write(json.dumps(payload).encode())
+        return mirrors[0].format(**substitutions)
+
+    return patch.object(registry, "download_from_mirrors", side_effect=fake_download)
+
+
+def test_registry_download_uses_shared_downloader() -> None:
+    """The metadata fetch delegates its retries and error reporting to
+    download_from_mirrors; failures surface unchanged."""
+    with (
+        patch.object(
+            registry,
+            "download_from_mirrors",
+            side_effect=EsphomeError("Failed to download from all mirrors"),
+        ) as mock_download,
+        pytest.raises(EsphomeError, match="Failed to download from all mirrors"),
+    ):
+        registry.registry_download("pkg", "1.0.0")
+    (mirrors, substitutions, _), _ = mock_download.call_args
+    assert mirrors == [registry._REGISTRY_URL]
+    assert substitutions == {"package": "pkg"}
+
+
+def test_registry_download_invalid_json_is_clean() -> None:
+    def fake_download(mirrors: list[str], substitutions: dict, target) -> str:
+        target.write(b"<html>not json</html>")
+        return "http://x"
+
+    with (
+        patch.object(registry, "download_from_mirrors", side_effect=fake_download),
+        pytest.raises(EsphomeError, match="invalid JSON"),
+    ):
+        registry.registry_download("pkg", "1.0.0")
+
+
+def test_registry_download_matches_system() -> None:
+    with (
+        _registry_response(
+            [
+                {"system": ["windows_amd64"], "download_url": "http://x/win"},
+                {
+                    "system": ["linux_x86_64"],
+                    "download_url": "http://x/linux",
+                    "checksum": {"sha256": "abc123"},
+                    "size": 42,
+                },
+            ]
+        ),
+        patch.object(registry, "get_systype", return_value="linux_x86_64"),
+    ):
+        assert registry.registry_download("pkg", "1.0.0") == (
+            "http://x/linux",
+            "abc123",
+            42,
+        )
+
+
+def test_registry_download_bare_string_system() -> None:
+    """A bare-string system tag is an exact match, not a substring test."""
+    with (
+        _registry_response(
+            [
+                {"system": "linux_x86", "download_url": "http://x/x86"},
+                {
+                    "system": "linux_x86_64",
+                    "download_url": "http://x/x86_64",
+                    "checksum": {"sha256": "abc"},
+                },
+            ]
+        ),
+        patch.object(registry, "get_systype", return_value="linux_x86_64"),
+    ):
+        assert registry.registry_download("pkg", "1.0.0")[0] == "http://x/x86_64"
+
+
+def test_registry_download_wildcard_system() -> None:
+    with _registry_response(
+        [
+            {
+                "system": "*",
+                "download_url": "http://x/any",
+                "checksum": {"sha256": "abc"},
+                "size": 7,
+            }
+        ]
+    ):
+        assert registry.registry_download("pkg", "1.0.0") == (
+            "http://x/any",
+            "abc",
+            7,
+        )
+
+
+def test_registry_download_missing_checksum_raises() -> None:
+    """An unverifiable archive is refused, never silently extracted."""
+    with (
+        _registry_response([{"system": "*", "download_url": "http://x/any"}]),
+        pytest.raises(EsphomeError, match="no sha256"),
+    ):
+        registry.registry_download("pkg", "1.0.0")
+
+
+def test_registry_download_no_system_match() -> None:
+    with (
+        _registry_response(
+            [{"system": ["windows_amd64"], "download_url": "http://x/win"}]
+        ),
+        patch.object(registry, "get_systype", return_value="linux_x86_64"),
+        pytest.raises(EsphomeError, match="No pkg 1.0.0 build"),
+    ):
+        registry.registry_download("pkg", "1.0.0")
+
+
+def test_registry_download_version_not_found() -> None:
+    def fake_download(mirrors: list[str], substitutions: dict, target) -> str:
+        target.write(
+            json.dumps({"versions": [{"name": "2.0.0", "files": []}]}).encode()
+        )
+        return "http://x"
+
+    with (
+        patch.object(registry, "download_from_mirrors", side_effect=fake_download),
+        pytest.raises(EsphomeError, match="not found"),
+    ):
+        registry.registry_download("pkg", "1.0.0")
+
+
+def test_install_package_skips_when_marker_exists(tmp_path: Path) -> None:
+    dest = tmp_path / "pkg"
+    (dest / "payload").mkdir(parents=True)
+    (dest / ".esphome_extracted").touch()
+    with patch.object(registry, "download_from_mirrors") as mock_download:
+        registry.install_package(
+            "pkg", "1.0.0", dest, [], tmp_path / "dl", expect=("payload",)
+        )
+    mock_download.assert_not_called()
+
+
+def test_install_package_marker_hit_rechecks_layout(tmp_path: Path) -> None:
+    """A marked install that later lost files fails by name instead of
+    surfacing as an opaque toolchain error."""
+    dest = tmp_path / "pkg"
+    dest.mkdir()
+    (dest / ".esphome_extracted").touch()
+    with pytest.raises(EsphomeError, match="missing the expected payload"):
+        registry.install_package(
+            "pkg", "1.0.0", dest, [], tmp_path / "dl", expect=("payload",)
+        )
+
+
+def test_install_package_downloads_via_mirrors(tmp_path: Path) -> None:
+    dest = tmp_path / "pkg"
+    mirrors = ["http://mirror/{VERSION}/{SYSTEM}.tar.gz"]
+    with (
+        patch.object(registry, "download_from_mirrors") as mock_download,
+        patch.object(registry, "archive_extract_all") as mock_extract,
+        patch.object(registry, "get_systype", return_value="linux_x86_64"),
+    ):
+        # Extraction is expected to create the directory
+        mock_extract.side_effect = lambda *_a, **_kw: (dest / "payload").mkdir(
+            parents=True
+        )
+        registry.install_package(
+            "pkg", "1.0.0", dest, mirrors, tmp_path / "dl", expect=("payload",)
+        )
+    assert mock_download.call_args[0][0] is mirrors
+    assert mock_download.call_args[0][1] == {
+        "VERSION": "1.0.0",
+        "SYSTEM": "linux_x86_64",
+    }
+    assert (dest / ".esphome_extracted").is_file()
+
+
+def test_install_package_downloads_via_registry(tmp_path: Path) -> None:
+    """The registry path downloads with the registry's sha256 and size."""
+    dest = tmp_path / "pkg"
+    with (
+        patch.object(registry, "download_with_resume") as mock_download,
+        patch.object(registry, "archive_extract_all") as mock_extract,
+        patch.object(
+            registry,
+            "registry_download",
+            return_value=("http://x/pkg.tar.gz", "abc123", 42),
+        ),
+    ):
+        mock_extract.side_effect = lambda *_a, **_kw: (dest / "payload").mkdir(
+            parents=True
+        )
+        registry.install_package(
+            "pkg", "1.0.0", dest, [], tmp_path / "dl", expect=("payload",)
+        )
+    assert mock_download.call_args[0][0] == "http://x/pkg.tar.gz"
+    assert mock_download.call_args[1] == {"sha256": "abc123", "size": 42}
+
+
+def test_install_package_validates_expected_layout(tmp_path: Path) -> None:
+    """The success marker is only written when the extracted tree is usable."""
+    dest = tmp_path / "pkg"
+    with (
+        patch.object(registry, "download_from_mirrors"),
+        patch.object(registry, "archive_extract_all") as mock_extract,
+        patch.object(registry, "get_systype", return_value="linux_x86_64"),
+    ):
+        mock_extract.side_effect = lambda *_a, **_kw: (dest / "bin").mkdir(parents=True)
+        registry.install_package(
+            "pkg", "1.0.0", dest, ["http://m"], tmp_path / "dl", expect=("bin",)
+        )
+    assert (dest / ".esphome_extracted").is_file()
+
+
+def test_install_package_unexpected_layout_raises(tmp_path: Path) -> None:
+    dest = tmp_path / "pkg"
+    with (
+        patch.object(registry, "download_from_mirrors"),
+        patch.object(registry, "archive_extract_all") as mock_extract,
+        patch.object(registry, "get_systype", return_value="linux_x86_64"),
+        pytest.raises(EsphomeError, match="missing the expected bin"),
+    ):
+        mock_extract.side_effect = lambda *_a, **_kw: (dest / "payload").mkdir(
+            parents=True
+        )
+        registry.install_package(
+            "pkg", "1.0.0", dest, ["http://m"], tmp_path / "dl", expect=("bin",)
+        )
+    assert not (dest / ".esphome_extracted").exists()
+
+
+def test_install_package_marker_rechecked_under_lock(tmp_path: Path) -> None:
+    """A concurrent install finishing while we wait for the lock is detected."""
+    dest = tmp_path / "pkg"
+    marker = dest / ".esphome_extracted"
+
+    @contextmanager
+    def _fake_lock(*_a, **_kw):
+        dest.mkdir(parents=True, exist_ok=True)
+        marker.touch()
+        yield
+
+    with (
+        patch("filelock.FileLock", _fake_lock),
+        patch.object(registry, "download_from_mirrors") as mock_download,
+        patch.object(registry, "rmdir") as mock_rmdir,
+    ):
+        registry.install_package(
+            "pkg", "1.0.0", dest, ["http://m"], tmp_path / "dl", expect=("payload",)
+        )
+    mock_download.assert_not_called()
+    mock_rmdir.assert_not_called()
+
+
+def test_install_package_uses_hard_lock(tmp_path: Path) -> None:
+    """The install lock must never degrade to a soft (existence) lock."""
+    dest = tmp_path / "pkg"
+    with (
+        patch("filelock.FileLock") as mock_lock,
+        patch.object(registry, "download_from_mirrors"),
+        patch.object(registry, "archive_extract_all") as mock_extract,
+        patch.object(registry, "get_systype", return_value="linux_x86_64"),
+    ):
+        mock_extract.side_effect = lambda *_a, **_kw: (dest / "payload").mkdir(
+            parents=True, exist_ok=True
+        )
+        registry.install_package(
+            "pkg", "1.0.0", dest, ["http://m"], tmp_path / "dl", expect=("payload",)
+        )
+    assert mock_lock.call_args.kwargs["fallback_to_soft"] is False
+
+
+def test_registry_download_empty_system_list_does_not_match() -> None:
+    """An explicitly empty system list must not act as a wildcard."""
+    with (
+        _registry_response([{"system": [], "download_url": "http://x/any"}]),
+        patch.object(registry, "get_systype", return_value="linux_x86_64"),
+        pytest.raises(EsphomeError, match="No pkg 1.0.0 build"),
+    ):
+        registry.registry_download("pkg", "1.0.0")
+
+
+def test_registry_download_unexpected_payload_is_named() -> None:
+    """An error envelope without a versions list is not 'version not found'."""
+
+    def fake_download(mirrors: list[str], substitutions: dict, target) -> str:
+        target.write(json.dumps({"message": "rate limited"}).encode())
+        return "http://x"
+
+    with (
+        patch.object(registry, "download_from_mirrors", side_effect=fake_download),
+        pytest.raises(EsphomeError, match="Unexpected package registry response"),
+    ):
+        registry.registry_download("pkg", "1.0.0")
+
+
+def test_registry_download_missing_system_key_matches_any() -> None:
+    """A file with no system key at all serves every host."""
+    with _registry_response(
+        [{"download_url": "http://x/any", "checksum": {"sha256": "abc"}, "size": 1}]
+    ):
+        assert registry.registry_download("pkg", "1.0.0") == ("http://x/any", "abc", 1)
+
+
+def test_registry_download_missing_files_list_is_named() -> None:
+    """A version entry without a files list is an unexpected payload, not a
+    missing platform build."""
+    with (
+        _registry_response(None),
+        pytest.raises(EsphomeError, match="Unexpected package registry response"),
+    ):
+        registry.registry_download("pkg", "1.0.0")
+
+
+def test_registry_download_missing_download_url_is_named() -> None:
+    with (
+        _registry_response([{"system": "*", "checksum": {"sha256": "abc"}, "size": 1}]),
+        pytest.raises(EsphomeError, match="no download URL"),
+    ):
+        registry.registry_download("pkg", "1.0.0")
+
+
+def test_install_package_empty_expect_rejected(tmp_path: Path) -> None:
+    """Layout validation is the only guard before marker.touch(), so an
+    empty expect is a caller bug, not a lenient install."""
+    with pytest.raises(ValueError, match="non-empty expect"):
+        registry.install_package(
+            "pkg", "1.0.0", tmp_path / "pkg", [], tmp_path / "dl", expect=()
+        )
+
+
+def test_registry_download_non_dict_version_entry_is_named() -> None:
+    """A versions list of bare strings is an unexpected payload, not an
+    AttributeError traceback."""
+
+    def fake_download(mirrors: list[str], substitutions: dict, target) -> str:
+        target.write(json.dumps({"versions": ["1.0.0", "2.0.0"]}).encode())
+        return "http://x"
+
+    with (
+        patch.object(registry, "download_from_mirrors", side_effect=fake_download),
+        pytest.raises(EsphomeError, match="Unexpected package registry response"),
+    ):
+        registry.registry_download("pkg", "1.0.0")
+
+
+def test_registry_download_non_dict_file_entry_is_named() -> None:
+    def fake_download(mirrors: list[str], substitutions: dict, target) -> str:
+        target.write(
+            json.dumps(
+                {"versions": [{"name": "1.0.0", "files": ["a.tar.gz"]}]}
+            ).encode()
+        )
+        return "http://x"
+
+    with (
+        patch.object(registry, "download_from_mirrors", side_effect=fake_download),
+        pytest.raises(EsphomeError, match="Unexpected package registry response"),
+    ):
+        registry.registry_download("pkg", "1.0.0")
+
+
+def test_registry_download_non_dict_payload_is_named() -> None:
+    """A JSON array answer is an unexpected payload at the outermost level."""
+
+    def fake_download(mirrors: list[str], substitutions: dict, target) -> str:
+        target.write(json.dumps(["1.0.0"]).encode())
+        return "http://x"
+
+    with (
+        patch.object(registry, "download_from_mirrors", side_effect=fake_download),
+        pytest.raises(EsphomeError, match="Unexpected package registry response"),
+    ):
+        registry.registry_download("pkg", "1.0.0")
+
+
+def test_registry_download_non_list_system_is_named() -> None:
+    """A system field that is neither missing, str, nor list is an
+    unexpected payload, not a TypeError from the ``in`` test."""
+    with (
+        _registry_response([{"system": 5, "checksum": {"sha256": "abc"}, "size": 1}]),
+        pytest.raises(EsphomeError, match="Unexpected package registry response"),
+    ):
+        registry.registry_download("pkg", "1.0.0")
+
+
+def _resolve_for(sizes: dict[str, int | None]):
+    def resolve(name: str, version: str):
+        size = sizes[name]
+        if size == -1:
+            raise EsphomeError("registry down")
+        return (f"http://x/{name}.tar.gz", "abc123", size)
+
+    return resolve
+
+
+def test_prefetch_packages_downloads_pending_in_parallel(tmp_path: Path) -> None:
+    """Two uninstalled packages download together under one combined bar,
+    with the registry's sha256 and size and a batch progress tracker."""
+    with (
+        patch.object(registry, "download_with_resume") as mock_download,
+        patch.object(
+            registry, "registry_download", side_effect=_resolve_for({"a": 10, "b": 20})
+        ),
+    ):
+        registry.prefetch_packages(
+            [
+                ("a", "1.0", tmp_path / "a", []),
+                ("b", "2.0", tmp_path / "b", []),
+            ],
+            tmp_path / "dl",
+        )
+    assert mock_download.call_count == 2
+    # Locking makes worker completion order nondeterministic
+    calls = sorted(mock_download.call_args_list, key=lambda c: c[0][0])
+    for call, (name, version, size) in zip(
+        calls, [("a", "1.0", 10), ("b", "2.0", 20)], strict=True
+    ):
+        assert call[0][0] == f"http://x/{name}.tar.gz"
+        assert call[0][1] == tmp_path / "dl" / f"{name}-{version}"
+        assert call[1]["sha256"] == "abc123"
+        assert call[1]["size"] == size
+        assert callable(call[1]["progress"])
+
+
+def test_prefetch_packages_dedupes_duplicate_entries(tmp_path: Path) -> None:
+    """Duplicate (name, version) entries would race each other between two
+    workers; only one survives (and one is too few to parallelize)."""
+    with (
+        patch.object(registry, "download_with_resume") as mock_download,
+        patch.object(
+            registry, "registry_download", side_effect=_resolve_for({"a": 10})
+        ),
+    ):
+        registry.prefetch_packages(
+            [
+                ("a", "1.0", tmp_path / "a", []),
+                ("a", "1.0", tmp_path / "a", []),
+            ],
+            tmp_path / "dl",
+        )
+    mock_download.assert_not_called()
+
+
+def test_prefetch_packages_single_pending_skips(tmp_path: Path) -> None:
+    """One pending package has nothing to parallelize; the sequential
+    install keeps its own bar."""
+    marker_dest = tmp_path / "a"
+    marker_dest.mkdir()
+    (marker_dest / ".esphome_extracted").touch()
+    with (
+        patch.object(registry, "download_with_resume") as mock_download,
+        patch.object(
+            registry, "registry_download", side_effect=_resolve_for({"b": 20})
+        ),
+    ):
+        registry.prefetch_packages(
+            [
+                ("a", "1.0", marker_dest, []),
+                ("b", "2.0", tmp_path / "b", []),
+            ],
+            tmp_path / "dl",
+        )
+    mock_download.assert_not_called()
+
+
+def test_prefetch_packages_mirror_and_sizeless_stay_sequential(
+    tmp_path: Path,
+) -> None:
+    """Mirror overrides and size-less registry entries are left to the
+    sequential path so its per-file bars stay trustworthy."""
+    with (
+        patch.object(registry, "download_with_resume") as mock_download,
+        patch.object(
+            registry,
+            "registry_download",
+            side_effect=_resolve_for({"b": None, "c": 30}),
+        ),
+    ):
+        registry.prefetch_packages(
+            [
+                ("a", "1.0", tmp_path / "a", ["http://mirror/{VERSION}"]),
+                ("b", "2.0", tmp_path / "b", []),
+                ("c", "3.0", tmp_path / "c", []),
+            ],
+            tmp_path / "dl",
+        )
+    mock_download.assert_not_called()
+
+
+def test_prefetch_packages_resolve_failure_defers_to_install(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A registry failure only skips the prefetch; install_package reports
+    the real error with context."""
+    caplog.set_level("DEBUG")
+    with (
+        patch.object(registry, "download_with_resume") as mock_download,
+        patch.object(
+            registry, "registry_download", side_effect=_resolve_for({"a": -1, "b": 20})
+        ),
+    ):
+        registry.prefetch_packages(
+            [
+                ("a", "1.0", tmp_path / "a", []),
+                ("b", "2.0", tmp_path / "b", []),
+            ],
+            tmp_path / "dl",
+        )
+    mock_download.assert_not_called()
+    assert "Prefetch resolve for a failed" in caplog.text
+
+
+def test_prefetch_packages_complete_archive_skipped(tmp_path: Path) -> None:
+    """An archive already fully downloaded is not re-fetched."""
+    dl = tmp_path / "dl"
+    dl.mkdir()
+    (dl / "a-1.0").write_bytes(b"x" * 10)
+    with (
+        patch.object(registry, "download_with_resume") as mock_download,
+        patch.object(
+            registry, "registry_download", side_effect=_resolve_for({"a": 10, "b": 20})
+        ),
+    ):
+        registry.prefetch_packages(
+            [
+                ("a", "1.0", tmp_path / "a", []),
+                ("b", "2.0", tmp_path / "b", []),
+            ],
+            dl,
+        )
+    mock_download.assert_not_called()
+
+
+def test_prefetch_packages_download_failure_is_debug(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A failed prefetch download is logged and left for install_package."""
+    caplog.set_level("DEBUG")
+    with (
+        patch.object(
+            registry, "download_with_resume", side_effect=OSError("boom")
+        ) as mock_download,
+        patch.object(
+            registry, "registry_download", side_effect=_resolve_for({"a": 10, "b": 20})
+        ),
+    ):
+        registry.prefetch_packages(
+            [
+                ("a", "1.0", tmp_path / "a", []),
+                ("b", "2.0", tmp_path / "b", []),
+            ],
+            tmp_path / "dl",
+        )
+    assert mock_download.call_count == 2
+    assert "Prefetch of a failed" in caplog.text
+    assert "Prefetch of b failed" in caplog.text
+
+
+def test_prefetch_packages_unexpected_failure_warns(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A programming error (not a download failure) surfaces at WARNING
+    instead of becoming a permanent silent no-op."""
+    with (
+        patch.object(
+            registry, "download_with_resume", side_effect=TypeError("bad call")
+        ),
+        patch.object(
+            registry, "registry_download", side_effect=_resolve_for({"a": 10, "b": 20})
+        ),
+    ):
+        registry.prefetch_packages(
+            [
+                ("a", "1.0", tmp_path / "a", []),
+                ("b", "2.0", tmp_path / "b", []),
+            ],
+            tmp_path / "dl",
+        )
+    assert "TypeError" in caplog.text

tests/unit_tests/test_platformio_toolchain.py

@@ -431,8 +431,8 @@ def test_ccache_env_enabled_by_default(setup_core: Path) -> None:
 
     with (
         patch.dict(os.environ, {}, clear=True),
-        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
-        patch.object(toolchain.subprocess, "run"),
+        patch("shutil.which", return_value="/usr/bin/ccache"),
+        patch("esphome.framework_helpers.subprocess.run"),
     ):
         env = toolchain._ccache_env()
 
@@ -469,7 +469,7 @@ def test_ccache_env_disabled_without_binary(
 
     with (
         patch.dict(os.environ, env_vars, clear=True),
-        patch.object(toolchain.shutil, "which", return_value=None),
+        patch("shutil.which", return_value=None),
         caplog.at_level("WARNING"),
     ):
         env = toolchain._ccache_env()
@@ -494,8 +494,8 @@ def test_ccache_env_disabled_when_probe_fails(
 
     with (
         patch.dict(os.environ, {}, clear=True),
-        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
-        patch.object(toolchain.subprocess, "run", side_effect=probe_error),
+        patch("shutil.which", return_value="/usr/bin/ccache"),
+        patch("esphome.framework_helpers.subprocess.run", side_effect=probe_error),
     ):
         env = toolchain._ccache_env()
 
@@ -508,8 +508,8 @@ def test_ccache_env_forced_on_skips_probe(setup_core: Path) -> None:
 
     with (
         patch.dict(os.environ, {"ESPHOME_CCACHE_ENABLE": "1"}, clear=True),
-        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
-        patch.object(toolchain.subprocess, "run") as mock_probe,
+        patch("shutil.which", return_value="/usr/bin/ccache"),
+        patch("esphome.framework_helpers.subprocess.run") as mock_probe,
     ):
         env = toolchain._ccache_env()
 
@@ -537,9 +537,9 @@ def test_ccache_env_strips_win_long_path_prefix(setup_core: Path) -> None:
         patch.dict(os.environ, {}, clear=True),
         # shutil.which is patched, so the win32 code path of the real
         # implementation (which crashes on a POSIX host) is never reached.
-        patch("esphome.platformio.toolchain.sys.platform", "win32"),
-        patch.object(toolchain.shutil, "which", return_value=prefixed),
-        patch.object(toolchain.subprocess, "run") as mock_probe,
+        patch("esphome.framework_helpers.sys.platform", "win32"),
+        patch("shutil.which", return_value=prefixed),
+        patch("esphome.framework_helpers.subprocess.run") as mock_probe,
     ):
         env = toolchain._ccache_env()
 
@@ -555,7 +555,7 @@ def test_ccache_env_opt_out(setup_core: Path) -> None:
 
     with (
         patch.dict(os.environ, {"ESPHOME_CCACHE_ENABLE": "0"}, clear=True),
-        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
+        patch("shutil.which", return_value="/usr/bin/ccache"),
     ):
         env = toolchain._ccache_env()
 
@@ -568,7 +568,7 @@ def test_ccache_env_normalizes_enable_value(setup_core: Path) -> None:
 
     with (
         patch.dict(os.environ, {"ESPHOME_CCACHE_ENABLE": "yes"}, clear=True),
-        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
+        patch("shutil.which", return_value="/usr/bin/ccache"),
     ):
         env = toolchain._ccache_env()
 
@@ -587,8 +587,8 @@ def test_ccache_env_respects_user_values_and_refreshes_basedir(
 
     with (
         patch.dict(os.environ, user_env, clear=True),
-        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
-        patch.object(toolchain.subprocess, "run"),
+        patch("shutil.which", return_value="/usr/bin/ccache"),
+        patch("esphome.framework_helpers.subprocess.run"),
     ):
         env = toolchain._ccache_env()
 
@@ -606,8 +606,8 @@ def test_run_platformio_cli_passes_ccache_env_to_subprocess_only(
 
     with (
         patch.dict(os.environ, {}, clear=False),
-        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
-        patch.object(toolchain.subprocess, "run"),
+        patch("shutil.which", return_value="/usr/bin/ccache"),
+        patch("esphome.framework_helpers.subprocess.run"),
     ):
         os.environ.pop("ESPHOME_CCACHE_ENABLE", None)
         mock_run_external_process.return_value = 0
@@ -628,8 +628,8 @@ def test_ccache_env_requires_build_path(setup_core: Path) -> None:
 
     with (
         patch.dict(os.environ, {}, clear=True),
-        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
-        patch.object(toolchain.subprocess, "run"),
+        patch("shutil.which", return_value="/usr/bin/ccache"),
+        patch("esphome.framework_helpers.subprocess.run"),
         pytest.raises(ValueError, match="CORE.build_path must be set"),
     ):
         toolchain._ccache_env()
@@ -642,8 +642,8 @@ def test_run_platformio_cli_merges_caller_env(
     CORE.build_path = str(setup_core / "build" / "test")
 
     with (
-        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
-        patch.object(toolchain.subprocess, "run"),
+        patch("shutil.which", return_value="/usr/bin/ccache"),
+        patch("esphome.framework_helpers.subprocess.run"),
     ):
         mock_run_external_process.return_value = 0
         toolchain.run_platformio_cli(
@@ -800,9 +800,7 @@ def test_ccache_env_real_probe_runs_stripped_path(setup_core: Path) -> None:
 
     with (
         patch.dict(os.environ, {}, clear=False),
-        patch.object(
-            toolchain.shutil, "which", return_value="\\\\?\\" + sys.executable
-        ),
+        patch("shutil.which", return_value="\\\\?\\" + sys.executable),
     ):
         os.environ.pop("ESPHOME_CCACHE_ENABLE", None)
         env = toolchain._ccache_env()
@@ -843,40 +841,6 @@ def test_ccache_wrapper_through_cmd_exe(
         assert marker.read_text() == "compiled"
 
 
-@pytest.mark.parametrize(
-    ("platform", "input_path", "expected"),
-    [
-        # win32: drive-letter extended-length prefix is stripped
-        (
-            "win32",
-            "\\\\?\\C:\\Users\\jesse\\AppData\\Local\\ESPHome Builder\\python\\python.exe",
-            "C:\\Users\\jesse\\AppData\\Local\\ESPHome Builder\\python\\python.exe",
-        ),
-        # win32: UNC extended-length prefix is translated to a regular UNC path
-        (
-            "win32",
-            "\\\\?\\UNC\\server\\share\\python.exe",
-            "\\\\server\\share\\python.exe",
-        ),
-        # win32: paths without the prefix are returned unchanged
-        (
-            "win32",
-            "C:\\Users\\jesse\\AppData\\Local\\ESPHome Builder\\python\\python.exe",
-            "C:\\Users\\jesse\\AppData\\Local\\ESPHome Builder\\python\\python.exe",
-        ),
-        # non-win32: prefix is left alone (no-op)
-        ("linux", "\\\\?\\C:\\python.exe", "\\\\?\\C:\\python.exe"),
-        ("darwin", "/usr/bin/python3", "/usr/bin/python3"),
-    ],
-)
-def test_strip_win_long_path_prefix(
-    platform: str, input_path: str, expected: str
-) -> None:
-    r"""``\\?\`` and ``\\?\UNC\`` prefixes are stripped only on win32."""
-    with patch("esphome.platformio.toolchain.sys.platform", platform):
-        assert toolchain._strip_win_long_path_prefix(input_path) == expected
-
-
 def test_run_platformio_cli_strips_win_long_path_prefix(
     setup_core: Path, mock_run_external_process: Mock
 ) -> None:
@@ -900,7 +864,7 @@ def test_run_platformio_cli_strips_win_long_path_prefix(
         # so the stdlib sees it too) would send shutil.which down the Windows
         # code path, which crashes on a POSIX host.
         patch.dict(os.environ, {"ESPHOME_CCACHE_ENABLE": "0"}, clear=False),
-        patch("esphome.platformio.toolchain.sys.platform", "win32"),
+        patch("esphome.framework_helpers.sys.platform", "win32"),
         patch("esphome.platformio.toolchain.sys.executable", prefixed_exe),
     ):
         # Pop any pre-existing PYTHONEXEPATH so the assertion below reflects
@@ -932,7 +896,7 @@ def test_run_platformio_cli_does_not_set_pythonexepath_without_strip(
 
     with (
         patch.dict(os.environ, {}, clear=False),
-        patch("esphome.platformio.toolchain.sys.platform", "linux"),
+        patch("esphome.framework_helpers.sys.platform", "linux"),
         patch("esphome.platformio.toolchain.sys.executable", plain_exe),
     ):
         os.environ.pop("PYTHONEXEPATH", None)
@@ -1977,10 +1941,3 @@ def test_run_platformio_cli_invokes_heal(
     with patch.object(toolchain, "heal_platformio_python_env") as mock_heal:
         toolchain.run_platformio_cli("test")
     mock_heal.assert_called_once()
-
-
-def test_ccache_probe_spawns_with_close_fds_false() -> None:
-    """The probe follows the repo-wide posix_spawn convention."""
-    with patch("subprocess.run") as mock_run:
-        assert toolchain._ccache_runs("/usr/bin/ccache") is True
-    assert mock_run.call_args.kwargs["close_fds"] is False

tests/unit_tests/test_writer.py

@@ -68,7 +68,8 @@ def _isolate_platformio_paths(tmp_path_factory: pytest.TempPathFactory) -> Any:
     test_clean_all_partial_exists) install their own inner patch which
     stacks on top of this one and wins for the duration of their block.
 
-    Also pin ``ESPHOME_ESP_IDF_PREFIX`` and ``ESPHOME_SDK_NRF_PREFIX`` to
+    Also pin ``ESPHOME_ESP_IDF_PREFIX``, ``ESPHOME_SDK_NRF_PREFIX`` and
+    ``ESPHOME_ARDUINO8266_PREFIX`` to
     nonexistent tmp dirs, and patch ``platformdirs.user_cache_dir``, for the
     same reason: ``clean_all`` removes the machine-global toolchain installs
     and their default cache root, which otherwise resolve to the real
@@ -77,6 +78,7 @@ def _isolate_platformio_paths(tmp_path_factory: pytest.TempPathFactory) -> Any:
     pio_root = tmp_path_factory.mktemp("isolated_pio") / "nonexistent"
     idf_root = tmp_path_factory.mktemp("isolated_idf") / "nonexistent"
     sdk_nrf_root = tmp_path_factory.mktemp("isolated_sdk_nrf") / "nonexistent"
+    arduino8266_root = tmp_path_factory.mktemp("isolated_arduino8266") / "nonexistent"
     cache_root = tmp_path_factory.mktemp("isolated_cache") / "nonexistent"
     mock_cfg = MagicMock()
     mock_cfg.get.side_effect = lambda section, option: (
@@ -92,6 +94,7 @@ def _isolate_platformio_paths(tmp_path_factory: pytest.TempPathFactory) -> Any:
             {
                 "ESPHOME_ESP_IDF_PREFIX": str(idf_root),
                 "ESPHOME_SDK_NRF_PREFIX": str(sdk_nrf_root),
+                "ESPHOME_ARDUINO8266_PREFIX": str(arduino8266_root),
             },
         ),
         patch("platformdirs.user_cache_dir", return_value=str(cache_root)),
@@ -1033,6 +1036,28 @@ def test_clean_all_removes_global_idf_install(
     assert str(idf_install.resolve()) in caplog.text
 
 
+@patch("esphome.writer.CORE")
+def test_clean_all_removes_global_arduino8266_install(
+    mock_core: MagicMock,
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """clean_all removes the machine-global native arduino8266 install dir."""
+    arduino8266_install = tmp_path / "arduino8266_install"
+    (arduino8266_install / "frameworks").mkdir(parents=True)
+    monkeypatch.setenv("ESPHOME_ARDUINO8266_PREFIX", str(arduino8266_install))
+
+    config_dir = tmp_path / "config"
+    config_dir.mkdir()
+
+    with caplog.at_level("INFO"):
+        clean_all([str(config_dir)])
+
+    assert not arduino8266_install.exists()
+    assert str(arduino8266_install.resolve()) in caplog.text
+
+
 @patch("esphome.writer.CORE")
 def test_clean_all_removes_global_sdk_nrf_install(
     mock_core: MagicMock,