Record bootloader hooks from the configured build, make config name sets immutable, never delete an entry on a transient read error

esphome/build_gen/espidf.py

@@ -291,6 +291,13 @@ include($ENV{{IDF_PATH}}/tools/cmake/project.cmake)
 
 project({CORE.name})
 
+# Record the bootloader customization channels the cache cannot replay;
+# esphome.espidf.bootloader bypasses the cache when any is non-empty.
+idf_build_get_property(esphome_bl_dirs BOOTLOADER_EXTRA_COMPONENT_DIRS)
+idf_build_get_property(esphome_bl_args EXTRA_CMAKE_ARGS)
+file(WRITE ${{CMAKE_BINARY_DIR}}/esphome_bootloader_hooks.txt
+     "${{esphome_bl_dirs}}${{BOOTLOADER_IGNORE_EXTRA_COMPONENT}}${{esphome_bl_args}}")
+
 # Emit per-memory-type JSON size data for ESPHome to read post-build.
 # json2 stays small; raw dumps every symbol (~2s on a large map) and
 # this command runs inside the link edge, blocking everything downstream.

esphome/espidf/__init__.py

@@ -12,6 +12,10 @@ from typing import TYPE_CHECKING
 if TYPE_CHECKING:
     from pathlib import Path
 
+# Written by the generated CMakeLists after project(); holds the bootloader
+# customization channels the cache cannot replay (see espidf.bootloader).
+BOOTLOADER_HOOKS_FILE = "esphome_bootloader_hooks.txt"
+
 
 def variant_to_idf_target(variant: str) -> str:
     """Map an esp32 variant name (e.g. "ESP32S3") to its ESP-IDF target name."""

esphome/espidf/bootloader.py

@@ -11,7 +11,6 @@ from __future__ import annotations
 
 from collections.abc import Iterable
 import contextlib
-from dataclasses import dataclass
 import hashlib
 import json
 import logging
@@ -25,15 +24,9 @@ import time
 from esphome.build_helpers.ccache import parse_enable_env
 from esphome.build_helpers.tool_runner import run_build_tool
 from esphome.core import CORE, EsphomeError
-from esphome.espidf import framework, parse_sdkconfig, toolchain
+from esphome.espidf import BOOTLOADER_HOOKS_FILE, framework, parse_sdkconfig, toolchain
 from esphome.framework_helpers import _rename_with_retry
-from esphome.helpers import (
-    copy_file_if_changed,
-    file_compare,
-    read_json_file,
-    rmtree,
-    write_file,
-)
+from esphome.helpers import copy_file_if_changed, file_compare, rmtree, write_file
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -41,9 +34,6 @@ BOOTLOADER_CACHE_ENV = "ESPHOME_BOOTLOADER_CACHE"
 # Bump when the build or storage recipe changes; old entries then miss.
 _CACHE_SCHEMA = "1"
 _SECURE_OPTION = re.compile(r"CONFIG_(SECURE_|FLASH_ENCRYPTION)")
-_BOOTLOADER_PROPERTY = re.compile(
-    r"BOOTLOADER_(EXTRA_COMPONENT_DIRS|IGNORE_EXTRA_COMPONENT)"
-)
 _DISABLED_VALUES = frozenset(("n", "0", ""))
 # Marks a config name the app's sdkconfig does not define.
 _ABSENT = "\x00absent"
@@ -79,7 +69,9 @@ def _compute_enabled() -> bool:
         if CORE.relative_build_path("bootloader_components").exists():
             # Project-local bootloader overrides are inputs the key can't see.
             return False
-        if _managed_bootloader_hook() is not None:
+        if _recorded_bootloader_hooks():
+            # The last configure saw a component customizing the bootloader.
+            _LOGGER.info("Bootloader cache off: build customizes the bootloader")
             return False
         if not os.access(tools := framework.get_idf_tools_path(), os.W_OK):
             # A read-only shared prefix would fail the cache on every build.
