[store_yaml] Gate YAML discovery on component presence, fail on unreadable files, warn on unresolved includes

esphome/components/api/api_connection.cpp

@@ -1191,6 +1191,9 @@ void APIConnection::try_send_store_yaml_() {
     return;
   auto *comp = store_yaml::global_store_yaml;
   if (comp == nullptr) {
+    // Defensive only: the global is set once in setup() and never cleared,
+    // and streaming can only start after on_get_yaml_request() saw it
+    // non-null, so this cannot happen mid-stream.
     this->store_yaml_pos_ = std::numeric_limits<size_t>::max();
     return;
   }

esphome/components/store_yaml/__init__.py

@@ -94,6 +94,14 @@ def _gather_files(include_secrets: bool) -> list[tuple[str, bytes]]:
             "did not populate CORE.data['yaml_sources']."
         )
 
+    if discovered.unresolved:
+        _LOGGER.warning(
+            "store_yaml: %d !include path(s) use substitutions and cannot be "
+            "captured (%s); the embedded recovery data will not contain them",
+            len(discovered.unresolved),
+            ", ".join(discovered.unresolved),
+        )
+
     config_path = Path(CORE.config_path).resolve()
     root = config_path.parent
     secret_paths = discovered.secrets
@@ -109,8 +117,11 @@ def _gather_files(include_secrets: bool) -> list[tuple[str, bytes]]:
             try:
                 content = path.read_bytes()
             except OSError as err:
-                _LOGGER.warning("store_yaml: skipping unreadable %s (%s)", path, err)
-                continue
+                # A silently partial recovery blob defeats the feature; fail
+                # the build instead of embedding an incomplete file set.
+                raise EsphomeError(
+                    f"store_yaml: cannot read tracked YAML file {path}: {err}"
+                ) from err
 
         try:
             rel_str = path.relative_to(root).as_posix()

esphome/config.py

@@ -1344,10 +1344,12 @@ def _load_config(
     # Discover the user's on-disk YAML files via a fresh re-parse — same
     # pattern bundle.py uses. Doing it post-validation (rather than keeping a
     # listener installed across validation) avoids capturing framework YAML
-    # that components load internally (e.g. LVGL's `hello_world.yaml`). The
-    # result is consumed by components like store_yaml that want to embed the
-    # user's configuration in firmware for recovery.
-    CORE.data["yaml_sources"] = yaml_util.discover_user_yaml_files(CORE.config_path)
+    # that components load internally (e.g. LVGL's `hello_world.yaml`). Only
+    # run when a consumer is configured: the re-parse force-loads every
+    # include and would otherwise tax every compile. `result` is the fully
+    # merged config, so store_yaml pulled in via packages is seen here too.
+    if "store_yaml" in result:
+        CORE.data["yaml_sources"] = yaml_util.discover_user_yaml_files(CORE.config_path)
     return result
 
 

esphome/yaml_util.py

@@ -272,7 +272,8 @@ def force_load_include_files(
     *,
     warn_on_unresolved: bool = True,
     _seen: set[int] | None = None,
-) -> None:
+    _unresolved: list[str] | None = None,
+) -> list[str]:
     """Recursively resolve any deferred ``IncludeFile`` instances in a YAML tree.
 
     Nested ``!include`` returns a deferred ``IncludeFile`` that is only resolved
@@ -285,14 +286,17 @@ def force_load_include_files(
     variables cannot be loaded. By default a warning is logged for each one;
     pass ``warn_on_unresolved=False`` (used by discovery paths that run on a
     fresh re-parse where substitutions haven't been applied yet) to demote it
-    to a debug log.
+    to a debug log. Returns the path strings of those unloadable includes so
+    callers can tell the walk was incomplete.
     """
     if _seen is None:
         _seen = set()
+    if _unresolved is None:
+        _unresolved = []
 
     if isinstance(obj, IncludeFile):
         if id(obj) in _seen:
-            return
+            return _unresolved
         _seen.add(id(obj))
         if obj.has_unresolved_expressions():
             log = _LOGGER.warning if warn_on_unresolved else _LOGGER.debug
@@ -301,7 +305,8 @@ def force_load_include_files(
                 obj.file,
                 obj.parent_file,
             )
-            return
+            _unresolved.append(str(obj.file))
+            return _unresolved
         try:
             loaded = obj.load()
         except EsphomeError as err:
@@ -311,26 +316,36 @@ def force_load_include_files(
                 obj.parent_file,
                 err,
             )
-            return
+            return _unresolved
         force_load_include_files(
-            loaded, warn_on_unresolved=warn_on_unresolved, _seen=_seen
+            loaded,
+            warn_on_unresolved=warn_on_unresolved,
+            _seen=_seen,
+            _unresolved=_unresolved,
         )
     elif isinstance(obj, dict):
         if id(obj) in _seen:
-            return
+            return _unresolved
         _seen.add(id(obj))
         for value in obj.values():
             force_load_include_files(
-                value, warn_on_unresolved=warn_on_unresolved, _seen=_seen
+                value,
+                warn_on_unresolved=warn_on_unresolved,
+                _seen=_seen,
+                _unresolved=_unresolved,
             )
     elif isinstance(obj, (list, tuple)):
         if id(obj) in _seen:
