[store_yaml] Simplify: lazy YAML discovery in to_code, zero-copy streaming, shared envelope decoder

esphome/components/api/api_connection.cpp

@@ -316,7 +316,10 @@ void APIConnection::loop() {
 #endif
 
 #ifdef USE_STORE_YAML
-  this->try_send_store_yaml_();
+  // Guard inline so the idle hot path pays a compare, not a function call.
+  if (this->store_yaml_pos_ != std::numeric_limits<size_t>::max()) {
+    this->try_send_store_yaml_();
+  }
 #endif
 }
 
@@ -1166,17 +1169,20 @@ void APIConnection::on_camera_image_request(const CameraImageRequest &msg) {
 // Chunk size per GetYamlResponse. Small enough to leave room for the protobuf frame
 // inside the 65535-byte API limit and friendly to TCP MSS.
 static constexpr size_t STORE_YAML_CHUNK_SIZE = 512;
-// Scratch buffer used to copy a chunk from PROGMEM (needed on ESP8266; harmless
-// elsewhere). Shared across connections is safe because the API loop is
-// single-threaded and each chunk is filled and consumed atomically inside one
-// `try_send_store_yaml_` iteration.
+#ifdef USE_ESP8266
+// On ESP8266 the blob lives in instruction flash and can't be read directly, so
+// each chunk is bounced through this static buffer via progmem_memcpy. Shared
+// across connections is safe because the API loop is single-threaded and each
+// chunk is filled and consumed atomically inside one `try_send_store_yaml_`
+// iteration. Every other platform sends straight from the blob, zero-copy.
 // NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
 static uint8_t store_yaml_chunk_buf[STORE_YAML_CHUNK_SIZE];
+#endif
 
 void APIConnection::on_get_yaml_request() {
-  auto *comp = store_yaml::global_store_yaml;
-  if (comp == nullptr || comp->get_size() == 0) {
-    // No blob — send a single done=true response so the client doesn't hang.
+  if (store_yaml::global_store_yaml == nullptr) {
+    // Request arrived before the component's setup() ran — send a single
+    // done=true response so the client doesn't hang.
     GetYamlResponse resp;
     resp.done = true;
     this->send_message(resp);
@@ -1186,18 +1192,10 @@ void APIConnection::on_get_yaml_request() {
   this->try_send_store_yaml_();
 }
 
+// Caller guarantees: store_yaml_pos_ != SIZE_MAX (a stream is in progress), which
+// implies on_get_yaml_request() already saw the never-cleared global non-null.
 void APIConnection::try_send_store_yaml_() {
-  if (this->store_yaml_pos_ == std::numeric_limits<size_t>::max())
-    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;
-  }
-
   const size_t total = comp->get_size();
 
   // Camera-style streaming: advance the position only after a successful send,
@@ -1207,14 +1205,15 @@ void APIConnection::try_send_store_yaml_() {
       return;
 
     const size_t remaining = total - this->store_yaml_pos_;
-    const size_t to_send = remaining < STORE_YAML_CHUNK_SIZE ? remaining : STORE_YAML_CHUNK_SIZE;
-
-    // Copy a chunk out of PROGMEM into a stack buffer; on ESP8266 this routes
-    // through progmem_read_byte, on every other platform it's a plain byte copy.
-    comp->read_chunk(this->store_yaml_pos_, store_yaml_chunk_buf, to_send);
+    const size_t to_send = std::min(remaining, STORE_YAML_CHUNK_SIZE);
 
     GetYamlResponse resp;
+#ifdef USE_ESP8266
+    progmem_memcpy(store_yaml_chunk_buf, comp->get_data() + this->store_yaml_pos_, to_send);
     resp.set_data(store_yaml_chunk_buf, to_send);
+#else
+    resp.set_data(comp->get_data() + this->store_yaml_pos_, to_send);
+#endif
     if (this->store_yaml_pos_ == 0) {
       resp.total_size = static_cast<uint32_t>(total);
       resp.encoding = StringRef(store_yaml::ENCODING);
@@ -1876,7 +1875,8 @@ bool APIConnection::send_device_info_response_() {
   resp.has_deep_sleep = deep_sleep::global_has_deep_sleep;
 #endif
 #ifdef USE_STORE_YAML
-  resp.has_store_yaml = store_yaml::global_store_yaml != nullptr && store_yaml::global_store_yaml->get_size() > 0;
+  // Codegen always embeds a non-empty blob, so presence of the component implies data.
+  resp.has_store_yaml = store_yaml::global_store_yaml != nullptr;
 #endif
 #ifdef ESPHOME_PROJECT_NAME
 #ifdef USE_ESP8266

esphome/components/store_yaml/__init__.py

@@ -4,15 +4,21 @@ import logging
 import os
 from pathlib import Path
 import struct
-from types import ModuleType
 
+from esphome import yaml_util
 import esphome.codegen as cg
+from esphome.components.api import CONF_ENCRYPTION
 import esphome.config_validation as cv
 from esphome.const import CONF_API, CONF_ID, CONF_RAW_DATA_ID
 from esphome.core import CORE, EsphomeError, HexInt
 import esphome.final_validate as fv
 from esphome.types import ConfigType
 
+try:
+    from compression import zstd  # Python 3.14+ stdlib
+except ImportError:
+    from backports import zstd  # pinned in requirements.txt for Python < 3.14
+
 _LOGGER = logging.getLogger(__name__)
 
 CODEOWNERS = ["@bdraco"]
@@ -47,11 +53,11 @@ CONFIG_SCHEMA = cv.Schema(
 def _final_validate(config: ConfigType) -> ConfigType:
     """Require API encryption: an unauthenticated client could otherwise pull
     the embedded YAML (which may include Wi-Fi credentials or opted-in
-    secrets). The escape hatch ``allow_unencrypted_api: true`` exists for
+    secrets). The escape hatch ``allow_unencrypted: true`` exists for
     isolated lab setups where the user has accepted the trade-off."""
     full = fv.full_config.get()
     api_conf = full.get(CONF_API, {})
-    if api_conf.get("encryption"):
+    if api_conf.get(CONF_ENCRYPTION):
         return config
     if config.get(CONF_ALLOW_UNENCRYPTED):
         _LOGGER.warning(
@@ -71,27 +77,14 @@ def _final_validate(config: ConfigType) -> ConfigType:
 FINAL_VALIDATE_SCHEMA = _final_validate
 
 
-def _import_zstd() -> ModuleType:
-    try:
-        from compression import zstd  # noqa: PLC0415 — Python 3.14+ stdlib
-    except ImportError:
-        try:
-            from backports import zstd  # noqa: PLC0415
-        except ImportError as err:
-            raise EsphomeError(
-                "store_yaml requires zstd compression. Install backports.zstd for "
-                "Python < 3.14 or upgrade to Python 3.14+."
-            ) from err
-    return zstd
-
-
-def _gather_files(include_secrets: bool) -> list[tuple[str, bytes]]:
-    """Read each YAML file the config loader touched, return (relative_path, content) pairs."""
-    discovered = CORE.data.get("yaml_sources")
-    if not discovered or not discovered.files:
+def _gather_files(
+    discovered: yaml_util.DiscoveredYamlFiles, include_secrets: bool
+) -> list[tuple[str, bytes]]:
+    """Read each discovered YAML file, return (relative_path, content) pairs."""
+    if not discovered.files:
         raise EsphomeError(
-            "store_yaml could not find any tracked YAML files; the config loader "
-            "did not populate CORE.data['yaml_sources']."
+            "store_yaml could not discover any YAML files for "
+            f"{CORE.config_path}; nothing to embed."
         )
 
     if discovered.unresolved:
@@ -157,12 +150,39 @@ def _pack_envelope(files: list[tuple[str, bytes]]) -> bytes:
     return b"".join(parts)
 
 
+def unpack_envelope(blob: bytes) -> dict[str, bytes]:
+    """Inverse of `_pack_envelope`: the reference decoder for the EHY1 envelope,
+    used by tests and client-side recovery tooling."""
+    if blob[:4] != ENVELOPE_MAGIC:
+        raise EsphomeError("envelope must start with EHY1 magic")
+    pos = 4
+    (count,) = struct.unpack_from("<I", blob, pos)
+    pos += 4
+    files: dict[str, bytes] = {}
+    for _ in range(count):
+        (path_len,) = struct.unpack_from("<H", blob, pos)
+        pos += 2
+        path = blob[pos : pos + path_len].decode("utf-8")
+        pos += path_len
+        (content_len,) = struct.unpack_from("<I", blob, pos)
+        pos += 4
+        files[path] = blob[pos : pos + content_len]
+        pos += content_len
+    if pos != len(blob):
+        raise EsphomeError("envelope has trailing bytes")
+    return files
+
+
 async def to_code(config: ConfigType) -> None:
     cg.add_define("USE_STORE_YAML")
 
-    zstd = _import_zstd()
-
-    files = _gather_files(config[CONF_INCLUDE_SECRETS])
+    # Discover the user's on-disk YAML files via a fresh re-parse — same
+    # pattern bundle.py uses. Running at codegen time (rather than keeping a
+    # listener installed across validation) avoids capturing framework YAML
+    # that components load internally (e.g. LVGL's `hello_world.yaml`), and
+    # costs nothing on validate-only runs or configs without this component.
+    discovered = yaml_util.discover_user_yaml_files(CORE.config_path)
+    files = _gather_files(discovered, config[CONF_INCLUDE_SECRETS])
     envelope = _pack_envelope(files)
     compressed = zstd.compress(envelope, level=ZSTD_LEVEL)
 
@@ -171,7 +191,7 @@ async def to_code(config: ConfigType) -> None:
         len(files),
         len(compressed),
         len(envelope),
-        100.0 * len(compressed) / max(1, len(envelope)),
+        100.0 * len(compressed) / len(envelope),
     )
 
     rhs = [HexInt(b) for b in compressed]

esphome/components/store_yaml/store_yaml.cpp

@@ -3,10 +3,6 @@
 #ifdef USE_STORE_YAML
 
 #include "esphome/core/log.h"
-#include <cstring>
-#ifdef USE_ESP8266
-#include <pgmspace.h>
-#endif
 
 namespace esphome::store_yaml {
 
@@ -26,18 +22,6 @@ void StoreYamlComponent::dump_config() {
                 this->size_, this->uncompressed_size_, ENCODING);
 }
 
-void StoreYamlComponent::read_chunk(size_t pos, uint8_t *dst, size_t len) const {
-#ifdef USE_ESP8266
-  // ESP8266 needs `memcpy_P` for aligned bulk flash reads; the byte-by-byte
-  // `progmem_read_byte` loop would otherwise emit ~4x as many flash accesses.
-  memcpy_P(dst, this->data_ + pos, len);
-#else
-  // PROGMEM is a no-op everywhere else and the data lives in normal address
-  // space, so a plain `std::memcpy` is correct and the fast path.
-  std::memcpy(dst, this->data_ + pos, len);
-#endif
-}
-
 }  // namespace esphome::store_yaml
 
 #endif  // USE_STORE_YAML

esphome/components/store_yaml/store_yaml.h

@@ -23,18 +23,14 @@ class StoreYamlComponent : public Component {
     this->uncompressed_size_ = uncompressed_size;
   }
   size_t get_size() const { return this->size_; }
-  size_t get_uncompressed_size() const { return this->uncompressed_size_; }
 
-  // Copy `len` bytes from the PROGMEM blob at offset `pos` into `dst`.
-  // Hides the platform-specific read (no-op everywhere except ESP8266, where the
-  // blob lives in code space and must be read through `progmem_read_byte`).
-  void read_chunk(size_t pos, uint8_t *dst, size_t len) const;
+  // Raw pointer to the PROGMEM blob. On ESP8266 the address is in instruction
+  // flash and must be read via `progmem_memcpy`; everywhere else PROGMEM is a
+  // no-op and the data is directly addressable.
+  const uint8_t *get_data() const { return this->data_; }
 
  protected:
-  // Points to a `const uint8_t[] PROGMEM` array emitted by codegen. On ESP8266 this
-  // address is in instruction flash and must be accessed via `progmem_read_byte` —
-  // hence the `read_chunk` accessor above. There is no public getter for the raw
-  // pointer; callers must go through `read_chunk`.
+  // Points to a `const uint8_t[] PROGMEM` array emitted by codegen.
   const uint8_t *data_{nullptr};
   size_t size_{0};
   size_t uncompressed_size_{0};

esphome/config.py

@@ -1332,26 +1332,13 @@ def _load_config(
         raise InvalidYAMLError(e) from e
 
     try:
-        result = validate_config(
-            config, command_line_substitutions, skip_external_update
-        )
+        return validate_config(config, command_line_substitutions, skip_external_update)
     except EsphomeError:
         raise
     except Exception:
         _LOGGER.error("Unexpected exception while reading configuration:")
         raise
 
-    # 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`). 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
-
 
 def load_config(
     command_line_substitutions: dict[str, Any], skip_external_update: bool = False

esphome/yaml_util.py

@@ -271,8 +271,6 @@ def force_load_include_files(
     obj: Any,
     *,
     warn_on_unresolved: bool = True,
-    _seen: set[int] | None = None,
-    _unresolved: list[str] | None = None,
 ) -> list[str]:
     """Recursively resolve any deferred ``IncludeFile`` instances in a YAML tree.
 
@@ -289,63 +287,43 @@ def force_load_include_files(
     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 = []
+    seen: set[int] = set()
+    unresolved: list[str] = []
 
-    if isinstance(obj, IncludeFile):
-        if id(obj) in _seen:
-            return _unresolved
-        _seen.add(id(obj))
-        if obj.has_unresolved_expressions():
-            log = _LOGGER.warning if warn_on_unresolved else _LOGGER.debug
-            log(
-                "Cannot resolve !include %s (referenced from %s) with substitutions in path",
-                obj.file,
-                obj.parent_file,
-            )
-            _unresolved.append(str(obj.file))
-            return _unresolved
-        try:
-            loaded = obj.load()
-        except EsphomeError as err:
-            _LOGGER.warning(
-                "Failed to load !include %s (referenced from %s): %s",
-                obj.file,
-                obj.parent_file,
-                err,
-            )
-            return _unresolved
-        force_load_include_files(
-            loaded,
-            warn_on_unresolved=warn_on_unresolved,
-            _seen=_seen,
-            _unresolved=_unresolved,
-        )
-    elif isinstance(obj, dict):
-        if id(obj) in _seen:
-            return _unresolved
-        _seen.add(id(obj))
-        for value in obj.values():
-            force_load_include_files(
-                value,
-                warn_on_unresolved=warn_on_unresolved,
-                _seen=_seen,
-                _unresolved=_unresolved,
-            )
-    elif isinstance(obj, (list, tuple)):
-        if id(obj) in _seen:
-            return _unresolved
-        _seen.add(id(obj))
-        for item in obj:
-            force_load_include_files(
-                item,
-                warn_on_unresolved=warn_on_unresolved,
-                _seen=_seen,
-                _unresolved=_unresolved,
-            )
-    return _unresolved
+    def walk(node: Any) -> None:
+        if not isinstance(node, (IncludeFile, dict, list, tuple)) or id(node) in seen:
+            return
+        seen.add(id(node))
+        if isinstance(node, IncludeFile):
+            if node.has_unresolved_expressions():
+                log = _LOGGER.warning if warn_on_unresolved else _LOGGER.debug
+                log(
+                    "Cannot resolve !include %s (referenced from %s) with substitutions in path",
+                    node.file,
+                    node.parent_file,
+                )
+                unresolved.append(str(node.file))
+                return
+            try:
+                loaded = node.load()
+            except EsphomeError as err:
+                _LOGGER.warning(
+                    "Failed to load !include %s (referenced from %s): %s",
+                    node.file,
+                    node.parent_file,
+                    err,
+                )
+                return
+            walk(loaded)
+        elif isinstance(node, dict):
+            for value in node.values():
+                walk(value)
+        else:
+            for item in node:
+                walk(item)
+
+    walk(obj)
+    return unresolved
 
 
 @dataclass(slots=True)

tests/integration/test_store_yaml_recovery.py

@@ -16,7 +16,6 @@ from __future__ import annotations
 
 import asyncio
 import contextlib
-import struct
 
 import pytest
 
@@ -25,6 +24,8 @@ try:
 except ImportError:
     from backports import zstd  # type: ignore[import-not-found, no-redef]
 
+from esphome.components.store_yaml import unpack_envelope
+
 from .types import RunCompiledFunction
 
 # Message IDs from esphome/components/api/api.proto.
@@ -33,8 +34,6 @@ HELLO_RESPONSE = 2
 GET_YAML_REQUEST = 149
 GET_YAML_RESPONSE = 150
 
-ENVELOPE_MAGIC = b"EHY1"
-
 
 def _encode_varint(value: int) -> bytes:
     """Encode an unsigned integer as a protobuf varint."""
@@ -94,25 +93,16 @@ def _parse_get_yaml_response(payload: bytes) -> tuple[bytes, bool, int, str]:
     return data, done, total_size, encoding
 
 
-def _unpack_envelope(blob: bytes) -> dict[str, bytes]:
-    """Inverse of `_pack_envelope` in `esphome/components/store_yaml/__init__.py`."""
-    assert blob[:4] == ENVELOPE_MAGIC, "envelope must start with EHY1 magic"
-    pos = 4
-    (count,) = struct.unpack_from("<I", blob, pos)
-    pos += 4
-    files: dict[str, bytes] = {}
-    for _ in range(count):
-        (path_len,) = struct.unpack_from("<H", blob, pos)
-        pos += 2
-        path = blob[pos : pos + path_len].decode("utf-8")
-        pos += path_len
-        (content_len,) = struct.unpack_from("<I", blob, pos)
-        pos += 4
-        content = blob[pos : pos + content_len]
-        pos += content_len
-        files[path] = content
-    assert pos == len(blob), "envelope must consume all bytes"
-    return files
+async def _read_varint_from(reader: asyncio.StreamReader) -> int:
+    """Read a protobuf varint byte-by-byte from a stream."""
+    result = 0
+    shift = 0
+    while True:
+        byte = (await reader.readexactly(1))[0]
+        result |= (byte & 0x7F) << shift
+        if not (byte & 0x80):
+            return result
+        shift += 7
 
 
 class _PlaintextClient:
@@ -137,18 +127,8 @@ class _PlaintextClient:
         preamble = await self._reader.readexactly(1)
         assert preamble == b"\x00", f"unexpected preamble {preamble!r}"
 
-        async def _read_varint_stream() -> int:
-            result = 0
-            shift = 0
-            while True:
-                byte = (await self._reader.readexactly(1))[0]
-                result |= (byte & 0x7F) << shift
-                if not (byte & 0x80):
-                    return result
-                shift += 7
-
-        payload_size = await _read_varint_stream()
-        msg_type = await _read_varint_stream()
+        payload_size = await _read_varint_from(self._reader)
+        msg_type = await _read_varint_from(self._reader)
         payload = await self._reader.readexactly(payload_size) if payload_size else b""
         return msg_type, payload
 
@@ -215,7 +195,7 @@ async def test_store_yaml_recovery(
     )
 
     envelope = zstd.decompress(compressed)
-    files = _unpack_envelope(envelope)
+    files = unpack_envelope(envelope)
 
     assert files, "envelope should contain at least one file"
     combined = b"\n".join(files.values())

tests/unit_tests/components/test_store_yaml.py

@@ -3,41 +3,19 @@
 from __future__ import annotations
 
 from pathlib import Path
-import struct
 
 import pytest
 
 from esphome.components.store_yaml import (
-    ENVELOPE_MAGIC,
     REDACTED_PLACEHOLDER,
     _gather_files,
     _pack_envelope,
+    unpack_envelope,
 )
 from esphome.core import CORE, EsphomeError
 from esphome.yaml_util import DiscoveredYamlFiles
 
 
-def _unpack_envelope(blob: bytes) -> dict[str, bytes]:
-    """Inverse of `_pack_envelope` for assertions in tests."""
-    assert blob[:4] == ENVELOPE_MAGIC, "envelope must start with EHY1 magic"
-    pos = 4
-    (count,) = struct.unpack_from("<I", blob, pos)
-    pos += 4
-    files: dict[str, bytes] = {}
-    for _ in range(count):
-        (path_len,) = struct.unpack_from("<H", blob, pos)
-        pos += 2
-        path = blob[pos : pos + path_len].decode("utf-8")
-        pos += path_len
-        (content_len,) = struct.unpack_from("<I", blob, pos)
-        pos += 4
-        content = blob[pos : pos + content_len]
-        pos += content_len
-        files[path] = content
-    assert pos == len(blob), "envelope must consume all bytes"
-    return files
-
-
 @pytest.fixture
 def project(tmp_path: Path) -> Path:
     """Lay out a tiny ESPHome-like project: entry yaml, an include, and a secrets file."""
@@ -49,31 +27,24 @@ def project(tmp_path: Path) -> Path:
     return project_dir
 
 
-@pytest.fixture(autouse=True)
-def _reset_core() -> None:
-    CORE.data.pop("yaml_sources", None)
-    CORE.config_path = None
-    yield
-    CORE.data.pop("yaml_sources", None)
-    CORE.config_path = None
-
-
-def _set_sources(project_dir: Path, *names: str, secrets: tuple[str, ...] = ()) -> None:
+def _sources(
+    project_dir: Path, *names: str, secrets: tuple[str, ...] = ()
+) -> DiscoveredYamlFiles:
     CORE.config_path = project_dir / "entry.yaml"
     files = [project_dir / name for name in names]
     secret_paths = {(project_dir / name).resolve() for name in secrets}
-    CORE.data["yaml_sources"] = DiscoveredYamlFiles(files, secret_paths)
+    return DiscoveredYamlFiles(files, secret_paths)
 
 
 def test_gather_redacts_secrets_by_default(project: Path) -> None:
-    _set_sources(
+    discovered = _sources(
         project,
         "entry.yaml",
         "wifi.yaml",
         "secrets.yaml",
         secrets=("secrets.yaml",),
     )
-    files = dict(_gather_files(include_secrets=False))
+    files = dict(_gather_files(discovered, include_secrets=False))
     assert files["secrets.yaml"] == REDACTED_PLACEHOLDER
     assert b"SUPER_SECRET" not in files["secrets.yaml"]
     assert files["wifi.yaml"] == (project / "wifi.yaml").read_bytes()
@@ -82,8 +53,10 @@ def test_gather_redacts_secrets_by_default(project: Path) -> None:
 def test_gather_redacts_yml_extension(project: Path) -> None:
     yml = project / "secrets.yml"
     yml.write_text("api_key: OTHER_SECRET\n")
-    _set_sources(project, "entry.yaml", "secrets.yml", secrets=("secrets.yml",))
-    files = dict(_gather_files(include_secrets=False))
+    discovered = _sources(
+        project, "entry.yaml", "secrets.yml", secrets=("secrets.yml",)
+    )
+    files = dict(_gather_files(discovered, include_secrets=False))
     assert files["secrets.yml"] == REDACTED_PLACEHOLDER
 
 
@@ -101,15 +74,17 @@ def test_gather_redacts_secret_symlinked_to_other_name(
     # but stores the resolved path; mimic that here.
     resolved = link.resolve()
     CORE.config_path = project / "entry.yaml"
-    CORE.data["yaml_sources"] = DiscoveredYamlFiles([resolved], {resolved})
-    files = dict(_gather_files(include_secrets=False))
+    discovered = DiscoveredYamlFiles([resolved], {resolved})
+    files = dict(_gather_files(discovered, include_secrets=False))
     assert REDACTED_PLACEHOLDER in files.values()
     assert b"FROM_SYMLINK" not in b"".join(files.values())
 
 
 def test_gather_embeds_secrets_when_opted_in(project: Path) -> None:
-    _set_sources(project, "entry.yaml", "secrets.yaml", secrets=("secrets.yaml",))
-    files = dict(_gather_files(include_secrets=True))
+    discovered = _sources(
+        project, "entry.yaml", "secrets.yaml", secrets=("secrets.yaml",)
+    )
+    files = dict(_gather_files(discovered, include_secrets=True))
     assert b"SUPER_SECRET" in files["secrets.yaml"]
 
 
@@ -120,10 +95,8 @@ def test_gather_uses_relative_path_for_external_files(
     sibling = tmp_path / "outside.yaml"
     sibling.write_text("foo: bar\n")
     CORE.config_path = project / "entry.yaml"
-    CORE.data["yaml_sources"] = DiscoveredYamlFiles(
-        [project / "entry.yaml", sibling], set()
-    )
-    files = dict(_gather_files(include_secrets=False))
+    discovered = DiscoveredYamlFiles([project / "entry.yaml", sibling], set())
+    files = dict(_gather_files(discovered, include_secrets=False))
     # project root is `tmp_path/project`, sibling is in `tmp_path` so it
     # resolves to `../outside.yaml`.
     assert "../outside.yaml" in files
@@ -132,7 +105,7 @@ def test_gather_uses_relative_path_for_external_files(
 def test_gather_raises_when_no_sources(project: Path) -> None:
     CORE.config_path = project / "entry.yaml"
     with pytest.raises(EsphomeError):
-        _gather_files(include_secrets=False)
+        _gather_files(DiscoveredYamlFiles(), include_secrets=False)
 
 
 def test_gather_raises_on_unreadable_file(
@@ -140,7 +113,7 @@ def test_gather_raises_on_unreadable_file(
 ) -> None:
     """An unreadable tracked file fails the build instead of producing a
     silently partial recovery blob."""
-    _set_sources(project, "entry.yaml", "wifi.yaml")
+    discovered = _sources(project, "entry.yaml", "wifi.yaml")
     orig_read_bytes = Path.read_bytes
 
     def fake_read_bytes(self: Path) -> bytes:
@@ -150,7 +123,7 @@ def test_gather_raises_on_unreadable_file(
 
     monkeypatch.setattr(Path, "read_bytes", fake_read_bytes)
     with pytest.raises(EsphomeError, match="wifi.yaml"):
-        _gather_files(include_secrets=False)
+        _gather_files(discovered, include_secrets=False)
 
 
 def test_gather_warns_on_unresolved_includes(
@@ -159,11 +132,9 @@ def test_gather_warns_on_unresolved_includes(
     """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"]
-    )
+    discovered = DiscoveredYamlFiles([project / "entry.yaml"], set(), ["${board}.yaml"])
     with caplog.at_level("WARNING", logger="esphome.components.store_yaml"):
-        files = _gather_files(include_secrets=False)
+        files = _gather_files(discovered, include_secrets=False)
     assert len(files) == 1
     assert any(
         "${board}.yaml" in r.message and "not contain" in r.message
@@ -177,16 +148,21 @@ def test_pack_envelope_roundtrip() -> None:
         ("wifi.yaml", b"ssid: a\n"),
     ]
     blob = _pack_envelope(files)
-    assert _unpack_envelope(blob) == dict(files)
+    assert unpack_envelope(blob) == dict(files)
 
 
 def test_pack_envelope_handles_utf8_paths() -> None:
     files = [("dossiers/maison.yaml", b"foo: bar\n")]
     blob = _pack_envelope(files)
-    assert _unpack_envelope(blob) == dict(files)
+    assert unpack_envelope(blob) == dict(files)
 
 
 def test_pack_envelope_rejects_overlong_path() -> None:
     long_path = "a" * (0xFFFF + 1)
     with pytest.raises(EsphomeError):
         _pack_envelope([(long_path, b"")])
+
+
+def test_unpack_envelope_rejects_bad_magic() -> None:
+    with pytest.raises(EsphomeError):
+        unpack_envelope(b"NOPE" + b"\x00" * 4)

tests/unit_tests/test_config_normalization.py

@@ -151,27 +151,6 @@ def test_validate_config_warns_on_dropped_merge_key(
     assert yaml_util.take_dropped_merge_keys() == []
 
 
-@pytest.mark.parametrize("store_yaml_enabled", [False, True])
-def test_load_config_gates_yaml_discovery_on_store_yaml(
-    tmp_path: Path, store_yaml_enabled: bool
-) -> None:
-    """YAML source discovery (a second full parse) only runs when a consumer
-    like store_yaml is configured."""
-    content = "esphome:\n  name: test\nhost:\napi:\n"
-    if store_yaml_enabled:
-        content += "store_yaml:\n  allow_unencrypted: true\n"
-    main = tmp_path / "main.yaml"
-    main.write_text(content)
-    CORE.config_path = main
-
-    result = config.load_config({})
-
-    assert not result.errors
-    assert ("yaml_sources" in CORE.data) == store_yaml_enabled
-    if store_yaml_enabled:
-        assert main.resolve() in CORE.data["yaml_sources"].files
-
-
 def test_validate_config_suppresses_merge_warning(
     tmp_path: Path, caplog: pytest.LogCaptureFixture
 ) -> None:

tests/unit_tests/test_yaml_util.py

@@ -1111,7 +1111,7 @@ def test_force_load_include_files_warns_on_load_failure(
 
 
 def test_discovered_yaml_files_holds_files_and_secrets() -> None:
-    """`DiscoveredYamlFiles` is a small data carrier; both fields are mandatory."""
+    """`DiscoveredYamlFiles` is a small data carrier."""
     files = [Path("/tmp/a.yaml")]
     secrets = {Path("/tmp/a.yaml")}
     discovered = DiscoveredYamlFiles(files, secrets)