@@ -100,19 +92,19 @@ def _compute_enabled() -> bool:
         return False
 
 
-def _managed_bootloader_hook() -> str | None:
-    """The managed component wiring the bootloader build, or None.
+def _recorded_bootloader_hooks() -> str | None:
+    """The bootloader customization channels the last configure recorded.
 
-    Checked again after configure: a cold tree has no managed_components
-    yet when the enable check first runs.
+    The generated CMakeLists dumps them after project(), so every source
+    counts: managed and local components, main, and EXTRA_CMAKE_ARGS.
+    None means no configure has recorded them yet; rechecked after
+    configure, since the pre-configure read can be stale or missing.
     """
-    managed = CORE.relative_build_path("managed_components")
-    hooks = managed.glob("*/project_include.cmake") if managed.is_dir() else ()
-    for hook in hooks:
-        if _BOOTLOADER_PROPERTY.search(hook.read_text(encoding="utf-8")):
-            _LOGGER.info("Bootloader cache off: %s customizes it", hook.parent.name)
-            return hook.parent.name
-    return None
+    path = toolchain._build_dir() / BOOTLOADER_HOOKS_FILE
+    try:
+        return path.read_text(encoding="utf-8").strip()
+    except FileNotFoundError:
+        return None
 
 
 def _snapshot_blocks_cache(sdkconfig_path: Path) -> bool:
@@ -148,69 +140,50 @@ def _cache_root() -> Path:
     )
 
 
-def _load_config_names() -> list[str] | None:
-    """The known union of config option names the bootloader consumes."""
-    path = _cache_root() / "config_names.json"
-    names = read_json_file(path)
-    if isinstance(names, list) and names and all(isinstance(n, str) for n in names):
-        return names
-    if names is not None:
-        # The next merge rewrites the full union; this only costs a miss.
-        _LOGGER.info("Ignoring corrupt %s", path)
-    return None
+def _load_config_names() -> list[str]:
+    """The known union of config option names the bootloader consumes.
+
+    Each harvest is an immutable file named by its digest; the union over
+    them is monotonic without any cross-process locking.
+    """
+    union: set[str] = set()
+    for path in _cache_root().glob("config_names/*.json"):
+        try:
+            names = json.loads(path.read_text(encoding="utf-8"))
+        except (OSError, ValueError) as err:
+            _LOGGER.info("Ignoring unreadable %s: %s", path, err)
+            continue
+        if isinstance(names, list) and all(isinstance(n, str) for n in names):
+            union.update(names)
+        else:
+            _LOGGER.info("Ignoring corrupt %s", path)
+    return sorted(union)
 
 
 def _merge_config_names(names: Iterable[str]) -> list[str]:
-    """Union-merge harvested names; a corrupt file is replaced outright.
+    """Record this build's harvested names and return the resulting union.
 
-    write_file renames into place, and re-merging until the file covers our
-    names keeps the union monotonic when two builds publish at once.
+    A set that was already recorded is not rewritten; concurrent publishes
+    can only add files, never lose another's discovered names.
     """
-    path = _cache_root() / "config_names.json"
-    merged = sorted(names)
-    for _ in range(3):
-        merged = sorted(set(_load_config_names() or ()) | set(merged))
+    merged = sorted(set(names))
+    path = _cache_root() / "config_names" / f"{_compute_key(merged)}.json"
+    if not path.is_file():
         write_file(path, json.dumps(merged, separators=(",", ":")))
