Merge pull request #16282 from esphome/bump-2026.4.5

2026.4.5

Doxyfile

@@ -48,7 +48,7 @@ PROJECT_NAME           = ESPHome
 # could be handy for archiving the generated documentation or if some version
 # control system is used.
 
-PROJECT_NUMBER         = 2026.4.4
+PROJECT_NUMBER         = 2026.4.5
 
 # Using the PROJECT_BRIEF tag one can provide an optional one line description
 # for a project that appears at the top of each page and should give viewer a

docker/ha-addon-rootfs/etc/cont-init.d/40-device-builder.sh

@@ -0,0 +1,22 @@
+#!/usr/bin/with-contenv bashio
+# ==============================================================================
+# Installs the latest prerelease of esphome-device-builder when the
+# `use_new_device_builder` config option is enabled.
+# This is a temporary install-on-boot step until esphome-device-builder
+# becomes a direct dependency of esphome.
+# ==============================================================================
+
+if ! bashio::config.true 'use_new_device_builder'; then
+  exit 0
+fi
+
+bashio::log.info "Installing latest prerelease of esphome-device-builder..."
+if command -v uv > /dev/null; then
+  uv pip install --system --no-cache-dir --prerelease=allow --upgrade \
+    esphome-device-builder ||
+    bashio::exit.nok "Failed installing esphome-device-builder."
+else
+  pip install --no-cache-dir --pre --upgrade esphome-device-builder ||
+    bashio::exit.nok "Failed installing esphome-device-builder."
+fi
+bashio::log.info "Installed esphome-device-builder."

docker/ha-addon-rootfs/etc/s6-overlay/s6-rc.d/esphome/run

@@ -49,5 +49,12 @@ if bashio::fs.directory_exists '/config/esphome/.esphome'; then
     rm -rf /config/esphome/.esphome
 fi
 
+if bashio::config.true 'use_new_device_builder'; then
+    bashio::log.info "Starting ESPHome Device Builder..."
+    exec esphome-device-builder /config/esphome \
+        --ha-addon \
+        --ingress-port "$(bashio::addon.ingress_port)"
+fi
+
 bashio::log.info "Starting ESPHome dashboard..."
 exec esphome dashboard /config/esphome --socket /var/run/esphome.sock --ha-addon

docker/ha-addon-rootfs/etc/s6-overlay/s6-rc.d/init-nginx/run

@@ -4,6 +4,14 @@
 # Community Hass.io Add-ons: ESPHome
 # Configures NGINX for use with ESPHome
 # ==============================================================================
+
+# When the new device builder is enabled it serves HA ingress directly,
+# so nginx is not used at all -- skip configuration.
+if bashio::config.true 'use_new_device_builder'; then
+    bashio::log.info "Skipping NGINX setup: new device builder serves ingress directly."
+    bashio::exit.ok
+fi
+
 mkdir -p /var/log/nginx
 
 # Generate Ingress configuration

docker/ha-addon-rootfs/etc/s6-overlay/s6-rc.d/nginx/run

@@ -5,6 +5,14 @@
 # Runs the NGINX proxy
 # ==============================================================================
 
+# The new device builder handles HA ingress itself, so nginx is bypassed.
+# Block the longrun forever so s6 keeps the dependency satisfied and does
+# not respawn it.
+if bashio::config.true 'use_new_device_builder'; then
+    bashio::log.info "NGINX bypassed: new device builder serves ingress directly."
+    exec sleep infinity
+fi
+
 bashio::log.info "Waiting for ESPHome dashboard to come up..."
 
 while [[ ! -S /var/run/esphome.sock ]]; do

esphome/bundle.py

@@ -98,11 +98,13 @@ _KNOWN_FILE_EXTENSIONS = frozenset(
 )
 
 
-# Matches !secret references in YAML text.  This is intentionally a simple
+# Matches !secret references in YAML text.  An optional surrounding
-# regex scan rather than a YAML parse — it may match inside comments or
+# quote pair around the key is allowed and ignored: YAML treats
-# multi-line strings, which is the conservative direction (include more
+# ``!secret 'foo'`` and ``!secret foo`` as the same key.  This is
-# secrets rather than fewer).
+# intentionally a simple regex scan rather than a YAML parse — it may
-_SECRET_RE = re.compile(r"!secret\s+(\S+)")
+# 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]:

esphome/components/nextion/nextion.cpp

@@ -641,6 +641,7 @@ void Nextion::process_nextion_commands_() {
         } else {
           ESP_LOGN(TAG, "String resp: '%s' id: %s type: %s", to_process.c_str(), component->get_variable_name().c_str(),
                    component->get_queue_type_string());
+          component->set_state_from_string(to_process, true, false);
         }
 
         delete nb;  // NOLINT(cppcoreguidelines-owning-memory)

esphome/components/substitutions/__init__.py

@@ -297,7 +297,18 @@ def _push_context(
         """Resolve a variable, recursively resolving any dependencies it references."""
         value = unresolved_vars.pop(key, Missing)
         if value is Missing:
-            return Missing
+            # Either already resolved (in resolved_vars) or currently being
+            # resolved (self-reference from inside a dict-valued substitution).
+            # Returning what we have lets sibling references inside a dict
+            # value, e.g. ``${device.manufacturer}`` inside ``device.name``,
+            # see literal sibling values during their own resolution.
+            return resolved_vars.get(key, Missing)
+        if isinstance(value, dict):
+            # Dict-valued substitutions form a namespace; eagerly publish the
+            # original mapping so its members can reference each other while
+            # the dict's own substitution pass is still running. The entry is
+            # replaced with the fully-substituted dict once recursion returns.
+            resolved_vars[key] = value
         try:
             value = substitute(value, [], resolver_context, True)
         except UndefinedError as err:

esphome/const.py

@@ -4,7 +4,7 @@ from enum import Enum
 
 from esphome.enum import StrEnum
 
-__version__ = "2026.4.4"
+__version__ = "2026.4.5"
 
 ALLOWED_NAME_CHARS = "abcdefghijklmnopqrstuvwxyz0123456789-_"
 VALID_SUBSTITUTIONS_CHARACTERS = (

esphome/core/config.py

@@ -562,14 +562,9 @@ async def _add_controller_registry_define() -> None:
 
 @coroutine_with_priority(CoroPriority.FINAL)
 async def _add_looping_components() -> None:
-    # Emit a constexpr that computes the looping component count at C++ compile time
+    # Emit ESPHOME_LOOPING_COMPONENT_COUNT. Sizing of looping_components_
-    # and pre-init the FixedVector with the exact capacity. Uses std::is_same_v to
+    # happens in core to_code() so it lands before safe_mode's early return.
-    # detect loop() overrides. The constexpr goes in main.cpp's global section where
-    # all component types are in scope. calculate_looping_components_() then skips
-    # the counting pass and only does the two population passes.
     entries = CORE.data.get("looping_component_entries", [])
-    if not entries:
-        return
 
     # Build constexpr sum for the exact count, deduplicating by type
     # Uses HasLoopOverride<T> which handles ambiguous &T::loop from multiple inheritance
@@ -577,7 +572,7 @@ async def _add_looping_components() -> None:
     terms = [
         f"({count} * HasLoopOverride<{cpp_type}>::value)"
         for cpp_type, count in type_counts.items()
-    ]
+    ] or ["0"]
     constexpr_expr = " + \\\n  ".join(terms)
     cg.add_global(
         cg.RawStatement(
@@ -586,14 +581,6 @@ async def _add_looping_components() -> None:
         )
     )
 
-    # Pre-init FixedVector with exact capacity so calculate_looping_components_()
-    # can skip the counting pass
-    cg.add(
-        cg.RawExpression(
-            "App.looping_components_.init(ESPHOME_LOOPING_COMPONENT_COUNT)"
-        )
-    )
-
 
 @coroutine_with_priority(CoroPriority.CORE)
 async def to_code(config: ConfigType) -> None:
@@ -636,6 +623,14 @@ async def to_code(config: ConfigType) -> None:
     # Define component count for static allocation
     cg.add_define("ESPHOME_COMPONENT_COUNT", len(CORE.component_ids))
 
+    # Pre-init FixedVector with exact capacity so calculate_looping_components_()
+    # can skip the counting pass
+    cg.add(
+        cg.RawExpression(
+            "App.looping_components_.init(ESPHOME_LOOPING_COMPONENT_COUNT)"
+        )
+    )
+
     CORE.add_job(_add_platform_defines)
     CORE.add_job(_add_controller_registry_define)
     CORE.add_job(_add_looping_components)

tests/integration/conftest.py

@@ -501,14 +501,15 @@ async def _read_stream_lines(
 
 
 @asynccontextmanager
-async def run_binary_and_wait_for_port(
+async def run_binary(
     binary_path: Path,
-    host: str,
-    port: int,
-    timeout: float = PORT_WAIT_TIMEOUT,
     line_callback: Callable[[str], None] | None = None,
-) -> AsyncGenerator[None]:
+) -> AsyncGenerator[tuple[asyncio.subprocess.Process, list[str]]]:
-    """Run a binary, wait for it to open a port, and clean up on exit."""
+    """Run a binary under a PTY, capture log output, and clean up on exit.
+
+    Yields the running ``Process`` and a live list of captured log lines.
+    No port wait -- callers that need that should use
+    ``run_binary_and_wait_for_port``."""
     # Create a pseudo-terminal to make the binary think it's running interactively
     # This is needed because the ESPHome host logger checks isatty()
     controller_fd, device_fd = pty.openpty()
@@ -535,7 +536,6 @@ async def run_binary_and_wait_for_port(
     controller_transport, _ = await loop.connect_read_pipe(
         lambda: controller_protocol, os.fdopen(controller_fd, "rb", 0)
     )
-    output_reader = controller_reader
 
     if process.returncode is not None:
         raise RuntimeError(
@@ -543,27 +543,59 @@ async def run_binary_and_wait_for_port(
             "Ensure the binary is valid and can run successfully."
         )
 
-    # Wait for the API server to start listening
-    loop = asyncio.get_running_loop()
-    start_time = loop.time()
-
-    # Start collecting output
     stdout_lines: list[str] = []
-    output_tasks: list[asyncio.Task] = []
+    output_task = asyncio.create_task(
+        _read_stream_lines(controller_reader, stdout_lines, sys.stdout, line_callback)
+    )
 
     try:
-        # Read from output stream
-        output_tasks = [
-            asyncio.create_task(
-                _read_stream_lines(
-                    output_reader, stdout_lines, sys.stdout, line_callback
-                )
-            )
-        ]
-
         # Small yield to ensure the process has a chance to start
         await asyncio.sleep(0)
+        yield process, stdout_lines
+    finally:
+        output_task.cancel()
+        result = await asyncio.gather(output_task, return_exceptions=True)
+        if isinstance(result[0], Exception) and not isinstance(
+            result[0], asyncio.CancelledError
+        ):
+            print(f"Error reading from PTY: {result[0]}", file=sys.stderr)
 
+        # Close the PTY transport (Unix only)
+        if controller_transport is not None:
+            controller_transport.close()
+
+        # Cleanup: terminate the process gracefully
+        if process.returncode is None:
+            # Send SIGINT (Ctrl+C) for graceful shutdown
+            process.send_signal(signal.SIGINT)
+            try:
+                await asyncio.wait_for(process.wait(), timeout=SIGINT_TIMEOUT)
+            except TimeoutError:
+                # If SIGINT didn't work, try SIGTERM
+                process.terminate()
+                try:
+                    await asyncio.wait_for(process.wait(), timeout=SIGTERM_TIMEOUT)
+                except TimeoutError:
+                    # Last resort: SIGKILL
+                    process.kill()
+                    await process.wait()
+
+
+@asynccontextmanager
+async def run_binary_and_wait_for_port(
+    binary_path: Path,
+    host: str,
+    port: int,
+    timeout: float = PORT_WAIT_TIMEOUT,
+    line_callback: Callable[[str], None] | None = None,
+) -> AsyncGenerator[None]:
+    """Run a binary, wait for it to open a port, and clean up on exit."""
+    async with run_binary(binary_path, line_callback=line_callback) as (
+        process,
+        stdout_lines,
+    ):
+        loop = asyncio.get_running_loop()
+        start_time = loop.time()
         while loop.time() - start_time < timeout:
             try:
                 # Try to connect to the port
@@ -593,41 +625,6 @@ async def run_binary_and_wait_for_port(
 
         raise TimeoutError(error_msg)
 
-    finally:
-        # Cancel output collection tasks
-        for task in output_tasks:
-            task.cancel()
-        # Wait for tasks to complete and check for exceptions
-        results = await asyncio.gather(*output_tasks, return_exceptions=True)
-        for i, result in enumerate(results):
-            if isinstance(result, Exception) and not isinstance(
-                result, asyncio.CancelledError
-            ):
-                print(
-                    f"Error reading from PTY: {result}",
-                    file=sys.stderr,
-                )
-
-        # Close the PTY transport (Unix only)
-        if controller_transport is not None:
-            controller_transport.close()
-
-        # Cleanup: terminate the process gracefully
-        if process.returncode is None:
-            # Send SIGINT (Ctrl+C) for graceful shutdown
-            process.send_signal(signal.SIGINT)
-            try:
-                await asyncio.wait_for(process.wait(), timeout=SIGINT_TIMEOUT)
-            except TimeoutError:
-                # If SIGINT didn't work, try SIGTERM
-                process.terminate()
-                try:
-                    await asyncio.wait_for(process.wait(), timeout=SIGTERM_TIMEOUT)
-                except TimeoutError:
-                    # Last resort: SIGKILL
-                    process.kill()
-                    await process.wait()
-
 
 @asynccontextmanager
 async def run_compiled_context(

tests/integration/fixtures/safe_mode_loop_runs.yaml

@@ -0,0 +1,25 @@
+esphome:
+  name: safe-mode-loop-runs
+
+host:
+
+logger:
+
+safe_mode:
+  num_attempts: 10
+  on_safe_mode:
+    - lambda: |-
+        // Spawn a detached thread that logs a unique marker. The
+        // non-main-thread log goes through the task log buffer, which
+        // is only drained by Logger::loop(). If looping components
+        // weren't initialized (the bug fixed in #16269), the buffer is
+        // never read and the marker never reaches the console.
+        struct MarkerThread {
+          static void *thread_func(void *) {
+            ESP_LOGI("safe_mode_test", "looping component ran in safe mode");
+            return nullptr;
+          }
+        };
+        pthread_t t;
+        pthread_create(&t, nullptr, MarkerThread::thread_func, nullptr);
+        pthread_detach(t);

tests/integration/host_prefs.py

@@ -0,0 +1,39 @@
+"""Helpers for manipulating the host platform's preferences file.
+
+ESPHome's host platform stores preferences in
+``~/.esphome/prefs/<app_name>.prefs`` using a simple binary layout that
+mirrors ``HostPreferences::sync()``:
+``[uint32_t key][uint8_t len][uint8_t data[len]]`` per entry.
+
+Tests use these helpers to pre-populate state the binary will see at
+boot (e.g. forcing safe mode) or to clear stale state between runs.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+import struct
+
+
+def host_prefs_path(device_name: str) -> Path:
+    """Return the on-disk prefs file path for a host-platform device."""
+    return Path.home() / ".esphome" / "prefs" / f"{device_name}.prefs"
+
+
+def clear_host_prefs(device_name: str) -> None:
+    """Delete the prefs file for a host-platform device, if it exists."""
+    host_prefs_path(device_name).unlink(missing_ok=True)
+
+
+def write_host_pref(device_name: str, key: int, data: bytes) -> Path:
+    """Write a single preference entry, replacing the file's contents.
+
+    Returns the path that was written.
+    """
+    if len(data) > 255:
+        raise ValueError(f"Preference data too long: {len(data)} bytes (max 255)")
+    path = host_prefs_path(device_name)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    payload = struct.pack("<IB", key, len(data)) + data
+    path.write_bytes(payload)
+    return path

tests/integration/test_safe_mode_loop_runs.py

@@ -0,0 +1,94 @@
+"""Regression test for safe_mode + looping_components init ordering.
+
+Reproduces the bug fixed in https://github.com/esphome/esphome/pull/16269:
+``App.looping_components_.init(...)`` was emitted at ``CoroPriority.FINAL``,
+which placed it *after* the ``safe_mode`` early-return in ``setup_app()``.
+When safe mode was entered, the ``FixedVector`` backing the looping-component
+list was never sized, ``looping_components_active_end_`` stayed at 0, and
+``loop()`` iterated zero components -- so any looping component above
+``CoroPriority.APPLICATION`` (e.g. wifi, logger) never ran.
+
+The test forces safe mode by writing ``ENTER_SAFE_MODE_MAGIC`` to the host
+preferences file before booting, then asserts that ``Logger::loop()`` runs
+by logging from a non-main thread. Non-main-thread logs are buffered in
+``TaskLogBuffer`` and only emitted to the console when ``Logger::loop()``
+drains the buffer. Without the fix, the marker stays in the buffer
+forever; with the fix, it reaches the console.
+
+The API server (``CoroPriority.WEB``, 40) is registered below safe_mode
+(``CoroPriority.APPLICATION``, 50), so it's never set up when safe mode
+is active and ``run_compiled`` would hang waiting for the API port.
+This test uses ``run_binary`` directly to skip the port wait.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import re
+import struct
+
+import pytest
+
+from .conftest import run_binary
+from .host_prefs import clear_host_prefs, write_host_pref
+from .types import CompileFunction, ConfigWriter
+
+# Must match esphome::safe_mode::RTC_KEY in safe_mode.h
+SAFE_MODE_RTC_KEY = 233825507
+# Must match esphome::safe_mode::SafeModeComponent::ENTER_SAFE_MODE_MAGIC
+ENTER_SAFE_MODE_MAGIC = 0x5AFE5AFE
+
+DEVICE_NAME = "safe-mode-loop-runs"
+THREAD_LOG_MARKER = "looping component ran in safe mode"
+
+
+@pytest.mark.asyncio
+async def test_safe_mode_loop_runs(
+    yaml_config: str,
+    write_yaml_config: ConfigWriter,
+    compile_esphome: CompileFunction,
+) -> None:
+    """When safe mode is active, ``App.loop()`` must still iterate looping
+    components -- proven here by a thread-logged marker reaching the
+    console (which requires ``Logger::loop()`` to run)."""
+    config_path = await write_yaml_config(yaml_config)
+    binary_path = await compile_esphome(config_path)
+
+    # Compile finished successfully; pre-populate prefs so the *next* run
+    # enters safe mode immediately.
+    write_host_pref(
+        DEVICE_NAME, SAFE_MODE_RTC_KEY, struct.pack("<I", ENTER_SAFE_MODE_MAGIC)
+    )
+
+    try:
+        loop = asyncio.get_running_loop()
+        safe_mode_active = loop.create_future()
+        thread_log_seen = loop.create_future()
+        safe_mode_pattern = re.compile(r"SAFE MODE IS ACTIVE")
+        thread_log_pattern = re.compile(re.escape(THREAD_LOG_MARKER))
+
+        def on_log(line: str) -> None:
+            if not safe_mode_active.done() and safe_mode_pattern.search(line):
+                safe_mode_active.set_result(True)
+            if not thread_log_seen.done() and thread_log_pattern.search(line):
+                thread_log_seen.set_result(True)
+
+        async with run_binary(binary_path, line_callback=on_log):
+            try:
+                await asyncio.wait_for(safe_mode_active, timeout=15.0)
+            except TimeoutError:
+                pytest.fail(
+                    "Did not observe 'SAFE MODE IS ACTIVE' -- safe mode "
+                    "didn't trigger, so this test isn't exercising the bug."
+                )
+            try:
+                await asyncio.wait_for(thread_log_seen, timeout=10.0)
+            except TimeoutError:
+                pytest.fail(
+                    f"Did not observe thread-logged marker {THREAD_LOG_MARKER!r} "
+                    "within timeout. Logger::loop() never drained the task "
+                    "log buffer, meaning App.looping_components_ was never "
+                    "sized -- this is the regression #16269 fixed."
+                )
+    finally:
+        clear_host_prefs(DEVICE_NAME)

tests/integration/test_template_text_save.py

@@ -9,7 +9,6 @@ Tests that:
 from __future__ import annotations
 
 import asyncio
-from pathlib import Path
 import socket
 from typing import Any
 
@@ -17,9 +16,12 @@ from aioesphomeapi import TextInfo, TextState
 import pytest
 
 from .conftest import run_binary_and_wait_for_port, wait_and_connect_api_client
+from .host_prefs import clear_host_prefs
 from .state_utils import InitialStateHelper, require_entity
 from .types import CompileFunction, ConfigWriter
 
+DEVICE_NAME = "host-template-text-save-test"
+
 
 @pytest.mark.asyncio
 async def test_template_text_save(
@@ -32,11 +34,7 @@ async def test_template_text_save(
     port, port_socket = reserved_tcp_port
 
     # Clean up any stale preference file from previous runs
-    prefs_file = (
+    clear_host_prefs(DEVICE_NAME)
-        Path.home() / ".esphome" / "prefs" / "host-template-text-save-test.prefs"
-    )
-    if prefs_file.exists():
-        prefs_file.unlink()
 
     # Write and compile once
     config_path = await write_yaml_config(yaml_config)
@@ -59,7 +57,7 @@ async def test_template_text_save(
         wait_and_connect_api_client(port=port) as client,
     ):
         device_info = await client.device_info()
-        assert device_info.name == "host-template-text-save-test"
+        assert device_info.name == DEVICE_NAME
 
         entities, _ = await client.list_entities_services()
         text_entity = require_entity(
@@ -127,5 +125,4 @@ async def test_template_text_save(
         )
 
     # Clean up preference file
-    if prefs_file.exists():
+    clear_host_prefs(DEVICE_NAME)
-        prefs_file.unlink()

tests/unit_tests/core/test_config.py

@@ -10,6 +10,7 @@ from unittest.mock import MagicMock, Mock, patch
 import pytest
 
 from esphome import config_validation as cv, core
+from esphome.components.safe_mode import to_code as safe_mode_to_code
 from esphome.const import (
     CONF_AREA,
     CONF_AREAS,
@@ -312,6 +313,75 @@ def test_add_platform_defines_priority() -> None:
     )
 
 
+def test_to_code_priority_above_safe_mode() -> None:
+    """Test that core to_code emits the looping_components_ init before safe_mode.
+
+    Regression test for https://github.com/esphome/esphome/issues/16262.
+    safe_mode emits an `if (should_enter_safe_mode(...)) return;` line in main()
+    at APPLICATION priority. The `App.looping_components_.init(...)` call must be
+    emitted at a higher priority than APPLICATION so it lands in main() before
+    the early return; otherwise the FixedVector is never sized when safe mode is
+    active and loop() never runs (Wi-Fi never connects).
+    """
+    assert config.to_code.priority > safe_mode_to_code.priority, (
+        f"core to_code priority ({config.to_code.priority}) must be greater than "
+        f"safe_mode to_code priority ({safe_mode_to_code.priority}) so that "
+        "App.looping_components_.init() is emitted before safe_mode's early return"
+    )
+
+
+@pytest.mark.asyncio
+async def test_add_looping_components_handles_empty_entries() -> None:
+    """Test that _add_looping_components emits a valid constexpr when there are
+    no looping component entries.
+
+    With zero entries the generated constexpr must still be syntactically valid
+    C++ (`= 0;`), not an empty expression (`= ;`). This guards the empty-list
+    case that would otherwise produce uncompilable main.cpp output.
+    """
+    CORE.data["looping_component_entries"] = []
+
+    await config._add_looping_components()
+
+    constexpr_lines = [
+        str(s)
+        for s in CORE.global_statements
+        if "ESPHOME_LOOPING_COMPONENT_COUNT" in str(s)
+    ]
+    assert len(constexpr_lines) == 1
+    text = constexpr_lines[0]
+    assert "static constexpr size_t ESPHOME_LOOPING_COMPONENT_COUNT" in text
+    # The right-hand side must contain a literal `0`, not be empty.
+    rhs = text.split("=", 1)[1]
+    assert "0" in rhs
+    assert rhs.strip().rstrip(";").strip(), (
+        f"constexpr right-hand side must not be empty, got: {text!r}"
+    )
+
+
+@pytest.mark.asyncio
+async def test_add_looping_components_with_entries() -> None:
+    """Test that _add_looping_components builds a HasLoopOverride sum from entries."""
+    CORE.data["looping_component_entries"] = [
+        "esphome::wifi::WiFiComponent",
+        "esphome::logger::Logger",
+        "esphome::wifi::WiFiComponent",
+    ]
+
+    await config._add_looping_components()
+
+    constexpr_lines = [
+        str(s)
+        for s in CORE.global_statements
+        if "ESPHOME_LOOPING_COMPONENT_COUNT" in str(s)
+    ]
+    assert len(constexpr_lines) == 1
+    text = constexpr_lines[0]
+    # Deduplicated by type, with per-type counts as multiplier.
+    assert "(2 * HasLoopOverride<esphome::wifi::WiFiComponent>::value)" in text
+    assert "(1 * HasLoopOverride<esphome::logger::Logger>::value)" in text
+
+
 def test_valid_include_with_angle_brackets() -> None:
     """Test valid_include accepts angle bracket includes."""
     assert valid_include("<ArduinoJson.h>") == "<ArduinoJson.h>"

tests/unit_tests/fixtures/substitutions/18-dict_self_reference.approved.yaml

@@ -0,0 +1,16 @@
+substitutions:
+  device:
+    manufacturer: espressif
+    model: esp32
+    mac_suffix: ffffff
+    name: espressif-esp32-ffffff
+  network:
+    host: example.com
+    port: 8080
+    url: http://example.com:8080/api
+esphome:
+  name: espressif-esp32-ffffff
+test_list:
+  - espressif-esp32-ffffff
+  - http://example.com:8080/api
+  - espressif/esp32

tests/unit_tests/fixtures/substitutions/18-dict_self_reference.input.yaml

@@ -0,0 +1,18 @@
+substitutions:
+  device:
+    manufacturer: "espressif"
+    model: "esp32"
+    mac_suffix: "ffffff"
+    name: ${device.manufacturer}-${device.model}-${device.mac_suffix}
+  network:
+    host: "example.com"
+    port: 8080
+    url: "http://${network.host}:${network.port}/api"
+
+esphome:
+  name: ${device.name}
+
+test_list:
+  - ${device.name}
+  - ${network.url}
+  - "${device.manufacturer}/${device.model}"

tests/unit_tests/test_bundle.py

@@ -170,6 +170,23 @@ def test_find_used_secret_keys_deduplicates(tmp_path: Path) -> None:
     assert keys == {"key1"}
 
 
+def test_find_used_secret_keys_quoted(tmp_path: Path) -> None:
+    """Quoted !secret keys should resolve to the same key as unquoted form.
+
+    YAML strips surrounding quotes during parsing, so the secrets.yaml
+    lookup uses the unquoted key. The bundle scan must do the same.
+    """
+    yaml1 = tmp_path / "a.yaml"
+    yaml1.write_text(
+        "single: !secret 'wifi_ssid'\n"
+        'double: !secret "wifi_pw"\n'
+        "bare: !secret api_key\n"
+    )
+
+    keys = _find_used_secret_keys([yaml1])
+    assert keys == {"wifi_ssid", "wifi_pw", "api_key"}
+
+
 # ---------------------------------------------------------------------------
 # _add_bytes_to_tar
 # ---------------------------------------------------------------------------
@@ -1217,6 +1234,35 @@ def test_create_bundle_filters_secrets(tmp_path: Path) -> None:
     assert "should_not_appear" not in secrets_data
 
 
+def test_create_bundle_filters_secrets_quoted(tmp_path: Path) -> None:
+    """Bundling must include secrets.yaml when !secret keys are quoted.
+
+    Regression test for issue 16259: quoted !secret references previously
+    captured the quotes as part of the key, so no key matched secrets.yaml
+    entries and the secrets file was dropped from the bundle entirely.
+    """
+    config_dir = _setup_config_dir(tmp_path)
+
+    secrets = config_dir / "secrets.yaml"
+    secrets.write_text("ota_password: hunter2\nunused: should_not_appear\n")
+
+    config_yaml = "ota:\n  password: !secret 'ota_password'\n"
+    (config_dir / "test.yaml").write_text(config_yaml)
+
+    creator = ConfigBundleCreator({})
+    result = creator.create_bundle()
+
+    assert result.manifest[ManifestKey.HAS_SECRETS] is True
+
+    buf = io.BytesIO(result.data)
+    with tarfile.open(fileobj=buf, mode="r:gz") as tar:
+        secrets_data = tar.extractfile("secrets.yaml").read().decode()
+
+    assert "ota_password" in secrets_data
+    assert "hunter2" in secrets_data
+    assert "unused" not in secrets_data
+
+
 def test_create_bundle_no_secrets(tmp_path: Path) -> None:
     _setup_config_dir(tmp_path)