[api] Move runtime log client out of the component package (#18043)

esphome/__main__.py

@@ -1429,7 +1429,7 @@ def show_logs(config: ConfigType, args: ArgsProtocol, devices: list[str]) -> int
     if has_api() and (
         network_devices := _resolve_network_devices(devices, config, args)
     ):
-        from esphome.components.api.client import run_logs
+        from esphome.api_client import run_logs
 
         return run_logs(
             config,

esphome/api_client.py

@@ -0,0 +1,173 @@
+from __future__ import annotations
+
+import asyncio
+from contextlib import suppress
+from datetime import datetime
+import importlib
+import logging
+from typing import TYPE_CHECKING, Any
+import warnings
+
+# Suppress protobuf version warnings
+with warnings.catch_warnings():
+    warnings.filterwarnings(
+        "ignore", category=UserWarning, message=".*Protobuf gencode version.*"
+    )
+    from aioesphomeapi import APIClient, parse_log_message
+    from aioesphomeapi.log_runner import async_run
+
+from esphome.const import CONF_ENCRYPTION, CONF_KEY, CONF_PORT, __version__
+from esphome.core import CORE
+from esphome.util import safe_print
+
+if TYPE_CHECKING:
+    from aioesphomeapi.api_pb2 import (
+        SubscribeLogsResponse,  # pylint: disable=no-name-in-module
+    )
+
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class _LogLineProcessor:
+    """Feeds incoming log lines to the stack-trace decoder.
+
+    Two responsibilities beyond just calling the decoder:
+    1. Catch everything the decoder can raise. aioesphomeapi isolates
+       exceptions raised by log handlers, so an escaping one no longer
+       kills the session, but it does log a full traceback per line. A
+       crash dump carries a PC line plus one per backtrace frame, so the
+       tracebacks bury the dump the user is trying to read. Decoding is a
+       diagnostic nicety; nothing it raises is worth that noise.
+    2. Disable decoding after the first failure. _decode_pc shells out to
+       the toolchain to resolve addr2line, which is expensive; a single
+       crash dump can contain many PC/BT lines and we don't want to retry
+       the failing subprocess for each one. This only works if every
+       failure is caught, which is why 1 is not narrowed to EsphomeError.
+    """
+
+    def __init__(self, config: dict[str, Any], platform_handler: Any | None) -> None:
+        self._config = config
+        self._platform_handler = platform_handler
+        self._decode_enabled = platform_handler is not None
+        self.backtrace_state = False
+
+    def process_line(self, raw_line: str) -> None:
+        if not self._decode_enabled:
+            return
+        try:
+            self.backtrace_state = self._platform_handler(
+                self._config, raw_line, self.backtrace_state
+            )
+        except Exception as exc:  # noqa: BLE001  # pylint: disable=broad-except
+            self._decode_enabled = False
+            self.backtrace_state = False
+            # _run_idedata raises EsphomeError with no message; fall back
+            # to a generic explanation when str(exc) is empty.
+            detail = str(exc) or "build artifacts not found locally"
+            _LOGGER.debug("Stack-trace decoding failed", exc_info=True)
+            _LOGGER.warning(
+                "Crash trace decoding unavailable: %s. "
+                "Run 'esphome compile' for this device to enable PC decoding.",
+                detail,
+            )
+
+
+async def async_run_logs(
+    config: dict[str, Any],
+    addresses: list[str],
+    subscribe_states: bool = True,
+) -> None:
+    """Run the logs command in the event loop."""
+    conf = config["api"]
+    name = config["esphome"]["name"]
+    port: int = int(conf[CONF_PORT])
+    noise_psk: str | None = None
+    if (encryption := conf.get(CONF_ENCRYPTION)) and (key := encryption.get(CONF_KEY)):
+        noise_psk = key
+
+    _LOGGER.info(
+        "Starting log output from %s using esphome API", " or ".join(addresses)
+    )
+
+    cli = APIClient(
+        addresses[0],  # Primary address for compatibility
+        port,
+        "",  # Password auth removed in 2026.1.0
+        client_info=f"ESPHome Logs {__version__}",
+        noise_psk=noise_psk,
+        addresses=addresses,  # Pass all addresses for automatic retry
+        provide_time=False,
+    )
+
+    # Try platform-specific stacktrace handler first, fall back to generic
+    platform_process_stacktrace = None
+    try:
+        module = importlib.import_module("esphome.components." + CORE.target_platform)
+        platform_process_stacktrace = module.process_stacktrace
+    except (AttributeError, ImportError):
+        # Distinguish "platform has no analyzer" from a genuinely broken
+        # platform package when debugging.
+        _LOGGER.debug("Stacktrace analyzer lookup failed", exc_info=True)
+        _LOGGER.info(
+            'Stacktrace analysis is unavailable: no compatible analyzer found for target platform "%s".',
+            CORE.target_platform,
+        )
+
+    processor = _LogLineProcessor(config, platform_process_stacktrace)
+
+    def on_log(msg: SubscribeLogsResponse) -> None:
+        """Handle a new log message."""
+        time_ = datetime.now().astimezone()
+        message: bytes = msg.message
+        text = message.decode("utf8", "backslashreplace")
+        nanoseconds = time_.microsecond // 1000
+        timestamp = (
+            f"[{time_.hour:02}:{time_.minute:02}:{time_.second:02}.{nanoseconds:03}]"
+        )
+        for parsed_msg in parse_log_message(text, timestamp):
+            # safe_print handles the dashboard \033 escaping and falls back
+            # to backslashreplace encoding on stdouts that can't represent
+            # the wifi signal-bar block characters (Windows redirected
+            # cp1252 pipe).
+            safe_print(parsed_msg)
+        for raw_line in text.splitlines():
+            processor.process_line(raw_line)
+
+    # Safe to fall back to plaintext here only for this diagnostics use
+    # case: the stream is one-way from device to client, and this code
+    # never accepts commands or acts on any message the device sends.
+    # An on-path attacker could still both inject fabricated log lines
+    # and passively read the device's log output (and any state data
+    # delivered when subscribe_states is enabled), so this does lose
+    # confidentiality as well as authentication/integrity. That tradeoff
+    # is acceptable for operator-visible logs, which aioesphomeapi also
+    # warns may come from an unverified device. Never mirror this opt-in
+    # for any connection that sends data to the device or uses Home
+    # Assistant actions.
+    stop = await async_run(
+        cli,
+        on_log,
+        name=name,
+        subscribe_states=subscribe_states,
+        allow_plaintext_fallback=True,
+        # 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,
+    )
+    try:
+        await asyncio.Event().wait()
+    finally:
+        await stop()
+
+
+def run_logs(
+    config: dict[str, Any],
+    addresses: list[str],
+    subscribe_states: bool = True,
+) -> None:
+    """Run the logs command."""
+    with suppress(KeyboardInterrupt):
+        asyncio.run(
+            async_run_logs(config, addresses, subscribe_states=subscribe_states)
+        )

esphome/components/api/__init__.py

@@ -13,6 +13,7 @@ from esphome.const import (
     CONF_CAPTURE_RESPONSE,
     CONF_DATA,
     CONF_DATA_TEMPLATE,
+    CONF_ENCRYPTION,
     CONF_EVENT,
     CONF_ID,
     CONF_KEY,
@@ -102,7 +103,6 @@ SERVICE_ARG_FALLBACK_TYPES: dict[str, MockObj] = {
         for name, t in _SERVICE_ARG_SCALAR_TYPES.items()
     },
 }
-CONF_ENCRYPTION = "encryption"
 CONF_BATCH_DELAY = "batch_delay"
 CONF_CUSTOM_SERVICES = "custom_services"
 CONF_HOMEASSISTANT_SERVICES = "homeassistant_services"

esphome/components/api/client.py

@@ -1,177 +1,10 @@
-from __future__ import annotations
+"""Backward-compatibility shim; the log client lives in esphome.api_client.
 
-import asyncio
+Importing this module executes the whole api component package, which pulls
-from datetime import datetime
+in the validation stack. CLI code paths should import esphome.api_client
-import importlib
+directly so the logs fast path stays light.
-import logging
+"""
-from typing import TYPE_CHECKING, Any
-import warnings
 
-# Suppress protobuf version warnings
+from esphome.api_client import async_run_logs, run_logs
-with warnings.catch_warnings():
-    warnings.filterwarnings(
-        "ignore", category=UserWarning, message=".*Protobuf gencode version.*"
-    )
-    from aioesphomeapi import APIClient, parse_log_message
-    from aioesphomeapi.log_runner import async_run
 
-import contextlib
+__all__ = ["async_run_logs", "run_logs"]
-
-from esphome.const import CONF_KEY, CONF_PORT, __version__
-from esphome.core import CORE
-from esphome.util import safe_print
-
-from . import CONF_ENCRYPTION
-
-if TYPE_CHECKING:
-    from aioesphomeapi.api_pb2 import (
-        SubscribeLogsResponse,  # pylint: disable=no-name-in-module
-    )
-
-
-_LOGGER = logging.getLogger(__name__)
-
-
-class _LogLineProcessor:
-    """Feeds incoming log lines to the stack-trace decoder.
-
-    Two responsibilities beyond just calling the decoder:
-    1. Catch everything the decoder can raise. aioesphomeapi isolates
-       exceptions raised by log handlers, so an escaping one no longer
-       kills the session, but it does log a full traceback per line. A
-       crash dump carries a PC line plus one per backtrace frame, so the
-       tracebacks bury the dump the user is trying to read. Decoding is a
-       diagnostic nicety; nothing it raises is worth that noise.
-    2. Disable decoding after the first failure. _decode_pc shells out to
-       the toolchain to resolve addr2line, which is expensive; a single
-       crash dump can contain many PC/BT lines and we don't want to retry
-       the failing subprocess for each one. This only works if every
-       failure is caught, which is why 1 is not narrowed to EsphomeError.
-    """
-
-    def __init__(self, config: dict[str, Any], platform_handler: Any | None) -> None:
-        self._config = config
-        self._platform_handler = platform_handler
-        self._decode_enabled = True
-        self.backtrace_state = False
-
-    def process_line(self, raw_line: str) -> None:
-        if not self._decode_enabled:
-            return
-        try:
-            if self._platform_handler is not None:
-                self.backtrace_state = self._platform_handler(
-                    self._config, raw_line, self.backtrace_state
-                )
-        except Exception as exc:  # noqa: BLE001  # pylint: disable=broad-except
-            self._decode_enabled = False
-            self.backtrace_state = False
-            # _run_idedata raises EsphomeError with no message; fall back
-            # to a generic explanation when str(exc) is empty.
-            detail = str(exc) or "build artifacts not found locally"
-            _LOGGER.debug("Stack-trace decoding failed", exc_info=True)
-            _LOGGER.warning(
-                "Crash trace decoding unavailable: %s. "
-                "Run 'esphome compile' for this device to enable PC decoding.",
-                detail,
-            )
-
-
-async def async_run_logs(
-    config: dict[str, Any],
-    addresses: list[str],
-    subscribe_states: bool = True,
-) -> None:
-    """Run the logs command in the event loop."""
-    conf = config["api"]
-    name = config["esphome"]["name"]
-    port: int = int(conf[CONF_PORT])
-    noise_psk: str | None = None
-    if (encryption := conf.get(CONF_ENCRYPTION)) and (key := encryption.get(CONF_KEY)):
-        noise_psk = key
-
-    if len(addresses) == 1:
-        _LOGGER.info("Starting log output from %s using esphome API", addresses[0])
-    else:
-        _LOGGER.info(
-            "Starting log output from %s using esphome API", " or ".join(addresses)
-        )
-
-    cli = APIClient(
-        addresses[0],  # Primary address for compatibility
-        port,
-        "",  # Password auth removed in 2026.1.0
-        client_info=f"ESPHome Logs {__version__}",
-        noise_psk=noise_psk,
-        addresses=addresses,  # Pass all addresses for automatic retry
-        provide_time=False,
-    )
-
-    # Try platform-specific stacktrace handler first, fall back to generic
-    platform_process_stacktrace = None
-    try:
-        module = importlib.import_module("esphome.components." + CORE.target_platform)
-        platform_process_stacktrace = module.process_stacktrace
-    except (AttributeError, ImportError):
-        _LOGGER.info(
-            'Stacktrace analysis is unavailable: no compatible analyzer found for target platform "%s".',
-            CORE.target_platform,
-        )
-
-    processor = _LogLineProcessor(config, platform_process_stacktrace)
-
-    def on_log(msg: SubscribeLogsResponse) -> None:
-        """Handle a new log message."""
-        time_ = datetime.now().astimezone()
-        message: bytes = msg.message
-        text = message.decode("utf8", "backslashreplace")
-        nanoseconds = time_.microsecond // 1000
-        timestamp = (
-            f"[{time_.hour:02}:{time_.minute:02}:{time_.second:02}.{nanoseconds:03}]"
-        )
-        for parsed_msg in parse_log_message(text, timestamp):
-            # safe_print handles the dashboard \033 escaping and falls back
-            # to backslashreplace encoding on stdouts that can't represent
-            # the wifi signal-bar block characters (Windows redirected
-            # cp1252 pipe).
-            safe_print(parsed_msg)
-        for raw_line in text.splitlines():
-            processor.process_line(raw_line)
-
-    # Safe to fall back to plaintext here only for this diagnostics use
-    # case: the stream is one-way from device to client, and this code
-    # never accepts commands or acts on any message the device sends.
-    # An on-path attacker could still both inject fabricated log lines
-    # and passively read the device's log output (and any state data
-    # delivered when subscribe_states is enabled), so this does lose
-    # confidentiality as well as authentication/integrity. That tradeoff
-    # is acceptable for operator-visible logs, which aioesphomeapi also
-    # warns may come from an unverified device. Never mirror this opt-in
-    # for any connection that sends data to the device or uses Home
-    # Assistant actions.
-    stop = await async_run(
-        cli,
-        on_log,
-        name=name,
-        subscribe_states=subscribe_states,
-        allow_plaintext_fallback=True,
-        # 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,
-    )
-    try:
-        await asyncio.Event().wait()
-    finally:
-        await stop()
-
-
-def run_logs(
-    config: dict[str, Any],
-    addresses: list[str],
-    subscribe_states: bool = True,
-) -> None:
-    """Run the logs command."""
-    with contextlib.suppress(KeyboardInterrupt):
-        asyncio.run(
-            async_run_logs(config, addresses, subscribe_states=subscribe_states)
-        )

esphome/components/packet_transport/__init__.py

@@ -4,12 +4,12 @@ import hashlib
 import logging
 
 import esphome.codegen as cg
-from esphome.components.api import CONF_ENCRYPTION
 from esphome.components.binary_sensor import BinarySensor
 from esphome.components.sensor import Sensor
 import esphome.config_validation as cv
 from esphome.const import (
     CONF_BINARY_SENSORS,
+    CONF_ENCRYPTION,
     CONF_ID,
     CONF_INTERNAL,
     CONF_KEY,

esphome/components/udp/packet_transport/__init__.py

@@ -1,12 +1,11 @@
 import esphome.codegen as cg
-from esphome.components.api import CONF_ENCRYPTION
 from esphome.components.packet_transport import (
     CONF_PING_PONG_ENABLE,
     PacketTransport,
     new_packet_transport,
     transport_schema,
 )
-from esphome.const import CONF_BINARY_SENSORS, CONF_SENSORS
+from esphome.const import CONF_BINARY_SENSORS, CONF_ENCRYPTION, CONF_SENSORS
 from esphome.cpp_types import PollingComponent
 
 from .. import UDP_SCHEMA, register_udp_client, udp_ns

esphome/const.py

@@ -384,6 +384,7 @@ CONF_ENABLE_PIN = "enable_pin"
 CONF_ENABLE_PRIVATE_NETWORK_ACCESS = "enable_private_network_access"
 CONF_ENABLE_RRM = "enable_rrm"
 CONF_ENABLE_TIME = "enable_time"
+CONF_ENCRYPTION = "encryption"
 CONF_ENERGY = "energy"
 CONF_ENTITY_CATEGORY = "entity_category"
 CONF_ENTITY_ID = "entity_id"

script/ci-custom.py

@@ -557,7 +557,7 @@ def lint_constants_usage():
 # Maximum allowed CONF_ constants in esphome/const.py.
 # This file is frozen — new constants go in esphome/components/const/__init__.py.
 # Decrease this number when constants are moved out of const.py.
-CONST_PY_MAX_CONF = 1015
+CONST_PY_MAX_CONF = 1016
 
 
 @lint_content_check(include=["esphome/const.py"])

tests/unit_tests/test_api_client.py

@@ -1,17 +1,34 @@
-"""Tests for esphome.components.api.client."""
+"""Tests for esphome.api_client."""
 
 from __future__ import annotations
 
-from unittest.mock import AsyncMock, patch
+import asyncio
+from unittest.mock import AsyncMock, Mock, patch
 
 import pytest
 
+from esphome import api_client
 from esphome.components import esp32
-from esphome.components.api import client as api_client
+from esphome.const import (
-from esphome.const import CONF_PORT, KEY_CORE, KEY_TARGET_PLATFORM
+    CONF_ENCRYPTION,
+    CONF_KEY,
+    CONF_PORT,
+    KEY_CORE,
+    KEY_TARGET_PLATFORM,
+)
 from esphome.core import CORE, EsphomeError
 
 
+def test_component_shim_reexports_runtime_client() -> None:
+    """The old import paths must keep working for external code."""
+    from esphome.components import api
+    from esphome.components.api import client as shim
+
+    assert shim.run_logs is api_client.run_logs
+    assert shim.async_run_logs is api_client.async_run_logs
+    assert api.CONF_ENCRYPTION is CONF_ENCRYPTION
+
+
 def test_decoder_swallows_esphome_error() -> None:
     """A failing stack-trace decode must not propagate.
 
@@ -166,3 +183,73 @@ 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_full_flow(caplog) -> None:
+    """Drive async_run_logs end to end with a fake connection.
+
+    Covers the encryption key extraction, the multi-address banner, the
+    missing-stacktrace-analyzer fallback, the on_log handler, and the
+    stop() cleanup in the finally block.
+    """
+    caplog.set_level("INFO", logger="esphome.api_client")
+    CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: "host"}
+    config = {
+        "esphome": {"name": "test"},
+        "api": {CONF_PORT: 6053, CONF_ENCRYPTION: {CONF_KEY: "psk123"}},
+    }
+
+    stop = AsyncMock()
+    run_started = asyncio.Event()
+
+    async def fake_async_run(*args, **kwargs):
+        run_started.set()
+        return stop
+
+    mock_run = AsyncMock(side_effect=fake_async_run)
+    printed: list[str] = []
+
+    with (
+        patch.object(api_client, "async_run", mock_run),
+        patch.object(api_client, "APIClient") as mock_client,
+        patch.object(api_client, "safe_print", printed.append),
+    ):
+        task = asyncio.get_running_loop().create_task(
+            api_client.async_run_logs(config, ["1.2.3.4", "5.6.7.8"])
+        )
+        # Let the task run up to the forever-wait; the timeout fails the
+        # test instead of hanging it if the task dies early.
+        async with asyncio.timeout(1):
+            await run_started.wait()
+        on_log = mock_run.call_args.args[1]
+        on_log(Mock(message=b"[I][main:001] hello world\nPC: 0x40104960"))
+        # Cancellation is the real termination path; stop() must still run.
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+
+    # Both addresses reach APIClient, along with the noise key.
+    assert mock_client.call_args.kwargs["noise_psk"] == "psk123"
+    assert mock_client.call_args.kwargs["addresses"] == ["1.2.3.4", "5.6.7.8"]
+    assert "1.2.3.4 or 5.6.7.8" in caplog.text
+    # host has no stacktrace analyzer; the fallback message is logged.
+    assert "Stacktrace analysis is unavailable" in caplog.text
+    # The log message was printed with a timestamp prefix.
+    assert any("hello world" in line for line in printed)
+    # stop() ran in the finally block despite the cancellation.
+    stop.assert_awaited_once()
+
+
+def test_run_logs_suppresses_keyboard_interrupt() -> None:
+    """Ctrl-C during log streaming exits cleanly instead of tracebacking."""
+    with patch.object(
+        api_client,
+        "async_run_logs",
+        AsyncMock(side_effect=KeyboardInterrupt),
+    ) as mock_run:
+        api_client.run_logs(
+            {"esphome": {"name": "test"}}, ["1.2.3.4"], subscribe_states=False
+        )
+
+    assert mock_run.call_args.kwargs["subscribe_states"] is False

tests/unit_tests/test_lazy_imports.py

@@ -31,11 +31,16 @@ HEAVY_MODULES = (
 )
 
 
-def test_main_module_does_not_import_heavy_modules() -> None:
+def _leaked_heavy_modules(module: str) -> str:
-    """A bare ``import esphome.__main__`` must not drag in validation/codegen."""
+    """Import ``module`` in a subprocess and report the heavy modules it pulled.
+
+    Any ``esphome.components.*`` package counts as heavy: executing a
+    component package drags in codegen/validation machinery by design.
+    """
     check = (
-        "import sys; import esphome.__main__; "
+        f"import sys; import {module}; "
         f"leaked = [m for m in {HEAVY_MODULES!r} if m in sys.modules]; "
+        "leaked += [m for m in sys.modules if m.startswith('esphome.components.')]; "
         "print(','.join(leaked))"
     )
     result = subprocess.run(
@@ -44,10 +49,30 @@ def test_main_module_does_not_import_heavy_modules() -> None:
         text=True,
         check=True,
     )
-    leaked = result.stdout.strip()
+    return result.stdout.strip()
+
+
+def test_main_module_does_not_import_heavy_modules() -> None:
+    """A bare ``import esphome.__main__`` must not drag in validation/codegen."""
+    leaked = _leaked_heavy_modules("esphome.__main__")
     assert not leaked, (
         f"esphome.__main__ imports heavy modules at top level: {leaked}. "
         "Import them lazily inside the command that needs them instead; "
         "every esphome invocation (including each parallel dashboard "
         "upload subprocess) pays for top-level imports."
     )
+
+
+def test_api_client_does_not_import_heavy_modules() -> None:
+    """``esphome.api_client`` is on the logs fast path and must stay light.
+
+    Importing it must not execute any component package (the api package
+    pulls the whole validation stack: logger, esp32, writer, config,
+    jinja2, voluptuous).
+    """
+    leaked = _leaked_heavy_modules("esphome.api_client")
+    assert not leaked, (
+        f"esphome.api_client imports heavy modules at top level: {leaked}. "
+        "The logs fast path skips validation; importing the validation "
+        "stack anyway defeats the validated-config cache."
+    )

tests/unit_tests/test_main.py

@@ -2918,7 +2918,7 @@ def test_show_logs_no_logger() -> None:
         show_logs(CORE.config, args, devices)
 
 
-@patch("esphome.components.api.client.run_logs")
+@patch("esphome.api_client.run_logs")
 def test_show_logs_api(
     mock_run_logs: Mock,
 ) -> None:
@@ -2944,7 +2944,7 @@ def test_show_logs_api(
     )
 
 
-@patch("esphome.components.api.client.run_logs")
+@patch("esphome.api_client.run_logs")
 def test_show_logs_api_no_states(
     mock_run_logs: Mock,
 ) -> None:
@@ -2971,7 +2971,7 @@ def test_show_logs_api_no_states(
     )
 
 
-@patch("esphome.components.api.client.run_logs")
+@patch("esphome.api_client.run_logs")
 def test_show_logs_api_with_fqdn_mdns_disabled(
     mock_run_logs: Mock,
 ) -> None:
@@ -2998,7 +2998,7 @@ def test_show_logs_api_with_fqdn_mdns_disabled(
     )
 
 
-@patch("esphome.components.api.client.run_logs")
+@patch("esphome.api_client.run_logs")
 def test_show_logs_api_with_mqtt_fallback(
     mock_run_logs: Mock,
     mock_mqtt_get_ip: Mock,
@@ -4974,7 +4974,7 @@ def test_upload_program_ota_mqttip_deduplication(
     assert "192.168.1.100" in call_args[0]
 
 
-@patch("esphome.components.api.client.run_logs")
+@patch("esphome.api_client.run_logs")
 def test_show_logs_api_static_ip_with_mqttip(
     mock_run_logs: Mock,
     mock_mqtt_get_ip: Mock,
@@ -5013,7 +5013,7 @@ def test_show_logs_api_static_ip_with_mqttip(
     )
 
 
-@patch("esphome.components.api.client.run_logs")
+@patch("esphome.api_client.run_logs")
 def test_show_logs_api_multiple_mqttip_resolves_once(
     mock_run_logs: Mock,
     mock_mqtt_get_ip: Mock,
@@ -5096,7 +5096,7 @@ def test_upload_program_ota_mqtt_timeout_fallback(
     )
 
 
-@patch("esphome.components.api.client.run_logs")
+@patch("esphome.api_client.run_logs")
 def test_show_logs_api_mqtt_timeout_fallback(
     mock_run_logs: Mock,
     mock_mqtt_get_ip: Mock,
@@ -6468,7 +6468,7 @@ def test_should_subscribe_states_no_flag_overrides_env() -> None:
         assert _should_subscribe_states(args) is False
 
 
-@patch("esphome.components.api.client.run_logs")
+@patch("esphome.api_client.run_logs")
 def test_command_run_passes_no_states_to_show_logs(
     mock_run_logs: Mock,
 ) -> None:
@@ -6506,7 +6506,7 @@ def test_command_run_passes_no_states_to_show_logs(
     )
 
 
-@patch("esphome.components.api.client.run_logs")
+@patch("esphome.api_client.run_logs")
 def test_command_run_defaults_subscribe_states_true(
     mock_run_logs: Mock,
 ) -> None: