Merge remote-tracking branch 'origin/dev' into jesserockz-2026-503

tests/unit_tests/analyze_memory/test_build_artifacts.py

@@ -22,7 +22,7 @@ def _make_build_dir(tmp_path: Path, name: str = "mydevice") -> Path:
 
 def _touch(path: Path) -> Path:
     path.parent.mkdir(parents=True, exist_ok=True)
-    path.write_text("")
+    path.write_text("", encoding="utf-8")
     return path
 
 

tests/unit_tests/build_gen/test_espidf.py

@@ -184,6 +184,18 @@ def test_get_component_cmakelists_compile_flags_excluded_from_link_opts() -> Non
     assert "-Wl,--gc-sections" in content
 
 
+def test_get_component_cmakelists_globs_alternate_cpp_extensions() -> None:
+    """Both app_sources glob variants include .cc/.cxx/.c++ so vendored sources
+    are compiled, matching the extensions PlatformIO's builder globs by default."""
+    CORE.build_flags = set()
+    from esphome.build_gen.espidf import get_component_cmakelists
+
+    content = get_component_cmakelists()
+    for ext in ("cc", "cxx", "c++"):
+        assert content.count(f'"${{CMAKE_CURRENT_SOURCE_DIR}}/*.{ext}"') == 2
+        assert content.count(f'"${{CMAKE_CURRENT_SOURCE_DIR}}/esphome/*.{ext}"') == 2
+
+
 def test_get_project_cmakelists_emits_managed_components_property(
     tmp_path: Path,
 ) -> None:

tests/unit_tests/components/test_esp8266.py

@@ -1,9 +1,15 @@
 """Tests for ESP8266 component."""
 
+from __future__ import annotations
+
+from collections.abc import Generator
+from unittest.mock import MagicMock, patch
+
 import pytest
 
-from esphome.components.esp8266 import lambdas_use_scanf_float
-from esphome.core import Lambda
+from esphome.components import esp8266
+from esphome.components.esp8266 import check_rosetta, lambdas_use_scanf_float
+from esphome.core import EsphomeError, Lambda
 from esphome.types import ConfigType
 
 
@@ -60,3 +66,54 @@ def test_lambdas_use_scanf_float_nested() -> None:
     """Test detection in deeply nested config."""
     config: ConfigType = {"a": {"b": {"c": [Lambda('sscanf(buf, "%f", &v)')]}}}
     assert lambdas_use_scanf_float(config) is True
+
+
+@pytest.fixture
+def apple_silicon_run(monkeypatch: pytest.MonkeyPatch) -> Generator[MagicMock]:
+    """Simulate an Apple Silicon Mac and yield the mocked subprocess.run."""
+    monkeypatch.setattr(esp8266, "IS_MACOS", True)
+    with (
+        patch("esphome.components.esp8266.platform.machine", return_value="arm64"),
+        patch("esphome.components.esp8266.subprocess.run") as mock_run,
+    ):
+        yield mock_run
+
+
+@pytest.mark.parametrize(
+    ("is_macos", "machine"),
+    [
+        (False, "arm64"),
+        (True, "x86_64"),
+    ],
+)
+def test_check_rosetta_skips_other_systems(
+    monkeypatch: pytest.MonkeyPatch, is_macos: bool, machine: str
+) -> None:
+    """The check only probes on Apple Silicon Macs."""
+    monkeypatch.setattr(esp8266, "IS_MACOS", is_macos)
+    with (
+        patch("esphome.components.esp8266.platform.machine", return_value=machine),
+        patch("esphome.components.esp8266.subprocess.run") as mock_run,
+    ):
+        check_rosetta()
+    mock_run.assert_not_called()
+
+
+def test_check_rosetta_installed(apple_silicon_run: MagicMock) -> None:
+    """No error when the x86_64 probe succeeds (Rosetta present)."""
+    apple_silicon_run.return_value = MagicMock(returncode=0)
+    check_rosetta()
+    apple_silicon_run.assert_called_once()
+
+
+def test_check_rosetta_missing(apple_silicon_run: MagicMock) -> None:
+    """A failing x86_64 probe raises an actionable error."""
+    apple_silicon_run.return_value = MagicMock(returncode=1)
+    with pytest.raises(EsphomeError, match="softwareupdate --install-rosetta"):
+        check_rosetta()
+
+
+def test_check_rosetta_arch_unavailable(apple_silicon_run: MagicMock) -> None:
+    """The build proceeds when arch(1) cannot be executed."""
+    apple_silicon_run.side_effect = OSError("no such file")
+    check_rosetta()

tests/unit_tests/components/test_esp_stacktrace.py

