[core] Don't block logs startup on MQTT IP discovery when addresses are known (#18313)

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,38 +644,42 @@ 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(
+                mqtt_ips = _mqtt_get_ip_or_warn(
                     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
+                    addr for addr in mqtt_ips if addr not in network_devices
                 )
+                mqtt_resolved = True
+            continue
+
+        _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,
         )
-                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)
-
-    return network_devices
+        return []
 
 
 def run_miniterm(config: ConfigType, port: str, args) -> int:
@@ -1438,16 +1482,36 @@ 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)
-    ):
+    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),
+                mqtt_resolver=mqtt_resolver,
             )
 
     if port_type in (PortType.NETWORK, PortType.MQTT) and has_mqtt_logging():

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,10 +146,37 @@ 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:
+        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()
 
 
@@ -109,9 +184,15 @@ 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/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,32 +187,58 @@ 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)
 
+            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:
+            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):
@@ -215,22 +246,54 @@ def get_esphome_device_ip(
         _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
 
+    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:
+                if dev_ip is not None or failed:
+                    break
+                if stop_event.wait(0.250):
+                    stopped = True
                     break
                 timeout -= 0.250
-        time.sleep(0.250)
-    mqtt_client.loop_stop()
+    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
 
 

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_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()
+    )