[espidf] Mirror registry components locally so configures skip the registry (#19949)

esphome/espidf/component_mirror.py

@@ -0,0 +1,431 @@
+"""Local mirror of IDF component-registry packages.
+
+The component manager contacts the registry on every cmake configure, even
+with an unchanged ``dependencies.lock``, but checks
+``IDF_COMPONENT_LOCAL_STORAGE_URL`` mirrors first and stops on a hit.
+Mirroring the pinned components with its own ``registry sync`` removes all
+registry traffic and makes builds work offline.
+"""
+
+from __future__ import annotations
+
+from contextlib import suppress
+from dataclasses import dataclass, field
+import json
+import logging
+import os
+from pathlib import Path
+import subprocess
+from typing import TYPE_CHECKING, NamedTuple
+import zipfile
+
+from esphome.build_helpers.tools_cache import IDF_TOOLS_CACHE, tools_cache_path
+from esphome.core import EsphomeError
+from esphome.framework_helpers import rename_with_retry
+from esphome.helpers import rmtree, write_file
+
+if TYPE_CHECKING:
+    from collections.abc import Callable, Iterator, Mapping
+
+_LOGGER = logging.getLogger(__name__)
+
+# Bump the name if the manager's storage layout ever changes incompatibly;
+# the stale directory is removed with the tools cache on clean-all.
+_MIRROR_DIR_NAME = "component_mirror"
+
+_ENV_LOCAL_STORAGE_URL = "IDF_COMPONENT_LOCAL_STORAGE_URL"
+_ENV_CHECK_NEW_VERSION = "IDF_COMPONENT_CHECK_NEW_VERSION"
+
+_DEFAULT_REGISTRY_URL = "https://components.espressif.com"
+# Everything a missing, unreadable or wrongly-shaped index file can raise.
+_BAD_INDEX_ERRORS = (OSError, ValueError, TypeError, KeyError, AttributeError)
+_SYNC_LOCK_NAME = ".sync.lock"
+_STAGING_DIR_NAME = ".staging"
+_SYNC_TIMEOUT_S = 120
+
+
+class ServiceDep(NamedTuple):
+    """A registry ("service") dependency pinned to an exact version."""
+
+    namespace: str
+    name: str
+    version: str
+
+    @property
+    def spec(self) -> str:
+        return f"{self.namespace}/{self.name}=={self.version}"
+
+
+def get_mirror_path() -> Path:
+    """The machine-global mirror directory."""
+    return tools_cache_path(*IDF_TOOLS_CACHE) / _MIRROR_DIR_NAME
+
+
+def component_mirror_env() -> dict[str, str]:
+    """Environment additions that serve the mirror to the component manager.
+
+    A user local storage list keeps precedence; the new-version check (an
+    extra solve whose answer exact pins make moot) is off unless user-set.
+    """
+    mirror = get_mirror_path()
+    try:
+        mirror.mkdir(parents=True, exist_ok=True)
+    except OSError as err:
+        # Offline support is quietly lost otherwise; say so.
+        _LOGGER.warning("Component mirror unavailable at %s: %s", mirror, err)
+        return {}
+    _LOGGER.info("Serving IDF components from the local mirror at %s", mirror)
+    local_storage = mirror.as_uri()
+    if user_value := os.environ.get(_ENV_LOCAL_STORAGE_URL):
+        local_storage = f"{user_value};{local_storage}"
+    env = {_ENV_LOCAL_STORAGE_URL: local_storage}
+    if _ENV_CHECK_NEW_VERSION not in os.environ:
+        env[_ENV_CHECK_NEW_VERSION] = "0"
+    return env
+
+
+def _load_yaml_dict(path: Path) -> dict | None:
+    """Read a small YAML mapping; missing file or bad content is None."""
+    # Deferred: keeps pyyaml off the serial upload fast path.
+    import yaml
+
+    try:
+        text = path.read_text(encoding="utf-8")
+    except FileNotFoundError:
+        return None
+    except (OSError, UnicodeDecodeError) as err:
+        _LOGGER.warning("Could not read %s: %s", path, err)
+        return None
+    try:
+        data = yaml.safe_load(text)
+    except yaml.YAMLError as err:
+        _LOGGER.warning("Could not parse %s: %s", path, err)
+        return None
+    return data if isinstance(data, dict) else None
+
+
+def _iter_deps(path: Path) -> Iterator[tuple[str, str, dict]]:
+    """(namespace, name, entry) for each namespaced dependency in a YAML file."""
+    data = _load_yaml_dict(path)
+    deps = data.get("dependencies") if data else None
+    if not isinstance(deps, dict):
+        return
+    for key, entry in deps.items():
+        if isinstance(key, str) and "/" in key and isinstance(entry, dict):
+            # The registry stores lowercase paths; match a mixed-case key.
+            namespace, _, name = key.lower().partition("/")
+            yield namespace, name, entry
+
+
+def parse_lock_service_deps(lock_path: Path) -> list[ServiceDep]:
+    """Pinned default-registry dependencies from a dependencies.lock.
+
+    Git, local and idf sources cannot be mirrored; non-default registries
+    are left to the manager.
+    """
+    deps: list[ServiceDep] = []
+    for namespace, name, entry in _iter_deps(lock_path):
+        source = entry.get("source")
+        if not isinstance(source, dict) or source.get("type") != "service":
+            continue
+        registry_url = str(source.get("registry_url") or _DEFAULT_REGISTRY_URL)
+        if registry_url.rstrip("/") != _DEFAULT_REGISTRY_URL:
+            _LOGGER.debug("Not mirroring %s/%s: non-default registry", namespace, name)
+            continue
+        if isinstance(version := entry.get("version"), str):
+            deps.append(ServiceDep(namespace, name, version))
+    return deps
+
+
+def parse_manifest_service_deps(manifest_path: Path) -> list[ServiceDep]:
+    """Exactly-pinned registry dependencies from an idf_component.yml.
+
+    The manifest exists before the first configure, so a fresh solve can
+    install from the mirror; range specs are left to the solver.
+    """
+    deps: list[ServiceDep] = []
+    for namespace, name, entry in _iter_deps(manifest_path):
+        if entry.keys() != {"version"} or not isinstance(
+            version := entry["version"], str
+        ):
+            continue
+        # The YAML shorthand keeps its operator: "==1.2.3" is an exact pin.
+        version = version.removeprefix("==")
+        # Exact versions start with a digit ("1.12.0", "1.3.3~1"); range
+        # operators are prefixes and wildcards contain "*".
+        if version[:1].isdigit() and "*" not in version:
+            deps.append(ServiceDep(namespace, name, version))
+    return deps
+
+
+def parse_manifest_ranged_specs(manifest_path: Path) -> list[str]:
+    """Compote specs for the manifest's range-pinned entries.
+
+    Refreshed before every fresh solve, as a range resolves on dev; the
+    previous candidate stays the offline fallback.
+    """
+    specs: list[str] = []
+    for namespace, name, entry in _iter_deps(manifest_path):
+        if entry.keys() != {"version"} or not isinstance(
+            version := entry["version"], str
+        ):
+            continue
+        # [:1] alone would match "": the empty string is in every string.
+        if version.startswith(("^", "~", "<", ">", "!")) or (
+            version.startswith("=") and not version.startswith("==")
+        ):
+            specs.append(f"{namespace}/{name}{version}")
+    return specs
+
+
+def project_service_deps(lock_path: Path, manifest_path: Path) -> list[ServiceDep]:
+    """The mirrorable dependencies of a build; the lock wins, the manifest
+    covers the build whose lock has not been written yet."""
+    deps = parse_lock_service_deps(lock_path)
+    seen = {(dep.namespace, dep.name) for dep in deps}
+    for dep in parse_manifest_service_deps(manifest_path):
+        if (dep.namespace, dep.name) not in seen:
+            deps.append(dep)
+    return deps
+
+
+def _mirror_has(mirror: Path, dep: ServiceDep) -> bool:
+    """Whether the mirror holds the dependency's index entry and files.
+
+    The manager fetches every file the entry names with no registry
+    fallback once the version is found locally.
+    """
+    json_path = mirror / "components" / dep.namespace / f"{dep.name}.json"
+    try:
+        for entry in json.loads(json_path.read_text(encoding="utf-8"))["versions"]:
+            if entry.get("version") == dep.version:
+                checksums = entry.get("checksums")
+                # is_zipfile rejects an archive torn by a crash or an
+                # old race, so the next sync replaces it (self-heal).
+                archive = mirror / entry["url"]
+                if (
+                    archive.suffix == ".zip"
+                    and archive.is_file()
+                    and not (zipfile.is_zipfile(archive))
+                ):
+                    # Deleted, the manager falls back to the registry even
+                    # when the heal sync cannot run; kept, it installs the
+                    # torn file and the configure fails.
+                    with suppress(OSError):
+                        archive.unlink()
+                    return False
+                return archive.is_file() and (
+                    not checksums or (mirror / checksums).is_file()
+                )
+    except _BAD_INDEX_ERRORS:
+        return False
+    return False
+
+
+def missing_deps(mirror: Path, deps: list[ServiceDep]) -> list[ServiceDep]:
+    return [dep for dep in deps if not _mirror_has(mirror, dep)]
+
+
+def _read_versions(path: Path) -> list[dict]:
+    """The version entries of an index; [] when missing or corrupt, so
+    publishing heals it. An I/O error raises: the content may be valid,
+    so the caller retries later instead of dropping its versions."""
+    try:
+        entries = json.loads(path.read_text(encoding="utf-8"))["versions"]
+        return [entry for entry in entries if isinstance(entry, dict)]
+    except FileNotFoundError:
+        return []
+    except (ValueError, TypeError, KeyError) as err:
+        _LOGGER.debug("Ignoring the corrupt index %s: %s", path, err)
+        return []
+
+
+def _publish_index(src: Path, dst: Path) -> None:
+    """Publish the staged index merged with the live one, atomically.
+
+    The staged index lists only the versions this sync fetched; the
+    atomic write keeps the live index intact when it fails, and raises.
+    """
+    staged = json.loads(src.read_text(encoding="utf-8"))
+    versions = [entry for entry in staged["versions"] if isinstance(entry, dict)]
+    known = {entry.get("version") for entry in versions}
+    staged["versions"] = versions + [
+        entry for entry in _read_versions(dst) if entry.get("version") not in known
+    ]
+    write_file(dst, json.dumps(staged))
+
+
+def _is_component_index(rel: Path) -> bool:
+    return (
+        rel.parts[0] == "components" and len(rel.parts) == 3 and rel.suffix == ".json"
+    )
+
+
+def _promote(staging: Path, mirror: Path) -> None:
+    """Move the synced files into the mirror, atomically per file.
+
+    Indexes go last, so a concurrent configure never reads a version
+    entry whose files have not landed yet.
+    """
+    files = sorted(
+        (path for path in staging.rglob("*") if path.is_file()),
+        key=lambda path: (_is_component_index(path.relative_to(staging)), path),
+    )
+    for src in files:
+        rel = src.relative_to(staging)
+        dst = mirror / rel
+        dst.parent.mkdir(parents=True, exist_ok=True)
+        if _is_component_index(rel):
+            _publish_index(src, dst)
+        else:
+            rename_with_retry(src, dst, overwrite=True)
+
+
+@dataclass
+class _SyncRun:
+    """Tools and progress of one registry sync."""
+
+    python: str
+    env: dict[str, str]
+    staging: Path
+    mirror: Path
+    synced: int = field(default=0, init=False)
+
+    def _attempt(self, specs: list[str]) -> bool:
+        """One registry sync invocation; promotes its staging on success."""
+        # Raises on an undeletable tree; never promote stale files.
+        rmtree(self.staging)
+        cmd = [self.python, "-m", "idf_component_manager", "registry", "sync"]
+        # latest: the default "all" follows ranged transitive specs and
+        # syncs every matching version.
+        cmd += ["--resolution", "latest"]
+        for spec in specs:
+            cmd += ["--component", spec]
+        cmd.append(str(self.staging))
+        result = subprocess.run(
+            cmd,
+            env=self.env,
+            capture_output=True,
+            # Not text=True: the locale codec can raise UnicodeDecodeError.
+            encoding="utf-8",
+            errors="replace",
+            timeout=_SYNC_TIMEOUT_S,
+            check=False,
+        )
+        if result.returncode != 0:
+            tail = "\n".join((result.stderr or result.stdout).strip().splitlines()[-5:])
+            _LOGGER.warning(
+                "Could not mirror %s (exit %d):\n%s",
+                ", ".join(specs),
+                result.returncode,
+                tail,
+            )
+            return False
+        _promote(self.staging, self.mirror)
+        return True
+
+    def sync(self, specs: list[str]) -> bool:
+        """The whole set in one invocation, per component after a timeout."""
+        try:
+            # One invocation for the whole set: one manager startup, which
+            # matters on slow hosts.
+            if not self._attempt(specs):
+                return False
+            self.synced = len(specs)
+            return True
+        except subprocess.TimeoutExpired:
+            if len(specs) == 1:
+                raise
+            _LOGGER.warning(
+                "Mirroring timed out after %d s; retrying one component at a time",
+                _SYNC_TIMEOUT_S,
+            )
+        # Per component the timeout bounds each one, finished ones stay
+        # promoted, and the first failure stops the loop.
+        for spec in specs:
+            if not self._attempt([spec]):
+                return False
+            self.synced += 1
+        return True
+
+
+def sync_component_mirror(
+    lock_path: Path,
+    manifest_path: Path,
+    get_python: Callable[[], str],
+    get_env: Callable[[], Mapping[str, str]],
+) -> bool:
+    """Mirror any of the build's registry components the mirror lacks.
+
+    Returns False on a failed attempt so callers can skip retrying this
+    run; ``get_python``/``get_env`` are only called when a sync is needed.
+    """
+    mirror = get_mirror_path()
+    to_sync = missing_deps(mirror, project_service_deps(lock_path, manifest_path))
+    # Before a fresh solve, ranged pins re-resolve as they would on dev.
+    refresh = [] if lock_path.exists() else parse_manifest_ranged_specs(manifest_path)
+    if not to_sync and not refresh:
+        return True
+    try:
+        python = get_python()
+        env = dict(get_env())
+        mirror.mkdir(parents=True, exist_ok=True)
+    except (OSError, EsphomeError) as err:
+        _LOGGER.warning("Could not mirror IDF components: %s", err)
+        return False
+    # Staged: the manager writes in place, and a configure in another
+    # process must never read a truncated file from the live mirror.
+    staging = mirror / _STAGING_DIR_NAME
+    # Lazy import, as in git.py: keeps filelock off the CLI startup path.
+    from filelock import FileLock, Timeout
+
+    lock = FileLock(str(mirror / _SYNC_LOCK_NAME), fallback_to_soft=False)
+    try:
+        lock.acquire(blocking=False)
+    except Timeout:
+        # Another esphome process is already filling the shared mirror.
+        _LOGGER.debug("Component mirror sync skipped: already in progress")
+        return True
+    except OSError as err:
+        # A broken cache, not contention; the retry guard should apply.
+        _LOGGER.warning("Could not lock the component mirror: %s", err)
+        return False
+    specs = [dep.spec for dep in to_sync] + refresh
+    _LOGGER.info("Mirroring %d IDF component(s) for offline builds...", len(specs))
+    run = _SyncRun(python, env, staging, mirror)
+    try:
+        if not run.sync(specs):
+            return False
+        if still := missing_deps(mirror, to_sync):
+            # A name the registry spells differently syncs clean yet
+            # covers nothing; retrying would loop forever.
+            _LOGGER.warning(
+                "Mirror sync left %d component(s) uncovered: %s",
+                len(still),
+                ", ".join(dep.spec for dep in still),
+            )
+            return False
+    except subprocess.TimeoutExpired:
+        _LOGGER.warning(
+            "Mirroring timed out after %d s; kept %d of %d component(s)",
+            _SYNC_TIMEOUT_S,
+            run.synced,
+            len(specs),
+        )
+        return False
+    except (*_BAD_INDEX_ERRORS, EsphomeError, subprocess.SubprocessError) as err:
+        # Includes a failed index publish; the live index is intact.
+        # A shape error is a bug: keep its traceback.
+        _LOGGER.warning(
+            "Could not mirror IDF components: %s",
+            err,
+            exc_info=isinstance(err, (TypeError, KeyError, AttributeError)),
+        )
+        return False
+    finally:
+        # Cleanup only; a leftover tree is removed by the next attempt.
+        with suppress(OSError):
+            rmtree(staging)
+        lock.release()
+    _LOGGER.info("Mirrored %d IDF component(s) for offline builds", run.synced)
+    return True

esphome/espidf/toolchain.py

@@ -24,6 +24,7 @@ from esphome.const import (
 )
 from esphome.core import CORE, EsphomeError
 from esphome.espidf import variant_to_idf_target
+from esphome.espidf.component_mirror import component_mirror_env, sync_component_mirror
 from esphome.espidf.framework import check_esp_idf_install, get_framework_env
 from esphome.espidf.size_summary import print_summary
 from esphome.helpers import add_git_ceiling_directory, get_bool_env, rmtree, write_file
@@ -43,6 +44,7 @@ class _CacheData:
     cmake_output: dict[Path, str] = field(default_factory=dict)
     skip_bootloader: bool | None = None
     cmake_tools: dict[Path, dict[str, Path]] = field(default_factory=dict)
+    mirror_sync_failed: bool = False
 
 
 def _cache() -> _CacheData:
@@ -110,6 +112,11 @@ def _get_idf_path(version: str | None = None) -> Path | None:
     return Path(_get_esphome_esp_idf_paths(version)[0])
 
 
+def _esphome_manages_idf() -> bool:
+    """A checkout supplied through IDF_PATH is the user's, not ESPHome's."""
+    return "IDF_PATH" not in os.environ
+
+
 def _get_idf_env(version: str | None = None) -> dict[str, str]:
     """Get environment variables needed for ESP-IDF build."""
     version = version or _get_core_framework_version()
@@ -120,10 +127,12 @@ def _get_idf_env(version: str | None = None) -> dict[str, str]:
         env_cache[version].pop("PYTHONPATH", None)
 
         # Use provided IDF framework if available
-        if "IDF_PATH" not in os.environ:
+        if _esphome_manages_idf():
             env_cache[version] |= get_framework_env(
                 *_get_esphome_esp_idf_paths(version)
             )
+            # Serve the component manager from the local registry mirror.
+            env_cache[version] |= component_mirror_env()
 
         # Cap git's repo search at the config directory so ESP-IDF's
         # `git describe` for the app version can't error out on an
@@ -369,6 +378,8 @@ def run_reconfigure(verbose: bool = False) -> int:
             if (path := build_dir / stale).is_dir():
                 rmtree(path)
         get_factory_firmware_path().unlink(missing_ok=True)
+    # First, so the configure (even a first solve) installs from the mirror.
+    _sync_component_mirror()
     cmd = [_get_idf_tool("cmake"), "-G", "Ninja"]
     if _idf_py().binary_dir_arg:
         cmd += ["-B", str(build_dir)]
@@ -387,6 +398,9 @@ def run_reconfigure(verbose: bool = False) -> int:
         (build_dir / "CMakeCache.txt").unlink(missing_ok=True)
         _LOGGER.error("CMake configure failed with exit code %d", rc)
         _print_hints(log_path)
+        return rc
+    # Mirror what the solve added to the lock (transitive dependencies).
+    _sync_component_mirror()
     return rc
 
 
@@ -481,6 +495,24 @@ def _print_hints(log_path: Path) -> None:
         _LOGGER.warning("%s", hints)
 
 
+def _sync_component_mirror() -> None:
+    """Best-effort update of the local registry mirror; never fails the build."""
+    if not _esphome_manages_idf():
+        # _get_idf_env injects no mirror env, so nothing would read a sync.
+        return
+    cache = _cache()
+    if cache.mirror_sync_failed:
+        return
+    if not sync_component_mirror(
+        CORE.relative_build_path("dependencies.lock"),
+        CORE.relative_build_path("src/idf_component.yml"),
+        lambda: _get_idf_tool("python"),
+        _get_idf_env,
+    ):
+        # One failed attempt (e.g. offline) is enough per run.
+        cache.mirror_sync_failed = True
+
+
 def _builtin_component_cache_path() -> Path | None:
     """Cache file for this build's built-in component list.
 
@@ -491,7 +523,7 @@ def _builtin_component_cache_path() -> Path | None:
     of CONFIG_* options and only gate their sources on them. A checkout
     supplied through IDF_PATH is not managed by ESPHome and is never cached.
     """
-    if "IDF_PATH" in os.environ:
+    if not _esphome_manages_idf():
         return None
     target = variant_to_idf_target(CORE.data[KEY_ESP32][KEY_VARIANT])
     excluded = CORE.cmake_args.get("EXCLUDE_COMPONENTS", "")
@@ -793,6 +825,8 @@ def run_compile(config, verbose: bool) -> int:
             return rc
     else:
         _LOGGER.info("Build configuration is up to date")
+        # Ninja can still re-run cmake on its own; keep the mirror current.
+        _sync_component_mirror()
 
     if not get_bool_env("ESPHOME_LDGEN_FULL_DEPS"):
         _warn_if_app_archive_mapped()

esphome/framework_helpers.py

@@ -461,7 +461,7 @@ def _zip_extract_all(
             report(1)
 
 
-def _rename_with_retry(
+def rename_with_retry(
     src: Path, dst: Path, attempts: int = 5, overwrite: bool = False
 ) -> None:
     """Rename ``src`` to ``dst`` with backoff retries on Windows sharing violations.
@@ -561,7 +561,7 @@ def _7z_extract_all(
                     rmtree(dest)
                 else:
                     dest.unlink()
-            _rename_with_retry(item, dest)
+            rename_with_retry(item, dest)
     finally:
         # staging is created before the try, so it always exists here; the
         # guard is defensive cleanup and its False branch is unreachable.
@@ -1193,7 +1193,7 @@ def download_with_resume(
             # retries fail, keep the verified part so the next attempt (or
             # run) only has to redo the rename, not the download.
             try:
-                _rename_with_retry(part, dest, overwrite=True)
+                rename_with_retry(part, dest, overwrite=True)
             except PermissionError as e:
                 _LOGGER.debug("Could not move %s into place: %s", part, e)
                 last_error = e

tests/unit_tests/test_espidf_component_mirror.py

@@ -0,0 +1,801 @@
+"""Tests for the local IDF component-registry mirror."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+import json
+from pathlib import Path
+import subprocess
+from unittest.mock import MagicMock, patch
+import zipfile
+
+from filelock import FileLock
+import pytest
+
+from esphome.core import EsphomeError
+from esphome.espidf import component_mirror
+
+_LOCK_TEXT = """\
+dependencies:
+  bblanchon/arduinojson:
+    component_hash: 5e6aff2bc5bc988b36c8e90c01e637495ee895ad3af93911860c21ec6da79b6a
+    dependencies: []
+    source:
+      registry_url: https://components.espressif.com/
+      type: service
+    version: 7.4.3
+  espressif/mdns:
+    component_hash: 3ba256ac95e07c274be53cbd73f06cb846c403b61e8fbdf1be57bdb79db7a63e
+    dependencies:
+    - name: idf
+      require: private
+      version: '>=5.0'
+    source:
+      registry_url: https://components.espressif.com
+      type: service
+    version: 1.12.0
+  esphome/libsodium:
+    source:
+      path: /pio_components/idf/cb5ad5b7/esphome/libsodium
+      type: local
+    version: '*'
+  fastled/fastled:
+    source:
+      git: https://github.com/fastled/fastled
+      type: git
+    version: 3.7.0
+  other/private:
+    source:
+      registry_url: https://registry.example.com
+      type: service
+    version: 1.0.0
+  idf:
+    source:
+      type: idf
+    version: 5.5.5
+direct_dependencies:
+- bblanchon/arduinojson
+- espressif/mdns
+manifest_hash: f4e38c031cbb70d5070c5806fc580cce194b8c9a0b919f52a7e952f7d496dda4
+target: esp32
+version: 2.0.0
+"""
+
+_MANIFEST_TEXT = """\
+dependencies:
+  espressif/mdns:
+    version: 1.12.0
+  espressif/esp-tflite-micro:
+    version: 1.3.3~1
+  fastled/fastled:
+    version: 3.7.0
+    git: https://github.com/fastled/fastled
+  esphome/libsodium:
+    version: '*'
+    override_path: /stub/libsodium
+  espressif/ranged:
+    version: ^1.2.0
+  esphome/shorthand:
+    version: ==2.1.0
+  esphome/unquoted:
+    version: 2
+"""
+
+_ARDUINOJSON = component_mirror.ServiceDep("bblanchon", "arduinojson", "7.4.3")
+_MDNS = component_mirror.ServiceDep("espressif", "mdns", "1.12.0")
+_TFLITE = component_mirror.ServiceDep("espressif", "esp-tflite-micro", "1.3.3~1")
+_SHORTHAND = component_mirror.ServiceDep("esphome", "shorthand", "2.1.0")
+_MANIFEST_DEPS = [_MDNS, _TFLITE, _SHORTHAND]
+_NS_CMP_1 = component_mirror.ServiceDep("ns", "cmp", "1.0.0")
+_NS_CMP_2 = component_mirror.ServiceDep("ns", "cmp", "2.0.0")
+
+
+@pytest.fixture(autouse=True)
+def _isolate_mirror(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("ESPHOME_ESP_IDF_PREFIX", str(tmp_path / "idf_install"))
+
+
+def _write_lock(project_dir: Path, text: str = _LOCK_TEXT) -> Path:
+    lock = project_dir / "dependencies.lock"
+    lock.parent.mkdir(parents=True, exist_ok=True)
+    lock.write_text(text, encoding="utf-8")
+    return lock
+
+
+def _ns_cmp_lock(version: str) -> str:
+    return (
+        "dependencies:\n"
+        "  ns/cmp:\n"
+        "    source:\n"
+        "      type: service\n"
+        f"    version: {version}\n"
+    )
+
+
+def _write_manifest(project_dir: Path, text: str = _MANIFEST_TEXT) -> Path:
+    manifest = project_dir / "src" / "idf_component.yml"
+    manifest.parent.mkdir(parents=True, exist_ok=True)
+    manifest.write_text(text, encoding="utf-8")
+    return manifest
+
+
+def _add_to_mirror(mirror: Path, dep: component_mirror.ServiceDep) -> None:
+    """Lay out one component the way `registry sync` does."""
+    version_dir = f"components/{dep.namespace}/{dep.name}/{dep.version}"
+    archive = f"{version_dir}/{dep.namespace}__{dep.name}-v{dep.version}.zip"
+    checksums = f"{version_dir}/CHECKSUMS.json"
+    json_path = mirror / "components" / dep.namespace / f"{dep.name}.json"
+    json_path.parent.mkdir(parents=True, exist_ok=True)
+    json_path.write_text(
+        json.dumps(
+            {
+                "name": dep.name,
+                "namespace": dep.namespace,
+                "versions": [
+                    {"version": dep.version, "url": archive, "checksums": checksums}
+                ],
+            }
+        ),
+        encoding="utf-8",
+    )
+    archive_path = mirror / archive
+    archive_path.parent.mkdir(parents=True, exist_ok=True)
+    with zipfile.ZipFile(archive_path, "w") as zf:
+        zf.writestr("idf_component.yml", "")
+    (mirror / checksums).write_text("{}", encoding="utf-8")
+
+
+# ---------------------------------------------------------------------------
+# parse_lock_service_deps / parse_manifest_service_deps / project_service_deps
+# ---------------------------------------------------------------------------
+
+
+def test_parse_lock_keeps_only_default_registry_service_deps(tmp_path: Path) -> None:
+    """git, local, idf and non-default registry sources cannot be mirrored."""
+    deps = component_mirror.parse_lock_service_deps(_write_lock(tmp_path))
+    assert deps == [_ARDUINOJSON, _MDNS]
+
+
+def test_parse_lock_missing_file(tmp_path: Path) -> None:
+    assert component_mirror.parse_lock_service_deps(tmp_path / "none.lock") == []
+
+
+@pytest.mark.parametrize(
+    "text",
+    [
+        pytest.param("{unbalanced", id="corrupt-yaml"),
+        pytest.param("[]", id="not-a-mapping"),
+        pytest.param("dependencies:\n  espressif/mdns: not-a-dict\n", id="bad-entry"),
+        pytest.param(_ns_cmp_lock("1.2"), id="non-string-version"),
+    ],
+)
+def test_parse_lock_tolerates_bad_content(tmp_path: Path, text: str) -> None:
+    assert component_mirror.parse_lock_service_deps(_write_lock(tmp_path, text)) == []
+
+
+def test_parse_lock_unreadable_file_warns(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Read trouble other than a missing file says so instead of raising."""
+    lock = tmp_path / "dependencies.lock"
+    lock.mkdir()  # read_text raises OSError, not FileNotFoundError
+    assert component_mirror.parse_lock_service_deps(lock) == []
+    assert "Could not read" in caplog.text
+
+
+def test_parse_lock_non_utf8_file_warns(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A non-UTF-8 lock is a warning, never a failed build."""
+    lock = tmp_path / "dependencies.lock"
+    lock.write_bytes(b"dependencies:\n  # caf\xe9\n")
+    assert component_mirror.parse_lock_service_deps(lock) == []
+    assert "Could not read" in caplog.text
+
+
+def test_parse_lock_defaults_registry_url(tmp_path: Path) -> None:
+    """A service entry without registry_url is a default-registry dependency."""
+    lock = _write_lock(tmp_path, _ns_cmp_lock("1.0.0"))
+    assert component_mirror.parse_lock_service_deps(lock) == [_NS_CMP_1]
+
+
+def test_parse_manifest_keeps_only_exact_registry_pins(tmp_path: Path) -> None:
+    """Only exact pins survive, including the == shorthand prefix."""
+    deps = component_mirror.parse_manifest_service_deps(_write_manifest(tmp_path))
+    assert deps == _MANIFEST_DEPS
+
+
+def test_parse_manifest_missing_file(tmp_path: Path) -> None:
+    assert component_mirror.parse_manifest_service_deps(tmp_path / "none.yml") == []
+
+
+def test_parse_manifest_lowercases_mixed_case_keys(tmp_path: Path) -> None:
+    """The registry stores lowercase paths; a mixed-case key must match them."""
+    manifest = tmp_path / "idf_component.yml"
+    manifest.write_text("dependencies:\n  Espressif/MDNS:\n    version: 1.2.0\n")
+    deps = component_mirror.parse_manifest_service_deps(manifest)
+    assert deps == [component_mirror.ServiceDep("espressif", "mdns", "1.2.0")]
+
+
+def test_parse_manifest_ranged_specs(tmp_path: Path) -> None:
+    """Ranges, not exact pins, git/override entries or an empty version,
+    become refresh specs."""
+    manifest = _write_manifest(
+        tmp_path, _MANIFEST_TEXT + "  ns/empty:\n    version: ''\n"
+    )
+    specs = component_mirror.parse_manifest_ranged_specs(manifest)
+    assert specs == ["espressif/ranged^1.2.0"]
+
+
+def test_project_service_deps_merges_lock_and_manifest(tmp_path: Path) -> None:
+    """The lock wins for a component in both; the manifest fills the rest."""
+    lock = _write_lock(tmp_path)
+    manifest = _write_manifest(tmp_path)
+    assert component_mirror.project_service_deps(lock, manifest) == [
+        _ARDUINOJSON,
+        _MDNS,
+        _TFLITE,
+        _SHORTHAND,
+    ]
+
+
+def test_project_service_deps_manifest_only(tmp_path: Path) -> None:
+    """A fresh build has no lock yet; the manifest alone drives the sync."""
+    manifest = _write_manifest(tmp_path)
+    assert (
+        component_mirror.project_service_deps(tmp_path / "dependencies.lock", manifest)
+        == _MANIFEST_DEPS
+    )
+
+
+# ---------------------------------------------------------------------------
+# missing_deps
+# ---------------------------------------------------------------------------
+
+
+def test_missing_deps_covered_and_not(tmp_path: Path) -> None:
+    mirror = tmp_path / "mirror"
+    _add_to_mirror(mirror, _MDNS)
+    deps = [
+        _MDNS,
+        component_mirror.ServiceDep("espressif", "mdns", "1.13.0"),
+        _ARDUINOJSON,
+    ]
+    assert component_mirror.missing_deps(mirror, deps) == deps[1:]
+
+
+@pytest.mark.parametrize(
+    "damage",
+    [
+        pytest.param(
+            lambda m: next(m.rglob("*.zip")).unlink(),
+            id="archive-gone",
+        ),
+        pytest.param(
+            lambda m: (m / "components/espressif/mdns/1.12.0/CHECKSUMS.json").unlink(),
+            id="checksums-gone",
+        ),
+        pytest.param(
+            lambda m: (m / "components/espressif/mdns.json").write_text("{broken"),
+            id="corrupt-index",
+        ),
+        pytest.param(
+            lambda m: next(m.rglob("*.zip")).write_bytes(b"PK\x03\x04torn"),
+            id="truncated-archive",
+        ),
+    ],
+)
+def test_missing_deps_damaged_mirror(
+    tmp_path: Path, damage: Callable[[Path], object]
+) -> None:
+    """Every file the index references must exist, and a broken index
+    counts as missing, not covered."""
+    mirror = tmp_path / "mirror"
+    _add_to_mirror(mirror, _MDNS)
+    damage(mirror)
+    assert component_mirror.missing_deps(mirror, [_MDNS]) == [_MDNS]
+
+
+def test_missing_deps_deletes_a_torn_archive(tmp_path: Path) -> None:
+    """A torn archive is removed on detection, so the manager falls back to
+    the registry even before a heal sync runs."""
+    mirror = tmp_path / "mirror"
+    _add_to_mirror(mirror, _MDNS)
+    torn = next(mirror.rglob("*.zip"))
+    torn.write_bytes(b"PK\x03\x04torn")
+    assert component_mirror.missing_deps(mirror, [_MDNS]) == [_MDNS]
+    assert not torn.exists()
+
+
+def test_missing_deps_covered_without_checksums_field(tmp_path: Path) -> None:
+    """An older payload without a checksums field only needs its archive."""
+    mirror = tmp_path / "mirror"
+    _add_to_mirror(mirror, _MDNS)
+    index = mirror / "components/espressif/mdns.json"
+    doc = json.loads(index.read_text())
+    del doc["versions"][0]["checksums"]
+    index.write_text(json.dumps(doc))
+    assert component_mirror.missing_deps(mirror, [_MDNS]) == []
+
+
+# ---------------------------------------------------------------------------
+# _publish_index / _promote
+# ---------------------------------------------------------------------------
+
+
+def _staged_index(tmp_path: Path, versions: list) -> Path:
+    src = tmp_path / "staged.json"
+    src.write_text(json.dumps({"versions": versions}), encoding="utf-8")
+    return src
+
+
+def test_publish_index_first_sync(tmp_path: Path) -> None:
+    """No live index yet: the staged one is published as is."""
+    src = _staged_index(tmp_path, [{"version": "2.0.0"}])
+    dst = tmp_path / "live.json"
+    component_mirror._publish_index(src, dst)
+    assert json.loads(dst.read_text()) == {"versions": [{"version": "2.0.0"}]}
+
+
+@pytest.mark.parametrize(
+    "live", ["not json", '{"versions": null}', '{"versions": {"a": 1}}']
+)
+def test_publish_index_replaces_a_corrupt_live_index(tmp_path: Path, live: str) -> None:
+    """An unreadable live index is replaced wholesale; that is the heal."""
+    src = _staged_index(tmp_path, [{"version": "2.0.0"}])
+    dst = tmp_path / "live.json"
+    dst.write_text(live)
+    component_mirror._publish_index(src, dst)
+    assert json.loads(dst.read_text()) == {"versions": [{"version": "2.0.0"}]}
+
+
+def test_publish_index_merges_and_filters(tmp_path: Path) -> None:
+    """Unfetched live versions survive; non-dict entries drop, not raise."""
+    src = _staged_index(tmp_path, [{"version": "2.0.0"}, "junk"])
+    dst = tmp_path / "live.json"
+    dst.write_text(
+        json.dumps({"versions": [{"version": "1.0.0"}, {"version": "2.0.0"}, "bad"]})
+    )
+    component_mirror._publish_index(src, dst)
+    assert json.loads(dst.read_text()) == {
+        "versions": [{"version": "2.0.0"}, {"version": "1.0.0"}]
+    }
+
+
+def test_publish_index_unreadable_live_index_raises(tmp_path: Path) -> None:
+    """An I/O error is not corruption; raising keeps the live versions and
+    a later run retries."""
+    src = _staged_index(tmp_path, [{"version": "2.0.0"}])
+    dst = tmp_path / "live.json"
+    dst.mkdir()  # read_text raises IsADirectoryError, not FileNotFoundError
+    with pytest.raises(OSError):
+        component_mirror._publish_index(src, dst)
+
+
+def test_promote_moves_archives_before_indexes(tmp_path: Path) -> None:
+    """A concurrent reader must never see an index entry without its files."""
+    staging = tmp_path / "staging"
+    _add_to_mirror(staging, _NS_CMP_2)
+    order: list[str] = []
+    real_rename = component_mirror.rename_with_retry
+    real_write = component_mirror.write_file
+
+    def recording_rename(src: Path, dst: Path, **kwargs) -> None:
+        order.append(src.name)
+        real_rename(src, dst, **kwargs)
+
+    def recording_write(path: Path, text: str) -> None:
+        order.append(path.name)
+        real_write(path, text)
+
+    with (
+        patch.object(component_mirror, "rename_with_retry", recording_rename),
+        patch.object(component_mirror, "write_file", recording_write),
+    ):
+        component_mirror._promote(staging, tmp_path / "mirror")
+
+    assert order[-1] == "cmp.json"
+    assert set(order[:-1]) == {"CHECKSUMS.json", "ns__cmp-v2.0.0.zip"}
+
+
+# ---------------------------------------------------------------------------
+# component_mirror_env
+# ---------------------------------------------------------------------------
+
+
+def test_component_mirror_env_sets_file_url(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.delenv("IDF_COMPONENT_LOCAL_STORAGE_URL", raising=False)
+    monkeypatch.delenv("IDF_COMPONENT_CHECK_NEW_VERSION", raising=False)
+    env = component_mirror.component_mirror_env()
+    mirror = component_mirror.get_mirror_path()
+    assert env["IDF_COMPONENT_LOCAL_STORAGE_URL"] == mirror.as_uri()
+    assert env["IDF_COMPONENT_CHECK_NEW_VERSION"] == "0"
+    assert mirror.is_dir()
+
+
+def test_component_mirror_env_preserves_user_values(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A user local storage list keeps precedence; their new-version choice wins."""
+    monkeypatch.setenv("IDF_COMPONENT_LOCAL_STORAGE_URL", "file:///user/mirror")
+    monkeypatch.setenv("IDF_COMPONENT_CHECK_NEW_VERSION", "1")
+    env = component_mirror.component_mirror_env()
+    mirror_uri = component_mirror.get_mirror_path().as_uri()
+    assert env["IDF_COMPONENT_LOCAL_STORAGE_URL"] == f"file:///user/mirror;{mirror_uri}"
+    assert "IDF_COMPONENT_CHECK_NEW_VERSION" not in env
+
+
+def test_component_mirror_env_unwritable_cache(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A read-only cache disables the feature with a warning, not a failure."""
+    with patch.object(Path, "mkdir", side_effect=OSError("read-only")):
+        assert component_mirror.component_mirror_env() == {}
+    assert "Component mirror unavailable" in caplog.text
+
+
+# ---------------------------------------------------------------------------
+# sync_component_mirror
+# ---------------------------------------------------------------------------
+
+
+def _run_sync(
+    project_dir: Path,
+    *,
+    returncode: int = 0,
+    stderr: str = "",
+    side_effect: Exception | Callable | None = None,
+    get_python=lambda: "/penv/python",
+    get_env=lambda: {"PATH": "/penv"},
+) -> tuple[bool, MagicMock]:
+    with patch.object(
+        component_mirror.subprocess,
+        "run",
+        return_value=subprocess.CompletedProcess([], returncode, "", stderr),
+        side_effect=side_effect,
+    ) as mock_run:
+        ok = component_mirror.sync_component_mirror(
+            project_dir / "dependencies.lock",
+            project_dir / "src" / "idf_component.yml",
+            get_python,
+            get_env,
+        )
+    return ok, mock_run
+
+
+def _fake_registry_sync(returncode: int = 0):
+    """A subprocess.run stand-in that lays files out like `registry sync`."""
+
+    def run(cmd, **kwargs) -> subprocess.CompletedProcess:
+        _add_to_mirror(Path(cmd[-1]), _NS_CMP_2)
+        return subprocess.CompletedProcess(cmd, returncode, "", "sync failed")
+
+    return run
+
+
+def _assert_sync_lock_released() -> None:
+    lock = FileLock(str(component_mirror.get_mirror_path() / ".sync.lock"))
+    lock.acquire(blocking=False)
+    lock.release()
+
+
+def test_sync_runs_the_manager_for_missing_deps(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    _write_lock(tmp_path)
+    ok, mock_run = _run_sync(tmp_path)
+    # The stub stages nothing, so the post-promote recheck reports failure.
+    assert not ok
+    assert "uncovered" in caplog.text
+    mirror = component_mirror.get_mirror_path()
+    # The whole set goes in one invocation: one manager startup.
+    mock_run.assert_called_once()
+    assert mock_run.call_args.args[0] == [
+        "/penv/python",
+        "-m",
+        "idf_component_manager",
+        "registry",
+        "sync",
+        "--resolution",
+        "latest",
+        "--component",
+        "bblanchon/arduinojson==7.4.3",
+        "--component",
+        "espressif/mdns==1.12.0",
+        str(mirror / ".staging"),
+    ]
+    assert mock_run.call_args.kwargs["env"] == {"PATH": "/penv"}
+    _assert_sync_lock_released()
+
+
+def test_sync_skips_when_covered(tmp_path: Path) -> None:
+    """A covered project resolves no environment and runs no subprocess."""
+    _write_lock(tmp_path)
+    mirror = component_mirror.get_mirror_path()
+    _add_to_mirror(mirror, _MDNS)
+    _add_to_mirror(mirror, _ARDUINOJSON)
+
+    def _boom() -> str:
+        raise AssertionError("environment resolved for a covered mirror")
+
+    ok, mock_run = _run_sync(tmp_path, get_python=_boom, get_env=_boom)
+    assert ok
+    mock_run.assert_not_called()
+
+
+def test_sync_failure_is_tolerated(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    _write_lock(tmp_path)
+    ok, mock_run = _run_sync(tmp_path, returncode=1, stderr="boom")
+    assert not ok
+    mock_run.assert_called_once()
+    assert "Could not mirror" in caplog.text
+    _assert_sync_lock_released()
+
+
+def test_sync_subprocess_errors_are_tolerated(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    _write_lock(tmp_path)
+    ok, _ = _run_sync(tmp_path, side_effect=OSError("no such file"))
+    assert not ok
+    assert "Could not mirror" in caplog.text
+    _assert_sync_lock_released()
+
+
+def test_sync_refreshes_ranged_specs_only_before_a_fresh_solve(
+    tmp_path: Path,
+) -> None:
+    """Without a lock the batch also re-resolves ranged pins, as dev would;
+    with a lock the ranged candidate is left alone."""
+    _write_manifest(tmp_path)
+    ok, mock_run = _run_sync(tmp_path)
+    assert not ok  # the stub stages nothing, so the recheck fails
+    assert "espressif/ranged^1.2.0" in mock_run.call_args.args[0]
+
+    _write_lock(tmp_path)
+    ok, mock_run = _run_sync(tmp_path)
+    assert not ok
+    assert "espressif/ranged^1.2.0" not in mock_run.call_args.args[0]
+
+
+def test_sync_ranged_refresh_runs_with_everything_else_covered(
+    tmp_path: Path,
+) -> None:
+    """A covered build still refreshes its ranged pins when the lock is
+    gone, in one invocation."""
+    manifest = _write_manifest(tmp_path)
+    mirror = component_mirror.get_mirror_path()
+    for dep in component_mirror.parse_manifest_service_deps(manifest):
+        _add_to_mirror(mirror, dep)
+    ok, mock_run = _run_sync(tmp_path)
+    assert ok
+    mock_run.assert_called_once()
+    assert "espressif/ranged^1.2.0" in mock_run.call_args.args[0]
+
+
+def test_sync_timeout_retries_ranged_specs_too(tmp_path: Path) -> None:
+    """The fallback covers refresh specs, not only the exact pins."""
+    _write_manifest(
+        tmp_path,
+        "dependencies:\n  ns/one:\n    version: ^1.0\n  ns/two:\n    version: ^2.0\n",
+    )
+    effects = [
+        subprocess.TimeoutExpired(cmd=[], timeout=120),
+        subprocess.CompletedProcess([], 0, "", ""),
+        subprocess.CompletedProcess([], 0, "", ""),
+    ]
+    ok, mock_run = _run_sync(tmp_path, side_effect=effects)
+    assert ok
+    retried = [call.args[0][-2] for call in mock_run.call_args_list[1:]]
+    assert retried == ["ns/one^1.0", "ns/two^2.0"]
+
+
+def test_sync_timeout_falls_back_to_one_component_at_a_time(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A timed-out batch retries per component, so the ones that finish
+    stay promoted and later runs fetch only the rest."""
+    _write_lock(tmp_path)
+
+    def slow_sync(cmd, **kwargs):
+        specs = [arg for arg in cmd if "==" in arg]
+        if specs != ["bblanchon/arduinojson==7.4.3"]:
+            # The batch, and later the mdns retry, exceed the timeout.
+            raise subprocess.TimeoutExpired(cmd=cmd, timeout=120)
+        namespace, _, rest = specs[0].partition("/")
+        name, _, version = rest.partition("==")
+        _add_to_mirror(
+            Path(cmd[-1]), component_mirror.ServiceDep(namespace, name, version)
+        )
+        return subprocess.CompletedProcess(cmd, 0, "", "")
+
+    ok, mock_run = _run_sync(tmp_path, side_effect=slow_sync)
+    assert not ok
+    assert mock_run.call_count == 3  # batch, arduinojson, mdns
+    assert "retrying one component" in caplog.text
+    assert "kept 1 of 2" in caplog.text
+    mirror = component_mirror.get_mirror_path()
+    assert component_mirror.missing_deps(mirror, [_ARDUINOJSON, _MDNS]) == [_MDNS]
+    assert not (mirror / ".staging").exists()
+    _assert_sync_lock_released()
+
+
+def test_sync_timeout_then_per_component_completes(tmp_path: Path) -> None:
+    """When every component fits on its own, the fallback finishes the job."""
+    _write_lock(tmp_path)
+
+    def slow_sync(cmd, **kwargs):
+        specs = [arg for arg in cmd if "==" in arg]
+        if len(specs) > 1:
+            raise subprocess.TimeoutExpired(cmd=cmd, timeout=120)
+        namespace, _, rest = specs[0].partition("/")
+        name, _, version = rest.partition("==")
+        _add_to_mirror(
+            Path(cmd[-1]), component_mirror.ServiceDep(namespace, name, version)
+        )
+        return subprocess.CompletedProcess(cmd, 0, "", "")
+
+    ok, mock_run = _run_sync(tmp_path, side_effect=slow_sync)
+    assert ok
+    assert mock_run.call_count == 3  # batch, then one call per component
+    mirror = component_mirror.get_mirror_path()
+    assert component_mirror.missing_deps(mirror, [_ARDUINOJSON, _MDNS]) == []
+
+
+def test_sync_timeout_then_component_error_stops(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A hard error in the fallback loop stops it like any other failure."""
+    _write_lock(tmp_path)
+    effects = [
+        subprocess.TimeoutExpired(cmd=[], timeout=120),
+        subprocess.CompletedProcess([], 1, "", "boom"),
+    ]
+    ok, mock_run = _run_sync(tmp_path, side_effect=effects)
+    assert not ok
+    assert mock_run.call_count == 2
+    assert "Could not mirror" in caplog.text
+    _assert_sync_lock_released()
+
+
+def test_sync_unexpected_error_logs_a_traceback(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """An operational failure warns in one line; a bug keeps its traceback."""
+    _write_lock(tmp_path)
+    with patch.object(component_mirror, "_promote", side_effect=KeyError("url")):
+        ok, _ = _run_sync(tmp_path)
+    assert not ok
+    record = next(r for r in caplog.records if "Could not mirror" in r.message)
+    assert record.exc_info
+
+
+def test_sync_single_component_timeout_is_not_retried(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """With one missing component the batch was already the retry; a build
+    pays the timeout once."""
+    _write_lock(tmp_path, _ns_cmp_lock("2.0.0"))
+    ok, mock_run = _run_sync(
+        tmp_path, side_effect=subprocess.TimeoutExpired(cmd=[], timeout=120)
+    )
+    assert not ok
+    mock_run.assert_called_once()
+    assert "kept 0 of 1" in caplog.text
+    _assert_sync_lock_released()
+
+
+def test_sync_environment_resolution_failure_is_tolerated(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    _write_lock(tmp_path)
+
+    def _boom() -> str:
+        raise EsphomeError("python not found")
+
+    ok, mock_run = _run_sync(tmp_path, get_python=_boom)
+    assert not ok
+    mock_run.assert_not_called()
+    assert "Could not mirror" in caplog.text
+
+
+def test_sync_skips_when_another_process_holds_the_lock(tmp_path: Path) -> None:
+    _write_lock(tmp_path)
+    mirror = component_mirror.get_mirror_path()
+    mirror.mkdir(parents=True)
+    held = FileLock(str(mirror / ".sync.lock"))
+    held.acquire(blocking=False)
+    try:
+        ok, mock_run = _run_sync(tmp_path)
+    finally:
+        held.release()
+    assert ok
+    mock_run.assert_not_called()
+
+
+def test_sync_ignores_a_leftover_lock_file(tmp_path: Path) -> None:
+    """A lock file from a dead process does not block: the OS lock is gone."""
+    _write_lock(tmp_path)
+    mirror = component_mirror.get_mirror_path()
+    mirror.mkdir(parents=True)
+    (mirror / ".sync.lock").touch()
+    _, mock_run = _run_sync(tmp_path)
+    mock_run.assert_called()  # the dead lock did not block the sync
+
+
+def test_sync_lock_oserror_is_a_failure(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A broken cache is not contention; the retry guard should apply."""
+    _write_lock(tmp_path)
+    with patch("filelock.FileLock.acquire", side_effect=PermissionError("ro")):
+        ok, mock_run = _run_sync(tmp_path)
+    assert not ok
+    mock_run.assert_not_called()
+    assert "Could not lock" in caplog.text
+
+
+def test_sync_undeletable_staging_is_a_failure(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """An undeletable staging tree aborts the sync; stale files must
+    never be promoted."""
+    _write_lock(tmp_path)
+    with patch.object(component_mirror, "rmtree", side_effect=OSError("stuck staging")):
+        ok, mock_run = _run_sync(tmp_path)
+    assert not ok
+    mock_run.assert_not_called()
+    assert "Could not mirror" in caplog.text
+    _assert_sync_lock_released()
+
+
+def test_sync_promotes_staged_files_and_merges_the_index(tmp_path: Path) -> None:
+    """New files land atomically and existing versions survive the merge."""
+    mirror = component_mirror.get_mirror_path()
+    _add_to_mirror(mirror, _NS_CMP_1)
+    _write_lock(tmp_path, _ns_cmp_lock("2.0.0"))
+
+    ok, _ = _run_sync(tmp_path, side_effect=_fake_registry_sync())
+
+    assert ok
+    doc = json.loads((mirror / "components" / "ns" / "cmp.json").read_text())
+    assert {entry["version"] for entry in doc["versions"]} == {"1.0.0", "2.0.0"}
+    assert (mirror / "components/ns/cmp/2.0.0/ns__cmp-v2.0.0.zip").is_file()
+    assert (mirror / "components/ns/cmp/1.0.0/ns__cmp-v1.0.0.zip").is_file()
+    assert not (mirror / ".staging").exists()
+
+
+def test_sync_failure_leaves_no_staging_behind(tmp_path: Path) -> None:
+    """A failed sync promotes nothing and removes its staging directory."""
+    _write_lock(tmp_path, _ns_cmp_lock("2.0.0"))
+
+    ok, _ = _run_sync(tmp_path, side_effect=_fake_registry_sync(returncode=1))
+
+    assert not ok
+    mirror = component_mirror.get_mirror_path()
+    assert not (mirror / ".staging").exists()
+    assert not (mirror / "components" / "ns").exists()
+
+
+def test_sync_failed_index_publish_keeps_the_live_index(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A failed publish keeps the live index; a later run retries."""
+    mirror = component_mirror.get_mirror_path()
+    _add_to_mirror(mirror, _NS_CMP_1)
+    _write_lock(tmp_path, _ns_cmp_lock("2.0.0"))
+
+    with patch.object(
+        component_mirror, "write_file", side_effect=EsphomeError("disk full")
+    ):
+        ok, _ = _run_sync(tmp_path, side_effect=_fake_registry_sync())
+
+    assert not ok
+    assert "Could not mirror" in caplog.text
+    doc = json.loads((mirror / "components" / "ns" / "cmp.json").read_text())
+    assert {entry["version"] for entry in doc["versions"]} == {"1.0.0"}
+    _assert_sync_lock_released()

tests/unit_tests/test_espidf_toolchain.py

@@ -215,6 +215,38 @@ def test_get_idf_env_sets_git_ceiling_directories(setup_core: Path) -> None:
     assert str(CORE.config_dir) in env["GIT_CEILING_DIRECTORIES"].split(os.pathsep)
 
 
+def test_get_idf_env_serves_the_component_mirror(setup_core: Path) -> None:
+    """The mirror env rides the managed-IDF branch, next to its sync gate."""
+    toolchain._cache().env.clear()
+    with (
+        patch.object(
+            toolchain,
+            "get_framework_env",
+            return_value={"PATH": "/penv"},
+        ),
+        patch.object(toolchain, "_get_esphome_esp_idf_paths", return_value=((), {})),
+        patch.object(
+            toolchain,
+            "component_mirror_env",
+            return_value={"IDF_COMPONENT_LOCAL_STORAGE_URL": "file:///mirror"},
+        ),
+    ):
+        env = toolchain._get_idf_env(version="5.5.4")
+    assert env["IDF_COMPONENT_LOCAL_STORAGE_URL"] == "file:///mirror"
+
+
+def test_get_idf_env_user_idf_skips_the_mirror(setup_core: Path) -> None:
+    """A user-managed IDF gets neither the framework env nor the mirror."""
+    toolchain._cache().env.clear()
+    with (
+        patch.dict(os.environ, {"IDF_PATH": str(setup_core)}),
+        patch.object(toolchain, "component_mirror_env") as mock_env,
+    ):
+        env = toolchain._get_idf_env(version="5.5.4")
+    mock_env.assert_not_called()
+    assert "IDF_COMPONENT_LOCAL_STORAGE_URL" not in env
+
+
 def test_get_idf_env_pops_inherited_pythonpath(setup_core: Path) -> None:
     """A PYTHONPATH from the parent environment must not reach idf.py.
 
@@ -362,6 +394,73 @@ def test_run_compile_discovery_without_cmakecache(setup_core: Path) -> None:
     assert not CORE.relative_build_path("build/CMakeCache.txt").exists()
 
 
+@pytest.mark.parametrize(
+    ("cmake_rc", "expected"),
+    [(0, ["sync", "cmake", "sync"]), (1, ["sync", "cmake"])],
+)
+def test_run_reconfigure_syncs_mirror_around_the_configure(
+    setup_core: Path, cmake_rc: int, expected: list[str]
+) -> None:
+    """Every configure path syncs: before, so the solve installs from the
+    mirror; after a success, so solver-added lock entries are mirrored."""
+    _setup_build(setup_core)
+    calls: list[str] = []
+
+    with (
+        _fake_tools() as mock_run,
+        patch.object(
+            toolchain,
+            "_sync_component_mirror",
+            side_effect=lambda: calls.append("sync"),
+        ),
+        # The real hint printer resolves an IDF install (or downloads one).
+        patch.object(toolchain, "_print_hints"),
+    ):
+        mock_run.side_effect = lambda *a, **k: calls.append("cmake") or cmake_rc
+        assert toolchain.run_reconfigure() == cmake_rc
+
+    assert calls == expected
+
+
+def test_run_compile_syncs_mirror_when_up_to_date(setup_core: Path) -> None:
+    """No reconfigure still refreshes the mirror once for ninja-driven
+    cmake re-runs."""
+    _setup_build(setup_core)
+    config = {CONF_ESPHOME: {}}
+
+    with (
+        _up_to_date_compile(),
+        patch.object(toolchain, "_sync_component_mirror") as mock_sync,
+    ):
+        assert toolchain.run_compile(config, verbose=False) == 0
+
+    mock_sync.assert_called_once()
+
+
+def test_sync_component_mirror_skips_user_idf(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A user-managed IDF never gets the mirror env, so a sync is unread."""
+    monkeypatch.setenv("IDF_PATH", "/opt/esp-idf")
+    with patch.object(toolchain, "sync_component_mirror") as mock_sync:
+        toolchain._sync_component_mirror()
+    mock_sync.assert_not_called()
+
+
+def test_sync_component_mirror_failure_not_retried(
+    setup_core: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """One failed attempt (e.g. offline) must not stall every later call."""
+    monkeypatch.delenv("IDF_PATH", raising=False)
+    _setup_build(setup_core)
+    with patch.object(
+        toolchain, "sync_component_mirror", return_value=False
+    ) as mock_sync:
+        toolchain._sync_component_mirror()
+        toolchain._sync_component_mirror()
+    mock_sync.assert_called_once()
+
+
 def test_run_compile_reconfigures_after_full_write_outside_testing_mode(
     setup_core: Path,
 ) -> None:

tests/unit_tests/test_framework_helpers.py

@@ -27,7 +27,6 @@ from esphome.framework_helpers import (
     _7z_extract_all,
     _BatchDownloadProgress,
     _detect_archive_root,
-    _rename_with_retry,
     _tar_extract_all,
     _zip_extract_all,
     archive_extract_all,
@@ -40,6 +39,7 @@ from esphome.framework_helpers import (
     get_project_link_flags,
     get_python_env_executable_path,
     get_system_python_path,
+    rename_with_retry,
     rmdir,
     run_batch_downloads,
     run_command,
@@ -1013,7 +1013,7 @@ class TestDownloadWithResume:
         with (
             patch("requests.get", return_value=_mock_response(b"data")) as mock_get,
             patch(
-                "esphome.framework_helpers._rename_with_retry",
+                "esphome.framework_helpers.rename_with_retry",
                 side_effect=[PermissionError("locked"), None],
             ) as rename,
         ):
@@ -1030,7 +1030,7 @@ class TestDownloadWithResume:
         with (
             patch("requests.get", return_value=_mock_response(b"data")),
             patch(
-                "esphome.framework_helpers._rename_with_retry",
+                "esphome.framework_helpers.rename_with_retry",
                 side_effect=PermissionError("locked"),
             ),
             pytest.raises(EsphomeError, match="after 1 attempts"),
@@ -2019,7 +2019,7 @@ class TestZipExtractAllBranches:
 
 
 # ---------------------------------------------------------------------------
-# _rename_with_retry
+# rename_with_retry
 # ---------------------------------------------------------------------------
 
 
@@ -2028,7 +2028,7 @@ class TestRenameWithRetry:
         src = tmp_path / "src.txt"
         src.write_text("data")
         dst = tmp_path / "dst.txt"
-        _rename_with_retry(src, dst)
+        rename_with_retry(src, dst)
         assert dst.read_text() == "data"
         assert not src.exists()
 
@@ -2050,7 +2050,7 @@ class TestRenameWithRetry:
             patch.object(Path, "rename", flaky_rename),
             patch("esphome.framework_helpers.time.sleep"),
         ):
-            _rename_with_retry(src, dst, attempts=3)
+            rename_with_retry(src, dst, attempts=3)
         assert dst.read_text() == "data"
 
     def test_raises_after_all_attempts_fail(self, tmp_path: Path) -> None:
@@ -2062,14 +2062,14 @@ class TestRenameWithRetry:
             patch("esphome.framework_helpers.time.sleep"),
             pytest.raises(PermissionError),
         ):
-            _rename_with_retry(src, dst, attempts=3)
+            rename_with_retry(src, dst, attempts=3)
 
     def test_attempts_zero_is_noop(self, tmp_path: Path) -> None:
         """Zero attempts means the for-loop body never runs; src is untouched."""
         src = tmp_path / "src.txt"
         src.write_text("data")
         dst = tmp_path / "dst.txt"
-        _rename_with_retry(src, dst, attempts=0)
+        rename_with_retry(src, dst, attempts=0)
         assert src.exists()
         assert not dst.exists()
 

tests/unit_tests/test_lazy_imports.py

@@ -192,7 +192,9 @@ def test_native_upload_fast_path_does_not_import_heavy_modules(
         fixture_path,
         probe_env,
         "native_upload_fast_path.py",
-        extra=("esphome.components.esp8266",),
+        # The cache-hit modules guard the espidf toolchain import chain:
+        # a top-level yaml import there would tax every upload subprocess.
+        extra=("esphome.components.esp8266", *CACHE_HIT_HEAVY_MODULES),
     )
     assert not leaked, (
         f"native upload_using_esptool pulls in heavy modules: {leaked}. "