Merge remote-tracking branch 'origin/dev' into mqtt-async-client-topic-length

2026-10-18 20:22:49 +00:00 · 2026-10-14 11:07:01 -10:00
parent f6830cc719 ac74796263
commit 07a491608c
22 changed files with 406 additions and 44 deletions
@@ -1,4 +1,7 @@
 #include "api_connection.h"
+#if defined(USE_LOGGER) && defined(USE_CRASH_HANDLER)
+#include "esphome/components/logger/logger.h"
+#endif
 #ifdef USE_API
 #include "api_connection_buffer.h"  // for the APIServer-dependent APIConnection inlines
 #ifdef USE_API_NOISE
@@ -451,6 +454,15 @@ bool APIConnection::send_disconnect_response_() {
  DisconnectResponse resp;
  return this->send_message(resp);
 }
+void APIConnection::on_subscribe_logs_request(const SubscribeLogsRequest &msg) {
+  this->flags_.log_subscription = msg.level;
+  if (msg.dump_config)
+    App.schedule_dump_config();
+#if defined(USE_LOGGER) && defined(USE_CRASH_HANDLER)
+  logger::log_crash_to_client();
+#endif
+}
+
 void APIConnection::on_disconnect_response() {
  // Don't close socket here, let APIServer::loop() do it
  // so getpeername() still works for the disconnect trigger
@@ -15,15 +15,6 @@
 #include "subscribe_state.h"
 #include "esphome/core/application.h"
 #include "esphome/core/component.h"
-#ifdef USE_ESP32_CRASH_HANDLER
-#include "esphome/components/esp32/crash_handler.h"
-#endif
-#ifdef USE_RP2_CRASH_HANDLER
-#include "esphome/components/rp2/crash_handler.h"
-#endif
-#ifdef USE_ESP8266_CRASH_HANDLER
-#include "esphome/components/esp8266/crash_handler.h"
-#endif
 #include "esphome/core/entity_base.h"
 #include "esphome/core/log.h"
 #include "esphome/core/string_ref.h"
@@ -298,21 +289,7 @@ class APIConnection final : public APIServerConnectionBase {
      this->begin_iterator_(ActiveIterator::INITIAL_STATE);
    }
  }
-  void on_subscribe_logs_request(const SubscribeLogsRequest &msg) {
-    this->flags_.log_subscription = msg.level;
-    if (msg.dump_config)
-      App.schedule_dump_config();
-#ifdef USE_ESP32_CRASH_HANDLER
-    esp32::crash_handler_log();
-    esp32::crash_handler_clear();
-#endif
-#ifdef USE_RP2_CRASH_HANDLER
-    rp2::crash_handler_log();
-#endif
-#ifdef USE_ESP8266_CRASH_HANDLER
-    esp8266::crash_handler_log();
-#endif
-  }
+  void on_subscribe_logs_request(const SubscribeLogsRequest &msg);
 #ifdef USE_API_HOMEASSISTANT_SERVICES
  void on_subscribe_homeassistant_services_request() { this->flags_.service_call_subscription = true; }
 #endif
@@ -2920,6 +2920,7 @@ async def _finalize_arduino_aware_flags():
    if not arduino_linked:
        cg.add_build_flag("-Wl,--wrap=esp_panic_handler")
        cg.add_define("USE_ESP32_CRASH_HANDLER")
+        cg.add_define("USE_CRASH_HANDLER")
    elif not CORE.using_arduino:
        cg.add_build_flag("-DUSE_ARDUINO")

