[core] Auto-clean the PlatformIO build environment when the Python version changes (#17671)

Co-authored-by: pre-commit-ci-lite[bot] <117423508+pre-commit-ci-lite[bot]@users.noreply.github.com>

esphome/platformio/toolchain.py

@@ -1,17 +1,35 @@
+from collections.abc import Iterable
 import json
 import logging
 import os
 from pathlib import Path
 import re
 import sys
+from typing import TYPE_CHECKING
 
 from esphome.const import CONF_COMPILE_PROCESS_LIMIT, CONF_ESPHOME, KEY_CORE
 from esphome.core import CORE, EsphomeError
-from esphome.helpers import add_git_ceiling_directory
+from esphome.helpers import add_git_ceiling_directory, rmtree, write_file
 from esphome.util import FlashImage, run_external_process
 
+if TYPE_CHECKING:
+    from platformio.project.config import ProjectConfig
+
 _LOGGER = logging.getLogger(__name__)
 
+# PlatformIO cache subdirs resolved via ProjectConfig. A full ``clean-all`` wipes
+# these plus the whole ``core_dir``; a Python-version heal wipes these plus the
+# penv while keeping ``core_dir`` (so the sibling stamp/lock survive).
+_PIO_CACHE_DIRS = ("cache_dir", "packages_dir", "platforms_dir")
+
+# Marker recording the Python major.minor the PlatformIO cache was provisioned
+# under, plus the lock guarding the check/wipe. Both live in the dir resolved
+# by ``_pio_stamp_dir`` (NOT wiped by the heal), so they survive the wipe and
+# are rewritten after it.
+_PIO_PYTHON_STAMP_FILE = ".esphome.pio.stamp.json"
+_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``.
@@ -44,7 +62,174 @@ def _strip_win_long_path_prefix(path: str) -> str:
     return path
 
 
+def get_platformio_config() -> "ProjectConfig | None":
+    """Return PlatformIO's ``ProjectConfig``, or None when PlatformIO is absent."""
+    try:
+        from platformio.project.config import ProjectConfig
+    except ImportError:
+        return None
+    return ProjectConfig.get_instance()
+
+
+def _pio_stamp_dir(config: "ProjectConfig") -> Path:
+    """Return the persistent home for the python-version stamp and lock.
+
+    The parent of ``platforms_dir``, not ``core_dir``: the container/add-on
+    images relocate the platform/package caches to a persistent volume while
+    ``core_dir`` stays at the ephemeral default (its ``appstate.json`` must not
+    move), so a stamp under ``core_dir`` would be wiped on every image update
+    while the stale cache it guards survives. Everywhere else ``platforms_dir``
+    sits inside ``core_dir`` and this resolves to ``core_dir``.
+    """
+    return Path(config.get("platformio", "platforms_dir")).parent
+
+
+def _delete_platformio_dirs(config: "ProjectConfig", pio_dirs: Iterable[str]) -> None:
+    """Delete each named PlatformIO dir resolved from *config*."""
+    for pio_dir in pio_dirs:
+        path = Path(config.get("platformio", pio_dir))
+        if path.is_dir():
+            _LOGGER.info("Deleting PlatformIO %s %s", pio_dir, path)
+            rmtree(path)
+
+
+def clean_platformio_cache() -> None:
+    """Wipe the whole PlatformIO cache (cache/packages/platforms/core).
+
+    The full set ``clean-all`` (Reset Build Environment) clears. No-op when
+    PlatformIO is unavailable.
+    """
+    config = get_platformio_config()
+    if config is None:
+        return
+    _delete_platformio_dirs(config, [*_PIO_CACHE_DIRS, "core_dir"])
+
+
+def _clean_platformio_python_env(config: "ProjectConfig", core_dir: Path) -> None:
+    """Wipe the cache subdirs + penv for a Python-version change.
+
+    Keeps ``core_dir`` itself (and the stamp/lock siblings under it); otherwise
+    the same cache set ``clean-all`` clears.
+    """
+    _delete_platformio_dirs(config, _PIO_CACHE_DIRS)
+    penv = core_dir / "penv"
+    if penv.is_dir():
+        _LOGGER.info("Deleting PlatformIO penv %s", penv)
+        rmtree(penv)
+
+
+def _current_python_minor() -> str:
+    """Return the running interpreter's ``major.minor`` (e.g. ``3.13``)."""
+    return f"{sys.version_info.major}.{sys.version_info.minor}"
+
+
+def _read_pio_stamp_python(stamp_file: Path) -> str | None:
+    """Return the ``python_version`` recorded in *stamp_file*, or None."""
+    try:
+        with stamp_file.open(encoding="utf-8") as f:
+            data = json.load(f)
+    except FileNotFoundError:
+        return None
+    except (json.JSONDecodeError, OSError) as err:
+        # A present-but-unreadable stamp is a distinct signal from an absent
+        # one, and it drives a cache clean; surface why at normal verbosity.
+        _LOGGER.warning("Could not read %s: %s", stamp_file, err)
+        return None
+    if not isinstance(data, dict):
+        return None
+    version = data.get("python_version")
+    return version if isinstance(version, str) else None
+
+
+def _write_pio_stamp_python(stamp_file: Path, python_version: str) -> None:
+    """Atomically write the PlatformIO python-version stamp."""
+    write_file(
+        stamp_file,
+        json.dumps(
+            {
+                "schema_version": _PIO_PYTHON_STAMP_SCHEMA,
+                "python_version": python_version,
+            }
+        ),
+    )
+
+
+def heal_platformio_python_env() -> None:
+    """Wipe the PlatformIO cache unless it is stamped for the running Python.
+
+    A PlatformIO platform/tool package pins the Python versions it accepts when
+    it is provisioned, and ESPHome pins platforms to exact, immutable versions,
+    so a later interpreter bump (a container upgrading its base Python) leaves
+    the cached platform rejecting the new interpreter ("Python version must be
+    between ...") until the cache is wiped. A stamp records the ``major.minor``
+    the cache was provisioned for; when it doesn't match the running
+    interpreter (or has never been written for an existing cache), the same
+    PlatformIO dirs ``clean-all`` wipes are cleaned so PlatformIO
+    re-provisions, matching Reset Build Environment automatically. The native
+    ESP-IDF toolchain already self-heals through its own stamp; this covers the
+    PlatformIO path. No-op when PlatformIO is unavailable.
+    """
+    config = get_platformio_config()
+    if config is None:
+        return
+    try:
+        _check_platformio_python_stamp(config)
+    except (EsphomeError, OSError) as err:
+        # The check is a best-effort repair; a full or read-only cache volume
+        # must not abort a build that might otherwise work. The stamp write
+        # surfaces as EsphomeError (write_file wraps OSError).
+        _LOGGER.warning("PlatformIO build environment check failed: %s", err)
+
+
+def _check_platformio_python_stamp(config: "ProjectConfig") -> None:
+    """Compare the stamp to the running interpreter; wipe and restamp on mismatch."""
+    current = _current_python_minor()
+    stamp_dir = _pio_stamp_dir(config)
+    # Host the stamp/lock even before PlatformIO's first run creates the dir.
+    stamp_dir.mkdir(parents=True, exist_ok=True)
+    stamp_file = stamp_dir / _PIO_PYTHON_STAMP_FILE
+
+    from filelock import FileLock
+
+    with FileLock(str(stamp_dir / _PIO_PYTHON_STAMP_LOCK)):
+        provisioned = _read_pio_stamp_python(stamp_file)
+        if provisioned == current:
+            return
+        core_dir = Path(config.get("platformio", "core_dir"))
+        has_cache = (
+            any(
+                Path(config.get("platformio", pio_dir)).is_dir()
+                for pio_dir in _PIO_CACHE_DIRS
+            )
+            or (core_dir / "penv").is_dir()
+        )
+        if has_cache:
+            if provisioned is None:
+                # An existing cache with no stamp predates the stamp: its
+                # provisioning interpreter is unknown, so clean once rather
+                # than leave a possibly-stale cache failing every build.
+                _LOGGER.info(
+                    "Cleaning the PlatformIO build environment once so it "
+                    "re-provisions for Python %s",
+                    current,
+                )
+            else:
+                _LOGGER.info(
+                    "Python version changed (%s -> %s); cleaning PlatformIO "
+                    "build environment so it re-provisions for the new "
+                    "interpreter",
+                    provisioned,
+                    current,
+                )
+            _clean_platformio_python_env(config, core_dir)
+        _write_pio_stamp_python(stamp_file, current)
+
+
 def run_platformio_cli(*args, **kwargs) -> str | int:
+    # Re-provision the PlatformIO cache if the interpreter's major.minor changed
+    # since it was last built; a stale platform otherwise rejects the new Python
+    # with "Python version must be between ..." until Reset Build Environment.
+    heal_platformio_python_env()
     os.environ["PLATFORMIO_FORCE_COLOR"] = "true"
     os.environ["PLATFORMIO_BUILD_DIR"] = str(CORE.relative_pioenvs_path().absolute())
     os.environ.setdefault(

esphome/writer.py

@@ -670,18 +670,9 @@ def clean_all(configuration: list[str]):
             rmtree(install_path)
 
     # Clean PlatformIO project files
-    try:
-        from platformio.project.config import ProjectConfig
-    except ImportError:
-        # PlatformIO is not available, skip cleaning
-        pass
-    else:
-        config = ProjectConfig.get_instance()
-        for pio_dir in ["cache_dir", "packages_dir", "platforms_dir", "core_dir"]:
-            path = Path(config.get("platformio", pio_dir))
-            if path.is_dir():
-                _LOGGER.info("Deleting PlatformIO %s %s", pio_dir, path)
-                rmtree(path)
+    from esphome.platformio.toolchain import clean_platformio_cache
+
+    clean_platformio_cache()
 
 
 GITIGNORE_CONTENT = """# Gitignore settings for ESPHome

requirements.txt

@@ -27,6 +27,7 @@ smpclient==7.2.0
 requests==2.34.2
 py7zr==1.1.3
 platformdirs==4.10.0  # native esp-idf toolchain global cache dir
+filelock==3.29.0  # lock guarding the PlatformIO python-version cache heal
 
 # esp-idf >= 5.0 requires this
 pyparsing >= 3.3.2

tests/unit_tests/test_platformio_toolchain.py

@@ -2,12 +2,14 @@
 
 # pylint: disable=protected-access
 
+from collections.abc import Generator
 from contextlib import contextmanager
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 import json
 import os
 from pathlib import Path
 import shutil
+import sys
 import threading
 from types import SimpleNamespace
 from unittest.mock import MagicMock, Mock, call, patch
@@ -1093,3 +1095,361 @@ def test_filter_platformio_lines_blocks_noisy_messages(msg: str) -> None:
 def test_filter_platformio_lines_allows_other_messages(msg: str) -> None:
     """Test that non-noisy platformio output lines pass through RedirectText."""
     assert _filter_through_redirect(msg) == msg + "\n"
+
+
+# ---------------------------------------------------------------------------
+# PlatformIO python-version cache heal
+# ---------------------------------------------------------------------------
+
+_CURRENT_MINOR = f"{sys.version_info.major}.{sys.version_info.minor}"
+# Captured before the autouse guard patches the name, so tests can exercise the
+# real implementation.
+_REAL_GET_PLATFORMIO_CONFIG = toolchain.get_platformio_config
+
+
+@pytest.fixture(autouse=True)
+def _guard_real_platformio() -> Generator[None, None, None]:
+    """Default the PlatformIO config lookup to None so no test in this module
+    touches a real ~/.platformio; the heal tests re-patch it at a temp dir."""
+    with patch.object(toolchain, "get_platformio_config", return_value=None):
+        yield
+
+
+def _pio_layout(core_dir: Path) -> dict[str, Path]:
+    """Return the PlatformIO dir layout with cache/packages/platforms under core."""
+    return {
+        "core_dir": core_dir,
+        "packages_dir": core_dir / "packages",
+        "platforms_dir": core_dir / "platforms",
+        "cache_dir": core_dir / ".cache",
+    }
+
+
+def _split_pio_layout(tmp_path: Path) -> dict[str, Path]:
+    """Container-shape layout: caches on a persistent root, core_dir ephemeral."""
+    persistent = tmp_path / "data" / "platformio"
+    return {
+        "core_dir": tmp_path / "root" / ".platformio",
+        "platforms_dir": persistent / "platforms",
+        "packages_dir": persistent / "packages",
+        "cache_dir": persistent / "cache",
+    }
+
+
+def _seed_layout(layout: dict[str, Path]) -> None:
+    """Populate each cache dir (and the core penv) with a marker file."""
+    for key in ("platforms_dir", "packages_dir", "cache_dir"):
+        layout[key].mkdir(parents=True, exist_ok=True)
+        (layout[key] / "marker").write_text("x", encoding="utf-8")
+    penv = layout["core_dir"] / "penv"
+    penv.mkdir(parents=True, exist_ok=True)
+    (penv / "marker").write_text("x", encoding="utf-8")
+
+
+def _make_pio_config(layout: dict[str, Path] | Path) -> MagicMock:
+    """A ProjectConfig stand-in resolving platformio dir options from *layout*."""
+    resolved = _pio_layout(layout) if isinstance(layout, Path) else layout
+    config = MagicMock()
+    config.get.side_effect = lambda section, option: (
+        str(resolved[option]) if section == "platformio" else ""
+    )
+    return config
+
+
+@contextmanager
+def _use_pio_config(layout: dict[str, Path] | Path) -> Generator[MagicMock, None, None]:
+    """Point ``get_platformio_config`` at a temp layout for the block."""
+    config = _make_pio_config(layout)
+    with patch.object(toolchain, "get_platformio_config", return_value=config):
+        yield config
+
+
+def _stamp_version(core_dir: Path) -> str | None:
+    """Read the python version recorded in the heal stamp under *core_dir*."""
+    return toolchain._read_pio_stamp_python(core_dir / toolchain._PIO_PYTHON_STAMP_FILE)
+
+
+def _cache_wiped(core_dir: Path) -> bool:
+    """True when the seeded cache subdir markers are gone."""
+    return not any(
+        (core_dir / sub / "marker").exists()
+        for sub in ("packages", "platforms", ".cache")
+    )
+
+
+@pytest.fixture
+def pio_core_dir(tmp_path: Path) -> Path:
+    """A populated PlatformIO core dir (packages/platforms/.cache/penv seeded)."""
+    core = tmp_path / "dot-platformio"
+    for sub in ("packages", "platforms", ".cache", "penv"):
+        seeded = core / sub
+        seeded.mkdir(parents=True)
+        (seeded / "marker").write_text("x", encoding="utf-8")
+    return core
+
+
+def test_current_python_minor_matches_running_interpreter() -> None:
+    """_current_python_minor returns major.minor of the running interpreter."""
+    assert toolchain._current_python_minor() == _CURRENT_MINOR
+
+
+def test_pio_stamp_round_trip(tmp_path: Path) -> None:
+    """The stamp writer/reader round-trips and records the schema version."""
+    stamp = tmp_path / toolchain._PIO_PYTHON_STAMP_FILE
+    toolchain._write_pio_stamp_python(stamp, "3.13")
+    assert toolchain._read_pio_stamp_python(stamp) == "3.13"
+    assert json.loads(stamp.read_text()) == {
+        "schema_version": toolchain._PIO_PYTHON_STAMP_SCHEMA,
+        "python_version": "3.13",
+    }
+
+
+def test_read_pio_stamp_missing(tmp_path: Path) -> None:
+    """A missing stamp file yields None."""
+    assert toolchain._read_pio_stamp_python(tmp_path / "nope.json") is None
+
+
+def test_read_pio_stamp_malformed(tmp_path: Path) -> None:
+    """A corrupt stamp file yields None instead of raising."""
+    stamp = tmp_path / "bad.json"
+    stamp.write_text("{not json", encoding="utf-8")
+    assert toolchain._read_pio_stamp_python(stamp) is None
+
+
+def test_read_pio_stamp_unreadable_logs_warning(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A present-but-unreadable stamp yields None and warns."""
+    stamp = tmp_path / "stamp.json"
+    stamp.mkdir()
+    with caplog.at_level("WARNING"):
+        assert toolchain._read_pio_stamp_python(stamp) is None
+    assert "Could not read" in caplog.text
+
+
+def test_read_pio_stamp_without_python_version(tmp_path: Path) -> None:
+    """A stamp missing python_version yields None."""
+    stamp = tmp_path / "s.json"
+    stamp.write_text(json.dumps({"schema_version": "0"}), encoding="utf-8")
+    assert toolchain._read_pio_stamp_python(stamp) is None
+
+
+@pytest.mark.parametrize("payload", ["42", '"x"', "[1, 2]", "null"])
+def test_read_pio_stamp_non_object_json(tmp_path: Path, payload: str) -> None:
+    """Valid-but-non-object JSON in the stamp yields None, not a crash."""
+    stamp = tmp_path / "s.json"
+    stamp.write_text(payload, encoding="utf-8")
+    assert toolchain._read_pio_stamp_python(stamp) is None
+
+
+def test_clean_platformio_cache_none_config_is_noop() -> None:
+    """clean_platformio_cache is a no-op when PlatformIO is unavailable."""
+    with patch.object(toolchain, "get_platformio_config", return_value=None):
+        toolchain.clean_platformio_cache()
+
+
+def test_clean_platformio_cache_wipes_everything(pio_core_dir: Path) -> None:
+    """clean_platformio_cache removes cache/packages/platforms and core_dir."""
+    with _use_pio_config(pio_core_dir):
+        toolchain.clean_platformio_cache()
+    assert not pio_core_dir.exists()
+
+
+def test_heal_none_config_is_noop() -> None:
+    """Heal is a no-op (no error) when PlatformIO is unavailable."""
+    with patch.object(toolchain, "get_platformio_config", return_value=None):
+        toolchain.heal_platformio_python_env()
+
+
+def test_heal_fresh_cache_stamps_without_wipe(tmp_path: Path) -> None:
+    """A fresh core dir (no stamp, no penv) is stamped, not wiped."""
+    core = tmp_path / "pio"
+    with _use_pio_config(core):
+        toolchain.heal_platformio_python_env()
+    assert _stamp_version(core) == _CURRENT_MINOR
+
+
+def test_heal_stamp_matches_current_no_wipe(pio_core_dir: Path) -> None:
+    """A stamp matching the running interpreter leaves the cache untouched."""
+    toolchain._write_pio_stamp_python(
+        pio_core_dir / toolchain._PIO_PYTHON_STAMP_FILE, _CURRENT_MINOR
+    )
+    with _use_pio_config(pio_core_dir):
+        toolchain.heal_platformio_python_env()
+    assert not _cache_wiped(pio_core_dir)
+    assert (pio_core_dir / "penv" / "marker").exists()
+
+
+def test_heal_stale_stamp_wipes_and_restamps(pio_core_dir: Path) -> None:
+    """A stamp from an older interpreter triggers a wipe + restamp; core_dir stays."""
+    toolchain._write_pio_stamp_python(
+        pio_core_dir / toolchain._PIO_PYTHON_STAMP_FILE, "2.7"
+    )
+    with _use_pio_config(pio_core_dir):
+        toolchain.heal_platformio_python_env()
+    assert _cache_wiped(pio_core_dir)
+    assert not (pio_core_dir / "penv").exists()
+    assert pio_core_dir.is_dir()
+    assert _stamp_version(pio_core_dir) == _CURRENT_MINOR
+
+
+def test_heal_no_stamp_existing_cache_wipes_once(
+    pio_core_dir: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """An existing cache with no stamp is cleaned once and stamped."""
+    with _use_pio_config(pio_core_dir), caplog.at_level("INFO"):
+        toolchain.heal_platformio_python_env()
+    assert _cache_wiped(pio_core_dir)
+    assert not (pio_core_dir / "penv").exists()
+    assert _stamp_version(pio_core_dir) == _CURRENT_MINOR
+    assert "once" in caplog.text
+
+
+def test_heal_no_stamp_penv_only_counts_as_cache(tmp_path: Path) -> None:
+    """A core dir holding only a penv still triggers the one-time clean."""
+    core = tmp_path / "pio"
+    penv = core / "penv"
+    penv.mkdir(parents=True)
+    (penv / "marker").write_text("x", encoding="utf-8")
+    with _use_pio_config(core):
+        toolchain.heal_platformio_python_env()
+    assert not penv.exists()
+    assert _stamp_version(core) == _CURRENT_MINOR
+
+
+def test_heal_oserror_is_nonfatal(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A filesystem failure during the check warns instead of aborting the build."""
+    blocker = tmp_path / "pio"
+    blocker.write_text("not a directory", encoding="utf-8")
+    with _use_pio_config(blocker), caplog.at_level("WARNING"):
+        toolchain.heal_platformio_python_env()
+    assert "build environment check failed" in caplog.text
+
+
+def test_heal_stamp_write_failure_is_nonfatal(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A failed stamp write (EsphomeError from write_file) warns, not aborts."""
+    with (
+        _use_pio_config(tmp_path / "pio"),
+        patch.object(
+            toolchain,
+            "_write_pio_stamp_python",
+            side_effect=EsphomeError("disk full"),
+        ),
+        caplog.at_level("WARNING"),
+    ):
+        toolchain.heal_platformio_python_env()
+    assert "build environment check failed" in caplog.text
+
+
+def test_heal_is_idempotent_across_runs(pio_core_dir: Path) -> None:
+    """After a heal writes the stamp, a re-provisioned cache is not wiped again."""
+    with _use_pio_config(pio_core_dir):
+        toolchain.heal_platformio_python_env()
+        repop = pio_core_dir / "packages"
+        repop.mkdir(exist_ok=True)
+        (repop / "marker").write_text("x", encoding="utf-8")
+        toolchain.heal_platformio_python_env()
+    assert (pio_core_dir / "packages" / "marker").exists()
+
+
+def test_pio_stamp_dir_is_platforms_parent(tmp_path: Path) -> None:
+    """The stamp home is the parent of platforms_dir, not core_dir."""
+    layout = _split_pio_layout(tmp_path)
+    config = _make_pio_config(layout)
+    assert toolchain._pio_stamp_dir(config) == layout["platforms_dir"].parent
+    nested = _make_pio_config(tmp_path / "pio")
+    assert toolchain._pio_stamp_dir(nested) == tmp_path / "pio"
+
+
+def test_heal_container_layout_stamps_persistent_root(tmp_path: Path) -> None:
+    """Container shape: the stamp lands on the persistent cache root."""
+    layout = _split_pio_layout(tmp_path)
+    with _use_pio_config(layout):
+        toolchain.heal_platformio_python_env()
+    persistent = layout["platforms_dir"].parent
+    assert _stamp_version(persistent) == _CURRENT_MINOR
+    assert not (layout["core_dir"] / toolchain._PIO_PYTHON_STAMP_FILE).exists()
+
+
+def test_heal_container_layout_stale_stamp_wipes_persistent_cache(
+    tmp_path: Path,
+) -> None:
+    """Container shape: a stale stamp wipes the relocated persistent caches."""
+    layout = _split_pio_layout(tmp_path)
+    _seed_layout(layout)
+    persistent = layout["platforms_dir"].parent
+    toolchain._write_pio_stamp_python(
+        persistent / toolchain._PIO_PYTHON_STAMP_FILE, "2.7"
+    )
+    with _use_pio_config(layout):
+        toolchain.heal_platformio_python_env()
+    for key in ("platforms_dir", "packages_dir", "cache_dir"):
+        assert not layout[key].exists()
+    assert not (layout["core_dir"] / "penv").exists()
+    assert _stamp_version(persistent) == _CURRENT_MINOR
+
+
+def test_heal_container_layout_survives_core_dir_wipe(tmp_path: Path) -> None:
+    """A python change is still detected after an image update wiped core_dir."""
+    layout = _split_pio_layout(tmp_path)
+    _seed_layout(layout)
+    shutil.rmtree(layout["core_dir"])
+    persistent = layout["platforms_dir"].parent
+    toolchain._write_pio_stamp_python(
+        persistent / toolchain._PIO_PYTHON_STAMP_FILE, "2.7"
+    )
+    with _use_pio_config(layout):
+        toolchain.heal_platformio_python_env()
+    for key in ("platforms_dir", "packages_dir", "cache_dir"):
+        assert not layout[key].exists()
+    assert _stamp_version(persistent) == _CURRENT_MINOR
+
+
+def test_get_platformio_config_returns_project_config() -> None:
+    """The real lookup returns a usable ProjectConfig when PlatformIO is present."""
+    config = _REAL_GET_PLATFORMIO_CONFIG()
+    assert config is not None
+    assert hasattr(config, "get")
+
+
+def test_get_platformio_config_none_when_platformio_absent() -> None:
+    """The lookup returns None when PlatformIO cannot be imported."""
+    with patch.dict(sys.modules, {"platformio.project.config": None}):
+        assert _REAL_GET_PLATFORMIO_CONFIG() is None
+
+
+def test_delete_platformio_dirs_skips_missing(tmp_path: Path) -> None:
+    """A named dir that does not exist is skipped without error."""
+    (tmp_path / "packages").mkdir()
+    (tmp_path / "packages" / "marker").write_text("x", encoding="utf-8")
+    config = _make_pio_config(tmp_path)
+    # platforms_dir does not exist; packages_dir does.
+    toolchain._delete_platformio_dirs(config, ["packages_dir", "platforms_dir"])
+    assert not (tmp_path / "packages").exists()
+
+
+def test_heal_stale_stamp_wipes_when_penv_absent(pio_core_dir: Path) -> None:
+    """The penv wipe is skipped cleanly when no penv exists."""
+    shutil.rmtree(pio_core_dir / "penv")
+    toolchain._write_pio_stamp_python(
+        pio_core_dir / toolchain._PIO_PYTHON_STAMP_FILE, "2.7"
+    )
+    with _use_pio_config(pio_core_dir):
+        toolchain.heal_platformio_python_env()
+    assert _cache_wiped(pio_core_dir)
+    assert _stamp_version(pio_core_dir) == _CURRENT_MINOR
+
+
+def test_run_platformio_cli_invokes_heal(
+    setup_core: Path, mock_run_external_process: Mock
+) -> None:
+    """run_platformio_cli runs the heal before spawning PlatformIO."""
+    CORE.build_path = str(setup_core / "build" / "test")
+    mock_run_external_process.return_value = 0
+    with patch.object(toolchain, "heal_platformio_python_env") as mock_heal:
+        toolchain.run_platformio_cli("test")
+    mock_heal.assert_called_once()