[store_yaml] Recover secrets as !secret references using cv.sensitive

esphome/bundle.py

@@ -98,15 +98,6 @@ _KNOWN_FILE_EXTENSIONS = frozenset(
 )
 
 
-# Matches !secret references in YAML text.  An optional surrounding
-# quote pair around the key is allowed and ignored: YAML treats
-# ``!secret 'foo'`` and ``!secret foo`` as the same key.  This is
-# intentionally a simple regex scan rather than a YAML parse — it may
-# match inside comments or multi-line strings, which is the conservative
-# direction (include more secrets rather than fewer).
-_SECRET_RE = re.compile(r"""!secret\s+['"]?([^\s'"]+)""")
-
-
 def _find_used_secret_keys(yaml_files: list[Path]) -> set[str]:
     """Scan YAML files for ``!secret <key>`` references."""
     keys: set[str] = set()
@@ -115,8 +106,7 @@ def _find_used_secret_keys(yaml_files: list[Path]) -> set[str]:
             text = fpath.read_text(encoding="utf-8")
         except (OSError, UnicodeDecodeError):
             continue
-        for match in _SECRET_RE.finditer(text):
-            keys.add(match.group(1))
+        keys |= yaml_util.find_secret_references(text)
     return keys
 
 

esphome/components/api/api_connection.cpp

@@ -1180,9 +1180,10 @@ static uint8_t store_yaml_chunk_buf[STORE_YAML_CHUNK_SIZE];
 #endif
 
 void APIConnection::on_get_yaml_request() {
-  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.
+  auto *comp = store_yaml::global_store_yaml;
+  if (comp == nullptr || comp->get_size() == 0) {
+    // No component yet (request before setup()) or empty blob — send a single
+    // done=true response so the client always gets a terminal frame.
     GetYamlResponse resp;
     resp.done = true;
     this->send_message(resp);

esphome/components/store_yaml/__init__.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from dataclasses import dataclass
 import logging
 import os
 from pathlib import Path
@@ -37,8 +38,12 @@ StoreYamlComponent = store_yaml_ns.class_("StoreYamlComponent", cg.Component)
 ZSTD_LEVEL = 22
 # Envelope magic: "EHY1" = ESPHome YAML, version 1.
 ENVELOPE_MAGIC = b"EHY1"
-# Replacement content when secrets are not included.
-REDACTED_PLACEHOLDER = b"# redacted\n"
+# Replacement content for secrets files: a fill-in skeleton listing every
+# `!secret` key the recovered config needs.
+SECRETS_SKELETON_HEADER = (
+    "# Redacted by store_yaml. Fill in these values and the recovered\n"
+    "# config is ready to flash.\n"
+)
 
 CONFIG_SCHEMA = cv.Schema(
     {
@@ -78,15 +83,28 @@ FINAL_VALIDATE_SCHEMA = _final_validate
 
 
 def _gather_files(
-    discovered: yaml_util.DiscoveredYamlFiles, include_secrets: bool
-) -> list[tuple[str, bytes]]:
-    """Read each discovered YAML file, return (relative_path, content) pairs."""
+    discovered: yaml_util.DiscoveredYamlFiles,
+) -> tuple[list[tuple[str, bytes]], set[str]]:
+    """Read each discovered YAML file verbatim.
+
+    Returns (relative_path, content) pairs plus the subset of relative paths
+    that are secrets files (matched upstream on the *un-resolved* basename, so
+    a `secrets.yaml` symlinked to a differently-named target is still flagged).
+    """
     if not discovered.files:
         raise EsphomeError(
             "store_yaml could not discover any YAML files for "
             f"{CORE.config_path}; nothing to embed."
         )
 
+    if discovered.load_errors:
+        # A silently partial recovery blob defeats the feature; fail the build
+        # instead of embedding an incomplete file set.
+        raise EsphomeError(
+            "store_yaml: could not load all configuration files: "
+            + "; ".join(discovered.load_errors)
+        )
+
     if discovered.unresolved:
         _LOGGER.warning(
             "store_yaml: %d !include path(s) use substitutions and cannot be "
@@ -97,24 +115,16 @@ def _gather_files(
 
     config_path = Path(CORE.config_path).resolve()
     root = config_path.parent
-    secret_paths = discovered.secrets
 
     files: list[tuple[str, bytes]] = []
+    secret_rels: set[str] = set()
     for path in discovered.files:
-        # `secret_paths` was collected from the *un-resolved* basename, so a
-        # `secrets.yaml` symlinked to a differently-named target is still
-        # treated as secrets here.
-        if path in secret_paths and not include_secrets:
-            content = REDACTED_PLACEHOLDER
-        else:
-            try:
-                content = path.read_bytes()
-            except OSError as err:
-                # 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:
+            content = path.read_bytes()
+        except OSError as err:
+            raise EsphomeError(
+                f"store_yaml: cannot read tracked YAML file {path}: {err}"
+            ) from err
 
         try:
             rel_str = path.relative_to(root).as_posix()
@@ -125,9 +135,119 @@ def _gather_files(
             # different directories with the same basename don't collide.
             rel_str = os.path.relpath(path, root).replace(os.sep, "/")
 
+        if path in discovered.secrets:
+            secret_rels.add(rel_str)
         files.append((rel_str, content))
 
-    return files
+    return files, secret_rels
+
+
+def _iter_sensitive_values(node: object, path: tuple[str, ...] = ()):
+    """Yield (config_path, value) for every cv.sensitive value in a config tree."""
+    if isinstance(node, yaml_util.SensitiveStr):
+        yield path, str(node)
+    elif isinstance(node, dict):
+        for key, value in node.items():
+            yield from _iter_sensitive_values(value, (*path, str(key)))
+    elif isinstance(node, (list, tuple)):
+        for item in node:
+            yield from _iter_sensitive_values(item, path)
+
+
+@dataclass
+class _SensitiveValue:
+    secret_name: str
+    config_path: str  # dotted path, for warnings (never log the value itself)
+    from_secret: bool  # already loaded via !secret somewhere
+
+
+def _collect_sensitive_values(reserved_names: set[str]) -> dict[str, _SensitiveValue]:
+    """Map each cv.sensitive value in the validated config to the `!secret`
+    name it should be recovered as.
+
+    Values that already come from `!secret` keep their existing name; inline
+    values get a name generated from their config path, avoiding
+    `reserved_names`.
+    """
+    used = set(reserved_names)
+    result: dict[str, _SensitiveValue] = {}
+    for path, value in _iter_sensitive_values(CORE.config):
+        if not value or value in result:
+            continue
+        if existing := yaml_util.is_secret(value):
+            name = existing
+        else:
+            base = "_".join(path) or "secret"
+            name = base
+            counter = 2
+            while name in used:
+                name = f"{base}_{counter}"
+                counter += 1
+        used.add(name)
+        result[value] = _SensitiveValue(name, ".".join(path), bool(existing))
+    return result
+
+
+def _build_secrets_skeleton(keys: set[str]) -> bytes:
+    parts = [SECRETS_SKELETON_HEADER]
+    parts.extend(f'{key}: ""\n' for key in sorted(keys))
+    return "".join(parts).encode("utf-8")
+
+
+def _generate_redacted_files(
+    files: list[tuple[str, bytes]], secret_rels: set[str]
+) -> list[tuple[str, bytes]]:
+    """Re-generate each captured file from its parse tree with cv.sensitive
+    values emitted as `!secret <name>` references, and replace secrets files
+    with a fill-in skeleton — the recovered config is flashable once the user
+    restores their secrets.yaml values.
+
+    The swap happens inside the YAML dumper (`represent_stringify` consults
+    the registered secret values), not by mutating text afterwards. Nested
+    `!include` references round-trip via the dumper's IncludeFile support;
+    comments and formatting of the originals are not preserved.
+    """
+    sensitive = _collect_sensitive_values(yaml_util.registered_secret_names())
+    inline = {
+        value: info.secret_name
+        for value, info in sensitive.items()
+        if not info.from_secret
+    }
+
+    config_path = Path(CORE.config_path).resolve()
+    root = config_path.parent
+    texts: dict[str, str] = {}
+    with yaml_util.secret_values_registered(inline):
+        for rel, _ in files:
+            if rel in secret_rels:
+                continue
+            tree = yaml_util.load_yaml(root / rel, clear_secrets=False)
+            texts[rel] = yaml_util.dump(tree)
+
+    skeleton_keys: set[str] = set()
+    for text in texts.values():
+        skeleton_keys |= yaml_util.find_secret_references(text)
+
+    for info in sensitive.values():
+        if not info.from_secret and info.secret_name not in skeleton_keys:
+            _LOGGER.warning(
+                "store_yaml: could not locate the sensitive value of '%s' in the "
+                "source YAML (built via substitutions?); it may still be "
+                "embedded verbatim",
+                info.config_path,
+            )
+
+    skeleton = _build_secrets_skeleton(skeleton_keys)
+    result = [
+        (rel, skeleton if rel in secret_rels else texts[rel].encode("utf-8"))
+        for rel, _ in files
+    ]
+    if skeleton_keys and not secret_rels:
+        # The generated files reference `!secret` keys but the project has no
+        # secrets file (all secrets were inline) — ship a synthetic one so the
+        # recovered config is complete.
+        result.append(("secrets.yaml", skeleton))
+    return result
 
 
 def _pack_envelope(files: list[tuple[str, bytes]]) -> bytes:
@@ -156,18 +276,25 @@ def unpack_envelope(blob: bytes) -> dict[str, bytes]:
     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)
+    try:
+        (count,) = struct.unpack_from("<I", blob, pos)
         pos += 4
-        files[path] = blob[pos : pos + content_len]
-        pos += content_len
+        for _ in range(count):
+            (path_len,) = struct.unpack_from("<H", blob, pos)
+            pos += 2
+            if pos + path_len > len(blob):
+                raise EsphomeError("truncated envelope")
+            path = blob[pos : pos + path_len].decode("utf-8")
+            pos += path_len
+            (content_len,) = struct.unpack_from("<I", blob, pos)
+            pos += 4
+            if pos + content_len > len(blob):
+                raise EsphomeError("truncated envelope")
+            files[path] = blob[pos : pos + content_len]
+            pos += content_len
+    except struct.error as err:
+        raise EsphomeError(f"truncated envelope: {err}") from err
     if pos != len(blob):
         raise EsphomeError("envelope has trailing bytes")
     return files
@@ -182,7 +309,9 @@ async def to_code(config: ConfigType) -> None:
     # 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])
+    files, secret_rels = _gather_files(discovered)
+    if not config[CONF_INCLUDE_SECRETS]:
+        files = _generate_redacted_files(files, secret_rels)
     envelope = _pack_envelope(files)
     compressed = zstd.compress(envelope, level=ZSTD_LEVEL)
 

esphome/yaml_util.py

@@ -11,7 +11,8 @@ import logging
 import math
 import os
 from pathlib import Path
-from typing import Any
+import re
+from typing import Any, NamedTuple
 import uuid
 
 import yaml
@@ -267,11 +268,37 @@ class IncludeFile:
         return has_substitution_or_expression(str(self.file))
 
 
+# Matches !secret references in YAML text.  An optional surrounding
+# quote pair around the key is allowed and ignored: YAML treats
+# ``!secret 'foo'`` and ``!secret foo`` as the same key.  This is
+# intentionally a simple regex scan rather than a YAML parse — it may
+# match inside comments or multi-line strings, which is the conservative
+# direction (include more secrets rather than fewer).
+_SECRET_REFERENCE_RE = re.compile(r"""!secret\s+['"]?([^\s'"]+)""")
+
+
+def find_secret_references(text: str) -> set[str]:
+    """Return the ``!secret <key>`` names referenced in a YAML document text."""
+    return {match.group(1) for match in _SECRET_REFERENCE_RE.finditer(text)}
+
+
+class ForceLoadResult(NamedTuple):
+    """Outcome of :func:`force_load_include_files`.
+
+    ``unresolved`` lists ``!include`` path strings that contain substitution
+    variables and therefore could not be loaded; ``errors`` lists includes
+    that failed to load. Either being non-empty means the walk was incomplete.
+    """
+
+    unresolved: list[str]
+    errors: list[str]
+
+
 def force_load_include_files(
     obj: Any,
     *,
     warn_on_unresolved: bool = True,
-) -> list[str]:
+) -> ForceLoadResult:
     """Recursively resolve any deferred ``IncludeFile`` instances in a YAML tree.
 
     Nested ``!include`` returns a deferred ``IncludeFile`` that is only resolved
@@ -284,11 +311,11 @@ 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. Returns the path strings of those unloadable includes so
-    callers can tell the walk was incomplete.
+    to a debug log.
     """
     seen: set[int] = set()
     unresolved: list[str] = []
+    errors: list[str] = []
 
     def walk(node: Any) -> None:
         if not isinstance(node, (IncludeFile, dict, list, tuple)) or id(node) in seen:
@@ -313,6 +340,7 @@ def force_load_include_files(
                     node.parent_file,
                     err,
                 )
+                errors.append(f"{node.file}: {err}")
                 return
             walk(loaded)
         elif isinstance(node, dict):
@@ -323,7 +351,7 @@ def force_load_include_files(
                 walk(item)
 
     walk(obj)
-    return unresolved
+    return ForceLoadResult(unresolved, errors)
 
 
 @dataclass(slots=True)
@@ -335,13 +363,15 @@ class DiscoveredYamlFiles:
     *un-resolved* filename matched :data:`esphome.const.SECRETS_FILES` (so
     a ``secrets.yaml`` symlinked to a differently-named target is still
     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.
+    contain substitution variables and therefore could not be loaded, and
+    ``load_errors`` lists files that failed to parse or load — consumers
+    should treat ``files`` as incomplete when either is non-empty.
     """
 
     files: list[Path] = field(default_factory=list)
     secrets: set[Path] = field(default_factory=set)
     unresolved: list[str] = field(default_factory=list)
+    load_errors: list[str] = field(default_factory=list)
 
 
 def discover_user_yaml_files(config_path: Path) -> DiscoveredYamlFiles:
@@ -371,9 +401,16 @@ def discover_user_yaml_files(config_path: Path) -> DiscoveredYamlFiles:
         try:
             try:
                 data = load_yaml(config_path)
-            except EsphomeError:
-                return DiscoveredYamlFiles(list(loaded), secrets)
-            unresolved = force_load_include_files(data, warn_on_unresolved=False)
+            except EsphomeError as err:
+                _LOGGER.warning(
+                    "YAML discovery failed to parse %s: %s", config_path, err
+                )
+                return DiscoveredYamlFiles(
+                    list(loaded), secrets, load_errors=[f"{config_path}: {err}"]
+                )
+            unresolved, load_errors = force_load_include_files(
+                data, warn_on_unresolved=False
+            )
         finally:
             _load_listeners.remove(_capture_secret)
 
@@ -384,7 +421,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, unresolved)
+    return DiscoveredYamlFiles(unique, secrets, unresolved, load_errors)
 
 
 def _add_data_ref(fn):
@@ -836,6 +873,28 @@ def _load_yaml_internal_with_type(
         loader.dispose()
 
 
+def registered_secret_names() -> set[str]:
+    """Names of all ``!secret`` keys the loader has seen since the last clear."""
+    return set(_SECRET_VALUES.values())
+
+
+@contextmanager
+def secret_values_registered(values: dict[str, str]) -> Generator[None]:
+    """Temporarily register value→name mappings so :func:`dump` renders those
+    scalars as ``!secret <name>``.
+
+    Mappings already present in ``_SECRET_VALUES`` (values loaded through a
+    real ``!secret``) win over the supplied ones and are left untouched.
+    """
+    added = {v: n for v, n in values.items() if v not in _SECRET_VALUES}
+    _SECRET_VALUES.update(added)
+    try:
+        yield
+    finally:
+        for value in added:
+            _SECRET_VALUES.pop(value, None)
+
+
 def dump(dict_, show_secrets=False, sort_keys=False):
     """Dump YAML to a string and remove null."""
     if show_secrets:

tests/integration/fixtures/store_yaml_recovery.yaml

@@ -10,5 +10,9 @@ logger:
 
 api:
 
+ota:
+  - platform: esphome
+    password: recoverme123
+
 store_yaml:
   allow_unencrypted: true

tests/integration/test_store_yaml_recovery.py

@@ -205,3 +205,16 @@ async def test_store_yaml_recovery(
     assert b"store_yaml:" in combined, (
         "expected the store_yaml config line to be in the recovery blob"
     )
+
+    # The inline cv.sensitive OTA password must be recovered as a `!secret`
+    # reference, never as its raw value, and the synthetic secrets.yaml
+    # skeleton must list the key so the recovered config is flashable.
+    assert b"recoverme123" not in envelope, (
+        "inline sensitive value leaked into the recovery blob"
+    )
+    assert b"!secret 'ota_password'" in combined, (
+        "expected the inline OTA password to be recovered as a !secret reference"
+    )
+    assert b'ota_password: ""' in files["secrets.yaml"], (
+        "expected the secrets.yaml skeleton to list the ota_password key"
+    )

tests/unit_tests/components/test_store_yaml.py

@@ -1,4 +1,5 @@
-"""Tests for the store_yaml component's file gathering and envelope packing."""
+"""Tests for the store_yaml component's file gathering, secret redaction, and
+envelope packing."""
 
 from __future__ import annotations
 
@@ -6,14 +7,16 @@ from pathlib import Path
 
 import pytest
 
+from esphome import yaml_util
 from esphome.components.store_yaml import (
-    REDACTED_PLACEHOLDER,
+    SECRETS_SKELETON_HEADER,
     _gather_files,
+    _generate_redacted_files,
     _pack_envelope,
     unpack_envelope,
 )
 from esphome.core import CORE, EsphomeError
-from esphome.yaml_util import DiscoveredYamlFiles
+from esphome.yaml_util import DiscoveredYamlFiles, SensitiveStr
 
 
 @pytest.fixture
@@ -21,12 +24,21 @@ def project(tmp_path: Path) -> Path:
     """Lay out a tiny ESPHome-like project: entry yaml, an include, and a secrets file."""
     project_dir = tmp_path / "project"
     project_dir.mkdir()
-    (project_dir / "entry.yaml").write_text("esphome:\n  name: test\n")
+    (project_dir / "entry.yaml").write_text(
+        "esphome:\n  name: test\napi:\n  encryption:\n    key: !secret api_key\n"
+    )
     (project_dir / "wifi.yaml").write_text("ssid: my_ssid\npassword: my_password\n")
     (project_dir / "secrets.yaml").write_text("api_key: SUPER_SECRET\n")
     return project_dir
 
 
+@pytest.fixture(autouse=True)
+def _clear_config() -> None:
+    CORE.config = {}
+    yield
+    yaml_util._SECRET_VALUES.clear()
+
+
 def _sources(
     project_dir: Path, *names: str, secrets: tuple[str, ...] = ()
 ) -> DiscoveredYamlFiles:
@@ -36,34 +48,30 @@ def _sources(
     return DiscoveredYamlFiles(files, secret_paths)
 
 
-def test_gather_redacts_secrets_by_default(project: Path) -> None:
+def _gather_redacted(discovered: DiscoveredYamlFiles) -> dict[str, bytes]:
+    files, secret_rels = _gather_files(discovered)
+    return dict(_generate_redacted_files(files, secret_rels))
+
+
+# ---------------------------------------------------------------------------
+# _gather_files
+# ---------------------------------------------------------------------------
+
+
+def test_gather_returns_verbatim_content_and_flags_secrets(project: Path) -> None:
     discovered = _sources(
-        project,
-        "entry.yaml",
-        "wifi.yaml",
-        "secrets.yaml",
-        secrets=("secrets.yaml",),
+        project, "entry.yaml", "secrets.yaml", secrets=("secrets.yaml",)
     )
-    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()
+    files, secret_rels = _gather_files(discovered)
+    contents = dict(files)
+    assert contents["secrets.yaml"] == b"api_key: SUPER_SECRET\n"
+    assert secret_rels == {"secrets.yaml"}
 
 
-def test_gather_redacts_yml_extension(project: Path) -> None:
-    yml = project / "secrets.yml"
-    yml.write_text("api_key: OTHER_SECRET\n")
-    discovered = _sources(
-        project, "entry.yaml", "secrets.yml", secrets=("secrets.yml",)
-    )
-    files = dict(_gather_files(discovered, include_secrets=False))
-    assert files["secrets.yml"] == REDACTED_PLACEHOLDER
-
-
-def test_gather_redacts_secret_symlinked_to_other_name(
+def test_gather_flags_secret_symlinked_to_other_name(
     project: Path, tmp_path: Path
 ) -> None:
-    """A `secrets.yaml` symlinked to a non-secrets-named target is still redacted
+    """A `secrets.yaml` symlinked to a non-secrets-named target is still flagged
     because the un-resolved basename was captured upstream."""
     target = tmp_path / "actual_creds.yaml"
     target.write_text("api_key: FROM_SYMLINK\n")
@@ -74,20 +82,10 @@ 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"
-    discovered = DiscoveredYamlFiles([resolved], {resolved})
-    files = dict(_gather_files(discovered, include_secrets=False))
-    assert REDACTED_PLACEHOLDER in files.values()
+    files = _gather_redacted(DiscoveredYamlFiles([resolved], {resolved}))
     assert b"FROM_SYMLINK" not in b"".join(files.values())
 
 
-def test_gather_embeds_secrets_when_opted_in(project: Path) -> None:
-    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"]
-
-
 def test_gather_uses_relative_path_for_external_files(
     project: Path, tmp_path: Path
 ) -> None:
@@ -96,16 +94,27 @@ def test_gather_uses_relative_path_for_external_files(
     sibling.write_text("foo: bar\n")
     CORE.config_path = project / "entry.yaml"
     discovered = DiscoveredYamlFiles([project / "entry.yaml", sibling], set())
-    files = dict(_gather_files(discovered, include_secrets=False))
+    files, _ = _gather_files(discovered)
     # project root is `tmp_path/project`, sibling is in `tmp_path` so it
     # resolves to `../outside.yaml`.
-    assert "../outside.yaml" in files
+    assert "../outside.yaml" in dict(files)
 
 
 def test_gather_raises_when_no_sources(project: Path) -> None:
     CORE.config_path = project / "entry.yaml"
     with pytest.raises(EsphomeError):
-        _gather_files(DiscoveredYamlFiles(), include_secrets=False)
+        _gather_files(DiscoveredYamlFiles())
+
+
+def test_gather_raises_on_load_errors(project: Path) -> None:
+    """A failed include load during discovery fails the build instead of
+    embedding an incomplete recovery bundle."""
+    CORE.config_path = project / "entry.yaml"
+    discovered = DiscoveredYamlFiles(
+        [project / "entry.yaml"], set(), load_errors=["oops.yaml: boom"]
+    )
+    with pytest.raises(EsphomeError, match="oops.yaml"):
+        _gather_files(discovered)
 
 
 def test_gather_raises_on_unreadable_file(
@@ -123,7 +132,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(discovered, include_secrets=False)
+        _gather_files(discovered)
 
 
 def test_gather_warns_on_unresolved_includes(
@@ -134,7 +143,7 @@ def test_gather_warns_on_unresolved_includes(
     CORE.config_path = project / "entry.yaml"
     discovered = DiscoveredYamlFiles([project / "entry.yaml"], set(), ["${board}.yaml"])
     with caplog.at_level("WARNING", logger="esphome.components.store_yaml"):
-        files = _gather_files(discovered, include_secrets=False)
+        files, _ = _gather_files(discovered)
     assert len(files) == 1
     assert any(
         "${board}.yaml" in r.message and "not contain" in r.message
@@ -142,6 +151,163 @@ def test_gather_warns_on_unresolved_includes(
     )
 
 
+# ---------------------------------------------------------------------------
+# _generate_redacted_files
+# ---------------------------------------------------------------------------
+
+
+def test_redacted_secrets_file_becomes_skeleton(project: Path) -> None:
+    """The secrets file is replaced by a fill-in skeleton listing every
+    referenced `!secret` key, so the recovered config is flashable."""
+    discovered = _sources(
+        project, "entry.yaml", "secrets.yaml", secrets=("secrets.yaml",)
+    )
+    files = _gather_redacted(discovered)
+    skeleton = files["secrets.yaml"].decode()
+    assert skeleton.startswith(SECRETS_SKELETON_HEADER)
+    assert 'api_key: ""' in skeleton
+    assert b"SUPER_SECRET" not in files["secrets.yaml"]
+    # The entry's own `!secret` reference is re-emitted as a reference.
+    assert "key: !secret 'api_key'" in files["entry.yaml"].decode()
+
+
+def test_redacted_inline_sensitive_value_becomes_secret_ref(project: Path) -> None:
+    """An inline cv.sensitive value is generated as `!secret <path-derived
+    name>` and lands in the skeleton."""
+    CORE.config = {"wifi": [{"password": SensitiveStr("my_password")}]}
+    discovered = _sources(
+        project, "wifi.yaml", "secrets.yaml", secrets=("secrets.yaml",)
+    )
+    files = _gather_redacted(discovered)
+    text = files["wifi.yaml"].decode()
+    assert "my_password" not in text
+    assert "password: !secret 'wifi_password'" in text
+    assert 'wifi_password: ""' in files["secrets.yaml"].decode()
+
+
+@pytest.mark.parametrize("quote", ['"', "'"])
+def test_redacted_quoted_inline_value(project: Path, quote: str) -> None:
+    """Quoting in the source doesn't matter — the swap happens on the parsed
+    scalar, not the text."""
+    (project / "wifi.yaml").write_text(f"password: {quote}my_password{quote}\n")
+    CORE.config = {"wifi": [{"password": SensitiveStr("my_password")}]}
+    discovered = _sources(project, "wifi.yaml")
+    files = _gather_redacted(discovered)
+    assert files["wifi.yaml"] == b"password: !secret 'wifi_password'\n"
+
+
+def test_redacted_swap_is_whole_scalar_and_value_keyed(project: Path) -> None:
+    """Every whole scalar equal to the sensitive value is swapped (value-keyed,
+    like `!secret` itself); substrings inside other scalars are never touched.
+    The recovered config stays semantically identical once the secret is filled."""
+    (project / "wifi.yaml").write_text(
+        "platform: esp32\nnote: esp32 is great\npassword: esp32\n"
+    )
+    CORE.config = {"wifi": [{"password": SensitiveStr("esp32")}]}
+    discovered = _sources(project, "wifi.yaml")
+    files = _gather_redacted(discovered)
+    text = files["wifi.yaml"].decode()
+    assert "password: !secret 'wifi_password'" in text
+    assert "platform: !secret 'wifi_password'" in text
+    assert "note: esp32 is great" in text
+
+
+def test_redacted_include_reference_round_trips(project: Path) -> None:
+    """A nested `!include` stays a reference in the generated file."""
+    (project / "entry.yaml").write_text(
+        "esphome:\n  name: test\nwifi: !include wifi.yaml\n"
+    )
+    discovered = _sources(project, "entry.yaml", "wifi.yaml")
+    files = _gather_redacted(discovered)
+    assert "wifi: !include 'wifi.yaml'" in files["entry.yaml"].decode()
+
+
+def test_redacted_reuses_existing_secret_name_for_duplicated_value(
+    project: Path,
+) -> None:
+    """A value that comes from `!secret` somewhere but is ALSO written inline
+    elsewhere is generated with the existing secret name."""
+    (project / "wifi.yaml").write_text("password: SUPER_SECRET\n")
+    CORE.config = {"wifi": [{"password": SensitiveStr("SUPER_SECRET")}]}
+    yaml_util._SECRET_VALUES["SUPER_SECRET"] = "api_key"
+    discovered = _sources(
+        project, "wifi.yaml", "secrets.yaml", secrets=("secrets.yaml",)
+    )
+    files = _gather_redacted(discovered)
+    assert files["wifi.yaml"] == b"password: !secret 'api_key'\n"
+
+
+def test_redacted_warns_when_value_not_locatable(
+    project: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A sensitive value that never appears as a whole scalar (e.g. composed
+    via substitutions) produces a warning naming the config path, not the value."""
+    CORE.config = {"wifi": [{"password": SensitiveStr("not_in_any_file")}]}
+    discovered = _sources(project, "wifi.yaml")
+    with caplog.at_level("WARNING", logger="esphome.components.store_yaml"):
+        _gather_redacted(discovered)
+    assert any(
+        "wifi.password" in r.message and "could not locate" in r.message
+        for r in caplog.records
+    )
+    assert not any("not_in_any_file" in r.message for r in caplog.records)
+
+
+def test_redacted_does_not_warn_for_secret_only_values(
+    project: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A value that only exists via `!secret` legitimately never appears inline."""
+    CORE.config = {"api": {"encryption": {"key": SensitiveStr("SUPER_SECRET")}}}
+    yaml_util._SECRET_VALUES["SUPER_SECRET"] = "api_key"
+    discovered = _sources(
+        project, "entry.yaml", "secrets.yaml", secrets=("secrets.yaml",)
+    )
+    with caplog.at_level("WARNING", logger="esphome.components.store_yaml"):
+        _gather_redacted(discovered)
+    assert not any("could not locate" in r.message for r in caplog.records)
+
+
+def test_redacted_skips_empty_sensitive_values(project: Path) -> None:
+    """Empty defaults (e.g. mqtt password) are never swapped."""
+    (project / "wifi.yaml").write_text("ssid: my_ssid\n")
+    CORE.config = {"mqtt": {"password": SensitiveStr("")}}
+    discovered = _sources(project, "wifi.yaml")
+    files = _gather_redacted(discovered)
+    assert files["wifi.yaml"] == b"ssid: my_ssid\n"
+
+
+def test_redacted_adds_synthetic_secrets_file_when_none_captured(
+    project: Path,
+) -> None:
+    """Inline secrets in a project without a secrets.yaml still produce a
+    skeleton so the recovered config is complete."""
+    CORE.config = {"wifi": [{"password": SensitiveStr("my_password")}]}
+    discovered = _sources(project, "wifi.yaml")
+    files = _gather_redacted(discovered)
+    assert 'wifi_password: ""' in files["secrets.yaml"].decode()
+
+
+def test_redacted_generates_unique_names_on_collision(project: Path) -> None:
+    """Two different inline values whose paths collide get distinct names."""
+    (project / "wifi.yaml").write_text("password: first_pw\n")
+    (project / "wifi2.yaml").write_text("password: second_pw\n")
+    CORE.config = {
+        "wifi": [
+            {"password": SensitiveStr("first_pw")},
+            {"password": SensitiveStr("second_pw")},
+        ]
+    }
+    discovered = _sources(project, "wifi.yaml", "wifi2.yaml")
+    files = _gather_redacted(discovered)
+    assert files["wifi.yaml"] == b"password: !secret 'wifi_password'\n"
+    assert files["wifi2.yaml"] == b"password: !secret 'wifi_password_2'\n"
+
+
+# ---------------------------------------------------------------------------
+# envelope pack/unpack
+# ---------------------------------------------------------------------------
+
+
 def test_pack_envelope_roundtrip() -> None:
     files = [
         ("entry.yaml", b"esphome:\n  name: test\n"),
@@ -166,3 +332,16 @@ def test_pack_envelope_rejects_overlong_path() -> None:
 def test_unpack_envelope_rejects_bad_magic() -> None:
     with pytest.raises(EsphomeError):
         unpack_envelope(b"NOPE" + b"\x00" * 4)
+
+
+@pytest.mark.parametrize("cut", [5, 9, 12, -1])
+def test_unpack_envelope_rejects_truncated_input(cut: int) -> None:
+    blob = _pack_envelope([("entry.yaml", b"esphome:\n")])
+    with pytest.raises(EsphomeError, match="truncated"):
+        unpack_envelope(blob[:cut])
+
+
+def test_unpack_envelope_rejects_trailing_bytes() -> None:
+    blob = _pack_envelope([("entry.yaml", b"esphome:\n")])
+    with pytest.raises(EsphomeError, match="trailing"):
+        unpack_envelope(blob + b"\x00")

tests/unit_tests/test_yaml_util.py

@@ -1091,8 +1091,9 @@ def test_force_load_include_files_returns_unresolved_paths(
     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)]
+    result = force_load_include_files({"a": templated, "b": plain})
+    assert result.unresolved == [str(templated.file)]
+    assert result.errors == []
     assert plain.load_calls == 1
 
 
@@ -1100,14 +1101,17 @@ def test_force_load_include_files_warns_on_load_failure(
     patch_include_file: None,
     caplog: pytest.LogCaptureFixture,
 ) -> None:
-    """An `EsphomeError` raised by `load()` is caught and logged, not propagated."""
+    """An `EsphomeError` raised by `load()` is caught, logged, and reported to
+    the caller — not propagated."""
     stub = _StubInclude("missing.yaml", raise_on_load=EsphomeError("boom"))
     with caplog.at_level("WARNING", logger="esphome.yaml_util"):
-        force_load_include_files({"k": stub})
+        result = force_load_include_files({"k": stub})
     assert any(
         "Failed to load !include" in r.message and "missing.yaml" in r.message
         for r in caplog.records
     )
+    assert result.errors == [f"{stub.file}: boom"]
+    assert result.unresolved == []
 
 
 def test_discovered_yaml_files_holds_files_and_secrets() -> None:
@@ -1174,11 +1178,18 @@ def test_discover_user_yaml_files_flags_secrets_symlink(tmp_path: Path) -> None:
     assert target.resolve() in discovered.secrets
 
 
-def test_discover_user_yaml_files_swallows_parse_errors(tmp_path: Path) -> None:
-    """A YAML parse failure returns whatever was tracked so far without raising."""
+def test_discover_user_yaml_files_reports_parse_errors(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A YAML parse failure is logged and surfaced in `.load_errors` (not
+    raised), so consumers can tell the file set is incomplete."""
     entry = _write(tmp_path, "entry.yaml", "esphome: [unterminated\n")
-    discovered = discover_user_yaml_files(entry)
+    with caplog.at_level("WARNING", logger="esphome.yaml_util"):
+        discovered = discover_user_yaml_files(entry)
     assert isinstance(discovered, DiscoveredYamlFiles)
+    assert len(discovered.load_errors) == 1
+    assert "entry.yaml" in discovered.load_errors[0]
+    assert any("discovery failed to parse" in r.message for r in caplog.records)
 
 
 def test_discover_user_yaml_files_reports_unresolved_includes(
@@ -1419,6 +1430,40 @@ def test_dump__redaction_flag_does_not_leak_between_calls() -> None:
     assert "\\033[8m" in redacted_again
 
 
+def test_secret_values_registered_swaps_scalars_in_dump() -> None:
+    """Registered value→name mappings make dump() emit `!secret <name>` for
+    matching scalars, and are removed again on exit."""
+    with yaml_util.secret_values_registered({"hunter2": "wifi_password"}):
+        out = yaml_util.dump({"password": make_data_base("hunter2")})
+        assert "password: !secret 'wifi_password'" in out
+        assert "hunter2" not in out
+    out_after = yaml_util.dump({"password": make_data_base("hunter2")})
+    assert "hunter2" in out_after
+    assert "!secret" not in out_after
+    assert yaml_util.is_secret("hunter2") is None
+
+
+def test_secret_values_registered_does_not_clobber_real_secrets() -> None:
+    """A value already mapped by a real `!secret` keeps its original name."""
+    yaml_util._SECRET_VALUES["hunter2"] = "original_name"
+    try:
+        with yaml_util.secret_values_registered({"hunter2": "generated_name"}):
+            out = yaml_util.dump({"password": make_data_base("hunter2")})
+            assert "!secret 'original_name'" in out
+        # The pre-existing mapping survives the context exit.
+        assert yaml_util.is_secret("hunter2") == "original_name"
+    finally:
+        yaml_util._SECRET_VALUES.clear()
+
+
+def test_registered_secret_names() -> None:
+    yaml_util._SECRET_VALUES["value_a"] = "name_a"
+    try:
+        assert "name_a" in yaml_util.registered_secret_names()
+    finally:
+        yaml_util._SECRET_VALUES.clear()
+
+
 @pytest.fixture(autouse=True)
 def clear_dropped_merge_keys() -> None:
     """Reset the dropped-merge-key queue between tests."""