-        if set(_load_config_names() or ()) >= set(merged):
-            break
-    return merged
-
-
-@dataclass(frozen=True, kw_only=True)
-class _SubprojectContract:
-    """How the pinned IDF's bootloader ExternalProject configures the subproject."""
-
-    self_append_bootloader_dir: bool  # EXTRA_COMPONENT_DIRS gets the bootloader dir
-    forward_idf_build_v2: bool  # IDF_BUILD_V2 forwarded (empty: never opted in)
-
-
-_SUBPROJECT_5 = _SubprojectContract(
-    self_append_bootloader_dir=True,
-    forward_idf_build_v2=False,
-)
-_SUBPROJECT_6 = _SubprojectContract(
-    self_append_bootloader_dir=False,
-    forward_idf_build_v2=True,
-)
-
-
-def _subproject() -> _SubprojectContract:
-    from esphome.components.esp32 import idf_version
-    import esphome.config_validation as cv
-
-    return _SUBPROJECT_6 if idf_version() >= cv.Version(6, 0, 0) else _SUBPROJECT_5
+    # The union guards against our own file being unreadable right back.
+    return sorted(set(merged) | set(_load_config_names()))
 
 
 def _subproject_cmake_args(
     cmake: str, python: str, sdkconfig: str, idf_path: str, project_dir: str
 ) -> list[str]:
     """The configure argv, mirroring IDF's bootloader ExternalProject args."""