@@ -137,3 +137,45 @@ def test_process_stacktrace_esp32_crash_handler(
     state = process_stacktrace(config, line_bt1, False)
     mock_esp32_decode_pc.assert_called_once_with(config, "42005ABC")
     assert state is False
+
+    mock_esp32_decode_pc.reset_mock()
+
+    # Reason line carries no address, must not trigger a decode
+    line_reason = "[E][esp32.crash:079]:   Reason: Fault - LoadProhibited (cause 28)"
+    state = process_stacktrace(config, line_reason, False)
+    mock_esp32_decode_pc.assert_not_called()
+    assert state is False
+
+    mock_esp32_decode_pc.reset_mock()
+
+    # EXCVADDR pointing at code (e.g. jumping through a corrupted pointer) decodes
+    line_excvaddr = "[E][esp32.crash:081]:   EXCVADDR: 0x400D9ABC  (faulting address)"
+    state = process_stacktrace(config, line_excvaddr, False)
+    mock_esp32_decode_pc.assert_called_once_with(config, "400D9ABC")
+    assert state is False
+
+    mock_esp32_decode_pc.reset_mock()
+
+    # EXCVADDR pointing at data (heap/null) is not a code address, must be ignored
+    line_excvaddr_data = (
+        "[E][esp32.crash:081]:   EXCVADDR: 0x0000001C  (faulting address)"
+    )
+    state = process_stacktrace(config, line_excvaddr_data, False)
+    mock_esp32_decode_pc.assert_not_called()
+    assert state is False
+
+    mock_esp32_decode_pc.reset_mock()
+
+    # RISC-V MTVAL pointing at code decodes
+    line_mtval = "[E][esp32.crash:081]:   MTVAL: 0x42001234  (faulting address)"
+    state = process_stacktrace(config, line_mtval, False)
+    mock_esp32_decode_pc.assert_called_once_with(config, "42001234")
+    assert state is False
+
+    mock_esp32_decode_pc.reset_mock()
+
+    # RISC-V MTVAL pointing at data must be ignored
+    line_mtval_data = "[E][esp32.crash:081]:   MTVAL: 0x3FC80123  (faulting address)"
+    state = process_stacktrace(config, line_mtval_data, False)
+    mock_esp32_decode_pc.assert_not_called()
+    assert state is False

tests/unit_tests/fixtures/idf_tools_stub/idf_tools.py

@@ -0,0 +1,120 @@
+"""Minimal idf_tools stand-in for get_tool_downloads.py tests."""
+
+from collections.abc import Iterable
+import os
+
+CURRENT_PLATFORM = "linux-amd64"
+TOOLS_FILE = "tools/tools.json"
+
+
+class ToolBinaryError(RuntimeError):
+    pass
+
+
+class _G:
+    idf_path: str | None = None
+    idf_tools_path: str | None = None
+    tools_json: str | None = None
+
+
+g = _G()
+
+
+class IDFEnv:
+    @classmethod
+    def get_idf_env(cls) -> "IDFEnv":
+        return cls()
+
+
+def add_and_check_targets(idf_env_obj: IDFEnv, targets_str: str) -> list[str]:
+    return targets_str.split(",")
+
+
+class _Download:
+    def __init__(self, url: str, size: int, sha256: str, rename_dist: str = "") -> None:
+        self.url = url
+        self.size = size
+        self.sha256 = sha256
+        self.rename_dist = rename_dist
+
+
+class _Version:
+    def __init__(self, download: _Download | None) -> None:
+        self._download = download
+
+    def get_download_for_platform(self, platform_name: str) -> _Download | None:
+        return self._download
+
+
+class _Tool:
+    def __init__(
+        self,
+        versions: dict[str, _Version],
+        recommended: str | None,
+        installed: Iterable[str] = (),
+        broken: bool = False,
+    ) -> None:
+        self.versions = versions
+        self._recommended = recommended
+        self.versions_installed = list(installed)
+        self._broken = broken
+
+    def compatible_with_platform(self) -> bool:
+        return True
+
+    def get_recommended_version(self) -> str | None:
+        return self._recommended
+
+    def find_installed_versions(self) -> None:
+        if self._broken:
+            raise ToolBinaryError("broken binary")
+
+
+_TOOLS = {
+    "cmake": _Tool(
+        {"3.30.2": _Version(_Download("https://gh.test/cmake.tar.gz", 11, "aa"))},
+        "3.30.2",
+    ),
+    "ninja": _Tool(
+        {
+            "1.12.1": _Version(
+                _Download("https://gh.test/ninja-mac.zip", 22, "bb", "ninja-v1.zip")
+            )
+        },
+        "1.12.1",
+    ),
+    "installed-tool": _Tool(
+        {"1.0": _Version(_Download("https://gh.test/x.tar.gz", 33, "cc"))},
+        "1.0",
+        installed=["1.0"],
+    ),
+    "broken-tool": _Tool(
+        {"2.0": _Version(_Download("https://gh.test/y.tar.gz", 44, "dd"))},
+        "2.0",
+        broken=True,
+    ),
+    "no-recommended-tool": _Tool({"3.0": _Version(None)}, None),
+    "no-download-tool": _Tool({"4.0": _Version(None)}, "4.0"),
+}
+
+
+def load_tools_info() -> dict[str, _Tool]:
+    return _TOOLS
+
+
+def expand_tools_arg(
+    tools_spec: list[str], overall_tools: dict[str, _Tool], targets: list[str]
+) -> list[str]:
+    if "required" in tools_spec:
+        return list(overall_tools)
+    return [t for t in tools_spec if "@" not in t] + [t for t in tools_spec if "@" in t]
+
+
+def get_idf_download_url_apply_mirrors(
+    args: object = None, download_url: str = ""
+) -> str:
+    print(f"Changed download URL: {download_url}")  # noise on stdout, like idf_tools
+    prefix = os.environ.get("TEST_MIRROR_PREFIX")
+    if prefix:
+        return prefix + download_url
+    return download_url

tests/unit_tests/test_bundle.py

@@ -27,6 +27,7 @@ from esphome.bundle import (
     is_bundle_path,
     prepare_bundle_for_compile,
     read_bundle_manifest,
+    remap_bundle_path,
 )
 from esphome.core import CORE, EsphomeError
 from esphome.yaml_util import force_load_include_files
@@ -478,7 +479,10 @@ def test_read_bundle_manifest_corrupted_tar(tmp_path: Path) -> None:
 def test_read_bundle_manifest(tmp_path: Path) -> None:
     bundle_path = _make_bundle(
         tmp_path,
-        manifest_overrides={ManifestKey.HAS_SECRETS: True},
+        manifest_overrides={
+            ManifestKey.HAS_SECRETS: True,
+            ManifestKey.CONFIG_DIR: "/original/config",
+        },
         extra_files={"secrets.yaml": b"wifi: test\n"},
     )
 
@@ -489,6 +493,7 @@ def test_read_bundle_manifest(tmp_path: Path) -> None:
     assert manifest.esphome_version == "2026.2.0-test"
     assert manifest.config_filename == "test.yaml"
     assert manifest.has_secrets is True
+    assert manifest.config_dir == "/original/config"
 
 
 def test_read_bundle_manifest_minimal(tmp_path: Path) -> None:
@@ -508,6 +513,266 @@ def test_read_bundle_manifest_minimal(tmp_path: Path) -> None:
     assert result.esphome_version == "unknown"
     assert not result.files
     assert result.has_secrets is False
+    assert result.config_dir is None
+
+
+def test_read_bundle_manifest_non_string_config_dir(tmp_path: Path) -> None:
+    """A malformed config_dir value is dropped rather than propagated."""
+    bundle_path = _make_bundle(
+        tmp_path, manifest_overrides={ManifestKey.CONFIG_DIR: 42}
+    )
+
+    assert read_bundle_manifest(bundle_path).config_dir is None
+
+
+# ---------------------------------------------------------------------------
+# remap_bundle_path
+# ---------------------------------------------------------------------------
+
+
+ORIGINAL_CONFIG_DIR = "/original/config"
+
+
+def _bundle_manifest_dict(**overrides: Any) -> dict[str, Any]:
+    """Manifest content an extracted bundle would contain."""
+    manifest: dict[str, Any] = {
+        ManifestKey.MANIFEST_VERSION: CURRENT_MANIFEST_VERSION,
+        ManifestKey.CONFIG_FILENAME: "test.yaml",
+        ManifestKey.CONFIG_DIR: ORIGINAL_CONFIG_DIR,
+    }
+    manifest.update(overrides)
+    return manifest
+
+
+def _setup_extracted_dir(
+    tmp_path: Path,
+    manifest: dict[str, Any] | str | None,
+    files: dict[str, str] | None = None,
+) -> Path:
+    """Create a directory shaped like an extracted bundle and point CORE at it."""
+    extract_dir = _setup_config_dir(tmp_path, files)
+    if manifest is not None:
+        content = manifest if isinstance(manifest, str) else json.dumps(manifest)
+        (extract_dir / MANIFEST_FILENAME).write_text(content)
+    return extract_dir
+
+
+def test_remap_bundle_path_success(tmp_path: Path) -> None:
+    """A stale absolute path resolves to the bundled copy next to the config."""
+    extract_dir = _setup_extracted_dir(
+        tmp_path, _bundle_manifest_dict(), files={"boards/partitions.csv": "csv\n"}
+    )
+
+    remapped = remap_bundle_path(f"{ORIGINAL_CONFIG_DIR}/boards/partitions.csv")
+
+    assert remapped == extract_dir / "boards" / "partitions.csv"
+    assert remapped.is_file()
+
+
+@pytest.mark.parametrize(
+    "value",
+    [
+        pytest.param(r"C:\Users\nick\esphome\boards\partitions.csv", id="backslashes"),
+        pytest.param("C:/Users/nick/esphome/boards/partitions.csv", id="forward"),
+        pytest.param(r"c:\users\NICK\esphome\boards\partitions.csv", id="case"),
+    ],
+)
+def test_remap_bundle_path_windows_bundle_on_posix(tmp_path: Path, value: str) -> None:
+    """A bundle created on Windows remaps on a build server with another layout."""
+    extract_dir = _setup_extracted_dir(
+        tmp_path,
+        _bundle_manifest_dict(**{ManifestKey.CONFIG_DIR: r"C:\Users\nick\esphome"}),
+        files={"boards/partitions.csv": "csv\n"},
+    )
+
+    remapped = remap_bundle_path(value)
+
+    assert remapped == extract_dir / "boards" / "partitions.csv"
+    assert remapped.is_file()
+
+
+def test_remap_bundle_path_windows_bundle_path_not_under_config_dir(
+    tmp_path: Path,
+) -> None:
+    """A Windows path outside the original config dir is left alone."""
+    _setup_extracted_dir(
+        tmp_path,
+        _bundle_manifest_dict(**{ManifestKey.CONFIG_DIR: r"C:\Users\nick\esphome"}),
+        files={"partitions.csv": "csv\n"},
+    )
+
+    assert remap_bundle_path(r"D:\other\partitions.csv") is None
+
+
+def test_remap_bundle_path_windows_profile_with_spaces(tmp_path: Path) -> None:
+    r"""A Windows profile like C:\Users\First Last remaps like any other dir."""
+    extract_dir = _setup_extracted_dir(
+        tmp_path,
+        _bundle_manifest_dict(
+            **{ManifestKey.CONFIG_DIR: r"C:\Users\First Last\esphome"}
+        ),
+        files={"boards/my partitions.csv": "csv\n"},
+    )
+
+    remapped = remap_bundle_path(
+        r"C:\Users\First Last\esphome\boards\my partitions.csv"
+    )
+
+    assert remapped == extract_dir / "boards" / "my partitions.csv"
+    assert remapped.is_file()
+
+
+def test_remap_bundle_path_unc_config_dir(tmp_path: Path) -> None:
+    """A bundle created from a UNC share remaps like any other Windows path."""
+    extract_dir = _setup_extracted_dir(
+        tmp_path,
+        _bundle_manifest_dict(**{ManifestKey.CONFIG_DIR: r"\\server\share\esphome"}),
+        files={"partitions.csv": "csv\n"},
+    )
+
+    remapped = remap_bundle_path(r"\\server\share\esphome\partitions.csv")
+
+    assert remapped == extract_dir / "partitions.csv"
+
+
+def test_remap_bundle_path_flavor_mismatch(tmp_path: Path) -> None:
+    """A POSIX style value cannot come from a Windows config dir; no remap."""
+    _setup_extracted_dir(
+        tmp_path,
+        _bundle_manifest_dict(**{ManifestKey.CONFIG_DIR: r"C:\Users\nick\esphome"}),
+        files={"partitions.csv": "csv\n"},
+    )
+
+    assert remap_bundle_path("/original/config/partitions.csv") is None
+
+
+def test_remap_bundle_path_rejects_traversal(tmp_path: Path) -> None:
+    """A remap may never escape the extracted config tree."""
+    extract_dir = _setup_extracted_dir(tmp_path, _bundle_manifest_dict())
+    (tmp_path / "outside.csv").write_text("csv\n")
+    assert (extract_dir / ".." / "outside.csv").resolve().is_file()
+
+    assert remap_bundle_path(f"{ORIGINAL_CONFIG_DIR}/../outside.csv") is None
+
+
+def test_remap_bundle_path_relative_value(tmp_path: Path) -> None:
+    """Relative references resolve normally and are never remapped."""
+    _setup_extracted_dir(tmp_path, _bundle_manifest_dict())
+
+    assert remap_bundle_path("missing.csv") is None
+
+
+def test_remap_bundle_path_no_manifest(tmp_path: Path) -> None:
+    """A config dir without a manifest is not an extracted bundle."""
+    _setup_extracted_dir(tmp_path, None, files={"partitions.csv": "csv\n"})
+
+    assert remap_bundle_path(f"{ORIGINAL_CONFIG_DIR}/partitions.csv") is None
+
+
+@pytest.mark.parametrize(
+    "manifest",
+    [
+        pytest.param("{not json", id="malformed_json"),
+        pytest.param("[]", id="not_a_dict"),
+        pytest.param(
+            _bundle_manifest_dict(**{ManifestKey.MANIFEST_VERSION: "x"}),
+            id="version_not_int",
+        ),
+        pytest.param(
+            _bundle_manifest_dict(**{ManifestKey.MANIFEST_VERSION: 0}),
+            id="version_zero",
+        ),
+        pytest.param(
+            _bundle_manifest_dict(**{ManifestKey.CONFIG_FILENAME: "other.yaml"}),
+            id="config_filename_mismatch",
+        ),
+        pytest.param(
+            {
+                ManifestKey.MANIFEST_VERSION: CURRENT_MANIFEST_VERSION,
+                ManifestKey.CONFIG_FILENAME: "test.yaml",
+            },
+            id="config_dir_missing",
+        ),
+        pytest.param(
+            _bundle_manifest_dict(**{ManifestKey.CONFIG_DIR: ""}),
+            id="config_dir_empty",
+        ),
+    ],
+)
+def test_remap_bundle_path_untrusted_manifest(
+    tmp_path: Path, manifest: dict[str, Any] | str
+) -> None:
+    """Manifests that do not look like this bundle's manifest are ignored."""
+    _setup_extracted_dir(tmp_path, manifest, files={"partitions.csv": "csv\n"})
+
+    assert remap_bundle_path(f"{ORIGINAL_CONFIG_DIR}/partitions.csv") is None
+
+
+def test_remap_bundle_path_unreadable_manifest_warns(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A present but broken manifest is reported, not silently ignored."""
+    _setup_extracted_dir(tmp_path, "{not json", files={"partitions.csv": "csv\n"})
+
+    assert remap_bundle_path(f"{ORIGINAL_CONFIG_DIR}/partitions.csv") is None
+    assert "ignoring unreadable" in caplog.text
+
+
+def test_remap_bundle_path_outside_original_config_dir(tmp_path: Path) -> None:
+    """Paths that were not under the original config dir are left alone."""
+    _setup_extracted_dir(tmp_path, _bundle_manifest_dict())
+
+    assert remap_bundle_path("/elsewhere/partitions.csv") is None
+
+
+def test_remap_bundle_path_bundled_copy_missing(tmp_path: Path) -> None:
+    """No remap when the bundle does not contain the file."""
+    _setup_extracted_dir(tmp_path, _bundle_manifest_dict())
+
+    assert remap_bundle_path(f"{ORIGINAL_CONFIG_DIR}/partitions.csv") is None
+
+
+def test_remap_bundle_path_manifest_read_once(tmp_path: Path) -> None:
+    """The manifest lookup result is cached for the rest of the run."""
+    extract_dir = _setup_extracted_dir(
+        tmp_path, _bundle_manifest_dict(), files={"partitions.csv": "csv\n"}
+    )
+
+    first = remap_bundle_path(f"{ORIGINAL_CONFIG_DIR}/partitions.csv")
+    assert first == extract_dir / "partitions.csv"
+
+    (extract_dir / MANIFEST_FILENAME).unlink()
+    second = remap_bundle_path(f"{ORIGINAL_CONFIG_DIR}/partitions.csv")
+    assert second == first
+
+
+def test_remap_bundle_path_round_trip(tmp_path: Path) -> None:
+    """A file referenced by absolute path survives bundle create and extract.
+
+    Reproduces https://github.com/esphome/esphome/issues/17755: the config
+    names its partitions csv by absolute path, the bundle is extracted on a
+    machine where that path does not exist, and the reference must resolve
+    to the bundled copy.
+    """
+    config_dir = _setup_config_dir(tmp_path, files={"partitions.csv": "nvs,data\n"})
+    abs_path = (config_dir / "partitions.csv").resolve()
+
+    creator = ConfigBundleCreator({"esp32": {"partitions": abs_path}})
+    result = creator.create_bundle()
+
+    bundle_path = tmp_path / f"device{BUNDLE_EXTENSION}"
+    bundle_path.write_bytes(result.data)
+    target = tmp_path / "build_server"
+    config_path = extract_bundle(bundle_path, target)
+
+    # Simulate the build server: fresh run, original config dir gone
+    CORE.reset()
+    CORE.config_path = config_path
+    shutil.rmtree(config_dir)
+
+    remapped = remap_bundle_path(str(abs_path))
+    assert remapped == target.resolve() / "partitions.csv"
+    assert remapped.is_file()
 
 
 # ---------------------------------------------------------------------------
@@ -1261,7 +1526,7 @@ def test_create_bundle_produces_valid_archive(tmp_path: Path) -> None:
 
 
 def test_create_bundle_manifest_content(tmp_path: Path) -> None:
-    _setup_config_dir(tmp_path)
+    config_dir = _setup_config_dir(tmp_path)
 
     creator = ConfigBundleCreator({})
     result = creator.create_bundle()
@@ -1269,6 +1534,7 @@ def test_create_bundle_manifest_content(tmp_path: Path) -> None:
     manifest = result.manifest
     assert manifest[ManifestKey.MANIFEST_VERSION] == CURRENT_MANIFEST_VERSION
     assert manifest[ManifestKey.CONFIG_FILENAME] == "test.yaml"
+    assert manifest[ManifestKey.CONFIG_DIR] == str(config_dir.resolve())
     assert "test.yaml" in manifest[ManifestKey.FILES]
 
 

tests/unit_tests/test_compiled_config.py

@@ -74,13 +74,13 @@ def _write_storage(
         "framework": "arduino",
         "core_platform": core_platform,
     }
-    storage_path.write_text(json.dumps(data))
+    storage_path.write_text(json.dumps(data), encoding="utf-8")
 
 
 def _write_cache(cache_path: Path, body: str = _VALIDATED_CONFIG_YAML) -> Path:
     """Write the cache file and return it."""
     cache_path.parent.mkdir(parents=True, exist_ok=True)
-    cache_path.write_text(body)
+    cache_path.write_text(body, encoding="utf-8")
     return cache_path
 
 
@@ -220,7 +220,7 @@ def test_run_esphome_upload_and_logs_use_cache_when_fresh(
 
     with (
         caplog.at_level("INFO", logger="esphome.__main__"),
-        patch("esphome.__main__.read_config") as mock_read,
+        patch("esphome.config.read_config") as mock_read,
         patch.dict("esphome.__main__.POST_CONFIG_ACTIONS", {command: _stub}),
     ):
         assert run_esphome(["esphome", command, str(fresh_cache_files)]) == 0
@@ -242,7 +242,7 @@ def test_run_esphome_upload_and_logs_fall_back_when_no_cache(
     yaml_path.write_text("esphome:\n  name: lite_test\n")
 
     with (
-        patch("esphome.__main__.read_config", return_value=None) as mock_read,
+        patch("esphome.config.read_config", return_value=None) as mock_read,
         patch.dict(
             "esphome.__main__.POST_CONFIG_ACTIONS",
             {command: lambda args, config: 0},
@@ -266,7 +266,7 @@ def test_run_esphome_upload_does_not_refresh_cache_without_sidecar(
 
     with (
         patch(
-            "esphome.__main__.read_config",
+            "esphome.config.read_config",
             return_value={"esphome": {"name": "lite_test"}},
         ),
         patch("esphome.compiled_config.save_compiled_config") as mock_save,
@@ -299,7 +299,7 @@ def test_run_esphome_upload_and_logs_refresh_cache_on_fallback(
     fresh_config = {"esphome": {"name": "lite_test"}, "logger": {}}
 
     with (
-        patch("esphome.__main__.read_config", return_value=fresh_config),
+        patch("esphome.config.read_config", return_value=fresh_config),
         patch(
             "esphome.compiled_config.save_compiled_config", wraps=save_compiled_config
         ) as mock_save,
@@ -322,7 +322,7 @@ def test_run_esphome_upload_with_substitution_does_not_refresh_cache(
     """`-s` substitutions skip the cache on both read and write -- saving
     here would clobber the cache with a substitution-specific config."""
     with (
-        patch("esphome.__main__.read_config", return_value={"esphome": {}}),
+        patch("esphome.config.read_config", return_value={"esphome": {}}),
         patch("esphome.compiled_config.save_compiled_config") as mock_save,
         patch.dict(
             "esphome.__main__.POST_CONFIG_ACTIONS",
@@ -341,7 +341,7 @@ def test_run_esphome_compile_does_not_refresh_cache_via_fallback(
     upload/logs fallback path -- the fallback save would skip the
     storage_should_clean check."""
     with (
-        patch("esphome.__main__.read_config", return_value={"esphome": {}}),
+        patch("esphome.config.read_config", return_value={"esphome": {}}),
         patch("esphome.compiled_config.save_compiled_config") as mock_save,
         patch.dict(
             "esphome.__main__.POST_CONFIG_ACTIONS",
@@ -360,7 +360,7 @@ def test_run_esphome_upload_with_substitution_skips_cache(
     against the prior substitution set, so reusing it would silently
     ignore the override."""
     with (
-        patch("esphome.__main__.read_config", return_value=None) as mock_read,
+        patch("esphome.config.read_config", return_value=None) as mock_read,
         patch.dict(
             "esphome.__main__.POST_CONFIG_ACTIONS",
             {"upload": lambda args, config: 0},
@@ -374,7 +374,7 @@ def test_run_esphome_upload_with_substitution_skips_cache(
 def test_run_esphome_compile_does_not_use_cache(fresh_cache_files: Path) -> None:
     """The compile subcommand always re-validates -- it's what writes the cache."""
     with (
-        patch("esphome.__main__.read_config", return_value=None) as mock_read,
+        patch("esphome.config.read_config", return_value=None) as mock_read,
         patch.dict(
             "esphome.__main__.POST_CONFIG_ACTIONS",
             {"compile": lambda args, config: 0},

tests/unit_tests/test_config_validation.py

@@ -1,3 +1,5 @@
+import json
+import logging
 from pathlib import Path
 import string
 
@@ -2912,3 +2914,135 @@ def test_rename_key_present() -> None:
 
 def test_rename_key_absent() -> None:
     assert cv.rename_key("old", "new")({"other": 5}) == {"other": 5}
+
+
+def test_rename_key_no_removed_in_is_silent(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    with caplog.at_level(logging.WARNING, logger="esphome.config_validation"):
+        assert cv.rename_key("old", "new")({"old": 5}) == {"new": 5}
+    assert not caplog.records
+
+
+def test_rename_key_removed_in_renames_and_warns(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    with caplog.at_level(logging.WARNING, logger="esphome.config_validation"):
+        result = cv.rename_key("old", "new", removed_in="2026.8.0")({"old": 5})
+    assert result == {"new": 5}
+    assert "'old' is deprecated, use 'new'. Will be removed in 2026.8.0" in caplog.text
+
+
+def test_rename_key_removed_in_absent_key_no_warning(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    with caplog.at_level(logging.WARNING, logger="esphome.config_validation"):
+        result = cv.rename_key("old", "new", removed_in="2026.8.0")({"other": 5})
+    assert result == {"other": 5}
+    assert not caplog.records
+
+
+def test_rename_key_removed_in_with_component_prefixes_warning(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    with caplog.at_level(logging.WARNING, logger="esphome.config_validation"):
+        result = cv.rename_key(
+            "old", "new", removed_in="2026.8.0", component="my_component"
+        )({"old": 5})
+    assert result == {"new": 5}
+    assert (
+        "[my_component] 'old' is deprecated, use 'new'. Will be removed in 2026.8.0"
+        in caplog.text
+    )
+
+
+def test_rename_key_both_keys_rejected() -> None:
+    with pytest.raises(Invalid, match="Cannot specify more than one of"):
+        cv.rename_key("old", "new")({"old": 5, "new": 6})
+
+
+def test_rename_key_both_keys_rejected_with_removed_in(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    with (
+        caplog.at_level(logging.WARNING, logger="esphome.config_validation"),
+        pytest.raises(Invalid, match="Cannot specify more than one of"),
+    ):
+        cv.rename_key("old", "new", removed_in="2026.8.0")({"old": 5, "new": 6})
+    assert not caplog.records
+
+
+def test_file__existing_relative_path(setup_core: Path) -> None:
+    (setup_core / "partitions.csv").write_text("csv\n")
+
+    assert cv.file_("partitions.csv") == setup_core / "partitions.csv"
+
+
+def test_file__missing_raises(setup_core: Path) -> None:
+    with pytest.raises(Invalid, match="Could not find file"):
+        cv.file_("partitions.csv")
+
+
+def test_file__remaps_bundle_absolute_path(setup_core: Path) -> None:
+    """A stale absolute path in an extracted bundle resolves to the bundled copy."""
+    manifest = {
+        "manifest_version": 1,
+        "config_filename": "test.yaml",
+        "config_dir": "/original/config",
+    }
+    (setup_core / "manifest.json").write_text(json.dumps(manifest))
+    (setup_core / "partitions.csv").write_text("csv\n")
+
+    assert cv.file_("/original/config/partitions.csv") == setup_core / "partitions.csv"
+
+
+def test_file__missing_absolute_path_without_bundle(setup_core: Path) -> None:
+    with pytest.raises(Invalid, match="Could not find file"):
+        cv.file_("/original/config/partitions.csv")
+
+
+def test_file__remaps_windows_bundle_absolute_path(setup_core: Path) -> None:
+    """A bundle created on Windows resolves on a host with another layout."""
+    manifest = {
+        "manifest_version": 1,
+        "config_filename": "test.yaml",
+        "config_dir": "C:\\Users\\nick\\esphome",
+    }
+    (setup_core / "manifest.json").write_text(json.dumps(manifest))
+    (setup_core / "partitions.csv").write_text("csv\n")
+
+    result = cv.file_("C:\\Users\\nick\\esphome\\partitions.csv")
+
+    assert result == setup_core / "partitions.csv"
+
+
+def test_directory_remaps_bundle_absolute_path(setup_core: Path) -> None:
+    """A stale absolute directory in an extracted bundle resolves to the bundled copy."""
+    manifest = {
+        "manifest_version": 1,
+        "config_filename": "test.yaml",
+        "config_dir": "/original/config",
+    }
+    (setup_core / "manifest.json").write_text(json.dumps(manifest))
+    (setup_core / "headers").mkdir()
+
+    assert cv.directory("/original/config/headers") == setup_core / "headers"
+
+
+def test_directory_missing_raises(setup_core: Path) -> None:
+    with pytest.raises(Invalid, match="Could not find directory"):
+        cv.directory("/original/config/headers")
+
+
+def test_file__remapped_path_is_directory_raises(setup_core: Path) -> None:
+    """A remapped path that is a directory still fails file validation."""
+    manifest = {
+        "manifest_version": 1,
+        "config_filename": "test.yaml",
+        "config_dir": "/original/config",
+    }
+    (setup_core / "manifest.json").write_text(json.dumps(manifest))
+    (setup_core / "headers").mkdir()
+
+    with pytest.raises(Invalid, match="is not a file"):
+        cv.file_("/original/config/headers")

tests/unit_tests/test_espidf_framework.py

@@ -3,10 +3,14 @@
 # pylint: disable=protected-access
 
 from contextlib import contextmanager
+import importlib.util
 import io
 import json
 import logging
+import os
 from pathlib import Path
+import runpy
+import subprocess
 import sys
 import tarfile
 from types import SimpleNamespace
@@ -15,7 +19,10 @@ from unittest.mock import patch
 import pytest
 
 from esphome.espidf.framework import (
+    ESPHOME_STAMP_FILE,
+    STAMP_SCHEMA_VERSION,
     _ccache_env,
+    _check_esphome_idf_framework_install,
     _check_stamp,
     _check_windows_path_length,
     _clone_idf_with_submodules,
@@ -25,8 +32,11 @@ from esphome.espidf.framework import (
     _get_python_env_path,
     _get_python_version,
     _parse_git_source,
-    _patch_tools_json_demote_openocd,
+    _patch_tools_json_demote_unused_tools,
     _patch_tools_json_for_linux_arm64,
+    _prefetch_idf_tool_archives,
+    _read_stamp,
+    _stamp_covers,
     _windows_long_paths_enabled,
     _write_idf_version_txt,
     _write_stamp,
@@ -132,10 +142,17 @@ def test_parse_git_source_rejected(source: str) -> None:
     assert _parse_git_source(source) is None
 
 
-def _make_idf_tree(framework_path: Path) -> None:
-    """Create the minimum tree _clone_idf_with_submodules sanity-checks for."""
+def _make_idf_tree(framework_path: Path, *, gitmodules: bool = True) -> None:
+    """Create the minimum tree _clone_idf_with_submodules sanity-checks for.
+
+    ``gitmodules=False`` simulates a fork that vendors components in-tree
+    instead of declaring submodules; update_submodules skips the git call
+    when that file is missing.
+    """
     (framework_path / "tools").mkdir(parents=True)
     (framework_path / "tools" / "idf_tools.py").write_text("# stub\n")
+    if gitmodules:
+        (framework_path / ".gitmodules").write_text("# stub\n")
 
 
 def test_clone_idf_with_submodules_without_ref(tmp_path: Path) -> None:
@@ -209,6 +226,28 @@ def test_clone_idf_with_submodules_raises_when_tree_missing(
         )
 
 
+def test_clone_idf_accepts_flattened_fork_without_gitmodules(
+    tmp_path: Path,
+) -> None:
+    """A fork that vendors components in-tree instead of as submodules is valid.
+
+    No .gitmodules means the submodule step is skipped entirely.
+    """
+    framework_path = tmp_path / "idf"
+    framework_path.mkdir()
+    _make_idf_tree(framework_path, gitmodules=False)
+
+    with patch("esphome.git.run_git_command", return_value="") as run_git_command_mock:
+        _clone_idf_with_submodules(
+            framework_path,
+            "https://github.com/example/flattened-esp-idf.git",
+            None,
+        )
+
+    calls = [c.args[0] for c in run_git_command_mock.call_args_list]
+    assert not any(c[1] == "submodule" for c in calls)
+
+
 # ---------------------------------------------------------------------------
 # Helpers for _tar_extract_all hard-link prefix-stripping tests
 # ---------------------------------------------------------------------------
@@ -311,6 +350,21 @@ class TestTarExtractHardLinkPrefixStripping:
 _IDF_VERSION = "5.1.2"
 
 
+def _fake_download_from_mirrors(
+    mirrors: list[str],
+    substitutions: dict[str, str],
+    target: object,
+    **kwargs: object,
+) -> str:
+    """Stand-in for download_from_mirrors that creates path targets, since
+    the framework code opens the downloaded tarball afterwards."""
+    if isinstance(target, (str, os.PathLike)):
+        path = Path(target)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.touch()
+    return "https://example.com/idf.tar.xz"
+
+
 @pytest.fixture
 def espidf_mocks(setup_core: Path):
     """Patch the heavy I/O of check_esp_idf_install and pre-create the framework dir."""
@@ -321,7 +375,7 @@ def espidf_mocks(setup_core: Path):
         patch("esphome.espidf.framework.rmdir") as rmdir_mock,
         patch(
             "esphome.espidf.framework.download_from_mirrors",
-            return_value="https://example.com/idf.tar.xz",
+            side_effect=_fake_download_from_mirrors,
         ) as download,
         patch("esphome.espidf.framework.archive_extract_all") as extract,
         patch("esphome.espidf.framework.create_venv") as venv,
@@ -332,9 +386,11 @@ def espidf_mocks(setup_core: Path):
         patch("esphome.espidf.framework._clone_idf_with_submodules") as clone,
         patch("esphome.espidf.framework._write_idf_version_txt"),
         patch("esphome.espidf.framework._patch_tools_json_for_linux_arm64"),
-        patch("esphome.espidf.framework._patch_tools_json_demote_openocd"),
+        patch("esphome.espidf.framework._patch_tools_json_demote_unused_tools"),
+        patch("esphome.espidf.framework._prefetch_idf_tool_archives"),
         patch("esphome.espidf.framework._write_stamp"),
         patch("esphome.espidf.framework._check_stamp", return_value=True),
+        patch("esphome.espidf.framework._stamp_covers", return_value=True),
         patch("esphome.espidf.framework._get_idf_version", return_value=_IDF_VERSION),
         patch("esphome.espidf.framework._get_python_version", return_value="3.11.0"),
         patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
@@ -413,6 +469,20 @@ def test_check_esp_idf_install_already_installed(espidf_mocks: SimpleNamespace)
     espidf_mocks.venv.assert_not_called()
 
 
+def test_corrupt_tarball_removed_when_extraction_fails(
+    espidf_mocks: SimpleNamespace,
+) -> None:
+    """A tarball that fails to extract (e.g. torn by an unclean shutdown) is
+    deleted so the next run re-downloads instead of failing forever."""
+    espidf_mocks.extract.side_effect = RuntimeError("xz: unexpected end of input")
+    tarball = get_idf_tools_path() / "dist" / f"esp-idf-{_IDF_VERSION}.tar.xz"
+
+    with pytest.raises(RuntimeError, match="unexpected end of input"):
+        check_esp_idf_install(_IDF_VERSION, force=True)
+
+    assert not tarball.exists()
+
+
 def test_check_esp_idf_install_framework_failure(espidf_mocks: SimpleNamespace) -> None:
     """A failing idf_tools install raises."""
     espidf_mocks.run_ok.side_effect = [False]
@@ -450,13 +520,17 @@ def _mark_installed() -> None:
 def test_check_esp_idf_install_stamp_mismatch_reinstalls(
     espidf_mocks: SimpleNamespace,
 ) -> None:
-    """A stamp mismatch reinstalls tools (marker present, so no re-extract)."""
+    """A stamp mismatch reinstalls tools (marker present, so no re-extract).
+
+    The python env is left alone: it depends on the framework version and
+    features, not on which toolchains are installed.
+    """
     _mark_installed()
-    with patch("esphome.espidf.framework._check_stamp", return_value=False):
+    with patch("esphome.espidf.framework._stamp_covers", return_value=False):
         check_esp_idf_install(_IDF_VERSION)
 
     espidf_mocks.extract.assert_not_called()  # marker present -> no re-extract
-    espidf_mocks.venv.assert_called_once()  # tools reinstall -> venv rebuilt
+    espidf_mocks.venv.assert_not_called()  # tools-only install -> venv kept
 
 
 def test_check_esp_idf_install_check_command_failure_reinstalls(
@@ -469,7 +543,7 @@ def test_check_esp_idf_install_check_command_failure_reinstalls(
     check_esp_idf_install(_IDF_VERSION, features=["fb"])
 
     espidf_mocks.extract.assert_not_called()
-    espidf_mocks.venv.assert_called_once()
+    espidf_mocks.venv.assert_not_called()  # tools-only install -> venv kept
 
 
 def test_check_esp_idf_install_unknown_python_version_reinstalls(
@@ -489,8 +563,8 @@ def test_check_esp_idf_install_python_stamp_mismatch_rebuilds_venv(
 ) -> None:
     """Framework stamp matches but the python-env stamp does not -> venv rebuilt."""
 
-    # _check_stamp passes for the framework (no python_version key) and fails
-    # for the python env (carries python_version), so only the venv rebuilds.
+    # _check_stamp only guards the python env now (the framework uses
+    # _stamp_covers, patched True by the fixture); failing it rebuilds the venv.
     def stamp_ok(_stamp_file, info: dict) -> bool:
         return "python_version" not in info
 
@@ -502,6 +576,146 @@ def test_check_esp_idf_install_python_stamp_mismatch_rebuilds_venv(
     espidf_mocks.venv.assert_called_once()
 
 
+def _requested_stamp(targets: list[str], tools: list[str] | None = None) -> dict:
+    return {
+        "schema_version": STAMP_SCHEMA_VERSION,
+        "targets": targets,
+        "tools": tools or ["required"],
+    }
+
+
+@pytest.mark.parametrize(
+    ("stored", "targets", "expected"),
+    [
+        # a stored "all" covers any target
+        (_requested_stamp(["all"]), ["esp32"], True),
+        # exact match and superset both cover
+        (_requested_stamp(["esp32"]), ["esp32"], True),
+        (_requested_stamp(["esp32", "esp32c3"]), ["esp32"], True),
+        # a new target is not covered
+        (_requested_stamp(["esp32"]), ["esp32c3"], False),
+        # tools and schema_version must match exactly
+        (_requested_stamp(["all"], tools=["cmake", "required"]), ["esp32"], False),
+        (_requested_stamp(["all"]) | {"schema_version": "no"}, ["esp32"], False),
+        # an unknown extra field participates in invalidation by default
+        (_requested_stamp(["all"]) | {"module_version": 1}, ["esp32"], False),
+        # missing/corrupt stamps never cover
+        (None, ["esp32"], False),
+        (
+            {"schema_version": STAMP_SCHEMA_VERSION, "tools": ["required"]},
+            ["esp32"],
+            False,
+        ),
+    ],
+)
+def test_stamp_covers(stored: dict | None, targets: list[str], expected: bool) -> None:
+    assert _stamp_covers(stored, _requested_stamp(targets)) is expected
+
+
+@contextmanager
+def _framework_install_patches():
+    """Patches for calling _check_esphome_idf_framework_install directly with
+    real stamp files (unlike espidf_mocks, which stubs the stamp layer)."""
+    with (
+        patch("esphome.espidf.framework.run_command_ok", return_value=True) as run_ok,
+        patch("esphome.espidf.framework._get_idf_tool_paths", return_value=([], {})),
+        patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
+        patch("esphome.espidf.framework.rmdir"),
+    ):
+        yield run_ok
+
+
+def _extracted_framework_with_stamp(stamp: dict) -> Path:
+    framework_path = _get_framework_path(_IDF_VERSION)
+    framework_path.mkdir(parents=True, exist_ok=True)
+    (framework_path / ".esphome_extracted").touch()
+    _write_stamp(framework_path / ESPHOME_STAMP_FILE, stamp)
+    return framework_path
+
+
+def test_framework_install_target_subset_skips_install() -> None:
+    """A stamp holding a superset of the requested targets skips the installer."""
+    framework_path = _extracted_framework_with_stamp(_requested_stamp(["all"]))
+
+    with _framework_install_patches() as run_ok:
+        _, fresh_extract = _check_esphome_idf_framework_install(
+            _IDF_VERSION, ["esp32"], ["required"]
+        )
+
+    run_ok.assert_not_called()
+    assert fresh_extract is False
+    # the stamp is untouched
+    stamp = json.loads((framework_path / ESPHOME_STAMP_FILE).read_text())
+    assert stamp["targets"] == ["all"]
+
+
+def test_framework_install_new_target_installs_and_merges_stamp() -> None:
+    """A new target runs the installer for just that target and the stamp
+    records the union of everything installed so far."""
+    framework_path = _extracted_framework_with_stamp(_requested_stamp(["esp32"]))
+
+    with _framework_install_patches() as run_ok:
+        _, fresh_extract = _check_esphome_idf_framework_install(
+            _IDF_VERSION, ["esp32c3"], ["required"]
+        )
+
+    assert fresh_extract is False
+    assert "--targets=esp32c3" in run_ok.call_args[0][0]
+    stamp = json.loads((framework_path / ESPHOME_STAMP_FILE).read_text())
+    assert stamp["targets"] == ["esp32", "esp32c3"]
+
+
+def test_check_esp_idf_install_env_targets_override_wins(
+    espidf_mocks: SimpleNamespace,
+) -> None:
+    """An explicitly set ESPHOME_IDF_DEFAULT_TARGETS overrides per-variant targets."""
+    with patch("esphome.espidf.framework._IDF_DEFAULT_TARGETS_EXPLICIT", True):
+        check_esp_idf_install(_IDF_VERSION, force=True, targets=["esp32"])
+
+    install_cmd = espidf_mocks.run_ok.call_args_list[0][0][0]
+    assert "--targets=all" in install_cmd
+
+
+def test_check_esp_idf_install_uses_requested_targets(
+    espidf_mocks: SimpleNamespace,
+) -> None:
+    """Without the env override, the caller's per-variant targets are installed."""
+    check_esp_idf_install(_IDF_VERSION, force=True, targets=["esp32"])
+
+    install_cmd = espidf_mocks.run_ok.call_args_list[0][0][0]
+    assert "--targets=esp32" in install_cmd
+
+
+def test_framework_install_all_request_collapses_merged_stamp_to_all() -> None:
+    """Requesting "all" over a per-variant stamp merges and collapses to
+    ["all"], not ["all", "esp32"], so the stamp shape stays canonical."""
+    framework_path = _extracted_framework_with_stamp(_requested_stamp(["esp32"]))
+
+    with _framework_install_patches() as run_ok:
+        _check_esphome_idf_framework_install(_IDF_VERSION, ["all"], ["required"])
+
+    run_ok.assert_called_once()
+    stamp = json.loads((framework_path / ESPHOME_STAMP_FILE).read_text())
+    assert stamp["targets"] == ["all"]
+
+
+def test_framework_install_tools_change_resets_stamp_targets() -> None:
+    """A reinstall triggered by a tools change must not carry the old stamp's
+    targets forward: the installer only ran for this build's targets, so a
+    merged stamp would let other variants skip the reinstall they need."""
+    framework_path = _extracted_framework_with_stamp(
+        _requested_stamp(["all"], tools=["cmake", "required"])
+    )
+
+    with _framework_install_patches() as run_ok:
+        _check_esphome_idf_framework_install(_IDF_VERSION, ["esp32"], ["required"])
+
+    run_ok.assert_called_once()
+    stamp = json.loads((framework_path / ESPHOME_STAMP_FILE).read_text())
+    assert stamp["targets"] == ["esp32"]
+    assert stamp["tools"] == ["required"]
+
+
 @pytest.mark.parametrize(
     ("lib", "expect_hint"),
     [
@@ -637,28 +851,377 @@ def test_patch_tools_json_already_patched_is_noop(tmp_path: Path) -> None:
 
 
 # ---------------------------------------------------------------------------
-# _patch_tools_json_demote_openocd (openocd-esp32 made optional)
+# _prefetch_idf_tool_archives
 # ---------------------------------------------------------------------------
 
 
-def test_demote_openocd_patches_install_type(tmp_path: Path) -> None:
+_PREFETCH_JSON = json.dumps(
+    [
+        {
+            "name": "cmake@3.30.2",
+            "url": "https://example.com/cmake.tar.gz",
+            "size": 123,
+            "sha256": "ab" * 32,
+            "dest": "cmake-3.30.2.tar.gz",
+        },
+        {
+            "name": "ninja@1.12.1",
+            "url": "https://example.com/ninja.zip",
+            "size": 45,
+            "sha256": "cd" * 32,
+            "dest": "ninja.zip",
+        },
+    ]
+)
+
+
+def test_prefetch_downloads_each_archive_with_resume(tmp_path: Path) -> None:
+    with (
+        patch(
+            "esphome.espidf.framework.run_command",
+            return_value=(True, _PREFETCH_JSON, ""),
+        ),
+        patch("esphome.espidf.framework.download_with_resume") as download,
+        patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
+    ):
+        _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
+
+    dist = get_idf_tools_path() / "dist"
+    assert download.call_count == 2
+    assert download.call_args_list[0][0] == (
+        "https://example.com/cmake.tar.gz",
+        dist / "cmake-3.30.2.tar.gz",
+    )
+    assert download.call_args_list[0][1] == {"sha256": "ab" * 32, "size": 123}
+
+
+def test_prefetch_skips_already_downloaded_archives(tmp_path: Path) -> None:
+    dist = get_idf_tools_path() / "dist"
+    dist.mkdir(parents=True)
+    (dist / "cmake-3.30.2.tar.gz").write_bytes(b"cached")
+    with (
+        patch(
+            "esphome.espidf.framework.run_command",
+            return_value=(True, _PREFETCH_JSON, ""),
+        ),
+        patch("esphome.espidf.framework.download_with_resume") as download,
+        patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
+    ):
+        _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
+
+    # only the missing archive is downloaded
+    assert download.call_count == 1
+    assert download.call_args[0][1] == dist / "ninja.zip"
+
+
+@pytest.mark.parametrize(
+    ("run_result", "download_error", "expected_log"),
+    [
+        ((False, "", "script exploded"), None, "tool downloads"),  # script failure
+        ((True, "{ not json", ""), None, "prefetch failed"),  # unparsable output
+        (
+            (True, _PREFETCH_JSON, ""),
+            OSError("network down"),
+            "Could not prefetch",
+        ),  # download failure
+    ],
+)
+def test_prefetch_failures_never_raise(
+    tmp_path: Path,
+    caplog: pytest.LogCaptureFixture,
+    run_result: tuple[bool, str, str],
+    download_error: Exception | None,
+    expected_log: str,
+) -> None:
+    """The prefetch is best-effort; idf_tools downloads whatever is missing."""
+    with (
+        patch("esphome.espidf.framework.run_command", return_value=run_result),
+        patch(
+            "esphome.espidf.framework.download_with_resume",
+            side_effect=download_error,
+        ),
+        patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
+    ):
+        _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
+
+    assert expected_log in caplog.text
+
+
+def test_prefetch_one_failed_archive_does_not_stop_the_rest(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A single archive failing its download must not abort the prefetch of
+    the remaining archives."""
+    with (
+        patch(
+            "esphome.espidf.framework.run_command",
+            return_value=(True, _PREFETCH_JSON, ""),
+        ),
+        patch(
+            "esphome.espidf.framework.download_with_resume",
+            side_effect=[OSError("network down"), None],
+        ) as download,
+        patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
+    ):
+        _prefetch_idf_tool_archives(tmp_path, "esp32", ["required"], None)
+
+    assert download.call_count == 2
+    assert "Could not prefetch cmake@3.30.2" in caplog.text
+
+
+def test_prefetch_passes_targets_and_tools_to_script(tmp_path: Path) -> None:
+    with (
+        patch(
+            "esphome.espidf.framework.run_command", return_value=(True, "[]", "")
+        ) as run,
+        patch("esphome.espidf.framework.get_system_python_path", return_value="python"),
+    ):
+        _prefetch_idf_tool_archives(
+            tmp_path, "esp32,esp32c3", ["required", "cmake"], {"IDF_TOOLS_PATH": "/x"}
+        )
+
+    cmd = run.call_args[0][0]
+    assert cmd[-3:] == ["esp32,esp32c3", "required", "cmake"]
+    assert cmd[1].endswith("get_tool_downloads.py")
+    # the script inherits the caller's env plus the framework tools PYTHONPATH
+    env = run.call_args[1]["env"]
+    assert env["IDF_TOOLS_PATH"] == "/x"
+    assert env["PYTHONPATH"] == str(tmp_path / "tools")
+
+
+def test_framework_install_prefetches_before_installer(
+    espidf_mocks: SimpleNamespace,
+) -> None:
+    """The prefetch runs before idf_tools.py install so the installer finds
+    the archives already in dist/."""
+    calls: list[str] = []
+    with (
+        patch(
+            "esphome.espidf.framework._prefetch_idf_tool_archives",
+            side_effect=lambda *a, **k: calls.append("prefetch"),
+        ),
+    ):
+        espidf_mocks.run_ok.side_effect = lambda *a, **k: (
+            calls.append("install") or True
+        )
+        check_esp_idf_install(_IDF_VERSION, force=True)
+
+    assert calls.index("prefetch") < calls.index("install")
+
+
+# ---------------------------------------------------------------------------
+# get_tool_downloads.py (against the stub idf_tools module in fixtures/)
+# ---------------------------------------------------------------------------
+
+
+_IDF_TOOLS_STUB_DIR = Path(__file__).parent / "fixtures" / "idf_tools_stub"
+
+
+def _run_downloads_script(
+    tmp_path: Path, *args: str, env_extra: dict[str, str] | None = None
+) -> subprocess.CompletedProcess[str]:
+    """Run the real get_tool_downloads.py against the stub idf_tools module."""
+    script = Path(__file__).parents[2] / "esphome" / "espidf" / "get_tool_downloads.py"
+    env = os.environ | {
+        "PYTHONPATH": str(_IDF_TOOLS_STUB_DIR),
+        "IDF_TOOLS_PATH": str(tmp_path / "tp"),
+    }
+    if env_extra:
+        env |= env_extra
+    return subprocess.run(
+        [sys.executable, str(script), str(tmp_path / "fw"), *args],
+        capture_output=True,
+        text=True,
+        env=env,
+        check=False,
+    )
+
+
+def test_get_tool_downloads_lists_missing_tools(tmp_path: Path) -> None:
+    """Installed versions are skipped, tools that fail their binary check are
+    still listed, rename_dist decides the dist filename, and idf_tools' stdout
+    chatter stays off the JSON channel."""
+    result = _run_downloads_script(tmp_path, "esp32", "required")
+
+    assert result.returncode == 0, result.stderr
+    downloads = {d["name"]: d for d in json.loads(result.stdout)}
+    # installed-tool@1.0 is already installed and must not be listed
+    assert set(downloads) == {"cmake@3.30.2", "ninja@1.12.1", "broken-tool@2.0"}
+    assert downloads["cmake@3.30.2"]["dest"] == "cmake.tar.gz"
+    assert downloads["cmake@3.30.2"]["size"] == 11
+    assert downloads["cmake@3.30.2"]["sha256"] == "aa"
+    # rename_dist overrides the URL basename
+    assert downloads["ninja@1.12.1"]["dest"] == "ninja-v1.zip"
+    # the stub prints informational lines; they must be on stderr
+    assert "Changed download URL" in result.stderr
+
+
+def test_get_tool_downloads_applies_mirror_rewrite(tmp_path: Path) -> None:
+    result = _run_downloads_script(
+        tmp_path,
+        "esp32",
+        "required",
+        env_extra={"TEST_MIRROR_PREFIX": "https://mirror.test/"},
+    )
+
+    assert result.returncode == 0, result.stderr
+    downloads = json.loads(result.stdout)
+    assert all(d["url"].startswith("https://mirror.test/") for d in downloads)
+
+
+def _run_downloads_inprocess(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+    *args: str,
+) -> list[dict]:
+    """Execute get_tool_downloads.py in-process against the stub idf_tools.
+
+    Unlike the subprocess variant this runs under coverage, exercising the
+    script's own lines.
+    """
+    spec = importlib.util.spec_from_file_location(
+        "idf_tools", _IDF_TOOLS_STUB_DIR / "idf_tools.py"
+    )
+    stub = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(stub)
+    monkeypatch.setitem(sys.modules, "idf_tools", stub)
+    monkeypatch.setenv("IDF_TOOLS_PATH", str(tmp_path / "tp"))
+    script = Path(__file__).parents[2] / "esphome" / "espidf" / "get_tool_downloads.py"
+    monkeypatch.setattr(sys, "argv", [str(script), str(tmp_path / "fw"), *args])
+    runpy.run_path(str(script))
+    return json.loads(capsys.readouterr().out)
+
+
+def test_get_tool_downloads_inprocess_full_flow(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """In-process run covering the whole script: required expansion,
+    installed/broken tools, rename_dist, and version pinning via tool@version."""
+    downloads = {
+        d["name"]: d
+        for d in _run_downloads_inprocess(
+            tmp_path, monkeypatch, capsys, "esp32", "required"
+        )
+    }
+    assert set(downloads) == {"cmake@3.30.2", "ninja@1.12.1", "broken-tool@2.0"}
+    assert downloads["ninja@1.12.1"]["dest"] == "ninja-v1.zip"
+    assert downloads["cmake@3.30.2"]["url"] == "https://gh.test/cmake.tar.gz"
+
+
+def test_get_tool_downloads_inprocess_explicit_tool_specs(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """Explicit tool names and tool@version specs resolve; unknown tools and
+    unknown versions are skipped."""
+    downloads = _run_downloads_inprocess(
+        tmp_path,
+        monkeypatch,
+        capsys,
+        "esp32",
+        "cmake@3.30.2",
+        "no-such-tool",
+        "cmake@9.9.9",
+    )
+    assert [d["name"] for d in downloads] == ["cmake@3.30.2"]
+
+
+# ---------------------------------------------------------------------------
+# _patch_tools_json_demote_unused_tools (openocd, gdb, ULP toolchain optional)
+# ---------------------------------------------------------------------------
+
+
+def test_demote_unused_tools_patches_install_type(tmp_path: Path) -> None:
     tools_json = _write_tools_json(
         tmp_path,
         {
             "tools": [
                 {"name": "openocd-esp32", "install": "always"},
-                {"name": "cmake", "install": "always"},
+                {"name": "xtensa-esp-elf-gdb", "install": "always"},
+                {"name": "riscv32-esp-elf-gdb", "install": "always"},
+                {"name": "esp32ulp-elf", "install": "always"},
+                {"name": "xtensa-esp-elf", "install": "always"},
+                {"name": "esp-rom-elfs", "install": "always"},
             ]
         },
     )
-    _patch_tools_json_demote_openocd(tmp_path)
+    _patch_tools_json_demote_unused_tools(tmp_path)
+
+    data = json.loads(tools_json.read_text(encoding="utf-8"))
+    install_types = {t["name"]: t["install"] for t in data["tools"]}
+    assert install_types == {
+        "openocd-esp32": "on_request",
+        "xtensa-esp-elf-gdb": "on_request",
+        "riscv32-esp-elf-gdb": "on_request",
+        "esp32ulp-elf": "on_request",
+        # the compiler toolchain and ROM ELFs stay required
+        "xtensa-esp-elf": "always",
+        "esp-rom-elfs": "always",
+    }
+
+
+def test_demote_unused_tools_drops_xtensa_from_riscv_targets(tmp_path: Path) -> None:
+    """riscv32-esp-elf loses the xtensa chips (ULP-RISC-V only, which ESPHome
+    never builds) but keeps its RISC-V targets; other tools are untouched."""
+    tools_json = _write_tools_json(
+        tmp_path,
+        {
+            "tools": [
+                {
+                    "name": "riscv32-esp-elf",
+                    "install": "always",
+                    "supported_targets": ["esp32s2", "esp32s3", "esp32c3", "esp32p4"],
+                },
+                {
+                    "name": "xtensa-esp-elf",
+                    "install": "always",
+                    "supported_targets": ["esp32", "esp32s2", "esp32s3"],
+                },
+            ]
+        },
+    )
+    _patch_tools_json_demote_unused_tools(tmp_path)
+
+    data = json.loads(tools_json.read_text(encoding="utf-8"))
+    riscv = next(t for t in data["tools"] if t["name"] == "riscv32-esp-elf")
+    xtensa = next(t for t in data["tools"] if t["name"] == "xtensa-esp-elf")
+    assert riscv["supported_targets"] == ["esp32c3", "esp32p4"]
+    assert riscv["install"] == "always"
+    assert xtensa["supported_targets"] == ["esp32", "esp32s2", "esp32s3"]
+
+
+def test_demote_unused_tools_bad_supported_targets_type_still_demotes(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A non-list supported_targets on riscv32-esp-elf must not abort the
+    other demotions; the targets patch is best-effort and logs the skip so a
+    silently resumed riscv download is diagnosable."""
+    tools_json = _write_tools_json(
+        tmp_path,
+        {
+            "tools": [
+                {
+                    "name": "riscv32-esp-elf",
+                    "install": "always",
+                    "supported_targets": None,
+                },
+                {"name": "openocd-esp32", "install": "always"},
+            ]
+        },
+    )
+    with caplog.at_level(logging.WARNING, logger="esphome.espidf.framework"):
+        _patch_tools_json_demote_unused_tools(tmp_path)
 
     data = json.loads(tools_json.read_text(encoding="utf-8"))
     openocd = next(t for t in data["tools"] if t["name"] == "openocd-esp32")
-    cmake = next(t for t in data["tools"] if t["name"] == "cmake")
+    riscv = next(t for t in data["tools"] if t["name"] == "riscv32-esp-elf")
     assert openocd["install"] == "on_request"
-    # other tools are left untouched
-    assert cmake["install"] == "always"
+    assert riscv["supported_targets"] is None
+    assert "Unexpected supported_targets" in caplog.text
 
 
 def test_patch_tools_json_unexpected_structure_warns_and_skips(
@@ -670,16 +1233,29 @@ def test_patch_tools_json_unexpected_structure_warns_and_skips(
     tools_json = tools_dir / "tools.json"
     tools_json.write_text('["not", "a", "dict"]', encoding="utf-8")
     before = tools_json.read_text(encoding="utf-8")
-    _patch_tools_json_demote_openocd(tmp_path)  # AttributeError -> skip
+    _patch_tools_json_demote_unused_tools(tmp_path)  # AttributeError -> skip
     assert tools_json.read_text(encoding="utf-8") == before
 
 
-def test_demote_openocd_already_patched_is_noop(tmp_path: Path) -> None:
+def test_demote_unused_tools_already_patched_is_noop(tmp_path: Path) -> None:
     tools_json = _write_tools_json(
-        tmp_path, {"tools": [{"name": "openocd-esp32", "install": "on_request"}]}
+        tmp_path,
+        {
+            "tools": [
+                {"name": "openocd-esp32", "install": "on_request"},
+                {"name": "xtensa-esp-elf-gdb", "install": "on_request"},
+                {"name": "riscv32-esp-elf-gdb", "install": "on_request"},
+                {"name": "esp32ulp-elf", "install": "on_request"},
+                {
+                    "name": "riscv32-esp-elf",
+                    "install": "always",
+                    "supported_targets": ["esp32c3", "esp32p4"],
+                },
+            ]
+        },
     )
     before = tools_json.read_text(encoding="utf-8")
-    _patch_tools_json_demote_openocd(tmp_path)
+    _patch_tools_json_demote_unused_tools(tmp_path)
     assert tools_json.read_text(encoding="utf-8") == before
 
 
@@ -909,6 +1485,54 @@ def test_check_stamp_corrupt_file(tmp_path: Path) -> None:
     assert _check_stamp(f, {"a": "1"}) is False
 
 
+def test_read_stamp_corrupt_file_warns(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    # A corrupt stamp forces a full reinstall on every build, so it warns
+    # where the normal missing-file case stays silent.
+    f = tmp_path / "s.json"
+    f.write_text("{ not json", encoding="utf-8")
+    with caplog.at_level(logging.WARNING, logger="esphome.espidf.framework"):
+        assert _read_stamp(f) is None
+    assert "Ignoring corrupt stamp file" in caplog.text
+
+
+def test_read_stamp_unreadable_file_warns(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    # An I/O fault (permissions, disk error) is distinguished from a simply
+    # missing stamp with a warning before falling back to reinstall.
+    f = tmp_path / "s.json"
+    f.write_text(json.dumps({"a": "1"}), encoding="utf-8")
+    with (
+        patch.object(Path, "open", side_effect=PermissionError("denied")),
+        caplog.at_level(logging.WARNING, logger="esphome.espidf.framework"),
+    ):
+        assert _read_stamp(f) is None
+    assert "Could not read stamp file" in caplog.text
+
+
+def test_read_stamp_non_dict_warns(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    # Well-formed JSON that is not an object is a fault, not a first install;
+    # it must leave a trace before forcing reinstalls.
+    f = tmp_path / "s.json"
+    f.write_text("null", encoding="utf-8")
+    with caplog.at_level(logging.WARNING, logger="esphome.espidf.framework"):
+        assert _read_stamp(f) is None
+    assert "unexpected type NoneType" in caplog.text
+
+
+def test_read_stamp_missing_file_is_silent(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    # Missing stamps are the normal first-install case and must not log.
+    with caplog.at_level(logging.DEBUG, logger="esphome.espidf.framework"):
+        assert _read_stamp(tmp_path / "nope.json") is None
+    assert "stamp file" not in caplog.text
+
+
 def test_write_idf_version_txt_writes_when_missing(tmp_path: Path) -> None:
     _write_idf_version_txt(tmp_path, "5.1.2")
     assert (tmp_path / "version.txt").read_text(encoding="utf-8") == "v5.1.2\n"

tests/unit_tests/test_espidf_toolchain.py

@@ -10,7 +10,13 @@ from unittest.mock import patch
 
 import pytest
 
-from esphome.const import CONF_FRAMEWORK, CONF_SOURCE
+from esphome.components.esp32.const import KEY_ESP32, KEY_VARIANT
+from esphome.const import (
+    CONF_COMPILE_PROCESS_LIMIT,
+    CONF_ESPHOME,
+    CONF_FRAMEWORK,
+    CONF_SOURCE,
+)
 from esphome.core import CORE, EsphomeError
 from esphome.espidf import toolchain
 
@@ -50,7 +56,7 @@ def test_get_esphome_esp_idf_paths_forwards_source_override():
         toolchain, "check_esp_idf_install", return_value=("/fw", "/penv")
     ) as mock_install:
         toolchain._get_esphome_esp_idf_paths("5.5.4")
-    mock_install.assert_called_once_with("5.5.4", source_url=url)
+    mock_install.assert_called_once_with("5.5.4", targets=None, source_url=url)
 
 
 def test_get_esphome_esp_idf_paths_no_override():
@@ -61,7 +67,28 @@ def test_get_esphome_esp_idf_paths_no_override():
         toolchain, "check_esp_idf_install", return_value=("/fw", "/penv")
     ) as mock_install:
         toolchain._get_esphome_esp_idf_paths("5.5.4")
-    mock_install.assert_called_once_with("5.5.4", source_url=None)
+    mock_install.assert_called_once_with("5.5.4", targets=None, source_url=None)
+
+
+def test_get_configured_targets_from_variant(monkeypatch: pytest.MonkeyPatch):
+    """The configured variant restricts the toolchain install to its target."""
+    monkeypatch.delenv("CI", raising=False)
+    CORE.data[KEY_ESP32] = {KEY_VARIANT: "ESP32S3"}
+    assert toolchain._get_configured_targets() == ["esp32s3"]
+
+
+def test_get_configured_targets_without_variant(monkeypatch: pytest.MonkeyPatch):
+    """No stored variant (e.g. tooling outside a build) keeps the default."""
+    monkeypatch.delenv("CI", raising=False)
+    CORE.data.pop(KEY_ESP32, None)
+    assert toolchain._get_configured_targets() is None
+
+
+def test_get_configured_targets_ci_installs_all(monkeypatch: pytest.MonkeyPatch):
+    """CI installs every target so the shared cache covers all variants."""
+    monkeypatch.setenv("CI", "true")
+    CORE.data[KEY_ESP32] = {KEY_VARIANT: "ESP32S3"}
+    assert toolchain._get_configured_targets() is None
 
 
 def _setup_build(setup_core: Path) -> tuple[Path, Path]:
@@ -309,6 +336,58 @@ def test_get_cmake_output_missing_build_does_not_resolve_idf_env(
     mock_run.assert_not_called()
 
 
+def test_run_idf_py_jobs_sets_build_jobs_env(setup_core: Path) -> None:
+    """The jobs argument is exported to idf.py as IDF_PY_BUILD_JOBS."""
+    _setup_build(setup_core)
+
+    with (
+        patch.object(toolchain, "_get_idf_path", return_value=Path("/idf")),
+        patch.object(toolchain, "_get_idf_env", return_value={"PATH": "/bin"}),
+        patch.object(toolchain, "_get_idf_tool", return_value="python"),
+        patch.object(toolchain.subprocess, "run") as mock_run,
+    ):
+        mock_run.return_value.returncode = 0
+
+        toolchain.run_idf_py("build", jobs=2)
+        env = mock_run.call_args.kwargs["env"]
+        assert env["IDF_PY_BUILD_JOBS"] == "2"
+        assert env["PATH"] == "/bin"
+
+        toolchain.run_idf_py("build")
+        env = mock_run.call_args.kwargs["env"]
+        assert "IDF_PY_BUILD_JOBS" not in env
+
+
+def test_run_compile_passes_compile_process_limit(setup_core: Path) -> None:
+    """compile_process_limit is forwarded to run_idf_py as the job limit."""
+    _setup_build(setup_core)
+    config = {CONF_ESPHOME: {CONF_COMPILE_PROCESS_LIMIT: 1}}
+
+    with (
+        patch.object(toolchain, "need_reconfigure", return_value=False),
+        patch.object(toolchain, "run_idf_py", return_value=0) as mock_run,
+        patch.object(toolchain, "print_summary"),
+    ):
+        assert toolchain.run_compile(config, verbose=False) == 0
+
+    mock_run.assert_called_once_with("build", "size", jobs=1)
+
+
+def test_run_compile_without_compile_process_limit(setup_core: Path) -> None:
+    """When no compile_process_limit is set, no job limit is passed to idf.py."""
+    _setup_build(setup_core)
+    config = {CONF_ESPHOME: {}}
+
+    with (
+        patch.object(toolchain, "need_reconfigure", return_value=False),
+        patch.object(toolchain, "run_idf_py", return_value=0) as mock_run,
+        patch.object(toolchain, "print_summary"),
+    ):
+        assert toolchain.run_compile(config, verbose=False) == 0
+
+    mock_run.assert_called_once_with("build", "size", jobs=None)
+
+
 def test_get_core_framework_version_from_core_data():
     """The version is read from CORE.data when validation populated it."""
     from esphome.components.esp32.const import KEY_ESP32, KEY_IDF_VERSION

tests/unit_tests/test_framework_helpers.py

@@ -2,8 +2,10 @@
 
 # pylint: disable=protected-access
 
+import hashlib
 import importlib.util
 import io
+import json
 import logging
 import os
 from pathlib import Path
@@ -16,6 +18,7 @@ import zipfile
 import pytest
 import requests as req
 
+from esphome import framework_helpers
 from esphome.core import EsphomeError
 from esphome.framework_helpers import (
     _7z_extract_all,
@@ -26,6 +29,7 @@ from esphome.framework_helpers import (
     archive_extract_all,
     create_venv,
     download_from_mirrors,
+    download_with_resume,
     get_project_compile_flags,
     get_project_cxx_compile_flags,
     get_project_link_flags,
@@ -507,7 +511,7 @@ class TestArchiveExtractAll:
 
 
 # ---------------------------------------------------------------------------
-# download_from_mirrors
+# download_from_mirrors / download_with_resume
 # ---------------------------------------------------------------------------
 
 
@@ -515,6 +519,8 @@ def _mock_response(content: bytes, ok: bool = True) -> MagicMock:
     r = MagicMock()
     r.__enter__.return_value = r
     r.__exit__.return_value = False
+    r.status_code = 200
+    r.ok = ok
     if ok:
         r.raise_for_status.return_value = None
     else:
@@ -524,6 +530,563 @@ def _mock_response(content: bytes, ok: bool = True) -> MagicMock:
     return r
 
 
+def _interrupted_response(content: bytes, etag: str | None = None) -> MagicMock:
+    """A response whose body yields ``content`` and then drops mid-stream.
+
+    ``etag`` makes the response resumable: without a validator the retry
+    logic restarts from zero rather than stitching unverified bytes.
+    """
+
+    def body(chunk_size):
+        yield content
+        raise req.exceptions.ChunkedEncodingError("connection dropped")
+
+    r = _mock_response(b"")
+    if etag is not None:
+        r.headers = {**r.headers, "ETag": etag}
+    r.iter_content.side_effect = body
+    return r
+
+
+def _resumed_response(content: bytes) -> MagicMock:
+    """An HTTP 206 response continuing an interrupted download."""
+    r = _mock_response(content)
+    r.status_code = 206
+    return r
+
+
+class TestOpenRanged:
+    def test_fresh_download_sends_no_range(self) -> None:
+        with patch("requests.get", return_value=_mock_response(b"x")) as mock_get:
+            resp, offset = framework_helpers._open_ranged("https://e.com/f", 0, 30)
+        assert offset == 0
+        assert mock_get.call_args[1]["headers"] == {}
+        assert resp is mock_get.return_value
+
+    def test_resume_kept_on_206(self) -> None:
+        with patch("requests.get", return_value=_resumed_response(b"x")):
+            _, offset = framework_helpers._open_ranged("https://e.com/f", 7, 30)
+        assert offset == 7
+
+    def test_resume_downgraded_on_200(self) -> None:
+        """A server that ignores the Range header forces a restart."""
+        with patch("requests.get", return_value=_mock_response(b"x")):
+            _, offset = framework_helpers._open_ranged("https://e.com/f", 7, 30)
+        assert offset == 0
+
+    def test_http_error_closes_response_and_raises(self) -> None:
+        r = _mock_response(b"", ok=False)
+        with (
+            patch("requests.get", return_value=r),
+            pytest.raises(req.HTTPError),
+        ):
+            framework_helpers._open_ranged("https://e.com/f", 0, 30)
+        r.close.assert_called_once()
+
+    def test_connect_error_propagates(self) -> None:
+        with (
+            patch("requests.get", side_effect=req.ConnectionError("refused")),
+            pytest.raises(req.ConnectionError),
+        ):
+            framework_helpers._open_ranged("https://e.com/f", 0, 30)
+
+
+class TestDownloadWithResume:
+    def test_downloads_and_renames(self, tmp_path: Path) -> None:
+        dest = tmp_path / "tool.tar.gz"
+        with patch("requests.get", return_value=_mock_response(b"data")) as mock_get:
+            download_with_resume("https://example.com/t", dest)
+        assert dest.read_bytes() == b"data"
+        assert not (tmp_path / "tool.tar.gz.part").exists()
+        # a fresh download must not send a Range header
+        assert "Range" not in mock_get.call_args[1]["headers"]
+
+    def test_mid_stream_drop_resumes_with_range(self, tmp_path: Path) -> None:
+        dest = tmp_path / "tool.tar.gz"
+        first = _interrupted_response(b"1234", etag='"v1"')
+        first.headers = {**first.headers, "content-length": "8"}
+        with patch(
+            "requests.get",
+            side_effect=[first, _resumed_response(b"5678")],
+        ) as mock_get:
+            download_with_resume("https://example.com/t", dest)
+        # earlier bytes were kept, remainder appended conditionally
+        assert dest.read_bytes() == b"12345678"
+        assert mock_get.call_args_list[1][1]["headers"] == {
+            "Range": "bytes=4-",
+            "If-Range": '"v1"',
+        }
+
+    def test_unverifiable_drop_without_length_restarts(self, tmp_path: Path) -> None:
+        """A validator alone is not enough to stitch when nothing can prove
+        the stitched file complete (no sha/size and no content-length)."""
+        dest = tmp_path / "tool.tar.gz"
+        with patch(
+            "requests.get",
+            side_effect=[
+                _interrupted_response(b"1234", etag='"v1"'),
+                _mock_response(b"full"),
+            ],
+        ) as mock_get:
+            download_with_resume("https://example.com/t", dest)
+        assert dest.read_bytes() == b"full"
+        assert "Range" not in mock_get.call_args_list[1][1]["headers"]
+
+    def test_resumed_clean_but_short_body_discarded(self, tmp_path: Path) -> None:
+        """A resumed stream that ends cleanly but short of the advertised
+        total is rejected and re-downloaded, not promoted."""
+        dest = tmp_path / "tool.tar.gz"
+        first = _interrupted_response(b"abcd", etag='"v1"')
+        first.headers = {**first.headers, "content-length": "8"}
+        # resume ends cleanly after only 2 of the 4 missing bytes
+        short = _resumed_response(b"ef")
+        full = _mock_response(b"abcdefgh")
+        full.headers = {**full.headers, "content-length": "8"}
+        with patch("requests.get", side_effect=[first, short, full]) as mock_get:
+            download_with_resume("https://example.com/t", dest)
+        assert dest.read_bytes() == b"abcdefgh"
+        # the short stitch was discarded; the final attempt started fresh
+        assert "Range" not in mock_get.call_args_list[2][1]["headers"]
+
+    def test_unverifiable_drop_without_validator_restarts(self, tmp_path: Path) -> None:
+        """No sha/size and no server validator: the retry must not stitch."""
+        dest = tmp_path / "tool.tar.gz"
+        with patch(
+            "requests.get",
+            side_effect=[_interrupted_response(b"1234"), _mock_response(b"full")],
+        ) as mock_get:
+            download_with_resume("https://example.com/t", dest)
+        assert dest.read_bytes() == b"full"
+        assert "Range" not in mock_get.call_args_list[1][1]["headers"]
+
+    def test_resume_across_invocations_from_part_file(self, tmp_path: Path) -> None:
+        """A .part file left by a previous run is resumed, not restarted,
+        when sha/size verification will vouch for the stitched result."""
+        dest = tmp_path / "tool.tar.gz"
+        (tmp_path / "tool.tar.gz.part").write_bytes(b"12345")
+        good = hashlib.sha256(b"12345678").hexdigest()
+        with patch("requests.get", return_value=_resumed_response(b"678")) as mock_get:
+            download_with_resume("https://example.com/t", dest, sha256=good, size=8)
+        assert dest.read_bytes() == b"12345678"
+        assert mock_get.call_args[1]["headers"] == {"Range": "bytes=5-"}
+
+    def test_unverifiable_leftover_part_file_ignored(self, tmp_path: Path) -> None:
+        """Without sha/size there is no way to vouch for a cross-run stitch,
+        so a leftover part file starts over."""
+        dest = tmp_path / "tool.tar.gz"
+        (tmp_path / "tool.tar.gz.part").write_bytes(b"12345")
+        with patch("requests.get", return_value=_mock_response(b"fresh")) as mock_get:
+            download_with_resume("https://example.com/t", dest)
+        assert dest.read_bytes() == b"fresh"
+        assert "Range" not in mock_get.call_args[1]["headers"]
+
+    def test_server_without_range_support_restarts(self, tmp_path: Path) -> None:
+        """HTTP 200 in response to a Range request truncates and restarts."""
+        dest = tmp_path / "tool.tar.gz"
+        (tmp_path / "tool.tar.gz.part").write_bytes(b"sta")
+        good = hashlib.sha256(b"fresh").hexdigest()
+        with patch("requests.get", return_value=_mock_response(b"fresh")) as mock_get:
+            download_with_resume("https://example.com/t", dest, sha256=good, size=5)
+        # the Range request was sent (verifiable resume) and downgraded
+        assert mock_get.call_args[1]["headers"] == {"Range": "bytes=3-"}
+        assert dest.read_bytes() == b"fresh"
+
+    def test_size_only_leftover_part_restarts(self, tmp_path: Path) -> None:
+        """A size alone cannot detect a same-length content change on the
+        server, so a cross-run part without sha256 restarts from zero."""
+        dest = tmp_path / "tool.tar.gz"
+        (tmp_path / "tool.tar.gz.part").write_bytes(b"12")
+        with patch("requests.get", return_value=_mock_response(b"1234")) as mock_get:
+            download_with_resume("https://example.com/t", dest, size=4)
+        assert "Range" not in mock_get.call_args[1]["headers"]
+        assert dest.read_bytes() == b"1234"
+
+    def test_size_only_in_run_drop_resumes_with_validator(self, tmp_path: Path) -> None:
+        """Within a run the If-Range validator proves identity, so size-only
+        callers still resume mid-stream drops."""
+        dest = tmp_path / "tool.tar.gz"
+        with patch(
+            "requests.get",
+            side_effect=[
+                _interrupted_response(b"12", etag='"v1"'),
+                _resumed_response(b"34"),
+            ],
+        ) as mock_get:
+            download_with_resume("https://example.com/t", dest, size=4)
+        assert dest.read_bytes() == b"1234"
+        assert mock_get.call_args_list[1][1]["headers"] == {
+            "Range": "bytes=2-",
+            "If-Range": '"v1"',
+        }
+
+    def test_unverifiable_download_logged(
+        self, tmp_path: Path, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """No sha, no size, no content-length: the download is promoted with
+        a debug note (routine for e.g. the constraints host, so not a
+        warning) that completeness could not be verified."""
+        dest = tmp_path / "tool.tar.gz"
+        with (
+            caplog.at_level(logging.DEBUG, logger="esphome.framework_helpers"),
+            patch("requests.get", return_value=_mock_response(b"data")),
+        ):
+            download_with_resume("https://example.com/t", dest)
+        assert dest.read_bytes() == b"data"
+        assert "without any way to verify completeness" in caplog.text
+
+    def test_416_promotes_complete_part_when_size_unknown(self, tmp_path: Path) -> None:
+        """sha256-only caller with a byte-complete part file: the server's
+        416 confirms nothing is missing, verification promotes in place, and
+        the 416 must not loop as a retryable error."""
+        dest = tmp_path / "tool.tar.gz"
+        (tmp_path / "tool.tar.gz.part").write_bytes(b"data")
+        good = hashlib.sha256(b"data").hexdigest()
+        r416 = _mock_response(b"", ok=False)
+        r416.status_code = 416
+        with patch("requests.get", return_value=r416) as mock_get:
+            download_with_resume("https://example.com/t", dest, sha256=good)
+        assert mock_get.call_count == 1
+        r416.close.assert_called_once()
+        assert dest.read_bytes() == b"data"
+
+    def test_416_with_corrupt_part_discards_and_redownloads(
+        self, tmp_path: Path
+    ) -> None:
+        dest = tmp_path / "tool.tar.gz"
+        (tmp_path / "tool.tar.gz.part").write_bytes(b"bad!")
+        good = hashlib.sha256(b"data").hexdigest()
+        r416 = _mock_response(b"", ok=False)
+        r416.status_code = 416
+        with patch(
+            "requests.get", side_effect=[r416, _mock_response(b"data")]
+        ) as mock_get:
+            download_with_resume("https://example.com/t", dest, sha256=good)
+        assert "Range" not in mock_get.call_args_list[1][1]["headers"]
+        assert dest.read_bytes() == b"data"
+
+    def test_hash_mismatch_discards_and_retries(self, tmp_path: Path) -> None:
+        dest = tmp_path / "tool.tar.gz"
+        good = hashlib.sha256(b"good").hexdigest()
+        with patch(
+            "requests.get",
+            side_effect=[_mock_response(b"bad!"), _mock_response(b"good")],
+        ) as mock_get:
+            download_with_resume("https://example.com/t", dest, sha256=good, size=4)
+        assert dest.read_bytes() == b"good"
+        # the corrupt part file was discarded, so the retry starts fresh
+        assert "Range" not in mock_get.call_args_list[1][1]["headers"]
+
+    def test_size_mismatch_discards_part(self, tmp_path: Path) -> None:
+        dest = tmp_path / "tool.tar.gz"
+        with (
+            patch("requests.get", return_value=_mock_response(b"xx")),
+            pytest.raises(EsphomeError, match="after 2 attempts"),
+        ):
+            download_with_resume("https://example.com/t", dest, size=99, attempts=2)
+        assert not (tmp_path / "tool.tar.gz.part").exists()
+        assert not dest.exists()
+
+    def test_attempts_exhausted_keeps_part_file(self, tmp_path: Path) -> None:
+        """Mid-stream failures keep the partial file so a later run resumes."""
+        dest = tmp_path / "tool.tar.gz"
+        first = _interrupted_response(b"12", etag='"v1"')
+        first.headers = {**first.headers, "content-length": "4"}
+        second = _interrupted_response(b"34")
+        second.status_code = 206
+        with (
+            patch("requests.get", side_effect=[first, second]),
+            pytest.raises(EsphomeError, match="after 2 attempts"),
+        ):
+            download_with_resume("https://example.com/t", dest, attempts=2)
+        assert (tmp_path / "tool.tar.gz.part").read_bytes() == b"1234"
+
+    def test_multiple_drops_accumulate_across_attempts(self, tmp_path: Path) -> None:
+        """Each attempt appends its bytes; three partial responses complete
+        the file."""
+        dest = tmp_path / "tool.tar.gz"
+        first = _interrupted_response(b"ab", etag='"v1"')
+        first.headers = {**first.headers, "content-length": "6"}
+        second = _interrupted_response(b"cd")
+        second.status_code = 206
+        third = _resumed_response(b"ef")
+        with patch(
+            "requests.get",
+            side_effect=[first, second, third],
+        ) as mock_get:
+            download_with_resume("https://example.com/t", dest)
+        assert dest.read_bytes() == b"abcdef"
+        expected = {"Range": "bytes=2-", "If-Range": '"v1"'}
+        assert mock_get.call_args_list[1][1]["headers"] == expected
+        expected = {"Range": "bytes=4-", "If-Range": '"v1"'}
+        assert mock_get.call_args_list[2][1]["headers"] == expected
+
+    def test_connect_error_then_success(self, tmp_path: Path) -> None:
+        """A connect error (no response at all) consumes an attempt and the
+        next attempt succeeds."""
+        dest = tmp_path / "tool.tar.gz"
+        with patch(
+            "requests.get",
+            side_effect=[req.ConnectionError("refused"), _mock_response(b"data")],
+        ):
+            download_with_resume("https://example.com/t", dest)
+        assert dest.read_bytes() == b"data"
+
+    def test_http_error_keeps_part_file(self, tmp_path: Path) -> None:
+        """A transient HTTP error (e.g. 503) must not discard resume state."""
+        dest = tmp_path / "tool.tar.gz"
+        (tmp_path / "tool.tar.gz.part").write_bytes(b"keep")
+        error = _mock_response(b"", ok=False)
+        error.status_code = 503
+        with (
+            patch("requests.get", return_value=error),
+            pytest.raises(EsphomeError, match="after 1 attempts"),
+        ):
+            download_with_resume("https://example.com/t", dest, attempts=1)
+        assert (tmp_path / "tool.tar.gz.part").read_bytes() == b"keep"
+
+    def test_creates_missing_parent_directories(self, tmp_path: Path) -> None:
+        dest = tmp_path / "dist" / "nested" / "tool.tar.gz"
+        with patch("requests.get", return_value=_mock_response(b"data")):
+            download_with_resume("https://example.com/t", dest)
+        assert dest.read_bytes() == b"data"
+
+    def test_verifies_both_size_and_sha(self, tmp_path: Path) -> None:
+        dest = tmp_path / "tool.tar.gz"
+        good = hashlib.sha256(b"data").hexdigest()
+        with patch("requests.get", return_value=_mock_response(b"data")):
+            download_with_resume("https://example.com/t", dest, sha256=good, size=4)
+        assert dest.read_bytes() == b"data"
+
+    def test_corrupt_partial_resumed_then_discarded_then_redownloaded(
+        self, tmp_path: Path
+    ) -> None:
+        """The full recovery cycle for a corrupted partial download: the
+        resume completes it, verification fails, the poisoned part file is
+        discarded, and the next attempt re-downloads from scratch."""
+        dest = tmp_path / "tool.tar.gz"
+        # a previous run left a corrupted 4-byte prefix behind
+        (tmp_path / "tool.tar.gz.part").write_bytes(b"BAD!")
+        good = hashlib.sha256(b"data66").hexdigest()
+        with patch(
+            "requests.get",
+            side_effect=[
+                _resumed_response(b"66"),  # resume "completes" the bad part
+                _mock_response(b"data66"),  # clean retry from zero
+            ],
+        ) as mock_get:
+            download_with_resume("https://example.com/t", dest, sha256=good, size=6)
+        # first attempt resumed at the corrupt offset, failed verification;
+        # second attempt started fresh (no Range header) and succeeded
+        assert mock_get.call_args_list[0][1]["headers"] == {"Range": "bytes=4-"}
+        assert "Range" not in mock_get.call_args_list[1][1]["headers"]
+        assert dest.read_bytes() == b"data66"
+        assert not (tmp_path / "tool.tar.gz.part").exists()
+
+    def test_existing_dest_passing_verification_kept(self, tmp_path: Path) -> None:
+        """A dest completed by an earlier run is reused without any request."""
+        dest = tmp_path / "tool.tar.gz"
+        dest.write_bytes(b"data")
+        good = hashlib.sha256(b"data").hexdigest()
+        with patch("requests.get") as mock_get:
+            download_with_resume("https://example.com/t", dest, sha256=good, size=4)
+        mock_get.assert_not_called()
+        assert dest.read_bytes() == b"data"
+
+    @pytest.mark.parametrize(
+        "stale",
+        [
+            pytest.param(b"corrupt!", id="wrong-size"),
+            pytest.param(b"bad!", id="right-size-wrong-hash"),
+        ],
+    )
+    def test_existing_dest_failing_verification_redownloaded(
+        self, tmp_path: Path, stale: bytes
+    ) -> None:
+        dest = tmp_path / "tool.tar.gz"
+        dest.write_bytes(stale)
+        good = hashlib.sha256(b"data").hexdigest()
+        with patch("requests.get", return_value=_mock_response(b"data")):
+            download_with_resume("https://example.com/t", dest, sha256=good, size=4)
+        assert dest.read_bytes() == b"data"
+
+    def test_existing_dest_with_size_only_kept(self, tmp_path: Path) -> None:
+        dest = tmp_path / "tool.tar.gz"
+        dest.write_bytes(b"data")
+        with patch("requests.get") as mock_get:
+            download_with_resume("https://example.com/t", dest, size=4)
+        mock_get.assert_not_called()
+
+    def test_existing_dest_with_sha_only_kept(self, tmp_path: Path) -> None:
+        """sha-only verification also authorizes reusing a completed dest."""
+        dest = tmp_path / "tool.tar.gz"
+        dest.write_bytes(b"data")
+        good = hashlib.sha256(b"data").hexdigest()
+        with patch("requests.get") as mock_get:
+            download_with_resume("https://example.com/t", dest, sha256=good)
+        mock_get.assert_not_called()
+
+    def test_meta_write_failure_is_best_effort(self, tmp_path: Path) -> None:
+        """A failure to persist the resume sidecar must not fail the
+        download itself."""
+        dest = tmp_path / "f.tar.xz"
+        first = _mock_response(b"data")
+        first.headers = {**first.headers, "ETag": '"v1"', "content-length": "4"}
+        with (
+            patch("requests.get", return_value=first),
+            patch.object(Path, "write_text", side_effect=OSError("read-only")),
+        ):
+            download_with_resume("https://example.com/f", dest)
+        assert dest.read_bytes() == b"data"
+
+    def test_meta_sidecar_written_and_removed(self, tmp_path: Path) -> None:
+        """The validator sidecar appears while downloading and is cleaned up
+        with the promotion."""
+        dest = tmp_path / "f.tar.xz"
+        meta = tmp_path / "f.tar.xz.part.meta"
+        seen: list[bool] = []
+        first = _interrupted_response(b"1234", etag='"v1"')
+        first.headers = {**first.headers, "content-length": "8"}
+        responses = [first]
+
+        def get(*args: object, **kwargs: object) -> MagicMock:
+            if responses:
+                return responses.pop(0)
+            # the resume request: the sidecar written by the first response
+            # must already be on disk at this point
+            seen.append(meta.is_file())
+            return _resumed_response(b"5678")
+
+        with patch("requests.get", side_effect=get):
+            download_with_resume("https://example.com/f", dest)
+        assert dest.read_bytes() == b"12345678"
+        assert seen == [True]  # sidecar existed during the resume attempt
+        assert not meta.exists()  # cleaned up on success
+
+    def test_locked_promotion_keeps_verified_part(self, tmp_path: Path) -> None:
+        """A rename that stays blocked (e.g. a long-lived Windows file lock)
+        must not delete the verified download; the next attempt retries just
+        the rename without touching the network."""
+        dest = tmp_path / "tool.tar.gz"
+        good = hashlib.sha256(b"data").hexdigest()
+        with (
+            patch("requests.get", return_value=_mock_response(b"data")) as mock_get,
+            patch(
+                "esphome.framework_helpers._rename_with_retry",
+                side_effect=[PermissionError("locked"), None],
+            ) as rename,
+        ):
+            download_with_resume("https://example.com/t", dest, sha256=good, size=4)
+        # one download; the second attempt only redid the rename
+        assert mock_get.call_count == 1
+        assert rename.call_count == 2
+
+    def test_locked_promotion_exhausted_keeps_part_for_next_run(
+        self, tmp_path: Path
+    ) -> None:
+        dest = tmp_path / "tool.tar.gz"
+        good = hashlib.sha256(b"data").hexdigest()
+        with (
+            patch("requests.get", return_value=_mock_response(b"data")),
+            patch(
+                "esphome.framework_helpers._rename_with_retry",
+                side_effect=PermissionError("locked"),
+            ),
+            pytest.raises(EsphomeError, match="after 1 attempts"),
+        ):
+            download_with_resume(
+                "https://example.com/t", dest, sha256=good, size=4, attempts=1
+            )
+        # the verified bytes survive for the next run
+        assert (tmp_path / "tool.tar.gz.part").read_bytes() == b"data"
+
+    def test_meta_sidecar_resumes_across_runs_without_sha(self, tmp_path: Path) -> None:
+        """A later run resumes an unfinished download using the validator the
+        first run stored — the cross-run fix for the framework tarball."""
+        dest = tmp_path / "f.tar.xz"
+        (tmp_path / "f.tar.xz.part").write_bytes(b"1234")
+        (tmp_path / "f.tar.xz.part.meta").write_text(
+            json.dumps(
+                {"url": "https://example.com/f", "validator": '"v1"', "total": 8}
+            )
+        )
+        with patch("requests.get", return_value=_resumed_response(b"5678")) as mock_get:
+            download_with_resume("https://example.com/f", dest)
+        assert dest.read_bytes() == b"12345678"
+        assert mock_get.call_args[1]["headers"] == {
+            "Range": "bytes=4-",
+            "If-Range": '"v1"',
+        }
+
+    def test_meta_sidecar_for_other_url_ignored(self, tmp_path: Path) -> None:
+        """Metadata from a different mirror URL must not authorize a stitch."""
+        dest = tmp_path / "f.tar.xz"
+        (tmp_path / "f.tar.xz.part").write_bytes(b"1234")
+        (tmp_path / "f.tar.xz.part.meta").write_text(
+            json.dumps({"url": "https://other.com/f", "validator": '"v1"', "total": 8})
+        )
+        full = _mock_response(b"12345678")
+        with patch("requests.get", return_value=full) as mock_get:
+            download_with_resume("https://example.com/f", dest)
+        assert "Range" not in mock_get.call_args[1]["headers"]
+        assert dest.read_bytes() == b"12345678"
+
+    def test_complete_part_file_promoted_without_network(self, tmp_path: Path) -> None:
+        """A .part holding every byte (killed between write and rename) is
+        verified in place and promoted; no request is made, so no 416 loop."""
+        dest = tmp_path / "tool.tar.gz"
+        (tmp_path / "tool.tar.gz.part").write_bytes(b"data")
+        good = hashlib.sha256(b"data").hexdigest()
+        with patch("requests.get") as mock_get:
+            download_with_resume("https://example.com/t", dest, sha256=good, size=4)
+        mock_get.assert_not_called()
+        assert dest.read_bytes() == b"data"
+
+    def test_complete_but_corrupt_part_file_redownloaded(self, tmp_path: Path) -> None:
+        """A full-size .part with a wrong hash is discarded and re-downloaded
+        from scratch."""
+        dest = tmp_path / "tool.tar.gz"
+        (tmp_path / "tool.tar.gz.part").write_bytes(b"bad!")
+        good = hashlib.sha256(b"data").hexdigest()
+        with patch("requests.get", return_value=_mock_response(b"data")) as mock_get:
+            download_with_resume("https://example.com/t", dest, sha256=good, size=4)
+        assert "Range" not in mock_get.call_args[1]["headers"]
+        assert dest.read_bytes() == b"data"
+
+    def test_oversized_part_file_discarded(self, tmp_path: Path) -> None:
+        """A .part larger than the expected size fails verification and is
+        replaced by a fresh download."""
+        dest = tmp_path / "tool.tar.gz"
+        (tmp_path / "tool.tar.gz.part").write_bytes(b"toolong")
+        good = hashlib.sha256(b"data").hexdigest()
+        with patch("requests.get", return_value=_mock_response(b"data")):
+            download_with_resume("https://example.com/t", dest, sha256=good, size=4)
+        assert dest.read_bytes() == b"data"
+
+    def test_malformed_content_length_degrades_gracefully(self, tmp_path: Path) -> None:
+        """A garbage Content-Length must not crash the attempt; it means
+        "unknown", so a drop restarts instead of stitching and a clean
+        download still succeeds."""
+        dest = tmp_path / "tool.tar.gz"
+        first = _interrupted_response(b"1234", etag='"v1"')
+        first.headers = {**first.headers, "content-length": "explode"}
+        retry = _mock_response(b"full")
+        retry.headers = {**retry.headers, "content-length": "explode"}
+        with patch("requests.get", side_effect=[first, retry]) as mock_get:
+            download_with_resume("https://example.com/t", dest)
+        assert dest.read_bytes() == b"full"
+        # unknown length -> completeness unprovable -> no resume attempted
+        assert "Range" not in mock_get.call_args_list[1][1]["headers"]
+
+    def test_zero_byte_part_file_sends_no_range(self, tmp_path: Path) -> None:
+        """An empty leftover part file is a fresh download, not a resume."""
+        dest = tmp_path / "tool.tar.gz"
+        (tmp_path / "tool.tar.gz.part").write_bytes(b"")
+        with patch("requests.get", return_value=_mock_response(b"data")) as mock_get:
+            download_with_resume("https://example.com/t", dest)
+        assert mock_get.call_args[1]["headers"] == {}
+        assert dest.read_bytes() == b"data"
+
+
 class TestDownloadFromMirrors:
     def test_success_returns_url_and_writes_content(self, tmp_path: Path) -> None:
         target = tmp_path / "out.bin"
@@ -640,7 +1203,8 @@ class TestDownloadFromMirrors:
             ei.value
         )
 
-    def test_falls_back_to_second_mirror(self, tmp_path: Path) -> None:
+    def test_falls_back_to_second_mirror(self) -> None:
+        buf = io.BytesIO()
         with patch(
             "requests.get",
             side_effect=[_mock_response(b"", ok=False), _mock_response(b"second")],
@@ -648,14 +1212,152 @@ class TestDownloadFromMirrors:
             url = download_from_mirrors(
                 ["https://mirror1.com/f", "https://mirror2.com/f"],
                 {},
-                tmp_path / "out.bin",
+                buf,
             )
         assert url == "https://mirror2.com/f"
-        assert (tmp_path / "out.bin").read_bytes() == b"second"
+        assert buf.getvalue() == b"second"
 
-    def test_all_mirrors_fail_raises_error_listing_every_attempt(
-        self, tmp_path: Path
-    ) -> None:
+    def test_mid_stream_drop_resumes_same_mirror(self) -> None:
+        """A mid-stream failure retries the same mirror with Range and
+        If-Range headers, keeping the bytes already received, before falling
+        to the next."""
+        first = _interrupted_response(b"1234", etag='"v1"')
+        first.headers = {**first.headers, "content-length": "8"}
+        buf = io.BytesIO()
+        with patch(
+            "requests.get",
+            side_effect=[first, _resumed_response(b"5678")],
+        ) as mock_get:
+            url = download_from_mirrors(
+                ["https://mirror1.com/f", "https://mirror2.com/f"],
+                {},
+                buf,
+            )
+        assert url == "https://mirror1.com/f"
+        assert buf.getvalue() == b"12345678"
+        assert mock_get.call_count == 2
+        assert mock_get.call_args_list[1][0][0] == "https://mirror1.com/f"
+        # the resume is conditional on the content being unchanged
+        assert mock_get.call_args_list[1][1]["headers"] == {
+            "Range": "bytes=4-",
+            "If-Range": '"v1"',
+        }
+
+    def test_mid_stream_drop_without_validator_restarts(self) -> None:
+        """A server offering no ETag/Last-Modified cannot be resumed safely;
+        the retry restarts from zero instead of stitching unverified bytes."""
+        buf = io.BytesIO()
+        with patch(
+            "requests.get",
+            side_effect=[_interrupted_response(b"1234"), _mock_response(b"full")],
+        ) as mock_get:
+            download_from_mirrors(["https://mirror1.com/f"], {}, buf)
+        assert buf.getvalue() == b"full"
+        assert "Range" not in mock_get.call_args_list[1][1]["headers"]
+
+    def test_drop_after_last_byte_recovers_via_416(self) -> None:
+        """A connection drop after the final body byte leaves a complete file;
+        the retry's 416 answer plus the length check turn it into success
+        instead of a wasted refetch."""
+        first = _interrupted_response(b"1234", etag='"v1"')
+        first.headers = {**first.headers, "content-length": "4"}
+        r416 = _mock_response(b"", ok=False)
+        r416.status_code = 416
+        buf = io.BytesIO()
+        with patch("requests.get", side_effect=[first, r416]) as mock_get:
+            url = download_from_mirrors(["https://mirror1.com/f"], {}, buf)
+        assert url == "https://mirror1.com/f"
+        assert buf.getvalue() == b"1234"
+        assert mock_get.call_count == 2
+
+    def test_mirror_drop_without_length_restarts(self) -> None:
+        """With no content-length there is no way to prove a stitched file
+        complete, so the retry restarts even though a validator exists."""
+        buf = io.BytesIO()
+        with patch(
+            "requests.get",
+            side_effect=[
+                _interrupted_response(b"1234", etag='"v1"'),
+                _mock_response(b"full"),
+            ],
+        ) as mock_get:
+            download_from_mirrors(["https://mirror1.com/f"], {}, buf)
+        assert buf.getvalue() == b"full"
+        assert "Range" not in mock_get.call_args_list[1][1]["headers"]
+
+    def test_path_target_resumes_across_runs(self, tmp_path: Path) -> None:
+        """A path target routes through download_with_resume: a part file and
+        metadata from a previous run resume instead of restarting."""
+        dest = tmp_path / "idf.tar.xz"
+        (tmp_path / "idf.tar.xz.part").write_bytes(b"1234")
+        (tmp_path / "idf.tar.xz.part.meta").write_text(
+            json.dumps(
+                {"url": "https://mirror1.com/f", "validator": '"v1"', "total": 8}
+            )
+        )
+        with patch("requests.get", return_value=_resumed_response(b"5678")) as mock_get:
+            url = download_from_mirrors(["https://mirror1.com/f"], {}, dest)
+        assert url == "https://mirror1.com/f"
+        assert dest.read_bytes() == b"12345678"
+        assert mock_get.call_args[1]["headers"] == {
+            "Range": "bytes=4-",
+            "If-Range": '"v1"',
+        }
+
+    def test_path_target_falls_back_to_next_mirror(self, tmp_path: Path) -> None:
+        dest = tmp_path / "idf.tar.xz"
+        with patch(
+            "requests.get",
+            side_effect=[req.ConnectionError("down"), _mock_response(b"data")],
+        ):
+            url = download_from_mirrors(
+                ["https://mirror1.com/f", "https://mirror2.com/f"], {}, dest
+            )
+        assert url == "https://mirror2.com/f"
+        assert dest.read_bytes() == b"data"
+
+    def test_resumed_short_body_fails_length_check(self) -> None:
+        """A stitched file whose final length disagrees with the advertised
+        total is rejected instead of reported as success."""
+        first = _interrupted_response(b"1234", etag='"v1"')
+        first.headers = {**first.headers, "content-length": "8"}
+        # the resume ends early (5 of 8 bytes); the poisoned part is then
+        # discarded and the fresh retry also delivers a short body
+        short_resume = _resumed_response(b"5")
+        short_fresh = _mock_response(b"56")
+        short_fresh.headers = {**short_fresh.headers, "content-length": "8"}
+        buf = io.BytesIO()
+        with (
+            patch("requests.get", side_effect=[first, short_resume, short_fresh]),
+            pytest.raises(EsphomeError, match="all mirrors"),
+        ):
+            download_from_mirrors(["https://mirror1.com/f"], {}, buf)
+
+    def test_failed_mirror_leftovers_not_kept_for_next_mirror(self) -> None:
+        """Bytes from a mirror that failed all attempts must not leak into the
+        next mirror's download (no bogus Range request, fresh content)."""
+        exhausted = [_interrupted_response(b"AAAA", etag='"a1"')]
+        for _ in range(2):
+            r = _interrupted_response(b"BB")
+            r.status_code = 206
+            exhausted.append(r)
+        buf = io.BytesIO()
+        with patch(
+            "requests.get",
+            side_effect=exhausted + [_mock_response(b"clean")],
+        ) as mock_get:
+            url = download_from_mirrors(
+                ["https://mirror1.com/f", "https://mirror2.com/f"],
+                {},
+                buf,
+            )
+        assert url == "https://mirror2.com/f"
+        assert buf.getvalue() == b"clean"
+        # the second mirror starts fresh, without a Range header
+        assert mock_get.call_args_list[3][0][0] == "https://mirror2.com/f"
+        assert "Range" not in mock_get.call_args_list[3][1]["headers"]
+
+    def test_all_mirrors_fail_raises_error_listing_every_attempt(self) -> None:
         with (
             patch(
                 "requests.get",
@@ -666,7 +1368,7 @@ class TestDownloadFromMirrors:
             download_from_mirrors(
                 ["https://mirror1.com/f", "https://mirror2.com/f"],
                 {},
-                tmp_path / "out.bin",
+                io.BytesIO(),
             )
         # Every attempted URL appears in the message, and the first mirror's
         # exception (the primary URL, usually the one that matters) is chained.
@@ -731,8 +1433,10 @@ def test_importing_framework_helpers_does_not_import_requests() -> None:
         [
             sys.executable,
             "-c",
-            "import sys\nimport esphome.framework_helpers\n"
-            "print('\\n'.join(sys.modules))",
+            (
+                "import sys\nimport esphome.framework_helpers\n"
+                "print('\\n'.join(sys.modules))"
+            ),
         ],
         capture_output=True,
         text=True,

tests/unit_tests/test_git.py

@@ -1,8 +1,10 @@
 """Tests for git.py module."""
 
 from collections.abc import Callable
+import logging
 import os
 from pathlib import Path
+import subprocess
 import time
 from typing import Any
 from unittest.mock import Mock, patch
@@ -71,19 +73,40 @@ def _simulate_cloned_repo(repo_dir: Path) -> None:
     (repo_dir / ".git").mkdir(exist_ok=True)
 
 
-def _make_clone_side_effect(repo_dir: Path) -> Callable[..., str]:
-    """Return a run_git_command side effect whose clone creates the repo dir."""
+def _make_clone_side_effect(
+    repo_dir: Path, gitmodules: bool = False
+) -> Callable[..., str]:
+    """Return a run_git_command side effect whose clone creates the repo dir.
+
+    With ``gitmodules`` the cloned repo also declares submodules.
+    """
 
     def git_command_side_effect(
         cmd: list[str], cwd: str | None = None, **kwargs: Any
     ) -> str:
         if _get_git_command_type(cmd) == "clone":
             _simulate_cloned_repo(repo_dir)
+            if gitmodules:
+                (repo_dir / ".gitmodules").write_text("test")
         return ""
 
     return git_command_side_effect
 
 
+def _submodule_calls(mock: Mock) -> list[Any]:
+    """Return the mock's `git submodule` calls."""
+    return [
+        c for c in mock.call_args_list if _get_git_command_type(c[0][0]) == "submodule"
+    ]
+
+
+def _assert_submodule_runs_without_isolation(call: Any, repo_dir: Path) -> None:
+    """Assert a git submodule call ran with plain cwd, not GIT_DIR/GIT_WORK_TREE
+    isolation, which breaks the submodule porcelain on some installations."""
+    assert call.kwargs.get("git_dir") is None
+    assert call.kwargs.get("cwd") == repo_dir
+
+
 def test_run_git_command_success(tmp_path: Path) -> None:
     """Test that run_git_command returns output on success."""
     # Create a simple git repo to test with
@@ -100,6 +123,22 @@ def test_run_git_command_success(tmp_path: Path) -> None:
     assert isinstance(result, str)
 
 
+def test_run_git_command_debug_log_redacts_credentials(
+    tmp_path: Path, mock_subprocess_run: Mock, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Embedded URL credentials never reach the debug log; -v output is
+    routinely pasted into public issues. subprocess is mocked so no real
+    git ever sees the URL (the path is not creatable on Windows)."""
+    mock_subprocess_run.return_value = Mock(returncode=0, stdout=b"", stderr=b"")
+    with caplog.at_level(logging.DEBUG, logger="esphome.git"):
+        git.run_git_command(
+            ["git", "clone", "https://user:hunter2@github.com/test/repo"],
+            cwd=tmp_path,
+        )
+    assert "hunter2" not in caplog.text
+    assert "://***@github.com/test/repo" in caplog.text
+
+
 def test_run_git_command_with_git_dir_isolation(
     tmp_path: Path, mock_subprocess_run: Mock
 ) -> None:
@@ -116,10 +155,17 @@ def test_run_git_command_with_git_dir_isolation(
         stderr=b"",
     )
 
-    result = git.run_git_command(
-        ["git", "rev-parse", "HEAD"],
-        git_dir=repo_dir,
-    )
+    # Ambient repo-scoping vars simulate a git hook invoking ESPHome; an
+    # ambient GIT_INDEX_FILE surviving into a git_dir invocation fails
+    # silently (git operates on the caller's index and exits 0).
+    with patch.dict(
+        os.environ,
+        {"GIT_INDEX_FILE": "/caller/index", "GIT_OBJECT_DIRECTORY": "/caller/objects"},
+    ):
+        result = git.run_git_command(
+            ["git", "rev-parse", "HEAD"],
+            git_dir=repo_dir,
+        )
 
     # Verify subprocess.run was called
     assert mock_subprocess_run.called
@@ -131,6 +177,9 @@ def test_run_git_command_with_git_dir_isolation(
     assert "GIT_WORK_TREE" in env
     assert env["GIT_DIR"] == str(repo_dir / ".git")
     assert env["GIT_WORK_TREE"] == str(repo_dir)
+    # The ambient scoping vars must be stripped, not passed through.
+    assert "GIT_INDEX_FILE" not in env
+    assert "GIT_OBJECT_DIRECTORY" not in env
 
     assert result == "test output"
 
@@ -216,6 +265,89 @@ def test_run_git_command_without_git_dir(mock_subprocess_run: Mock) -> None:
     assert result == "Cloning into 'test_repo'..."
 
 
+@pytest.mark.parametrize("relative", [False, True], ids=["absolute", "relative"])
+def test_run_git_command_with_cwd_runs_in_dir_without_isolation(
+    tmp_path: Path,
+    mock_subprocess_run: Mock,
+    monkeypatch: pytest.MonkeyPatch,
+    relative: bool,
+) -> None:
+    """The cwd parameter sets the working directory without GIT_DIR/GIT_WORK_TREE.
+
+    Ambient GIT_DIR/GIT_WORK_TREE (e.g. from a git hook or CI wrapper) must be
+    stripped too, and GIT_CEILING_DIRECTORIES must stop git from walking up to
+    an enclosing repository if the target repo's .git is missing or corrupt.
+    Git silently ignores a relative ceiling entry, so the variable must come
+    out absolute even when the given cwd is relative.
+    """
+    repo_dir = tmp_path / "test_repo"
+    repo_dir.mkdir()
+    if relative:
+        monkeypatch.chdir(tmp_path)
+        cwd_arg = Path("test_repo")
+    else:
+        cwd_arg = repo_dir
+
+    mock_subprocess_run.return_value = Mock(
+        returncode=0,
+        stdout=b"test output",
+        stderr=b"",
+    )
+
+    with patch.dict(
+        os.environ,
+        {
+            "GIT_DIR": "/ambient/.git",
+            "GIT_WORK_TREE": "/ambient",
+            "GIT_INDEX_FILE": "/ambient/.git/index",
+        },
+    ):
+        result = git.run_git_command(["git", "submodule", "update"], cwd=cwd_arg)
+
+    call_args = mock_subprocess_run.call_args
+    env = call_args[1]["env"]
+    assert "GIT_DIR" not in env
+    assert "GIT_WORK_TREE" not in env
+    assert "GIT_INDEX_FILE" not in env
+    ceiling = Path(env["GIT_CEILING_DIRECTORIES"])
+    assert ceiling.is_absolute()
+    assert ceiling.samefile(tmp_path)
+    assert call_args[1]["cwd"] == cwd_arg
+    assert result == "test output"
+
+
+def test_run_git_command_raises_on_nonfatal_stderr(
+    tmp_path: Path, mock_subprocess_run: Mock
+) -> None:
+    """Nonzero exit with stderr lacking a fatal: prefix raises with full stderr."""
+    mock_subprocess_run.return_value = Mock(
+        returncode=1,
+        stdout=b"",
+        stderr=b"error: pathspec 'nope' did not match any file(s)\n",
+    )
+
+    with pytest.raises(GitCommandError, match="did not match"):
+        git.run_git_command(["git", "checkout", "nope"], git_dir=tmp_path)
+
+
+def test_run_git_command_raises_on_nonzero_exit_without_stderr(
+    tmp_path: Path, mock_subprocess_run: Mock
+) -> None:
+    """A nonzero exit must raise even when git printed nothing to stderr.
+
+    Silent nonzero exits were previously treated as success, which is how
+    broken checkouts could be cached as complete.
+    """
+    mock_subprocess_run.return_value = Mock(
+        returncode=1,
+        stdout=b"",
+        stderr=b"",
+    )
+
+    with pytest.raises(GitCommandError, match="exited with code 1"):
+        git.run_git_command(["git", "submodule", "update"], cwd=tmp_path)
+
+
 def test_run_git_command_without_git_dir_raises_error(
     mock_subprocess_run: Mock,
 ) -> None:
@@ -1156,46 +1288,6 @@ def test_clone_with_ref_uses_shallow_fetch(
     assert ref in fetch_calls[0][0][0]
 
 
-def test_clone_with_submodules_uses_shallow_submodule_update(
-    tmp_path: Path, mock_run_git_command: Mock
-) -> None:
-    """Submodule init on a fresh clone should use --depth=1."""
-    CORE.config_path = tmp_path / "test.yaml"
-
-    url = "https://github.com/test/repo"
-    domain = "test"
-    repo_dir = _compute_repo_dir(url, None, domain)
-
-    def git_command_side_effect(
-        cmd: list[str], cwd: str | None = None, **kwargs: Any
-    ) -> str:
-        if _get_git_command_type(cmd) == "clone":
-            repo_dir.mkdir(parents=True, exist_ok=True)
-            (repo_dir / ".git").mkdir(exist_ok=True)
-        return ""
-
-    mock_run_git_command.side_effect = git_command_side_effect
-
-    git.clone_or_update(
-        url=url,
-        ref=None,
-        refresh=None,
-        domain=domain,
-        submodules=["components/foo"],
-    )
-
-    submodule_calls = [
-        c for c in mock_run_git_command.call_args_list if "submodule" in c[0][0]
-    ]
-    assert len(submodule_calls) == 1
-    cmd = submodule_calls[0][0][0]
-    assert "--depth=1" in cmd
-    assert "components/foo" in cmd
-    # The `--` terminator must precede the submodule paths so a path
-    # beginning with `-` cannot be parsed as an option.
-    assert cmd.index("--") < cmd.index("components/foo")
-
-
 def test_refresh_fetch_is_shallow(tmp_path: Path, mock_run_git_command: Mock) -> None:
     """The refresh-path fetch should use --depth=1."""
     CORE.config_path = tmp_path / "test.yaml"
@@ -1220,10 +1312,91 @@ def test_refresh_fetch_is_shallow(tmp_path: Path, mock_run_git_command: Mock) ->
     assert cmd[-1] == ref
 
 
-def test_refresh_submodule_update_is_shallow(
+@pytest.mark.parametrize(
+    "refresh", [None, TimePeriodSeconds(days=1)], ids=["clone", "refresh"]
+)
+def test_all_submodules_skipped_without_gitmodules(
+    tmp_path: Path, mock_run_git_command: Mock, refresh: TimePeriodSeconds | None
+) -> None:
+    """init_submodules is a no-op for repos with no .gitmodules.
+
+    This is the esp-idf toolchain library scenario from issue #17860: the
+    PlatformIO library converter requests "all submodules" for every git
+    library, and most libraries declare none. The git submodule porcelain
+    must not run at all in that case — it fails outright on some git
+    installations.
+    """
+    CORE.config_path = tmp_path / "test.yaml"
+
+    url = "https://github.com/test/repo"
+    domain = "test"
+    repo_dir = _compute_repo_dir(url, None, domain)
+
+    if refresh is None:
+        mock_run_git_command.side_effect = _make_clone_side_effect(repo_dir)
+    else:
+        _setup_old_repo(repo_dir)
+        mock_run_git_command.return_value = "abc123"
+
+    git.clone_or_update(
+        url=url,
+        ref=None,
+        refresh=refresh,
+        domain=domain,
+        init_submodules=True,
+    )
+
+    assert not _submodule_calls(mock_run_git_command)
+
+
+@pytest.mark.parametrize(
+    "refresh", [None, TimePeriodSeconds(days=1)], ids=["clone", "refresh"]
+)
+def test_all_submodules_updated_with_gitmodules(
+    tmp_path: Path, mock_run_git_command: Mock, refresh: TimePeriodSeconds | None
+) -> None:
+    """init_submodules initializes all submodules when .gitmodules exists."""
+    CORE.config_path = tmp_path / "test.yaml"
+
+    url = "https://github.com/test/repo"
+    domain = "test"
+    repo_dir = _compute_repo_dir(url, None, domain)
+
+    if refresh is None:
+        mock_run_git_command.side_effect = _make_clone_side_effect(
+            repo_dir, gitmodules=True
+        )
+    else:
+        _setup_old_repo(repo_dir)
+        (repo_dir / ".gitmodules").write_text("test")
+        mock_run_git_command.return_value = "abc123"
+
+    git.clone_or_update(
+        url=url,
+        ref=None,
+        refresh=refresh,
+        domain=domain,
+        init_submodules=True,
+    )
+
+    submodule_calls = _submodule_calls(mock_run_git_command)
+    # Which submodules get populated is git's own policy, so no status
+    # verification follows the update.
+    assert len(submodule_calls) == 1
+    cmd = submodule_calls[0][0][0]
+    assert cmd[2] == "update"
+    assert "--depth=1" in cmd
+    # Recursive, mirroring PlatformIO's recursive library clones.
+    assert "--recursive" in cmd
+    _assert_submodule_runs_without_isolation(submodule_calls[0], repo_dir)
+
+
+def test_recovery_reclone_keeps_credentials_and_cache_key(
     tmp_path: Path, mock_run_git_command: Mock
 ) -> None:
-    """The refresh-path submodule update should use --depth=1."""
+    """The recovery re-clone must not re-apply credentials to the already
+    rewritten URL (no doubled userinfo) and must land in the same cache
+    directory, or a credentialed private repo re-clones on every run."""
     CORE.config_path = tmp_path / "test.yaml"
 
     url = "https://github.com/test/repo"
@@ -1231,24 +1404,235 @@ def test_refresh_submodule_update_is_shallow(
     repo_dir = _compute_repo_dir(url, None, domain)
 
     _setup_old_repo(repo_dir)
-    mock_run_git_command.return_value = "abc123"
+    (repo_dir / ".gitmodules").write_text("test")
 
-    git.clone_or_update(
+    calls = {"submodule": 0}
+
+    def git_command_side_effect(
+        cmd: list[str], cwd: str | None = None, **kwargs: Any
+    ) -> str:
+        if _get_git_command_type(cmd) == "clone":
+            _simulate_cloned_repo(repo_dir)
+        if _get_git_command_type(cmd) == "submodule":
+            calls["submodule"] += 1
+            if calls["submodule"] == 1:
+                raise git.GitCommandError("git submodule update exited with code 1")
+        return ""
+
+    mock_run_git_command.side_effect = git_command_side_effect
+
+    recovered_dir, _ = git.clone_or_update(
         url=url,
         ref=None,
         refresh=TimePeriodSeconds(days=1),
         domain=domain,
-        submodules=["components/foo"],
+        username="user",
+        password="hunter2",
+        init_submodules=True,
     )
 
-    submodule_calls = [
-        c for c in mock_run_git_command.call_args_list if "submodule" in c[0][0]
+    assert recovered_dir == repo_dir
+    clone_cmds = [
+        c[0][0]
+        for c in mock_run_git_command.call_args_list
+        if _get_git_command_type(c[0][0]) == "clone"
     ]
-    assert len(submodule_calls) == 1
-    cmd = submodule_calls[0][0][0]
-    assert "--depth=1" in cmd
-    assert "components/foo" in cmd
-    assert cmd.index("--") < cmd.index("components/foo")
+    assert clone_cmds
+    clone_url = clone_cmds[0][-2]
+    assert clone_url == "https://user:hunter2@github.com/test/repo"
+    assert clone_url.count("@") == 1
+
+
+def test_refresh_submodule_failure_recovers_then_raises(
+    tmp_path: Path, mock_run_git_command: Mock
+) -> None:
+    """A refresh-path submodule failure routes through the recovery re-clone.
+
+    The broken repo is removed and re-cloned; when the submodule update fails
+    again on the fresh clone the cache entry is removed and the error
+    propagates, instead of leaving behind a repo the refresh window would
+    silently accept on the next run.
+    """
+    CORE.config_path = tmp_path / "test.yaml"
+
+    url = "https://github.com/test/repo"
+    domain = "test"
+    repo_dir = _compute_repo_dir(url, None, domain)
+
+    _setup_old_repo(repo_dir)
+    (repo_dir / ".gitmodules").write_text("test")
+
+    def git_command_side_effect(
+        cmd: list[str], cwd: str | None = None, **kwargs: Any
+    ) -> str:
+        if _get_git_command_type(cmd) == "clone":
+            _simulate_cloned_repo(repo_dir)
+            (repo_dir / ".gitmodules").write_text("test")
+        if _get_git_command_type(cmd) == "submodule":
+            raise git.GitCommandError("git submodule update exited with code 1")
+        return ""
+
+    mock_run_git_command.side_effect = git_command_side_effect
+
+    with pytest.raises(git.GitCommandError, match="exited with code 1"):
+        git.clone_or_update(
+            url=url,
+            ref=None,
+            refresh=TimePeriodSeconds(days=1),
+            domain=domain,
+            init_submodules=True,
+        )
+
+    assert not repo_dir.is_dir()
+    # Recovery removed the repo and re-cloned before failing again.
+    assert any(
+        _get_git_command_type(c[0][0]) == "clone"
+        for c in mock_run_git_command.call_args_list
+    )
+
+
+def _real_git(*args: str, cwd: Path) -> None:
+    """Run real git to build a test fixture repository."""
+    subprocess.run(
+        [
+            "git",
+            "-c",
+            "user.email=test@test.invalid",
+            "-c",
+            "user.name=test",
+            "-c",
+            "commit.gpgsign=false",
+            "-c",
+            "protocol.file.allow=always",
+            *args,
+        ],
+        cwd=cwd,
+        check=True,
+        capture_output=True,
+    )
+
+
+# Git blocks file-protocol submodules by default (CVE-2022-39253); the e2e
+# tests allow them via GIT_CONFIG_* environment variables, which reach the
+# child git processes through run_git_command's filtered environment.
+_ALLOW_FILE_PROTOCOL_ENV = {
+    "GIT_CONFIG_COUNT": "1",
+    "GIT_CONFIG_KEY_0": "protocol.file.allow",
+    "GIT_CONFIG_VALUE_0": "always",
+}
+
+
+def _make_real_repo(path: Path, filename: str) -> None:
+    """Create a real git repository containing one committed file."""
+    path.mkdir()
+    _real_git("init", "-q", cwd=path)
+    (path / filename).write_text("content")
+    _real_git("add", filename, cwd=path)
+    _real_git("commit", "-q", "-m", "init", cwd=path)
+
+
+def _add_submodule(
+    repo: Path, url: Path, path: str, *, update_none: bool = False
+) -> None:
+    """Add ``url`` as a submodule of ``repo`` at ``path`` and commit it."""
+    _real_git("submodule", "add", str(url), path, cwd=repo)
+    if update_none:
+        _real_git(
+            "config", "-f", ".gitmodules", f"submodule.{path}.update", "none", cwd=repo
+        )
+        _real_git("add", ".gitmodules", cwd=repo)
+    _real_git("commit", "-q", "-m", f"add submodule {path}", cwd=repo)
+
+
+def test_clone_or_update_real_git_without_submodules(tmp_path: Path) -> None:
+    """End-to-end with real git: a repo with no .gitmodules clones cleanly.
+
+    This is the issue #17860 scenario: requesting "all submodules" on a
+    submodule-less repository must not invoke the git submodule porcelain
+    and must produce a usable checkout.
+    """
+    CORE.config_path = tmp_path / "test.yaml"
+
+    upstream = tmp_path / "upstream"
+    _make_real_repo(upstream, "README.md")
+
+    repo_dir, _ = git.clone_or_update(
+        url=str(upstream),
+        ref=None,
+        refresh=None,
+        domain="test_e2e",
+        init_submodules=True,
+    )
+
+    assert (repo_dir / "README.md").is_file()
+
+
+def test_clone_or_update_real_git_initializes_submodules(tmp_path: Path) -> None:
+    """End-to-end with real git: submodules are actually checked out.
+
+    Exercises the real `git submodule update` invocation, including the
+    env handling in run_git_command that the mocked tests cannot cover.
+    """
+    CORE.config_path = tmp_path / "test.yaml"
+
+    sub_repo = tmp_path / "sub"
+    _make_real_repo(sub_repo, "sub_file.txt")
+
+    upstream = tmp_path / "upstream"
+    _make_real_repo(upstream, "README.md")
+    _add_submodule(upstream, sub_repo, "vendor/sub")
+
+    with patch.dict(os.environ, _ALLOW_FILE_PROTOCOL_ENV):
+        repo_dir, _ = git.clone_or_update(
+            url=str(upstream),
+            ref=None,
+            refresh=None,
+            domain="test_e2e",
+            init_submodules=True,
+        )
+
+    assert (repo_dir / "vendor" / "sub" / "sub_file.txt").is_file()
+
+
+def test_clone_or_update_real_git_honors_update_none_submodule(
+    tmp_path: Path,
+) -> None:
+    """End-to-end with real git: submodules declared `update = none` stay skipped.
+
+    Shows git itself skipping the declared paths at both nesting levels
+    (and exiting 0) while the regular submodules check out.
+    """
+    CORE.config_path = tmp_path / "test.yaml"
+
+    sub_repo = tmp_path / "sub"
+    _make_real_repo(sub_repo, "sub_file.txt")
+
+    # Intermediate submodule that itself declares a skipped nested submodule.
+    mid_repo = tmp_path / "mid"
+    _make_real_repo(mid_repo, "mid_file.txt")
+    _add_submodule(mid_repo, sub_repo, "vendor/leaf", update_none=True)
+
+    upstream = tmp_path / "upstream"
+    _make_real_repo(upstream, "README.md")
+    _add_submodule(upstream, sub_repo, "vendor/sub")
+    _add_submodule(upstream, sub_repo, "vendor/skipped", update_none=True)
+    _add_submodule(upstream, mid_repo, "vendor/mid")
+
+    with patch.dict(os.environ, _ALLOW_FILE_PROTOCOL_ENV):
+        repo_dir, _ = git.clone_or_update(
+            url=str(upstream),
+            ref=None,
+            refresh=None,
+            domain="test_e2e",
+            init_submodules=True,
+        )
+
+    assert (repo_dir / "vendor" / "sub" / "sub_file.txt").is_file()
+    assert not (repo_dir / "vendor" / "skipped" / "sub_file.txt").exists()
+    assert (repo_dir / "vendor" / "mid" / "mid_file.txt").is_file()
+    assert not (
+        repo_dir / "vendor" / "mid" / "vendor" / "leaf" / "sub_file.txt"
+    ).exists()
 
 
 def test_refresh_picks_up_new_remote_commits(

tests/unit_tests/test_lazy_imports.py

@@ -0,0 +1,53 @@
+"""Guard the lazy-import contract of ``esphome.__main__``.
+
+Every ``esphome`` invocation pays for whatever ``esphome.__main__``
+imports at module level before the requested command runs. The
+dashboard and device-builder spawn one ``esphome upload`` subprocess
+per device, so keeping validation/codegen machinery out of the
+top-level import directly lowers the RAM cost of each concurrent
+upload (the upload/logs fast path in ``esphome.compiled_config``
+never needs them).
+
+``script/check_import_time.py`` budgets import *time* in CI; this
+test pins down *which* heavy modules must stay out entirely.
+"""
+
+from __future__ import annotations
+
+import subprocess
+import sys
+
+# Modules that must only load for the commands that actually use them
+# (compile/config validation, shell completion), never from a bare
+# ``import esphome.__main__``.
+HEAVY_MODULES = (
+    "argcomplete",
+    "esphome.codegen",
+    "esphome.config",
+    "esphome.config_validation",
+    "esphome.cpp_generator",
+    "esphome.loader",
+    "voluptuous",
+)
+
+
+def test_main_module_does_not_import_heavy_modules() -> None:
+    """A bare ``import esphome.__main__`` must not drag in validation/codegen."""
+    check = (
+        "import sys; import esphome.__main__; "
+        f"leaked = [m for m in {HEAVY_MODULES!r} if m in sys.modules]; "
+        "print(','.join(leaked))"
+    )
+    result = subprocess.run(
+        [sys.executable, "-c", check],
+        capture_output=True,
+        text=True,
+        check=True,
+    )
+    leaked = result.stdout.strip()
+    assert not leaked, (
+        f"esphome.__main__ imports heavy modules at top level: {leaked}. "
+        "Import them lazily inside the command that needs them instead; "
+        "every esphome invocation (including each parallel dashboard "
+        "upload subprocess) pays for top-level imports."
+    )

tests/unit_tests/test_main.py

@@ -442,6 +442,46 @@ def test_redact_with_legacy_fallback__does_not_match_fragment_as_suffix(
     assert not any("legacy substring" in rec.message for rec in caplog.records)
 
 
+@pytest.mark.parametrize("field", ["public_key", "peer_public_key"])
+def test_redact_with_legacy_fallback__skips_public_key_fields(
+    field: str,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Public keys are not secret; fields with a ``public`` name segment
+    must pass through unredacted and without the migration warning
+    (see issue #17718)."""
+    text = f"{field}: c29tZXB1YmxpY2tleQ==\n"
+    with caplog.at_level(logging.WARNING, logger="esphome.__main__"):
+        out = _redact_with_legacy_fallback(text)
+    assert out == text
+    assert not any("legacy substring" in rec.message for rec in caplog.records)
+
+
+def test_redact_with_legacy_fallback__public_substitution_still_redacted(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Substitution keys are user-named with no schema behind them, so the
+    public-key exemption does not apply there; a ``public``-named substitution
+    keeps the conservative silent redaction."""
+    text = "substitutions:\n  public_key: something\nesphome:\n  name: x\n"
+    with caplog.at_level(logging.WARNING, logger="esphome.__main__"):
+        out = _redact_with_legacy_fallback(text)
+    assert "public_key: \\033[8msomething\\033[28m" in out
+    assert not any("legacy substring" in rec.message for rec in caplog.records)
+
+
+def test_redact_with_legacy_fallback__public_must_be_a_whole_segment(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """The exemption matches ``public`` as an underscore-separated segment,
+    not a substring; an unrelated name like ``republic_key`` keeps the
+    conservative redaction."""
+    with caplog.at_level(logging.WARNING, logger="esphome.__main__"):
+        out = _redact_with_legacy_fallback("republic_key: abc\n")
+    assert "republic_key: \\033[8mabc\\033[28m" in out
+    assert any("'republic_key'" in rec.message for rec in caplog.records)
+
+
 def test_redact_with_legacy_fallback__substitutions_redacted_without_warning(
     caplog: pytest.LogCaptureFixture,
 ) -> None:
@@ -578,7 +618,7 @@ def test_command_config__no_defaults_skips_strip_default_ids(
     validated.user_config = {"sensor": [{"name": "x"}]}
 
     with patch(
-        "esphome.__main__.strip_default_ids", side_effect=AssertionError
+        "esphome.config.strip_default_ids", side_effect=AssertionError
     ) as mock_strip:
         result = command_config(args, validated)
 
@@ -5410,6 +5450,43 @@ def _setup_build_info_test(
     return build_info_path, firmware_path
 
 
+def test_compile_program_esp8266_runs_rosetta_check(tmp_path: Path) -> None:
+    """Test that compile_program runs the Rosetta preflight for ESP8266 targets."""
+    setup_core(platform=PLATFORM_ESP8266, tmp_path=tmp_path, name="test_device")
+
+    config: dict[str, Any] = {CONF_ESPHOME: {CONF_NAME: "test_device"}}
+    args = MockArgs()
+
+    with (
+        patch(
+            "esphome.components.esp8266.check_rosetta",
+            side_effect=EsphomeError("Rosetta 2 is not installed"),
+        ) as mock_check,
+        pytest.raises(EsphomeError, match="Rosetta 2 is not installed"),
+    ):
+        compile_program(args, config)
+
+    mock_check.assert_called_once()
+
+
+def test_compile_program_skips_rosetta_check_on_other_platforms(
+    tmp_path: Path,
+    mock_compile_build_info_run_compile: Mock,
+    mock_compile_build_info_get_idedata: Mock,
+) -> None:
+    """Test that the Rosetta preflight does not run for non-ESP8266 targets."""
+    _setup_build_info_test(tmp_path, firmware_first=True)
+
+    config: dict[str, Any] = {CONF_ESPHOME: {CONF_NAME: "test_device"}}
+    args = MockArgs()
+
+    with patch("esphome.components.esp8266.check_rosetta") as mock_check:
+        result = compile_program(args, config)
+
+    assert result == 0
+    mock_check.assert_not_called()
+
+
 def test_compile_program_emits_build_info_when_firmware_rebuilt(
     tmp_path: Path,
     caplog: pytest.LogCaptureFixture,
@@ -6124,7 +6201,7 @@ def test_run_esphome_bundle_detection(tmp_path: Path) -> None:
             "esphome.bundle.prepare_bundle_for_compile",
             return_value=extracted_yaml,
         ) as mock_prepare,
-        patch("esphome.__main__.read_config", return_value=None),
+        patch("esphome.config.read_config", return_value=None),
     ):
         result = run_esphome(["esphome", "compile", str(bundle_path)])
 
@@ -6142,7 +6219,7 @@ def test_run_esphome_non_bundle_skips_extraction(tmp_path: Path) -> None:
     with (
         patch("esphome.bundle.is_bundle_path", return_value=False) as mock_is_bundle,
         patch("esphome.bundle.prepare_bundle_for_compile") as mock_prepare,
-        patch("esphome.__main__.read_config", return_value=None),
+        patch("esphome.config.read_config", return_value=None),
     ):
         result = run_esphome(["esphome", "compile", str(yaml_file)])
 
@@ -6170,7 +6247,7 @@ def test_run_esphome_skip_external_update_per_command(
     yaml_file = tmp_path / "device.yaml"
     yaml_file.write_text("esphome:\n  name: test\n")
 
-    with patch("esphome.__main__.read_config", return_value=None) as mock_read:
+    with patch("esphome.config.read_config", return_value=None) as mock_read:
         run_esphome(["esphome", command, str(yaml_file)])
 
     mock_read.assert_called_once()
@@ -6328,6 +6405,23 @@ def test_parse_args_logs_states() -> None:
     assert args.states is True
 
 
+def test_parse_args_argcomplete_only_runs_when_completing() -> None:
+    """Only import and invoke argcomplete when _ARGCOMPLETE is set.
+
+    The shell-completion machinery sets _ARGCOMPLETE when it invokes the
+    CLI; a normal invocation must skip the import entirely so every
+    esphome subprocess (e.g. parallel dashboard uploads) avoids paying
+    for it.
+    """
+    fake_argcomplete = MagicMock()
+    with (
+        patch.dict(os.environ, {"_ARGCOMPLETE": "1"}),
+        patch.dict(sys.modules, {"argcomplete": fake_argcomplete}),
+    ):
+        parse_args(["esphome", "version"])
+    fake_argcomplete.autocomplete.assert_called_once()
+
+
 def test_should_subscribe_states_default() -> None:
     """Test that states are shown by default when nothing is set."""
     from esphome.__main__ import _should_subscribe_states

tests/unit_tests/test_nrf52_framework.py

@@ -201,6 +201,24 @@ class TestCheckAndInstall:
         assert mock_nrf52_ops.download_from_mirrors.call_count == 2
         assert mock_nrf52_ops.archive_extract_all.call_count == 2
 
+    def test_framework_clone_is_shallow(
+        self,
+        nrf52_dirs: SimpleNamespace,
+        mock_nrf52_ops: SimpleNamespace,
+    ) -> None:
+        """Both the manifest repository and every project are fetched at depth 1."""
+        _mark_venv_ready(nrf52_dirs.python_env)
+
+        check_and_install()
+
+        init_cmd, update_cmd = (
+            call.args[0] for call in mock_nrf52_ops.run_command_ok.call_args_list[:2]
+        )
+        assert "init" in init_cmd
+        assert "-o=--depth=1" in init_cmd
+        assert "update" in update_cmd
+        assert "--fetch-opt=--depth=1" in update_cmd
+
     def test_requirements_install_failure_raises(
         self,
         nrf52_dirs: SimpleNamespace,

tests/unit_tests/test_platformio_library.py

@@ -261,15 +261,18 @@ def test_convert_libraries_raises_when_manifest_missing_after_retry(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:
     # If the forced re-download still yields no manifest, the error is raised
-    # after exactly one retry (no retry loop).
+    # after exactly one retry (no retry loop). The error must name the cache
+    # directory so users can find the broken entry instead of guessing where
+    # the library was unpacked.
     calls = _patch_download_without_manifest(
         monkeypatch, tmp_path, manifest_on_force=False
     )
 
-    with pytest.raises(RuntimeError, match="Invalid PIO library"):
+    with pytest.raises(RuntimeError, match="Invalid PIO library") as excinfo:
         convert_libraries([Library("esphome/A", "1.0.0", None)], _backend())
 
     assert calls == [False, True]
+    assert str(tmp_path / "esphome__A") in str(excinfo.value)
 
 
 @pytest.mark.parametrize(

tests/unit_tests/test_platformio_toolchain.py

@@ -322,6 +322,149 @@ def test_run_platformio_cli_sets_environment_variables(
         assert "arg" in args
 
 
+def test_ccache_env_enabled_by_default(setup_core: Path) -> None:
+    """Ccache is enabled when the binary is on PATH and no override is set."""
+    CORE.build_path = setup_core / "build" / "test"
+
+    with (
+        patch.dict(os.environ, {}, clear=True),
+        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
+    ):
+        env = toolchain._ccache_env()
+
+    assert env["ESPHOME_CCACHE_ENABLE"] == "1"
+    assert env["CCACHE_BASEDIR"] == str((setup_core / "build" / "test").resolve())
+    assert env["CCACHE_DIR"].endswith("platformio-ccache")
+    assert env["CCACHE_NOHASHDIR"] == "true"
+    # Nothing may leak into os.environ: a later ESP-IDF build in the same
+    # process would otherwise skip its own ccache defaults.
+    assert "CCACHE_BASEDIR" not in os.environ
+    assert "ESPHOME_CCACHE_ENABLE" not in os.environ
+
+
+def test_ccache_env_disabled_without_binary(setup_core: Path) -> None:
+    """Ccache stays off when the binary is not on PATH."""
+    CORE.build_path = setup_core / "build" / "test"
+
+    with (
+        patch.dict(os.environ, {}, clear=True),
+        patch.object(toolchain.shutil, "which", return_value=None),
+    ):
+        env = toolchain._ccache_env()
+
+    assert env == {"ESPHOME_CCACHE_ENABLE": "0"}
+
+
+def test_ccache_env_opt_out(setup_core: Path) -> None:
+    """ESPHOME_CCACHE_ENABLE=0 disables ccache even with the binary present."""
+    CORE.build_path = setup_core / "build" / "test"
+
+    with (
+        patch.dict(os.environ, {"ESPHOME_CCACHE_ENABLE": "0"}, clear=True),
+        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
+    ):
+        env = toolchain._ccache_env()
+
+    assert env == {"ESPHOME_CCACHE_ENABLE": "0"}
+
+
+def test_ccache_env_normalizes_enable_value(setup_core: Path) -> None:
+    """A truthy override value is normalized to "1" for the build scripts."""
+    CORE.build_path = setup_core / "build" / "test"
+
+    with (
+        patch.dict(os.environ, {"ESPHOME_CCACHE_ENABLE": "yes"}, clear=True),
+        patch.object(toolchain.shutil, "which", return_value=None),
+    ):
+        env = toolchain._ccache_env()
+
+    assert env["ESPHOME_CCACHE_ENABLE"] == "1"
+
+
+def test_ccache_env_respects_user_values_and_refreshes_basedir(
+    setup_core: Path,
+) -> None:
+    """User CCACHE_* values win, but CCACHE_BASEDIR follows the build dir."""
+    user_env = {
+        "CCACHE_DIR": "/custom/cache",
+        "CCACHE_BASEDIR": "/stale/other-device",
+    }
+    CORE.build_path = setup_core / "build" / "test"
+
+    with (
+        patch.dict(os.environ, user_env, clear=True),
+        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
+    ):
+        env = toolchain._ccache_env()
+
+    # CCACHE_DIR is not returned, so the user's os.environ value applies in
+    # the subprocess; CCACHE_BASEDIR is always refreshed to the build dir.
+    assert "CCACHE_DIR" not in env
+    assert env["CCACHE_BASEDIR"] == str((setup_core / "build" / "test").resolve())
+
+
+def test_run_platformio_cli_passes_ccache_env_to_subprocess_only(
+    setup_core: Path, mock_run_external_process: Mock
+) -> None:
+    """The ccache settings reach the subprocess env without touching os.environ."""
+    CORE.build_path = str(setup_core / "build" / "test")
+
+    with (
+        patch.dict(os.environ, {}, clear=False),
+        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
+    ):
+        os.environ.pop("ESPHOME_CCACHE_ENABLE", None)
+        mock_run_external_process.return_value = 0
+        toolchain.run_platformio_cli("test", "arg")
+
+        env = mock_run_external_process.call_args[1]["env"]
+        assert env["ESPHOME_CCACHE_ENABLE"] == "1"
+        assert env["CCACHE_BASEDIR"] == str((setup_core / "build" / "test").resolve())
+        assert "ESPHOME_CCACHE_ENABLE" not in os.environ
+        assert "CCACHE_BASEDIR" not in os.environ
+
+
+def test_ccache_env_requires_build_path(setup_core: Path) -> None:
+    """Enabling ccache without a build path fails loudly."""
+    CORE.build_path = None
+
+    with (
+        patch.dict(os.environ, {}, clear=True),
+        patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"),
+        pytest.raises(ValueError, match="CORE.build_path must be set"),
+    ):
+        toolchain._ccache_env()
+
+
+def test_run_platformio_cli_merges_caller_env(
+    setup_core: Path, mock_run_external_process: Mock
+) -> None:
+    """A caller-supplied env is the base and gains the ccache settings."""
+    CORE.build_path = str(setup_core / "build" / "test")
+
+    with patch.object(toolchain.shutil, "which", return_value="/usr/bin/ccache"):
+        mock_run_external_process.return_value = 0
+        toolchain.run_platformio_cli(
+            "test", env={"CUSTOM_VAR": "1", "ESPHOME_CCACHE_ENABLE": "0"}
+        )
+
+    env = mock_run_external_process.call_args[1]["env"]
+    assert env["CUSTOM_VAR"] == "1"
+    # The normalized enable flag still lands in the subprocess env.
+    assert "ESPHOME_CCACHE_ENABLE" in env
+
+
+def test_copy_ccache_script(setup_core: Path) -> None:
+    """The shared ccache pre-script is copied into the build dir."""
+    CORE.build_path = setup_core / "build" / "test"
+
+    toolchain.copy_ccache_script()
+
+    dest = setup_core / "build" / "test" / "ccache.py"
+    source = Path(toolchain.__file__).parent / "ccache.py.script"
+    assert dest.read_text() == source.read_text()
+
+
 @pytest.mark.parametrize(
     ("platform", "input_path", "expected"),
     [
@@ -375,7 +518,10 @@ def test_run_platformio_cli_strips_win_long_path_prefix(
     )
 
     with (
-        patch.dict(os.environ, {}, clear=False),
+        # Pin ccache off: patching sys.platform to win32 (sys is a singleton,
+        # 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.platformio.toolchain.sys.executable", prefixed_exe),
     ):