[core] Assert a frame came back before each resync() in the FrameReader tests

AGENTS.md

@@ -216,9 +216,10 @@ file does, and it is the authority when they disagree. The most useful starting
         the helper's arguments instead of writing a loop. 7-byte BCD RTC calendars use `time::decode_bcd_calendar<LAYOUT>`
         and `encode_bcd_calendar<LAYOUT>` with `EPSON_BCD_CALENDAR`, `DS_BCD_CALENDAR` or `PCF_BCD_CALENDAR`.
     *   **Bus helpers:** UART drains with `read_available(on_byte)` and `discard_available()`, text lines with
-        `LineReader<N>`; I2C has `update_register`, `write_registers` (PROGMEM `I2CRegValue` table), `check_chip_id`
-        and `read_bytes_16_le`. `if (!this->status_comm_result(ok)) return;` sets or clears the warning.
-        A child sensor that skips repeated values is `SUB_SENSOR_WITH_DEDUP(name, type)`.
+        `LineReader<N>`, fixed-length binary frames with `FrameReader<N, HEADER...>`; I2C has `update_register`,
+        `write_registers` (PROGMEM `I2CRegValue` table), `check_chip_id` and `read_bytes_16_le`.
+        `if (!this->status_comm_result(ok)) return;` sets or clears the warning. A child sensor that skips repeated
+        values is `SUB_SENSOR_WITH_DEDUP(name, type)`.
     *   **An unset float is `NAN`**, not `optional<float>`; the optional costs 4 to 8 bytes of padding per member.
     *   **`PollingComponent` first runs `update()`** after a random share of the interval; call `this->update()` at the
         end of `setup()` when the entities need a state at boot.

esphome/config_validation.py

@@ -570,7 +570,8 @@ class SensitiveValidator:
     def __call__(self, value: typing.Any) -> typing.Any:
         validated = self.inner(value)
         # Already-tagged values are returned as-is so nested cv.sensitive is
-        # idempotent; bool is an int subclass and must stay bool.
+        # idempotent; bool is an int subclass and must stay bool. Other types
+        # (e.g. lambdas) pass through unmasked.
         if isinstance(validated, (Sensitive, bool)):
             return validated
         if isinstance(validated, str):
@@ -579,9 +580,6 @@ class SensitiveValidator:
             return (SensitiveHexInt if isinstance(validated, HexInt) else SensitiveInt)(
                 validated
             )
-        if isinstance(validated, Lambda):
-            # Marked, not masked: the dumper keeps the lambda representer.
-            return add_class_to_obj(validated, Sensitive)
         return validated
 
     def __repr__(self) -> str:

esphome/core/frame_reader.h

@@ -0,0 +1,67 @@
+#pragma once
+
+#include <cstddef>
+#include <cstdint>
+#include <type_traits>
+
+namespace esphome {
+
+/// Collects fixed length binary frames of N bytes that start with the header bytes HEADER...; the binary twin of
+/// LineReader. Bytes before a header are skipped; checksums are left to the caller.
+template<size_t N, uint8_t... HEADER> class FrameReader {
+  using pos_t = std::conditional_t<(N <= 0xFF), uint8_t, uint16_t>;
+  static constexpr size_t HEADER_LEN = sizeof...(HEADER);
+  static_assert(HEADER_LEN >= 1 && HEADER_LEN <= 8, "FrameReader takes 1 to 8 header bytes");  // bounds replay_() depth
+  static_assert(HEADER_LEN < N && N <= 0xFFFF, "FrameReader size out of range");
+
+ public:
+  /// Feed one byte; returns the completed frame, header included, valid until the next call, or nullptr.
+  const uint8_t *feed(uint8_t c) {
+    pos_t pos = this->pos_;
+    if (pos < HEADER_LEN && c != header_byte(pos)) {
+      // A later byte of the broken header, or c, may start the real one
+      this->buf_[pos] = c;
+      this->replay_(pos);
+      return nullptr;
+    }
+    this->buf_[pos++] = c;
+    if (pos < N) {
+      this->pos_ = pos;
+      return nullptr;
+    }
+    this->pos_ = 0;
+    return this->buf_;
+  }
+
+  /// Number of bytes of the partial frame; 0 after a frame is returned and while looking for a header.
+  size_t length() const { return this->pos_; }
+
+  /// Forget the partial frame and look for a header from the next byte.
+  void reset() { this->pos_ = 0; }
+
+  /// For a frame the caller rejected (bad checksum): drop its first byte and realign on a header in the rest.
+  /// Call only right after feed() returned that frame; use reset() otherwise.
+  void resync() { this->replay_(N - 1); }
+
+ protected:
+  // A fold, not an array, so no header table lands in RAM on ESP8266
+  static constexpr uint8_t header_byte(size_t i) {
+    size_t j = 0;
+    uint8_t r = 0;
+    ((r = (j++ == i) ? HEADER : r), ...);
+    return r;
+  }
+
+  // Start over and feed buf_[1..last] again; never completes a frame as last < N
+  void replay_(size_t last) {
+    this->pos_ = 0;
+    for (size_t i = 1; i <= last; i++) {
+      this->feed(this->buf_[i]);
+    }
+  }
+
+  uint8_t buf_[N];
+  pos_t pos_{0};
+};
+
+}  // namespace esphome

esphome/git.py

@@ -145,17 +145,13 @@ class GitRepositoryError(GitException):
     """Exception raised when a git repository is in an invalid state."""
 
 
-# The userinfo part of a URL, which can carry credentials.
-URL_USERINFO_RE = re.compile(r"://([^/@\s]+)@")
-
-
 def _redact_url_credentials(text: str) -> str:
     """Mask userinfo in any URLs embedded in ``text``.
 
     Users can put credentials directly in a git URL, and log output is
     routinely pasted into public issues.
     """
-    return URL_USERINFO_RE.sub("://***@", text)
+    return re.sub(r"://[^/@\s]+@", "://***@", text)
 
 
 def _is_transient_git_error(stderr: str) -> bool:

esphome/yaml_util.py

@@ -54,7 +54,7 @@ _LOADED_SECRET_NAMES: set[str] = set()
 # Set while a secret_values_registered context is active; the dumper records
 # every emitted `!secret` name into it.
 _EMITTED_SECRET_NAMES: set[str] | None = None
-# Called with every file the YAML loader opens.
+# Called with every file the YAML loader opens; ``track_yaml_loads`` builds on it.
 # Used by device-builder (esphome/device-builder), whose listener filters by
 # thread because the hook is process-global; keep the name and the per-file
 # ``Path`` signature. Not otherwise a supported API: use at your own risk.
@@ -118,6 +118,25 @@ class SensitiveHexInt(core.HexInt, SensitiveInt):
     """Keep hexadecimal formatting and code generation for sensitive integers."""
 
 
+@contextmanager
+def track_yaml_loads() -> Generator[list[Path]]:
+    """Context manager that records every file loaded by the YAML loader.
+
+    Yields a list that is populated with resolved Path objects for every
+    file loaded through ``_load_yaml_internal`` while the context is active.
+    """
+    loaded: list[Path] = []
+
+    def _on_load(fname: Path) -> None:
+        loaded.append(Path(fname).resolve())
+
+    load_listeners.append(_on_load)
+    try:
+        yield loaded
+    finally:
+        load_listeners.remove(_on_load)
+
+
 class ESPHomeDataBase:
     @property
     def esp_range(self):
@@ -498,7 +517,7 @@ def force_load_include_files(
     Nested ``!include`` returns a deferred ``IncludeFile`` that is only resolved
     later (substitution / packages pass). Callers that need every referenced
     file to actually load — bundle discovery, on-device YAML recovery — invoke
-    this while a ``load_listeners`` callback is active so the underlying
+    this while a :func:`track_yaml_loads` listener is active so the underlying
     loader fires and records every reachable file.
 
     ``IncludeFile`` instances whose path contains unresolved substitution
@@ -520,17 +539,6 @@ def force_load_include_files(
     return result
 
 
-def include_enabled(include: IncludeFile) -> bool:
-    """False only for a statically false condition; an unresolved or malformed one
-    still loads."""
-    from voluptuous import Invalid
-
-    try:
-        return include.should_load()
-    except Invalid:
-        return True
-
-
 def _force_load_include_files(
     obj: Any,
     *,
@@ -560,8 +568,6 @@ def _force_load_include_files(
         if id(obj) in _seen:
             return
         _seen.add(id(obj))
-        if not include_enabled(obj):
-            return
         if obj.has_unresolved_file():
             _load_include_candidates(
                 obj,
@@ -632,27 +638,13 @@ class DiscoveredYamlFiles:
     *un-resolved* filename matched :data:`esphome.const.SECRETS_FILES` (so
     a ``secrets.yaml`` symlinked to a differently-named target is still
     flagged as secrets). ``files`` is incomplete when ``unresolved`` (templated
-    include paths) or ``load_errors`` is non-empty. ``aliases`` maps a resolved
-    path to every :func:`named_path` the YAML reached it by; ``links`` maps each
-    directory symlink a path went through to its target. Neither covers files a
-    templated include matched, nor links on a path through ``..``.
+    include paths) or ``load_errors`` 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)
-    aliases: dict[Path, list[Path]] = field(default_factory=dict)
-    links: dict[Path, Path] = field(default_factory=dict)
-
-
-def named_path(path: Path) -> Path:
-    """``path`` with its directory resolved but its own name kept: a symlinked file
-    keeps its name, and ``..`` after a symlinked directory goes where the OS does."""
-    path = Path(path)
-    if os.name == "nt":  # Windows collapses `..` before it follows a directory link
-        path = Path(os.path.normpath(path))
-    return path.parent.resolve() / path.name
 
 
 def discover_user_yaml_files(config_path: Path) -> DiscoveredYamlFiles:
@@ -671,56 +663,35 @@ def discover_user_yaml_files(config_path: Path) -> DiscoveredYamlFiles:
     """
     from esphome.const import SECRETS_FILES
 
-    root = Path(config_path).absolute().parent
     secrets: set[Path] = set()
-    # Insertion ordered, so its keys are the loaded files in first-seen order.
-    aliases: dict[Path, list[Path]] = {}
-    links: dict[Path, Path] = {}
 
-    def _on_load(fname: Path) -> None:
-        fname = Path(fname).absolute()
-        resolved = fname.resolve()
-        if fname.name in SECRETS_FILES:
-            secrets.add(resolved)
-        names = aliases.setdefault(resolved, [])
-        if (alias := named_path(fname)) not in names:
-            names.append(alias)
-        # The outermost directory symlink below the config directory; `..` after
-        # one is the OS's call.
-        if ".." not in fname.parts:
-            for directory in reversed((fname.parent, *fname.parent.parents)):
-                if (
-                    directory.is_relative_to(root)
-                    and directory != root
-                    and directory.is_symlink()
-                ):
-                    links[directory] = directory.resolve()
-                    break
+    def _capture_secret(fname: Path) -> None:
+        if Path(fname).name in SECRETS_FILES:
+            secrets.add(Path(fname).resolve())
 
-    unresolved: list[str] = []
-    load_listeners.append(_on_load)
-    try:
+    with track_yaml_loads() as loaded:
+        load_listeners.append(_capture_secret)
         try:
-            data = load_yaml(config_path)
-        except EsphomeError as err:
-            load_errors = [f"{config_path}: {err}"]
-        else:
+            try:
+                data = load_yaml(config_path)
+            except EsphomeError as err:
+                return DiscoveredYamlFiles(
+                    list(loaded), secrets, load_errors=[f"{config_path}: {err}"]
+                )
             unresolved, load_errors = force_load_include_files(
                 data, warn_on_unresolved=False, warn_on_load_error=False
             )
-    finally:
-        load_listeners.remove(_on_load)
+        finally:
+            load_listeners.remove(_capture_secret)
 
-    # A `!secret` probe beside the including file is reported even when absent.
-    files = [path for path in aliases if path.is_file()]
-    return DiscoveredYamlFiles(
-        files,
-        secrets & set(files),
-        unresolved,
-        load_errors,
-        {path: aliases[path] for path in files},
-        links,
-    )
+    # Deduplicate while preserving first-seen order.
+    seen: set[Path] = set()
+    unique: list[Path] = []
+    for path in loaded:
+        if path not in seen:
+            seen.add(path)
+            unique.append(path)
+    return DiscoveredYamlFiles(unique, secrets, unresolved, load_errors)
 
 
 def _add_data_ref(fn):
@@ -1179,13 +1150,8 @@ def parse_yaml(
     yaml_loader=None,
     *,
     track_document_range: bool = True,
-    resolve_document_include: bool = True,
 ) -> Any:
-    """Parse a YAML file.
-
-    resolve_document_include=False keeps a document that is one ``!include``
-    as the deferred IncludeFile, with its ``vars`` and ``condition``.
-    """
+    """Parse a YAML file."""
     if yaml_loader is None:
         # Nested loads (!include, !secret, !include_dir_*) inherit the
         # same tracking mode.
@@ -1215,7 +1181,7 @@ def parse_yaml(
             track_document_range=track_document_range,
         )
     # Resolved after the fallback so a failed first parse never loads the include twice
-    return _resolve_include(res) if resolve_document_include else res
+    return _resolve_include(res)
 
 
 def _load_yaml_internal_with_type(
@@ -1268,11 +1234,6 @@ def registered_secret_names() -> set[str]:
     return set(_LOADED_SECRET_NAMES)
 
 
-def registered_secret_values() -> dict[str, str]:
-    """Each registered secret value mapped to its ``!secret`` name."""
-    return dict(_SECRET_VALUES)
-
-
 @contextmanager
 def secret_values_registered(values: dict[str, str]) -> Generator[set[str]]:
     """Make :func:`dump` write these values as ``!secret <name>`` while active.
@@ -1602,23 +1563,6 @@ class ESPHomeDumper(yaml.SafeDumper):
             return self.represent_secret(value.value)
         return self.represent_scalar(tag="!lambda", value=value.value, style="|")
 
-    def represent_data(self, data):
-        # str, dict and list precede the ESPLiteralValue mixin in the MRO.
-        if isinstance(data, ESPLiteralValue):
-            # Not aliased, so the enclosing container's alias entry stays its own
-            self.alias_key = None
-            return self.represent_literal(data)
-        return super().represent_data(data)
-
-    def represent_literal(self, value: ESPLiteralValue) -> yaml.Node:
-        if isinstance(value, dict):
-            return self.represent_mapping("!literal", value)
-        if isinstance(value, list):
-            return self.represent_sequence("!literal", value)
-        if is_secret(value):
-            return self._represent_tagged_secret("!literal", value)
-        return self.represent_scalar(tag="!literal", value=str(value))
-
     def _represent_tagged_secret(self, tag: str, value: Any) -> yaml.ScalarNode:
         # Writing the secret beats leaking the value, but the tag is lost, so
         # the dumped config no longer means the same thing when loaded back.

tests/components/core/frame_reader_test.cpp

@@ -0,0 +1,143 @@
+#include <gtest/gtest.h>
+#include <vector>
+#include "esphome/core/frame_reader.h"
+
+namespace esphome::core::testing {
+
+using Bytes = std::vector<uint8_t>;
+using Frames = std::vector<Bytes>;
+
+// Feeds bytes and collects every completed frame.
+template<size_t N, uint8_t... H> static Frames feed_all(FrameReader<N, H...> &reader, const Bytes &input) {
+  Frames frames;
+  for (uint8_t c : input) {
+    const uint8_t *frame = reader.feed(c);
+    if (frame != nullptr) {
+      EXPECT_EQ(reader.length(), 0u);
+      frames.emplace_back(frame, frame + N);
+    }
+  }
+  return frames;
+}
+
+static_assert(sizeof(FrameReader<4, 0xFF>) == 5);
+static_assert(sizeof(FrameReader<9, 0x55, 0x01, 0x01, 0x04>) == 10);
+static_assert(sizeof(FrameReader<0xFF, 0xAA>) == 0x100);
+static_assert(sizeof(FrameReader<0x100, 0xAA>) == 0x102);
+
+TEST(FrameReaderTest, SingleByteHeader) {
+  FrameReader<4, 0xFF> reader;
+  EXPECT_EQ(feed_all(reader, {0xFF, 0x01, 0x02, 0x02, 0xFF, 0x03, 0x04, 0x06}),
+            (Frames{{0xFF, 0x01, 0x02, 0x02}, {0xFF, 0x03, 0x04, 0x06}}));
+}
+
+TEST(FrameReaderTest, SkipsBytesBeforeHeader) {
+  FrameReader<3, 0xFF> reader;
+  EXPECT_EQ(feed_all(reader, {0x01, 0x02, 0xFF, 0x10, 0x20}), (Frames{{0xFF, 0x10, 0x20}}));
+}
+
+TEST(FrameReaderTest, HeaderByteInsidePayloadIsData) {
+  FrameReader<4, 0xFF> reader;
+  EXPECT_EQ(feed_all(reader, {0xFF, 0xFF, 0xFF, 0x00}), (Frames{{0xFF, 0xFF, 0xFF, 0x00}}));
+}
+
+TEST(FrameReaderTest, LengthTracksPartialFrame) {
+  FrameReader<4, 0xAA, 0xBB> reader;
+  EXPECT_EQ(reader.length(), 0u);
+  reader.feed(0x00);
+  EXPECT_EQ(reader.length(), 0u);
+  reader.feed(0xAA);
+  EXPECT_EQ(reader.length(), 1u);
+  reader.feed(0xBB);
+  reader.feed(0x01);
+  EXPECT_EQ(reader.length(), 3u);
+  reader.reset();
+  EXPECT_EQ(reader.length(), 0u);
+  EXPECT_EQ(feed_all(reader, {0x02, 0xAA, 0xBB, 0x03, 0x04}), (Frames{{0xAA, 0xBB, 0x03, 0x04}}));
+}
+
+TEST(FrameReaderTest, MultiByteHeaderMismatchRestarts) {
+  FrameReader<6, 0x55, 0x01, 0x01, 0x04> reader;
+  EXPECT_EQ(feed_all(reader, {0x55, 0x01, 0x02, 0x55, 0x01, 0x01, 0x04, 0x11, 0x22}),
+            (Frames{{0x55, 0x01, 0x01, 0x04, 0x11, 0x22}}));
+}
+
+TEST(FrameReaderTest, MismatchedByteCanStartHeader) {
+  FrameReader<6, 0x55, 0x01, 0x01, 0x04> reader;
+  // The second 0x55 breaks the first header and starts the real one
+  EXPECT_EQ(feed_all(reader, {0x55, 0x01, 0x55, 0x01, 0x01, 0x04, 0x11, 0x22}),
+            (Frames{{0x55, 0x01, 0x01, 0x04, 0x11, 0x22}}));
+}
+
+TEST(FrameReaderTest, RepeatedHeaderPrefixIsKept) {
+  // AA AA AA BB: the mismatch at the third byte still leaves AA AA matched
+  FrameReader<4, 0xAA, 0xAA, 0xBB> reader;
+  EXPECT_EQ(feed_all(reader, {0xAA, 0xAA, 0xAA, 0xBB, 0x01}), (Frames{{0xAA, 0xAA, 0xBB, 0x01}}));
+}
+
+TEST(FrameReaderTest, OverlappingHeaderPrefix) {
+  // A1 B2 A1 B2: the mismatch on the last B2 keeps the trailing A1 B2 as the start of the header
+  FrameReader<5, 0xA1, 0xB2, 0xA1, 0xC3> reader;
+  EXPECT_EQ(feed_all(reader, {0xA1, 0xB2, 0xA1, 0xB2, 0xA1, 0xC3, 0x07}), (Frames{{0xA1, 0xB2, 0xA1, 0xC3, 0x07}}));
+}
+
+TEST(FrameReaderTest, ResyncFindsHeaderInRejectedFrame) {
+  FrameReader<4, 0xFF> reader;
+  // Started mid-frame: the 0xFF taken as a header was payload; the real frame starts at the second 0xFF
+  const uint8_t *frame = nullptr;
+  for (uint8_t c : {0xFF, 0x12, 0xFF, 0x01}) {
+    frame = reader.feed(c);
+  }
+  ASSERT_NE(frame, nullptr);
+  reader.resync();
+  EXPECT_EQ(reader.length(), 2u);
+  EXPECT_EQ(feed_all(reader, {0x02, 0x02}), (Frames{{0xFF, 0x01, 0x02, 0x02}}));
+}
+
+TEST(FrameReaderTest, ResyncWithoutHeaderStartsOver) {
+  FrameReader<4, 0xFF> reader;
+  const uint8_t *frame = nullptr;
+  for (uint8_t c : {0xFF, 0x01, 0x02, 0x03}) {
+    frame = reader.feed(c);
+  }
+  ASSERT_NE(frame, nullptr);
+  reader.resync();
+  EXPECT_EQ(reader.length(), 0u);
+}
+
+TEST(FrameReaderTest, ResyncKeepsPartialHeaderAtEnd) {
+  FrameReader<5, 0x3C, 0x32> reader;
+  const uint8_t *frame = nullptr;
+  for (uint8_t c : {0x3C, 0x32, 0x00, 0x00, 0x3C}) {
+    frame = reader.feed(c);
+  }
+  ASSERT_NE(frame, nullptr);
+  reader.resync();
+  EXPECT_EQ(reader.length(), 1u);
+  EXPECT_EQ(feed_all(reader, {0x32, 0x01, 0x02, 0x03}), (Frames{{0x3C, 0x32, 0x01, 0x02, 0x03}}));
+}
+
+TEST(FrameReaderTest, ResyncMultiByteHeader) {
+  FrameReader<6, 0x3C, 0x32> reader;
+  const uint8_t *frame = nullptr;
+  for (uint8_t c : {0x3C, 0x32, 0x3C, 0x3C, 0x32, 0x09}) {
+    frame = reader.feed(c);
+  }
+  ASSERT_NE(frame, nullptr);
+  reader.resync();
+  EXPECT_EQ(reader.length(), 3u);
+  EXPECT_EQ(feed_all(reader, {0x0A, 0x0B, 0x0C}), (Frames{{0x3C, 0x32, 0x09, 0x0A, 0x0B, 0x0C}}));
+}
+
+TEST(FrameReaderTest, SixteenBitIndex) {
+  FrameReader<300, 0xAA, 0x55> reader;
+  Bytes input{0x00, 0xAA, 0x55};
+  input.resize(input.size() + 298, 0x42);
+  Frames frames = feed_all(reader, input);
+  ASSERT_EQ(frames.size(), 1u);
+  EXPECT_EQ(frames[0][0], 0xAA);
+  EXPECT_EQ(frames[0][1], 0x55);
+  EXPECT_EQ(frames[0][299], 0x42);
+}
+
+}  // namespace esphome::core::testing

tests/unit_tests/test_bundle.py

@@ -1393,7 +1393,6 @@ def test_force_load_skips_duplicate_include_file() -> None:
         def __init__(self) -> None:
             self.file = Path("dup.yaml")
             self.parent_file = Path("root.yaml")
-            self.should_load = lambda: True
             self.load_calls = 0
 
         def has_unresolved_file(self) -> bool:

tests/unit_tests/test_config_validation.py

@@ -70,11 +70,9 @@ from esphome.schema_extractors import SCHEMA_EXTRACT
 from esphome.util import Registry
 from esphome.yaml_util import (
     ESPHomeDataBase,
-    Sensitive,
     SensitiveHexInt,
     SensitiveInt,
     SensitiveStr,
-    dump,
     load_yaml,
     make_data_base,
     parse_yaml,
@@ -206,17 +204,6 @@ def test_sensitive__wraps_string_result_in_sensitive_str() -> None:
     assert result == "hunter2"
 
 
-def test_sensitive__marks_a_lambda_without_changing_its_value() -> None:
-    """A lambda credential is marked so store_yaml can find it; dumps still see a lambda."""
-    result = cv.sensitive(cv.templatable(cv.string))(Lambda("return id(pw).state;"))
-
-    assert isinstance(result, Lambda)
-    assert isinstance(result, Sensitive)
-    assert (
-        dump({"password": result}) == "password: !lambda |-\n  return id(pw).state;\n"
-    )
-
-
 @pytest.mark.parametrize(
     "pre_tagged", [SensitiveStr("hunter2"), SensitiveInt(5), SensitiveHexInt(0x12)]
 )

tests/unit_tests/test_yaml_util.py

@@ -1,11 +1,10 @@
 from collections.abc import Callable, Generator
 import importlib.util
 import io
-import os
 from pathlib import Path
 import shutil
 import sys
-from types import ModuleType, SimpleNamespace
+from types import ModuleType
 from unittest.mock import patch
 
 import pytest
@@ -456,6 +455,70 @@ def test_dump_sort_keys() -> None:
     assert sorted_dump.index("a_key:") < sorted_dump.index("z_key:")
 
 
+# ---------------------------------------------------------------------------
+# track_yaml_loads
+# ---------------------------------------------------------------------------
+
+
+def test_track_yaml_loads_records_files(
+    tmp_path: Path, yaml_util_module: ModuleType
+) -> None:
+    """yaml_util_module.track_yaml_loads records every file loaded inside the context."""
+    yaml_file = tmp_path / "test.yaml"
+    yaml_file.write_text("key: value\n")
+
+    with yaml_util_module.track_yaml_loads() as loaded:
+        yaml_util_module.load_yaml(yaml_file)
+
+    assert len(loaded) == 1
+    assert loaded[0] == yaml_file.resolve()
+
+
+def test_track_yaml_loads_records_includes(
+    tmp_path: Path, yaml_util_module: ModuleType
+) -> None:
+    """yaml_util_module.track_yaml_loads records nested !include files."""
+    inc = tmp_path / "included.yaml"
+    inc.write_text("included_key: 42\n")
+    main = tmp_path / "main.yaml"
+    main.write_text("child: !include included.yaml\n")
+
+    with yaml_util_module.track_yaml_loads() as loaded:
+        result = yaml_util_module.load_yaml(main)
+        # !include is deferred; resolve it to trigger the nested load
+        result["child"].load()
+
+    resolved = [p.name for p in loaded]
+    assert "main.yaml" in resolved
+    assert "included.yaml" in resolved
+
+
+def test_track_yaml_loads_empty_outside_context(
+    tmp_path: Path, yaml_util_module: ModuleType
+) -> None:
+    """Files loaded outside the context are not recorded."""
+    yaml_file = tmp_path / "test.yaml"
+    yaml_file.write_text("key: value\n")
+
+    with yaml_util_module.track_yaml_loads() as loaded:
+        pass  # load nothing inside
+
+    yaml_util_module.load_yaml(yaml_file)
+    assert loaded == []
+
+
+def test_track_yaml_loads_cleanup_on_exception(
+    tmp_path: Path, yaml_util_module: ModuleType
+) -> None:
+    """Listener is removed even if the body raises."""
+    before = len(yaml_util_module.load_listeners)
+
+    with pytest.raises(RuntimeError), yaml_util_module.track_yaml_loads():
+        raise RuntimeError("boom")
+
+    assert len(yaml_util_module.load_listeners) == before
+
+
 def test_load_listener_removing_itself_does_not_skip_the_next(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch, yaml_util_module: ModuleType
 ) -> None:
@@ -476,20 +539,15 @@ def test_load_listener_removing_itself_does_not_skip_the_next(
     assert listeners == [seen.append]
 
 
-def test_load_listeners_no_duplicate_load_on_top_level_include_failure(
+def test_track_yaml_loads_no_duplicate_load_on_top_level_include_failure(
     tmp_path: Path, yaml_util_module: ModuleType
 ) -> None:
-    """A failed top-level !include must not report any file to load_listeners twice."""
+    """A failed top-level !include must not record any file twice in yaml_util_module.track_yaml_loads."""
     main = tmp_path / "main.yaml"
     main.write_text("!include missing.yaml\n")
 
-    loaded: list[Path] = []
-    yaml_util_module.load_listeners.append(loaded.append)
-    try:
-        with pytest.raises(EsphomeError):
-            yaml_util_module.load_yaml(main)
-    finally:
-        yaml_util_module.load_listeners.remove(loaded.append)
+    with yaml_util_module.track_yaml_loads() as loaded, pytest.raises(EsphomeError):
+        yaml_util_module.load_yaml(main)
 
     assert len(loaded) == len(set(loaded)), (
         f"Files loaded more than once during a failed top-level include: {loaded}"
@@ -680,25 +738,6 @@ def test_represent_include_file_with_condition() -> None:
     )
 
 
-@pytest.mark.parametrize(
-    ("value", "secrets", "expected"),
-    [
-        ("${x}", {}, "key: !literal '${x}'\n"),
-        ({"a": "${x}"}, {}, "key: !literal\n  a: ${x}\n"),
-        (["${x}"], {}, "key: !literal\n  - ${x}\n"),
-        # A literal equal to a registered secret is written as the secret.
-        ("hunter2", {"hunter2": "pw"}, "key: !secret 'pw'\n"),
-    ],
-    ids=["scalar", "mapping", "sequence", "secret"],
-)
-def test_represent_literal(
-    value: object, secrets: dict[str, str], expected: str
-) -> None:
-    """`!literal` values keep their tag when dumped."""
-    yaml_util._SECRET_VALUES.update(secrets)
-    assert yaml_util.dump({"key": make_literal(value)}) == expected
-
-
 def test_represent_include_file_with_data_base_mixin() -> None:
     """Test that IncludeFile wrapped with ESPHomeDataBase mixin is also dumped correctly.
 
@@ -1308,7 +1347,6 @@ class _StubInclude:
         # stubs never glob real files during candidate expansion.
         self.file = file
         self.parent_file = parent_file or Path("/nonexistent/parent.yaml")
-        self.should_load = lambda: True
         self._unresolved = unresolved
         self._load_result = load_result if load_result is not None else {}
         self._raise = raise_on_load
@@ -1771,77 +1809,17 @@ def test_discover_user_yaml_files_tolerates_templated_top_level_include(
     assert (tmp_path / "wrapper.yaml").resolve() in discovered.files
 
 
-def test_discover_user_yaml_files_records_symlink_alias(tmp_path: Path) -> None:
-    """A file reached through a symlink is listed by its target and aliased to the link."""
-    target = _write(tmp_path, "shared/network.yaml", "ssid: x\n")
-    link = tmp_path / "wifi.yaml"
+def test_track_yaml_loads_records_resolved_paths(
+    tmp_path: Path, yaml_util_module: ModuleType
+) -> None:
+    """`track_yaml_loads` is the building block - sanity-check it resolves
+    symlinks so callers can dedupe by identity."""
+    target = _write(tmp_path, "actual.yaml", "esphome:\n  name: t\n")
+    link = tmp_path / "alias.yaml"
     link.symlink_to(target)
-    discovered = discover_user_yaml_files(_write_entry_including(tmp_path, "wifi.yaml"))
-    assert target.resolve() in discovered.files
-    assert discovered.aliases[target.resolve()] == [yaml_util.named_path(link)]
-
-
-def test_parse_yaml_can_keep_a_document_include(tmp_path: Path) -> None:
-    """resolve_document_include=False returns the deferred include with its vars."""
-    _write(tmp_path, "base.yaml", "esphome:\n  name: ${name}\n")
-    entry = _write(
-        tmp_path, "entry.yaml", "!include\n  file: base.yaml\n  vars:\n    name: k\n"
-    )
-    with entry.open(encoding="utf-8") as handle:
-        loaded = yaml_util.parse_yaml(entry, handle, resolve_document_include=False)
-    assert isinstance(loaded, yaml_util.IncludeFile)
-    assert loaded.file == "base.yaml"
-    assert loaded.vars == {"name": "k"}
-    assert yaml_util.load_yaml(entry) == {"esphome": {"name": "${name}"}}
-
-
-@pytest.mark.parametrize("condition", ["false", "true"])
-def test_discover_user_yaml_files_static_include_condition(
-    tmp_path: Path, condition: str
-) -> None:
-    """A missing file behind `condition: false` is not an error; behind `true` it is."""
-    entry = _write(
-        tmp_path,
-        "entry.yaml",
-        f"packages:\n  opt: !include {{file: missing.yaml, condition: {condition}}}\n",
-    )
-    discovered = discover_user_yaml_files(entry)
-    assert bool(discovered.load_errors) == (condition == "true")
-
-
-def test_discover_user_yaml_files_skips_a_disabled_templated_include(
-    tmp_path: Path,
-) -> None:
-    """A statically disabled include is skipped before its templated name is expanded."""
-    _write(tmp_path, "optional-x.yaml", "broken: [\n")
-    entry = _write(
-        tmp_path,
-        "entry.yaml",
-        "packages:\n"
-        '  opt: !include {file: "optional-${board}.yaml", condition: false}\n',
-    )
-    discovered = discover_user_yaml_files(entry)
-    assert not discovered.load_errors
-    assert all(p.name != "optional-x.yaml" for p in discovered.files)
-
-
-def test_discover_user_yaml_files_skips_missing_secrets_probe(
-    tmp_path: Path,
-) -> None:
-    """A `!secret` in a package first probes a secrets.yaml next to that file;
-    when only the entry directory has one, the missing probe is not reported."""
-    _write(tmp_path, "pkg/common.yaml", "wifi:\n  password: !secret pw\n")
-    secrets = _write(tmp_path, "cfg/secrets.yaml", "pw: value\n")
-    entry = _write(
-        tmp_path,
-        "cfg/entry.yaml",
-        "esphome:\n  name: test\npackages:\n  common: !include ../pkg/common.yaml\n",
-    )
-    with patch.object(core.CORE, "config_path", entry):
-        discovered = discover_user_yaml_files(entry)
-    assert discovered.load_errors == []
-    assert discovered.secrets == {secrets.resolve()}
-    assert all(p.is_file() for p in discovered.files)
+    with yaml_util_module.track_yaml_loads() as loaded:
+        yaml_util_module.load_yaml(link)
+    assert target.resolve() in loaded
 
 
 # ---------------------------------------------------------------------------
@@ -2750,71 +2728,3 @@ def test_dump_secret_valued_float_lambda_and_id() -> None:
     assert "!secret 'float_secret'" in dumped
     assert "!secret 'lambda_secret'" in dumped
     assert "!secret 'id_secret'" in dumped
-
-
-def test_named_path_collapses_dotdot_lexically_on_windows(
-    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
-) -> None:
-    """Windows resolves `..` before following a directory link, so named_path does too."""
-    (tmp_path / "real" / "sub").mkdir(parents=True)
-    # Nested, so following the link (real) and collapsing `..` (tmp_path) differ
-    (tmp_path / "link").symlink_to(tmp_path / "real" / "sub")
-    path = tmp_path / "link" / ".." / "x.yaml"
-    if os.name != "nt":  # POSIX follows the link first
-        assert yaml_util.named_path(path) == tmp_path.resolve() / "real" / "x.yaml"
-    # Only yaml_util sees Windows; pathlib keeps resolving with the real os
-    monkeypatch.setattr(yaml_util, "os", SimpleNamespace(name="nt", path=os.path))
-    assert yaml_util.named_path(path) == tmp_path.resolve() / "x.yaml"
-
-
-def test_discover_user_yaml_files_loads_an_include_with_a_templated_condition(
-    tmp_path: Path,
-) -> None:
-    """A condition that is not static yet is loaded anyway, so its errors still show."""
-    entry = _write(
-        tmp_path,
-        "entry.yaml",
-        'packages:\n  opt: !include {file: missing.yaml, condition: "${enable}"}\n',
-    )
-    assert discover_user_yaml_files(entry).load_errors
-
-
-def test_discover_user_yaml_files_records_directory_links(tmp_path: Path) -> None:
-    """A file reached through a symlinked directory below the config directory
-    records that link and its target."""
-    shared = tmp_path / "shared"
-    _write(shared, "wifi.yaml", "ssid: x\n")
-    config_dir = tmp_path / "config"
-    entry = _write(config_dir, "entry.yaml", "wifi: !include linked/wifi.yaml\n")
-    (config_dir / "linked").symlink_to(shared)
-    discovered = discover_user_yaml_files(entry)
-    assert discovered.links == {config_dir.resolve() / "linked": shared.resolve()}
-
-
-def test_registered_secret_values_maps_a_loaded_secret_to_its_name(
-    tmp_path: Path,
-) -> None:
-    """Values loaded through `!secret` are listed with the name they came from."""
-    _write(tmp_path, "secrets.yaml", "wifi_password: hunter2\n")
-    entry = _write(tmp_path, "entry.yaml", "wifi:\n  password: !secret wifi_password\n")
-    yaml_util.load_yaml(entry)
-    assert yaml_util.registered_secret_values()["hunter2"] == "wifi_password"
-
-
-def test_discover_user_yaml_files_lists_only_existing_files_after_a_load_error(
-    tmp_path: Path,
-) -> None:
-    """A failed top-level load still reports only files that exist."""
-    entry = _write(tmp_path, "entry.yaml", "!include missing.yaml\n")
-    discovered = discover_user_yaml_files(entry)
-    assert discovered.load_errors
-    assert discovered.files == [entry.resolve()]
-    assert list(discovered.aliases) == [entry.resolve()]
-
-
-def test_dump_literal_inside_a_shared_container_keeps_the_container() -> None:
-    """A literal must not take over the alias of the list that holds it."""
-    shared = [yaml_util.make_literal({"x": "y"})]
-    out = yaml_util.dump({"a": shared, "b": shared})
-    loaded = yaml.load(out.replace("!literal", ""), Loader=yaml.SafeLoader)
-    assert loaded == {"a": [{"x": "y"}], "b": [{"x": "y"}]}