@@ -161,7 +161,7 @@ static bool s_crash_data_valid = false;  // NOLINT(cppcoreguidelines-avoid-non-c

 namespace esphome::esp32 {

-ESPHOME_LOG_TAG(TAG, "esp32.crash");
+ESPHOME_LOG_TAG(TAG, CRASH_LOG_TAG);

 // RAM copy of the build timestamp. The generated constant lives in flash,
 // which the panic handler must not read (cache may be disabled during
@@ -175,7 +175,8 @@ static uint32_t s_current_build_time = static_cast<uint32_t>(ESPHOME_BUILD_TIME)

 // Validate the NOINIT record. Runs on every has_data() call; re-running is
 // harmless and the magic is left alone so the record survives an OTA
-// rollback reboot, crash_handler_clear() drops it once an API client has it.
+// rollback reboot, crash_handler_clear() drops it once a log client has it
+// (logger::log_crash_to_client(), for the API and MQTT).
 static void read_crash_data() {
  if (s_raw_crash_data.magic == CRASH_MAGIC && s_raw_crash_data.version == CRASH_DATA_VERSION) {
    s_crash_data_valid = true;
@@ -4,6 +4,9 @@

 namespace esphome::esp32 {

+/// The tag the crash record is logged under, so a client can check the level in force for it
+inline constexpr const char *CRASH_LOG_TAG = "esp32.crash";
+
 /// Log crash data if a crash was detected on previous boot.
 void crash_handler_log();

@@ -420,6 +420,7 @@ async def to_code(config: ConfigType) -> None:
    cg.add_define("ESPHOME_VARIANT", "ESP8266")
    cg.add_define(ThreadModel.SINGLE)
    cg.add_define("USE_ESP8266_CRASH_HANDLER")
+    cg.add_define("USE_CRASH_HANDLER")

    enable_scanf_float = config.get(CONF_ENABLE_SCANF_FLOAT)
    if enable_scanf_float is None and lambdas_use_scanf_float(CORE.config):
@@ -142,6 +142,8 @@ void Logger::log_vprintf_(uint8_t level, const char *tag, int line, const __Flas
 }
 #endif  // USE_ESP8266

+uint8_t Logger::get_log_level(const char *tag) { return this->level_for(tag); }
+
 inline uint8_t Logger::level_for(const char *tag) {
 #ifdef USE_LOGGER_RUNTIME_TAG_LEVELS
  // Skip the lookup when no per-tag levels are set
@@ -178,6 +178,8 @@ class Logger final : public Component {
  void set_log_level(const char *tag, uint8_t log_level);
 #endif
  uint8_t get_log_level() { return this->current_level_; }
+  /// The level in force for a tag: its own when one is set, else the default
+  uint8_t get_log_level(const char *tag);

  // ========== INTERNAL METHODS ==========
  // (In most use cases you won't need these)
@@ -511,6 +513,14 @@ class LoggerMessageTrigger final : public Trigger<uint8_t, const char *, const c
 #endif
 };

+#ifdef USE_CRASH_HANDLER
+/// Whether a crash record from before this boot is waiting to be reported
+bool crash_log_pending();
+/// Log the previous crash for a client that just subscribed to the logs. On the ESP32 the record
+/// is kept until a client has seen it, so this call also clears it there.
+void log_crash_to_client();
+#endif
+
 }  // namespace esphome::logger

 // Platform-specific inline implementations of write_msg_()
@@ -162,5 +162,16 @@ const LogString *Logger::get_uart_selection_() {
  }
 }

+#ifdef USE_CRASH_HANDLER
+bool crash_log_pending() { return esp32::crash_handler_has_data(); }
+void log_crash_to_client() {
+  // A level that drops the lines would throw the record away unseen; keep it for a client that can see it
+  if (global_logger->get_log_level(esp32::CRASH_LOG_TAG) < ESPHOME_LOG_LEVEL_ERROR)
+    return;
+  esp32::crash_handler_log();
+  esp32::crash_handler_clear();
+}
+#endif
+
 }  // namespace esphome::logger
 #endif
@@ -48,6 +48,11 @@ const LogString *Logger::get_uart_selection_() {
 #endif
 }

+#ifdef USE_CRASH_HANDLER
+bool crash_log_pending() { return esp8266::crash_handler_has_data(); }
+void log_crash_to_client() { esp8266::crash_handler_log(); }
+#endif
+
 }  // namespace esphome::logger

 #if !defined(USE_ESP8266_LOGGER_SERIAL) && !defined(USE_ESP8266_LOGGER_SERIAL1)
@@ -49,5 +49,10 @@ const LogString *Logger::get_uart_selection_() {
  }
 }

+#ifdef USE_CRASH_HANDLER
+bool crash_log_pending() { return rp2::crash_handler_has_data(); }
+void log_crash_to_client() { rp2::crash_handler_log(); }
+#endif
+
 }  // namespace esphome::logger
 #endif  // USE_RP2
@@ -50,6 +50,13 @@ MQTTClientComponent::MQTTClientComponent() {
 }

 // Connection
