Merge pull request #18083 from esphome/bump-2026.7.4

2026.7.4

Doxyfile

@@ -48,7 +48,7 @@ PROJECT_NAME           = ESPHome
 # could be handy for archiving the generated documentation or if some version
 # control system is used.
 
-PROJECT_NUMBER         = 2026.7.3
+PROJECT_NUMBER         = 2026.7.4
 
 # Using the PROJECT_BRIEF tag one can provide an optional one line description
 # for a project that appears at the top of each page and should give viewer a

docker/Dockerfile

@@ -22,7 +22,7 @@ RUN \
     -r /requirements.txt
 
 # Install the ESPHome Device Builder dashboard.
-RUN uv pip install --no-cache-dir esphome-device-builder==1.7.0
+RUN uv pip install --no-cache-dir esphome-device-builder==1.9.2
 
 RUN \
     platformio settings set enable_telemetry No \

esphome/components/api/api_overflow_buffer.cpp

@@ -12,6 +12,22 @@ APIOverflowBuffer::~APIOverflowBuffer() {
 }
 
 ssize_t APIOverflowBuffer::try_drain(socket::Socket *socket) {
+  // socket->write() can re-enter this function: a log message emitted from an
+  // lwip callback during the write goes out over the API and lands back in the
+  // frame helper's write/drain path. If a nested drain ran here it would send
+  // and free the entry the outer drain is still holding, causing a double free.
+  // Report "no progress" instead; the outer drain keeps draining, and the
+  // nested send is enqueued behind the existing backlog.
+  if (this->draining_)
+    return 0;
+
+  // RAII so the flag is cleared on every return path
+  struct DrainGuard {
+    explicit DrainGuard(bool &flag) : flag_(flag) { flag_ = true; }
+    ~DrainGuard() { this->flag_ = false; }
+    bool &flag_;
+  } guard(this->draining_);
+
   while (this->count_ > 0) {
     Entry *front = this->queue_[this->head_];
 
@@ -29,11 +45,12 @@ ssize_t APIOverflowBuffer::try_drain(socket::Socket *socket) {
       return sent;
     }
 
-    // Entry fully sent — free it and advance
-    Entry::destroy(front);
+    // Entry fully sent — unlink it before freeing so a freed pointer is never
+    // reachable from the queue
     this->queue_[this->head_] = nullptr;
     this->head_ = (this->head_ + 1) % API_MAX_SEND_QUEUE;
     this->count_--;
+    Entry::destroy(front);
   }
 
   return 0;  // All drained

esphome/components/api/api_overflow_buffer.h

@@ -69,6 +69,10 @@ class APIOverflowBuffer {
   uint8_t head_{0};
   uint8_t tail_{0};
   uint8_t count_{0};
+  // Guards against re-entrant drains: socket->write() can re-enter the API
+  // send path (e.g. a log message emitted from an lwip callback), and a nested
+  // drain would free the entry the outer drain is still holding.
+  bool draining_{false};
 };
 
 }  // namespace esphome::api

esphome/components/epaper_spi/models/__init__.py

@@ -15,7 +15,7 @@ class EpaperModel:
         self,
         name: str,
         class_name: str,
-        initsequence=None,
+        initsequence=(),
         **defaults,
     ):
         name = name.upper()

esphome/components/esp32/__init__.py

@@ -628,7 +628,6 @@ class NetworkSdkconfigData:
     wifi_ap: bool = False  # WiFi AP mode configured
     ethernet: bool = False  # Ethernet component active
     bluetooth: bool = False  # any BLE component active
-    ble_42: bool = False  # BLE 4.2 features needed
     software_coexistence: bool = False  # WiFi/BT software coexistence requested
     # esp32 advanced enable_lwip_dhcp_server option (True/False/None=unset)
     enable_lwip_dhcp_server: bool | None = None
@@ -654,12 +653,10 @@ def request_ethernet() -> None:
     _network_sdkconfig().ethernet = True
 
 
-def request_bluetooth(ble_42: bool = False) -> None:
-    """Request the Bluetooth controller. Pass ble_42=True for 4.2 features."""
+def request_bluetooth() -> None:
+    """Request the Bluetooth controller."""
     net = _network_sdkconfig()
     net.bluetooth = True
-    if ble_42:
-        net.ble_42 = True
 
 
 def request_software_coexistence() -> None:
@@ -1029,7 +1026,9 @@ def _check_esp_idf_versions(config: ConfigType) -> ConfigType:
 
 
 def _validate_toolchain(value) -> Toolchain:
-    return Toolchain(cv.one_of(*(t.value for t in Toolchain), lower=True)(value))
+    return Toolchain(
+        cv.one_of(Toolchain.PLATFORMIO, Toolchain.ESP_IDF, lower=True)(value)
+    )
 
 
 def _resolve_toolchain(value: ConfigType) -> ConfigType:
@@ -2044,12 +2043,12 @@ async def _reconcile_network_sdkconfig() -> None:
         if name not in opts:
             add_idf_sdkconfig_option(name, value)
 
-    # Bluetooth: only ever enable when requested. The IDF default is off and
-    # nothing sets these False today, so never write False here.
+    # Bluetooth: only ever enable when requested. The IDF default is off.
+    # According to the IDF docs, only one of 4.2 or 5.0 should be enabled.
     if net.bluetooth:
         set_opt("CONFIG_BT_ENABLED", True)
-        if net.ble_42:
-            set_opt("CONFIG_BT_BLE_42_FEATURES_SUPPORTED", True)
+        set_opt("CONFIG_BT_BLE_42_FEATURES_SUPPORTED", True)
+        set_opt("CONFIG_BT_BLE_50_FEATURES_SUPPORTED", False)
 
     # WiFi stack: disable only when Ethernet is present and WiFi is not. WiFi
     # relies on the IDF default (enabled), so it is never written True here.

esphome/components/esp32_ble/__init__.py

@@ -604,7 +604,7 @@ async def to_code(config):
     max_connections = config.get(CONF_MAX_CONNECTIONS, DEFAULT_MAX_CONNECTIONS)
     cg.add_define("USE_ESP32_BLE_MAX_CONNECTIONS", max_connections)
 
-    request_bluetooth(ble_42=True)
+    request_bluetooth()
 
     # When PSRAM and BT are used together, Bluedroid should prefer SPIRAM for
     # heap allocations and use dynamic (heap-based) environment memory tables

esphome/components/esp32_ble_beacon/__init__.py

@@ -86,4 +86,4 @@ async def to_code(config):
 
     cg.add_define("USE_ESP32_BLE_ADVERTISING")
 
-    request_bluetooth(ble_42=True)
+    request_bluetooth()

esphome/components/packages/__init__.py

@@ -1,6 +1,7 @@
 from collections import UserDict
 from collections.abc import Callable
 from functools import reduce
+import logging
 from pathlib import Path
 from typing import Any
 
@@ -35,6 +36,8 @@ from esphome.const import (
 )
 from esphome.core import EsphomeError
 
+_LOGGER = logging.getLogger(__name__)
+
 DOMAIN = CONF_PACKAGES
 # Guard against infinite include chains (e.g. A includes B includes A).
 MAX_INCLUDE_DEPTH = 20
@@ -267,8 +270,23 @@ def _process_remote_package(config: dict[str, Any]) -> dict[str, Any]:
         # If loading fails, the cached checkout may be stale — revert and retry once.
         try:
             return {CONF_PACKAGES: get_packages(files)}
-        except cv.Invalid:
-            revert()
+        except cv.Invalid as err:
+            if not revert():
+                # The pre-update content is out of reach (lock timeout, the
+                # checkout moved, or the reset failed; see the log), so a
+                # retry could not see it.
+                raise cv.Invalid(
+                    f"Failed to load packages and could not revert the cached "
+                    f"checkout to retry. {err}",
+                    path=err.path,
+                ) from err
+            # If the retry succeeds this is the only trace that the
+            # refreshed upstream content was broken.
+            _LOGGER.warning(
+                "Loading packages failed (%s), reverted the cached checkout "
+                "and retrying",
+                err,
+            )
         try:
             return {CONF_PACKAGES: get_packages(files)}
         except cv.Invalid as err:

esphome/const.py

@@ -4,7 +4,7 @@ from enum import Enum
 
 from esphome.enum import StrEnum
 
-__version__ = "2026.7.3"
+__version__ = "2026.7.4"
 
 ALLOWED_NAME_CHARS = "abcdefghijklmnopqrstuvwxyz0123456789-_"
 VALID_SUBSTITUTIONS_CHARACTERS = (

esphome/espidf/component.py

@@ -92,6 +92,14 @@ def generate_cmakelists_txt(component: IDFComponent) -> str:
         # In CMakeLists.txt, backslashes need to be escaped
         return f'"{str(p)}"'.replace("\\", "\\\\")
 
+    def escape_path(p: PathType) -> str:
+        # CMake uses forward slashes for paths on every platform and treats
+        # backslashes as escape characters. On Windows os.path.relpath yields
+        # backslash paths, which break CMake's list re-parsing (e.g. "\b" in
+        # "src\backend" is an invalid character escape). Emit forward slashes,
+        # which Windows accepts too, so the generated CMakeLists is portable.
+        return f'"{str(p).replace(os.sep, "/")}"'
+
     # Extract the values
     build_src_dir = component.data.get("build", {}).get("srcDir", None)
     if not build_src_dir:
@@ -173,10 +181,10 @@ def generate_cmakelists_txt(component: IDFComponent) -> str:
     # Generate the component
     content = "idf_component_register(\n"
     if build_src_files:
-        str_srcs = " ".join([escape_entry(p) for p in sorted(build_src_files)])
+        str_srcs = " ".join([escape_path(p) for p in sorted(build_src_files)])
         content += f"  SRCS {str_srcs}\n"
     if build_include_dirs:
-        str_include_dirs = " ".join([escape_entry(p) for p in build_include_dirs])
+        str_include_dirs = " ".join([escape_path(p) for p in build_include_dirs])
         content += f"  INCLUDE_DIRS {str_include_dirs}\n"
     # Project-managed and built-in component lists are set per-project
     # via idf_build_set_property in the top-level CMakeLists; expanded
@@ -211,7 +219,7 @@ def generate_cmakelists_txt(component: IDFComponent) -> str:
     if link_directories:
         content += "target_link_directories(${COMPONENT_LIB} INTERFACE\n"
         for link_directory in link_directories:
-            str_build_flag = escape_entry(link_directory)
+            str_build_flag = escape_path(link_directory)
             content += f"  {str_build_flag}\n"
         content += ")\n"
 

esphome/git.py

@@ -1,5 +1,8 @@
-from collections.abc import Callable
+from collections.abc import Callable, Iterator
+from contextlib import contextmanager
 from dataclasses import dataclass
+from enum import Enum, auto
+import errno
 import hashlib
 import logging
 import os
@@ -8,23 +11,42 @@ import re
 import subprocess
 import sys
 import time
+from typing import TYPE_CHECKING
 import urllib.parse
 
 import esphome.config_validation as cv
 from esphome.core import CORE, EsphomeError, TimePeriodSeconds
 from esphome.helpers import add_git_ceiling_directory, rmtree, write_file
 
+if TYPE_CHECKING:
+    from filelock import FileLock
+
 _LOGGER = logging.getLogger(__name__)
 
 # Special value to indicate never refresh
 NEVER_REFRESH = TimePeriodSeconds(seconds=-1)
 
-# Written inside .git only after every clone step (clone, ref fetch, reset,
-# submodule init) has completed. A directory without it is an interrupted
-# clone (e.g. the process was killed mid-clone) and must be re-cloned; without
-# this check such a directory would be trusted forever when the caller uses
-# NEVER_REFRESH. Lives in .git so stash/reset/checkout can never touch it and
-# it does not pollute the worktree.
+# revert() runs on an already-failing path; bound its wait for the cache
+# entry lock so that recovery cannot hang forever behind another process.
+_REVERT_LOCK_TIMEOUT_SECONDS = 60
+
+# When a complete cache entry already exists, a caller does not wait forever
+# behind another process's stalled clone or update (git sets no network
+# timeouts): after this bound it uses the existing clone without refreshing
+# it. With no complete entry there is nothing to fall back to, so the wait
+# is unbounded.
+_COMPLETE_ENTRY_LOCK_TIMEOUT_SECONDS = 60
+
+# Written inside .git only while the entry is a complete, quiescent
+# checkout: after every clone step (clone, ref fetch, reset, submodule init)
+# has finished, and removed for the duration of a refresh's rewrite
+# (stash/fetch/reset). A directory without it is an interrupted clone or
+# update (e.g. the process was killed mid-clone) and must be re-cloned;
+# without this check such a directory would be trusted forever when the
+# caller uses NEVER_REFRESH, and the bounded-wait fallback would hand a
+# mid-rewrite tree to a timed-out peer. Lives in .git so
+# stash/reset/checkout can never touch it and it does not pollute the
+# worktree.
 _CLONE_COMPLETE_MARKER = "esphome_clone_complete"
 
 # Environment variables that scope git to a specific repository. Git hooks and
@@ -149,6 +171,16 @@ def run_git_command(
     return ret.stdout.decode("utf-8").strip()
 
 
+def _cache_key(url: str, ref: str | None) -> str:
+    """Cache key identifying one repository checkout.
+
+    The lock path and the entry directory both hash this, keeping them in
+    agreement. (micro_wake_word still rebuilds the format by hand to locate
+    manifests; fold it in here if the format ever changes.)
+    """
+    return f"{url}@{ref}"
+
+
 def _compute_destination_path(key: str, domain: str) -> Path:
     base_dir = Path(CORE.data_dir) / domain
     h = hashlib.new("sha256")
@@ -156,22 +188,195 @@ def _compute_destination_path(key: str, domain: str) -> Path:
     return base_dir / h.hexdigest()[:8]
 
 
+def _repo_entry_dir(key: str, domain: str, subpath: Path | None) -> Path:
+    """Worktree directory of one cache entry: the hash dir plus optional subpath."""
+    repo_dir = _compute_destination_path(key, domain)
+    if subpath:
+        repo_dir = repo_dir / subpath
+    return repo_dir
+
+
+def _repo_lock_path(key: str, domain: str) -> Path:
+    """Path of the lock file serializing all work on one cache entry.
+
+    Lives next to the hash directory, never inside it, so the removal of a
+    broken or incomplete clone can never delete a lock another process holds.
+    """
+    repo_dir = _compute_destination_path(key, domain)
+    return repo_dir.parent / f"{repo_dir.name}.lock"
+
+
+class _LockStatus(Enum):
+    ACQUIRED = auto()
+    # A bounded wait expired while another process held the lock.
+    TIMEOUT = auto()
+    # The lock could not be taken at all; callers proceed unlocked,
+    # matching the behavior before the lock existed.
+    UNAVAILABLE = auto()
+
+
+# Errnos that mean the filesystem genuinely cannot take file locks (NFS
+# without a lock daemon, some FUSE mounts). Any other OSError (permissions,
+# read-only volume, full disk) is a cache directory problem, which the git
+# commands themselves report clearly when it actually matters. EPERM is
+# deliberately absent: it usually means a permissions problem, so it takes
+# the generic message that names no cause. On Linux ENOTSUP and EOPNOTSUPP
+# are the same value; the set folds them.
+_NO_LOCK_SUPPORT_ERRNOS = frozenset(
+    {errno.ENOLCK, errno.ENOSYS, errno.EOPNOTSUPP, errno.ENOTSUP}
+)
+
+
+def _acquire_repo_lock(
+    lock: "FileLock",
+    safe_key: str,
+    timeout: float,
+    wait_message: str = "Waiting for another process to finish updating %s",
+) -> _LockStatus:
+    """Acquire ``lock``, logging ``wait_message`` when a wait actually begins.
+
+    ``timeout`` of -1 waits forever; a positive value bounds the wait and
+    can yield ``TIMEOUT``.
+    """
+    from filelock import Timeout
+
+    try:
+        try:
+            lock.acquire(blocking=False)
+        except Timeout:
+            # Waiting on another process's clone or update can take
+            # minutes; say so instead of appearing hung.
+            _LOGGER.info(wait_message, safe_key)
+            lock.acquire(timeout=timeout)
+    except Timeout:
+        return _LockStatus.TIMEOUT
+    except OSError as err:
+        if err.errno in _NO_LOCK_SUPPORT_ERRNOS:
+            _LOGGER.warning(
+                "The filesystem does not support locking the cache entry for "
+                "%s (%s), continuing without a lock",
+                safe_key,
+                err,
+            )
+        else:
+            # Not a locking problem (permissions, read-only volume, full
+            # disk). Still continue unlocked: a pre-seeded read-only cache
+            # with refresh disabled only reads and must keep working, and
+            # in every other case the git commands fail with the real error.
+            _LOGGER.warning(
+                "Could not take the cache entry lock for %s (%s), "
+                "continuing without a lock",
+                safe_key,
+                err,
+            )
+        return _LockStatus.UNAVAILABLE
+    return _LockStatus.ACQUIRED
+
+
+@contextmanager
+def _repo_cache_lock(
+    key: str, domain: str, repo_dir: Path
+) -> Iterator[tuple[bool, "FileLock | None"]]:
+    """Hold the cache entry lock for ``key`` over the with block.
+
+    Yields ``(use_existing, lock)``. ``use_existing`` is True when the lock
+    could not be acquired within the bounded wait but ``repo_dir`` is a
+    complete cache entry; the caller should use it as-is and do nothing
+    else. Otherwise ``lock`` is the held lock, released when the block
+    exits, or ``None`` when the lock could not be taken at all and the
+    caller proceeds unlocked.
+    """
+    # Lazy import: keeps filelock off the CLI startup import path.
+    from filelock import FileLock
+
+    safe_key = _redact_url_credentials(key)
+    # acquire() creates the lock file's directory itself; git clone later
+    # creates the hash directory next to it. fallback_to_soft would silently
+    # downgrade ENOSYS to a SoftFileLock, whose stale existence marker from
+    # another host on a shared cache could hang the unbounded wait forever;
+    # routing it through the OSError handler runs unlocked instead.
+    lock: FileLock | None = FileLock(
+        str(_repo_lock_path(key, domain)), fallback_to_soft=False
+    )
+    status = _acquire_repo_lock(lock, safe_key, _COMPLETE_ENTRY_LOCK_TIMEOUT_SECONDS)
+    if status is _LockStatus.TIMEOUT:
+        if _clone_complete_marker_path(repo_dir).is_file():
+            # Mutual exclusion matters most while no complete entry exists
+            # (initial clone, recovery re-clone); with one on disk, reading
+            # it beats hanging behind a stalled holder.
+            _LOGGER.warning(
+                "Timed out waiting for another process updating %s, proceeding "
+                "with the existing clone, which that process may still be "
+                "changing",
+                safe_key,
+            )
+            yield True, None
+            return
+        # Nothing to fall back to; the holder is producing the clone this
+        # caller needs.
+        status = _acquire_repo_lock(
+            lock,
+            safe_key,
+            timeout=-1,
+            wait_message="Still waiting for the clone of %s, "
+            "there is no existing clone to fall back on",
+        )
+    if status is not _LockStatus.ACQUIRED:
+        lock = None
+    try:
+        yield False, lock
+    finally:
+        if lock is not None:
+            lock.release()
+
+
 def _clone_complete_marker_path(repo_dir: Path) -> Path:
     return repo_dir / ".git" / _CLONE_COMPLETE_MARKER
 
 
+def _clear_clone_complete_marker(repo_dir: Path) -> None:
+    """Best-effort removal of the completion marker.
+
+    If the unlink fails (e.g. a file lock on Windows), the marker stays and
+    the entry keeps its previous trust level; every consumer of the marker
+    tolerates that.
+    """
+    try:
+        _clone_complete_marker_path(repo_dir).unlink(missing_ok=True)
+    except OSError as err:
+        _LOGGER.debug("Could not delete clone completion marker: %s", err)
+
+
+def _write_clone_complete_marker(
+    repo_dir: Path, key: str, hash_dir_name: str, safe_key: str
+) -> None:
+    """Mark the entry as a complete, quiescent checkout.
+
+    The key and hash dir name are recorded purely to make cache debugging
+    easier. The marker is only a validity signal, so a failed write must not
+    fail an otherwise complete clone or update: the only cost is a re-clone
+    on the next run.
+    """
+    try:
+        write_file(
+            _clone_complete_marker_path(repo_dir),
+            f"key={key}\nhash={hash_dir_name}\n",
+        )
+    except EsphomeError as err:
+        _LOGGER.warning(
+            "Could not write clone completion marker for %s: %s", safe_key, err
+        )
+
+
 def _remove_repo_dir(repo_dir: Path) -> None:
     """Remove a repo directory, deleting the completion marker first.
 
     Marker-first ordering guarantees an interrupted removal can never leave a
     marker behind next to a partially deleted worktree. The unlink is best
-    effort: if it fails (e.g. a file lock on Windows), rmtree below still
-    gets the chance to remove the directory, marker included.
+    effort: if it fails, rmtree below still gets the chance to remove the
+    directory, marker included.
     """
-    try:
-        _clone_complete_marker_path(repo_dir).unlink(missing_ok=True)
-    except OSError as err:
-        _LOGGER.debug("Could not delete clone completion marker first: %s", err)
+    _clear_clone_complete_marker(repo_dir)
     if repo_dir.is_dir():
         rmtree(repo_dir)
 
@@ -286,16 +491,79 @@ def resolve_symlink_stub(repo_dir: Path, file_path: Path) -> Path | None:
 def clone_or_update(
     *,
     url: str,
-    ref: str = None,
+    ref: str | None = None,
     refresh: TimePeriodSeconds | None,
     domain: str,
-    username: str = None,
-    password: str = None,
+    username: str | None = None,
+    password: str | None = None,
     init_submodules: bool = False,
     subpath: Path | None = None,
+) -> tuple[Path, Callable[[], bool] | None]:
+    """Clone a repository into the cache, or refresh an existing clone.
+
+    All work runs under a per-cache-entry inter-process file lock, so
+    concurrent resolutions of the same repository (two esphome processes, or
+    a subprocess plus an in-process load) serialize instead of interleaving.
+    Without the lock, ``repo_dir.is_dir()`` is true from the instant
+    ``git clone`` creates the directory: a second caller could read a half
+    populated worktree, or see the missing completion marker and delete the
+    clone in progress out from under the first caller.
+
+    The lock guards mutation of the cache entry only; it is released when
+    this function returns, so a caller still reading the worktree can
+    overlap a later refresh by another process. That residual window is
+    narrow (the refresh interval is re-checked under the lock) and predates
+    the lock.
+
+    Locking is best effort: on a filesystem that cannot take file locks a
+    warning is logged and the work proceeds unlocked, matching the behavior
+    before the lock existed. A complete cache entry also caps the wait: if
+    the holder is still busy after a bounded time (e.g. stalled on the
+    network), the existing clone is used without refreshing it, so a stuck
+    process cannot hang every peer that already has a good entry.
+    """
+    key = _cache_key(url, ref)
+    repo_dir = _repo_entry_dir(key, domain, subpath)
+    with _repo_cache_lock(key, domain, repo_dir) as (use_existing, lock):
+        if use_existing:
+            return repo_dir, None
+        return _clone_or_update_locked(
+            url=url,
+            ref=ref,
+            refresh=refresh,
+            domain=domain,
+            username=username,
+            password=password,
+            init_submodules=init_submodules,
+            subpath=subpath,
+            lock=lock,
+        )
+
+
+def _clone_or_update_locked(
+    *,
+    url: str,
+    ref: str | None,
+    refresh: TimePeriodSeconds | None,
+    domain: str,
+    username: str | None,
+    password: str | None,
+    init_submodules: bool,
+    subpath: Path | None,
+    lock: "FileLock | None",
     _recover_broken: bool = True,
-) -> tuple[Path, Callable[[], None] | None]:
-    key = f"{url}@{ref}"
+) -> tuple[Path, Callable[[], bool] | None]:
+    """Body of ``clone_or_update``; the caller holds ``lock``.
+
+    Split out because the broken-repository recovery below re-enters this
+    function: re-acquiring the already-held lock would deadlock, since OS
+    file locks taken on separate file descriptors conflict even within one
+    process. ``lock`` is only re-acquired by the returned ``revert``
+    callback, which runs after the wrapper's ``finally`` has released it.
+    ``lock`` is ``None`` when the filesystem cannot take file locks and the
+    wrapper fell back to running unlocked.
+    """
+    key = _cache_key(url, ref)
     # The user may have embedded credentials in the URL itself; log this
     # instead of key.
     safe_key = _redact_url_credentials(key)
@@ -309,10 +577,8 @@ def clone_or_update(
             "://", f"://{urllib.parse.quote(username)}:{urllib.parse.quote(password)}@"
         )
 
-    repo_dir = _compute_destination_path(key, domain)
-    hash_dir_name = repo_dir.name
-    if subpath:
-        repo_dir = repo_dir / subpath
+    hash_dir_name = _compute_destination_path(key, domain).name
+    repo_dir = _repo_entry_dir(key, domain, subpath)
 
     if repo_dir.is_dir() and not _clone_complete_marker_path(repo_dir).is_file():
         # The last clone never finished (killed process, container stop) or
@@ -353,19 +619,8 @@ def clone_or_update(
             _remove_repo_dir(repo_dir)
             raise
 
-        # Every git step succeeded; the key and hash dir name are recorded
-        # purely to make cache debugging easier. The marker is only a
-        # validity signal, so a failed write must not fail an otherwise
-        # complete clone: the only cost is a re-clone on the next run.
-        try:
-            write_file(
-                _clone_complete_marker_path(repo_dir),
-                f"key={key}\nhash={hash_dir_name}\n",
-            )
-        except EsphomeError as err:
-            _LOGGER.warning(
-                "Could not write clone completion marker for %s: %s", safe_key, err
-            )
+        # Every git step succeeded.
+        _write_clone_complete_marker(repo_dir, key, hash_dir_name, safe_key)
 
     else:
         if refresh == NEVER_REFRESH or CORE.skip_external_update:
@@ -396,6 +651,13 @@ def clone_or_update(
                 _LOGGER.info("Updating %s", safe_key)
                 _LOGGER.debug("Location: %s", repo_dir)
 
+                # The entry is about to be rewritten; drop the marker so a
+                # timed-out peer's fallback and the incomplete-entry check
+                # can tell a quiescent complete entry from one mid-rewrite,
+                # and so an update interrupted by a crash re-clones instead
+                # of being trusted.
+                _clear_clone_complete_marker(repo_dir)
+
                 # Stash local changes (if any)
                 # Use git_dir to ensure this only affects the specific repo
                 run_git_command(
@@ -425,6 +687,15 @@ def clone_or_update(
                 # refresh window would silently accept on the next run.
                 if init_submodules:
                     update_submodules(repo_dir, key)
+
+                # Recorded so revert() can tell whether the checkout is
+                # still the one this update produced.
+                new_sha = run_git_command(
+                    ["git", "rev-parse", "HEAD"], git_dir=repo_dir
+                )
+
+                # The rewrite finished; the entry is trustworthy again.
+                _write_clone_complete_marker(repo_dir, key, hash_dir_name, safe_key)
             except GitException as err:
                 # Repository is in a broken state or update failed
                 # Only attempt recovery once to prevent infinite recursion
@@ -444,9 +715,10 @@ def clone_or_update(
                 _remove_repo_dir(repo_dir)
                 _LOGGER.info("Successfully removed broken repository, re-cloning...")
 
-                # Recursively call clone_or_update to re-clone
-                # Set _recover_broken=False to prevent infinite recursion
-                result = clone_or_update(
+                # Re-clone while still holding the lock; going through the
+                # public wrapper would try to re-acquire it and deadlock.
+                # Set _recover_broken=False to prevent infinite recursion.
+                result = _clone_or_update_locked(
                     url=original_url,
                     ref=ref,
                     refresh=refresh,
@@ -455,14 +727,80 @@ def clone_or_update(
                     password=password,
                     init_submodules=init_submodules,
                     subpath=subpath,
+                    lock=lock,
                     _recover_broken=False,
                 )
                 _LOGGER.info("Repository %s successfully recovered", safe_key)
                 return result
 
-            def revert():
-                _LOGGER.info("Reverting changes to %s -> %s", safe_key, old_sha)
-                run_git_command(["git", "reset", "--hard", old_sha], git_dir=repo_dir)
+            def revert() -> bool:
+                """Reset the checkout to the pre-update SHA.
+
+                Returns False when the revert did not happen: the cache
+                entry lock could not be acquired in time, the checkout
+                moved since this update (another process refreshed it), or
+                the reset itself failed. A retry cannot reach the
+                pre-update content then.
+                """
+                if lock is None:
+                    # The wrapper already warned about the unlockable
+                    # filesystem; revert unlocked like everything else.
+                    status = _LockStatus.UNAVAILABLE
+                else:
+                    status = _acquire_repo_lock(
+                        lock, safe_key, _REVERT_LOCK_TIMEOUT_SECONDS
+                    )
+                if status is _LockStatus.TIMEOUT:
+                    # revert() only runs on an already-failing path; skip
+                    # rather than hang so the original error can surface.
+                    _LOGGER.warning(
+                        "Could not lock %s to revert to %s, skipping revert; "
+                        "the cached checkout keeps the un-reverted content "
+                        "until its next refresh",
+                        safe_key,
+                        old_sha,
+                    )
+                    return False
+                try:
+                    # Anything can happen between the wrapper releasing the
+                    # lock and revert() re-acquiring it; only undo this
+                    # process's own update, never a peer's newer refresh.
+                    head = run_git_command(
+                        ["git", "rev-parse", "HEAD"], git_dir=repo_dir
+                    )
+                    if head != new_sha:
+                        _LOGGER.warning(
+                            "Not reverting %s: the checkout moved since this "
+                            "update (another process refreshed it)",
+                            safe_key,
+                        )
+                        return False
+                    # Announced only once every skip check has passed, so
+                    # the log says exactly one thing per outcome.
+                    _LOGGER.info("Reverting changes to %s -> %s", safe_key, old_sha)
+                    run_git_command(
+                        ["git", "reset", "--hard", old_sha], git_dir=repo_dir
+                    )
+                except GitException as err:
+                    # GitException is a cv.Invalid; letting it escape would
+                    # replace the caller's original error with a bare git
+                    # message. Report the failed reset like the skip above,
+                    # and drop the marker: an entry whose reset fails cannot
+                    # be trusted, so the next use re-clones it instead of
+                    # the refresh window silently accepting it.
+                    _LOGGER.warning(
+                        "Could not revert %s to %s (%s), the entry will be "
+                        "re-cloned on next use",
+                        safe_key,
+                        old_sha,
+                        err,
+                    )
+                    _clear_clone_complete_marker(repo_dir)
+                    return False
+                finally:
+                    if status is _LockStatus.ACQUIRED:
+                        lock.release()
+                return True
 
             return repo_dir, revert
 

requirements.txt

@@ -26,8 +26,8 @@ bleak==2.1.1
 smpclient==7.2.0
 requests==2.34.2
 py7zr==1.1.3
-platformdirs==4.10.0  # native esp-idf toolchain global cache dir
-filelock==3.29.0  # lock guarding the PlatformIO python-version cache heal
+platformdirs==4.11.0  # native esp-idf toolchain global cache dir
+filelock==3.32.0  # inter-process locks (PlatformIO cache heal, git clone cache); >=3.32 for FileLock(fallback_to_soft=...), older versions silently drop the kwarg
 
 # esp-idf >= 5.0 requires this
 pyparsing >= 3.3.2

tests/component_tests/epaper_spi/config/t133a01_no_init_sequence.yaml

@@ -0,0 +1,26 @@
+esphome:
+  name: test
+
+esp32:
+  board: esp32-s3-devkitc-1
+  variant: esp32s3
+  framework:
+    type: esp-idf
+
+spi:
+  clk_pin: GPIO18
+  mosi_pin: GPIO19
+
+display:
+  - platform: epaper_spi
+    id: epaper_display
+    model: t133a01
+    dc_pin: GPIO21
+    reset_pin: GPIO38
+    cs_pin: GPIO10
+    cs1_pin: GPIO2
+    busy_pin: GPIO13
+    update_interval: never
+    dimensions:
+      width: 200
+      height: 200

tests/component_tests/epaper_spi/test_init.py

@@ -462,3 +462,24 @@ def test_enable_pin_code_generation(
     # Both pin objects must be passed to the display via set_enable_pins() as a
     # std::vector initializer list, in the configured order.
     assert f"set_enable_pins({{{pin_25}, {pin_26}}});" in main_cpp
+
+
+def test_model_with_no_default_init_sequence_generates(
+    generate_main: Callable[[str | Path], str],
+    component_config_path: Callable[[str], Path],
+) -> None:
+    """Test that code generation succeeds for a model with no default init sequence.
+
+    The base "t133a01" model (used directly, not via one of its `.extend()`
+    variants) doesn't override `get_init_sequence()` or pass `initsequence` to
+    its constructor, and the user didn't supply `init_sequence:` either.
+    `EpaperModel.get_init_sequence()` used to default to `None` in this case,
+    which made `flatten_sequence()` raise a `TypeError` during code
+    generation. Regression test for that crash.
+    """
+    main_cpp = generate_main(component_config_path("t133a01_no_init_sequence.yaml"))
+
+    # The generated constructor call takes (name, width, height, init_sequence,
+    # init_sequence_length, ...); a length of 0 confirms the empty init
+    # sequence array was generated instead of raising during code generation.
+    assert re.search(r"epaper_spi::EPaperT133A01\([^;]*,\s*\w+,\s*0\);", main_cpp)

tests/component_tests/esp32/test_esp32.py

@@ -108,6 +108,24 @@ def test_esp32_default_toolchain_is_esp_idf(
     assert CORE.toolchain == expected
 
 
+@pytest.mark.parametrize(
+    "config_toolchain",
+    [Toolchain.SDK_NRF.value, "nonsense"],
+)
+def test_esp32_rejects_unsupported_toolchains(
+    set_core_config: SetCoreConfigCallable,
+    config_toolchain: str,
+) -> None:
+    """Toolchains esp32 does not support are rejected at validation time."""
+    set_core_config(PlatformFramework.ESP32_IDF)
+
+    from esphome.components.esp32 import CONFIG_SCHEMA
+
+    CORE.toolchain = None
+    with pytest.raises(cv.Invalid, match="Unknown value"):
+        CONFIG_SCHEMA({"variant": VARIANT_ESP32, "toolchain": config_toolchain})
+
+
 @pytest.mark.parametrize(
     ("config", "error_match"),
     [
@@ -454,26 +472,18 @@ def test_flash_mode_unset_leaves_defaults(
         ),
         pytest.param(
             PlatformFramework.ESP32_IDF,
-            NetworkSdkconfigData(
-                wifi=True, bluetooth=True, ble_42=True, software_coexistence=True
-            ),
+            NetworkSdkconfigData(wifi=True, bluetooth=True, software_coexistence=True),
             {},
             {
                 "CONFIG_BT_ENABLED": True,
                 "CONFIG_BT_BLE_42_FEATURES_SUPPORTED": True,
+                "CONFIG_BT_BLE_50_FEATURES_SUPPORTED": False,
                 "CONFIG_SW_COEXIST_ENABLE": True,
                 "CONFIG_ESP_WIFI_SOFTAP_SUPPORT": False,
                 "CONFIG_LWIP_DHCPS": False,
             },
             id="idf_wifi_ble_tracker_coexistence",
         ),
-        pytest.param(
-            PlatformFramework.ESP32_IDF,
-            NetworkSdkconfigData(bluetooth=True),
-            {},
-            {"CONFIG_BT_ENABLED": True},
-            id="idf_ble_server_only_no_ble42",
-        ),
         # --- IDF: user sdkconfig_options always win ---
         pytest.param(
             PlatformFramework.ESP32_IDF,
@@ -594,6 +604,7 @@ def test_network_wifi_ble_coexistence_reconciles_end_to_end(
     sdkconfig = CORE.data[KEY_ESP32][KEY_SDKCONFIG_OPTIONS]
     assert sdkconfig.get("CONFIG_BT_ENABLED") is True
     assert sdkconfig.get("CONFIG_BT_BLE_42_FEATURES_SUPPORTED") is True
+    assert sdkconfig.get("CONFIG_BT_BLE_50_FEATURES_SUPPORTED") is False
     assert sdkconfig.get("CONFIG_SW_COEXIST_ENABLE") is True
     assert sdkconfig.get("CONFIG_ESP_WIFI_SOFTAP_SUPPORT") is False
     assert sdkconfig.get("CONFIG_LWIP_DHCPS") is False

tests/component_tests/packages/test_packages.py

@@ -1264,6 +1264,75 @@ def test_remote_packages_no_revert(
     ]
 
 
+@patch("esphome.yaml_util.load_yaml")
+@patch("pathlib.Path.is_file")
+@patch("esphome.git.clone_or_update")
+def test_remote_packages_skipped_revert_does_not_retry(
+    mock_clone_or_update, mock_is_file, mock_load_yaml
+) -> None:
+    """When revert() reports the rollback was skipped, the load is not
+    retried (the checkout is unchanged) and the error says so."""
+    mock_revert = MagicMock(return_value=False)
+    mock_clone_or_update.return_value = (Path("/tmp/noexists"), mock_revert)
+    mock_is_file.return_value = True
+    mock_load_yaml.side_effect = cv.Invalid("bad yaml")
+
+    config = {
+        CONF_PACKAGES: {
+            "pkg": {
+                CONF_URL: "https://github.com/esphome/repo",
+                CONF_REF: "main",
+                CONF_FILES: [{CONF_PATH: "file.yaml"}],
+                CONF_REFRESH: "1d",
+            }
+        }
+    }
+    with pytest.raises(cv.Invalid, match="could not revert the cached checkout"):
+        packages_pass(config)
+
+    assert mock_revert.call_count == 1
+    assert mock_load_yaml.call_count == 1
+
+
+@patch("esphome.yaml_util.load_yaml")
+@patch("pathlib.Path.is_file")
+@patch("esphome.git.clone_or_update")
+def test_remote_packages_successful_revert_retries(
+    mock_clone_or_update, mock_is_file, mock_load_yaml, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A successful revert retries the load against the reverted checkout and
+    logs the original error, the only trace that upstream was broken."""
+    mock_revert = MagicMock(return_value=True)
+    mock_clone_or_update.return_value = (Path("/tmp/noexists"), mock_revert)
+    mock_is_file.return_value = True
+    mock_load_yaml.side_effect = [
+        cv.Invalid("bad yaml"),
+        OrderedDict(
+            {CONF_SENSOR: [{CONF_PLATFORM: TEST_SENSOR_PLATFORM_1, CONF_NAME: "test"}]}
+        ),
+    ]
+
+    config = {
+        CONF_PACKAGES: {
+            "pkg": {
+                CONF_URL: "https://github.com/esphome/repo",
+                CONF_REF: "main",
+                CONF_FILES: [{CONF_PATH: "file.yaml"}],
+                CONF_REFRESH: "1d",
+            }
+        }
+    }
+    with caplog.at_level(logging.WARNING):
+        actual = packages_pass(config)
+
+    assert actual[CONF_SENSOR] == [
+        {CONF_PLATFORM: TEST_SENSOR_PLATFORM_1, CONF_NAME: "test"}
+    ]
+    assert mock_revert.call_count == 1
+    assert mock_load_yaml.call_count == 2
+    assert any("reverted the cached checkout" in r.getMessage() for r in caplog.records)
+
+
 def test_raw_config_contains_merged_esphome_from_package(tmp_path) -> None:
     """Test that CORE.raw_config contains esphome section from merged package.
 

tests/unit_tests/test_espidf_component.py

@@ -169,30 +169,57 @@ def test_generate_cmakelists_txt_with_flags(tmp_component, tmp_path):
     }
 
     content = generate_cmakelists_txt(tmp_component)
-    sep = "\\\\" if os.name == "nt" else "/"
+    # Paths are always emitted with forward slashes so the CMakeLists is
+    # portable; on Windows os.path.relpath would otherwise yield backslashes
+    # that break CMake's list re-parsing.
     assert (
         content
-        == f"""idf_component_register(
-  SRCS "src{sep}main.c"
+        == """idf_component_register(
+  SRCS "src/main.c"
   INCLUDE_DIRS "src"
-  REQUIRES dep ${{ESPHOME_PROJECT_MANAGED_COMPONENTS}} ${{ESPHOME_PROJECT_BUILTIN_COMPONENTS}}
+  REQUIRES dep ${ESPHOME_PROJECT_MANAGED_COMPONENTS} ${ESPHOME_PROJECT_BUILTIN_COMPONENTS}
 )
-target_compile_options(${{COMPONENT_LIB}} PUBLIC
+target_compile_options(${COMPONENT_LIB} PUBLIC
   "-DTEST"
 )
-target_compile_options(${{COMPONENT_LIB}} PRIVATE
+target_compile_options(${COMPONENT_LIB} PRIVATE
   "-Wall"
 )
-target_link_directories(${{COMPONENT_LIB}} INTERFACE
+target_link_directories(${COMPONENT_LIB} INTERFACE
   "lib"
 )
-target_link_libraries(${{COMPONENT_LIB}} INTERFACE
+target_link_libraries(${COMPONENT_LIB} INTERFACE
   "mylib"
 )
 """
     )
 
 
+def test_generate_cmakelists_txt_uses_forward_slashes_on_windows(
+    tmp_component, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # os.path.relpath yields backslash paths on Windows, which CMake rejects
+    # when it re-parses the SRCS list (e.g. "\b" in "src\backend" is an invalid
+    # character escape). Simulate that output and confirm the generated
+    # CMakeLists normalizes the separators to forward slashes.
+    src_dir = tmp_component.path / "src" / "backend"
+    src_dir.mkdir(parents=True)
+    (src_dir / "cipher.c").write_text("int f() {}")
+
+    tmp_component.data = {}
+
+    monkeypatch.setattr("esphome.espidf.component.os.sep", "\\")
+    monkeypatch.setattr(
+        "esphome.espidf.component.os.path.relpath",
+        lambda *args, **kwargs: "src\\backend\\cipher.c",
+    )
+
+    content = generate_cmakelists_txt(tmp_component)
+
+    assert 'SRCS "src/backend/cipher.c"' in content
+    assert "\\" not in content
+
+
 def test_generate_cmakelists_txt_multi_token_flag(tmp_component):
     # PlatformIO shell-lexes each build.flags entry, so a single entry can
     # carry a flag and its argument. The generated CMakeLists must emit them

tests/unit_tests/test_git.py

@@ -1,14 +1,17 @@
 """Tests for git.py module."""
 
 from collections.abc import Callable
+import errno
 import logging
 import os
 from pathlib import Path
 import subprocess
+import threading
 import time
 from typing import Any
 from unittest.mock import Mock, patch
 
+from filelock import FileLock
 import pytest
 
 from esphome import git
@@ -74,17 +77,23 @@ def _simulate_cloned_repo(repo_dir: Path) -> None:
 
 
 def _make_clone_side_effect(
-    repo_dir: Path, gitmodules: bool = False
+    repo_dir: Path,
+    gitmodules: bool = False,
+    on_clone: Callable[[], None] | None = None,
 ) -> Callable[..., str]:
     """Return a run_git_command side effect whose clone creates the repo dir.
 
-    With ``gitmodules`` the cloned repo also declares submodules.
+    With ``gitmodules`` the cloned repo also declares submodules. ``on_clone``
+    runs at clone time before the repo dir appears, so a test can probe or
+    block mid-clone.
     """
 
     def git_command_side_effect(
         cmd: list[str], cwd: str | None = None, **kwargs: Any
     ) -> str:
         if _get_git_command_type(cmd) == "clone":
+            if on_clone is not None:
+                on_clone()
             _simulate_cloned_repo(repo_dir)
             if gitmodules:
                 (repo_dir / ".gitmodules").write_text("test")
@@ -1491,6 +1500,591 @@ def test_refresh_submodule_failure_recovers_then_raises(
     )
 
 
+def _lock_path(url: str, ref: str | None, domain: str) -> Path:
+    """The lock file the implementation uses for one cache entry."""
+    return git._repo_lock_path(git._cache_key(url, ref), domain)
+
+
+class _SetEventOnWaitLog(logging.Handler):
+    """Set an event when the 'Waiting for another process' record is emitted,
+    so tests can react to a caller observably blocking on the lock instead of
+    racing a wall-clock timer."""
+
+    def __init__(self, event: threading.Event) -> None:
+        super().__init__()
+        self._event = event
+
+    def emit(self, record: logging.LogRecord) -> None:
+        if "Waiting for another process" in record.getMessage():
+            self._event.set()
+
+
+def test_clone_or_update_serializes_concurrent_clones(
+    tmp_path: Path, mock_run_git_command: Mock, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Two concurrent callers for the same uncached repo must not both clone.
+
+    Without the per-entry lock both callers pass the is_dir() check before
+    either clone finishes, so the second one either clones on top of the
+    first or reads a half populated worktree (device-builder issue 2425).
+    """
+    CORE.config_path = tmp_path / "test.yaml"
+
+    url = "https://github.com/test/repo"
+    domain = "test"
+    repo_dir = _compute_repo_dir(url, None, domain)
+
+    start_together = threading.Barrier(2)
+    other_caller_waiting = threading.Event()
+
+    def on_clone() -> None:
+        # Hold the lock until the other caller is observably blocked on it,
+        # so the interleaving is guaranteed rather than raced on a timer.
+        assert other_caller_waiting.wait(timeout=30)
+
+    mock_run_git_command.side_effect = _make_clone_side_effect(
+        repo_dir, on_clone=on_clone
+    )
+
+    results: list[Path] = []
+    errors: list[BaseException] = []
+
+    def call() -> None:
+        try:
+            start_together.wait()
+            result_dir, _ = git.clone_or_update(
+                url=url, ref=None, refresh=git.NEVER_REFRESH, domain=domain
+            )
+            results.append(result_dir)
+        except BaseException as err:  # noqa: BLE001 - re-raised via errors below
+            errors.append(err)
+
+    handler = _SetEventOnWaitLog(other_caller_waiting)
+    git_logger = logging.getLogger("esphome.git")
+    git_logger.addHandler(handler)
+    try:
+        with caplog.at_level(logging.INFO):
+            threads = [threading.Thread(target=call) for _ in range(2)]
+            for t in threads:
+                t.start()
+            for t in threads:
+                t.join()
+    finally:
+        git_logger.removeHandler(handler)
+
+    assert not errors
+    assert results == [repo_dir, repo_dir]
+    clone_calls = [
+        c
+        for c in mock_run_git_command.call_args_list
+        if _get_git_command_type(c[0][0]) == "clone"
+    ]
+    # The second caller waited for the lock, then saw the completed clone.
+    assert len(clone_calls) == 1
+    assert _marker_path(repo_dir).is_file()
+
+
+def test_clone_or_update_creates_lock_file_next_to_hash_dir(
+    tmp_path: Path, mock_run_git_command: Mock
+) -> None:
+    """The lock file lives beside the hash dir, never inside it.
+
+    Checked while the clone runs (the lock is held): filelock's Windows
+    backend deletes the lock file on release, so probing after the call
+    would only work on Unix.
+    """
+    CORE.config_path = tmp_path / "test.yaml"
+
+    url = "https://github.com/test/repo"
+    domain = "test"
+    repo_dir = _compute_repo_dir(url, None, domain)
+    lock_path = _lock_path(url, None, domain)
+    lock_held_during_clone: list[bool] = []
+
+    mock_run_git_command.side_effect = _make_clone_side_effect(
+        repo_dir, on_clone=lambda: lock_held_during_clone.append(lock_path.is_file())
+    )
+
+    result_dir, _ = git.clone_or_update(
+        url=url, ref=None, refresh=git.NEVER_REFRESH, domain=domain
+    )
+
+    assert result_dir == repo_dir
+    assert lock_path.parent == repo_dir.parent
+    assert lock_held_during_clone == [True]
+
+
+def test_clone_or_update_subpath_locks_at_hash_dir_level(
+    tmp_path: Path, mock_run_git_command: Mock
+) -> None:
+    """With a subpath the lock still guards the whole hash dir cache entry."""
+    CORE.config_path = tmp_path / "test.yaml"
+
+    url = "https://github.com/test/repo"
+    domain = "test"
+    hash_dir = _compute_repo_dir(url, None, domain)
+    repo_dir = hash_dir / "lib"
+    lock_path = _lock_path(url, None, domain)
+    lock_held_during_clone: list[bool] = []
+
+    mock_run_git_command.side_effect = _make_clone_side_effect(
+        repo_dir, on_clone=lambda: lock_held_during_clone.append(lock_path.is_file())
+    )
+
+    result_dir, _ = git.clone_or_update(
+        url=url,
+        ref=None,
+        refresh=git.NEVER_REFRESH,
+        domain=domain,
+        subpath=Path("lib"),
+    )
+
+    assert result_dir == repo_dir
+    assert lock_path == hash_dir.parent / f"{hash_dir.name}.lock"
+    assert lock_held_during_clone == [True]
+
+
+def test_clone_or_update_recovery_holds_lock_without_deadlock(
+    tmp_path: Path, mock_run_git_command: Mock
+) -> None:
+    """Recovery re-clones while holding the lock and must not re-acquire it.
+
+    A naive re-acquisition would deadlock here, since OS file locks taken on
+    separate descriptors conflict even within one process. The lock file
+    itself must survive the recovery rmtree of the broken repo dir: the
+    re-clone runs after the rmtree, so probing the lock file there proves
+    it. Probing after the call would only work on Unix, since filelock's
+    Windows backend deletes the lock file on release.
+    """
+    CORE.config_path = tmp_path / "test.yaml"
+
+    url = "https://github.com/test/repo"
+    domain = "test"
+    repo_dir = _compute_repo_dir(url, None, domain)
+    lock_path = _lock_path(url, None, domain)
+    lock_held_during_reclone: list[bool] = []
+
+    _setup_old_repo(repo_dir)
+
+    def git_command_side_effect(
+        cmd: list[str], cwd: str | None = None, **kwargs: Any
+    ) -> str:
+        if _get_git_command_type(cmd) == "stash":
+            raise git.GitCommandError("broken repository")
+        if _get_git_command_type(cmd) == "clone":
+            lock_held_during_reclone.append(lock_path.is_file())
+            _simulate_cloned_repo(repo_dir)
+        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
+    )
+
+    assert recovered_dir == repo_dir
+    assert lock_held_during_reclone == [True]
+
+
+def _hold_lock_in_thread(lock_path: Path) -> tuple[threading.Thread, threading.Event]:
+    """Hold the lock from another thread; returns the thread and its release event.
+
+    OS file locks taken on separate descriptors conflict even within one
+    process, so a second FileLock instance in a thread contends the same
+    way another process would.
+    """
+    held = threading.Event()
+    release = threading.Event()
+
+    def hold() -> None:
+        with FileLock(str(lock_path)):
+            held.set()
+            release.wait(timeout=30)
+
+    holder = threading.Thread(target=hold)
+    holder.start()
+    assert held.wait(timeout=30)
+    return holder, release
+
+
+def test_clone_or_update_logs_wait_on_contended_lock(
+    tmp_path: Path, mock_run_git_command: Mock, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A contended acquire logs a redacted waiting message instead of
+    silently blocking."""
+    CORE.config_path = tmp_path / "test.yaml"
+
+    url = "https://user:hunter2@github.com/test/repo"
+    domain = "test"
+    repo_dir = _compute_repo_dir(url, None, domain)
+    mock_run_git_command.side_effect = _make_clone_side_effect(repo_dir)
+
+    holder, release = _hold_lock_in_thread(_lock_path(url, None, domain))
+
+    # Free the lock only once the waiting message has been emitted, so the
+    # release is caused by the thing being asserted instead of racing it.
+    handler = _SetEventOnWaitLog(release)
+    git_logger = logging.getLogger("esphome.git")
+    git_logger.addHandler(handler)
+    try:
+        with caplog.at_level(logging.INFO):
+            result_dir, _ = git.clone_or_update(
+                url=url, ref=None, refresh=git.NEVER_REFRESH, domain=domain
+            )
+    finally:
+        git_logger.removeHandler(handler)
+        release.set()
+    holder.join()
+
+    assert result_dir == repo_dir
+    waiting = [
+        r.getMessage()
+        for r in caplog.records
+        if "Waiting for another process" in r.getMessage()
+    ]
+    assert len(waiting) == 1
+    assert "hunter2" not in waiting[0]
+    assert "://***@" in waiting[0]
+
+
+def test_revert_skips_on_contended_lock(
+    tmp_path: Path,
+    mock_run_git_command: Mock,
+    caplog: pytest.LogCaptureFixture,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """revert() only runs on an already-failing path; when it cannot get the
+    lock within the bounded timeout it warns and skips instead of hanging."""
+    CORE.config_path = tmp_path / "test.yaml"
+    monkeypatch.setattr(git, "_REVERT_LOCK_TIMEOUT_SECONDS", 0.05)
+
+    url = "https://github.com/test/repo"
+    domain = "test"
+    repo_dir = _compute_repo_dir(url, None, domain)
+    _setup_old_repo(repo_dir)
+
+    # A bare return value satisfies the rev-parse; the other commands'
+    # outputs are unused.
+    mock_run_git_command.return_value = "old_sha"
+
+    _, revert = git.clone_or_update(
+        url=url, ref=None, refresh=TimePeriodSeconds(days=1), domain=domain
+    )
+    assert revert is not None
+
+    holder, release = _hold_lock_in_thread(_lock_path(url, None, domain))
+    calls_before = len(mock_run_git_command.call_args_list)
+    with caplog.at_level(logging.INFO):
+        assert revert() is False
+    release.set()
+    holder.join()
+
+    # No git reset was issued; the wait and the skip were both logged.
+    assert len(mock_run_git_command.call_args_list) == calls_before
+    assert any("Waiting for another process" in r.getMessage() for r in caplog.records)
+    assert any("skipping revert" in r.getMessage() for r in caplog.records)
+
+
+def test_update_clears_marker_while_rewriting(
+    tmp_path: Path, mock_run_git_command: Mock
+) -> None:
+    """The completion marker is absent while a refresh rewrites the entry
+    and restored once the rewrite finishes, so a timed-out peer's fallback
+    never trusts a mid-rewrite tree and a crashed update re-clones."""
+    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)
+
+    marker_during_rewrite: list[bool] = []
+
+    def git_command_side_effect(
+        cmd: list[str], cwd: str | None = None, **kwargs: Any
+    ) -> str:
+        if _get_git_command_type(cmd) in ("stash", "fetch", "reset"):
+            marker_during_rewrite.append(_marker_path(repo_dir).is_file())
+        return ""
+
+    mock_run_git_command.side_effect = git_command_side_effect
+
+    git.clone_or_update(
+        url=url, ref=None, refresh=TimePeriodSeconds(days=1), domain=domain
+    )
+
+    assert marker_during_rewrite == [False, False, False]
+    assert _marker_path(repo_dir).is_file()
+
+
+def test_revert_skips_when_checkout_moved(
+    tmp_path: Path, mock_run_git_command: Mock, caplog: pytest.LogCaptureFixture
+) -> None:
+    """revert() only undoes this process's own update; when another process
+    refreshed the entry in the meantime it skips instead of rolling the
+    peer's newer checkout backwards."""
+    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)
+
+    # Pre-update SHA, post-update SHA, then a peer's SHA at revert time.
+    shas = iter(["old_sha", "new_sha", "peer_sha"])
+
+    def git_command_side_effect(
+        cmd: list[str], cwd: str | None = None, **kwargs: Any
+    ) -> str:
+        if _get_git_command_type(cmd) == "rev-parse":
+            return next(shas)
+        return ""
+
+    mock_run_git_command.side_effect = git_command_side_effect
+
+    _, revert = git.clone_or_update(
+        url=url, ref=None, refresh=TimePeriodSeconds(days=1), domain=domain
+    )
+    assert revert is not None
+
+    with caplog.at_level(logging.WARNING):
+        assert revert() is False
+
+    resets = [
+        c[0][0]
+        for c in mock_run_git_command.call_args_list
+        if _get_git_command_type(c[0][0]) == "reset" and c[0][0][-1] == "old_sha"
+    ]
+    assert resets == []
+    assert any("checkout moved" in r.getMessage() for r in caplog.records)
+
+
+def test_revert_returns_false_when_reset_fails(
+    tmp_path: Path, mock_run_git_command: Mock, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A failed git reset inside revert() is reported through the bool
+    contract instead of raising a cv.Invalid that would replace the
+    caller's original error."""
+    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)
+
+    def git_command_side_effect(
+        cmd: list[str], cwd: str | None = None, **kwargs: Any
+    ) -> str:
+        if _get_git_command_type(cmd) == "rev-parse":
+            return "old_sha"
+        # Only revert's reset targets the recorded SHA; the update path's
+        # reset targets FETCH_HEAD and must succeed.
+        if cmd[-1] == "old_sha":
+            raise git.GitCommandError("object not found")
+        return ""
+
+    mock_run_git_command.side_effect = git_command_side_effect
+
+    _, revert = git.clone_or_update(
+        url=url, ref=None, refresh=TimePeriodSeconds(days=1), domain=domain
+    )
+    assert revert is not None
+
+    with caplog.at_level(logging.WARNING):
+        assert revert() is False
+
+    assert any("Could not revert" in r.getMessage() for r in caplog.records)
+    # The entry cannot be trusted after a failed reset; the dropped marker
+    # forces a re-clone on the next use.
+    assert not _marker_path(repo_dir).is_file()
+
+
+def _raise_oserror_on_acquire(
+    monkeypatch: pytest.MonkeyPatch, code: int = errno.ENOLCK
+) -> None:
+    """Make every FileLock acquire fail with the given errno."""
+
+    def broken_acquire(self: FileLock, *args: Any, **kwargs: Any) -> None:
+        raise OSError(code, os.strerror(code))
+
+    monkeypatch.setattr(FileLock, "acquire", broken_acquire)
+
+
+@pytest.mark.parametrize(
+    ("code", "expected_fragment"),
+    [
+        # Genuinely missing lock support is reported as such.
+        (errno.ENOLCK, "does not support locking"),
+        # A cache directory problem is not blamed on lock support; git
+        # reports the real error when it actually matters.
+        (errno.EROFS, "Could not take the cache entry lock"),
+    ],
+)
+def test_clone_or_update_continues_unlocked_when_filesystem_cannot_lock(
+    tmp_path: Path,
+    mock_run_git_command: Mock,
+    caplog: pytest.LogCaptureFixture,
+    monkeypatch: pytest.MonkeyPatch,
+    code: int,
+    expected_fragment: str,
+) -> None:
+    """A filesystem where taking the lock fails (e.g. NFS without a lock
+    daemon) degrades to the old unlocked behavior instead of failing."""
+    CORE.config_path = tmp_path / "test.yaml"
+
+    url = "https://github.com/test/repo"
+    domain = "test"
+    repo_dir = _compute_repo_dir(url, None, domain)
+    mock_run_git_command.side_effect = _make_clone_side_effect(repo_dir)
+    _raise_oserror_on_acquire(monkeypatch, code)
+
+    with caplog.at_level(logging.WARNING):
+        result_dir, _ = git.clone_or_update(
+            url=url, ref=None, refresh=git.NEVER_REFRESH, domain=domain
+        )
+
+    assert result_dir == repo_dir
+    assert _marker_path(repo_dir).is_file()
+    warnings = [
+        r.getMessage()
+        for r in caplog.records
+        if "continuing without a lock" in r.getMessage()
+    ]
+    assert warnings
+    assert expected_fragment in warnings[0]
+
+
+@pytest.mark.parametrize("broken_from_start", [True, False])
+def test_revert_continues_unlocked_when_filesystem_cannot_lock(
+    tmp_path: Path,
+    mock_run_git_command: Mock,
+    caplog: pytest.LogCaptureFixture,
+    monkeypatch: pytest.MonkeyPatch,
+    broken_from_start: bool,
+) -> None:
+    """A revert still resets when locking is unavailable, whether the wrapper
+    already fell back to unlocked (revert sees no lock at all) or the
+    filesystem stops locking between the update and the revert."""
+    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)
+
+    # A bare return value satisfies the rev-parse; the other commands'
+    # outputs are unused.
+    mock_run_git_command.return_value = "old_sha"
+
+    if broken_from_start:
+        _raise_oserror_on_acquire(monkeypatch)
+
+    with caplog.at_level(logging.WARNING):
+        _, revert = git.clone_or_update(
+            url=url, ref=None, refresh=TimePeriodSeconds(days=1), domain=domain
+        )
+        assert revert is not None
+        if not broken_from_start:
+            _raise_oserror_on_acquire(monkeypatch)
+        # The reset ran (unlocked), so the revert reports success.
+        assert revert() is True
+
+    assert mock_run_git_command.call_args_list[-1][0][0] == [
+        "git",
+        "reset",
+        "--hard",
+        "old_sha",
+    ]
+    assert any("continuing without a lock" in r.getMessage() for r in caplog.records)
+
+
+def _script_acquire_statuses(
+    monkeypatch: pytest.MonkeyPatch, statuses: list["git._LockStatus"]
+) -> list[float]:
+    """Replace _acquire_repo_lock with a scripted sequence; returns the
+    timeouts it was called with."""
+    timeouts: list[float] = []
+    status_iter = iter(statuses)
+
+    def fake_acquire(
+        lock: FileLock, safe_key: str, timeout: float, **kwargs: Any
+    ) -> "git._LockStatus":
+        timeouts.append(timeout)
+        return next(status_iter)
+
+    monkeypatch.setattr(git, "_acquire_repo_lock", fake_acquire)
+    return timeouts
+
+
+@pytest.mark.parametrize("subpath", [None, Path("lib")])
+def test_clone_or_update_uses_complete_entry_when_lock_wait_times_out(
+    tmp_path: Path,
+    mock_run_git_command: Mock,
+    caplog: pytest.LogCaptureFixture,
+    monkeypatch: pytest.MonkeyPatch,
+    subpath: Path | None,
+) -> None:
+    """A bounded wait behind a stalled holder falls back to an existing
+    complete cache entry instead of hanging every peer."""
+    CORE.config_path = tmp_path / "test.yaml"
+
+    url = "https://github.com/test/repo"
+    domain = "test"
+    repo_dir = _compute_repo_dir(url, None, domain)
+    if subpath is not None:
+        repo_dir = repo_dir / subpath
+    _simulate_cloned_repo(repo_dir)
+    _mark_clone_complete(repo_dir)
+
+    timeouts = _script_acquire_statuses(monkeypatch, [git._LockStatus.TIMEOUT])
+
+    with caplog.at_level(logging.WARNING):
+        result_dir, revert = git.clone_or_update(
+            url=url,
+            ref=None,
+            refresh=TimePeriodSeconds(days=1),
+            domain=domain,
+            subpath=subpath,
+        )
+
+    assert result_dir == repo_dir
+    assert revert is None
+    # Nothing was cloned or refreshed; the existing entry was used as-is.
+    assert mock_run_git_command.call_args_list == []
+    assert timeouts == [git._COMPLETE_ENTRY_LOCK_TIMEOUT_SECONDS]
+    assert any(
+        "proceeding with the existing clone" in r.getMessage() for r in caplog.records
+    )
+
+
+def test_clone_or_update_waits_unbounded_without_complete_entry(
+    tmp_path: Path,
+    mock_run_git_command: Mock,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """With no complete entry there is nothing to fall back to, so after the
+    bounded wait expires the caller keeps waiting for the holder's clone."""
+    CORE.config_path = tmp_path / "test.yaml"
+
+    url = "https://github.com/test/repo"
+    domain = "test"
+    repo_dir = _compute_repo_dir(url, None, domain)
+    mock_run_git_command.side_effect = _make_clone_side_effect(repo_dir)
+
+    timeouts = _script_acquire_statuses(
+        monkeypatch, [git._LockStatus.TIMEOUT, git._LockStatus.ACQUIRED]
+    )
+
+    result_dir, _ = git.clone_or_update(
+        url=url, ref=None, refresh=git.NEVER_REFRESH, domain=domain
+    )
+
+    assert result_dir == repo_dir
+    assert timeouts == [git._COMPLETE_ENTRY_LOCK_TIMEOUT_SECONDS, -1]
+    # The clone proceeded normally once the lock was finally acquired.
+    assert _marker_path(repo_dir).is_file()
+
+
 def _real_git(*args: str, cwd: Path) -> None:
     """Run real git to build a test fixture repository."""
     subprocess.run(
@@ -1674,7 +2268,8 @@ def test_refresh_picks_up_new_remote_commits(
     # Verify the refresh sequence: rev-parse -> stash -> fetch (depth=1) -> reset
     call_list = mock_run_git_command.call_args_list
     cmd_sequence = [_get_git_command_type(c[0][0]) for c in call_list]
-    assert cmd_sequence == ["rev-parse", "stash", "fetch", "reset"]
+    # The trailing rev-parse records the post-update SHA for revert().
+    assert cmd_sequence == ["rev-parse", "stash", "fetch", "reset", "rev-parse"]
 
     fetch_cmd = call_list[2][0][0]
     assert "--depth=1" in fetch_cmd
@@ -1685,7 +2280,7 @@ def test_refresh_picks_up_new_remote_commits(
 
     # revert callback should reset back to the recorded pre-update SHA.
     assert revert is not None
-    revert()
+    assert revert() is True
     assert mock_run_git_command.call_args_list[-1][0][0] == [
         "git",
         "reset",