-            return
+            return _unresolved
         _seen.add(id(obj))
         for item in obj:
             force_load_include_files(
-                item, warn_on_unresolved=warn_on_unresolved, _seen=_seen
+                item,
+                warn_on_unresolved=warn_on_unresolved,
+                _seen=_seen,
+                _unresolved=_unresolved,
             )
+    return _unresolved
 
 
 @dataclass(slots=True)
@@ -341,11 +356,14 @@ class DiscoveredYamlFiles:
     were re-parsing the user's config; ``secrets`` is the subset whose
     *un-resolved* filename matched :data:`esphome.const.SECRETS_FILES` (so
     a ``secrets.yaml`` symlinked to a differently-named target is still
-    flagged as secrets).
+    flagged as secrets). ``unresolved`` lists ``!include`` path strings that
+    contain substitution variables and therefore could not be loaded —
+    consumers should treat ``files`` as incomplete when it is non-empty.
     """
 
     files: list[Path] = field(default_factory=list)
     secrets: set[Path] = field(default_factory=set)
+    unresolved: list[str] = field(default_factory=list)
 
 
 def discover_user_yaml_files(config_path: Path) -> DiscoveredYamlFiles:
@@ -377,7 +395,7 @@ def discover_user_yaml_files(config_path: Path) -> DiscoveredYamlFiles:
                 data = load_yaml(config_path)
             except EsphomeError:
                 return DiscoveredYamlFiles(list(loaded), secrets)
-            force_load_include_files(data, warn_on_unresolved=False)
+            unresolved = force_load_include_files(data, warn_on_unresolved=False)
         finally:
             _load_listeners.remove(_capture_secret)
 
@@ -388,7 +406,7 @@ def discover_user_yaml_files(config_path: Path) -> DiscoveredYamlFiles:
         if path not in seen:
             seen.add(path)
             unique.append(path)
-    return DiscoveredYamlFiles(unique, secrets)
+    return DiscoveredYamlFiles(unique, secrets, unresolved)
 
 
 def _add_data_ref(fn):

tests/unit_tests/components/test_store_yaml.py

@@ -135,6 +135,42 @@ def test_gather_raises_when_no_sources(project: Path) -> None:
         _gather_files(include_secrets=False)
 
 
+def test_gather_raises_on_unreadable_file(
+    project: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """An unreadable tracked file fails the build instead of producing a
+    silently partial recovery blob."""
+    _set_sources(project, "entry.yaml", "wifi.yaml")
+    orig_read_bytes = Path.read_bytes
+
+    def fake_read_bytes(self: Path) -> bytes:
+        if self.name == "wifi.yaml":
+            raise OSError("permission denied")
+        return orig_read_bytes(self)
+
+    monkeypatch.setattr(Path, "read_bytes", fake_read_bytes)
+    with pytest.raises(EsphomeError, match="wifi.yaml"):
+        _gather_files(include_secrets=False)
+
+
+def test_gather_warns_on_unresolved_includes(
+    project: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Substitution-pathed includes that discovery could not capture produce a
+    warning naming them, so the user knows the blob is incomplete."""
+    CORE.config_path = project / "entry.yaml"
+    CORE.data["yaml_sources"] = DiscoveredYamlFiles(
+        [project / "entry.yaml"], set(), ["${board}.yaml"]
+    )
+    with caplog.at_level("WARNING", logger="esphome.components.store_yaml"):
+        files = _gather_files(include_secrets=False)
+    assert len(files) == 1
+    assert any(
+        "${board}.yaml" in r.message and "not contain" in r.message
+        for r in caplog.records
+    )
+
+
 def test_pack_envelope_roundtrip() -> None:
     files = [
         ("entry.yaml", b"esphome:\n  name: test\n"),

tests/unit_tests/test_yaml_util.py

@@ -1084,6 +1084,18 @@ def test_force_load_include_files_unresolved_log_level(
     assert matching == [expect_level]
 
 
+def test_force_load_include_files_returns_unresolved_paths(
+    patch_include_file: None,
+) -> None:
+    """Includes with substitution-templated paths are reported back to the
+    caller; resolvable ones are not."""
+    templated = _StubInclude("${var}.yaml", unresolved=True)
+    plain = _StubInclude("ok.yaml")
+    unresolved = force_load_include_files({"a": templated, "b": plain})
+    assert unresolved == [str(templated.file)]
+    assert plain.load_calls == 1
+
+
 def test_force_load_include_files_warns_on_load_failure(
     patch_include_file: None,
     caplog: pytest.LogCaptureFixture,
@@ -1169,6 +1181,16 @@ def test_discover_user_yaml_files_swallows_parse_errors(tmp_path: Path) -> None:
     assert isinstance(discovered, DiscoveredYamlFiles)
 
 
+def test_discover_user_yaml_files_reports_unresolved_includes(
+    tmp_path: Path,
+) -> None:
+    """A substitution-templated `!include` path is surfaced in `.unresolved`."""
+    entry = _write_entry_including(tmp_path, "${board}.yaml")
+    discovered = discover_user_yaml_files(entry)
+    assert len(discovered.unresolved) == 1
+    assert "${board}.yaml" in discovered.unresolved[0]
+
+
 def test_discover_user_yaml_files_deduplicates(tmp_path: Path) -> None:
     """The same file referenced twice appears once in `.files`."""
     _write(tmp_path, "wifi.yaml", "ssid: a\n")