+// Room for "esphome/<kind>/<name>" with the longest kind, "discover"
+static constexpr size_t ESPHOME_TOPIC_BUFFER_SIZE = sizeof("esphome/discover/") + ESPHOME_DEVICE_NAME_MAX_LEN;
+
+static void format_esphome_topic(char (&topic)[ESPHOME_TOPIC_BUFFER_SIZE], const char *kind) {
+  buf_append_printf(topic, sizeof(topic), 0, "esphome/%s/%s", kind, App.get_name().c_str());
+}
+
 void MQTTClientComponent::setup() {
  this->mqtt_backend_.set_on_message(
      [this](const char *topic, const char *payload, size_t len, size_t index, size_t total) {
@@ -79,6 +86,13 @@ void MQTTClientComponent::setup() {
        this, [](void *self, uint8_t level, const char *tag, const char *message, size_t message_len) {
          static_cast<MQTTClientComponent *>(self)->on_log(level, tag, message, message_len);
        });
+#ifdef USE_MQTT_CRASH_LOG
+    // `esphome logs` announces itself here, like an API client's log subscription, and the record stays
+    // loggable until the next boot, so the offer does too. A log topic that drops errors would swallow the
+    // replay, so it is not offered. The topic is rebuilt from the device name when needed, so nothing is
+    // stored for it.
+    this->crash_log_offered_ = logger::crash_log_pending() && this->log_level_ >= ESPHOME_LOG_LEVEL_ERROR;
+#endif
  }
 #endif

@@ -88,10 +102,8 @@ void MQTTClientComponent::setup() {
        2);

    // Format topic on stack - subscribe() copies it
-    // "esphome/ping/" (13) + name (ESPHOME_DEVICE_NAME_MAX_LEN) + null (1)
-    constexpr size_t ping_topic_buffer_size = 13 + ESPHOME_DEVICE_NAME_MAX_LEN + 1;
-    char ping_topic[ping_topic_buffer_size];
-    buf_append_printf(ping_topic, sizeof(ping_topic), 0, "esphome/ping/%s", App.get_name().c_str());
+    char ping_topic[ESPHOME_TOPIC_BUFFER_SIZE];
+    format_esphome_topic(ping_topic, "ping");
    this->subscribe(
        ping_topic, [this](const std::string &topic, const std::string &payload) { this->send_device_info_(); }, 2);
  }
@@ -106,10 +118,8 @@ void MQTTClientComponent::send_device_info_() {
    return;
  }
  // Format topic on stack to avoid heap allocation
-  // "esphome/discover/" (17) + name (ESPHOME_DEVICE_NAME_MAX_LEN) + null (1)
-  constexpr size_t topic_buffer_size = 17 + ESPHOME_DEVICE_NAME_MAX_LEN + 1;
-  char topic[topic_buffer_size];
-  buf_append_printf(topic, sizeof(topic), 0, "esphome/discover/%s", App.get_name().c_str());
+  char topic[ESPHOME_TOPIC_BUFFER_SIZE];
+  format_esphome_topic(topic, "discover");

  // NOLINTBEGIN(clang-analyzer-cplusplus.NewDeleteLeaks) false positive with ArduinoJson
  this->publish_json(
@@ -366,6 +376,14 @@ void MQTTClientComponent::check_connected() {
  delay(100);  // NOLINT

  this->resubscribe_subscriptions_();
+#ifdef USE_MQTT_CRASH_LOG
+  // Best effort: a refused subscription is tried again at the next connection
+  if (this->crash_log_offered_) {
+    char topic[ESPHOME_TOPIC_BUFFER_SIZE];
+    format_esphome_topic(topic, "logs");
+    this->subscribe_(topic, 0);
+  }
+#endif
  this->send_device_info_();

  for (MQTTComponent *component : this->children_)
@@ -473,6 +491,19 @@ void MQTTClientComponent::resubscribe_subscriptions_() {
  }
 }

+#ifdef USE_MQTT_CRASH_LOG
+void MQTTClientComponent::handle_crash_log_request_(const std::string &topic, const std::string &payload) {
+  // Only the CLI's empty request counts: a log_topic pointing at this topic would otherwise replay the crash
+  // log for every line it logs
+  if (!this->crash_log_offered_ || !payload.empty())
+    return;
+  char logs_topic[ESPHOME_TOPIC_BUFFER_SIZE];
+  format_esphome_topic(logs_topic, "logs");
+  if (topic == logs_topic)
+    logger::log_crash_to_client();
+}
+#endif
+
 void MQTTClientComponent::add_subscription_(std::string &&topic, mqtt_subscription_callback_t callback, bool boxed,
                                            uint8_t qos) {
  auto &subscription = this->subscriptions_.emplace_back(MQTTSubscription{
@@ -680,6 +711,9 @@ void MQTTClientComponent::on_message(const std::string &topic, const std::string
  // DO NOT REMOVE THIS DEFER without understanding the above. It may appear to work
  // in simple tests but will cause crashes with complex automations.
  this->defer([this, topic, payload]() {
+#endif
+#ifdef USE_MQTT_CRASH_LOG
+    this->handle_crash_log_request_(topic, payload);
 #endif
    // Indexed with the count taken up front: a callback may subscribe, which can reallocate the
    // vector, and the new entry waits for the next message.
@@ -24,6 +24,11 @@

 #include <vector>

+// The crash log replay needs the logger, which mqtt does not depend on
+#if defined(USE_LOGGER) && defined(USE_CRASH_HANDLER)
+#define USE_MQTT_CRASH_LOG
+#endif
+
 namespace esphome::mqtt {

 /** Callback for MQTT events.
@@ -316,6 +321,9 @@ class MQTTClientComponent final : public Component {
    callback(topic, doc.as<JsonObject>());
  }
  void erase_removed_subscriptions_();
+#ifdef USE_MQTT_CRASH_LOG
+  void handle_crash_log_request_(const std::string &topic, const std::string &payload);
+#endif
  void add_subscription_(std::string &&topic, mqtt_subscription_callback_t callback, bool boxed, uint8_t qos);
  void resubscribe_subscriptions_();

@@ -359,6 +367,9 @@ class MQTTClientComponent final : public Component {
  bool dns_resolved_{false};
  bool dns_resolve_error_{false};
  bool enable_on_boot_{true};
+#ifdef USE_MQTT_CRASH_LOG
+  bool crash_log_offered_{false};
+#endif
  std::vector<MQTTComponent *> children_;
  uint32_t reboot_timeout_{300000};
  uint32_t connect_begin_;
@@ -392,6 +392,7 @@ async def to_code(config: ConfigType) -> None:
        "USE_RP2040_WATCHDOG_TIMEOUT", config[CONF_WATCHDOG_TIMEOUT]
    )  # back-compat alias
    cg.add_define("USE_RP2_CRASH_HANDLER")
+    cg.add_define("USE_CRASH_HANDLER")
    cg.add_define("USE_RP2040_CRASH_HANDLER")  # back-compat alias

    _configure_lwip()
@@ -58,6 +58,7 @@
 #define USE_CLIMATE_VISUAL_OVERRIDES
 #define USE_CONTROLLER_REGISTRY
 #define USE_COVER
+#define USE_CRASH_HANDLER
 #define USE_DATETIME
 #define USE_DATETIME_DATE
 #define USE_DATETIME_DATETIME
@@ -11,6 +11,7 @@ from typing import TYPE_CHECKING
 import paho.mqtt.client as mqtt

 from esphome.const import (
+    CONF_BIRTH_MESSAGE,
    CONF_BROKER,
    CONF_CERTIFICATE_AUTHORITY,
    CONF_CLIENT_CERTIFICATE,
@@ -21,7 +22,9 @@ from esphome.const import (
    CONF_MQTT,
    CONF_NAME,
    CONF_PASSWORD,
+    CONF_PAYLOAD,
    CONF_PORT,
+    CONF_RETAIN,
    CONF_SKIP_CERT_CN_CHECK,
    CONF_TOPIC,
    CONF_TOPIC_PREFIX,
@@ -319,13 +322,36 @@ def show_logs(config, topic=None, username=None, password=None, client_id=None):
        return 1
    _LOGGER.info("Starting log output from %s", topic)

-    def on_message(client, userdata, msg):
-        time_ = datetime.now().astimezone().time().strftime("[%H:%M:%S]")
-        payload = msg.payload.decode(errors="backslashreplace")
-        message = time_ + payload
-        safe_print(message)
+    logs_topic = f"esphome/logs/{config[CONF_ESPHOME][CONF_NAME]}"
+    # Ask when the retained birth message reports online: a request sent while the device is down is
+    # lost, and only a retained message tells a CLI that connects later that the device is already up.
+    # An empty payload deletes the retained message instead, so it counts as no birth message.
+    birth = config.get(CONF_MQTT, {}).get(CONF_BIRTH_MESSAGE) or {}
+    birth_payload = birth.get(CONF_PAYLOAD)
+    birth_topic = birth.get(CONF_TOPIC) if birth_payload else None
+    ask_on_connect = birth_topic is None or not birth[CONF_RETAIN]

-    return initialize(config, [topic], on_message, None, username, password, client_id)
+    def request_crash_log(client):
+        _LOGGER.debug("Requesting the crash log via topic: %s", logs_topic)
+        client.publish(logs_topic, None, retain=False)
+
+    def on_message(client, userdata, msg):
+        payload = msg.payload.decode(errors="backslashreplace")
+        if msg.topic == birth_topic and payload == birth_payload:
+            request_crash_log(client)
+        # The birth topic may sit inside the log subscription, so print by what was asked for
+        if mqtt.topic_matches_sub(topic, msg.topic):
+            time_ = datetime.now().astimezone().time().strftime("[%H:%M:%S]")
+            safe_print(time_ + payload)
+
+    def on_connect(client, userdata, flags, return_code):
+        if ask_on_connect:
+            request_crash_log(client)
+
+    subscriptions = [topic] if birth_topic is None else [topic, birth_topic]
+    return initialize(
+        config, subscriptions, on_message, on_connect, username, password, client_id
+    )


 def clear_topic(config, topic, username=None, password=None, client_id=None):
@@ -12,7 +12,7 @@ pyserial==3.5
 platformio==6.2.0
 esptool==5.5.0
 click==8.5.0
-aioesphomeapi==46.10.0
+aioesphomeapi==46.11.0
 aiohappyeyeballs==2.7.1  # Happy Eyeballs for requests downloads; already pulled in by aioesphomeapi
 zeroconf==0.151.5
 puremagic==2.2.0
@@ -49,6 +49,7 @@ Options:
 from __future__ import annotations

 import argparse
+import ast
 from collections import Counter
 from collections.abc import Callable
 from enum import StrEnum
@@ -58,6 +59,7 @@ import math
 import os
 from pathlib import Path
 import statistics
+import subprocess
 import sys
 from typing import Any

@@ -69,6 +71,7 @@ from clang_tidy_hash import (
 )
 from helpers import (
    CPP_FILE_EXTENSIONS,
+    ESPHOME_COMPONENTS_PATH,
    ESPHOME_TESTS_COMPONENTS_PATH,
    INTEGRATION_TESTS_PATH,
    PYTHON_FILE_EXTENSIONS,
@@ -805,6 +808,9 @@ def determine_cpp_unit_tests(
 # Paths within tests/benchmarks/ that contain component benchmark files
 BENCHMARKS_COMPONENTS_PATH = "tests/benchmarks/components"

+# The Python benchmark that imports components; its COMPONENTS tuple says which ones
+PYTHON_IMPORT_BENCHMARK_FILE = "tests/benchmarks/python/test_component_import_bench.py"
+
 # Files that, when changed, should trigger benchmark runs
 BENCHMARK_INFRASTRUCTURE_FILES = frozenset(
    {
@@ -815,6 +821,53 @@ BENCHMARK_INFRASTRUCTURE_FILES = frozenset(
 )


+def _benchmark_components_tuple(path: Path) -> frozenset[str]:
+    """Return the benchmark file's ``COMPONENTS`` tuple.
+
+    Raises when the file has none, so renaming it fails CI instead of quietly
+    turning the trigger off.
+    """
+    for node in ast.parse(path.read_text(encoding="utf-8")).body:
+        if isinstance(node, ast.Assign) and any(
+            isinstance(t, ast.Name) and t.id == "COMPONENTS" for t in node.targets
+        ):
+            return frozenset(ast.literal_eval(node.value))
+    raise RuntimeError(f"{PYTHON_IMPORT_BENCHMARK_FILE} has no COMPONENTS tuple")
+
+
+# Imports the given components and prints every component whose module got loaded
+_IMPORT_PROBE = (
+    "import importlib, sys\n"
+    "for c in sys.argv[1:]: importlib.import_module('esphome.components.' + c)\n"
+    "print(*{n.split('.')[2] for n in sys.modules if n.startswith('esphome.components.')})"
+)
+
+
+def python_import_benchmarked_components() -> frozenset[str] | None:
+    """Return the components loaded by importing the benchmark's ``COMPONENTS``.
+
+    Imported in a fresh interpreter from the checkout, so the set is exactly what
+    the benchmark times. None when the import fails.
+    """
+    path = Path(root_path) / PYTHON_IMPORT_BENCHMARK_FILE
+    if not path.is_file():
+        return frozenset()
+    result = subprocess.run(
+        [sys.executable, "-c", _IMPORT_PROBE, *_benchmark_components_tuple(path)],
+        cwd=root_path,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if result.returncode != 0:
+        print(
+            f"Benchmark import probe failed, running benchmarks:\n{result.stderr}",
+            file=sys.stderr,
+        )
+        return None
+    return frozenset(result.stdout.split())
+
+
 def should_run_benchmarks(branch: str | None = None) -> bool:
    """Determine if benchmarks (C++ and Python) should run based on changed files.

@@ -831,6 +884,8 @@ def should_run_benchmarks(branch: str | None = None) -> bool:
    4. A directly changed component has benchmark files (no dependency expansion)
    5. Benchmark infrastructure changed (tests/benchmarks/*, script/cpp_benchmark.py,
       script/build_helpers.py, script/setup_codspeed_lib.py)
+    6. A Python file changed in a component that importing the Python import
+       benchmark's ``COMPONENTS`` loads, even if it has no C++ benchmark

    Unlike unit tests, benchmarks do NOT expand to dependent components.
    Changing ``sensor`` does not trigger ``api`` benchmarks just because
@@ -862,6 +917,19 @@ def should_run_benchmarks(branch: str | None = None) -> bool:
    ):
        return True

+    # The Python import benchmark times importing its components and what they pull
+    # in, so their Python sources move its numbers even without a C++ benchmark.
+    # A failed import runs the benchmarks rather than risk skipping them.
+    changed_python = {
+        get_component_from_path(f)
+        for f in files
+        if f.startswith(ESPHOME_COMPONENTS_PATH) and f.endswith(PYTHON_FILE_EXTENSIONS)
+    }
+    if changed_python:
+        loaded = python_import_benchmarked_components()
+        if loaded is None or changed_python & loaded:
+            return True
+
    # Check if any directly changed component has benchmarks
    benchmarks_dir = Path(root_path) / BENCHMARKS_COMPONENTS_PATH
    if not benchmarks_dir.is_dir():
@@ -12,6 +12,7 @@ esphome:

 bq25186:
  - id: bq25186_charger
+    i2c_id: i2c_bus
    configure_on_boot: false
    fast_charge_current: 150mA
    battery_regulation_voltage: 4.2V
@@ -20,6 +21,7 @@ bq25186:
    charge_enabled: true
  # Second charger with the boot-time write enabled
  - id: bq25186_boot
+    i2c_id: i2c_bus
    address: 0x6B
    configure_on_boot: true
    fast_charge_current: 20mA
@@ -4,6 +4,7 @@
 #include "esphome/components/friedrich/friedrich.h"
 #include "esphome/components/remote_base/aeha_protocol.h"
 #include "esphome/core/helpers.h"
+#include "esphome/core/preferences.h"

 namespace esphome::friedrich::testing {

@@ -41,7 +42,12 @@ struct Fixture {
  TestFriedrich sut;
  CapturingTransmitter transmitter;

-  Fixture() { this->sut.set_transmitter(&this->transmitter); }
+  Fixture() {
+    // Climate tests share one preference key, so clear what an earlier test saved or setup restores it
+    if (global_preferences != nullptr)
+      global_preferences->reset();
+    this->sut.set_transmitter(&this->transmitter);
+  }

  // The last transmission decoded, or an empty frame if it could not be decoded
  AEHAData sent() { return AEHAProtocol().decode(rx_of(this->transmitter.last)).value_or(AEHAData{}); }
@@ -4,6 +4,7 @@ from collections.abc import Generator
 import importlib.util
 import json
 from pathlib import Path
+import subprocess
 import sys
 from unittest.mock import Mock, patch

@@ -2784,6 +2785,87 @@ def test_should_run_benchmarks_core_benchmark_file_change() -> None:
        assert determine_jobs.should_run_benchmarks() is True


+def test_python_import_benchmarked_components_imports_real_components() -> None:
+    """The set is what importing the real benchmark's components loads."""
+    loaded = determine_jobs.python_import_benchmarked_components()
+    # mqtt is not listed in COMPONENTS; sensor imports it at module scope
+    assert {"binary_sensor", "sensor", "logger", "mqtt"} <= loaded
+
+
+def test_should_run_benchmarks_python_import_benchmarked_component() -> None:
+    """A Python change in a component the import benchmark loads runs benchmarks.
+
+    logger has no C++ benchmark, so before this a change to its __init__.py
+    skipped the Python import benchmark that times it.
+    """
+    for changed, expected in [
+        ("esphome/components/logger/__init__.py", True),
+        ("esphome/components/logger/select/__init__.py", True),
+        ("esphome/components/mqtt/__init__.py", True),
+        ("esphome/components/logger/logger.cpp", False),
+        ("tests/components/logger/common.yaml", False),
+        ("esphome/components/wifi/__init__.py", False),
+    ]:
+        with (
+            patch.object(determine_jobs, "changed_files", return_value=[changed]),
+            patch.object(
+                determine_jobs,
+                "python_import_benchmarked_components",
+                return_value=frozenset({"logger", "mqtt"}),
+            ),
+        ):
+            assert determine_jobs.should_run_benchmarks() is expected, changed
+
+
+def test_should_run_benchmarks_when_import_probe_fails() -> None:
+    """A component import that fails runs the benchmarks instead of skipping them."""
+    with (
+        patch.object(
+            determine_jobs,
+            "changed_files",
+            return_value=["esphome/components/wifi/__init__.py"],
+        ),
+        patch.object(
+            determine_jobs, "python_import_benchmarked_components", return_value=None
+        ),
+    ):
+        assert determine_jobs.should_run_benchmarks() is True
+
+
+def test_python_import_benchmarked_components_logs_probe_failure(
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """A failed probe returns None and prints why on stderr, never stdout."""
+    failed = subprocess.CompletedProcess(
+        args=[], returncode=1, stdout="", stderr="ModuleNotFoundError: boom"
+    )
+    with patch.object(determine_jobs.subprocess, "run", return_value=failed):
+        assert determine_jobs.python_import_benchmarked_components() is None
+    captured = capsys.readouterr()
+    assert "ModuleNotFoundError: boom" in captured.err
+    assert captured.out == ""
+
+
+def test_python_import_benchmarked_components_missing_file(tmp_path: Path) -> None:
+    """Without the benchmark file no component is treated as benchmarked."""
+    with patch.object(determine_jobs, "root_path", str(tmp_path)):
+        assert determine_jobs.python_import_benchmarked_components() == frozenset()
+
+
+def test_python_import_benchmarked_components_requires_components(
+    tmp_path: Path,
+) -> None:
+    """A benchmark file without COMPONENTS fails loudly instead of disabling the rule."""
+    bench = tmp_path / determine_jobs.PYTHON_IMPORT_BENCHMARK_FILE
+    bench.parent.mkdir(parents=True)
+    bench.write_text('NAMES = ("logger",)\n')
+    with (
+        patch.object(determine_jobs, "root_path", str(tmp_path)),
+        pytest.raises(RuntimeError, match="COMPONENTS"),
+    ):
+        determine_jobs.python_import_benchmarked_components()
+
+
 def test_should_run_benchmarks_benchmarked_component_change(tmp_path: Path) -> None:
    """Test benchmarks trigger when a benchmarked component changes."""
    # Create a fake benchmarks directory with an 'api' component
@@ -9,9 +9,112 @@ from unittest.mock import MagicMock, patch

 import pytest

-from esphome.const import CONF_BROKER, CONF_ESPHOME, CONF_MQTT, CONF_NAME
+from esphome.const import (
+    CONF_BIRTH_MESSAGE,
+    CONF_BROKER,
+    CONF_ESPHOME,
+    CONF_MQTT,
+    CONF_NAME,
+    CONF_PAYLOAD,
+    CONF_RETAIN,
+    CONF_TOPIC,
+)
 from esphome.core import EsphomeError
-from esphome.mqtt import get_esphome_device_ip
+from esphome.mqtt import get_esphome_device_ip, show_logs
+
+
+@pytest.mark.parametrize(
+    "birth",
+    [
+        None,
+        {CONF_TOPIC: "test-device/status", CONF_PAYLOAD: "online", CONF_RETAIN: False},
+        {CONF_TOPIC: "test-device/status", CONF_PAYLOAD: "", CONF_RETAIN: True},
+    ],
+    ids=["no birth message", "unretained birth message", "empty birth payload"],
+)
+@patch("esphome.mqtt.initialize")
+def test_show_logs_asks_on_connect(
+    mock_initialize: MagicMock, birth: dict | None
+) -> None:
+    """Without a retained birth message nothing tells a later CLI the device is up, so it asks on connect."""
+    config = _discovery_config()
+    if birth is not None:
+        config[CONF_MQTT][CONF_BIRTH_MESSAGE] = birth
+
+    show_logs(config)
+
+    on_connect = mock_initialize.call_args.args[3]
+    client = MagicMock()
+    on_connect(client, None, None, 0)
+    client.publish.assert_called_once_with(
+        "esphome/logs/test-device", None, retain=False
+    )
+    if birth is None or not birth[CONF_PAYLOAD]:
+        assert mock_initialize.call_args.args[1] == ["test-device/debug"]
+
+
+@patch("esphome.mqtt.safe_print")
+@patch("esphome.mqtt.initialize")
+def test_show_logs_requests_crash_log_when_device_reports_online(
+    mock_initialize: MagicMock, mock_print: MagicMock
+) -> None:
+    """With a retained birth message the request follows the device coming online, not the CLI connecting."""
+    config = _discovery_config()
+    config[CONF_MQTT][CONF_BIRTH_MESSAGE] = {
+        CONF_TOPIC: "test-device/status",
+        CONF_PAYLOAD: "online",
+        CONF_RETAIN: True,
+    }
+
+    show_logs(config)
+
+    assert mock_initialize.call_args.args[1] == [
+        "test-device/debug",
+        "test-device/status",
+    ]
+    on_message = mock_initialize.call_args.args[2]
+    on_connect = mock_initialize.call_args.args[3]
+    client = MagicMock()
+    on_connect(client, None, None, 0)
+    client.publish.assert_not_called()
+
+    on_message(client, None, MagicMock(topic="test-device/status", payload=b"offline"))
+    client.publish.assert_not_called()
+    on_message(client, None, MagicMock(topic="test-device/status", payload=b"online"))
+    client.publish.assert_called_once_with(
+        "esphome/logs/test-device", None, retain=False
+    )
+    mock_print.assert_not_called()
+
+    on_message(
+        client, None, MagicMock(topic="test-device/debug", payload=b"[I][app]: hi")
+    )
+    assert mock_print.call_args.args[0].endswith("[I][app]: hi")
+    client.publish.assert_called_once()
+
+
+@patch("esphome.mqtt.safe_print")
+@patch("esphome.mqtt.initialize")
+def test_show_logs_prints_a_birth_message_inside_the_log_subscription(
+    mock_initialize: MagicMock, mock_print: MagicMock
+) -> None:
+    """A birth topic covered by the log topic is still printed, and still requests the crash log."""
+    config = _discovery_config()
+    config[CONF_MQTT][CONF_BIRTH_MESSAGE] = {
+        CONF_TOPIC: "test-device/status",
+        CONF_PAYLOAD: "online",
+        CONF_RETAIN: True,
+    }
+
+    show_logs(config, topic="test-device/#")
+
+    on_message = mock_initialize.call_args.args[2]
+    client = MagicMock()
+    on_message(client, None, MagicMock(topic="test-device/status", payload=b"online"))
+    client.publish.assert_called_once_with(
+        "esphome/logs/test-device", None, retain=False
+    )
+    assert mock_print.call_args.args[0].endswith("online")


 def test_get_esphome_device_ip_empty_broker() -> None: