[store_yaml] Cover scalar include secret swap, document secret registry ordering

# Conflicts:
#	esphome/components/api/api.proto
#	esphome/components/api/api_pb2.cpp
#	esphome/components/api/api_pb2.h
#	esphome/components/api/api_pb2_dump.cpp
#	esphome/yaml_util.py

CODEOWNERS

@@ -517,6 +517,7 @@ esphome/components/st7735/* @SenexCrenshaw
 esphome/components/st7789v/* @kbx81
 esphome/components/st7920/* @marsjan155
 esphome/components/statsd/* @Links2004
+esphome/components/store_yaml/* @bdraco
 esphome/components/stts22h/* @B48D81EFCC
 esphome/components/substitutions/* @esphome/core
 esphome/components/sun/* @OttoWinter

esphome/bundle.py

@@ -101,25 +101,21 @@ _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()
     for fpath in yaml_files:
         try:
             text = fpath.read_text(encoding="utf-8")
-        except (OSError, UnicodeDecodeError):
+        except (OSError, UnicodeDecodeError) as err:
+            _LOGGER.warning(
+                "Could not scan %s for !secret references (%s); the bundled "
+                "secret set may be incomplete",
+                fpath,
+                err,
+            )
             continue
-        for match in _SECRET_RE.finditer(text):
-            keys.add(match.group(1))
+        keys |= yaml_util.find_secret_references(text)
     return keys
 
 
@@ -393,6 +389,18 @@ class ConfigBundleCreator:
         must ship every candidate so the remote build can pick any one.
         """
         discovered = yaml_util.discover_user_yaml_files(self._config_path)
+        if discovered.load_errors:
+            _LOGGER.warning(
+                "Bundle may be incomplete; could not load all configuration files: %s",
+                "; ".join(discovered.load_errors),
+            )
+        if discovered.unresolved:
+            _LOGGER.warning(
+                "Bundle may be incomplete; %d !include path(s) use "
+                "substitutions and cannot be captured: %s",
+                len(discovered.unresolved),
+                ", ".join(discovered.unresolved),
+            )
         self._secrets_paths.update(discovered.secrets)
         config_resolved = self._config_path.resolve()
         for fpath in discovered.files:

esphome/components/api/api.proto

@@ -76,6 +76,8 @@ service APIConnection {
   rpc serial_proxy_set_modem_pins(SerialProxySetModemPinsRequest) returns (void) {}
   rpc serial_proxy_get_modem_pins(SerialProxyGetModemPinsRequest) returns (void) {}
   rpc serial_proxy_request(SerialProxyRequest) returns (void) {}
+
+  rpc get_yaml(GetYamlRequest) returns (void) {}
 }
 
 
@@ -315,6 +317,11 @@ message DeviceInfoResponse {
   // all-zeros PSK, so the api encryption key can be provisioned without being
   // sent in plaintext (protects against passive sniffing, not active MITM)
   bool api_encryption_provisionable = 26 [(field_ifdef) = "USE_API_NOISE"];
+
+  // Whether this firmware embeds its YAML configuration for recovery via
+  // `get_yaml`. Clients use this to skip the request entirely when the
+  // device cannot answer it instead of waiting for a timeout.
+  bool has_store_yaml = 27 [(field_ifdef) = "USE_STORE_YAML"];
 }
 
 message ListEntitiesRequest {
@@ -2752,3 +2759,31 @@ message BluetoothSetConnectionParamsResponse {
   uint64 address = 1;
   int32 error = 2;
 }
+
+// ==================== STORE YAML ====================
+// Embed the user's YAML in firmware and stream it back over the API so a lost
+// config can be recovered from a running device. The device only stores the
+// compressed bytes; decompression happens client-side.
+//
+// A GetYamlRequest received while a transfer is already streaming on the same
+// connection is ignored; the in-flight transfer continues undisturbed. To
+// restart a transfer, reconnect.
+message GetYamlRequest {
+  option (id) = 149;
+  option (source) = SOURCE_CLIENT;
+  option (ifdef) = "USE_STORE_YAML";
+  option (no_delay) = true;
+}
+
+message GetYamlResponse {
+  option (id) = 150;
+  option (source) = SOURCE_SERVER;
+  option (ifdef) = "USE_STORE_YAML";
+
+  bytes data = 1 [(force) = true];
+  bool done = 2 [(force) = true];
+  // Sent on the first chunk only — the client is expected to cache it.
+  // (firmware bandwidth/flash is expensive; the client has gigabytes.)
+  uint32 total_size = 3;
+  string encoding = 4 [(max_data_length) = 8];
+}

esphome/components/api/api_connection.cpp

@@ -56,6 +56,9 @@
 #ifdef USE_RADIO_FREQUENCY
 #include "esphome/components/radio_frequency/radio_frequency.h"
 #endif
+#ifdef USE_STORE_YAML
+#include "esphome/components/store_yaml/store_yaml.h"
+#endif
 
 namespace esphome::api {
 
@@ -346,6 +349,13 @@ void APIConnection::loop() {
   // (missing a frame is fine, missing a state update is not)
   this->try_send_camera_image_();
 #endif
+
+#ifdef USE_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
 }
 
 void APIConnection::check_keepalive_(uint32_t now) {
@@ -1191,6 +1201,89 @@ void APIConnection::on_camera_image_request(const CameraImageRequest &msg) {
 }
 #endif
 
+#ifdef USE_STORE_YAML
+#ifdef USE_ESP8266
+// On ESP8266 the blob lives in instruction flash and can't be read directly, so
+// each chunk is bounced through a heap buffer via progmem_memcpy. The buffer
+// exists only while a transfer is in flight; retrieval is rare, so no RAM is
+// held for the firmware's lifetime. Every other platform sends straight from
+// the blob, zero-copy, in MTU-sized chunks.
+static constexpr size_t STORE_YAML_CHUNK_SIZE = 512;
+#endif
+
+void APIConnection::on_get_yaml_request() {
+  // A re-request while a transfer is in flight is ignored; see the
+  // GetYamlRequest comment in api.proto. A client that wants to restart
+  // must reconnect.
+  if (this->store_yaml_pos_ != std::numeric_limits<size_t>::max())
+    return;
+#ifdef USE_ESP8266
+  // OOM aborts by design (NEW_OOM_ABORT): a heap that cannot supply this
+  // buffer is already failing, and the post-reset retry gets a clean heap.
+  this->store_yaml_chunk_buf_ = std::make_unique<uint8_t[]>(STORE_YAML_CHUNK_SIZE);
+#endif
+  // All responses go through the loop-driven retry below, so a full TX
+  // buffer at request time can't strand the client without a terminal frame.
+  this->store_yaml_pos_ = 0;
+  this->try_send_store_yaml_();
+}
+
+// Caller guarantees: store_yaml_pos_ != SIZE_MAX (a request is in flight).
+void APIConnection::try_send_store_yaml_() {
+  // A client connecting while later components are still setting up (the app
+  // loop runs for already-initialized components during setup) can request
+  // YAML before store_yaml's setup() registered the component. Leave the
+  // request pending; this is retried from loop() until the component appears.
+  auto *comp = store_yaml::global_store_yaml;
+  if (comp == nullptr)
+    return;
+  // Codegen always embeds a non-empty blob, so total > 0 here.
+  const size_t total = comp->get_size();
+
+#ifdef USE_ESP8266
+  const size_t chunk_size = STORE_YAML_CHUNK_SIZE;
+#else
+  const size_t chunk_size = MAX_BATCH_PACKET_SIZE;
+#endif
+
+  // Camera-style streaming: advance the position only after a successful send,
+  // so a WOULD_BLOCK simply retries the same chunk on the next loop iteration.
+  while (true) {
+    if (!this->helper_->can_write_without_blocking())
+      return;
+
+    const size_t remaining = total - this->store_yaml_pos_;
+    const size_t to_send = std::min(remaining, chunk_size);
+
+    GetYamlResponse resp;
+#ifdef USE_ESP8266
+    progmem_memcpy(this->store_yaml_chunk_buf_.get(), comp->get_data() + this->store_yaml_pos_, to_send);
+    resp.set_data(this->store_yaml_chunk_buf_.get(), 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);
+    }
+    resp.done = (this->store_yaml_pos_ + to_send) >= total;
+
+    if (!this->send_message(resp))
+      return;  // retry on next loop, pos unchanged
+
+    this->store_yaml_pos_ += to_send;
+    if (resp.done)
+      break;
+  }
+
+  // Final response (with done=true) sent successfully.
+  this->store_yaml_pos_ = std::numeric_limits<size_t>::max();
+#ifdef USE_ESP8266
+  this->store_yaml_chunk_buf_.reset();
+#endif
+}
+#endif
+
 #ifdef USE_HOMEASSISTANT_TIME
 void APIConnection::on_get_time_response(const GetTimeResponse &value) {
   if (homeassistant::global_homeassistant_time != nullptr) {
@@ -1842,6 +1935,12 @@ bool APIConnection::send_device_info_response_() {
 #ifdef USE_DEEP_SLEEP
   resp.has_deep_sleep = deep_sleep::global_has_deep_sleep;
 #endif
+#ifdef USE_STORE_YAML
+  // Compile-time knowledge: codegen always embeds a non-empty blob when the
+  // component is compiled in. Deriving this from the runtime pointer could
+  // report false to a client connecting before store_yaml's setup() ran.
+  resp.has_store_yaml = true;
+#endif
 #ifdef ESPHOME_PROJECT_NAME
 #ifdef USE_ESP8266
   static const char PROJECT_NAME_PROGMEM[] PROGMEM = ESPHOME_PROJECT_NAME;
@@ -2116,10 +2215,15 @@ bool APIConnection::try_to_clear_buffer_slow_(bool log_out_of_space) {
 bool APIConnection::send_message_(uint32_t payload_size, uint8_t message_type, MessageEncodeFn encode_fn,
                                   const void *msg) {
 #ifdef HAS_PROTO_MESSAGE_DUMP
-  // Skip dump for log messages (recursive logging risk) and camera frames (high-frequency noise)
+  // Skip dump for log messages (recursive logging risk), camera frames (high-frequency noise),
+  // and YAML recovery payloads (every chunk would log the embedded config, including any
+  // secrets the user opted into).
   if (message_type != SubscribeLogsResponse::MESSAGE_TYPE
 #ifdef USE_CAMERA
       && message_type != CameraImageResponse::MESSAGE_TYPE
+#endif
+#ifdef USE_STORE_YAML
+      && message_type != GetYamlResponse::MESSAGE_TYPE
 #endif
   ) {
     auto *proto_msg = static_cast<const ProtoMessage *>(msg);

esphome/components/api/api_connection.h

@@ -29,6 +29,7 @@
 
 #include <functional>
 #include <limits>
+#include <memory>
 #include <vector>
 
 namespace esphome {
@@ -121,6 +122,9 @@ class APIConnection final : public APIServerConnectionBase {
   void set_camera_state(std::shared_ptr<camera::CameraImage> image);
   void on_camera_image_request(const CameraImageRequest &msg);
 #endif
+#ifdef USE_STORE_YAML
+  void on_get_yaml_request();
+#endif
 #ifdef USE_CLIMATE
   bool send_climate_state(climate::Climate *climate);
   void on_climate_command_request(const ClimateCommandRequest &msg);
@@ -399,6 +403,18 @@ class APIConnection final : public APIServerConnectionBase {
   void try_send_camera_image_();
 #endif
 
+#ifdef USE_STORE_YAML
+  void try_send_store_yaml_();
+  // Streaming offset into the PROGMEM blob; max() means "not streaming".
+  size_t store_yaml_pos_{std::numeric_limits<size_t>::max()};
+#ifdef USE_ESP8266
+  // Bounce buffer for progmem_memcpy, alive only while a transfer is in
+  // flight; retrieval is rare, so the RAM is not held for the firmware's
+  // lifetime. Freed on the terminal frame or with the connection.
+  std::unique_ptr<uint8_t[]> store_yaml_chunk_buf_;
+#endif
+#endif
+
 #ifdef USE_API_HOMEASSISTANT_STATES
   void process_state_subscriptions_();
 #endif

esphome/components/api/api_pb2.cpp

@@ -173,6 +173,9 @@ uint8_t *DeviceInfoResponse::encode(ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_
 #endif
 #ifdef USE_API_NOISE
   ProtoEncode::encode_bool(pos PROTO_ENCODE_DEBUG_ARG, 26, this->api_encryption_provisionable);
+#endif
+#ifdef USE_STORE_YAML
+  ProtoEncode::encode_bool(pos PROTO_ENCODE_DEBUG_ARG, 27, this->has_store_yaml);
 #endif
   return pos;
 }
@@ -238,6 +241,9 @@ uint32_t DeviceInfoResponse::calculate_size() const {
 #endif
 #ifdef USE_API_NOISE
   size += ProtoSize::calc_bool(2, this->api_encryption_provisionable);
+#endif
+#ifdef USE_STORE_YAML
+  size += ProtoSize::calc_bool(2, this->has_store_yaml);
 #endif
   return size;
 }
@@ -4181,5 +4187,26 @@ uint32_t BluetoothSetConnectionParamsResponse::calculate_size() const {
   return size;
 }
 #endif
+#ifdef USE_STORE_YAML
+uint8_t *GetYamlResponse::encode(ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_PARAM) const {
+  uint8_t *__restrict__ pos = buffer.get_pos();
+  ProtoEncode::write_raw_byte(pos PROTO_ENCODE_DEBUG_ARG, 10);
+  ProtoEncode::encode_varint_raw(pos PROTO_ENCODE_DEBUG_ARG, this->data_len_);
+  ProtoEncode::encode_raw(pos PROTO_ENCODE_DEBUG_ARG, this->data_ptr_, this->data_len_);
+  ProtoEncode::write_raw_byte(pos PROTO_ENCODE_DEBUG_ARG, 16);
+  ProtoEncode::write_raw_byte(pos PROTO_ENCODE_DEBUG_ARG, this->done ? 0x01 : 0x00);
+  ProtoEncode::encode_uint32(pos PROTO_ENCODE_DEBUG_ARG, 3, this->total_size);
+  ProtoEncode::encode_string(pos PROTO_ENCODE_DEBUG_ARG, 4, this->encoding);
+  return pos;
+}
+uint32_t GetYamlResponse::calculate_size() const {
+  uint32_t size = 0;
+  size += ProtoSize::calc_length_force(1, this->data_len_);
+  size += ProtoSize::calc_bool_force(1);
+  size += ProtoSize::calc_uint32(1, this->total_size);
+  size += !this->encoding.empty() ? 2 + this->encoding.size() : 0;
+  return size;
+}
+#endif
 
 }  // namespace esphome::api

esphome/components/api/api_pb2.h

@@ -533,7 +533,7 @@ class SerialProxyInfo final : public ProtoMessage {
 class DeviceInfoResponse final : public ProtoMessage {
  public:
   static constexpr uint8_t MESSAGE_TYPE = 10;
-  static constexpr uint16_t ESTIMATED_SIZE = 312;
+  static constexpr uint16_t ESTIMATED_SIZE = 315;
 #ifdef HAS_PROTO_MESSAGE_DUMP
   const LogString *message_name() const override { return LOG_STR("device_info_response"); }
 #endif
@@ -591,6 +591,9 @@ class DeviceInfoResponse final : public ProtoMessage {
 #endif
 #ifdef USE_API_NOISE
   bool api_encryption_provisionable{false};
+#endif
+#ifdef USE_STORE_YAML
+  bool has_store_yaml{false};
 #endif
   uint8_t *encode(ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_PARAM) const;
   uint32_t calculate_size() const;
@@ -3328,5 +3331,31 @@ class BluetoothSetConnectionParamsResponse final : public ProtoMessage {
  protected:
 };
 #endif
+#ifdef USE_STORE_YAML
+class GetYamlResponse final : public ProtoMessage {
+ public:
+  static constexpr uint8_t MESSAGE_TYPE = 150;
+  static constexpr uint8_t ESTIMATED_SIZE = 34;
+#ifdef HAS_PROTO_MESSAGE_DUMP
+  const LogString *message_name() const override { return LOG_STR("get_yaml_response"); }
+#endif
+  const uint8_t *data_ptr_{nullptr};
+  size_t data_len_{0};
+  void set_data(const uint8_t *data, size_t len) {
+    this->data_ptr_ = data;
+    this->data_len_ = len;
+  }
+  bool done{false};
+  uint32_t total_size{0};
+  StringRef encoding{};
+  uint8_t *encode(ProtoWriteBuffer &buffer PROTO_ENCODE_DEBUG_PARAM) const;
+  uint32_t calculate_size() const;
+#ifdef HAS_PROTO_MESSAGE_DUMP
+  const char *dump_to(DumpBuffer &out) const override;
+#endif
+
+ protected:
+};
+#endif
 
 }  // namespace esphome::api

esphome/components/api/api_pb2_dump.cpp

@@ -985,6 +985,9 @@ const char *DeviceInfoResponse::dump_to(DumpBuffer &out) const {
 #endif
 #ifdef USE_API_NOISE
   dump_field(out, ESPHOME_PSTR("api_encryption_provisionable"), this->api_encryption_provisionable);
+#endif
+#ifdef USE_STORE_YAML
+  dump_field(out, ESPHOME_PSTR("has_store_yaml"), this->has_store_yaml);
 #endif
   return out.c_str();
 }
@@ -2732,6 +2735,16 @@ const char *BluetoothSetConnectionParamsResponse::dump_to(DumpBuffer &out) const
   return out.c_str();
 }
 #endif
+#ifdef USE_STORE_YAML
+const char *GetYamlResponse::dump_to(DumpBuffer &out) const {
+  MessageDumpHelper helper(out, ESPHOME_PSTR("GetYamlResponse"));
+  dump_bytes_field(out, ESPHOME_PSTR("data"), this->data_ptr_, this->data_len_);
+  dump_field(out, ESPHOME_PSTR("done"), this->done);
+  dump_field(out, ESPHOME_PSTR("total_size"), this->total_size);
+  dump_field(out, ESPHOME_PSTR("encoding"), this->encoding);
+  return out.c_str();
+}
+#endif
 
 }  // namespace esphome::api
 

esphome/components/api/api_pb2_service.cpp

@@ -704,6 +704,15 @@ void APIConnection::read_message_(uint32_t msg_size, uint32_t msg_type, const ui
       this->on_bluetooth_set_connection_params_request(msg);
       break;
     }
+#endif
+#ifdef USE_STORE_YAML
+    case 149 /* GetYamlRequest is empty */: {
+#ifdef HAS_PROTO_MESSAGE_DUMP
+      this->log_receive_message_(LOG_STR("on_get_yaml_request"));
+#endif
+      this->on_get_yaml_request();
+      break;
+    }
 #endif
     default:
       break;

esphome/components/api/api_pb2_service.h

@@ -236,6 +236,10 @@ class APIServerConnectionBase {
 #ifdef USE_BLUETOOTH_PROXY
   void on_bluetooth_set_connection_params_request(const BluetoothSetConnectionParamsRequest &value){};
 #endif
+
+#ifdef USE_STORE_YAML
+  void on_get_yaml_request(){};
+#endif
 };
 
 }  // namespace esphome::api

esphome/components/packages/__init__.py

@@ -1,5 +1,6 @@
 from collections import UserDict
 from collections.abc import Callable
+from dataclasses import dataclass, field
 from functools import reduce
 from pathlib import Path
 from typing import Any
@@ -33,9 +34,43 @@ from esphome.const import (
     CONF_VARS,
     __version__ as ESPHOME_VERSION,
 )
-from esphome.core import EsphomeError
+from esphome.core import CORE, EsphomeError
 
 DOMAIN = CONF_PACKAGES
+
+
+@dataclass(frozen=True)
+class RemotePackageSource:
+    """A remote source a package was fetched from while processing the config."""
+
+    url: str
+    ref: str | None
+
+
+@dataclass
+class PackagesData:
+    """Per-run package state, keyed under DOMAIN in CORE.data."""
+
+    remote_sources: list[RemotePackageSource] = field(default_factory=list)
+
+
+def _get_data() -> PackagesData:
+    if DOMAIN not in CORE.data:
+        CORE.data[DOMAIN] = PackagesData()
+    return CORE.data[DOMAIN]
+
+
+def get_remote_package_sources() -> list[RemotePackageSource]:
+    """Remote sources fetched while processing this config, in fetch order.
+
+    Consumers (e.g. store_yaml) use this to tell which parts of the config
+    came from remote repositories rather than local files.
+    """
+    if (data := CORE.data.get(DOMAIN)) is None:
+        return []
+    return data.remote_sources
+
+
 # Guard against infinite include chains (e.g. A includes B includes A).
 MAX_INCLUDE_DEPTH = 20
 