-    contract = _subproject()
-    extra_dirs = (
-        f"{idf_path}/components/bootloader"
-        if contract.self_append_bootloader_dir
-        else ""
-    )
+    from esphome.components.esp32 import idf_version
+    import esphome.config_validation as cv
+
+    # 5.x self-appends the bootloader dir to EXTRA_COMPONENT_DIRS; 6.x
+    # instead forwards IDF_BUILD_V2 (empty: the subproject never opts in).
+    idf6 = idf_version() >= cv.Version(6, 0, 0)
     args = [
         cmake,
         "-G",
@@ -220,9 +193,9 @@ def _subproject_cmake_args(
         f"-DIDF_TARGET={toolchain._idf_target()}",
         "-DPYTHON_DEPS_CHECKED=1",
         f"-DPYTHON={python}",
-        f"-DEXTRA_COMPONENT_DIRS={extra_dirs}",
+        f"-DEXTRA_COMPONENT_DIRS={'' if idf6 else f'{idf_path}/components/bootloader'}",
     ]
-    if contract.forward_idf_build_v2:
+    if idf6:
         args.append("-DIDF_BUILD_V2=")
     args += [
         f"-DPROJECT_SOURCE_DIR={project_dir}",
@@ -244,13 +217,13 @@ def _key_payload(names: list[str], app_config: dict, compiler: str, stamp: str)
         "args": _subproject_cmake_args(
             "cmake", "@PYTHON@", "@SDKCONFIG@", "@IDF@", "@PROJECT@"
         ),
-        "config": {n: app_config.get(n, _ABSENT) for n in sorted(names)},
+        "config": {n: app_config.get(n, _ABSENT) for n in names},
         "env": {n: os.environ.get(n, _ABSENT) for n in _FLAG_ENV_VARS},
     }
 
 
-def _compute_key(payload: dict) -> str:
-    """The cache entry name for this build's bootloader inputs."""
+def _compute_key(payload: dict | list) -> str:
+    """The digest naming a cache entry or a recorded config name set."""
     text = json.dumps(payload, sort_keys=True, separators=(",", ":"))
     return hashlib.sha256(text.encode()).hexdigest()[:16]
 
@@ -294,10 +267,17 @@ def _bin_sha256(path: Path) -> str:
 
 
 def _entry_intact(entry: Path) -> bool:
-    """Complete with the published bin bytes; anything else is a miss."""
+    """Complete with the published bin bytes; anything else is a miss.
+
+    False only for provable damage; a transient read error raises so a
+    valid shared entry is never mistaken for a corrupt one and deleted.
+    """
     if not all((entry / name).is_file() for name in _OUTPUTS):
         return False
-    meta = read_json_file(entry / "meta.json")
+    try:
+        meta = json.loads((entry / "meta.json").read_text(encoding="utf-8"))
+    except (FileNotFoundError, ValueError):
+        return False
     expected = meta.get("bootloader_bin_sha256") if isinstance(meta, dict) else None
     return expected == _bin_sha256(entry / "bootloader.bin")
 
@@ -305,8 +285,9 @@ def _entry_intact(entry: Path) -> bool:
 def _publish(build_dir: Path, key: str, payload: dict) -> Path:
     """Move the distilled outputs into the cache; losing a race is fine."""
     entry = _cache_root() / key
-    if not _entry_intact(entry):
+    if entry.is_dir() and not _entry_intact(entry):
         # A partial or corrupted entry (manual cleanup) must not win.
+        _LOGGER.debug("Discarding damaged entry %s", entry)
         _remove_dir(entry)
     stage = _work_dir(f".stage-{key}-")
     try:
@@ -347,6 +328,10 @@ def _install_into_build(entry: Path) -> Path:
     if (dest / "CMakeCache.txt").is_file():
         # A leftover in-tree sub-build; its ninja state must not linger.
         rmtree(dest)
+    if (stamps := dest.parent / "bootloader-prefix").is_dir():
+        # Stale ExternalProject stamps would make a later flip back to
+        # in-tree skip the sub-configure and fail on the missing cache.
+        rmtree(stamps)
     for name in _OUTPUTS:
         copy_file_if_changed(entry / name, dest / name)
     return dest
@@ -398,8 +383,10 @@ def ensure_cached_bootloader(verbose: bool = False) -> bool:
     bootloader raises instead, since in-tree would fail the same way.
     """
     try:
-        if _managed_bootloader_hook() is not None:
-            # Configure just downloaded it; the pre-configure scan ran too early.
+        if _recorded_bootloader_hooks() != "":
+            # Non-empty: configure just recorded a bootloader customization.
+            # None: this configure did not write the record at all.
+            _LOGGER.info("Bootloader customized by the build; building in-tree")
             return False
         app_config = toolchain._load_sdkconfig_json(toolchain._build_dir())
         offset = _bootloader_offset(app_config)
@@ -414,7 +401,7 @@ def ensure_cached_bootloader(verbose: bool = False) -> bool:
             )
             return False
         dest = _install_cached(app_config, compiler, verbose)
-    except (OSError, EsphomeError) as err:
+    except (OSError, ValueError, EsphomeError) as err:
         _LOGGER.warning("Bootloader cache failed: %s", err)
         return False
     if dest is None:
@@ -422,7 +409,7 @@ def ensure_cached_bootloader(verbose: bool = False) -> bool:
     try:
         # The size overflow raises out: the in-tree build would fail the same.
         _check_bootloader_size(dest / "bootloader.bin", offset, int(table_offset))
-    except OSError as err:
+    except (OSError, ValueError) as err:
         _LOGGER.warning("Bootloader cache failed: %s", err)
         return False
     return True
@@ -434,7 +421,7 @@ def _install_cached(app_config: dict, compiler: str, verbose: bool) -> Path | No
     if not (stamp := framework.read_idf_version_txt(toolchain._get_idf_path())):
         _LOGGER.info("Framework version.txt missing; cannot key the bootloader")
         return None
-    if (names := _load_config_names()) is not None:
+    if names := _load_config_names():
         entry = _cache_root() / _compute_key(
             _key_payload(names, app_config, compiler, stamp)
         )

script/check_idf_py_equivalence.py

@@ -184,10 +184,10 @@ def _bootloader_parity(build_path: Path) -> list[str]:
     # pylint: disable=protected-access
     from esphome.espidf import bootloader, toolchain
 
-    bin_path = build_path / "build" / "bootloader" / "bootloader.bin"
     if not bootloader.tree_uses_cached_bootloader(build_path / "build"):
         print("note: tree not in cached mode; bootloader parity not exercised")
         return []
+    bin_path = build_path / "build" / "bootloader" / "bootloader.bin"
     cached = bin_path.read_bytes()
     bin_path.unlink()  # the rebuild must produce it, never compare to itself
     os.environ[bootloader.BOOTLOADER_CACHE_ENV] = "0"

tests/unit_tests/build_gen/test_espidf.py

@@ -175,6 +175,20 @@ def test_get_project_cmakelists_size_command_uses_json2() -> None:
     assert "--format=json2" in content
 
 
+def test_get_project_cmakelists_records_bootloader_hooks() -> None:
+    """The post-project() dump feeds espidf.bootloader's bypass check."""
+    from esphome.espidf import BOOTLOADER_HOOKS_FILE
+
+    content = _render()
+    assert f"file(WRITE ${{CMAKE_BINARY_DIR}}/{BOOTLOADER_HOOKS_FILE}" in content
+    for source in (
+        "esphome_bl_dirs BOOTLOADER_EXTRA_COMPONENT_DIRS",
+        "${BOOTLOADER_IGNORE_EXTRA_COMPONENT}",
+        "esphome_bl_args EXTRA_CMAKE_ARGS",
+    ):
+        assert source in content
+
+
 def test_get_project_cmakelists_uses_supplied_builtin_components() -> None:
     """A cached list replaces project_description.json and is still filtered
     by EXCLUDE_COMPONENTS."""

tests/unit_tests/test_espidf_bootloader.py

@@ -33,6 +33,14 @@ def _fresh_cache(setup_core: Path, monkeypatch: pytest.MonkeyPatch) -> None:
     CORE.data.pop(toolchain.DOMAIN, None)
 
 
+def _write_hooks_record(text: str) -> Path:
+    """The hooks dump the generated CMakeLists writes on every configure."""
+    path = toolchain._build_dir() / "esphome_bootloader_hooks.txt"
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(text)
+    return path
+
+
 def _write_snapshot(text: str = "") -> Path:
     path = CORE.relative_build_path("sdkconfig.dev.esphomeinternal")
     path.parent.mkdir(parents=True, exist_ok=True)
@@ -117,22 +125,16 @@ def test_enabled_project_bootloader_components_disable(tmp_path: Path) -> None:
         assert bootloader._compute_enabled() is False
 
 
-def test_enabled_managed_component_bootloader_hook_disables(tmp_path: Path) -> None:
-    """A managed component wiring the bootloader build is an unseen input."""
+def test_enabled_recorded_bootloader_hooks_disable(tmp_path: Path) -> None:
+    """A customization recorded by the last configure is an unseen input."""
     _write_snapshot("CONFIG_APP_REPRODUCIBLE_BUILD=y\n")
-    hook = CORE.relative_build_path(
-        "managed_components", "vendor__boot", "project_include.cmake"
-    )
-    hook.parent.mkdir(parents=True)
-    hook.write_text(
-        'idf_build_set_property(BOOTLOADER_EXTRA_COMPONENT_DIRS "x" APPEND)'
-    )
+    _write_hooks_record("/some/component/dir")
     with (
         _tools_prefix(tmp_path),
         patch.object(build_gen, "idf_macro_matches", return_value=True),
     ):
         assert bootloader._compute_enabled() is False
-        hook.write_text("# nothing bootloader related")
+        _write_hooks_record("")
         assert bootloader._compute_enabled() is True
 
 
@@ -292,45 +294,43 @@ def test_merge_config_names_unions_and_sorts(tmp_path: Path) -> None:
         assert bootloader._load_config_names() == ["A", "B", "C"]
 
 
-def test_merge_config_names_remerges_after_a_racing_writer(tmp_path: Path) -> None:
-    """A name landed by a concurrent build must survive the merge."""
-    # merge-read, verify-read (racer landed), merge-read, verify-read
-    loads = iter([None, ["OTHER"], ["OTHER"], ["A", "B", "OTHER"]])
-    with (
-        patch.object(bootloader, "_load_config_names", side_effect=loads),
-        patch.object(bootloader, "write_file") as mock_write,
-    ):
-        assert bootloader._merge_config_names(["B", "A"]) == ["A", "B", "OTHER"]
-    assert mock_write.call_count == 2  # rewrote once with the racer's name
-
-
-def test_merge_config_names_gives_up_after_bounded_retries(tmp_path: Path) -> None:
-    """A pathological racer costs at most a miss, never a spin."""
-    with (
-        patch.object(bootloader, "_load_config_names", return_value=["X"]),
-        patch.object(bootloader, "write_file") as mock_write,
-    ):
-        assert bootloader._merge_config_names(["A"]) == ["A", "X"]
-    assert mock_write.call_count == 3
+def test_merge_config_names_skips_rewriting_a_recorded_set(tmp_path: Path) -> None:
+    """The same harvest recorded twice writes one immutable file."""
+    root = tmp_path / "cache"
+    with patch.object(bootloader, "_cache_root", return_value=root):
+        bootloader._merge_config_names(["A"])
+        with patch.object(bootloader, "write_file") as mock_write:
+            assert bootloader._merge_config_names(["A"]) == ["A"]
+    mock_write.assert_not_called()
 
 
 @pytest.mark.parametrize(
-    "content",
-    [None, "not json", "[]", '["a", 1]', '"str"'],
-    ids=["missing", "corrupt", "empty", "mixed-types", "not-a-list"],
+    ("content", "log"),
+    [
+        (None, ""),
+        ("not json", "Ignoring unreadable"),
+        ("[]", ""),
+        ('["a", 1]', "Ignoring corrupt"),
+        ('"str"', "Ignoring corrupt"),
+    ],
+    ids=["missing", "unreadable", "empty", "mixed-types", "not-a-list"],
 )
 def test_load_config_names_rejects_bad_files(
-    tmp_path: Path, content: str | None, caplog: pytest.LogCaptureFixture
+    tmp_path: Path, content: str | None, log: str, caplog: pytest.LogCaptureFixture
 ) -> None:
     if content is not None:
-        (tmp_path / "config_names.json").write_text(content)
+        (tmp_path / "config_names").mkdir()
+        (tmp_path / "config_names" / "x.json").write_text(content)
     with (
         patch.object(bootloader, "_cache_root", return_value=tmp_path),
         caplog.at_level("INFO"),
     ):
-        assert bootloader._load_config_names() is None
-    # A present-but-unusable file says so; a missing one is normal.
-    assert ("Ignoring corrupt" in caplog.text) is (content not in (None, "not json"))
+        assert bootloader._load_config_names() == []
+    # A present-but-unusable file says so; missing or empty is normal.
+    if log:
+        assert log in caplog.text
+    else:
+        assert "Ignoring" not in caplog.text
 
 
 # ------------------------------------------------------------- cmake args
@@ -508,9 +508,7 @@ def test_publish_keeps_the_original_error_when_the_intact_check_breaks(
     (build / "bootloader.map").unlink()  # the real failure
     with (
         patch.object(bootloader, "_cache_root", return_value=root),
-        patch.object(
-            bootloader, "_entry_intact", side_effect=[False, PermissionError()]
-        ),
+        patch.object(bootloader, "_entry_intact", side_effect=PermissionError()),
         # FileNotFoundError can only be the staging error, not the check's.
         pytest.raises(FileNotFoundError),
     ):
@@ -529,11 +527,15 @@ def test_install_into_build_replaces_stale_subbuild(tmp_path: Path) -> None:
     dest.mkdir(parents=True)
     (dest / "CMakeCache.txt").write_text("stale in-tree sub-build")
     (dest / "old.obj").write_bytes(b"x")
+    stamps = CORE.relative_build_path("build", "bootloader-prefix", "src")
+    stamps.mkdir(parents=True)
     assert bootloader._install_into_build(entry) == dest
     for name in bootloader._OUTPUTS:
         assert (dest / name).read_bytes() == name.encode()
     assert not (dest / "CMakeCache.txt").exists()
     assert not (dest / "old.obj").exists()
+    # Stale stamps would make a later in-tree flip skip the sub-configure.
+    assert not stamps.parent.exists()
 
 
 # -------------------------------------------------------------- size check
@@ -559,26 +561,35 @@ def test_size_check_overflow_raises(tmp_path: Path) -> None:
 # ------------------------------------------------------------ orchestration
 
 
-def test_ensure_rechecks_managed_hooks_after_configure(tmp_path: Path) -> None:
-    """A hook that configure just downloaded must stop the first build too."""
-    hook = CORE.relative_build_path(
-        "managed_components", "vendor__boot", "project_include.cmake"
-    )
-    hook.parent.mkdir(parents=True)
-    hook.write_text("idf_build_set_property(BOOTLOADER_IGNORE_EXTRA_COMPONENT x)")
+def test_ensure_rechecks_recorded_hooks_after_configure(tmp_path: Path) -> None:
+    """A customization configure just recorded must stop the first build too."""
     with (
         _orchestration_env(tmp_path),
         patch.object(bootloader, "_build_standalone") as mock_build,
     ):
+        _write_hooks_record("vendor__boot")
         assert bootloader.ensure_cached_bootloader() is False
     mock_build.assert_not_called()
 
 
-def test_ensure_fails_soft_when_hook_scan_errors(tmp_path: Path) -> None:
-    """An unreadable managed hook falls back instead of aborting the build."""
+def test_ensure_fails_soft_without_a_hooks_record(tmp_path: Path) -> None:
+    """A configure that wrote no record cannot prove the bootloader is stock."""
     with (
         _orchestration_env(tmp_path),
-        patch.object(bootloader, "_managed_bootloader_hook", side_effect=OSError("io")),
+        patch.object(bootloader, "_build_standalone") as mock_build,
+    ):
+        (toolchain._build_dir() / "esphome_bootloader_hooks.txt").unlink()
+        assert bootloader.ensure_cached_bootloader() is False
+    mock_build.assert_not_called()
+
+
+def test_ensure_fails_soft_when_hooks_record_is_unreadable(tmp_path: Path) -> None:
+    """An unreadable hooks record falls back instead of aborting the build."""
+    with (
+        _orchestration_env(tmp_path),
+        patch.object(
+            bootloader, "_recorded_bootloader_hooks", side_effect=OSError("io")
+        ),
     ):
         assert bootloader.ensure_cached_bootloader() is False
 
@@ -592,7 +603,7 @@ def test_ensure_fails_soft_when_size_check_cannot_stat(
         patch.object(bootloader, "_check_bootloader_size", side_effect=OSError("gone")),
     ):
         _make_entry(root / "deadbeefdeadbeef", b"\xe9" * 64)
-        (root / "config_names.json").write_text('["A"]')
+        _seed_names(root)
         assert bootloader.ensure_cached_bootloader() is False
     assert "Bootloader cache failed" in caplog.text
 
@@ -601,11 +612,18 @@ def test_ensure_fails_soft_without_configure_outputs(
     caplog: pytest.LogCaptureFixture,
 ) -> None:
     """No app config or compiler id means fall back, not crash."""
+    _write_hooks_record("")
     with caplog.at_level("INFO"):
         assert bootloader.ensure_cached_bootloader() is False
     assert "cache unusable" in caplog.text  # says which input is missing
 
 
+def _seed_names(root: Path) -> None:
+    """Record a harvested name set the way a previous publish would have."""
+    (root / "config_names").mkdir(parents=True, exist_ok=True)
+    (root / "config_names" / "seed.json").write_text('["A"]')
+
+
 @contextmanager
 def _orchestration_env(tmp_path: Path) -> Iterator[Path]:
     """A configured app tree plus the patches every ensure_* test shares;
@@ -613,6 +631,7 @@ def _orchestration_env(tmp_path: Path) -> Iterator[Path]:
     root = tmp_path / "cache-root"
     build = CORE.relative_build_path("build")
     build.mkdir(parents=True, exist_ok=True)
+    (build / "esphome_bootloader_hooks.txt").write_text("")
     (build / "config").mkdir(exist_ok=True)
     (build / "config" / "sdkconfig.json").write_text(
         json.dumps(
@@ -639,7 +658,7 @@ def test_ensure_cache_hit_installs_without_building(tmp_path: Path) -> None:
         patch.object(bootloader, "_build_standalone") as mock_build,
     ):
         _make_entry(root / "deadbeefdeadbeef", b"\xe9" * 64)
-        (root / "config_names.json").write_text('["A"]')
+        _seed_names(root)
         assert bootloader.ensure_cached_bootloader() is True
     mock_build.assert_not_called()
     installed = CORE.relative_build_path("build", "bootloader", "bootloader.bin")
@@ -656,12 +675,12 @@ def test_ensure_cache_miss_builds_and_publishes(tmp_path: Path) -> None:
         ) as mock_build,
     ):
         root.mkdir(parents=True)
-        (root / "config_names.json").write_text('["A"]')
+        _seed_names(root)
         assert bootloader.ensure_cached_bootloader() is True
+        assert "A" in bootloader._load_config_names()
     # Both probe builds ran; entry published, names harvested, work dirs gone.
     assert len(mock_build.call_args_list) == 2
     assert (root / "deadbeefdeadbeef" / "bootloader.bin").is_file()
-    assert "A" in json.loads((root / "config_names.json").read_text())
     assert not list(root.glob(".build-*"))
     installed = CORE.relative_build_path("build", "bootloader", "bootloader.bin")
     assert installed.is_file()
@@ -692,6 +711,37 @@ def test_ensure_soft_fails_when_build_yields_no_config(
     assert "produced no" in caplog.text  # names the expected path
 
 
+def test_entry_intact_false_on_corrupt_meta(tmp_path: Path) -> None:
+    """Provably damaged metadata is a miss and may be discarded."""
+    entry = tmp_path / "k"
+    _make_entry(entry)
+    (entry / "meta.json").write_text("not json")
+    assert bootloader._entry_intact(entry) is False
+
+
+def test_entry_intact_raises_on_an_unreadable_meta(tmp_path: Path) -> None:
+    """Transient read trouble must not look like corruption."""
+    entry = tmp_path / "k"
+    _make_entry(entry)
+    (entry / "meta.json").unlink()
+    (entry / "meta.json").mkdir()  # read_text raises OSError, not ValueError
+    with pytest.raises(OSError):
+        bootloader._entry_intact(entry)
+
+
+def test_publish_keeps_an_entry_it_cannot_verify(tmp_path: Path) -> None:
+    """A transient read error must never delete a valid shared entry."""
+    with patch.object(bootloader, "_cache_root", return_value=tmp_path):
+        entry = tmp_path / "k"
+        _make_entry(entry)
+        with (
+            patch.object(bootloader, "_entry_intact", side_effect=OSError("locked")),
+            pytest.raises(OSError),
+        ):
+            bootloader._publish(tmp_path / "build", "k", {})
+    assert (entry / "bootloader.bin").is_file()
+
+
 @pytest.mark.parametrize("damage", ["missing-elf", "corrupt-bin"])
 def test_ensure_rebuilds_a_damaged_entry(tmp_path: Path, damage: str) -> None:
     """A partial or corrupted entry is a miss and gets replaced."""
@@ -707,7 +757,7 @@ def test_ensure_rebuilds_a_damaged_entry(tmp_path: Path, damage: str) -> None:
             (entry / "bootloader.elf").unlink()
         else:
             (entry / "bootloader.bin").write_bytes(b"truncated")
-        (root / "config_names.json").write_text('["A"]')
+        _seed_names(root)
         assert bootloader.ensure_cached_bootloader() is True
     assert mock_build.call_count == 2  # built and probed, not served as a hit
     for name in bootloader._OUTPUTS: