Merge pull request #18334 from esphome/bump-2026.8.0b2

2026.8.0b2

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.8.0b1
+PROJECT_NUMBER         = 2026.8.0b2
 
 # 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/Dockerfile

@@ -22,7 +22,7 @@ RUN \
     -r /requirements.txt
 
 # Install the ESPHome Device Builder dashboard.
-RUN uv pip install --no-cache-dir esphome-device-builder==1.9.5
+RUN uv pip install --no-cache-dir esphome-device-builder==1.9.6
 
 RUN \
     platformio settings set enable_telemetry No \

esphome/__main__.py

@@ -10,7 +10,7 @@ from pathlib import Path
 import re
 import sys
 import time
-from typing import Protocol
+from typing import TYPE_CHECKING, Protocol
 
 # Note: Do not import modules from esphome.components here, as this would
 # cause them to be loaded before external components are processed, resulting
@@ -71,6 +71,9 @@ from esphome.util import (
     safe_print,
 )
 
+if TYPE_CHECKING:
+    import threading
+
 # Keep expensive imports (zeroconf, writer, yaml_util, etc.) out of this
 # module's top level. Every `esphome` invocation — including fast paths
 # like `esphome version` — pays the cost of what's imported here before
@@ -567,11 +570,48 @@ def has_name_add_mac_suffix() -> bool:
 
 
 def mqtt_get_ip(
-    config: ConfigType, username: str, password: str, client_id: str
+    config: ConfigType,
+    username: str,
+    password: str,
+    client_id: str,
+    stop_event: "threading.Event | None" = None,
 ) -> list[str]:
     from esphome import mqtt
 
-    return mqtt.get_esphome_device_ip(config, username, password, client_id)
+    return mqtt.get_esphome_device_ip(
+        config, username, password, client_id, stop_event=stop_event
+    )
+
+
+def _add_network_device(device: str, network_devices: list[str]) -> None:
+    """Append a device to the list, expanding it through ``CORE.address_cache``.
+
+    If the hostname is already in the address cache (e.g. populated by mDNS
+    discovery), substitute the cached IPs so aioesphomeapi doesn't open its
+    own Zeroconf to re-resolve it. Duplicates are dropped.
+    """
+    if CORE.address_cache and (cached := CORE.address_cache.get_addresses(device)):
+        network_devices.extend(addr for addr in cached if addr not in network_devices)
+    elif device not in network_devices:
+        network_devices.append(device)
+
+
+def _split_network_devices(devices: list[str]) -> tuple[list[str], bool]:
+    """Split the device list into direct addresses and an MQTT-lookup flag.
+
+    Direct addresses are expanded through ``CORE.address_cache`` and deduped
+    the same way ``_resolve_network_devices`` does; MQTT/MQTTIP magic strings
+    are not resolved, only reported via the returned bool so the caller can
+    defer the broker lookup.
+    """
+    network_devices: list[str] = []
+    has_mqtt_lookup = False
+    for device in devices:
+        if get_port_type(device) in _MQTT_PORT_TYPES:
+            has_mqtt_lookup = True
+        else:
+            _add_network_device(device, network_devices)
+    return network_devices, has_mqtt_lookup
 
 
 def _resolve_network_devices(
@@ -604,40 +644,44 @@ def _resolve_network_devices(
         if port_type in _MQTT_PORT_TYPES:
             # Only resolve MQTT once, even if multiple MQTT entries
             if not mqtt_resolved:
-                try:
-                    mqtt_ips = mqtt_get_ip(
-                        config, args.username, args.password, args.client_id
-                    )
-                    # pylint can't infer mqtt_get_ip's return through its
-                    # lazy ``from esphome import mqtt`` import, so it flags
-                    # the genexpr below.
-                    network_devices.extend(
-                        addr
-                        for addr in mqtt_ips  # pylint: disable=not-an-iterable
-                        if addr not in network_devices
-                    )
-                except EsphomeError as err:
-                    _LOGGER.warning(
-                        "MQTT IP discovery failed (%s), will try other devices if available",
-                        err,
-                    )
+                mqtt_ips = _mqtt_get_ip_or_warn(
+                    config, args.username, args.password, args.client_id
+                )
+                network_devices.extend(
+                    addr for addr in mqtt_ips if addr not in network_devices
+                )
                 mqtt_resolved = True
             continue
 
-        # If the hostname is already in the address cache (e.g. populated by
-        # mDNS discovery), substitute the cached IPs so aioesphomeapi doesn't
-        # open its own Zeroconf to re-resolve it.
-        if CORE.address_cache and (cached := CORE.address_cache.get_addresses(device)):
-            network_devices.extend(
-                addr for addr in cached if addr not in network_devices
-            )
-        elif device not in network_devices:
-            # Regular network address or IP - add if not already present
-            network_devices.append(device)
+        _add_network_device(device, network_devices)
 
     return network_devices
 
 
+def _mqtt_get_ip_or_warn(
+    config: ConfigType,
+    username: str,
+    password: str,
+    client_id: str,
+    stop_event: "threading.Event | None" = None,
+) -> list[str]:
+    """Look up the device IP via MQTT, returning [] with a warning on failure.
+
+    This owns the failure policy for MQTT IP discovery on paths that have
+    other addresses to fall back on: a broker problem must not abort the
+    operation. Also used as the deferred resolver handed to ``run_logs``,
+    where it runs in a worker thread.
+    """
+    try:
+        return mqtt_get_ip(config, username, password, client_id, stop_event=stop_event)
+    except EsphomeError as err:
+        _LOGGER.warning(
+            "MQTT IP discovery failed (%s), will try other devices if available",
+            err,
+        )
+        return []
+
+
 def run_miniterm(config: ConfigType, port: str, args) -> int:
     from datetime import datetime
 
@@ -1438,17 +1482,37 @@ def show_logs(config: ConfigType, args: ArgsProtocol, devices: list[str]) -> int
         return run_miniterm(config, port, args)
 
     # Check if we should use API for logging
-    # Resolve MQTT magic strings to actual IP addresses
-    if has_api() and (
-        network_devices := _resolve_network_devices(devices, config, args)
-    ):
-        from esphome.api_client import run_logs
+    if has_api():
+        network_devices, has_mqtt_lookup = _split_network_devices(devices)
+        mqtt_resolver = None
+        if has_mqtt_lookup:
+            if network_devices:
+                # Addresses are already known, so don't block startup on the
+                # MQTT broker lookup; hand it to run_logs as a deferred
+                # resolver that runs in the background and feeds discovered
+                # addresses into the running log client, keeping MQTT as a
+                # fallback for when the known addresses are stale (e.g. DHCP
+                # reassigned the IP).
+                mqtt_resolver = functools.partial(
+                    _mqtt_get_ip_or_warn,
+                    config,
+                    args.username,
+                    args.password,
+                    args.client_id,
+                )
+            else:
+                # The MQTT lookup is the only way to find the device; resolve
+                # it up front since the client needs an address to start with.
+                network_devices = _resolve_network_devices(devices, config, args)
+        if network_devices:
+            from esphome.api_client import run_logs
 
-        return run_logs(
-            config,
-            network_devices,
-            subscribe_states=_should_subscribe_states(args),
-        )
+            return run_logs(
+                config,
+                network_devices,
+                subscribe_states=_should_subscribe_states(args),
+                mqtt_resolver=mqtt_resolver,
+            )
 
     if port_type in (PortType.NETWORK, PortType.MQTT) and has_mqtt_logging():
         from esphome import mqtt

esphome/api_client.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import asyncio
 from contextlib import suppress
 import logging
+import threading
 from typing import TYPE_CHECKING, Any
 import warnings
 
@@ -20,6 +21,8 @@ from esphome.stacktrace import LogLineProcessor
 from esphome.util import safe_print
 
 if TYPE_CHECKING:
+    from collections.abc import Callable
+
     from aioesphomeapi.api_pb2 import (
         SubscribeLogsResponse,  # pylint: disable=no-name-in-module
     )
@@ -32,8 +35,18 @@ async def async_run_logs(
     config: dict[str, Any],
     addresses: list[str],
     subscribe_states: bool = True,
+    mqtt_resolver: Callable[[threading.Event], list[str]] | None = None,
 ) -> None:
-    """Run the logs command in the event loop."""
+    """Run the logs command in the event loop.
+
+    If ``mqtt_resolver`` is given, it is called in a worker thread (paho-mqtt
+    has no asyncio support on Windows) concurrently with the connection
+    attempts to ``addresses``, and any addresses it discovers are fed into
+    the running client. It owns its own failure handling (returning [] when
+    discovery fails) and must honor the ``threading.Event`` it is passed so
+    teardown is not delayed by the lookup's wait window; the initial broker
+    connect itself is only bounded by the socket timeout.
+    """
     from datetime import datetime
 
     conf = config["api"]
@@ -60,6 +73,41 @@ async def async_run_logs(
     # Decoder resolution policy lives in LogLineProcessor.
     processor = LogLineProcessor(config, CORE.target_platform)
 
+    mqtt_task: asyncio.Task[None] | None = None
+    mqtt_stop_event = threading.Event()
+
+    def _cancel_mqtt_discovery() -> None:
+        """Stop the broker lookup once a connection has been established.
+
+        Its answer is only useful while still disconnected: after that it
+        either duplicates the connected address or arrives too late to
+        matter, so don't keep an idle broker session open for it.
+        """
+        mqtt_stop_event.set()
+        if mqtt_task is not None and not mqtt_task.done():
+            mqtt_task.cancel()
+
+    async def _resolve_mqtt_addresses() -> None:
+        """Discover the device address via the MQTT broker in the background."""
+        try:
+            mqtt_ips = await asyncio.to_thread(mqtt_resolver, mqtt_stop_event)
+            if not mqtt_ips:
+                _LOGGER.debug(
+                    "MQTT discovery %s",
+                    "aborted" if mqtt_stop_event.is_set() else "found no addresses",
+                )
+                return
+            if cli.add_addresses(mqtt_ips):
+                _LOGGER.info("Discovered address(es) via MQTT: %s", ", ".join(mqtt_ips))
+            else:
+                _LOGGER.debug(
+                    "MQTT-discovered address(es) already known: %s", ", ".join(mqtt_ips)
+                )
+        except Exception:  # pylint: disable=broad-except
+            # A background task failure would otherwise stay invisible for
+            # the whole session and only re-raise at teardown
+            _LOGGER.exception("MQTT address discovery failed")
+
     def on_log(msg: SubscribeLogsResponse) -> None:
         """Handle a new log message."""
         time_ = datetime.now().astimezone()
@@ -98,20 +146,53 @@ async def async_run_logs(
         # A top-level ``deep_sleep:`` block means the device is only awake
         # briefly; cap the reconnect backoff so a wake window is not missed.
         deep_sleep="deep_sleep" in config,
+        on_connect=_cancel_mqtt_discovery if mqtt_resolver is not None else None,
     )
     try:
+        # Don't start (or keep) the broker lookup if a connection already
+        # succeeded; the stop event doubles as the not-needed-anymore latch
+        # and get_esphome_device_ip returns immediately when it is set.
+        if mqtt_resolver is not None and not mqtt_stop_event.is_set():
+            mqtt_task = asyncio.create_task(_resolve_mqtt_addresses())
         await asyncio.Event().wait()
     finally:
-        await stop()
+        try:
+            if mqtt_task is not None:
+                # Unblock the worker thread first so it can't hold up
+                # loop.shutdown_default_executor() for the full lookup timeout.
+                mqtt_stop_event.set()
+                # Give the worker a moment to exit through its own error
+                # handling; cancelling first would race out a late failure.
+                done, _ = await asyncio.wait([mqtt_task], timeout=1.0)
+                if not done:
+                    mqtt_task.cancel()
+                # return_exceptions keeps a CancelledError from the cancel()
+                # above from re-raising here and jumping over the stop() below.
+                # The task handles Exception itself, so only a BaseException
+                # escape (e.g. SystemExit from the worker) can land here.
+                (result,) = await asyncio.gather(mqtt_task, return_exceptions=True)
+                if isinstance(result, BaseException) and not isinstance(
+                    result, asyncio.CancelledError
+                ):
+                    _LOGGER.error("MQTT address discovery failed", exc_info=result)
+        finally:
+            # Must run even if a second cancellation lands mid-cleanup above
+            await stop()
 
 
 def run_logs(
     config: dict[str, Any],
     addresses: list[str],
     subscribe_states: bool = True,
+    mqtt_resolver: Callable[[threading.Event], list[str]] | None = None,
 ) -> None:
     """Run the logs command."""
     with suppress(KeyboardInterrupt):
         asyncio.run(
-            async_run_logs(config, addresses, subscribe_states=subscribe_states)
+            async_run_logs(
+                config,
+                addresses,
+                subscribe_states=subscribe_states,
+                mqtt_resolver=mqtt_resolver,
+            )
         )

esphome/components/ld2420/ld2420.cpp

@@ -746,7 +746,14 @@ void LD2420Component::set_reg_value(uint16_t reg, uint16_t value) {
   this->send_cmd_from_array(cmd_frame);
 }
 
-void LD2420Component::handle_cmd_error(uint8_t error) { ESP_LOGE(TAG, "Command failed: %s", ERR_MESSAGE[error]); }
+void LD2420Component::handle_cmd_error(uint16_t error) {
+  if (error < std::size(ERR_MESSAGE)) {
+    ESP_LOGE(TAG, "Command failed: %s", ERR_MESSAGE[error]);
+  } else {
+    // The error word comes from the device reply frame; unknown codes must not index ERR_MESSAGE
+    ESP_LOGE(TAG, "Command failed: error 0x%04X", error);
+  }
+}
 
 int LD2420Component::get_gate_threshold_(uint8_t gate) {
   uint8_t error;

esphome/components/ld2420/ld2420.h

@@ -108,7 +108,7 @@ class LD2420Component final : public Component, public uart::UARTDevice {
   float get_setup_priority() const override;
   int send_cmd_from_array(CmdFrameT cmd_frame);
   void report_gate_data();
-  void handle_cmd_error(uint8_t error);
+  void handle_cmd_error(uint16_t error);
   void set_operating_mode(const char *state);
   void auto_calibrate_sensitivity();
   void update_radar_data(uint16_t const *gate_energy, uint8_t sample_number);

esphome/components/sendspin/__init__.py

@@ -234,7 +234,7 @@ async def to_code(config: ConfigType) -> None:
         psram.request_external_task_stack()
 
     # sendspin-cpp library
-    esp32.add_idf_component(name="sendspin/sendspin-cpp", ref="0.7.1")
+    esp32.add_idf_component(name="sendspin/sendspin-cpp", ref="0.7.2")
 
     cg.add_define("USE_SENDSPIN", True)  # for MDNS
 

esphome/components/usb_uart/pl2303.cpp

@@ -292,8 +292,8 @@ bool USBUartTypePL2303::config_step(USBUartChannel *channel, uint8_t step, bool
       // Data bits
       line_coding[6] = channel->get_data_bits();
 
-      ESP_LOGD(TAG, "PL2303: SET_LINE_REQUEST baud=%u stop=%u parity=%u data=%u", baud, line_coding[4], line_coding[5],
-               line_coding[6]);
+      ESP_LOGD(TAG, "PL2303: SET_LINE_REQUEST baud=%" PRIu32 " stop=%u parity=%u data=%u", baud, line_coding[4],
+               line_coding[5], line_coding[6]);
 
       std::vector<uint8_t> lc_vec(line_coding, line_coding + 7);
       this->config_transfer_(SET_LINE_REQUEST_TYPE, SET_LINE_REQUEST, 0, iface, lc_vec);

esphome/const.py

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

esphome/framework_helpers.py

@@ -25,9 +25,13 @@ _LOGGER = logging.getLogger(__name__)
 
 # Attempts per mirror URL before falling through to the next mirror; only
 # mid-stream drops retry (resuming when the server gave a validator),
-# connect errors move on immediately.
+# connect errors move on to the next mirror immediately.
 _MIRROR_ATTEMPTS = 3
 
+# Passes over the whole mirror list when a transient network error is in
+# the mix; matches git.py's _NETWORK_MAX_ATTEMPTS (3 tries, 2s/4s backoff).
+_MIRROR_SWEEP_ATTEMPTS = 3
+
 
 def get_project_link_flags() -> list[str]:
     """Return the sorted -Wl, linker flags from the current build."""
@@ -887,37 +891,51 @@ def _failure_reason(e: Exception) -> str:
     return str(e).split(" for url: ", maxsplit=1)[0] or repr(e)
 
 
-def download_from_mirrors(
-    mirrors: list[str],
-    substitutions: dict[str, str],
-    target: io.RawIOBase | IO[bytes] | PathType,
-    timeout: int = 30,
-) -> str:
+def _spent_attempts_error(e: Exception, attempts: int) -> Exception:
+    """Wrap a failure whose mirror already consumed download attempts, so
+    the sweep classifies it as permanent."""
+    from esphome.core import EsphomeError
+
+    err = EsphomeError(f"failed after {attempts} attempts: {_failure_reason(e)}")
+    err.__cause__ = e
+    return err
+
+
+def _is_transient_download_error(e: Exception) -> bool:
+    """Return True when a download failure is worth retrying.
+
+    Connection-level failures and HTTP 429/5xx are transient. Other HTTP
+    errors, local errors, and exhausted-attempts EsphomeError wrappers
+    (their per-mirror retries are already spent) are permanent.
     """
-    Download file from multiple mirrors with substitution support.
+    # Imported lazily: requests is a heavy import (~85ms) and is only
+    # needed when actually downloading, never during config validation.
+    import requests
 
-    Args:
-        mirrors: list of mirror URLs
-        substitutions: Dictionary of substitutions to apply to URLs
-        target: Target file path or file-like object
-        timeout: Download timeout in seconds
+    if isinstance(e, requests.exceptions.HTTPError):
+        resp = e.response
+        return resp is not None and (resp.status_code == 429 or resp.status_code >= 500)
+    return isinstance(
+        e,
+        (
+            requests.exceptions.ConnectionError,
+            requests.exceptions.Timeout,
+            requests.exceptions.ChunkedEncodingError,
+        ),
+    )
 
-    Returns:
-        The source URL.
 
-    Mirror URL templates that reference a substitution not present in
-    ``substitutions`` are skipped, so callers can offer templates that only
-    apply to some downloads.
+def _try_mirrors_once(
+    urls: list[str],
+    path_target: Path | None,
+    f: IO[bytes] | None,
+    timeout: int,
+    failures: list[tuple[str, Exception]],
+) -> str | None:
+    """Single pass over the resolved mirror ``urls``, one try per URL.
 
-    A path target downloads through ``download_with_resume``, so an
-    interrupted download resumes on the next esphome run; a file-like target
-    only resumes mid-stream drops within this call.
-
-    Raises:
-        ValueError: If mirrors list is empty.
-        EsphomeError: If all download attempts fail; the message lists every
-            attempted URL with its individual failure reason. Also raised if
-            no template matched the provided substitutions.
+    Returns the source URL on success, or None with each URL's exception
+    appended to ``failures``.
     """
     # Imported lazily: requests is a heavy import (~85ms) and is only
     # needed when actually downloading, never during config validation.
@@ -925,43 +943,7 @@ def download_from_mirrors(
 
     from esphome.core import EsphomeError
 
-    ensure_happy_eyeballs()
-
-    # 1. Classify the target: filesystem path or open file object
-    path_target: Path | None = None
-    f: IO[bytes] | None = None
-    if isinstance(target, (str, os.PathLike)):
-        path_target = Path(target)
-    elif isinstance(target, (io.RawIOBase, io.IOBase)):
-        f = target
-    else:
-        raise TypeError(
-            f"target must be str, Path, or file-like object: {type(target)}"
-        )
-
-    # 2. Try each mirror in order
-    failures: list[tuple[str, Exception]] = []
-    skipped: list[tuple[str, str]] = []
-
-    for mirror in mirrors:
-        # 3. Apply substitutions to URL
-        try:
-            url = mirror.format(**substitutions)
-        except KeyError as e:
-            # The template references a substitution not provided for
-            # this download (e.g. SHORT_VERSION only exists for x.y.0
-            # versions) - expected, the template just doesn't apply.
-            _LOGGER.debug("Skipping mirror %s: %s not available", mirror, e)
-            skipped.append((mirror, f"not applicable ({e.args[0]} not available)"))
-            continue
-        except (IndexError, ValueError) as e:
-            # A malformed template (unbalanced braces, bad format spec)
-            # is an authoring error, not an expected fallthrough - warn
-            # even if a later mirror succeeds.
-            _LOGGER.warning("Skipping malformed mirror URL template %s: %r", mirror, e)
-            skipped.append((mirror, f"skipped ({e!r})"))
-            continue
-
+    for url in urls:
         _LOGGER.debug("Trying to download from %s", url)
 
         # Path targets delegate to download_with_resume so a partial
@@ -986,14 +968,14 @@ def download_from_mirrors(
                 failures.append((url, e))
                 continue
 
-        # 4. Download; mid-stream failures retry the same mirror with
-        # resume (see download_with_resume) instead of starting over.
-        # There is no checksum to verify a resumed file against, so a
-        # stitch is only trusted when the server proves consistency: the
-        # If-Range validator guarantees 206 only for unchanged content,
-        # and the expected total length (when the first response carried
-        # one) guards against short or shifted bodies. Without a
-        # validator the retry restarts from zero.
+        # File-like targets download here; mid-stream failures retry the
+        # same mirror with resume (see download_with_resume) instead of
+        # starting over. There is no checksum to verify a resumed file
+        # against, so a stitch is only trusted when the server proves
+        # consistency: the If-Range validator guarantees 206 only for
+        # unchanged content, and the expected total length (when the first
+        # response carried one) guards against short or shifted bodies.
+        # Without a validator the retry restarts from zero.
         offset = 0
         expected_total = 0
         validator = None
@@ -1001,9 +983,12 @@ def download_from_mirrors(
             try:
                 resp, offset = _open_ranged(url, offset, timeout, validator)
             except (requests.RequestException, OSError) as e:
-                # Connect/HTTP error, no bytes flowed — next mirror.
+                # Connect/HTTP error, no bytes flowed — next mirror. Wrap
+                # when earlier attempts were already spent on this mirror.
                 _LOGGER.debug("Failed to download %s: %s", url, str(e))
-                failures.append((url, e))
+                failures.append(
+                    (url, _spent_attempts_error(e, attempt + 1) if attempt else e)
+                )
                 break
 
             try:
@@ -1031,7 +1016,7 @@ def download_from_mirrors(
 
                 _LOGGER.debug("Downloaded successfully from: %s", url)
 
-                # 5. Reset file pointer and return
+                # Reset file pointer and return
                 f.seek(0)
                 return url
 
@@ -1054,16 +1039,124 @@ def download_from_mirrors(
                     )
                     offset = 0
                 if attempt == _MIRROR_ATTEMPTS - 1:
-                    failures.append((url, e))
+                    failures.append((url, _spent_attempts_error(e, _MIRROR_ATTEMPTS)))
 
-    # 6. Report every attempted URL if all mirrors failed. Falling back
-    # past an early mirror is normal (e.g. only one of the framework URL
-    # templates matches a given version's tag), so raising only the last
-    # error would hide the failure that actually matters.
-    if failures:
-        attempts = "".join(
-            f"\n  {url}\n    {_failure_reason(e)}" for url, e in failures
+    return None
+
+
+def download_from_mirrors(
+    mirrors: list[str],
+    substitutions: dict[str, str],
+    target: io.RawIOBase | IO[bytes] | PathType,
+    timeout: int = 30,
+) -> str:
+    """
+    Download file from multiple mirrors with substitution support.
+
+    Args:
+        mirrors: list of mirror URLs
+        substitutions: Dictionary of substitutions to apply to URLs
+        target: Target file path or file-like object
+        timeout: Download timeout in seconds
+
+    Returns:
+        The source URL.
+
+    Mirror URL templates that reference a substitution not present in
+    ``substitutions`` are skipped, so callers can offer templates that only
+    apply to some downloads.
+
+    A path target downloads through ``download_with_resume``, so an
+    interrupted download resumes on the next esphome run; a file-like target
+    only resumes mid-stream drops within this call.
+
+    When every mirror fails and at least one failure is transient (dropped
+    connection, timeout, HTTP 429/5xx), the whole list is retried with a
+    short backoff; permanent failures (e.g. 404) raise immediately.
+
+    Raises:
+        ValueError: If mirrors list is empty.
+        EsphomeError: If all download attempts fail; the message lists every
+            attempted URL with its individual failure reason. Also raised if
+            no template matched the provided substitutions.
+    """
+    from esphome.core import EsphomeError
+
+    ensure_happy_eyeballs()
+
+    # 1. Classify the target: filesystem path or open file object
+    path_target: Path | None = None
+    f: IO[bytes] | None = None
+    if isinstance(target, (str, os.PathLike)):
+        path_target = Path(target)
+    elif isinstance(target, (io.RawIOBase, io.IOBase)):
+        f = target
+    else:
+        raise TypeError(
+            f"target must be str, Path, or file-like object: {type(target)}"
         )
+
+    # 2. Resolve the mirror templates (invariant across retry sweeps)
+    urls: list[str] = []
+    skipped: list[tuple[str, str]] = []
+    for mirror in mirrors:
+        try:
+            urls.append(mirror.format(**substitutions))
+        except KeyError as e:
+            # The template references a substitution not provided for
+            # this download (e.g. SHORT_VERSION only exists for x.y.0
+            # versions) - expected, the template just doesn't apply.
+            _LOGGER.debug("Skipping mirror %s: %s not available", mirror, e)
+            skipped.append((mirror, f"not applicable ({e.args[0]} not available)"))
+        except (IndexError, ValueError) as e:
+            # A malformed template (unbalanced braces, bad format spec)
+            # is an authoring error, not an expected fallthrough - warn
+            # even if a later mirror succeeds.
+            _LOGGER.warning("Skipping malformed mirror URL template %s: %r", mirror, e)
+            skipped.append((mirror, f"skipped ({e!r})"))
+
+    # 3. Sweep the mirror list, retrying transient failures with backoff:
+    # a single pass keeps mirror failover fast, re-sweeping keeps one
+    # network blip from failing the build when only one mirror applies.
+    failures: list[tuple[str, Exception]] = []
+    for sweep in range(1, _MIRROR_SWEEP_ATTEMPTS + 1):
+        sweep_failures: list[tuple[str, Exception]] = []
+        if (
+            url := _try_mirrors_once(urls, path_target, f, timeout, sweep_failures)
+        ) is not None:
+            return url
+        failures.extend(sweep_failures)
+        # Permanent failures (404, verification mismatch) won't heal;
+        # only retry when a transient error is in the mix (as git.py does).
+        transient = next(
+            ((u, e) for u, e in sweep_failures if _is_transient_download_error(e)),
+            None,
+        )
+        if transient is None:
+            break
+        if sweep < _MIRROR_SWEEP_ATTEMPTS:
+            delay = 2**sweep
+            _LOGGER.warning(
+                "Download of %s failed (%s); retrying in %d seconds (attempt %d/%d)",
+                transient[0],
+                _failure_reason(transient[1]),
+                delay,
+                sweep + 1,
+                _MIRROR_SWEEP_ATTEMPTS,
+            )
+            time.sleep(delay)
+
+    # 4. Report every attempted URL if all mirrors failed. failures spans
+    # all sweeps (deduplicated by URL and reason), so neither an early
+    # mirror's failure nor an earlier sweep's failure mode is hidden.
+    if failures:
+        seen: set[tuple[str, str]] = set()
+        attempts = ""
+        for url, e in failures:
+            reason = _failure_reason(e)
+            if (url, reason) not in seen:
+                seen.add((url, reason))
+                attempts += f"\n  {url}\n    {reason}"
         attempts += "".join(f"\n  {mirror}\n    {reason}" for mirror, reason in skipped)
         raise EsphomeError(
             f"Failed to download from all mirrors:{attempts}"

esphome/idf_component.yml

@@ -98,7 +98,7 @@ dependencies:
   esp32async/asynctcp:
     version: 3.4.91
   sendspin/sendspin-cpp:
-    version: 0.7.1
+    version: 0.7.2
   lvgl/lvgl:
     version: 9.5.0
   fastled/FastLED:

esphome/mqtt.py

@@ -6,6 +6,7 @@ from pathlib import Path
 import ssl
 import tempfile
 import time
+from typing import TYPE_CHECKING
 
 import paho.mqtt.client as mqtt
 
@@ -31,6 +32,9 @@ from esphome.helpers import get_int_env, get_str_env
 from esphome.types import ConfigType
 from esphome.util import safe_print
 
+if TYPE_CHECKING:
+    import threading
+
 _LOGGER = logging.getLogger(__name__)
 
 
@@ -164,6 +168,7 @@ def get_esphome_device_ip(
     password: str | None = None,
     client_id: str | None = None,
     timeout: float = 25,
+    stop_event: "threading.Event | None" = None,
 ) -> list[str]:
     if CONF_MQTT not in config:
         raise EsphomeError(
@@ -182,55 +187,113 @@ def get_esphome_device_ip(
 
     dev_name = config[CONF_ESPHOME][CONF_NAME]
     dev_ip = None
+    failed = False
 
     topic = "esphome/discover/" + dev_name
     _LOGGER.info("Starting looking for IP in topic %s", topic)
 
     def on_message(client, userdata, msg):
-        nonlocal dev_ip
+        nonlocal dev_ip, failed
         time_ = datetime.now().astimezone().time().strftime("[%H:%M:%S]")
         payload = msg.payload.decode(errors="backslashreplace")
         if len(payload) > 0:
             message = time_ + " " + payload
             _LOGGER.debug(message)
 
-            data = json.loads(payload)
+            try:
+                data = json.loads(payload)
+            except ValueError:
+                data = None
+            if not isinstance(data, dict):
+                # A raise in this handler would kill paho's network thread
+                _LOGGER.warning("Ignoring unparsable discovery payload")
+                return
             if "name" not in data or data["name"] != dev_name:
                 _LOGGER.warning("Wrong device answer")
                 return
 
-            dev_ip = []
+            addresses = []
             key = "ip"
             n = 0
             while key in data:
-                dev_ip.append(data[key])
+                value = data[key]
+                if (
+                    isinstance(value, str)
+                    and (value := value.strip())
+                    and value.isprintable()
+                ):
+                    addresses.append(value)
+                else:
+                    # repr-escaped and truncated: must not forge log lines
+                    _LOGGER.warning(
+                        "Ignoring invalid address in discovery answer: %s",
+                        repr(value)[:100],
+                    )
                 n = n + 1
                 key = "ip" + str(n)
 
-            if dev_ip:
-                client.disconnect()
+            if not addresses:
+                _LOGGER.warning("Device answer did not include an IP address")
+                failed = True
+                return
+
+            dev_ip = addresses
+            failed = False  # a complete answer wins over an earlier empty one
+            client.disconnect()
 
     def on_connect(client, userdata, flags, return_code):
         topic = "esphome/ping/" + dev_name
         _LOGGER.info("Send discover via MQTT broker topic: %s", topic)
         client.publish(topic, None, retain=False)
 
+    if stop_event is not None and stop_event.is_set():
+        # Teardown already started; don't open a broker connection at all
+        return []
+
+    def on_disconnect(client, userdata, result_code):
+        nonlocal failed
+        if result_code != 0:
+            _LOGGER.warning("Disconnected from MQTT broker (%s)", result_code)
+            failed = True
+
     mqtt_client = prepare(
         config, [topic], on_message, on_connect, username, password, client_id
     )
+    # Discovery is one-shot; prepare()'s reconnect-forever on_disconnect runs
+    # on the network thread and would make loop_stop() below join forever.
+    mqtt_client.on_disconnect = on_disconnect
 
-    mqtt_client.loop_start()
-    while timeout > 0:
-        if dev_ip is not None:
-            break
-        timeout -= 0.250
-        time.sleep(0.250)
-    mqtt_client.loop_stop()
+    if stop_event is None:
+        import threading
+
+        stop_event = threading.Event()  # never set; wait() below is a plain sleep
+    stopped = stop_event.is_set()  # teardown may have started during connect
+    try:
+        if not stopped:
+            mqtt_client.loop_start()
+            while timeout > 0:
+                if dev_ip is not None or failed:
+                    break
+                if stop_event.wait(0.250):
+                    stopped = True
+                    break
+                timeout -= 0.250
+    finally:
+        # A cleanup failure must not replace the discovery result or its
+        # EsphomeError; a second disconnect after on_message's is harmless.
+        try:
+            mqtt_client.disconnect()
+        except Exception:  # pylint: disable=broad-except
+            _LOGGER.debug("Error disconnecting from MQTT broker", exc_info=True)
+        mqtt_client.loop_stop()  # only signals and joins; does not raise
 
     if dev_ip is None:
+        if stopped:
+            # Aborted by the caller, not a failure; stay quiet
+            return []
         raise EsphomeError("Failed to find IP via MQTT")
 
-    _LOGGER.info("Found IP: %s", dev_ip)
+    _LOGGER.info("Found IP via MQTT broker: %s", ", ".join(dev_ip))
     return dev_ip
 
 

requirements.txt

@@ -12,7 +12,7 @@ pyserial==3.5
 platformio==6.1.19
 esptool==5.3.1
 click==8.3.3
-aioesphomeapi==45.10.0
+aioesphomeapi==45.10.1
 aiohappyeyeballs==2.7.1  # Happy Eyeballs for requests downloads; already pulled in by aioesphomeapi
 zeroconf==0.150.0
 puremagic==2.2.0

tests/unit_tests/test_api_client.py

@@ -56,7 +56,7 @@ async def test_async_run_logs_full_flow(caplog) -> None:
 
     with (
         patch.object(api_client, "async_run", mock_run),
-        patch.object(api_client, "APIClient") as mock_client,
+        patch.object(api_client, "APIClient", autospec=True) as mock_client,
         patch.object(api_client, "safe_print", printed.append),
     ):
         task = asyncio.get_running_loop().create_task(
@@ -163,3 +163,324 @@ async def test_async_run_logs_passes_deep_sleep(
         await api_client.async_run_logs(config, ["1.2.3.4"])
 
     assert mock_run.call_args.kwargs["deep_sleep"] is expected_deep_sleep
+
+
+@pytest.mark.asyncio
+async def test_async_run_logs_mqtt_resolver_feeds_addresses(caplog) -> None:
+    """Addresses discovered via MQTT are fed into the running client."""
+    caplog.set_level("INFO", logger="esphome.api_client")
+    CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: "esp32"}
+    config = {"esphome": {"name": "test"}, "api": {CONF_PORT: 6053}}
+
+    stop = AsyncMock()
+    fed = asyncio.Event()
+
+    def resolver(stop_event):
+        return ["10.0.0.9", "10.0.0.10"]
+
+    with (
+        patch.object(api_client, "async_run", AsyncMock(return_value=stop)),
+        patch.object(api_client, "APIClient", autospec=True) as mock_client,
+    ):
+        mock_client.return_value.add_addresses.side_effect = lambda addrs: (
+            fed.set() or True
+        )
+        task = asyncio.get_running_loop().create_task(
+            api_client.async_run_logs(config, ["1.2.3.4"], mqtt_resolver=resolver)
+        )
+        async with asyncio.timeout(1):
+            await fed.wait()
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+
+    mock_client.return_value.add_addresses.assert_called_once_with(
+        ["10.0.0.9", "10.0.0.10"]
+    )
+    assert "Discovered address(es) via MQTT" in caplog.text
+    stop.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_async_run_logs_mqtt_resolver_no_addresses_keeps_running() -> None:
+    """A resolver returning nothing (failed lookup) leaves the session running."""
+    import threading
+
+    CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: "esp32"}
+    config = {"esphome": {"name": "test"}, "api": {CONF_PORT: 6053}}
+
+    stop = AsyncMock()
+    resolver_ran = threading.Event()
+
+    def resolver(stop_event):
+        # The resolver owns failure handling; a failed lookup returns []
+        resolver_ran.set()
+        return []
+
+    with (
+        patch.object(api_client, "async_run", AsyncMock(return_value=stop)),
+        patch.object(api_client, "APIClient", autospec=True) as mock_client,
+    ):
+        task = asyncio.get_running_loop().create_task(
+            api_client.async_run_logs(config, ["1.2.3.4"], mqtt_resolver=resolver)
+        )
+        await asyncio.to_thread(resolver_ran.wait, 1)
+        await asyncio.sleep(0)
+        assert not task.done()
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+
+    mock_client.return_value.add_addresses.assert_not_called()
+    stop.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_async_run_logs_mqtt_resolver_stopped_on_teardown() -> None:
+    """Teardown sets the resolver's stop event so the thread exits promptly."""
+    import threading
+
+    CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: "esp32"}
+    config = {"esphome": {"name": "test"}, "api": {CONF_PORT: 6053}}
+
+    stop = AsyncMock()
+    captured_event: threading.Event | None = None
+    resolver_started = threading.Event()
+
+    def resolver(stop_event):
+        nonlocal captured_event
+        captured_event = stop_event
+        resolver_started.set()
+        # Simulate a slow broker lookup that only ends via the stop event.
+        stop_event.wait(timeout=5)
+        return []
+
+    with (
+        patch.object(api_client, "async_run", AsyncMock(return_value=stop)),
+        patch.object(api_client, "APIClient", autospec=True),
+    ):
+        task = asyncio.get_running_loop().create_task(
+            api_client.async_run_logs(config, ["1.2.3.4"], mqtt_resolver=resolver)
+        )
+        await asyncio.to_thread(resolver_started.wait, 1)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+
+    assert captured_event is not None
+    assert captured_event.is_set()
+    stop.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_async_run_logs_mqtt_resolver_crash_still_stops_cleanly(caplog) -> None:
+    """A resolver raising unexpectedly must not skip stop() at teardown."""
+    import threading
+
+    CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: "esp32"}
+    config = {"esphome": {"name": "test"}, "api": {CONF_PORT: 6053}}
+
+    stop = AsyncMock()
+    resolver_ran = threading.Event()
+
+    def resolver(stop_event):
+        resolver_ran.set()
+        raise RuntimeError("resolver blew up")
+
+    with (
+        patch.object(api_client, "async_run", AsyncMock(return_value=stop)),
+        patch.object(api_client, "APIClient", autospec=True),
+    ):
+        task = asyncio.get_running_loop().create_task(
+            api_client.async_run_logs(config, ["1.2.3.4"], mqtt_resolver=resolver)
+        )
+        await asyncio.to_thread(resolver_ran.wait, 1)
+        await asyncio.sleep(0.05)
+        assert not task.done()
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+
+    assert "MQTT address discovery failed" in caplog.text
+    stop.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_async_run_logs_connect_cancels_mqtt_discovery() -> None:
+    """A successful connection stops the in-flight broker lookup."""
+    import threading
+
+    CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: "esp32"}
+    config = {"esphome": {"name": "test"}, "api": {CONF_PORT: 6053}}
+
+    stop = AsyncMock()
+    captured_event: threading.Event | None = None
+    resolver_started = threading.Event()
+
+    def resolver(stop_event):
+        nonlocal captured_event
+        captured_event = stop_event
+        resolver_started.set()
+        stop_event.wait(timeout=5)
+        return []
+
+    with (
+        patch.object(api_client, "async_run", AsyncMock(return_value=stop)) as mock_run,
+        patch.object(api_client, "APIClient", autospec=True) as mock_client,
+    ):
+        task = asyncio.get_running_loop().create_task(
+            api_client.async_run_logs(config, ["1.2.3.4"], mqtt_resolver=resolver)
+        )
+        await asyncio.to_thread(resolver_started.wait, 1)
+
+        # The runner reports a successful connection
+        on_connect = mock_run.call_args.kwargs["on_connect"]
+        on_connect()
+        await asyncio.sleep(0.05)
+
+        assert captured_event is not None
+        assert captured_event.is_set()
+        assert not task.done()
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+
+    mock_client.return_value.add_addresses.assert_not_called()
+    stop.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_async_run_logs_connect_before_discovery_skips_lookup() -> None:
+    """A connection during async_run startup prevents the lookup from starting."""
+    CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: "esp32"}
+    config = {"esphome": {"name": "test"}, "api": {CONF_PORT: 6053}}
+
+    stop = AsyncMock()
+    resolver = Mock(name="resolver")
+
+    async def fake_async_run(*args, **kwargs):
+        # Connection succeeds before async_run even returns
+        kwargs["on_connect"]()
+        return stop
+
+    with (
+        patch.object(api_client, "async_run", AsyncMock(side_effect=fake_async_run)),
+        patch.object(api_client, "APIClient", autospec=True),
+    ):
+        task = asyncio.get_running_loop().create_task(
+            api_client.async_run_logs(config, ["1.2.3.4"], mqtt_resolver=resolver)
+        )
+        await asyncio.sleep(0.05)
+        assert not task.done()
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+
+    resolver.assert_not_called()
+    stop.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_async_run_logs_mqtt_resolver_duplicate_addresses_logged(caplog) -> None:
+    """A discovery the client rejects as already known leaves a debug trace."""
+    import threading
+
+    caplog.set_level("DEBUG", logger="esphome.api_client")
+    CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: "esp32"}
+    config = {"esphome": {"name": "test"}, "api": {CONF_PORT: 6053}}
+
+    stop = AsyncMock()
+    fed = threading.Event()
+
+    def resolver(stop_event):
+        return ["1.2.3.4"]
+
+    with (
+        patch.object(api_client, "async_run", AsyncMock(return_value=stop)),
+        patch.object(api_client, "APIClient", autospec=True) as mock_client,
+    ):
+        mock_client.return_value.add_addresses.side_effect = lambda addrs: (
+            fed.set() or False
+        )
+        task = asyncio.get_running_loop().create_task(
+            api_client.async_run_logs(config, ["1.2.3.4"], mqtt_resolver=resolver)
+        )
+        await asyncio.to_thread(fed.wait, 1)
+        await asyncio.sleep(0.05)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+
+    mock_client.return_value.add_addresses.assert_called_once_with(["1.2.3.4"])
+    assert "MQTT-discovered address(es) already known: 1.2.3.4" in caplog.text
+    assert "Discovered address(es) via MQTT" not in caplog.text
+    stop.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_async_run_logs_base_exception_escape_logged_at_teardown(caplog) -> None:
+    """A BaseException escaping the worker is reported, and stop() still runs."""
+    import threading
+
+    CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: "esp32"}
+    config = {"esphome": {"name": "test"}, "api": {CONF_PORT: 6053}}
+
+    stop = AsyncMock()
+    resolver_ran = threading.Event()
+
+    class WorkerEscape(BaseException):
+        """Not an Exception, so the task-level guard must not catch it."""
+
+    def resolver(stop_event):
+        resolver_ran.set()
+        raise WorkerEscape("worker bailed")
+
+    with (
+        patch.object(api_client, "async_run", AsyncMock(return_value=stop)),
+        patch.object(api_client, "APIClient", autospec=True),
+    ):
+        task = asyncio.get_running_loop().create_task(
+            api_client.async_run_logs(config, ["1.2.3.4"], mqtt_resolver=resolver)
+        )
+        await asyncio.to_thread(resolver_ran.wait, 1)
+        await asyncio.sleep(0.05)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+
+    assert "MQTT address discovery failed" in caplog.text
+    stop.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_async_run_logs_stubborn_worker_cancelled_at_teardown() -> None:
+    """A worker that ignores the stop event is cancelled after the grace period."""
+    import threading
+
+    CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: "esp32"}
+    config = {"esphome": {"name": "test"}, "api": {CONF_PORT: 6053}}
+
+    stop = AsyncMock()
+    resolver_ran = threading.Event()
+    release = threading.Event()
+
+    def resolver(stop_event):
+        resolver_ran.set()
+        # Ignore stop_event entirely; only the test releases us
+        release.wait(timeout=10)
+        return []
+
+    with (
+        patch.object(api_client, "async_run", AsyncMock(return_value=stop)),
+        patch.object(api_client, "APIClient", autospec=True),
+    ):
+        task = asyncio.get_running_loop().create_task(
+            api_client.async_run_logs(config, ["1.2.3.4"], mqtt_resolver=resolver)
+        )
+        await asyncio.to_thread(resolver_ran.wait, 1)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        release.set()
+
+    stop.assert_awaited_once()

tests/unit_tests/test_framework_helpers.py

@@ -12,7 +12,7 @@ from pathlib import Path
 import subprocess
 import sys
 import tarfile
-from unittest.mock import MagicMock, Mock, patch
+from unittest.mock import MagicMock, Mock, call, patch
 import zipfile
 
 import pytest
@@ -23,6 +23,7 @@ from esphome.core import EsphomeError
 from esphome.framework_helpers import (
     _7z_extract_all,
     _detect_archive_root,
+    _is_transient_download_error,
     _rename_with_retry,
     _tar_extract_all,
     _zip_extract_all,
@@ -515,16 +516,23 @@ class TestArchiveExtractAll:
 # ---------------------------------------------------------------------------
 
 
-def _mock_response(content: bytes, ok: bool = True) -> MagicMock:
+def _mock_response(
+    content: bytes, ok: bool = True, status: int | None = None
+) -> MagicMock:
+    """A fake requests response. The HTTPError carries the response (as
+    ``raise_for_status`` on a real response) so the transient classifier
+    can see its ``status``; failures default to a permanent 404."""
+    if status is None:
+        status = 200 if ok else 404
     r = MagicMock()
     r.__enter__.return_value = r
     r.__exit__.return_value = False
-    r.status_code = 200
+    r.status_code = status
     r.ok = ok
     if ok:
         r.raise_for_status.return_value = None
     else:
-        r.raise_for_status.side_effect = req.HTTPError("503")
+        r.raise_for_status.side_effect = req.HTTPError(str(status), response=r)
     r.headers = {"content-length": "0"}  # suppress ProgressBar
     r.iter_content.return_value = [content] if content else []
     return r
@@ -1419,6 +1427,191 @@ class TestDownloadFromMirrors:
         assert target.exists()
         assert target.read_bytes() == b""
 
+    @pytest.mark.parametrize("target_kind", ["path", "file-like"])
+    def test_transient_failure_retries_mirror_sweep(
+        self, tmp_path: Path, target_kind: str
+    ) -> None:
+        """A transient connect error on the only applicable mirror retries the
+        whole mirror list with backoff instead of failing the build."""
+        target = tmp_path / "idf.tar.xz" if target_kind == "path" else io.BytesIO()
+        with (
+            patch(
+                "requests.get",
+                side_effect=[
+                    req.ConnectionError("Remote end closed connection"),
+                    _mock_response(b"data"),
+                ],
+            ) as mock_get,
+            patch("esphome.framework_helpers.time.sleep") as mock_sleep,
+        ):
+            url = download_from_mirrors(["https://mirror1.com/f"], {}, target)
+        assert url == "https://mirror1.com/f"
+        data = target.read_bytes() if target_kind == "path" else target.getvalue()
+        assert data == b"data"
+        assert mock_get.call_count == 2
+        mock_sleep.assert_called_once_with(2)
+
+    def test_permanent_failure_does_not_retry_sweep(self, tmp_path: Path) -> None:
+        """An HTTP 404 will not heal on its own; fail after a single pass."""
+        with (
+            patch(
+                "requests.get", return_value=_mock_response(b"", ok=False, status=404)
+            ) as mock_get,
+            patch("esphome.framework_helpers.time.sleep") as mock_sleep,
+            pytest.raises(EsphomeError, match="all mirrors"),
+        ):
+            download_from_mirrors(["https://mirror1.com/f"], {}, tmp_path / "out.bin")
+        assert mock_get.call_count == 1
+        mock_sleep.assert_not_called()
+
+    def test_transient_failure_exhausts_sweeps(self, tmp_path: Path) -> None:
+        """A persistent transient error gives up after the configured number
+        of passes, with 2s/4s backoff, and still lists the attempted URL."""
+        with (
+            patch("requests.get", side_effect=req.ConnectionError("down")) as mock_get,
+            patch("esphome.framework_helpers.time.sleep") as mock_sleep,
+            pytest.raises(EsphomeError, match="all mirrors") as ei,
+        ):
+            download_from_mirrors(["https://mirror1.com/f"], {}, tmp_path / "out.bin")
+        assert mock_get.call_count == 3
+        assert mock_sleep.call_args_list == [call(2), call(4)]
+        assert "https://mirror1.com/f" in str(ei.value)
+
+    def test_mixed_permanent_and_transient_retries_sweep(self, tmp_path: Path) -> None:
+        """One mirror 404s permanently while another hits a transient error;
+        the transient failure makes the whole list worth another pass."""
+        dest = tmp_path / "out.bin"
+        with (
+            patch(
+                "requests.get",
+                side_effect=[
+                    _mock_response(b"", ok=False, status=404),
+                    req.ConnectionError("down"),
+                    _mock_response(b"", ok=False, status=404),
+                    _mock_response(b"data"),
+                ],
+            ),
+            patch("esphome.framework_helpers.time.sleep") as mock_sleep,
+        ):
+            url = download_from_mirrors(
+                ["https://mirror1.com/f", "https://mirror2.com/f"], {}, dest
+            )
+        assert url == "https://mirror2.com/f"
+        assert dest.read_bytes() == b"data"
+        mock_sleep.assert_called_once_with(2)
+
+    def test_http_5xx_retries_sweep(self, tmp_path: Path) -> None:
+        """A real 5xx (response attached to the HTTPError) is transient."""
+        dest = tmp_path / "out.bin"
+        with (
+            patch(
+                "requests.get",
+                side_effect=[
+                    _mock_response(b"", ok=False, status=503),
+                    _mock_response(b"data"),
+                ],
+            ),
+            patch("esphome.framework_helpers.time.sleep") as mock_sleep,
+        ):
+            url = download_from_mirrors(["https://mirror1.com/f"], {}, dest)
+        assert url == "https://mirror1.com/f"
+        assert dest.read_bytes() == b"data"
+        mock_sleep.assert_called_once_with(2)
+
+    def test_error_reports_failure_modes_from_all_sweeps(self, tmp_path: Path) -> None:
+        """A failure mode that changes between sweeps stays in the final
+        error; the first failure (the one that started the retries) is
+        chained as the cause."""
+        with (
+            patch(
+                "requests.get",
+                side_effect=[
+                    req.ConnectionError("dropped by middlebox"),
+                    _mock_response(b"", ok=False, status=404),
+                ],
+            ),
+            patch("esphome.framework_helpers.time.sleep") as mock_sleep,
+            pytest.raises(EsphomeError, match="all mirrors") as ei,
+        ):
+            download_from_mirrors(["https://mirror1.com/f"], {}, tmp_path / "out.bin")
+        assert "dropped by middlebox" in str(ei.value)
+        assert "404" in str(ei.value)
+        assert isinstance(ei.value.__cause__, req.ConnectionError)
+        mock_sleep.assert_called_once_with(2)
+
+    def test_exhausted_mid_stream_attempts_not_swept(self) -> None:
+        """A file-like mirror that spent all its mid-stream attempts is not
+        retried again at the sweep level (unlike a path target, it has no
+        part file to resume from on a later sweep)."""
+        buf = io.BytesIO()
+        with (
+            patch(
+                "requests.get",
+                side_effect=[_interrupted_response(b"1234") for _ in range(3)],
+            ) as mock_get,
+            patch("esphome.framework_helpers.time.sleep") as mock_sleep,
+            pytest.raises(EsphomeError, match="failed after 3 attempts"),
+        ):
+            download_from_mirrors(["https://mirror1.com/f"], {}, buf)
+        assert mock_get.call_count == 3
+        mock_sleep.assert_not_called()
+
+    def test_mid_stream_drop_then_connect_error_not_swept(self) -> None:
+        """A connect error on a later attempt (after a mid-stream drop spent
+        one) also counts as spent budget and does not re-arm the sweep."""
+        buf = io.BytesIO()
+        with (
+            patch(
+                "requests.get",
+                side_effect=[
+                    _interrupted_response(b"1234"),
+                    req.ConnectionError("down"),
+                ],
+            ) as mock_get,
+            patch("esphome.framework_helpers.time.sleep") as mock_sleep,
+            pytest.raises(EsphomeError, match="failed after 2 attempts"),
+        ):
+            download_from_mirrors(["https://mirror1.com/f"], {}, buf)
+        assert mock_get.call_count == 2
+        mock_sleep.assert_not_called()
+
+
+def _http_error(status: int) -> req.HTTPError:
+    """An HTTPError carrying a response with the given status, as raised by
+    ``raise_for_status`` on a real response."""
+    resp = MagicMock()
+    resp.status_code = status
+    return req.HTTPError(str(status), response=resp)
+
+
+class TestIsTransientDownloadError:
+    def test_connection_errors_are_transient(self) -> None:
+        assert _is_transient_download_error(req.ConnectionError("reset"))
+        assert _is_transient_download_error(req.Timeout("timed out"))
+        assert _is_transient_download_error(
+            req.exceptions.ChunkedEncodingError("dropped")
+        )
+
+    def test_http_statuses(self) -> None:
+        assert not _is_transient_download_error(_http_error(404))
+        assert not _is_transient_download_error(_http_error(403))
+        assert _is_transient_download_error(_http_error(429))
+        assert _is_transient_download_error(_http_error(503))
+
+    def test_http_error_without_response_is_permanent(self) -> None:
+        assert not _is_transient_download_error(req.HTTPError("boom"))
+
+    def test_exhausted_resume_attempts_are_permanent(self) -> None:
+        """download_with_resume already spent its own resume attempts; its
+        EsphomeError wrapper is not retried again at the sweep level."""
+        wrapped = EsphomeError("Failed to download after 3 attempts")
+        wrapped.__cause__ = req.ConnectionError("down")
+        assert not _is_transient_download_error(wrapped)
+
+    def test_unrelated_errors_are_permanent(self) -> None:
+        assert not _is_transient_download_error(OSError("disk full"))
+        assert not _is_transient_download_error(EsphomeError("size mismatch"))
+
 
 def test_importing_framework_helpers_does_not_import_requests() -> None:
     """Importing framework_helpers must not drag in requests.

tests/unit_tests/test_main.py

@@ -25,6 +25,7 @@ from esphome.__main__ import (
     _make_crystal_freq_callback,
     _redact_with_legacy_fallback,
     _resolve_network_devices,
+    _split_network_devices,
     _unresolved_default_error,
     _validate_bootloader_binary,
     _validate_partition_table_binary,
@@ -2879,7 +2880,9 @@ def test_upload_program_ota_with_mqtt_resolution(
 
     assert exit_code == 0
     assert host == "192.168.1.100"
-    mock_mqtt_get_ip.assert_called_once_with(config, "user", "pass", "client")
+    mock_mqtt_get_ip.assert_called_once_with(
+        config, "user", "pass", "client", stop_event=None
+    )
     expected_firmware = (
         tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
     )
@@ -2926,7 +2929,9 @@ def test_upload_program_ota_with_mqtt_empty_broker(
     assert exit_code == 0
     assert host == "192.168.1.50"
     # Verify MQTT was attempted but failed gracefully
-    mock_mqtt_get_ip.assert_called_once_with(config, "user", "pass", "client")
+    mock_mqtt_get_ip.assert_called_once_with(
+        config, "user", "pass", "client", stop_event=None
+    )
     # Verify we fell back to the IP address
     expected_firmware = (
         tmp_path / ".esphome" / "build" / "test" / ".pioenvs" / "test" / "firmware.bin"
@@ -3015,7 +3020,10 @@ def test_show_logs_api(
 
     assert result == 0
     mock_run_logs.assert_called_once_with(
-        CORE.config, ["192.168.1.100", "192.168.1.101"], subscribe_states=True
+        CORE.config,
+        ["192.168.1.100", "192.168.1.101"],
+        subscribe_states=True,
+        mqtt_resolver=None,
     )
 
 
@@ -3042,7 +3050,7 @@ def test_show_logs_api_no_states(
 
     assert result == 0
     mock_run_logs.assert_called_once_with(
-        CORE.config, ["192.168.1.100"], subscribe_states=False
+        CORE.config, ["192.168.1.100"], subscribe_states=False, mqtt_resolver=None
     )
 
 
@@ -3069,7 +3077,7 @@ def test_show_logs_api_with_fqdn_mdns_disabled(
     assert result == 0
     # Should use the FQDN directly, not try MQTT lookup
     mock_run_logs.assert_called_once_with(
-        CORE.config, ["device.example.com"], subscribe_states=True
+        CORE.config, ["device.example.com"], subscribe_states=True, mqtt_resolver=None
     )
 
 
@@ -3097,9 +3105,44 @@ def test_show_logs_api_with_mqtt_fallback(
     result = show_logs(CORE.config, args, devices)
 
     assert result == 0
-    mock_mqtt_get_ip.assert_called_once_with(CORE.config, "user", "pass", "client")
+    mock_mqtt_get_ip.assert_called_once_with(
+        CORE.config, "user", "pass", "client", stop_event=None
+    )
     mock_run_logs.assert_called_once_with(
-        CORE.config, ["192.168.1.200"], subscribe_states=True
+        CORE.config, ["192.168.1.200"], subscribe_states=True, mqtt_resolver=None
+    )
+
+
+@patch("esphome.mqtt.show_logs")
+def test_show_logs_api_mqtt_only_resolve_failure_falls_back_to_mqtt_logs(
+    mock_mqtt_show_logs: Mock,
+    mock_mqtt_get_ip: Mock,
+) -> None:
+    """With no addresses at all after a failed MQTT lookup, MQTT logging is used."""
+    setup_core(
+        config={
+            "logger": {},
+            CONF_API: {},
+            CONF_MQTT: {CONF_BROKER: "mqtt.local"},
+        },
+        platform=PLATFORM_ESP32,
+    )
+    mock_mqtt_show_logs.return_value = 0
+    mock_mqtt_get_ip.side_effect = EsphomeError("Failed to find IP via MQTT")
+
+    args = MockArgs(
+        topic="esphome/logs", username="user", password="pass", client_id="client"
+    )
+    devices = ["MQTT", "MQTTIP"]
+
+    result = show_logs(CORE.config, args, devices)
+
+    assert result == 0
+    mock_mqtt_get_ip.assert_called_once_with(
+        CORE.config, "user", "pass", "client", stop_event=None
+    )
+    mock_mqtt_show_logs.assert_called_once_with(
+        CORE.config, "esphome/logs", "user", "pass", "client"
     )
 
 
@@ -3466,7 +3509,9 @@ def test_mqtt_get_ip() -> None:
         result = mqtt_get_ip(config, "user", "pass", "client-id")
 
         assert result == ["192.168.1.100", "192.168.1.101"]
-        mock_get_ip.assert_called_once_with(config, "user", "pass", "client-id")
+        mock_get_ip.assert_called_once_with(
+            config, "user", "pass", "client-id", stop_event=None
+        )
 
 
 def test_has_resolvable_address() -> None:
@@ -3847,6 +3892,37 @@ def test_resolve_network_devices_keeps_uncached_hosts(tmp_path: Path) -> None:
     assert result == ["unknown.local", "192.168.1.50"]
 
 
+def test_split_network_devices_direct_only(tmp_path: Path) -> None:
+    """Direct addresses pass through deduped, with no MQTT flag."""
+    setup_core(tmp_path=tmp_path)
+
+    assert _split_network_devices(["192.168.1.50", "device.local", "192.168.1.50"]) == (
+        ["192.168.1.50", "device.local"],
+        False,
+    )
+
+
+def test_split_network_devices_mqtt_only(tmp_path: Path) -> None:
+    """MQTT magic strings produce no direct addresses, only the flag."""
+    setup_core(tmp_path=tmp_path)
+
+    assert _split_network_devices(["MQTTIP", "MQTT"]) == ([], True)
+
+
+def test_split_network_devices_expands_cached_mdns_hosts(tmp_path: Path) -> None:
+    """Hostnames in ``CORE.address_cache`` are expanded like _resolve_network_devices."""
+    setup_core(tmp_path=tmp_path)
+    CORE.address_cache = AddressCache(
+        mdns_cache={
+            "device-abc123.local": ["10.0.0.1", "10.0.0.2"],
+        }
+    )
+
+    assert _split_network_devices(
+        ["device-abc123.local", "MQTTIP", "192.168.1.50", "device-abc123.local"]
+    ) == (["10.0.0.1", "10.0.0.2", "192.168.1.50"], True)
+
+
 def test_await_discovery_timeout_returns_empty(
     caplog: pytest.LogCaptureFixture,
 ) -> None:
@@ -5022,7 +5098,9 @@ def test_upload_program_ota_static_ip_with_mqttip(
     assert host == "192.168.1.100"
 
     # Verify MQTT was resolved
-    mock_mqtt_get_ip.assert_called_once_with(config, "user", "pass", "client")
+    mock_mqtt_get_ip.assert_called_once_with(
+        config, "user", "pass", "client", stop_event=None
+    )
 
     # Verify espota2.run_ota was called with both IPs
     expected_firmware = (
@@ -5069,7 +5147,9 @@ def test_upload_program_ota_multiple_mqttip_resolves_once(
     assert host == "192.168.2.50"
 
     # Verify MQTT was only resolved once despite multiple MQTT magic strings
-    mock_mqtt_get_ip.assert_called_once_with(config, "user", "pass", "client")
+    mock_mqtt_get_ip.assert_called_once_with(
+        config, "user", "pass", "client", stop_event=None
+    )
 
     # Verify espota2.run_ota was called with all unique IPs
     expected_firmware = (
@@ -5116,7 +5196,9 @@ def test_upload_program_ota_mqttip_deduplication(
     assert host == "192.168.1.100"
 
     # Verify MQTT was resolved
-    mock_mqtt_get_ip.assert_called_once_with(config, "user", "pass", "client")
+    mock_mqtt_get_ip.assert_called_once_with(
+        config, "user", "pass", "client", stop_event=None
+    )
 
     # Verify espota2.run_ota was called with deduplicated IPs (only one instance of 192.168.1.100)
     # Note: Current implementation doesn't dedupe, so we'll get the IP twice
@@ -5136,7 +5218,9 @@ def test_show_logs_api_static_ip_with_mqttip(
 
     This tests the scenario where a device has manual_ip (static IP) configured
     and MQTT is also configured. The devices list contains both the static IP
-    and "MQTTIP" magic string.
+    and "MQTTIP" magic string. The MQTT lookup must not block startup; it is
+    handed to run_logs as a deferred resolver instead (issue #18311), while
+    still being reachable as a fallback for a stale static IP.
     """
     setup_core(
         config={
@@ -5157,12 +5241,19 @@ def test_show_logs_api_static_ip_with_mqttip(
 
     assert result == 0
 
-    # Verify MQTT was resolved
-    mock_mqtt_get_ip.assert_called_once_with(CORE.config, "user", "pass", "client")
+    # The broker must not be contacted before run_logs starts
+    mock_mqtt_get_ip.assert_not_called()
 
-    # Verify run_logs was called with both IPs
-    mock_run_logs.assert_called_once_with(
-        CORE.config, ["192.168.1.100", "192.168.2.50"], subscribe_states=True
+    # run_logs gets the static IP immediately plus a deferred MQTT resolver
+    mock_run_logs.assert_called_once()
+    assert mock_run_logs.call_args.args == (CORE.config, ["192.168.1.100"])
+    assert mock_run_logs.call_args.kwargs["subscribe_states"] is True
+    resolver = mock_run_logs.call_args.kwargs["mqtt_resolver"]
+
+    # Invoking the resolver performs the MQTT lookup (the #11260 fallback)
+    assert resolver(None) == ["192.168.2.50"]
+    mock_mqtt_get_ip.assert_called_once_with(
+        CORE.config, "user", "pass", "client", stop_event=None
     )
 
 
@@ -5171,7 +5262,7 @@ def test_show_logs_api_multiple_mqttip_resolves_once(
     mock_run_logs: Mock,
     mock_mqtt_get_ip: Mock,
 ) -> None:
-    """Test that MQTT resolution only happens once for show_logs with multiple MQTT magic strings."""
+    """Test that multiple MQTT magic strings collapse into one deferred resolver."""
     setup_core(
         config={
             "logger": {},
@@ -5191,16 +5282,16 @@ def test_show_logs_api_multiple_mqttip_resolves_once(
 
     assert result == 0
 
-    # Verify MQTT was only resolved once despite multiple MQTT magic strings
-    mock_mqtt_get_ip.assert_called_once_with(CORE.config, "user", "pass", "client")
+    # Note: "MQTT" is a different magic string from "MQTTIP", but both defer
+    # to the same single resolver; the broker is not contacted eagerly
+    mock_mqtt_get_ip.assert_not_called()
+    mock_run_logs.assert_called_once()
+    assert mock_run_logs.call_args.args == (CORE.config, ["192.168.1.100"])
 
-    # Verify run_logs was called with all unique IPs (MQTT strings replaced with IPs)
-    # Note: "MQTT" is a different magic string from "MQTTIP", but both trigger MQTT resolution
-    # The _resolve_network_devices helper filters out both after first resolution
-    mock_run_logs.assert_called_once_with(
-        CORE.config,
-        ["192.168.2.50", "192.168.2.51", "192.168.1.100"],
-        subscribe_states=True,
+    resolver = mock_run_logs.call_args.kwargs["mqtt_resolver"]
+    assert resolver(None) == ["192.168.2.50", "192.168.2.51"]
+    mock_mqtt_get_ip.assert_called_once_with(
+        CORE.config, "user", "pass", "client", stop_event=None
     )
 
 
@@ -5238,7 +5329,9 @@ def test_upload_program_ota_mqtt_timeout_fallback(
     assert host == "192.168.1.100"
 
     # Verify MQTT was attempted
-    mock_mqtt_get_ip.assert_called_once_with(config, "user", "pass", "client")
+    mock_mqtt_get_ip.assert_called_once_with(
+        config, "user", "pass", "client", stop_event=None
+    )
 
     # Verify espota2.run_ota was called with only the static IP (MQTT failed)
     expected_firmware = (
@@ -5254,7 +5347,7 @@ def test_show_logs_api_mqtt_timeout_fallback(
     mock_run_logs: Mock,
     mock_mqtt_get_ip: Mock,
 ) -> None:
-    """Test show_logs falls back to other devices when MQTT times out."""
+    """Test show_logs proceeds with the static IP when MQTT times out."""
     setup_core(
         config={
             "logger": {},
@@ -5273,15 +5366,17 @@ def test_show_logs_api_mqtt_timeout_fallback(
 
     result = show_logs(CORE.config, args, devices)
 
-    # Should succeed using the static IP even though MQTT failed
+    # Logs start on the static IP without waiting for the broker
     assert result == 0
+    mock_run_logs.assert_called_once()
+    assert mock_run_logs.call_args.args == (CORE.config, ["192.168.1.100"])
 
-    # Verify MQTT was attempted
-    mock_mqtt_get_ip.assert_called_once_with(CORE.config, "user", "pass", "client")
-
-    # Verify run_logs was called with only the static IP (MQTT failed)
-    mock_run_logs.assert_called_once_with(
-        CORE.config, ["192.168.1.100"], subscribe_states=True
+    # The deferred resolver owns the failure policy: it logs a warning and
+    # returns no addresses so the session keeps running on the known ones
+    resolver = mock_run_logs.call_args.kwargs["mqtt_resolver"]
+    assert resolver(None) == []
+    mock_mqtt_get_ip.assert_called_once_with(
+        CORE.config, "user", "pass", "client", stop_event=None
     )
 
 
@@ -6764,7 +6859,7 @@ def test_command_run_passes_no_states_to_show_logs(
 
     assert result == 0
     mock_run_logs.assert_called_once_with(
-        CORE.config, ["192.168.1.100"], subscribe_states=False
+        CORE.config, ["192.168.1.100"], subscribe_states=False, mqtt_resolver=None
     )
 
 
@@ -6805,7 +6900,7 @@ def test_command_run_defaults_subscribe_states_true(
 
     assert result == 0
     mock_run_logs.assert_called_once_with(
-        CORE.config, ["192.168.1.100"], subscribe_states=True
+        CORE.config, ["192.168.1.100"], subscribe_states=True, mqtt_resolver=None
     )
 
 

tests/unit_tests/test_mqtt.py

@@ -2,6 +2,11 @@
 
 from __future__ import annotations
 
+import json
+import threading
+import time
+from unittest.mock import MagicMock, patch
+
 import pytest
 
 from esphome.const import CONF_BROKER, CONF_ESPHOME, CONF_MQTT, CONF_NAME
@@ -89,3 +94,260 @@ def test_get_esphome_device_ip_missing_name() -> None:
         match="Cannot discover IP via MQTT as the config does not include the device name:",
     ):
         get_esphome_device_ip(config)
+
+
+def _discovery_config() -> dict:
+    return {
+        CONF_MQTT: {
+            CONF_BROKER: "mqtt.local",
+        },
+        CONF_ESPHOME: {
+            CONF_NAME: "test-device",
+        },
+    }
+
+
+def _deliver_on_loop_start(mock_prepare, client, payload: bytes) -> None:
+    """Deliver a discovery answer as soon as the network loop starts."""
+
+    def deliver(*args, **kwargs):
+        msg = MagicMock()
+        msg.payload = payload
+        mock_prepare.call_args.args[2](client, None, msg)
+
+    client.loop_start.side_effect = deliver
+
+
+def test_get_esphome_device_ip_success() -> None:
+    """A device answer on the discovery topic returns its IPs."""
+    client = MagicMock()
+
+    with patch("esphome.mqtt.prepare", return_value=client) as mock_prepare:
+        _deliver_on_loop_start(
+            mock_prepare,
+            client,
+            json.dumps(
+                {"name": "test-device", "ip": "10.0.0.5", "ip1": "10.0.0.6"}
+            ).encode(),
+        )
+
+        result = get_esphome_device_ip(_discovery_config())
+
+    assert result == ["10.0.0.5", "10.0.0.6"]
+    client.loop_stop.assert_called_once_with()
+    # Once from on_message on receiving the answer, once from the finally
+    assert client.disconnect.call_count == 2
+
+
+def test_get_esphome_device_ip_preset_stop_event_skips_lookup() -> None:
+    """A stop event set before the call returns [] without touching the broker."""
+    stop_event = threading.Event()
+    stop_event.set()
+
+    with patch("esphome.mqtt.prepare") as mock_prepare:
+        result = get_esphome_device_ip(_discovery_config(), stop_event=stop_event)
+
+    assert result == []
+    mock_prepare.assert_not_called()
+
+
+def test_get_esphome_device_ip_stop_event_aborts_wait() -> None:
+    """A stop event set mid-wait exits quietly with no addresses."""
+    stop_event = threading.Event()
+    client = MagicMock()
+    # Simulate teardown starting right after the network loop spins up
+    client.loop_start.side_effect = stop_event.set
+
+    start = time.monotonic()
+    with patch("esphome.mqtt.prepare", return_value=client):
+        result = get_esphome_device_ip(_discovery_config(), stop_event=stop_event)
+
+    # An abort is not a failure and must be nowhere near the 25s timeout
+    assert result == []
+    assert time.monotonic() - start < 5
+    client.disconnect.assert_called_once_with()
+    client.loop_stop.assert_called_once_with()
+
+
+def test_get_esphome_device_ip_timeout_raises() -> None:
+    """No answer within the timeout raises EsphomeError (default stop event path)."""
+    client = MagicMock()
+    with (
+        patch("esphome.mqtt.prepare", return_value=client),
+        pytest.raises(EsphomeError, match="Failed to find IP via MQTT"),
+    ):
+        get_esphome_device_ip(_discovery_config(), timeout=0.25)
+
+    client.disconnect.assert_called_once_with()
+    client.loop_stop.assert_called_once_with()
+
+
+def test_get_esphome_device_ip_stop_during_connect_skips_wait() -> None:
+    """A stop event set while the broker connect is in flight still cleans up."""
+    stop_event = threading.Event()
+    client = MagicMock()
+
+    def prepare_and_stop(*args):
+        stop_event.set()
+        return client
+
+    with patch("esphome.mqtt.prepare", side_effect=prepare_and_stop):
+        result = get_esphome_device_ip(_discovery_config(), stop_event=stop_event)
+
+    assert result == []
+    client.loop_start.assert_not_called()
+    client.disconnect.assert_called_once_with()
+    client.loop_stop.assert_called_once_with()
+
+
+def test_get_esphome_device_ip_replaces_reconnect_handler(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """The one-shot discovery client must not inherit the reconnect-forever
+    handler, which would make loop_stop() join the network thread forever;
+    its replacement still reports a broker-initiated disconnect."""
+    client = MagicMock()
+    prepare_handler = MagicMock()
+    client.on_disconnect = prepare_handler
+
+    with (
+        patch("esphome.mqtt.prepare", return_value=client),
+        pytest.raises(EsphomeError, match="Failed to find IP via MQTT"),
+    ):
+        get_esphome_device_ip(_discovery_config(), timeout=0.25)
+
+    assert client.on_disconnect is not prepare_handler
+    client.on_disconnect(client, None, 0)
+    assert "Disconnected from MQTT broker" not in caplog.text
+    client.on_disconnect(client, None, 5)
+    assert "Disconnected from MQTT broker (5)" in caplog.text
+
+
+def test_get_esphome_device_ip_answer_without_ip_fails_fast(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A device answer with no IP fields fails promptly, not at the timeout."""
+    client = MagicMock()
+
+    with patch("esphome.mqtt.prepare", return_value=client) as mock_prepare:
+        _deliver_on_loop_start(
+            mock_prepare, client, json.dumps({"name": "test-device"}).encode()
+        )
+
+        start = time.monotonic()
+        with pytest.raises(EsphomeError, match="Failed to find IP via MQTT"):
+            get_esphome_device_ip(_discovery_config(), timeout=5)
+
+    assert time.monotonic() - start < 1
+    assert "Device answer did not include an IP address" in caplog.text
+
+
+@pytest.mark.parametrize("payload", [b"not json {", b"123", b"null"])
+def test_get_esphome_device_ip_unparsable_payload_ignored(
+    caplog: pytest.LogCaptureFixture,
+    payload: bytes,
+) -> None:
+    """Garbage on the discovery topic must not kill paho's network thread."""
+    client = MagicMock()
+
+    with patch("esphome.mqtt.prepare", return_value=client) as mock_prepare:
+        _deliver_on_loop_start(mock_prepare, client, payload)
+
+        with pytest.raises(EsphomeError, match="Failed to find IP via MQTT"):
+            get_esphome_device_ip(_discovery_config(), timeout=0)
+
+    assert "Ignoring unparsable discovery payload" in caplog.text
+
+
+def test_get_esphome_device_ip_broker_disconnect_fails_fast(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A broker-initiated disconnect aborts the wait instead of timing out."""
+    client = MagicMock()
+
+    with patch("esphome.mqtt.prepare", return_value=client):
+
+        def drop_connection(*args, **kwargs):
+            client.on_disconnect(client, None, 5)
+
+        client.loop_start.side_effect = drop_connection
+
+        start = time.monotonic()
+        with pytest.raises(EsphomeError, match="Failed to find IP via MQTT"):
+            get_esphome_device_ip(_discovery_config(), timeout=5)
+
+    assert time.monotonic() - start < 1
+    assert "Disconnected from MQTT broker (5)" in caplog.text
+
+
+def test_get_esphome_device_ip_sends_discovery_ping() -> None:
+    """Connecting publishes the discovery ping for the device."""
+    client = MagicMock()
+
+    with patch("esphome.mqtt.prepare", return_value=client) as mock_prepare:
+
+        def connect_then_answer(*args, **kwargs):
+            on_connect = mock_prepare.call_args.args[3]
+            on_connect(client, None, None, 0)
+            msg = MagicMock()
+            msg.payload = json.dumps({"name": "test-device", "ip": "10.0.0.5"}).encode()
+            mock_prepare.call_args.args[2](client, None, msg)
+
+        client.loop_start.side_effect = connect_then_answer
+
+        result = get_esphome_device_ip(_discovery_config())
+
+    assert result == ["10.0.0.5"]
+    client.publish.assert_called_once_with(
+        "esphome/ping/test-device", None, retain=False
+    )
+
+
+def test_get_esphome_device_ip_disconnect_error_does_not_mask_result(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A cleanup failure must not replace the discovery result."""
+    client = MagicMock()
+    # First disconnect (from on_message) succeeds; the finally's fails
+    client.disconnect.side_effect = [None, OSError("socket already closed")]
+
+    with patch("esphome.mqtt.prepare", return_value=client) as mock_prepare:
+        _deliver_on_loop_start(
+            mock_prepare,
+            client,
+            json.dumps({"name": "test-device", "ip": "10.0.0.5"}).encode(),
+        )
+
+        result = get_esphome_device_ip(_discovery_config())
+
+    assert result == ["10.0.0.5"]
+    client.loop_stop.assert_called_once_with()
+
+
+def test_get_esphome_device_ip_invalid_address_values_skipped(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Non-string or non-printable ip values are skipped, valid ones kept."""
+    client = MagicMock()
+
+    with patch("esphome.mqtt.prepare", return_value=client) as mock_prepare:
+        _deliver_on_loop_start(
+            mock_prepare,
+            client,
+            json.dumps(
+                {
+                    "name": "test-device",
+                    "ip": 1234,
+                    "ip1": "x\n[00:00:00][I][forged] fake line",
+                    "ip2": "  10.0.0.5  ",
+                }
+            ).encode(),
+        )
+
+        result = get_esphome_device_ip(_discovery_config())
+
+    assert result == ["10.0.0.5"]
+    assert caplog.text.count("Ignoring invalid address in discovery answer") == 2
+    assert "forged" not in "".join(
+        r.getMessage() for r in caplog.records if "Found IP" in r.getMessage()
+    )