@@ -189,6 +224,10 @@ def _process_remote_package(config: dict[str, Any]) -> dict[str, Any]:
         username=config.get(CONF_USERNAME),
         password=config.get(CONF_PASSWORD),
     )
+    source = RemotePackageSource(config[CONF_URL], config.get(CONF_REF))
+    remote_sources = _get_data().remote_sources
+    if source not in remote_sources:
+        remote_sources.append(source)
     files: list[dict[str, Any]] = []
 
     # ``repo_root`` is the directory containing ``.git`` and must be passed

esphome/components/store_yaml/__init__.py

@@ -0,0 +1,546 @@
+from __future__ import annotations
+
+from collections.abc import Generator
+from dataclasses import dataclass
+import logging
+from pathlib import Path
+import struct
+
+from esphome import yaml_util
+import esphome.codegen as cg
+from esphome.components import packages
+from esphome.components.api import CONF_ENCRYPTION
+from esphome.config_helpers import Extend, Remove
+import esphome.config_validation as cv
+from esphome.const import CONF_API, CONF_ID, CONF_KEY, CONF_RAW_DATA_ID
+from esphome.core import CORE, EsphomeError, HexInt, Lambda
+import esphome.final_validate as fv
+from esphome.helpers import ensure_unique_string
+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"]
+DEPENDENCIES = ["api"]
+
+CONF_INCLUDE_SECRETS = "include_secrets"
+CONF_ALLOW_UNENCRYPTED = "allow_unencrypted"
+
+store_yaml_ns = cg.esphome_ns.namespace("store_yaml")
+StoreYamlComponent = store_yaml_ns.class_("StoreYamlComponent", cg.Component)
+
+# Compression level for zstd; 22 is the max and gives ~70-90% reduction on YAML.
+ZSTD_LEVEL = 22
+# Envelope magic: "EHY1" = ESPHome YAML, version 1.
+ENVELOPE_MAGIC = b"EHY1"
+# 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"
+)
+# Envelope path of the note recording content that could not be captured.
+UNCAPTURED_NOTE_PATH = "store_yaml_uncaptured.yaml"
+
+CONFIG_SCHEMA = cv.Schema(
+    {
+        cv.GenerateID(): cv.declare_id(StoreYamlComponent),
+        cv.GenerateID(CONF_RAW_DATA_ID): cv.declare_id(cg.uint8),
+        cv.Optional(CONF_INCLUDE_SECRETS, default=False): cv.boolean,
+        cv.Optional(CONF_ALLOW_UNENCRYPTED, default=False): cv.boolean,
+    }
+).extend(cv.COMPONENT_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: 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, {})
+    # Explicitly require a configured key: a keyless provisionable
+    # `api: encryption:` accepts the well-known all-zeros PSK until
+    # provisioned, so anyone could provision it and pull the YAML.
+    encryption = api_conf.get(CONF_ENCRYPTION) or {}
+    if CONF_KEY in encryption:
+        return config
+    if config.get(CONF_ALLOW_UNENCRYPTED):
+        if config.get(CONF_INCLUDE_SECRETS):
+            _LOGGER.warning(
+                "store_yaml is enabled without API encryption and with "
+                "include_secrets; any client that can reach the device on the "
+                "network can pull the embedded YAML including the verbatim "
+                "contents of secrets.yaml."
+            )
+        else:
+            _LOGGER.warning(
+                "store_yaml is enabled without API encryption; any client that "
+                "can reach the device on the network can pull the embedded YAML."
+            )
+        return config
+    raise cv.Invalid(
+        "store_yaml requires API encryption (configure `api.encryption.key`). "
+        "Without encryption, the embedded YAML — which may contain Wi-Fi "
+        "credentials or opted-in secrets — can be read by any client that "
+        "reaches the device. Set `store_yaml.allow_unencrypted: true` to "
+        "override after acknowledging the risk."
+    )
+
+
+FINAL_VALIDATE_SCHEMA = _final_validate
+
+
+def _gather_files(
+    discovered: yaml_util.DiscoveredYamlFiles,
+) -> tuple[list[tuple[str, Path]], set[str]]:
+    """Map each discovered YAML file to its envelope path.
+
+    Returns (relative_path, source_path) 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 "
+            "captured (%s); the embedded recovery data will not contain them",
+            len(discovered.unresolved),
+            ", ".join(discovered.unresolved),
+        )
+
+    # Resolved (not just absolute) because discovery returns resolved paths
+    # and relative_to needs both sides on the same footing when symlinks are
+    # in play; CORE.config_dir is only absolute.
+    root = CORE.config_path.resolve().parent
+
+    entries: list[tuple[str, Path]] = []
+    secret_rels: set[str] = set()
+    for path in discovered.files:
+        # Files outside the project root (e.g. ../common.yaml or a secrets file
+        # in $HOME) keep their ".." components so the include graph is preserved
+        # and files from different directories with the same basename don't
+        # collide.
+        try:
+            rel_str = path.relative_to(root, walk_up=True).as_posix()
+        except ValueError as err:
+            # Different anchors (a Windows file on another drive) cannot be
+            # expressed relative to the config root.
+            raise EsphomeError(
+                f"store_yaml: cannot place {path} in the recovery envelope; "
+                f"it does not share a root with {root}: {err}"
+            ) from err
+
+        if path in discovered.secrets:
+            secret_rels.add(rel_str)
+        entries.append((rel_str, path))
+
+    return entries, secret_rels
+
+
+def _read_files_verbatim(entries: list[tuple[str, Path]]) -> list[tuple[str, bytes]]:
+    """Read each file's exact on-disk bytes (the `include_secrets: true` path)."""
+    files: list[tuple[str, bytes]] = []
+    for rel, path in entries:
+        try:
+            files.append((rel, 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
+    return files
+
+
+def _iter_nodes(
+    node: object, path: tuple[str, ...] = ()
+) -> Generator[tuple[tuple[str, ...], object, bool]]:
+    """Yield (config_path, value, is_key) for every mapping key and scalar in
+    a config tree. Keys are yielded at the path of their mapping.
+
+    Wrapper types the dumper renders as text are unwrapped so their payloads
+    are scanned too: `!lambda` bodies, `!extend`/`!remove` ids, and `!include`
+    file paths plus `vars:` values.
+    """
+    if isinstance(node, dict):
+        for key, value in node.items():
+            yield path, str(key), True
+            yield from _iter_nodes(value, (*path, str(key)))
+    elif isinstance(node, (list, tuple)):
+        for item in node:
+            yield from _iter_nodes(item, path)
+    elif isinstance(node, (Lambda, Extend, Remove)):
+        yield path, node.value, False
+    elif isinstance(node, yaml_util.IncludeFile):
+        yield path, str(node.file), False
+        if node.vars:
+            yield from _iter_nodes(node.vars, path)
+    elif isinstance(node, (str, int, float)) and not isinstance(node, bool):
+        yield path, node, False
+
+
+@dataclass
+class _SensitiveValue:
+    secret_name: str
+    config_path: str  # dotted path, for warnings (never log the value itself)
+    # Last path segments the value is sensitive at (e.g. {"password"}), used to
+    # tell the value's own occurrences apart from unrelated collisions.
+    sensitive_keys: set[str]
+
+
+def _check_sensitive_usage(
+    sensitive: dict[str, _SensitiveValue], trees: dict[str, object]
+) -> None:
+    """One pass over the parse trees that will be dumped, checking every place
+    a sensitive value shows up beyond its own whole-scalar occurrences.
+
+    - As a mapping key: fail the build. The dumper's value-keyed swap would
+      rewrite the key and corrupt the recovered structure.
+    - Strictly inside a larger scalar (a lambda body, a URL like
+      http://user:pw@host): the whole-scalar swap cannot redact it, so it
+      would ship verbatim. Inline sensitive values fail the build (the
+      move-into-!secret remedy applies); values already sourced from a real
+      `!secret` only warn, since overlaps like an SSID inside an entity name
+      are common and the value stays in the user's secrets.yaml either way.
+      Scanning tree scalars (not serialized text) means key names, tags, and
+      generated `!secret` references can never false-positive.
+    - As a whole scalar at an unrelated location: warn only. The swap rewrites
+      it to the `!secret` reference, which stays semantically identical until
+      the user fills in a different value during recovery. Occurrences under
+      the value's own sensitive key and swapped `substitutions:` definitions
+      (which keep `${...}` references working) are expected and stay silent.
+    """
+    if not sensitive:
+        return
+    key_hits: list[str] = []
+    embedded: list[str] = []
+    for rel, tree in trees.items():
+        for path, node, is_key in _iter_nodes(tree):
+            text = str(node)
+            info = sensitive.get(text)
+            if is_key:
+                if info is not None:
+                    key_hits.append(
+                        f"{info.config_path} (as the mapping key at "
+                        f"{'.'.join((*path, text))} in {rel})"
+                    )
+                continue
+            if info is not None:
+                if path and (
+                    path[0] == "substitutions" or path[-1] in info.sensitive_keys
+                ):
+                    continue
+                _LOGGER.warning(
+                    "store_yaml: the sensitive value at %s also matches the "
+                    "scalar at %s in %s; the recovered config will reference "
+                    "!secret %s there too",
+                    info.config_path,
+                    ".".join(path),
+                    rel,
+                    info.secret_name,
+                )
+                continue
+            for value, other in sensitive.items():
+                if value not in text:
+                    continue
+                if yaml_util.is_secret(value) is not None:
+                    # The value lives in a real secrets.yaml, so the
+                    # move-into-!secret remedy does not apply; overlaps like
+                    # an SSID inside an entity name are common and benign.
+                    # Warn naming the location, never the value.
+                    _LOGGER.warning(
+                        "store_yaml: the value of !secret %s (sensitive at %s) "
+                        "appears inside the value at %s in %s; substring "
+                        "occurrences are not redacted",
+                        other.secret_name,
+                        other.config_path,
+                        ".".join(path),
+                        rel,
+                    )
+                else:
+                    embedded.append(
+                        f"{other.config_path} (inside {'.'.join(path)} in {rel})"
+                    )
+    if key_hits:
+        raise EsphomeError(
+            "store_yaml: sensitive value(s) are also used as mapping keys: "
+            f"{', '.join(key_hits)}. The redaction swap would rewrite the key "
+            "and corrupt the recovered config. Change the value, or set "
+            "`include_secrets: true` to embed secrets deliberately."
+        )
+    if embedded:
+        raise EsphomeError(
+            "store_yaml: sensitive value(s) appear embedded inside larger "
+            f"values: {', '.join(embedded)}. Redaction only replaces whole "
+            "scalars, so these would ship unredacted. Move the value into a "
+            "`!secret` referenced on its own, or set `include_secrets: true` "
+            "to embed secrets deliberately."
+        )
+
+
+def _collect_sensitive_values() -> 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 names already
+    taken by real secrets.
+    """
+    used = yaml_util.registered_secret_names()
+    result: dict[str, _SensitiveValue] = {}
+    for path, node, is_key in _iter_nodes(CORE.config):
+        if is_key or not isinstance(node, yaml_util.SensitiveStr) or not node:
+            continue
+        value = str(node)
+        entry = result.get(value)
+        if entry is None:
+            name = yaml_util.is_secret(value)
+            if name is None:
+                name = ensure_unique_string("_".join(path) or "secret", used)
+            used.add(name)
+            entry = result[value] = _SensitiveValue(name, ".".join(path), set())
+        if path:
+            entry.sensitive_keys.add(path[-1])
+    return result
+
+
+def _uncaptured_note(
+    unresolved: list[str], remote_packages: list[str]
+) -> tuple[str, bytes] | None:
+    """Comment-only YAML entry listing content that could not be captured, so
+    a recovered config never silently appears complete. Emitted for both the
+    redacted and verbatim paths; user files are never modified to carry it.
+    Returns None when there is nothing to record."""
+    if not unresolved and not remote_packages:
+        return None
+    parts = ["# store_yaml: the following content could not be captured.\n"]
+    if unresolved:
+        parts.append(
+            "# These !include paths use substitutions; restore the files manually:\n"
+            + "".join(f"#   {inc}\n" for inc in unresolved)
+        )
+    if remote_packages:
+        parts.append(
+            "# These packages come from remote sources; re-fetch them to\n"
+            "# complete this config:\n"
+            + "".join(f"#   {pkg}\n" for pkg in remote_packages)
+        )
+    return (UNCAPTURED_NOTE_PATH, "".join(parts).encode("utf-8"))
+
+
+def _remote_package_descriptions() -> list[str]:
+    """Describe every remote source packages were fetched from.
+
+    Remote packages are downloaded while the config is processed; the packages
+    component records each source, and this formats that record. Their files
+    cannot be embedded, but the entry file still records the package config, so
+    the config is re-fetchable; this only makes the gap visible instead of
+    silent.
+    """
+    return [
+        f"{source.url}@{source.ref}" if source.ref else source.url
+        for source in packages.get_remote_package_sources()
+    ]
+
+
+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(
+    entries: list[tuple[str, Path]], 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()
+
+    trees = {
+        rel: yaml_util.load_yaml(path, clear_secrets=False)
+        for rel, path in entries
+        if rel not in secret_rels
+    }
+
+    _check_sensitive_usage(sensitive, trees)
+
+    registered = {value: info.secret_name for value, info in sensitive.items()}
+    with yaml_util.secret_values_registered(registered) as skeleton_keys:
+        texts = {rel: yaml_util.dump(tree) for rel, tree in trees.items()}
+
+    # skeleton_keys now holds exactly the `!secret` names the dumper emitted.
+    # A registered inline value whose name was never emitted was not found as
+    # a whole scalar in any captured file, so it would ship nowhere and the
+    # redaction promise cannot be verified — fail the build.
+    leaked = [
+        info.config_path
+        for value, info in sensitive.items()
+        if yaml_util.is_secret(value) is None and info.secret_name not in skeleton_keys
+    ]
+    if leaked:
+        remote = _remote_package_descriptions()
+        raise EsphomeError(
+            "store_yaml: could not redact the sensitive value(s) of "
+            f"{', '.join(leaked)}. The value was not found in any captured "
+            "file; it may be composed via substitutions, set on the command "
+            "line with -s, or defined inside a remote package"
+            + (f" ({', '.join(remote)})" if remote else "")
+            + ". Reference it with `!secret` in the YAML, or set "
+            "`include_secrets: true` to embed secrets deliberately."
+        )
+
+    skeleton = _build_secrets_skeleton(skeleton_keys)
+    result = [
+        (rel, skeleton if rel in secret_rels else texts[rel].encode("utf-8"))
+        for rel, _ in entries
+    ]
+    if skeleton_keys and yaml_util.SECRET_YAML not in secret_rels:
+        # The generated files reference `!secret` keys but no captured secrets
+        # file lands at the config root (none exists, or it resolves outside
+        # the root, e.g. a symlink target). `!secret` resolution looks for
+        # secrets.yaml beside the config, so ship a synthetic root skeleton to
+        # keep the recovered config loadable.
+        result.append((yaml_util.SECRET_YAML, skeleton))
+    return result
+
+
+def _pack_envelope(files: list[tuple[str, bytes]]) -> bytes:
+    """Pack files into the EHY1 envelope.
+
+    Layout: magic (4) | u32 file_count | repeat { u16 path_len | path_utf8 | u32 content_len | content_bytes }
+    All integers are little-endian.
+    """
+    parts: list[bytes] = [ENVELOPE_MAGIC, struct.pack("<I", len(files))]
+    seen: set[str] = set()
+    for path, content in files:
+        if path in seen:
+            # unpack_envelope builds a dict, so a duplicate would silently
+            # replace the earlier entry — fail the build instead.
+            raise EsphomeError(f"store_yaml: duplicate envelope path: {path}")
+        seen.add(path)
+        path_bytes = path.encode("utf-8")
+        if len(path_bytes) > 0xFFFF:
+            raise EsphomeError(
+                f"store_yaml: path too long ({len(path_bytes)} bytes): {path}"
+            )
+        parts.append(struct.pack("<H", len(path_bytes)))
+        parts.append(path_bytes)
+        parts.append(struct.pack("<I", len(content)))
+        parts.append(content)
+    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.
+
+    Absolute and drive-qualified paths are rejected: the packer never emits
+    them, so their presence means a malformed or hostile envelope. Relative
+    paths with ``..`` components are legitimate (the packer emits them for
+    files outside the config root), so callers that write files to disk must
+    still confine the resulting paths to their target directory."""
+    if blob[:4] != ENVELOPE_MAGIC:
+        raise EsphomeError("envelope must start with EHY1 magic")
+    pos = 4
+    files: dict[str, bytes] = {}
+
+    def take(n: int) -> bytes:
+        nonlocal pos
+        if pos + n > len(blob):
+            raise EsphomeError("truncated envelope")
+        chunk = blob[pos : pos + n]
+        pos += n
+        return chunk
+
+    (count,) = struct.unpack("<I", take(4))
+    for _ in range(count):
+        (path_len,) = struct.unpack("<H", take(2))
+        try:
+            path = take(path_len).decode("utf-8")
+        except UnicodeDecodeError as err:
+            raise EsphomeError(f"envelope path is not valid UTF-8: {err}") from err
+        if path.startswith(("/", "\\")) or (len(path) >= 2 and path[1] == ":"):
+            raise EsphomeError(f"envelope contains non-relative path: {path}")
+        if path in files:
+            raise EsphomeError(f"envelope contains duplicate path: {path}")
+        (content_len,) = struct.unpack("<I", take(4))
+        files[path] = take(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")
+
+    # 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.
+    # This re-parse (and the per-file loads in _generate_redacted_files) also
+    # repopulates yaml_util's secret registry, which save_compiled_config
+    # wiped earlier in write_cpp via dump(show_secrets=True); the redaction
+    # swap and is_secret() checks below rely on that registration.
+    discovered = yaml_util.discover_user_yaml_files(CORE.config_path)
+    entries, secret_rels = _gather_files(discovered)
+    if config[CONF_INCLUDE_SECRETS]:
+        files = _read_files_verbatim(entries)
+    else:
+        files = _generate_redacted_files(entries, secret_rels)
+    remote_packages = _remote_package_descriptions()
+    if remote_packages:
+        _LOGGER.warning(
+            "store_yaml: %d package(s) come from remote sources and cannot be "
+            "captured (%s); the embedded recovery data records the source so "
+            "they can be re-fetched",
+            len(remote_packages),
+            ", ".join(remote_packages),
+        )
+    if (note := _uncaptured_note(discovered.unresolved, remote_packages)) is not None:
+        files.append(note)
+    envelope = _pack_envelope(files)
+    compressed = zstd.compress(envelope, level=ZSTD_LEVEL)
+
+    _LOGGER.info(
+        "store_yaml: embedding %d file(s) as %d bytes (%d uncompressed, %.1f%% ratio)",
+        len(files),
+        len(compressed),
+        len(envelope),
+        100.0 * len(compressed) / len(envelope),
+    )
+
+    rhs = [HexInt(b) for b in compressed]
+    prog_arr = cg.progmem_array(config[CONF_RAW_DATA_ID], rhs)
+
+    var = cg.new_Pvariable(config[CONF_ID])
+    await cg.register_component(var, config)
+    cg.add(var.set_data(prog_arr, len(compressed), len(envelope)))

esphome/components/store_yaml/store_yaml.cpp

@@ -0,0 +1,27 @@
+#include "store_yaml.h"
+
+#ifdef USE_STORE_YAML
+
+#include "esphome/core/log.h"
+
+namespace esphome::store_yaml {
+
+static const char *const TAG = "store_yaml";
+
+// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+StoreYamlComponent *global_store_yaml = nullptr;
+
+void StoreYamlComponent::setup() { global_store_yaml = this; }
+
+void StoreYamlComponent::dump_config() {
+  ESP_LOGCONFIG(TAG,
+                "YAML:\n"
+                "  Compressed size: %zu bytes\n"
+                "  Uncompressed size: %zu bytes\n"
+                "  Encoding: %s",
+                this->size_, this->uncompressed_size_, ENCODING);
+}
+
+}  // namespace esphome::store_yaml
+
+#endif  // USE_STORE_YAML

esphome/components/store_yaml/store_yaml.h

@@ -0,0 +1,44 @@
+#pragma once
+
+#include "esphome/core/defines.h"
+#ifdef USE_STORE_YAML
+
+#include "esphome/core/component.h"
+#include "esphome/core/hal.h"
+
+namespace esphome::store_yaml {
+
+// "zstd" — published in GetYamlResponse.encoding so clients know how to decompress.
+constexpr const char *ENCODING = "zstd";
+
+class StoreYamlComponent : public Component {
+ public:
+  void setup() override;
+  void dump_config() override;
+
+  // Called once from codegen with the PROGMEM blob.
+  void set_data(const uint8_t *data, size_t size, size_t uncompressed_size) {
+    this->data_ = data;
+    this->size_ = size;
+    this->uncompressed_size_ = uncompressed_size;
+  }
+  size_t get_size() const { return this->size_; }
+
+  // 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.
+  const uint8_t *data_{nullptr};
+  size_t size_{0};
+  size_t uncompressed_size_{0};
+};
+
+// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+extern StoreYamlComponent *global_store_yaml;
+
+}  // namespace esphome::store_yaml
+
+#endif  // USE_STORE_YAML

esphome/core/defines.h

@@ -204,6 +204,7 @@
 #define USE_API_CUSTOM_SERVICES
 #define USE_API_USER_DEFINED_ACTION_RESPONSES
 #define USE_API_USER_DEFINED_ACTION_RESPONSES_JSON
+#define USE_STORE_YAML
 #define API_MAX_SEND_QUEUE 8
 #define MAX_API_CONNECTIONS 6
 #define USE_MD5

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
@@ -46,6 +47,10 @@ _LOGGER = logging.getLogger(__name__)
 SECRET_YAML = "secrets.yaml"
 _SECRET_CACHE = {}
 _SECRET_VALUES = {}
+# Stack of collectors (one per active secret_values_registered context); the
+# dumper records every emitted `!secret` name into the innermost one. YAML
+# processing is single-threaded.
+_EMITTED_SECRET_NAMES: list[set[str]] = []
 # Not thread-safe — config processing is single-threaded today.
 _load_listeners: list[Callable[[Path], None]] = []
 
@@ -267,12 +272,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,
-    _seen: set[int] | None = None,
-) -> None:
+) -> ForceLoadResult:
     """Recursively resolve any deferred ``IncludeFile`` instances in a YAML tree.
 
     Nested ``!include`` returns a deferred ``IncludeFile`` that is only resolved
@@ -287,50 +317,45 @@ def force_load_include_files(
     fresh re-parse where substitutions haven't been applied yet) to demote it
     to a debug log.
     """
-    if _seen is None:
-        _seen = set()
+    seen: set[int] = set()
+    unresolved: list[str] = []
+    errors: list[str] = []
 
-    if isinstance(obj, IncludeFile):
-        if id(obj) in _seen:
+    def walk(node: Any) -> None:
+        if not isinstance(node, (IncludeFile, dict, list, tuple)) or id(node) in seen:
             return
-        _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,
-            )
-            return
-        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
-        force_load_include_files(
-            loaded, warn_on_unresolved=warn_on_unresolved, _seen=_seen
-        )
-    elif isinstance(obj, dict):
-        if id(obj) in _seen:
-            return
-        _seen.add(id(obj))
-        for value in obj.values():
-            force_load_include_files(
-                value, warn_on_unresolved=warn_on_unresolved, _seen=_seen
-            )
-    elif isinstance(obj, (list, tuple)):
-        if id(obj) in _seen:
-            return
-        _seen.add(id(obj))
-        for item in obj:
-            force_load_include_files(
-                item, warn_on_unresolved=warn_on_unresolved, _seen=_seen
-            )
+        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,
+                )
+                errors.append(f"{node.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 ForceLoadResult(unresolved, errors)
 
 
 @dataclass(slots=True)
@@ -341,11 +366,16 @@ 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, 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:
@@ -375,9 +405,16 @@ def discover_user_yaml_files(config_path: Path) -> DiscoveredYamlFiles:
         try:
             try:
                 data = load_yaml(config_path)
-            except EsphomeError:
-                return DiscoveredYamlFiles(list(loaded), secrets)
-            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)
 
@@ -388,7 +425,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, load_errors)
 
 
 def _add_data_ref(fn):
@@ -840,6 +877,38 @@ 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[set[str]]:
+    """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.
+
+    Yields a set that collects the name of every ``!secret`` reference the
+    dumper emits while the context is active, so callers can tell exactly
+    which registered values were actually swapped.
+    """
+    added = {v: n for v, n in values.items() if v not in _SECRET_VALUES}
+    _SECRET_VALUES.update(added)
+    emitted: set[str] = set()
+    _EMITTED_SECRET_NAMES.append(emitted)
+    try:
+        yield emitted
+    finally:
+        # Contexts unwind LIFO, so the innermost collector is always last;
+        # pop() removes by position where remove() would match the first
+        # *equal* set and could strip an outer context's collector.
+        _EMITTED_SECRET_NAMES.pop()
+        for value in added:
+            _SECRET_VALUES.pop(value, None)
+
+
 def dump(dict_, show_secrets=False, sort_keys=False, relative_to: Path | None = None):
     """Dump YAML to a string and remove null.
 
@@ -1042,7 +1111,10 @@ class ESPHomeDumper(yaml.SafeDumper):
         return node
 
     def represent_secret(self, value):
-        return self.represent_scalar(tag="!secret", value=_SECRET_VALUES[str(value)])
+        name = _SECRET_VALUES[str(value)]
+        if _EMITTED_SECRET_NAMES:
+            _EMITTED_SECRET_NAMES[-1].add(name)
+        return self.represent_scalar(tag="!secret", value=name)
 
     def represent_stringify(self, value):
         if is_secret(value):
@@ -1117,17 +1189,27 @@ class ESPHomeDumper(yaml.SafeDumper):
         return self.represent_scalar(tag="!lambda", value=value.value, style="|")
 
     def represent_extend(self, value):
+        # Consult is_secret like the other scalar representers so a payload
+        # equal to a registered secret is never written out in cleartext.
+        if is_secret(value.value):
+            return self.represent_secret(value.value)
         return self.represent_scalar(tag="!extend", value=value.value)
 
     def represent_remove(self, value):
+        if is_secret(value.value):
+            return self.represent_secret(value.value)
         return self.represent_scalar(tag="!remove", value=value.value)
 
     def represent_include_file(self, value):
         if value.vars:
+            # The mapping values route through the regular representers,
+            # which already consult is_secret.
             mapping = {"file": value.file.as_posix(), "vars": value.vars}
             return self.represent_mapping(
                 tag="!include", mapping=mapping, flow_style=False
             )
+        if is_secret(value.file.as_posix()):
+            return self.represent_secret(value.file.as_posix())
         return self.represent_scalar(tag="!include", value=value.file.as_posix())
 
     def represent_id(self, value):

requirements.txt

@@ -29,6 +29,9 @@ py7zr==1.1.3
 platformdirs==4.11.0  # native esp-idf toolchain global cache dir
 filelock==3.32.0  # lock guarding the PlatformIO python-version cache heal
 
+# zstd compression for store_yaml component (stdlib in 3.14+)
+backports.zstd==1.5.0; python_version < "3.14"
+
 # esp-idf >= 5.0 requires this
 pyparsing >= 3.3.2
 

script/analyze_component_buses.py

@@ -94,6 +94,7 @@ ISOLATED_COMPONENTS = {
     "modbus_controller": "Defines multiple modbus buses for testing client/server functionality - conflicts with package modbus bus",
     "neopixelbus": "RMT type conflict with ESP32 Arduino/ESP-IDF headers (enum vs struct rmt_channel_t)",
     "packages": "cannot merge packages",
+    "store_yaml": "Embeds the whole merged config in firmware; grouping would make the blob and its secret redaction depend on every grouped component's test config",
     "tinyusb": "Conflicts with usb_host component - cannot be used together",
     "usb_cdc_acm": "Depends on tinyusb which conflicts with usb_host",
 }

tests/components/store_yaml/common.yaml

@@ -0,0 +1,4 @@
+api:
+
+store_yaml:
+  allow_unencrypted: true

tests/components/store_yaml/test.bk72xx-ard.yaml

@@ -0,0 +1,6 @@
+wifi:
+  ssid: MySSID
+  password: password1
+
+packages:
+  store_yaml: !include common.yaml

tests/components/store_yaml/test.esp32-idf.yaml

@@ -0,0 +1,6 @@
+wifi:
+  ssid: MySSID
+  password: password1
+
+packages:
+  store_yaml: !include common.yaml

tests/components/store_yaml/test.esp8266-ard.yaml

@@ -0,0 +1,6 @@
+wifi:
+  ssid: MySSID
+  password: password1
+
+packages:
+  store_yaml: !include common.yaml

tests/components/store_yaml/test.host.yaml

@@ -0,0 +1,4 @@
+network:
+
+packages:
+  store_yaml: !include common.yaml

tests/components/store_yaml/test.ln882x-ard.yaml

@@ -0,0 +1,6 @@
+wifi:
+  ssid: MySSID
+  password: password1
+
+packages:
+  store_yaml: !include common.yaml

tests/components/store_yaml/test.rp2040-ard.yaml

@@ -0,0 +1,6 @@
+wifi:
+  ssid: MySSID
+  password: password1
+
+packages:
+  store_yaml: !include common.yaml

tests/components/store_yaml/test.rtl87xx-ard.yaml

@@ -0,0 +1,6 @@
+wifi:
+  ssid: MySSID
+  password: password1
+
+packages:
+  store_yaml: !include common.yaml

tests/components/store_yaml/validate.esp32-idf.yaml

@@ -0,0 +1,9 @@
+wifi:
+  ssid: MySSID
+  password: password1
+
+api:
+  encryption:
+    key: AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=
+
+store_yaml:

tests/integration/conftest.py

@@ -10,6 +10,7 @@ import logging
 import os
 from pathlib import Path
 import platform
+import re
 import signal
 import socket
 import subprocess
@@ -178,10 +179,9 @@ async def yaml_config(request: pytest.FixtureRequest, unused_tcp_port: int) -> s
     loop = asyncio.get_running_loop()
     content = await loop.run_in_executor(None, fixture_path.read_text)
 
-    # Replace the port in the config if it contains api section
-    if "api:" in content:
-        # Add port configuration after api:
-        content = content.replace("api:", f"api:\n  port: {unused_tcp_port}")
+    # Replace the port in the config if it contains an api section. Anchored to
+    # the start of a line so keys that merely end in "api:" are left alone.
+    content = re.sub(r"(?m)^api:", f"api:\n  port: {unused_tcp_port}", content)
 
     # Add debug build flags for integration tests to enable assertions
     if "esphome:" in content and "platformio_options:" not in content:

tests/integration/fixtures/store_yaml_recovery.yaml

@@ -0,0 +1,18 @@
+esphome:
+  name: store-yaml-test
+  areas:
+    - id: living_room
+      name: "Living Room"
+
+host:
+
+logger:
+
+api:
+
+ota:
+  - platform: esphome
+    password: recoverme123
+
+store_yaml:
+  allow_unencrypted: true

tests/integration/test_store_yaml_recovery.py

@@ -0,0 +1,217 @@
+"""End-to-end test for the `store_yaml` recovery flow over the native API.
+
+Talks plaintext API to a host build directly via asyncio sockets rather than
+through aioesphomeapi: the released aioesphomeapi shipped with this PR does
+not yet know about `GetYamlRequest` / `GetYamlResponse`, so the high-level
+client would silently drop the streamed bytes as "unknown message type".
+
+The raw client implements just enough of the plaintext framing
+(``0x00 | varint(size) | varint(msg_type) | payload``, see
+``api_frame_helper_plaintext.cpp``) to send the empty `GetYamlRequest`
+(message type 149) and accumulate every `GetYamlResponse` (message type 150)
+until ``done=true``.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+
+import pytest
+
+# The component resolves the stdlib-vs-backport zstd import once; reuse it.
+from esphome.components.store_yaml import unpack_envelope, zstd
+from esphome.yaml_util import find_secret_references
+
+from .types import RunCompiledFunction
+
+# Message IDs from esphome/components/api/api.proto.
+HELLO_REQUEST = 1
+HELLO_RESPONSE = 2
+GET_YAML_REQUEST = 149
+GET_YAML_RESPONSE = 150
+
+
+def _encode_varint(value: int) -> bytes:
+    """Encode an unsigned integer as a protobuf varint."""
+    out = bytearray()
+    while True:
+        byte = value & 0x7F
+        value >>= 7
+        if value:
+            out.append(byte | 0x80)
+        else:
+            out.append(byte)
+            return bytes(out)
+
+
+def _read_varint(buf: bytes, pos: int) -> tuple[int, int]:
+    result = 0
+    shift = 0
+    while True:
+        b = buf[pos]
+        pos += 1
+        result |= (b & 0x7F) << shift
+        if not (b & 0x80):
+            return result, pos
+        shift += 7
+
+
+def _parse_get_yaml_response(payload: bytes) -> tuple[bytes, bool, int, str]:
+    """Hand-rolled parser for `GetYamlResponse`.
+
+    Returns ``(data, done, total_size, encoding)``.
+    """
+    data = b""
+    done = False
+    total_size = 0
+    encoding = ""
+    pos = 0
+    while pos < len(payload):
+        tag, pos = _read_varint(payload, pos)
+        field_number = tag >> 3
+        wire_type = tag & 0x07
+        if wire_type == 0:  # varint
+            value, pos = _read_varint(payload, pos)
+            if field_number == 2:
+                done = bool(value)
+            elif field_number == 3:
+                total_size = value
+        elif wire_type == 2:  # length-delimited
+            length, pos = _read_varint(payload, pos)
+            chunk = payload[pos : pos + length]
+            pos += length
+            if field_number == 1:
+                data = chunk
+            elif field_number == 4:
+                encoding = chunk.decode("utf-8")
+        else:
+            raise AssertionError(f"unexpected wire type {wire_type}")
+    return data, done, total_size, encoding
+
+
+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:
+    """Just-enough plaintext API client for one short streaming exchange."""
+
+    def __init__(
+        self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter
+    ) -> None:
+        self._reader = reader
+        self._writer = writer
+
+    async def send(self, msg_type: int, payload: bytes = b"") -> None:
+        # Frame: 0x00 | varint(payload_size) | varint(message_id) | payload
+        frame = (
+            b"\x00" + _encode_varint(len(payload)) + _encode_varint(msg_type) + payload
+        )
+        self._writer.write(frame)
+        await self._writer.drain()
+
+    async def recv(self) -> tuple[int, bytes]:
+        # Read preamble byte (must be 0x00 for plaintext).
+        preamble = await self._reader.readexactly(1)
+        assert preamble == b"\x00", f"unexpected preamble {preamble!r}"
+
+        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
+
+
+@pytest.mark.asyncio
+async def test_store_yaml_recovery(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    unused_tcp_port: int,
+) -> None:
+    """Compile a host build with `store_yaml`, ask it to stream the YAML back,
+    decompress, and verify the recovered file tree matches the source fixture."""
+    async with run_compiled(yaml_config):
+        # Open a raw TCP connection to the API server.
+        reader, writer = await asyncio.wait_for(
+            asyncio.open_connection("127.0.0.1", unused_tcp_port),
+            timeout=10.0,
+        )
+        client = _PlaintextClient(reader, writer)
+        try:
+            # HelloRequest: client_info (field 1, length-delimited string).
+            # Password auth (the old ConnectRequest/Response exchange at message
+            # IDs 3/4) was removed in 2026.1.0, so a successful HelloResponse is
+            # all the handshake we need before issuing application requests.
+            client_info = b"store_yaml integration test"
+            api_version = b"\x10\x01\x18\x0e"  # api_version_major=1, minor=14
+            hello_payload = (
+                b"\x0a" + _encode_varint(len(client_info)) + client_info + api_version
+            )
+            await client.send(HELLO_REQUEST, hello_payload)
+            msg_type, _ = await asyncio.wait_for(client.recv(), timeout=5.0)
+            assert msg_type == HELLO_RESPONSE, f"expected HelloResponse, got {msg_type}"
+
+            # The actual request under test.
+            await client.send(GET_YAML_REQUEST, b"")
+
+            chunks: list[bytes] = []
+            advertised_total: int | None = None
+            advertised_encoding: str | None = None
+            done = False
+            while not done:
+                msg_type, payload = await asyncio.wait_for(client.recv(), timeout=5.0)
+                if msg_type != GET_YAML_RESPONSE:
+                    # Tolerate intervening server messages (e.g. pings).
+                    continue
+                chunk, done, total_size, encoding = _parse_get_yaml_response(payload)
+                if encoding:
+                    advertised_encoding = encoding
+                if total_size and advertised_total is None:
+                    advertised_total = total_size
+                if chunk:
+                    chunks.append(chunk)
+        finally:
+            writer.close()
+            with contextlib.suppress(ConnectionError, OSError):
+                await writer.wait_closed()
+
+    compressed = b"".join(chunks)
+    assert advertised_encoding == "zstd", (
+        f"expected encoding 'zstd', got {advertised_encoding!r}"
+    )
+    assert advertised_total == len(compressed), (
+        f"server advertised {advertised_total} bytes but we received {len(compressed)}"
+    )
+
+    envelope = zstd.decompress(compressed)
+    files = unpack_envelope(envelope)
+
+    assert files, "envelope should contain at least one file"
+    combined = b"\n".join(files.values())
+    assert b"store-yaml-test" in combined, (
+        "expected the fixture's device name to round-trip through the recovery blob"
+    )
+    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 "ota_password" in find_secret_references(combined.decode()), (
+        "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

@@ -0,0 +1,642 @@
+"""Tests for the store_yaml component's file gathering, secret redaction, and
+envelope packing."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from esphome import yaml_util
+from esphome.components import packages
+from esphome.components.store_yaml import (
+    CONF_ALLOW_UNENCRYPTED,
+    SECRETS_SKELETON_HEADER,
+    UNCAPTURED_NOTE_PATH,
+    _final_validate,
+    _gather_files,
+    _generate_redacted_files,
+    _pack_envelope,
+    _read_files_verbatim,
+    _remote_package_descriptions,
+    _uncaptured_note,
+    unpack_envelope,
+)
+import esphome.config_validation as cv
+from esphome.core import CORE, EsphomeError
+import esphome.final_validate as fv
+from esphome.yaml_util import DiscoveredYamlFiles, SensitiveStr
+
+
+@pytest.fixture
+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\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 = {}
+
+
+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}
+    return DiscoveredYamlFiles(files, secret_paths)
+
+
+def _gather_redacted(discovered: DiscoveredYamlFiles) -> dict[str, bytes]:
+    entries, secret_rels = _gather_files(discovered)
+    return dict(_generate_redacted_files(entries, secret_rels))
+
+
+# ---------------------------------------------------------------------------
+# _gather_files
+# ---------------------------------------------------------------------------
+
+
+def test_gather_maps_rel_paths_and_flags_secrets(project: Path) -> None:
+    discovered = _sources(
+        project, "entry.yaml", "secrets.yaml", secrets=("secrets.yaml",)
+    )
+    entries, secret_rels = _gather_files(discovered)
+    assert dict(entries) == {
+        "entry.yaml": project / "entry.yaml",
+        "secrets.yaml": project / "secrets.yaml",
+    }
+    assert secret_rels == {"secrets.yaml"}
+
+
+def test_read_files_verbatim_returns_exact_bytes(project: Path) -> None:
+    """`include_secrets: true` embeds the on-disk bytes untouched."""
+    discovered = _sources(
+        project, "entry.yaml", "secrets.yaml", secrets=("secrets.yaml",)
+    )
+    entries, _ = _gather_files(discovered)
+    contents = dict(_read_files_verbatim(entries))
+    assert contents["secrets.yaml"] == (project / "secrets.yaml").read_bytes()
+    assert contents["entry.yaml"] == (project / "entry.yaml").read_bytes()
+
+
+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 flagged
+    because the un-resolved basename was captured upstream."""
+    target = tmp_path / "actual_creds.yaml"
+    target.write_text("api_key: FROM_SYMLINK\n")
+    link = project / "secrets.yaml"
+    link.unlink()  # remove the regular file laid down by the fixture
+    link.symlink_to(target)
+    # Discovery records the un-resolved listener fname under SECRETS_FILES
+    # but stores the resolved path; mimic that here.
+    resolved = link.resolve()
+    CORE.config_path = project / "entry.yaml"
+    files = _gather_redacted(DiscoveredYamlFiles([resolved], {resolved}))
+    assert b"FROM_SYMLINK" not in b"".join(files.values())
+
+
+def test_gather_uses_relative_path_for_external_files(
+    project: Path, tmp_path: Path
+) -> None:
+    """Files outside the project root use a ``..``-style relative path so they don't collide."""
+    sibling = tmp_path / "outside.yaml"
+    sibling.write_text("foo: bar\n")
+    CORE.config_path = project / "entry.yaml"
+    discovered = DiscoveredYamlFiles([project / "entry.yaml", sibling], set())
+    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 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())
+
+
+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_read_files_verbatim_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."""
+    discovered = _sources(project, "entry.yaml", "wifi.yaml")
+    entries, _ = _gather_files(discovered)
+    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"):
+        _read_files_verbatim(entries)
+
+
+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"
+    discovered = DiscoveredYamlFiles([project / "entry.yaml"], set(), ["${board}.yaml"])
+    with caplog.at_level("WARNING", logger="esphome.components.store_yaml"):
+        files, _ = _gather_files(discovered)
+    assert len(files) == 1
+    assert any(
+        "${board}.yaml" in r.message and "not contain" in r.message
+        for r in caplog.records
+    )
+
+
+# ---------------------------------------------------------------------------
+# _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). The recovered config stays semantically identical
+    once the secret is filled."""
+    (project / "wifi.yaml").write_text("platform: esp32\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
+
+
+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_raises_when_value_not_locatable(project: Path) -> None:
+    """A sensitive value that never appears as a whole scalar (e.g. composed
+    via substitutions) would ship verbatim — fail the build, naming the config
+    path but never the value."""
+    CORE.config = {"wifi": [{"password": SensitiveStr("not_in_any_file")}]}
+    discovered = _sources(project, "wifi.yaml")
+    with pytest.raises(EsphomeError, match="wifi.password") as err:
+        _gather_redacted(discovered)
+    assert "not_in_any_file" not in str(err.value)
+
+
+def test_redacted_accepts_secret_only_values(project: Path) -> 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",)
+    )
+    files = _gather_redacted(discovered)
+    assert 'api_key: ""' in files["secrets.yaml"].decode()
+
+
+def test_uncaptured_note_lists_missing_includes() -> None:
+    """Substitution-pathed includes that can't be captured are recorded in a
+    dedicated envelope entry (both modes), not just a compile-time log line."""
+    rel, content = _uncaptured_note(["${board}.yaml"], [])
+    assert rel == UNCAPTURED_NOTE_PATH
+    text = content.decode()
+    assert text.startswith("# store_yaml:")
+    assert "#   ${board}.yaml" in text
+
+
+def test_uncaptured_note_lists_remote_packages() -> None:
+    """Remote packages that can't be captured are recorded with their source
+    so the user knows to re-fetch them."""
+    rel, content = _uncaptured_note(
+        [], ["https://github.com/org/repo@main", "https://github.com/org/other"]
+    )
+    assert rel == UNCAPTURED_NOTE_PATH
+    text = content.decode()
+    assert "#   https://github.com/org/repo@main" in text
+    assert "#   https://github.com/org/other" in text
+
+
+def test_remote_package_descriptions_read_packages_record(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Remote sources recorded by the packages component during config
+    processing are formatted as url@ref (url alone when ref is absent)."""
+    monkeypatch.delitem(CORE.data, packages.DOMAIN, raising=False)
+    data = packages._get_data()
+    data.remote_sources.append(
+        packages.RemotePackageSource("https://github.com/org/repo", "main")
+    )
+    data.remote_sources.append(
+        packages.RemotePackageSource("https://github.com/org/other", None)
+    )
+    assert _remote_package_descriptions() == [
+        "https://github.com/org/repo@main",
+        "https://github.com/org/other",
+    ]
+
+
+def test_remote_package_descriptions_empty_without_packages() -> None:
+    CORE.data.pop(packages.DOMAIN, None)
+    assert _remote_package_descriptions() == []
+
+
+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"),
+        ("wifi.yaml", b"ssid: a\n"),
+    ]
+    blob = _pack_envelope(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)
+
+
+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)
+
+
+@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")
+
+
+def test_unpack_envelope_rejects_invalid_utf8_path() -> None:
+    """A corrupted envelope raises EsphomeError, never a bare UnicodeDecodeError."""
+    import struct
+
+    blob = (
+        b"EHY1"
+        + struct.pack("<I", 1)
+        + struct.pack("<H", 2)
+        + b"\xff\xfe"
+        + struct.pack("<I", 0)
+    )
+    with pytest.raises(EsphomeError, match="UTF-8"):
+        unpack_envelope(blob)
+
+
+def test_unpack_envelope_rejects_duplicate_paths() -> None:
+    """A tampered envelope with duplicate paths must not silently drop data."""
+    import struct
+
+    entry = struct.pack("<H", 6) + b"a.yaml" + struct.pack("<I", 1) + b"x"
+    blob = b"EHY1" + struct.pack("<I", 2) + entry + entry
+    with pytest.raises(EsphomeError, match="duplicate"):
+        unpack_envelope(blob)
+
+
+def test_pack_envelope_rejects_duplicate_paths() -> None:
+    """A duplicate path would silently clobber the earlier entry on unpack."""
+    with pytest.raises(EsphomeError, match="duplicate"):
+        _pack_envelope([("entry.yaml", b"a: 1\n"), ("entry.yaml", b"b: 2\n")])
+
+
+@pytest.mark.parametrize("path", ["/etc/passwd", "\\evil.yaml", "C:/evil.yaml"])
+def test_unpack_envelope_rejects_non_relative_paths(path: str) -> None:
+    """The packer never emits absolute or drive-qualified paths, so their
+    presence means a malformed or hostile envelope."""
+    blob = _pack_envelope([(path, b"boom\n")])
+    with pytest.raises(EsphomeError, match="non-relative"):
+        unpack_envelope(blob)
+
+
+# ---------------------------------------------------------------------------
+# embedded-value leak scan and collision warning
+# ---------------------------------------------------------------------------
+
+
+def test_redacted_embedded_sensitive_value_fails_build(project: Path) -> None:
+    """A sensitive value inside a larger scalar (URL, lambda body) is not
+    swapped by the whole-scalar redaction; the substring scan fails closed."""
+    (project / "wifi.yaml").write_text(
+        "password: my_password\nurl: http://user:my_password@host\n"
+    )
+    CORE.config = {"wifi": [{"password": SensitiveStr("my_password")}]}
+    discovered = _sources(project, "wifi.yaml")
+    with pytest.raises(EsphomeError, match="embedded"):
+        _gather_redacted(discovered)
+
+
+def test_redacted_warns_on_value_collision(
+    project: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """An unrelated scalar equal to a sensitive value gets rewritten by the
+    value-keyed swap; a warning documents the trap. The value's own
+    occurrence (under its sensitive key) does not warn."""
+    (project / "wifi.yaml").write_text("password: esp32\nplatform: esp32\n")
+    CORE.config = {"wifi": [{"password": SensitiveStr("esp32")}]}
+    discovered = _sources(project, "wifi.yaml")
+    files = _gather_redacted(discovered)
+    assert files["wifi.yaml"] == (
+        b"password: !secret 'wifi_password'\nplatform: !secret 'wifi_password'\n"
+    )
+    assert "also matches the scalar at platform in wifi.yaml" in caplog.text
+    assert "at password in wifi.yaml" not in caplog.text
+
+
+def test_redacted_no_warning_for_substitution_definition(
+    project: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A swapped `substitutions:` definition keeps `${...}` working in the
+    recovered config, so it is expected and does not warn."""
+    (project / "wifi.yaml").write_text(
+        "substitutions:\n  wifi_password: hunter2\nwifi:\n  password: ${wifi_password}\n"
+    )
+    CORE.config = {"wifi": [{"password": SensitiveStr("hunter2")}]}
+    discovered = _sources(project, "wifi.yaml")
+    _gather_redacted(discovered)
+    assert "also matches" not in caplog.text
+
+
+def test_redacted_sensitive_value_as_mapping_key_fails_build(project: Path) -> None:
+    """A sensitive value equal to a mapping key would be swapped in key
+    position and corrupt the recovered structure; the build fails instead."""
+    (project / "wifi.yaml").write_text("password: password\n")
+    CORE.config = {"ota": [{"password": SensitiveStr("password")}]}
+    discovered = _sources(project, "wifi.yaml")
+    with pytest.raises(EsphomeError, match="mapping key"):
+        _gather_redacted(discovered)
+
+
+def test_redacted_key_names_do_not_false_positive_embedded_scan(
+    project: Path,
+) -> None:
+    """The embedded scan runs on tree scalars, not serialized text, so a
+    sensitive value that is a substring of a key name (or of the generated
+    `!secret` reference text) does not fail the build."""
+    (project / "wifi.yaml").write_text("password: word\n")
+    CORE.config = {"wifi": [{"password": SensitiveStr("word")}]}
+    discovered = _sources(project, "wifi.yaml")
+    files = _gather_redacted(discovered)
+    assert files["wifi.yaml"] == b"password: !secret 'wifi_password'\n"
+
+
+# ---------------------------------------------------------------------------
+# _final_validate encryption gate
+# ---------------------------------------------------------------------------
+
+
+def _run_final_validate(full_config: dict, config: dict) -> dict:
+    token = fv.full_config.set(full_config)
+    try:
+        return _final_validate(config)
+    finally:
+        fv.full_config.reset(token)
+
+
+def test_final_validate_accepts_encrypted_api() -> None:
+    config = {CONF_ALLOW_UNENCRYPTED: False}
+    full = {
+        "api": {"encryption": {"key": "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="}}
+    }
+    assert _run_final_validate(full, config) is config
+
+
+def test_final_validate_rejects_unencrypted_api() -> None:
+    with pytest.raises(cv.Invalid, match="requires API encryption"):
+        _run_final_validate({"api": {}}, {CONF_ALLOW_UNENCRYPTED: False})
+
+
+def test_final_validate_allows_unencrypted_with_escape_hatch(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    config = {CONF_ALLOW_UNENCRYPTED: True}
+    assert _run_final_validate({"api": {}}, config) is config
+    assert "without API encryption" in caplog.text
+
+
+def test_redacted_lambda_body_leak_fails_build(project: Path) -> None:
+    """A sensitive value inside a !lambda body would be emitted verbatim by
+    the dumper; the scan must catch it."""
+    (project / "wifi.yaml").write_text(
+        "password: my_password\nx: !lambda 'return \"my_password\";'\n"
+    )
+    CORE.config = {"wifi": [{"password": SensitiveStr("my_password")}]}
+    discovered = _sources(project, "wifi.yaml")
+    with pytest.raises(EsphomeError, match="embedded"):
+        _gather_redacted(discovered)
+
+
+def test_redacted_include_vars_leak_fails_build(project: Path) -> None:
+    """A sensitive value inside an !include vars: mapping is emitted by the
+    dumper; the scan must catch it."""
+    (project / "entry.yaml").write_text(
+        "password: my_password\n"
+        "pkg: !include {file: wifi.yaml, vars: {url: 'http://user:my_password@host'}}\n"
+    )
+    CORE.config = {"wifi": [{"password": SensitiveStr("my_password")}]}
+    discovered = _sources(project, "entry.yaml", "wifi.yaml")
+    with pytest.raises(EsphomeError, match="embedded"):
+        _gather_redacted(discovered)
+
+
+def test_redacted_secret_sourced_overlap_warns_not_fails(
+    project: Path, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A value from a real !secret appearing inside another scalar (SSID in an
+    entity name) warns instead of failing; the remedy text of the hard error
+    does not apply to it."""
+    yaml_util._SECRET_VALUES["Kitchen"] = "wifi_ssid"
+    (project / "wifi.yaml").write_text("ssid: Kitchen\nname: Kitchen Temperature\n")
+    CORE.config = {"wifi": [{"ssid": SensitiveStr("Kitchen")}]}
+    discovered = _sources(project, "wifi.yaml")
+    files = _gather_redacted(discovered)
+    assert b"!secret 'wifi_ssid'" in files["wifi.yaml"]
+    assert "appears inside the value at name in wifi.yaml" in caplog.text
+
+
+def test_final_validate_rejects_keyless_encryption() -> None:
+    """A keyless provisionable `api: encryption:` accepts the all-zeros PSK
+    until provisioned, so it must not satisfy the gate."""
+    with pytest.raises(cv.Invalid, match="requires API encryption"):
+        _run_final_validate(
+            {"api": {"encryption": {}}}, {CONF_ALLOW_UNENCRYPTED: False}
+        )
+
+
+def test_redacted_outside_root_secrets_gets_root_skeleton(
+    project: Path, tmp_path: Path
+) -> None:
+    """A secrets file resolving outside the config root still yields a root
+    secrets.yaml skeleton, so the recovered config can resolve !secret."""
+    target = tmp_path / "actual_creds.yaml"
+    target.write_text("api_key: SUPER_SECRET\n")
+    link = project / "secrets.yaml"
+    link.unlink()
+    link.symlink_to(target)
+    resolved = link.resolve()
+    CORE.config_path = project / "entry.yaml"
+    files = _gather_redacted(
+        DiscoveredYamlFiles([project / "entry.yaml", resolved], {resolved})
+    )
+    assert "secrets.yaml" in files
+    assert 'api_key: ""' in files["secrets.yaml"].decode()
+    assert b"SUPER_SECRET" not in b"".join(files.values())
+
+
+def test_gather_raises_esphome_error_on_cross_anchor_path(
+    project: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A file that cannot be made relative to the config root (a Windows
+    cross-drive path) surfaces as EsphomeError, not a raw ValueError."""
+
+    def fake_relative_to(self: Path, other: Path, walk_up: bool = False) -> Path:
+        raise ValueError("paths have different anchors")
+
+    monkeypatch.setattr(Path, "relative_to", fake_relative_to)
+    discovered = _sources(project, "entry.yaml")
+    with pytest.raises(EsphomeError, match="does not share a root"):
+        _gather_files(discovered)
+
+
+def test_final_validate_unencrypted_with_secrets_names_secrets_yaml(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """allow_unencrypted combined with include_secrets warns about the
+    verbatim secrets.yaml specifically."""
+    config = {CONF_ALLOW_UNENCRYPTED: True, "include_secrets": True}
+    assert _run_final_validate({"api": {}}, config) is config
+    assert "verbatim contents of secrets.yaml" in caplog.text

tests/unit_tests/conftest.py

@@ -16,6 +16,7 @@ from unittest.mock import Mock, patch
 
 import pytest
 
+from esphome import yaml_util
 from esphome.core import CORE
 
 here = Path(__file__).parent
@@ -32,6 +33,16 @@ def reset_core():
     CORE.reset()
 
 
+@pytest.fixture(autouse=True)
+def clear_yaml_secrets():
+    """Isolate the yaml_util secrets registry between tests."""
+    yaml_util._SECRET_VALUES.clear()
+    yaml_util._SECRET_CACHE.clear()
+    yield
+    yaml_util._SECRET_VALUES.clear()
+    yaml_util._SECRET_CACHE.clear()
+
+
 @pytest.fixture
 def fixture_path() -> Path:
     """

tests/unit_tests/test_yaml_util.py

@@ -25,16 +25,6 @@ from esphome.yaml_util import (
 )
 
 
-@pytest.fixture(autouse=True)
-def clear_secrets_cache() -> None:
-    """Clear the secrets cache before each test."""
-    yaml_util._SECRET_VALUES.clear()
-    yaml_util._SECRET_CACHE.clear()
-    yield
-    yaml_util._SECRET_VALUES.clear()
-    yaml_util._SECRET_CACHE.clear()
-
-
 @pytest.fixture(autouse=True)
 def clear_core_frontmatter() -> None:
     """Reset CORE.frontmatter between tests."""
@@ -1084,22 +1074,38 @@ 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")
+    result = force_load_include_files({"a": templated, "b": plain})
+    assert result.unresolved == [str(templated.file)]
+    assert result.errors == []
+    assert plain.load_calls == 1
+
+
 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:
-    """`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)
@@ -1162,11 +1168,28 @@ 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(
+    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:
@@ -1448,6 +1471,49 @@ 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_collects_emitted_names() -> None:
+    """The context yields a set recording exactly the `!secret` names the
+    dumper emitted, including real secrets, and not names never swapped."""
+    yaml_util._SECRET_VALUES["real_value"] = "real_name"
+    with yaml_util.secret_values_registered({"hunter2": "wifi_password"}) as emitted:
+        yaml_util.dump(
+            {
+                "password": make_data_base("hunter2"),
+                "key": make_data_base("real_value"),
+                "plain": make_data_base("nothing"),
+            }
+        )
+    assert emitted == {"wifi_password", "real_name"}
+
+
+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"
+    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"
+
+
+def test_registered_secret_names() -> None:
+    yaml_util._SECRET_VALUES["value_a"] = "name_a"
+    assert "name_a" in yaml_util.registered_secret_names()
+
+
 @pytest.fixture(autouse=True)
 def clear_dropped_merge_keys() -> None:
     """Reset the dropped-merge-key queue between tests."""
@@ -1491,3 +1557,32 @@ def test_merge_include_no_overlap_records_nothing(tmp_path: Path) -> None:
     assert result["api"] == {"reboot_timeout": "5min"}
     assert result["logger"] == {"level": "DEBUG"}
     assert yaml_util.take_dropped_merge_keys() == []
+
+
+def test_wrapper_representers_consult_is_secret() -> None:
+    """!extend / !remove payloads and scalar !include paths equal to a
+    registered secret are swapped, never written in cleartext."""
+    from esphome.config_helpers import Extend, Remove
+
+    with yaml_util.secret_values_registered({"hunter2": "the_secret"}):
+        out = yaml_util.dump(
+            {
+                "a": Extend("hunter2"),
+                "b": Remove("hunter2"),
+                "c": Extend("plain_id"),
+            }
+        )
+    assert out.count("!secret 'the_secret'") == 2
+    assert "hunter2" not in out
+    assert "!extend 'plain_id'" in out
+
+
+def test_scalar_include_path_equal_to_secret_is_swapped() -> None:
+    """A scalar !include whose path equals a registered secret is swapped,
+    matching the other wrapper representers."""
+    include = yaml_util.IncludeFile(
+        Path("/fake/main.yaml"), "hunter2", None, lambda _: {}
+    )
+    with yaml_util.secret_values_registered({"hunter2": "the_secret"}):
+        out = yaml_util.dump({"key": include})
+    assert out == "key: !secret 'the_secret'\n"