[libretiny] Keep the RTL8720C frame bounds check from wrapping

Merge remote-tracking branch 'origin/dev' into rtl87xx-crash-handler
[libretiny] Explain the RTL8720C fault entry stub
2026-10-27 08:22:25 +00:00 · 2026-10-26 06:31:45 -10:00 · 2026-10-26 06:30:01 -10:00 · 2026-10-26 06:30:00 -10:00 · 2026-10-26 06:28:54 -10:00 · 2026-10-26 06:26:00 -10:00
71 changed files with 1073 additions and 2865 deletions
@@ -1305,10 +1305,7 @@ def upload_program(
                f"{option_string} is only supported with the esphome OTA platform; "
                "the web_server OTA path can only update the firmware image."
            )
-        binary = CORE.firmware_bin
-        if getattr(args, "file", None) is not None:
-            binary = Path(args.file)
-        return _upload_via_web_server(config, network_devices, binary)
+        return _upload_via_web_server(config, network_devices, _app_ota_image(args))

    return _upload_via_native_api(config, network_devices, args)

@@ -1364,6 +1361,16 @@ def _esphome_ota_conf(config: ConfigType) -> ConfigType | None:
    return next(iter(esphome_ota_items(config)), None)


+def _app_ota_image(args: ArgsProtocol) -> Path:
+    """The file an app update sends: ``--file`` or the build output, as the
+    target platform's ota_image hook picks it."""
+    binary = CORE.firmware_bin
+    if (file := getattr(args, "file", None)) is not None:
+        binary = Path(file)
+    ota_image = platform_hooks.get_platform_hook(CORE.target_platform, "ota_image")
+    return ota_image(binary) if ota_image is not None else binary
+
+
 def _upload_via_native_api(
    config: ConfigType, network_devices: list[str], args: ArgsProtocol
 ) -> tuple[int, str | None]:
@@ -1413,27 +1420,27 @@ def _upload_via_native_api(
                f"flash a build with the option enabled, and then retry {option_string}."
            )

-    binary = CORE.firmware_bin
    ota_type = espota2.OTA_TYPE_UPDATE_APP
+    file = getattr(args, "file", None)
    if getattr(args, "partition_table", False):
        # Fail fast if the resolved ESPHome OTA config does not enable allow_partition_access.
        # The device-side handshake also rejects this with "Device only supports app updates",
        # but checking here surfaces the misconfiguration before opening a network connection.
        check_partition_access("--partition-table")
-        binary = CORE.partition_table_bin
+        binary = Path(file) if file is not None else CORE.partition_table_bin
        ota_type = espota2.OTA_TYPE_UPDATE_PARTITION_TABLE
    elif getattr(args, "bootloader", False):
        check_partition_access("--bootloader")
        if (
-            getattr(args, "file", None) is None
+            file is None
            and (native := native_backend())
            and (hint := getattr(native, "missing_image_hint", lambda: None)())
        ):
            raise EsphomeError(hint)
-        binary = CORE.bootloader_bin
+        binary = Path(file) if file is not None else CORE.bootloader_bin
        ota_type = espota2.OTA_TYPE_UPDATE_BOOTLOADER
-    if getattr(args, "file", None) is not None:
-        binary = Path(args.file)
+    else:
+        binary = _app_ota_image(args)

    if ota_type == espota2.OTA_TYPE_UPDATE_PARTITION_TABLE:
        _validate_partition_table_binary(binary)
@@ -2233,12 +2240,6 @@ def command_analyze_memory(args: ArgsProtocol, config: ConfigType) -> int:
    return 0


-def command_rotate_key(args: ArgsProtocol, config: ConfigType) -> int | None:
-    from esphome.cli.rotate_key import command_rotate_key as run
-
-    return run(args, config)
-
-
 def command_rename(args: ArgsProtocol, config: ConfigType) -> int | None:
    from esphome.cli.rename import command_rename as run

@@ -2265,7 +2266,6 @@ POST_CONFIG_ACTIONS = {
    "clean-mqtt": command_clean_mqtt,
    "idedata": command_idedata,
    "rename": command_rename,
-    "rotate-key": command_rotate_key,
    "discover": command_discover,
    "analyze-memory": command_analyze_memory,
    "bundle": command_bundle,
@@ -2663,37 +2663,6 @@ def parse_args(argv):
    )
    parser_rename.add_argument("name", help="The new name for the device.", type=str)

-    parser_rotate_key = subparsers.add_parser(
-        "rotate-key",
-        help=(
-            "Change the OTA encryption key over the air: write a new key to the "
-            "yaml, build with it, upload with the current key, confirm."
-        ),
-        parents=[mqtt_options],
-    )
-    parser_rotate_key.add_argument(
-        "configuration", help="Your YAML configuration file.", nargs=1
-    )
-    parser_rotate_key.add_argument(
-        "--device",
-        action="append",
-        help=ARGUMENT_HELP_DEVICE,
-    )
-    parser_rotate_key.add_argument(
-        "--prompt-new-key",
-        action="store_true",
-        help=(
-            "Ask for the new key instead of generating one; read without echo, "
-            "or as one line from stdin when that is not a terminal."
-        ),
-    )
-    parser_rotate_key.add_argument(
-        "-y",
-        "--yes",
-        action="store_true",
-        help="Do not ask for confirmation when the api encryption key changes too.",
-    )
-
    parser_analyze_memory = subparsers.add_parser(
        "analyze-memory",
        help="Analyze memory usage by component.",
@@ -1,289 +0,0 @@
-"""``esphome rotate-key``."""
-
-from __future__ import annotations
-
-import argparse
-from pathlib import Path
-import sys
-
-from esphome import espota2
-from esphome.components.noise import (
-    generate_encryption_key,
-    static_encryption_key,
-    validate_encryption_key,
-)
-from esphome.config_validation import Invalid
-from esphome.const import CONF_API, CONF_PASSWORD, CONF_PORT
-from esphome.core import CORE, EsphomeError
-from esphome.espota2 import ProbeResult
-from esphome.log import AnsiFore, color
-from esphome.ota_key_edit import locate_key_edits, old_key_edit, with_sharers
-from esphome.types import ConfigType
-from esphome.upload_targets import PortType, get_port_type
-from esphome.util import (
-    ESPHOME_COMMAND,
-    read_secret_line,
-    run_external_process,
-    safe_input,
-    safe_print,
-)
-from esphome.yaml_edit import RestoreError, Snapshot, apply_line_edits, restore_files
-
-PRECHECK_TIMEOUT = 15.0
-# What to do when the device does not accept the key in the yaml before anything is written
-PRECHECK_ADVICE = {
-    ProbeResult.UNREACHABLE: (
-        "The device could not be reached; check that it is online, then try again."
-    ),
-    ProbeResult.REJECTED: (
-        "The device runs a different key than the yaml. Install the current "
-        "configuration first, presenting its key with --prompt-ota-key, then rotate."
-    ),
-    ProbeResult.NOT_OFFERED: (
-        "The device does not offer encryption: it runs a firmware older than "
-        "ESPHome 2026.9.0 or without a key. Install the current configuration "
-        "first, then rotate."
-    ),
-    ProbeResult.DEVICE_ERROR: (
-        "The device answered with an error (see the warning above); fix that first."
-    ),
-}
-# Why the device did not confirm the new key after the upload
-CONFIRM_FAILURE = {
-    ProbeResult.REJECTED: "The device still answers with the previous key.",
-    ProbeResult.NOT_OFFERED: "The device came back without encryption.",
-    ProbeResult.DEVICE_ERROR: "The device answered with an error (see the warning above).",
-    ProbeResult.UNREACHABLE: "The device did not come back to confirm the new key.",
-}
-
-
-def command_rotate_key(args: argparse.Namespace, config: ConfigType) -> int | None:
-    """Write a new key, build with it, upload with the current one, confirm."""
-    from esphome.__main__ import (
-        Purpose,
-        _esphome_ota_conf,
-        _resolve_network_devices,
-        choose_upload_log_host,
-    )
-
-    def fail(message: str) -> int:
-        safe_print(color(AnsiFore.BOLD_RED, message))
-        return 1
-
-    ota_conf = _esphome_ota_conf(config)
-    if ota_conf is None:
-        return fail("rotate-key needs the esphome OTA platform in the configuration")
-    old_key = static_encryption_key(ota_conf)
-    if old_key is None:
-        return fail(
-            "rotate-key needs 'encryption:' under the esphome OTA platform with a "
-            "key in the yaml; a key provisioned at runtime is rotated from Home "
-            "Assistant"
-        )
-    if CORE.is_host:
-        return fail("rotate-key is for devices updated over the air")
-    old_key = str(old_key)
-    remote_port = int(ota_conf[CONF_PORT])
-
-    devices = choose_upload_log_host(
-        default=args.device, check_default=None, purpose=Purpose.UPLOADING
-    )
-    if get_port_type(devices[0]) in (PortType.SERIAL, PortType.BOOTSEL):
-        return fail(
-            "rotate-key works over the air; on a serial connection change the "
-            "key in the yaml and run 'esphome run' instead"
-        )
-    network_devices = _resolve_network_devices(devices, config, args)
-
-    if args.prompt_new_key:
-        new_key = read_secret_line("New OTA encryption key: ")
-    else:
-        new_key = generate_encryption_key()
-    try:
-        validate_encryption_key(new_key)
-    except Invalid as err:
-        return fail(f"Invalid new key: {err}")
-    if new_key == old_key:
-        return fail("The new key is the same as the current one")
-
-    try:
-        edits = locate_key_edits(old_key, new_key)
-        # The previous key stays in the config as old_key so an install still
-        # reaches the device if it ends up running either key
-        edits += old_key_edit(old_key)
-        unchecked = with_sharers(edits)
-    except EsphomeError as err:
-        return fail(str(err))
-
-    warnings = []
-    # With the recommended shape the ota inherits the api key, so rotating
-    # it changes what Home Assistant connects with
-    if str(static_encryption_key(config.get(CONF_API) or {})) == old_key:
-        warnings.append(
-            "The api encryption key changes too; Home Assistant will ask for "
-            "the new key after the install."
-        )
-    shared: dict[str, set[str]] = {}
-    for edit in edits:
-        what = f"secret '{edit.secret}'" if edit.secret else edit.path.name
-        for path in edit.shared_with:
-            shared.setdefault(
-                str(path.relative_to(CORE.config_dir.resolve())), set()
-            ).add(what)
-    if unchecked:
-        warnings.append(
-            "Could not check every configuration for shared use: "
-            + "; ".join(unchecked)
-        )
-    if shared:
-        warnings.append(
-            "Other configurations share what is rewritten: "
-            + "; ".join(
-                f"{p} ({', '.join(sorted(w))})" for p, w in sorted(shared.items())
-            )
-            + ". Their devices keep the previous key until their next install; "
-            "make sure those configurations carry it as old_key."
-        )
-    if warnings:
-        for warning in warnings:
-            safe_print(color(AnsiFore.BOLD_YELLOW, warning))
-        if not args.yes:
-            if not sys.stdin.isatty():
-                return fail("Confirm with --yes when there is no terminal")
-            if safe_input("Continue? [y/N] ").strip().lower() not in ("y", "yes"):
-                return 1
-
-    safe_print("Checking that the device accepts the current key...")
-    found = espota2.probe_ota_key(
-        network_devices,
-        remote_port,
-        old_key,
-        timeout=PRECHECK_TIMEOUT,
-        retry_rejected=False,
-    )
-    if found is not ProbeResult.ACCEPTED:
-        return fail(PRECHECK_ADVICE[found])
-
-    # Both keys are shown only when a file could not be put back
-    keys = f"Previous key: {old_key}\nNew key: {new_key}"
-    try:
-        originals = apply_line_edits(edits)
-    except RestoreError as err:
-        safe_print(keys)
-        return fail(str(err))
-    except EsphomeError as err:
-        return fail(str(err))
-    except KeyboardInterrupt as err:
-        if notes := getattr(err, "__notes__", None):  # a rollback that failed too
-            safe_print(keys)
-            return fail("Interrupted; " + "; ".join(notes))
-        return fail("Interrupted before the upload")
-    # From here every exit restores or reports, so nothing sits outside the try
-    uploaded = False
-    try:
-        safe_print(
-            "Wrote the new key to "
-            + ", ".join(color(AnsiFore.CYAN, str(p)) for p in originals)
-        )
-        # Global options go before the subcommand; the compile parser is strict
-        cli_args = ["--dashboard"] if CORE.dashboard else []
-        if args.toolchain:
-            cli_args += ["--toolchain", str(args.toolchain)]
-        for key, value in args.substitution or []:
-            cli_args += ["-s", key, value]
-        cli_args += ["compile", str(CORE.config_path)]
-        if run_external_process(*ESPHOME_COMMAND, *cli_args) != 0:
-            return fail("Compiling with the new key failed")
-
-        def image_sent() -> None:
-            # The device accepted the session and may commit from here on;
-            # the edited config reaches it on either key
-            nonlocal uploaded
-            uploaded = True
-
-        rc, _ = espota2.run_ota(
-            network_devices,
-            remote_port,
-            ota_conf.get(CONF_PASSWORD),
-            CORE.firmware_bin,
-            noise_psk=old_key,
-            plaintext_fallback=False,
-            on_upload_start=image_sent,
-        )
-        if rc != 0:
-            return fail(
-                "Uploading with the current key failed. The configuration keeps "
-                "the new key with the previous one as old_key, so the next "
-                "install reaches the device whichever key it runs."
-                if uploaded
-                else "The upload failed before the image was sent (see the error "
-                "above); nothing changed on the device."
-            )
-        safe_print("Waiting for the device to come back with the new key...")
-        found = espota2.probe_ota_key(network_devices, remote_port, new_key)
-        if found is not ProbeResult.ACCEPTED:
-            return fail(
-                CONFIRM_FAILURE[found]
-                + " The configuration keeps the new key, with the previous one as "
-                "old_key, so the next install reaches the device either way."
-            )
-    except KeyboardInterrupt:
-        return fail(
-            "Interrupted; the device's key was not confirmed"
-            if uploaded
-            else "Interrupted before the upload"
-        )
-    finally:
-        # Before the upload nothing changed on the device; after it the device
-        # most likely runs the new key, and old_key covers the other case
-        if uploaded:
-            safe_print(f"New OTA encryption key: {color(AnsiFore.CYAN, new_key)}")
-        else:
-            _restore(originals, keys)
-
-    safe_print(color(AnsiFore.BOLD_GREEN, "SUCCESS"))
-    return 0
-
-
-def _restore(originals: dict[Path, Snapshot], keys: str) -> None:
-    """Put the previous key back, and drop the images a build made with the
-    new key; both keys are shown when a step fails."""
-    files = ", ".join(str(p) for p in originals)
-    try:
-        restore_files(originals)
-    except (EsphomeError, KeyboardInterrupt) as err:
-        # An interrupted restore carries its report as a note
-        why = str(err) or "; ".join(getattr(err, "__notes__", ()))
-        safe_print(color(AnsiFore.BOLD_RED, why or f"Interrupted restoring {files}"))
-        safe_print(keys)
-    else:
-        safe_print(color(AnsiFore.BOLD_YELLOW, f"Restored the previous key in {files}"))
-    build_dir = CORE.firmware_bin.parent
-    failed: list[str] = []
-    images: list[Path] = []
-    try:
-        images = [p for s in (".bin", ".uf2") for p in build_dir.glob(f"*{s}")]
-        for image in images:  # every one is tried, whatever the others did
-            try:
-                image.unlink()
-            except OSError as err:
-                failed.append(f"{image}: {err}")
-    except (OSError, KeyboardInterrupt) as err:
-        failed.append(f"{build_dir}: {str(err) or 'interrupted'}")
-    if failed:
-        safe_print(
-            color(
-                AnsiFore.BOLD_RED,
-                "Could not remove images that may hold the new key: "
-                + "; ".join(failed),
-            )
-        )
-        safe_print(keys)
-    elif images:
-        safe_print(
-            color(
-                AnsiFore.BOLD_YELLOW,
-                f"Removed the images in {build_dir}, which may hold the new key; "
-                "compile again before the next upload.",
-            )
-        )
@@ -297,9 +297,7 @@ void ADE7880::init_device_() {
 void ADE7880::reset_device_() {
  if (this->reset_pin_ != nullptr) {
    ESP_LOGD(TAG, "Reset device using RESET pin");
-    this->reset_pin_->digital_write(false);
-    delay(1);
-    this->reset_pin_->digital_write(true);
+    pulse_reset(this->reset_pin_, {.low_ms = 1});
  } else {
    ESP_LOGD(TAG, "Reset device using SWRST command");
    this->write_u16_register16_(CONFIG, CONFIG_SWRST);
@@ -16,13 +16,7 @@ constexpr static const uint8_t AXS_READ_TOUCHPAD[11] = {0xb5, 0xab, 0xa5, 0x5a,
  }

 void AXS15231Touchscreen::setup() {
-  if (this->reset_pin_ != nullptr) {
-    this->reset_pin_->setup();
-    this->reset_pin_->digital_write(false);
-    delay(5);
-    this->reset_pin_->digital_write(true);
-    delay(10);
-  }
+  pulse_reset(this->reset_pin_, {.low_ms = 5, .high_after_ms = 10});
  if (this->interrupt_pin_ != nullptr) {
    this->interrupt_pin_->pin_mode(gpio::FLAG_INPUT);
    this->interrupt_pin_->setup();
@@ -78,3 +78,7 @@ async def pin_to_code(config: ConfigType) -> MockObj:
 # Called by writer.py; delegates to the shared libretiny implementation.
 def copy_files() -> None:
    libretiny.copy_files()
+
+
+# Called by the OTA upload; the shared libretiny implementation.
+ota_image = libretiny.device_ota_image
@@ -6,12 +6,7 @@ ESPHOME_LOG_TAG(TAG, "cst226.touchscreen");

 void CST226Touchscreen::setup() {
  if (this->reset_pin_ != nullptr) {
-    this->reset_pin_->setup();
-    this->reset_pin_->digital_write(true);
-    delay(5);
-    this->reset_pin_->digital_write(false);
-    delay(5);
-    this->reset_pin_->digital_write(true);
+    pulse_reset(this->reset_pin_, {.high_before_ms = 5, .low_ms = 5});
    this->set_timeout(30, [this] { this->continue_setup_(); });
  } else {
    this->continue_setup_();
@@ -43,9 +43,7 @@ void CST328Touchscreen::setup() {
 }

 void CST328Touchscreen::reset_device_() {
-  this->reset_pin_->digital_write(false);
-  delay(5);
-  this->reset_pin_->digital_write(true);
+  pulse_reset(this->reset_pin_, {.low_ms = 5});
  this->set_timeout(CST328_TRANSITION_TIMEOUT, [this] { this->continue_setup_(); });
 }

@@ -52,12 +52,7 @@ void CST816Touchscreen::continue_setup_() {

 void CST816Touchscreen::setup() {
  if (this->reset_pin_ != nullptr) {
-    this->reset_pin_->setup();
-    this->reset_pin_->digital_write(true);
-    delay(5);
-    this->reset_pin_->digital_write(false);
-    delay(5);
-    this->reset_pin_->digital_write(true);
+    pulse_reset(this->reset_pin_, {.high_before_ms = 5, .low_ms = 5});
    this->set_timeout(30, [this] { this->continue_setup_(); });
  } else {
    this->continue_setup_();
@@ -10,14 +10,7 @@ static constexpr uint32_t REREAD_TIMEOUT_ID = 1;
 static constexpr uint32_t REREAD_DELAY_MS = 20;

 void CST9220Touchscreen::setup() {
-  if (this->reset_pin_ != nullptr) {
-    this->reset_pin_->setup();
-    this->reset_pin_->digital_write(true);
-    delay(5);
-    this->reset_pin_->digital_write(false);
-    delay(10);
-    this->reset_pin_->digital_write(true);
-  }
+  pulse_reset(this->reset_pin_, {.high_before_ms = 5, .low_ms = 10});
  // Wait for the controller to leave its bootloader before talking to it.
  this->set_timeout(30, [this] { this->continue_setup_(); });
 }
@@ -42,6 +42,7 @@ CONFIG_SCHEMA = cv.All(


 async def to_code(config: ConfigType) -> None:
+    cg.add_define("USE_LINE_READER")
    await uart.new_uart_component(config)


@@ -11,10 +11,10 @@
 namespace esphome::dfrobot_sen0395 {

 ESPHOME_LOG_TAG(TAG, "dfrobot_sen0395");
-static constexpr char ASCII_CR = 0x0D;
-static constexpr char ASCII_LF = 0x0A;
 static constexpr float DET_RANGE_STEP_M = 0.15f;
 static constexpr float LATENCY_STEP_S = 0.025f;
+// Sent after a reset with no line end
+static constexpr char PROMPT[] PROGMEM = "leapMMW:/>";

 // What each CommandType cannot do while the sensor runs; empty for commands that do not need it stopped
 PROGMEM_STRING_TABLE(BusyActions, "", "configure range config", "configure output latency",
@@ -121,9 +121,9 @@ bool DfrobotSen0395Component::run_command_(const Command &cmd, uint32_t now) {
    return false;  // Command not done yet
  }

-  if (this->read_message_()) {
-    if (ESPHOME_strstr_P(this->read_buffer_, ESPHOME_PSTR("is not recognized as a CLI command")) == nullptr) {
-      if (!this->handle_reply_(cmd))
+  if (const char *message = this->read_message_()) {
+    if (ESPHOME_strstr_P(message, ESPHOME_PSTR("is not recognized as a CLI command")) == nullptr) {
+      if (!this->handle_reply_(cmd, message))
        return false;         // Command not done yet
      this->read_message_();  // Skip the prompt
      return true;
@@ -146,12 +146,13 @@ bool DfrobotSen0395Component::run_command_(const Command &cmd, uint32_t now) {
 }

 void DfrobotSen0395Component::read_state_() {
-  if (!this->read_message_())
+  const char *message = this->read_message_();
+  if (message == nullptr)
    return;
-  if (ESPHOME_strstr_P(this->read_buffer_, ESPHOME_PSTR("$JYBSS,0, , , *")) != nullptr) {
+  if (ESPHOME_strstr_P(message, ESPHOME_PSTR("$JYBSS,0, , , *")) != nullptr) {
    this->set_detected_(false);
    this->set_active(true);
-  } else if (ESPHOME_strstr_P(this->read_buffer_, ESPHOME_PSTR("$JYBSS,1, , , *")) != nullptr) {
+  } else if (ESPHOME_strstr_P(message, ESPHOME_PSTR("$JYBSS,1, , , *")) != nullptr) {
    this->set_detected_(true);
    this->set_active(true);
  }
@@ -206,8 +207,7 @@ void DfrobotSen0395Component::format_command(const Command &cmd, char *buf, size
  }
 }

-bool DfrobotSen0395Component::handle_reply_(const Command &cmd) {
-  const char *message = this->read_buffer_;
+bool DfrobotSen0395Component::handle_reply_(const Command &cmd, const char *message) {
  auto is = [message](ESPHOME_PGM_P text) { return ESPHOME_strcmp_P(message, text) == 0; };

  // Commands with their own replies
@@ -230,7 +230,7 @@ bool DfrobotSen0395Component::handle_reply_(const Command &cmd) {
      }
      return true;
    case CommandType::COMMAND_TYPE_RESET_SYSTEM:
-      if (!is(ESPHOME_PSTR("leapMMW:/>")))
+      if (!is(PROMPT))
        return false;
      ESP_LOGI(TAG, "Restarted sensor.");
      return true;
@@ -300,34 +300,24 @@ bool DfrobotSen0395Component::handle_reply_(const Command &cmd) {
  return true;
 }

-bool DfrobotSen0395Component::read_message_() {
+const char *DfrobotSen0395Component::read_message_() {
  while (this->available()) {
    uint8_t byte;
    this->read_byte(&byte);
-
-    if (this->read_pos_ == MMWAVE_READ_BUFFER_LENGTH)
-      this->read_pos_ = 0;
-
-    ESP_LOGVV(TAG, "Buffer pos: %u %d", this->read_pos_, byte);
-
-    if (byte == ASCII_CR)
-      continue;
    if (byte >= 0x7F)
      byte = '?';  // needs to be valid utf8 string for log functions.
-    this->read_buffer_[this->read_pos_] = byte;
-
-    if (this->read_pos_ == 9 && byte == '>')
-      this->read_buffer_[++this->read_pos_] = ASCII_LF;
-
-    if (this->read_buffer_[this->read_pos_] == ASCII_LF) {
-      this->read_buffer_[this->read_pos_] = 0;
-      this->read_pos_ = 0;
-      ESP_LOGV(TAG, "Message: %s", this->read_buffer_);
-      return true;  // Full message in buffer
+    const char *line = this->line_reader_.feed(byte);
+    if (line == nullptr && byte == '>' && this->line_reader_.length() == sizeof(PROMPT) - 1) {
+      line = this->line_reader_.partial();
+      this->line_reader_.reset();
    }
-    this->read_pos_++;
+    if (line != nullptr) {
+      ESP_LOGV(TAG, "Message: %s", line);
+      return line;
+    }
+    this->line_reader_.log_if_overflowed(TAG);
  }
-  return false;  // No full message yet
+  return nullptr;
 }

 void DfrobotSen0395Component::set_detected_(bool detected) {
@@ -4,6 +4,7 @@
 #include "esphome/core/automation.h"
 #include "esphome/core/component.h"
 #include "esphome/core/helpers.h"
+#include "esphome/core/line_reader.h"

 #ifdef USE_BINARY_SENSOR
 #include "esphome/components/binary_sensor/binary_sensor.h"
@@ -14,7 +15,7 @@

 namespace esphome::dfrobot_sen0395 {

-static constexpr uint8_t MMWAVE_READ_BUFFER_LENGTH = 255;
+static constexpr uint8_t MMWAVE_READ_BUFFER_LENGTH = 253;  // largest LineReader size with a one byte position
 static constexpr uint8_t COMMAND_QUEUE_SIZE = 20;

 // Keep in sync with the BusyActions table in dfrobot_sen0395.cpp
@@ -121,8 +122,7 @@ class DfrobotSen0395Component final : public uart::UARTDevice, public Component
  bool led_active_{false};
  bool uart_presence_active_{false};
  bool start_after_boot_{false};
-  char read_buffer_[MMWAVE_READ_BUFFER_LENGTH];
-  size_t read_pos_{0};
+  LineReader<MMWAVE_READ_BUFFER_LENGTH> line_reader_;
  bool cmd_sent_{false};
  uint8_t retries_left_{2};
  StaticRingBuffer<Command, COMMAND_QUEUE_SIZE> cmd_queue_;
@@ -131,12 +131,13 @@ class DfrobotSen0395Component final : public uart::UARTDevice, public Component
  int16_t det_range_steps_[8]{};
  uint16_t latency_steps_[2]{};

-  bool read_message_();
+  /// The next complete line, valid until the next call, or nullptr
+  const char *read_message_();
  void read_state_();
  /// Run the command at the front of the queue; true when it is done
  bool run_command_(const Command &cmd, uint32_t now);
  /// Handle a reply to a sent command; true when the command is done
-  bool handle_reply_(const Command &cmd);
+  bool handle_reply_(const Command &cmd, const char *message);

  void set_detected_(bool detected);
 };
@@ -20,9 +20,7 @@ void EKTF2232Touchscreen::setup() {

  this->attach_interrupt_(this->interrupt_pin_, gpio::INTERRUPT_FALLING_EDGE);

-  this->reset_pin_->setup();
-
-  this->hard_reset_();
+  pulse_reset(this->reset_pin_, {.low_ms = 15, .high_after_ms = 15});
  if (!this->soft_reset_()) {
    ESP_LOGE(TAG, "Failed to soft reset EKT2232!");
    this->interrupt_pin_->detach_interrupt();
@@ -96,13 +94,6 @@ bool EKTF2232Touchscreen::get_power_state() {
  return (received[1] >> 3) & 1;
 }

-void EKTF2232Touchscreen::hard_reset_() {
-  this->reset_pin_->digital_write(false);
-  delay(15);
-  this->reset_pin_->digital_write(true);
-  delay(15);
-}
-
 bool EKTF2232Touchscreen::soft_reset_() {
  auto err = this->write(SOFT_RESET_CMD, 4);
  if (err != i2c::ERROR_OK)
@@ -22,7 +22,6 @@ class EKTF2232Touchscreen final : public Touchscreen, public i2c::I2CDevice {
  bool get_power_state();

 protected:
-  void hard_reset_();
  bool soft_reset_();
  void update_touches() override;

@@ -72,9 +72,7 @@ bool EPaperJD79660::reset() {

      // As commented on SLEEP_MS_RESET1: Reset pulse must happen within time window.
      // So do not use FSM loop, and avoid other calls/logs during pulse below.
-      this->reset_pin_->digital_write(false);
-      delay(SLEEP_MS_RESET1);
-      this->reset_pin_->digital_write(true);
+      pulse_reset(this->reset_pin_, {.low_ms = SLEEP_MS_RESET1});

      this->reset_duration_ = SLEEP_MS_RESET2;
      this->step_ = FSMState::RESET_STEP2_IDLECHECK;
@@ -34,10 +34,7 @@ void FT63X6Touchscreen::setup() {
    this->attach_interrupt_(this->interrupt_pin_, gpio::INTERRUPT_ANY_EDGE);
  }

-  if (this->reset_pin_ != nullptr) {
-    this->reset_pin_->setup();
-    this->hard_reset_();
-  }
+  pulse_reset(this->reset_pin_, {.low_ms = 10});

  // Get touch resolution
  if (this->x_raw_max_ == this->x_raw_min_) {
@@ -57,14 +54,6 @@ void FT63X6Touchscreen::setup() {
  this->write_byte(FT6X36_ADDR_TOUCHRATE_ACTIVE, 0x0E);
 }

-void FT63X6Touchscreen::hard_reset_() {
-  if (this->reset_pin_ != nullptr) {
-    this->reset_pin_->digital_write(false);
-    delay(10);
-    this->reset_pin_->digital_write(true);
-  }
-}
-
 void FT63X6Touchscreen::dump_config() {
  ESP_LOGCONFIG(TAG, "FT63X6 Touchscreen:");
  LOG_I2C_DEVICE(this);
@@ -27,7 +27,6 @@ class FT63X6Touchscreen final : public Touchscreen, public i2c::I2CDevice {
  void set_threshold(uint8_t threshold) { this->threshold_ = threshold; }

 protected:
-  void hard_reset_();
  void update_touches() override;

  InternalGPIOPin *interrupt_pin_{nullptr};
@@ -83,13 +83,7 @@ void GSL3670Touchscreen::update_touches() {
 void GSL3670Touchscreen::clear_reg_() {
  ESP_LOGD(TAG, "clear_reg");

-  // GPIO reset pulse
-  if (this->reset_pin_ != nullptr) {
-    this->reset_pin_->digital_write(false);
-    delay(1);
-    this->reset_pin_->digital_write(true);
-    delay(5);
-  }
+  pulse_reset(this->reset_pin_, {.low_ms = 1, .high_after_ms = 5});

  this->write_reg8_(0x88, 0x01);
  // delay(5);
@@ -106,12 +100,7 @@ void GSL3670Touchscreen::clear_reg_() {
 void GSL3670Touchscreen::reset_() {
  ESP_LOGD(TAG, "reset");

-  if (this->reset_pin_ != nullptr) {
-    this->reset_pin_->digital_write(false);
-    delay(1);
-    this->reset_pin_->digital_write(true);
-    delay(5);
-  }
+  pulse_reset(this->reset_pin_, {.low_ms = 1, .high_after_ms = 5});

  this->write_reg8_(0xe4, 0x04);

@@ -19,12 +19,7 @@ void ICNT86Touchscreen::setup() {
  }

  // Perform reset if necessary
-  if (this->reset_pin_ != nullptr) {
-    this->reset_pin_->setup();
-    this->reset_pin_->digital_write(false);
-    delay(10);
-    this->reset_pin_->digital_write(true);
-  }
+  pulse_reset(this->reset_pin_, {.low_ms = 10});

  if (this->x_raw_max_ == this->x_raw_min_) {
    this->x_raw_max_ = this->display_->get_native_width();
@@ -1,8 +1,7 @@
 """Shared base for the infrared and radio_frequency entity components."""

 import esphome.codegen as cg
-from esphome.components import remote_base
-from esphome.components.api import DOMAIN as API_DOMAIN
+from esphome.const import CONF_API
 from esphome.core import CORE
 from esphome.core.entity_helpers import queue_entity_register
 from esphome.types import ConfigType
@@ -24,8 +23,10 @@ async def attach_transmitter(var: cg.MockObj, config: ConfigType, key: str) -> N
    does. Without this call anywhere in the build a request is answered as
    soon as the frame is handed over.
    """
+    from esphome.components import remote_base
+
    await remote_base.register_transmittable(var, config, key)
-    if API_DOMAIN in CORE.config:
+    if CONF_API in CORE.config:
        cg.add_define("USE_IR_RF_TRANSMIT_COMPLETE")


@@ -26,7 +26,7 @@ from esphome.const import (
 )
 from esphome.core import CORE
 from esphome.core.config import BOARD_MAX_LENGTH
-from esphome.helpers import copy_file_if_changed
+from esphome.helpers import copy_file_if_changed, write_file
 from esphome.platformio.toolchain import copy_ccache_script
 from esphome.storage_json import StorageJSON

@@ -45,6 +45,7 @@ from .const import (
    FAMILY_COMPONENT,
    FAMILY_FRIENDLY,
    FAMILY_RTL8710B,
+    FAMILY_RTL8720C,
    KEY_BOARD,
    KEY_COMPONENT,
    KEY_COMPONENT_DATA,
@@ -59,6 +60,8 @@ from .const import (
 _LOGGER = logging.getLogger(__name__)
 # Fault handlers the crash handler wraps (BK7238, BK7231N)
 BK_TRAP_HANDLERS = ("bk_trap_udef", "bk_trap_pabt", "bk_trap_dabt")
+# Families with a crash handler; the RTL8720C installs it at run time (no wraps).
+CRASH_HANDLER_FAMILIES = (FAMILY_BK7231N, FAMILY_BK7238, FAMILY_RTL8720C)

 CODEOWNERS = ["@kuba2k2"]
 AUTO_LOAD = ["preferences"]
@@ -554,10 +557,11 @@ async def component_to_code(config):
    # RAM-executable output section and prints a post-link placement summary.
    if FAMILY_COMPONENT[config[CONF_FAMILY]] != COMPONENT_BK72XX:
        cg.add_platformio_option("extra_scripts", ["pre:patch_linker.py"])
-    if config[CONF_FAMILY] in (FAMILY_BK7231N, FAMILY_BK7238):
-        # Crash handler (crash_handler.cpp, patch_noinit.py.script)
+    if config[CONF_FAMILY] in CRASH_HANDLER_FAMILIES:
+        # Crash handler (crash_handler.cpp)
        cg.add_define("USE_LIBRETINY_CRASH_HANDLER")
        cg.add_define("USE_CRASH_HANDLER")
+    if config[CONF_FAMILY] in (FAMILY_BK7231N, FAMILY_BK7238):
        for trap in BK_TRAP_HANDLERS:
            cg.add_build_flag(f"-Wl,--wrap={trap}")
        cg.add_platformio_option("extra_scripts", ["pre:patch_noinit.py"])
@@ -656,6 +660,26 @@ async def component_to_code(config):
    await cg.register_component(var, config)


+# Called by the OTA upload through the platforms' ota_image hook
+def device_ota_image(firmware: Path) -> Path:
+    """The file to send for an app update: the UF2 without the images the
+    device skips, written next to the build output, or ``firmware`` itself
+    when there is nothing to drop."""
+    from .ota_uf2 import device_ota_uf2
+
+    data = firmware.read_bytes()
+    if (trimmed := device_ota_uf2(data)) is None:
+        return firmware
+    image = CORE.relative_pioenvs_path(CORE.name, "firmware.ota.uf2")
+    write_file(image, trimmed)
+    _LOGGER.info(
+        "Sending the images this device writes: %s of %s bytes",
+        len(trimmed),
+        len(data),
+    )
+    return image
+
+
 # Called by writer.py
 def copy_files() -> None:
    script_dir = Path(__file__).parent
@@ -1,8 +1,12 @@
-#ifdef USE_BK72XX
+#ifdef USE_LIBRETINY

 #include "esphome/core/defines.h"
 #ifdef USE_LIBRETINY_CRASH_HANDLER

+#if !defined(USE_BK72XX) && !defined(USE_LIBRETINY_VARIANT_RTL8720C)
+#error "The LibreTiny crash handler has no fault capture for this chip family"
+#endif
+
 #include "crash_handler.h"
 #include "core.h"
 #include "esphome/core/log.h"
@@ -13,6 +17,7 @@
 #include <cstdint>
 #include <cstdio>

+#ifdef USE_BK72XX
 extern "C" {
 #include <arm_mcu_pub.h>  // MCU_REG_BACKUP_SP_*
 #include <intc.h>         // struct arm_registers
@@ -39,28 +44,138 @@ extern char _itcmcode_ram_begin[];
 extern char _itcmcode_ram_end[];
 }
 // NOLINTEND(bugprone-reserved-identifier,cert-dcl37-c,cert-dcl51-cpp,readability-identifier-naming)
+#endif  // USE_BK72XX
+
+#ifdef USE_LIBRETINY_VARIANT_RTL8720C
+// NOLINTBEGIN(bugprone-reserved-identifier,cert-dcl37-c,cert-dcl51-cpp,readability-identifier-naming)
+extern "C" {
+// Bounds of the code in flash (XIP) and in RAM (linker symbols, rtl8710c_ram.ld).
+extern char __xip_code_text_start__[];
+extern char __xip_code_text_end__[];
+extern char __ram_entry_text_start__[];
+extern char __ram_code_text_end__[];
+}
+// NOLINTEND(bugprone-reserved-identifier,cert-dcl37-c,cert-dcl51-cpp,readability-identifier-naming)
+#endif  // USE_LIBRETINY_VARIANT_RTL8720C

 namespace esphome::libretiny {

 ESPHOME_LOG_TAG(TAG, "libretiny.crash");

 // 0xDEAD sentinel, then the record layout version.
-static constexpr uint32_t CRASH_MAGIC_V1 = 0xDEAD0001;
+static constexpr uint32_t CRASH_MAGIC_V2 = 0xDEAD0002;

+// BK72xx: the ARM9 exception taken.
 static constexpr uint32_t CRASH_REASON_UNDEFINED = 1;
 static constexpr uint32_t CRASH_REASON_PREFETCH_ABORT = 2;
 static constexpr uint32_t CRASH_REASON_DATA_ABORT = 3;
-PROGMEM_STRING_TABLE(CrashReasonStrings, "Unknown", "Undefined instruction", "Prefetch abort", "Data abort");
+// RTL8720C: the Cortex-M exception number (3 to 7) plus this offset.
+static constexpr uint32_t CRASH_REASON_CORTEX_M_OFFSET = 1;
+PROGMEM_STRING_TABLE(CrashReasonStrings, "Unknown", "Undefined instruction", "Prefetch abort", "Data abort",
+                     "Hard fault", "MemManage fault", "Bus fault", "Usage fault", "Secure fault");
+
+static constexpr size_t MAX_BACKTRACE = 8;
+// Words scanned above sp for return addresses.
+static constexpr size_t STACK_SCAN_WORDS = 64;
+
+struct CrashRecord {
+  uint32_t magic;
+  uint32_t reason;
+  uint32_t pc;
+  uint32_t lr;
+  uint32_t sp;
+  uint32_t fault_status;   // Cortex-M CFSR, 0 on BK72xx
+  uint32_t fault_address;  // Cortex-M BFAR or MMFAR when valid, else 0
+  uint32_t backtrace[MAX_BACKTRACE];
+};
+
+#ifdef USE_BK72XX
+// Kept in ITCM, see patch_noinit.py.script.
+#define CRASH_RECORD_SECTION ".noinit"
+#else
+// rtl8710c_ram.ld keeps .ram.bss.noinit* out of the startup zeroing.
+#define CRASH_RECORD_SECTION ".ram.bss.noinit.esphome_crash"
+#endif
+// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
+__attribute__((section(CRASH_RECORD_SECTION))) static CrashRecord s_record;
+
+static bool s_record_read = false;   // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
+static bool s_record_valid = false;  // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
+
+bool crash_handler_has_data() {
+  if (!s_record_read) {
+    s_record_read = true;
+    s_record_valid = s_record.magic == CRASH_MAGIC_V2;
+    // Clear the magic so the record is reported once.
+    s_record.magic = 0;
+  }
+  return s_record_valid;
+}
+
+// One ESP_LOGE per line: the CLI decodes per line and partial output still shows.
+void crash_handler_log() {
+  if (!crash_handler_has_data())
+    return;
+
+  ESP_LOGE(TAG, "*** CRASH DETECTED ON PREVIOUS BOOT ***");
+  ESP_LOGE(TAG, "  Reason: %s", LOG_STR_ARG(CrashReasonStrings::get_log_str(s_record.reason, 0)));
+  if (s_record.fault_status != 0) {
+    ESP_LOGE(TAG, "  CFSR: 0x%08" PRIX32 "  Address: 0x%08" PRIX32, s_record.fault_status, s_record.fault_address);
+  }
+  ESP_LOGE(TAG, "  PC:  0x%08" PRIX32 "  (fault location)", s_record.pc);
+  ESP_LOGE(TAG, "  LR:  0x%08" PRIX32 "  (return address)", s_record.lr);
+  ESP_LOGE(TAG, "  SP:  0x%08" PRIX32, s_record.sp);
+  uint8_t count = 0;
+  for (; count < MAX_BACKTRACE && s_record.backtrace[count] != 0; count++) {
+    ESP_LOGE(TAG, "  BT%d: 0x%08" PRIX32 "  (stack backtrace)", count, s_record.backtrace[count]);
+  }
+  // addr2line hint with every address
+  char hint[160];
+  int pos = snprintf(hint, sizeof(hint), "Use: addr2line -pfiaC -e raw_firmware.elf 0x%08" PRIX32 " 0x%08" PRIX32,
+                     s_record.pc, s_record.lr);
+  for (uint8_t i = 0; i < count && pos < (int) sizeof(hint) - 12; i++) {
+    pos += snprintf(hint + pos, sizeof(hint) - pos, " 0x%08" PRIX32, s_record.backtrace[i]);
+  }
+  ESP_LOGE(TAG, "%s", hint);
+}
+
+// --- Crash time: exception mode, a few dozen bytes of stack: no printing, no large locals.
+
+static inline bool is_code_addr(uint32_t val) {
+  uint32_t addr = val & ~1u;  // clear the Thumb bit of a return address
+#ifdef USE_BK72XX
+  return (addr >= (uint32_t) _vector_start && addr < (uint32_t) _data_flash_begin) ||
+         (addr >= (uint32_t) _itcmcode_ram_begin && addr < (uint32_t) _itcmcode_ram_end);
+#else
+  return (addr >= (uint32_t) __xip_code_text_start__ && addr < (uint32_t) __xip_code_text_end__) ||
+         (addr >= (uint32_t) __ram_entry_text_start__ && addr < (uint32_t) __ram_code_text_end__);
+#endif
+}
+
+// Return addresses above sp give the backtrace; then the magic marks the record complete.
+static void finish_record(uint32_t sp, uint32_t sram_start, uint32_t sram_end) {
+  uint32_t count = 0;
+  if (sp >= sram_start && sp < sram_end && (sp & 3) == 0) {
+    auto *scan = (const uint32_t *) sp;       // NOLINT(performance-no-int-to-ptr)
+    auto *end = (const uint32_t *) sram_end;  // NOLINT(performance-no-int-to-ptr)
+    for (size_t i = 0; i < STACK_SCAN_WORDS && scan < end && count < MAX_BACKTRACE; i++, scan++) {
+      uint32_t val = *scan;
+      if (is_code_addr(val) && val != s_record.lr && (val & ~1u) != s_record.pc)
+        s_record.backtrace[count++] = val;
+    }
+  }
+  if (count < MAX_BACKTRACE)
+    s_record.backtrace[count] = 0;  // terminator, the reader stops at the first 0
+  s_record.magic = CRASH_MAGIC_V2;
+}
+
+#ifdef USE_BK72XX

 // With Wi-Fi up the boot-time watchdog no longer fires: any interrupts-off spin hangs until power is
 // cycled (BK7238 and BK7231N; cause not in the open SDK). Re-armed here, long enough for the SDK's
 // register dump when sdk_silent is off.
 static constexpr uint32_t CRASH_REBOOT_WDT_MS = 500;

-static constexpr size_t MAX_BACKTRACE = 8;
-// Words scanned above sp for return addresses.
-static constexpr size_t STACK_SCAN_WORDS = 64;
-
 // SRAM: 192 KB on the BK7231N, 256 KB on the BK7238.
 static constexpr uint32_t SRAM_START = 0x00400000;
 #ifdef USE_LIBRETINY_VARIANT_BK7231N
@@ -79,65 +194,6 @@ static constexpr uint32_t CPSR_MODE_UND = 0x1B;
 static constexpr uint32_t CPSR_MODE_SYS = 0x1F;
 static constexpr uint32_t CPSR_THUMB = 1u << 5;

-struct CrashRecord {
-  uint32_t magic;
-  uint32_t reason;
-  uint32_t pc;
-  uint32_t lr;
-  uint32_t sp;
-  uint32_t backtrace[MAX_BACKTRACE];
-};
-
-// Kept in ITCM, see patch_noinit.py.script.
-// NOLINTNEXTLINE(cppcoreguidelines-avoid-non-const-global-variables)
-__attribute__((section(".noinit"))) static CrashRecord s_record;
-
-static bool s_record_read = false;   // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
-static bool s_record_valid = false;  // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
-
-bool crash_handler_has_data() {
-  if (!s_record_read) {
-    s_record_read = true;
-    s_record_valid = s_record.magic == CRASH_MAGIC_V1;
-    // Clear the magic so the record is reported once.
-    s_record.magic = 0;
-  }
-  return s_record_valid;
-}
-
-// One ESP_LOGE per line: the CLI decodes per line and partial output still shows.
-void crash_handler_log() {
-  if (!crash_handler_has_data())
-    return;
-
-  ESP_LOGE(TAG, "*** CRASH DETECTED ON PREVIOUS BOOT ***");
-  ESP_LOGE(TAG, "  Reason: %s", LOG_STR_ARG(CrashReasonStrings::get_log_str(s_record.reason, 0)));
-  ESP_LOGE(TAG, "  PC:  0x%08" PRIX32 "  (fault location)", s_record.pc);
-  ESP_LOGE(TAG, "  LR:  0x%08" PRIX32 "  (return address)", s_record.lr);
-  ESP_LOGE(TAG, "  SP:  0x%08" PRIX32, s_record.sp);
-  uint8_t count = 0;
-  for (; count < MAX_BACKTRACE && s_record.backtrace[count] != 0; count++) {
-    ESP_LOGE(TAG, "  BT%d: 0x%08" PRIX32 "  (stack backtrace)", count, s_record.backtrace[count]);
-  }
-  // addr2line hint with every address
-  char hint[160];
-  int pos = snprintf(hint, sizeof(hint), "Use: addr2line -pfiaC -e raw_firmware.elf 0x%08" PRIX32 " 0x%08" PRIX32,
-                     s_record.pc, s_record.lr);
-  for (uint8_t i = 0; i < count && pos < (int) sizeof(hint) - 12; i++) {
-    pos += snprintf(hint + pos, sizeof(hint) - pos, " 0x%08" PRIX32, s_record.backtrace[i]);
-  }
-  ESP_LOGE(TAG, "%s", hint);
-}
-
-// --- Crash time: exception mode, interrupts masked, a few dozen bytes of stack: no printing, no
-// large locals.
-
-static inline bool is_code_addr(uint32_t val) {
-  uint32_t addr = val & ~1u;  // clear the Thumb bit of a return address
-  return (addr >= (uint32_t) _vector_start && addr < (uint32_t) _data_flash_begin) ||
-         (addr >= (uint32_t) _itcmcode_ram_begin && addr < (uint32_t) _itcmcode_ram_end);
-}
-
 // PUSH_ALL_ARM_REG (boot_handlers.S) saves every mode's registers at the arm_mcu_pub.h addresses
 // before bk_trap_*; lr (r14) follows sp (r13).
 struct SavedBank {
@@ -174,21 +230,9 @@ static void record_crash(arm_registers *regs, uint32_t reason, uint32_t pc_offse
  s_record.pc = regs->pc - pc_offset;
  s_record.lr = lr;
  s_record.sp = sp;
-
-  // Return addresses above sp give the backtrace.
-  uint32_t count = 0;
-  if (sp >= SRAM_START && sp < SRAM_END && (sp & 3) == 0) {
-    auto *scan = (const uint32_t *) sp;       // NOLINT(performance-no-int-to-ptr)
-    auto *end = (const uint32_t *) SRAM_END;  // NOLINT(performance-no-int-to-ptr)
-    for (size_t i = 0; i < STACK_SCAN_WORDS && scan < end && count < MAX_BACKTRACE; i++, scan++) {
-      uint32_t val = *scan;
-      if (is_code_addr(val) && val != lr && (val & ~1u) != s_record.pc)
-        s_record.backtrace[count++] = val;
-    }
-  }
-  if (count < MAX_BACKTRACE)
-    s_record.backtrace[count] = 0;  // terminator, the reader stops at the first 0
-  s_record.magic = CRASH_MAGIC_V1;
+  s_record.fault_status = 0;
+  s_record.fault_address = 0;
+  finish_record(sp, SRAM_START, SRAM_END);

  lt_wdt_enable(CRASH_REBOOT_WDT_MS);
 }
@@ -211,7 +255,106 @@ extern "C" void __wrap_bk_trap_dabt(arm_registers *regs) {
 }
 // NOLINTEND(bugprone-reserved-identifier,cert-dcl37-c,cert-dcl51-cpp,readability-identifier-naming)

+#endif  // USE_BK72XX
+
+#ifdef USE_LIBRETINY_VARIANT_RTL8720C
+
+// The SDK points the fault vectors (3 HardFault to 7 SecureFault) of its RAM vector table (VTOR) at a
+// ROM handler that prints a register dump and then spins. crash_handler_install() puts
+// esphome_fault_entry() in front of it, as the SDK's own CONFIG_MIIO patch (app_start.c) does.
+static_assert(LT_HW_WATCHDOG, "the crash reboot relies on the hardware watchdog");
+// The ROM dump takes about 0.4 s at 115200 baud; without the re-arm the reset waits for the 10 s
+// watchdog from arch_init().
+static constexpr uint32_t CRASH_REBOOT_WDT_MS = 1000;
+static constexpr uint32_t FIRST_FAULT_VECTOR = 3;
+static constexpr uint32_t FAULT_VECTOR_COUNT = 5;
+
+// Architectural System Control Block registers (ARMv8-M).
+static constexpr uintptr_t SCB_VTOR = 0xE000ED08;
+static constexpr uintptr_t SCB_CFSR = 0xE000ED28;
+static constexpr uintptr_t SCB_MMFAR = 0xE000ED34;
+static constexpr uintptr_t SCB_BFAR = 0xE000ED38;
+static constexpr uint32_t CFSR_MMARVALID = 1u << 7;
+static constexpr uint32_t CFSR_BFARVALID = 1u << 15;
+
+// DTCM, where the app's stacks and heap live (rtl8710c_ram.ld).
+static constexpr uint32_t SRAM_START = 0x10000000;
+static constexpr uint32_t SRAM_END = 0x1003FA00;
+
+// Exception frame: r0-r3, r12, lr, pc, xPSR (the RTL8720C has no FPU, so never the extended frame).
+static constexpr uint32_t FRAME_LR = 5;
+static constexpr uint32_t FRAME_PC = 6;
+static constexpr uint32_t FRAME_XPSR = 7;
+static constexpr uint32_t FRAME_WORDS = 8;
+static constexpr uint32_t XPSR_STACK_ALIGNED = 1u << 9;
+
+// The ROM handlers fault_entry() continues into, by vector.
+static uint32_t s_rom_fault_vectors[FAULT_VECTOR_COUNT];  // NOLINT(cppcoreguidelines-avoid-non-const-global-variables)
+
+static inline uint32_t read_reg(uintptr_t addr) {
+  return *(const volatile uint32_t *) addr;  // NOLINT(performance-no-int-to-ptr)
+}
+
+// Returns the ROM handler to continue into. pc, lr and sp stay 0 when the frame is outside DTCM.
+extern "C" __attribute__((used)) uint32_t esphome_record_fault(const uint32_t *frame, uint32_t vector) {
+  uint32_t cfsr = read_reg(SCB_CFSR);
+  uint32_t address = 0;
+  if (cfsr & CFSR_BFARVALID) {
+    address = read_reg(SCB_BFAR);
+  } else if (cfsr & CFSR_MMARVALID) {
+    address = read_reg(SCB_MMFAR);
+  }
+  s_record.reason = vector + CRASH_REASON_CORTEX_M_OFFSET;
+  s_record.fault_status = cfsr;
+  s_record.fault_address = address;
+
+  auto frame_addr = (uint32_t) frame;
+  uint32_t sp = 0;
+  if (frame_addr >= SRAM_START && frame_addr <= SRAM_END - FRAME_WORDS * 4 && (frame_addr & 3) == 0) {
+    s_record.pc = frame[FRAME_PC];
+    s_record.lr = frame[FRAME_LR];
+    sp = frame_addr + FRAME_WORDS * 4 + ((frame[FRAME_XPSR] & XPSR_STACK_ALIGNED) ? 4 : 0);
+  } else {
+    s_record.pc = 0;
+    s_record.lr = 0;
+  }
+  s_record.sp = sp;
+  finish_record(sp, SRAM_START, SRAM_END);
+
+  lt_wdt_enable(CRASH_REBOOT_WDT_MS);
+  return s_rom_fault_vectors[vector - FIRST_FAULT_VECTOR];
+}
+
+// Fault vector: calls esphome_record_fault(frame, exception number), then jumps to the ROM handler it returns
+// with r0-r3 and lr restored. Asm because C would push onto MSP before the frame is found, and the ROM handler
+// reads EXC_RETURN from lr.
+extern "C" __attribute__((naked, used)) void esphome_fault_entry() {
+  __asm volatile("push {r0-r3, r4, lr}  \n"  // r4 only keeps the stack 8-byte aligned
+                 "tst  lr, #4           \n"
+                 "ite  eq               \n"
+                 "mrseq r0, msp         \n"
+                 "mrsne r0, psp         \n"
+                 "it   eq               \n"
+                 "addeq r0, r0, #24     \n"  // MSP: skip the six words pushed above
+                 "mrs  r1, ipsr         \n"
+                 "ubfx r1, r1, #0, #9   \n"  // exception number, 3 to 7
+                 "bl   esphome_record_fault \n"
+                 "mov  r12, r0          \n"  // ROM handler address
+                 "pop  {r0-r3, r4, lr}  \n"
+                 "bx   r12              \n");
+}
+
+void crash_handler_install() {
+  auto *vectors = (uint32_t *) read_reg(SCB_VTOR);  // NOLINT(performance-no-int-to-ptr)
+  for (uint32_t i = 0; i < FAULT_VECTOR_COUNT; i++) {
+    s_rom_fault_vectors[i] = vectors[FIRST_FAULT_VECTOR + i];
+    vectors[FIRST_FAULT_VECTOR + i] = (uint32_t) &esphome_fault_entry;
+  }
+}
+
+#endif  // USE_LIBRETINY_VARIANT_RTL8720C
+
 }  // namespace esphome::libretiny

 #endif  // USE_LIBRETINY_CRASH_HANDLER
-#endif  // USE_BK72XX
+#endif  // USE_LIBRETINY
@@ -1,6 +1,6 @@
 #pragma once

-#ifdef USE_BK72XX
+#ifdef USE_LIBRETINY

 #include "esphome/core/defines.h"

@@ -14,7 +14,12 @@ void crash_handler_log();
 /// Returns true if crash data was found this boot, reading it first if needed.
 bool crash_handler_has_data();

+#ifdef USE_LIBRETINY_VARIANT_RTL8720C
+/// Put the crash recorder in front of the SDK's fault handlers.
+void crash_handler_install();
+#endif
+
 }  // namespace esphome::libretiny

 #endif  // USE_LIBRETINY_CRASH_HANDLER
-#endif  // USE_BK72XX
+#endif  // USE_LIBRETINY
@@ -91,6 +91,10 @@ async def pin_to_code(config: ConfigType) -> MockObj:
 # Called by writer.py; delegates to the shared libretiny implementation.
 def copy_files() -> None:
    libretiny.copy_files()
+
+
+# Called by the OTA upload; the shared libretiny implementation.
+ota_image = libretiny.device_ota_image
 '''

 BASE_CODE_BOARDS = '''
@@ -152,6 +156,7 @@ COMPONENT_CODEOWNERS = {
 # Extra imports per component; the CLI looks up process_stacktrace on the platform package.
 COMPONENT_EXTRA_IMPORTS = {
    "bk72xx": "from esphome.stacktrace import process_crash_handler_line as process_stacktrace  # noqa: F401",
+    "rtl87xx": "from esphome.stacktrace import process_crash_handler_line as process_stacktrace  # noqa: F401",
 }


@@ -1,6 +1,7 @@
 #ifdef USE_LIBRETINY

 #include "core.h"
+#include "crash_handler.h"
 #include "esphome/core/hal.h"
 #include "preferences.h"

@@ -27,6 +28,9 @@ namespace esphome {
 void arch_init() {
  libretiny::setup_preferences();
  lt_wdt_enable(10000L);
+#if defined(USE_LIBRETINY_CRASH_HANDLER) && defined(USE_LIBRETINY_VARIANT_RTL8720C)
+  libretiny::crash_handler_install();
+#endif
 #ifdef USE_BK72XX
  // BK72xx SDK creates the main Arduino task at priority 3, which is lower than
  // all WiFi (4-5), LwIP (4), and TCP/IP (7) tasks. This causes ~100ms loop
@@ -0,0 +1,98 @@
+"""Trim a LibreTiny UF2 package to the images a device writes over OTA.
+
+A LibreTiny UF2 holds one image per OTA scheme: the device schemes (single,
+dual slot 1, dual slot 2) and the flasher schemes used over UART. The device
+skips every block of an image whose scheme it does not run (uf2ota), so an
+image no device scheme writes is only sent to be thrown away. On single OTA
+chips (BK72xx, LN882H) that is the whole flasher image, more than half the
+package. The legacy OTA format names the same device partitions, so devices
+that read it write the same images.
+"""
+
+import struct
+
+UF2_BLOCK_SIZE = 512
+UF2_MAGIC_HEAD = struct.pack("<2I", 0x0A324655, 0x9E5D5157)
+UF2_MAGIC_TAIL = struct.pack("<I", 0x0AB16F30)
+UF2_FLAG_HAS_TAGS = 0x8000
+# Header words after the magic: flags, target address, payload length, block
+# number, block count
+_HEADER_WORDS = struct.Struct("<5I")
+_SEQ_OFFSET = 20
+_DATA_START = 32
+_DATA_END = UF2_BLOCK_SIZE - len(UF2_MAGIC_TAIL)
+
+# One partition index nibble per scheme, high nibble first: device single,
+# dual 1 and dual 2 are the first three, the flasher schemes the next three
+UF2_TAG_OTA_PART_INFO = 0xC0EE0C
+# Tags uf2ota acts on beyond the image they sit in; one in an image that
+# would be dropped leaves the package untrimmed
+_KEEP_TAGS = frozenset(
+    {
+        0x5D57D0,  # OTA_FORMAT_1
+        0x6C8492,  # OTA_FORMAT_2
+        0x6EC68A,  # OTA_PART_LIST
+        0xB948DE,  # BINPATCH
+        0x8288ED,  # FAL_PTABLE
+    }
+)
+
+
+def _tags(block: bytes, length: int) -> dict[int, bytes] | None:
+    """The extension tags after a block's payload, or None if malformed."""
+    tags: dict[int, bytes] = {}
+    pos = _DATA_START + length
+    while pos + 4 <= _DATA_END and (size := block[pos]):
+        if size < 4 or pos + size > _DATA_END:
+            return None
+        tags[int.from_bytes(block[pos + 1 : pos + 4], "little")] = block[
+            pos + 4 : pos + size
+        ]
+        pos += (size + 3) & ~3
+    return tags
+
+
+def device_ota_uf2(data: bytes) -> bytes | None:
+    """``data`` without the images no device scheme writes.
+
+    None when there is nothing to drop, or when ``data`` is not a well
+    formed UF2, so the caller sends the file as it is.
+    """
+    if len(data) % UF2_BLOCK_SIZE:
+        return None
+    count = len(data) // UF2_BLOCK_SIZE
+    kept: list[bytes] = []
+    keep = True
+    for seq in range(count):
+        block = data[seq * UF2_BLOCK_SIZE : (seq + 1) * UF2_BLOCK_SIZE]
+        flags, _, length, block_seq, block_count = _HEADER_WORDS.unpack_from(block, 8)
+        # A missing or repeated block must stay visible to the device, which
+        # takes the update size from the block count
+        if (
+            block[:8] != UF2_MAGIC_HEAD
+            or block[_DATA_END:] != UF2_MAGIC_TAIL
+            or _DATA_START + length > _DATA_END
+            or (block_seq, block_count) != (seq, count)
+        ):
+            return None
+        tags = _tags(block, length) if flags & UF2_FLAG_HAS_TAGS else {}
+        if tags is None:
+            return None
+        if (part_info := tags.get(UF2_TAG_OTA_PART_INFO)) is not None:
+            # uf2ota rejects a selector without the whole scheme map
+            if len(part_info) < 3:
+                return None
+            # Any of the three device scheme nibbles names a partition
+            keep = int.from_bytes(part_info[:2], "big") >> 4 != 0
+        if keep:
+            kept.append(block)
+        elif not _KEEP_TAGS.isdisjoint(tags):
+            # Without its image's OTA_PART_INFO the device would write this
+            # block to the partition of the image before it
+            return None
+    if len(kept) == count:
+        return None
+    out = bytearray(b"".join(kept))
+    for seq in range(len(kept)):
+        struct.pack_into("<2I", out, seq * UF2_BLOCK_SIZE + _SEQ_OFFSET, seq, len(kept))
+    return bytes(out)
@@ -77,3 +77,7 @@ async def pin_to_code(config: ConfigType) -> MockObj:
 # Called by writer.py; delegates to the shared libretiny implementation.
 def copy_files() -> None:
    libretiny.copy_files()
+
+
+# Called by the OTA upload; the shared libretiny implementation.
+ota_image = libretiny.device_ota_image
@@ -2,7 +2,6 @@ from collections.abc import Callable

 from esphome import automation
 import esphome.codegen as cg
-from esphome.components import audio
 from esphome.components.const import CONF_BITRATE
 import esphome.config_validation as cv
 from esphome.const import (
@@ -116,6 +115,8 @@ def validate_preferred_format(
    """

    def validator(config: ConfigType) -> ConfigType:
+        from esphome.components import audio
+
        # Inherit settings from audio device if not manually set
        inherit_property_from(CONF_NUM_CHANNELS, audio_device_key)(config)
        inherit_property_from(CONF_SAMPLE_RATE, audio_device_key)(config)
@@ -155,6 +156,8 @@ def request_codecs_for_format_configs(

    Public API for external components. Do not remove.
    """
+    from esphome.components import audio
+
    needed_formats: set[str] = set()
    need_all = False

@@ -107,12 +107,7 @@ void MipiDsi::setup() {
  }
 #endif
  if (this->reset_pin_ != nullptr) {
-    this->reset_pin_->setup();
-    this->reset_pin_->digital_write(true);
-    delay(5);
-    this->reset_pin_->digital_write(false);
-    delay(5);
-    this->reset_pin_->digital_write(true);
+    pulse_reset(this->reset_pin_, {.high_before_ms = 5, .low_ms = 5});
  } else {
    esp_lcd_panel_io_tx_param(this->io_handle_, SW_RESET_CMD, nullptr, 0);
  }
@@ -33,14 +33,7 @@ void MipiRgb::setup_enables_() {
    }
    delay(10);
  }
-  if (this->reset_pin_ != nullptr) {
-    this->reset_pin_->setup();
-    this->reset_pin_->digital_write(true);
-    delay(5);
-    this->reset_pin_->digital_write(false);
-    delay(5);
-    this->reset_pin_->digital_write(true);
-  }
+  pulse_reset(this->reset_pin_, {.high_before_ms = 5, .low_ms = 5});
 }

 #ifdef USE_SPI
@@ -151,15 +151,8 @@ class MipiSpi : public display::Display,
      pin->setup();
      pin->digital_write(true);
    }
-    if (this->reset_pin_ != nullptr) {
-      this->reset_pin_->setup();
-      this->reset_pin_->digital_write(true);
-      delay(5);
-      this->reset_pin_->digital_write(false);
-      delay(5);
-      this->reset_pin_->digital_write(true);
-      // required delay after reset is already in the init sequence, don't duplicate
-    }
+    // required delay after reset is already in the init sequence, don't duplicate
+    pulse_reset(this->reset_pin_, {.high_before_ms = 5, .low_ms = 5});

    // need to know when the display is ready for SLPOUT command - will be 120ms after reset
    auto when = millis() + 120;
@@ -226,6 +226,12 @@ def set_sdkconfig_options(config: ConfigType) -> None:
    add_idf_sdkconfig_option("CONFIG_OPENTHREAD_DNS64_CLIENT", not border_router)
    add_idf_sdkconfig_option("CONFIG_OPENTHREAD_SRP_CLIENT", not border_router)
    if not border_router:
+        # The DNS64 hook breaks IPv6 literal lookups; see openthread_esp.cpp.
+        # Fixed upstream on ESP-IDF master, remove this once a release includes
+        # "fix(openthread): bypass DNS64 synthesis for IPv6 literals",
+        # https://github.com/espressif/esp-idf/commit/87b133ee1a7e91c1ee95176b4733b7f72236e5da
+        # (merged in https://github.com/espressif/esp-idf/commit/83239509832bf5d427935cd365f4b62a320e8765)
+        cg.add_build_flag("-Wl,--wrap=lwip_hook_dns_external_resolve")
        add_idf_sdkconfig_option("CONFIG_OPENTHREAD_SRP_CLIENT_MAX_SERVICES", 5)

    if border_router:
@@ -35,6 +35,10 @@
 #include "esp_vfs_eventfd.h"
 #include "freertos/FreeRTOS.h"
 #include "freertos/task.h"
+#ifndef USE_OPENTHREAD_BORDER_ROUTER
+#include "lwip/dns.h"
+#include "lwip/ip_addr.h"
+#endif

 ESPHOME_LOG_TAG(TAG, "openthread");

@@ -489,4 +493,26 @@ InstanceLock::~InstanceLock() {
 }

 }  // namespace esphome::openthread
+
+#ifndef USE_OPENTHREAD_BORDER_ROUTER
+// ESP-IDF's DNS64 hook runs before lwIP's literal address check and retries every lookup as IPv4,
+// so an IPv6 literal never resolves. Let lwIP handle those; linked with --wrap from __init__.py.
+// Fixed upstream on ESP-IDF master, remove this once a release includes
+// "fix(openthread): bypass DNS64 synthesis for IPv6 literals",
+// https://github.com/espressif/esp-idf/commit/87b133ee1a7e91c1ee95176b4733b7f72236e5da
+// (merged in https://github.com/espressif/esp-idf/commit/83239509832bf5d427935cd365f4b62a320e8765)
+// NOLINTBEGIN(bugprone-reserved-identifier,cert-dcl37-c,cert-dcl51-cpp,readability-identifier-naming)
+extern "C" {
+int __real_lwip_hook_dns_external_resolve(const char *name, ip_addr_t *addr, dns_found_callback found,
+                                          void *callback_arg, u8_t addrtype, err_t *err);
+int __wrap_lwip_hook_dns_external_resolve(const char *name, ip_addr_t *addr, dns_found_callback found,
+                                          void *callback_arg, u8_t addrtype, err_t *err) {
+  ip_addr_t literal;
+  if (ipaddr_aton(name, &literal) && IP_IS_V6_VAL(literal))
+    return 0;
+  return __real_lwip_hook_dns_external_resolve(name, addr, found, callback_arg, addrtype, err);
+}
+}  // extern "C"
+// NOLINTEND(bugprone-reserved-identifier,cert-dcl37-c,cert-dcl51-cpp,readability-identifier-naming)
+#endif  // USE_OPENTHREAD_BORDER_ROUTER
 #endif
@@ -9,23 +9,10 @@ ESPHOME_LOG_TAG(TAG, "pcd_8544");

 void PCD8544::setup_pins_() {
  this->spi_setup();
-  this->init_reset_();
+  pulse_reset(this->reset_pin_, {.high_before_ms = 1, .low_ms = 10});
  this->dc_pin_->setup();
 }

-void PCD8544::init_reset_() {
-  if (this->reset_pin_ != nullptr) {
-    this->reset_pin_->setup();
-    this->reset_pin_->digital_write(true);
-    delay(1);
-    // Trigger Reset
-    this->reset_pin_->digital_write(false);
-    delay(10);
-    // Wake up
-    this->reset_pin_->digital_write(true);
-  }
-}
-
 void PCD8544::initialize() {
  this->init_internal_(this->get_buffer_length_());

@@ -57,8 +57,6 @@ class PCD8544 final : public display::DisplayBuffer,

  void setup_pins_();

-  void init_reset_();
-
  size_t get_buffer_length_();

  void start_command_();
@@ -29,6 +29,7 @@ from esphome.components.libretiny.const import (
 )
 from esphome.core import CORE
 from esphome.cpp_generator import MockObj
+from esphome.stacktrace import process_crash_handler_line as process_stacktrace  # noqa: F401
 from esphome.types import ConfigType

 from .boards import RTL87XX_BOARD_PINS, RTL87XX_BOARDS
@@ -77,3 +78,7 @@ async def pin_to_code(config: ConfigType) -> MockObj:
 # Called by writer.py; delegates to the shared libretiny implementation.
 def copy_files() -> None:
    libretiny.copy_files()
+
+
+# Called by the OTA upload; the shared libretiny implementation.
+ota_image = libretiny.device_ota_image
@@ -122,18 +122,7 @@ void SH1122::fill(Color color) {
  const uint8_t fill = (color4 & SH1122_COLORMASK) | ((color4 & SH1122_COLORMASK) << SH1122_COLORSHIFT);
  memset(this->buffer_, fill, this->get_buffer_length_());
 }
-void SH1122::init_reset_() {
-  if (this->reset_pin_ != nullptr) {
-    this->reset_pin_->setup();
-    this->reset_pin_->digital_write(true);
-    delay(1);
-    // Trigger Reset
-    this->reset_pin_->digital_write(false);
-    delay(10);
-    // Wake up
-    this->reset_pin_->digital_write(true);
-  }
-}
+void SH1122::init_reset_() { pulse_reset(this->reset_pin_, {.high_before_ms = 1, .low_ms = 10}); }
 const LogString *SH1122::model_str_() {
  switch (this->model_) {
    case SH1122_MODEL_256_64:
@@ -355,18 +355,7 @@ void SSD1306::fill(Color color) {
  for (uint32_t i = 0; i < this->get_buffer_length_(); i++)
    this->buffer_[i] = fill;
 }
-void SSD1306::init_reset_() {
-  if (this->reset_pin_ != nullptr) {
-    this->reset_pin_->setup();
-    this->reset_pin_->digital_write(true);
-    delay(1);
-    // Trigger Reset
-    this->reset_pin_->digital_write(false);
-    delay(10);
-    // Wake up
-    this->reset_pin_->digital_write(true);
-  }
-}
+void SSD1306::init_reset_() { pulse_reset(this->reset_pin_, {.high_before_ms = 1, .low_ms = 10}); }
 const LogString *SSD1306::model_str_() {
  return ModelStrings::get_log_str(static_cast<uint8_t>(this->model_), ModelStrings::LAST_INDEX);
 }
@@ -184,18 +184,7 @@ void SSD1322::fill(Color color) {
  for (uint32_t i = 0; i < this->get_buffer_length_(); i++)
    this->buffer_[i] = fill;
 }
-void SSD1322::init_reset_() {
-  if (this->reset_pin_ != nullptr) {
-    this->reset_pin_->setup();
-    this->reset_pin_->digital_write(true);
-    delay(1);
-    // Trigger Reset
-    this->reset_pin_->digital_write(false);
-    delay(10);
-    // Wake up
-    this->reset_pin_->digital_write(true);
-  }
-}
+void SSD1322::init_reset_() { pulse_reset(this->reset_pin_, {.high_before_ms = 1, .low_ms = 10}); }
 const char *SSD1322::model_str_() {
  switch (this->model_) {
    case SSD1322_MODEL_256_64:
@@ -211,18 +211,7 @@ void SSD1325::fill(Color color) {
  for (uint32_t i = 0; i < this->get_buffer_length_(); i++)
    this->buffer_[i] = fill;
 }
-void SSD1325::init_reset_() {
-  if (this->reset_pin_ != nullptr) {
-    this->reset_pin_->setup();
-    this->reset_pin_->digital_write(true);
-    delay(1);
-    // Trigger Reset
-    this->reset_pin_->digital_write(false);
-    delay(10);
-    // Wake up
-    this->reset_pin_->digital_write(true);
-  }
-}
+void SSD1325::init_reset_() { pulse_reset(this->reset_pin_, {.high_before_ms = 1, .low_ms = 10}); }
 const char *SSD1325::model_str_() {
  switch (this->model_) {
    case SSD1325_MODEL_128_32:
@@ -160,18 +160,7 @@ void SSD1327::fill(Color color) {
  for (uint32_t i = 0; i < this->get_buffer_length_(); i++)
    this->buffer_[i] = fill;
 }
-void SSD1327::init_reset_() {
-  if (this->reset_pin_ != nullptr) {
-    this->reset_pin_->setup();
-    this->reset_pin_->digital_write(true);
-    delay(1);
-    // Trigger Reset
-    this->reset_pin_->digital_write(false);
-    delay(10);
-    // Wake up
-    this->reset_pin_->digital_write(true);
-  }
-}
+void SSD1327::init_reset_() { pulse_reset(this->reset_pin_, {.high_before_ms = 1, .low_ms = 10}); }
 const char *SSD1327::model_str_() {
  switch (this->model_) {
    case SSD1327_MODEL_128_128:
@@ -142,17 +142,6 @@ void SSD1331::fill(Color color) {
    }
  }
 }
-void SSD1331::init_reset_() {
-  if (this->reset_pin_ != nullptr) {
-    this->reset_pin_->setup();
-    this->reset_pin_->digital_write(true);
-    delay(1);
-    // Trigger Reset
-    this->reset_pin_->digital_write(false);
-    delay(10);
-    // Wake up
-    this->reset_pin_->digital_write(true);
-  }
-}
+void SSD1331::init_reset_() { pulse_reset(this->reset_pin_, {.high_before_ms = 1, .low_ms = 10}); }

 }  // namespace esphome::ssd1331_base
@@ -174,18 +174,7 @@ void SSD1351::fill(Color color) {
    }
  }
 }
-void SSD1351::init_reset_() {
-  if (this->reset_pin_ != nullptr) {
-    this->reset_pin_->setup();
-    this->reset_pin_->digital_write(true);
-    delay(1);
-    // Trigger Reset
-    this->reset_pin_->digital_write(false);
-    delay(10);
-    // Wake up
-    this->reset_pin_->digital_write(true);
-  }
-}
+void SSD1351::init_reset_() { pulse_reset(this->reset_pin_, {.high_before_ms = 1, .low_ms = 10}); }
 const char *SSD1351::model_str_() {
  switch (this->model_) {
    case SSD1351_MODEL_128_96:
@@ -9,12 +9,8 @@ ESPHOME_LOG_TAG(TAG, "st7123.touchscreen");

 void ST7123Touchscreen::setup() {
  if (this->reset_pin_ != nullptr) {
-    this->reset_pin_->setup();
-    this->reset_pin_->digital_write(true);
-    delay(5);
-    this->reset_pin_->digital_write(false);  // TP_RESX is active low, assert for at least tRSTW (2ms)
-    delay(5);
-    this->reset_pin_->digital_write(true);
+    // TP_RESX is active low, assert for at least tRSTW (2ms)
+    pulse_reset(this->reset_pin_, {.high_before_ms = 5, .low_ms = 5});
    // The controller needs up to 20ms to initialize after reset before it can be accessed.
    this->setup_time_ = millis() + 30;
  }
@@ -137,18 +137,7 @@ void ST7567::fill(Color color) {
  memset(buffer_, fill, this->get_buffer_length_());
 }

-void ST7567::init_reset_() {
-  if (this->reset_pin_ != nullptr) {
-    this->reset_pin_->setup();
-    this->reset_pin_->digital_write(true);
-    delay(1);
-    // Trigger Reset
-    this->reset_pin_->digital_write(false);
-    delay(10);
-    // Wake up
-    this->reset_pin_->digital_write(true);
-  }
-}
+void ST7567::init_reset_() { pulse_reset(this->reset_pin_, {.high_before_ms = 1, .low_ms = 10}); }

 const char *ST7567::model_str_() { return "ST7567 128x64"; }

@@ -54,10 +54,7 @@ void TT21100Touchscreen::setup() {
  }

  // Perform reset if necessary
-  if (this->reset_pin_ != nullptr) {
-    this->reset_pin_->setup();
-    this->reset_();
-  }
+  pulse_reset(this->reset_pin_, {.low_ms = 10, .high_after_ms = 10});

  // Update display dimensions if they were updated during display setup
  if (this->display_ != nullptr) {
@@ -122,15 +119,6 @@ void TT21100Touchscreen::update_touches() {
  }
 }

-void TT21100Touchscreen::reset_() {
-  if (this->reset_pin_ != nullptr) {
-    this->reset_pin_->digital_write(false);
-    delay(10);
-    this->reset_pin_->digital_write(true);
-    delay(10);
-  }
-}
-
 void TT21100Touchscreen::dump_config() {
  ESP_LOGCONFIG(TAG, "TT21100 Touchscreen:");
  LOG_I2C_DEVICE(this);
@@ -28,8 +28,6 @@ class TT21100Touchscreen final : public Touchscreen, public i2c::I2CDevice {
  void register_button_listener(TT21100ButtonListener *listener) { this->button_listeners_.push_back(listener); }

 protected:
-  void reset_();
-
  void update_touches() override;

  InternalGPIOPin *interrupt_pin_;
@@ -774,7 +774,7 @@
 #endif

 #ifdef USE_LIBRETINY
-#ifdef USE_BK72XX
+#if defined(USE_BK72XX) || defined(USE_LIBRETINY_VARIANT_RTL8720C)
 #define USE_LIBRETINY_CRASH_HANDLER
 #endif
 #define USE_BK72XX_BLE
@@ -1,4 +1,5 @@
 #include "esphome/core/gpio.h"
+#include "esphome/core/hal.h"
 #include "esphome/core/log.h"

 namespace esphome {
@@ -18,4 +19,18 @@ void log_pin(const char *tag, const char *prefix, GPIOPin *pin) {
 }
 #endif

+namespace internal {
+void pulse_reset_ms(GPIOPin *pin, uint32_t high_before_ms, uint32_t low_ms, uint32_t high_after_ms) {
+  if (pin == nullptr)
+    return;
+  pin->setup();
+  const uint32_t waits[] = {high_before_ms, low_ms, high_after_ms};
+  // Level for each step is high, low, high; a zero high_before_ms skips the first step
+  for (size_t i = high_before_ms == 0; i < 3; i++) {
+    pin->digital_write(i != 1);
+    delay(waits[i]);
+  }
+}
+}  // namespace internal
+
 }  // namespace esphome
@@ -144,4 +144,24 @@ void log_pin(const char *tag, const __FlashStringHelper *prefix, GPIOPin *pin);
 void log_pin(const char *tag, const char *prefix, GPIOPin *pin);
 #endif

+/// Waveform for pulse_reset(): high, low, then high again. Name the fields at the call site, e.g.
+/// pulse_reset(this->reset_pin_, {.high_before_ms = 1, .low_ms = 10}).
+struct ResetPulse {
+  uint32_t high_before_ms{0};  ///< Drive high this long first; skipped when 0
+  uint32_t low_ms{0};          ///< Hold the pin low (in reset) this long
+  uint32_t high_after_ms{0};   ///< Wait this long after releasing the pin
+};
+
+namespace internal {
+/// Out-of-line body of pulse_reset(); call that instead.
+void pulse_reset_ms(GPIOPin *pin, uint32_t high_before_ms, uint32_t low_ms, uint32_t high_after_ms);
+}  // namespace internal
+
+/// Set up an active-low reset pin and drive it through \p pulse. Blocks for the whole sequence;
+/// does nothing when pin is null. Inline so the fields reach pulse_reset_ms() in registers
+/// (RISC-V would otherwise pass the struct by reference).
+ESPHOME_ALWAYS_INLINE inline void pulse_reset(GPIOPin *pin, ResetPulse pulse) {
+  internal::pulse_reset_ms(pin, pulse.high_before_ms, pulse.low_ms, pulse.high_after_ms);
+}
+
 }  // namespace esphome
@@ -1,385 +0,0 @@
-"""Rewrite the OTA encryption key in place: the `key:` line the loader read
-it from, or the secrets.yaml line a `!secret` on it points to."""
-
-from __future__ import annotations
-
-from dataclasses import dataclass, field
-import logging
-import os
-from pathlib import Path
-import re
-
-from esphome.components.noise import static_encryption_key
-from esphome.const import CONF_API, CONF_ENCRYPTION, CONF_KEY
-from esphome.core import CORE, EsphomeError
-from esphome.espota2 import CONF_OLD_KEY, esphome_ota_items
-from esphome.types import ConfigType
-from esphome.yaml_edit import (
-    LineEdit,
-    editable_file,
-    field_line_re,
-    own_documents,
-    read_text,
-    rewrite,
-    source_line,
-    source_of,
-)
-from esphome.yaml_util import secrets_path_for
-
-_LOGGER = logging.getLogger(__name__)
-_DOMAIN = "ota_key_edit"
-
-
-@dataclass
-class KeyEdit(LineEdit):
-    """A line edit with the secret it holds and the other configs using it."""
-
-    secret: str | None = None
-    shared_with: list[Path] = field(default_factory=list)
-
-
-def _api_encryption() -> ConfigType:
-    return ((CORE.raw_config or {}).get(CONF_API) or {}).get(CONF_ENCRYPTION) or {}
-
-
-def _key_blocks(raw: ConfigType, key: str) -> list[ConfigType]:
-    """The api and esphome ota ``encryption:`` mappings whose key is ``key``."""
-    blocks = [raw.get(CONF_API) or {}, *esphome_ota_items(raw)]
-    found = []
-    for block in blocks:
-        if (node := static_encryption_key(block)) is None or str(node) != key:
-            continue
-        encryption = block[CONF_ENCRYPTION]
-        if source_of(encryption, CONF_KEY) is not None:
-            found.append(encryption)
-        elif hasattr(next(k for k in encryption if k == CONF_KEY), "esp_range"):
-            # Read from a file but not on a line of its own, as with a merge
-            # key; only a key validation added, which has no range at all,
-            # is passed over (a bare ota block takes the api key)
-            raise EsphomeError(
-                f"The '{CONF_KEY}' of an '{CONF_ENCRYPTION}:' block is not on a "
-                "line of its own; edit the key by hand"
-            )
-    return found
-
-
-_INDENT_RE = re.compile(r"\s*")
-_KEY_FIELD_RE = re.compile(rf"^\s*{CONF_KEY}\s*:")
-_BARE_BLOCK_RE = re.compile(rf"^\s*{CONF_ENCRYPTION}\s*:\s*(#.*)?$")
-_YAML_SUFFIXES = (".yaml", ".yml")
-_INCLUDE_DIR_RE = re.compile(r"!include_dir_\w+\s+(?P<dir>[^\s#]+)")
-# An include or package file whose path holds a substitution
-_SUBSTITUTED_INCLUDE_RE = re.compile(
-    r"(?:!include\w*\s+|\bfile:\s*)[\"']?(?P<target>[^\s#\"']*\$[^\s#\"']*)"
-)
-
-
-def _secret_ref(name: str = r"[^\s\"']+") -> str:
-    """`!secret name` as the whole scalar: `!secret ota key` is not `ota`, and
-    a comment needs whitespace before its `#`, so `a#b` is one name."""
-    return rf"!secret\s+(?P<q>[\"']?)(?P<name>{name})(?P=q)(?=\s+#|\s*$)"
-
-
-def _secret_field_re(name: str) -> re.Pattern[str]:
-    """``name: !secret <any>``."""
-    return re.compile(rf"^\s*{name}\s*:\s*{_secret_ref()}")
-
-
-def _secret_use_re(name: str) -> re.Pattern[str]:
-    """A use of the secret ``name`` anywhere on a line."""
-    return re.compile(_secret_ref(re.escape(name)))
-
-
-_KEY_SECRET_RE = _secret_field_re(CONF_KEY)
-_OLD_KEY_SECRET_RE = _secret_field_re(CONF_OLD_KEY)
-
-
-def _indent(text: str) -> str:
-    return _INDENT_RE.match(text).group()
-
-
-def _without_comment(text: str) -> str:
-    """The line up to its comment: a `#` after whitespace that is not inside
-    a quoted scalar, so `!include "shared # ota.yaml"` keeps its name."""
-    quote = ""
-    for i, char in enumerate(text):
-        if quote:
-            if char == quote:
-                quote = ""
-        elif char in "\"'":
-            quote = char
-        elif char == "#" and (i == 0 or text[i - 1].isspace()):
-            return text[:i]
-    return text
-
-
-def _file_use_re(file_name: str) -> re.Pattern[str]:
-    """The file name at a path boundary; `wifi_api.yaml` is not `api.yaml`."""
-    return re.compile(rf"(?<![\w.-]){re.escape(file_name)}(?![\w.])")
-
-
-def _root_indent(lines: list[str]) -> str:
-    """The indent of the first content line; a root mapping may be indented."""
-    content = (
-        t for t in lines if t.strip() and not t.strip().startswith(("#", "---", "%"))
-    )
-    return next((_indent(t) for t in content), "")
-
-
-def _config_texts() -> tuple[list[tuple[Path, str]], list[str]]:
-    """Every yaml file under the config dir with its text, and what could
-    not be read (a symlinked directory is not walked); walked once per run."""
-    if (found := CORE.data.get(_DOMAIN)) is None:
-        found = CORE.data[_DOMAIN] = _walk_config_texts()
-    return found
-
-
-def _walk_config_texts() -> tuple[list[tuple[Path, str]], list[str]]:
-    data_dir = CORE.data_dir.resolve()
-    found = []
-    skipped: list[str] = []
-
-    def skip(what: object) -> None:
-        skipped.append(str(what))
-        _LOGGER.debug("Skipping %s", what)
-
-    for dirpath, dirnames, filenames in os.walk(
-        CORE.config_dir.resolve(), onerror=skip
-    ):
-        for name in list(dirnames):
-            path = Path(dirpath, name)
-            if path == data_dir:
-                dirnames.remove(name)
-            elif path.is_symlink():
-                dirnames.remove(name)
-                skip(f"{path} is a link")
-        for filename in filenames:
-            path = Path(dirpath, filename)
-            if path.suffix not in _YAML_SUFFIXES:
-                continue
-            try:
-                text = read_text(path)
-            except EsphomeError as err:
-                skip(err)
-                continue
-            # Searched, never edited: comments do not count
-            found.append((path, "\n".join(map(_without_comment, text.splitlines()))))
-    return sorted(found), skipped
-
-
-def with_sharers(edits: list[KeyEdit]) -> list[str]:
-    """Fill in ``shared_with``: the other configs that use each rewritten
-    line's secret or include, directly or through further includes. Returns
-    what the scan could not follow (unreadable files, directory and
-    substituted includes)."""
-    main = CORE.config_path.resolve()
-    own = own_documents()
-    # This config's own includes stay in: another device may reach a line
-    # through one of them
-    texts, skipped = _config_texts()
-    texts = [(path, text) for path, text in texts if path.resolve() != main]
-    unchecked = [*skipped]
-    for path, text in texts:
-        if match := _INCLUDE_DIR_RE.search(text):
-            unchecked.append(f"{path} includes the directory {match['dir']}")
-        if match := _SUBSTITUTED_INCLUDE_RE.search(text):
-            unchecked.append(
-                f"{path} includes {match['target']} through a substitution"
-            )
-    for edit in edits:
-        if edit.insert_after and edit.secret:
-            continue  # a new secrets line has no users yet
-        ref = (
-            _secret_use_re(edit.secret) if edit.secret else _file_use_re(edit.path.name)
-        )
-        users = {path for path, text in texts if ref.search(text)}
-        frontier = users
-        while frontier:
-            frontier = {
-                path
-                for name in {p.name for p in frontier}
-                for path, text in texts
-                if path not in users and _file_use_re(name).search(text)
-            }
-            users |= frontier
-        edit.shared_with = sorted(p for p in users if p.resolve() not in own)
-    return unchecked
-
-
-def _secret_file(doc: Path) -> Path:
-    return editable_file(secrets_path_for(doc))
-
-
-def _secret_line(
-    secrets_path: Path, name: str, value: str | None = None
-) -> tuple[int, re.Match[str]] | None:
-    """The root level ``name:`` line, None when absent."""
-    lines = read_text(secrets_path).splitlines()
-    line_re = field_line_re(name, value, re.escape(_root_indent(lines)))
-    return next(
-        ((i, m) for i, text in enumerate(lines) if (m := line_re.match(text))), None
-    )
-
-
-def _secret_rewrite(
-    secrets_path: Path, name: str, value: str, expect: str | None = None
-) -> KeyEdit | None:
-    """Set the root level ``name:`` line to ``value``; None when it is not a
-    plain scalar holding ``expect``."""
-    if (hit := _secret_line(secrets_path, name, expect)) is None:
-        return None
-    i, m = hit
-    return KeyEdit(secrets_path, i, m.string, rewrite(m, value), secret=name)
-
-
-def _secret_insert(secrets_path: Path, after: str, name: str, value: str) -> KeyEdit:
-    """Add ``name: value`` under the root level ``after`` line."""
-    if (hit := _secret_line(secrets_path, after)) is None:
-        raise EsphomeError(
-            f"No plain '{after}:' line in {secrets_path}; edit it by hand"
-        )
-    i, m = hit
-    return KeyEdit(
-        secrets_path,
-        i,
-        m.string,
-        f'{_indent(m.string)}{name}: "{value}"',
-        insert_after=True,
-        secret=name,
-    )
-
-
-def locate_key_edits(old_key: str, new_key: str) -> list[KeyEdit]:
-    """The edits that replace the current key; refuses a substitution, flow
-    mapping or remote package."""
-    literal_re = field_line_re(CONF_KEY, old_key)
-    blocks = _key_blocks(CORE.raw_config or {}, old_key)
-    if not blocks:
-        raise EsphomeError("The current key was not found on a line of the yaml")
-    edits: list[KeyEdit] = []
-    key_lines: set[tuple[Path, int]] = set()
-    for block in blocks:
-        doc, line_no, text = source_line(block, CONF_KEY)
-        key_lines.add((doc.resolve(), line_no))
-        if match := _KEY_SECRET_RE.match(text):
-            name, secrets_path = match["name"], _secret_file(doc)
-            edit = _secret_rewrite(secrets_path, name, new_key, expect=old_key)
-            if edit is None:
-                raise EsphomeError(
-                    f"No '{name}:' line with the current key in {secrets_path}"
-                )
-        elif match := literal_re.match(text):
-            edit = KeyEdit(doc, line_no, text, rewrite(match, new_key))
-        else:
-            raise EsphomeError(
-                f"{doc}:{line_no + 1} does not hold the key as a plain value or "
-                "a !secret; edit the key by hand"
-            )
-        # The api and ota blocks may share one !secret line
-        if edit not in edits:
-            edits.append(edit)
-    for edit in edits:
-        if edit.secret:
-            _refuse_other_own_uses(edit.secret, key_lines)
-    return edits
-
-
-def _refuse_other_own_uses(name: str, allowed: set[tuple[Path, int]]) -> None:
-    """The secret may only feed the ``allowed`` lines; a wifi password on it
-    would change too."""
-    ref = _secret_use_re(name)
-    for doc in own_documents():
-        for i, text in enumerate(read_text(doc).splitlines()):
-            if ref.search(_without_comment(text)) and (doc, i) not in allowed:
-                raise EsphomeError(
-                    f"'{name}' is also used at {doc}:{i + 1}; a rotation would "
-                    "change that value too, edit the key by hand"
-                )
-
-
-def _secret_name(block: ConfigType) -> str | None:
-    """The secrets file name a ``key: !secret <name>`` line of ``block`` holds."""
-    if source_of(block, CONF_KEY) is None:
-        return None
-    match = _KEY_SECRET_RE.match(source_line(block, CONF_KEY)[2])
-    return match["name"] if match else None
-
-
-def old_key_edit(old_key: str) -> list[KeyEdit]:
-    """Set ``old_key:`` on the esphome ota block, rewriting or adding it. A
-    key from secrets.yaml stays there as ``<name>_old``. Of several entries
-    the one with its own old_key or key line gets it."""
-    items = [
-        (item, item[CONF_ENCRYPTION] or {})
-        for item in esphome_ota_items(CORE.raw_config or {})
-        if CONF_ENCRYPTION in item
-    ]
-    if not items:
-        raise EsphomeError("The esphome OTA platform has no 'encryption:' block")
-    item, encryption = next(
-        (
-            pair
-            for name in (CONF_OLD_KEY, CONF_KEY)
-            for pair in items
-            if source_of(pair[1], name) is not None
-        ),
-        items[0],
-    )
-    if (existing := encryption.get(CONF_OLD_KEY)) is not None:
-        doc, line_no, text = source_line(encryption, CONF_OLD_KEY)
-        if match := _OLD_KEY_SECRET_RE.match(text):
-            name, secrets_path = match["name"], _secret_file(doc)
-            if (edit := _secret_rewrite(secrets_path, name, old_key)) is None:
-                raise EsphomeError(
-                    f"No plain '{name}:' line in {secrets_path}; edit it by hand"
-                )
-            _refuse_other_own_uses(name, {(doc.resolve(), line_no)})
-            return [edit]
-        match = field_line_re(CONF_OLD_KEY, str(existing)).match(text)
-        if match is None:
-            raise EsphomeError(
-                f"{doc}:{line_no + 1} does not hold '{CONF_OLD_KEY}' as a plain "
-                "value; edit it by hand"
-            )
-        return [KeyEdit(doc, line_no, text, rewrite(match, old_key))]
-    # Indent from the block's own key line, else one level under `encryption:`
-    if source_of(encryption, CONF_KEY) is not None:
-        mapping, anchor, anchor_re, step = encryption, CONF_KEY, _KEY_FIELD_RE, ""
-    else:
-        mapping, anchor, anchor_re, step = item, CONF_ENCRYPTION, _BARE_BLOCK_RE, "  "
-    doc, line_no, text = source_line(mapping, anchor)
-    if not anchor_re.match(text):
-        raise EsphomeError(
-            f"{doc}:{line_no + 1} does not hold '{anchor}:' as a block line; "
-            "edit it by hand"
-        )
-    # The block's own key line decides; a bare block takes after the api key
-    keyed = encryption if mapping is encryption else _api_encryption()
-    if (secret := _secret_name(keyed)) is None:
-        rendered = f'{CONF_OLD_KEY}: "{old_key}"'
-        extra = []
-    else:
-        rendered = f"{CONF_OLD_KEY}: !secret {secret}_old"
-        secrets_path = _secret_file(doc)  # the file that gets old_key
-        # An existing `<secret>_old` line is reused only when nothing uses it
-        if (
-            kept := _secret_rewrite(secrets_path, f"{secret}_old", old_key)
-        ) is not None:
-            ref = _secret_use_re(f"{secret}_old")
-            texts, skipped = _config_texts()
-            if skipped:  # an unread file might use the line
-                raise EsphomeError(
-                    "Could not read every configuration to check the secret: "
-                    + skipped[0]
-                )
-            if users := [p for p, t in texts if ref.search(t)]:
-                raise EsphomeError(
-                    f"'{secret}_old:' in {secrets_path} is used by "
-                    + ", ".join(str(p) for p in users)
-                    + f"; rename it or add '{CONF_OLD_KEY}' by hand"
-                )
-        extra = [kept or _secret_insert(secrets_path, secret, f"{secret}_old", old_key)]
-    edit = KeyEdit(
-        doc, line_no, text, f"{_indent(text)}{step}{rendered}", insert_after=True
-    )
-    return [edit, *extra]
@@ -26,17 +26,20 @@ from esphome.const import (
    PLATFORM_BK72XX,
    PLATFORM_ESP32,
    PLATFORM_ESP8266,
+    PLATFORM_LN882X,
    PLATFORM_NRF52,
    PLATFORM_RP2,
+    PLATFORM_RTL87XX,
    Platform,
 )

 _LOGGER = logging.getLogger(__name__)

-# Hooks whose loss only degrades diagnostics; skipping one of these is
-# logged at debug, while skipping a hook that changes what the CLI does
-# (upload method, log transport) warns. A new hook is loud by default.
-COSMETIC_HOOKS: Final = frozenset({"process_stacktrace"})
+# Hooks whose loss only degrades diagnostics or upload size; skipping one
+# of these is logged at debug, while skipping a hook that changes what the
+# CLI does (upload method, log transport) warns. A new hook is loud by
+# default.
+COSMETIC_HOOKS: Final = frozenset({"process_stacktrace", "ota_image"})

 # Platforms whose firmware prints the crash handler report share one gate.
 _CRASH_HANDLER_GATE = r"0x[0-9a-fA-F]{3,}\b|CRASH DETECTED ON PREVIOUS BOOT"
@@ -64,6 +67,7 @@ STACKTRACE_GATES: Final[dict[str, str]] = {
    ),
    PLATFORM_RP2: _CRASH_HANDLER_GATE,
    PLATFORM_BK72XX: _CRASH_HANDLER_GATE,
+    PLATFORM_RTL87XX: _CRASH_HANDLER_GATE,
    PLATFORM_NRF52: r"0x[0-9a-fA-F]{3,}\b|Last crash:",
 }

@@ -71,6 +75,8 @@ PLATFORM_HOOKS: Final[dict[str, frozenset[str]]] = {
    "show_logs": frozenset({PLATFORM_NRF52}),
    "upload_program": frozenset({PLATFORM_NRF52}),
    "process_stacktrace": frozenset(STACKTRACE_GATES),
+    # The file an app update sends, when it is not the build output itself
+    "ota_image": frozenset({PLATFORM_BK72XX, PLATFORM_LN882X, PLATFORM_RTL87XX}),
 }


@@ -152,7 +152,7 @@ def _addr2line(tool: str, elf: Path, addr: str) -> str:
 def process_crash_handler_line(
    config: ConfigType, line: str, backtrace_state: bool
 ) -> bool:
-    """Decode a crash handler report with addr2line; the process_stacktrace hook of rp2 and bk72xx."""
+    """Decode a crash handler report with addr2line; the process_stacktrace hook of rp2, bk72xx and rtl87xx."""
    if CRASH_HANDLER_RE.search(line):
        _LOGGER.error("Crash detected - decoding addresses")
        return True
@@ -966,7 +966,9 @@ class ESPHomeLoaderMixin:
            if self.name == CORE.config_path:
                raise e
            try:
-                secrets = self.yaml_loader(CORE.config_path.parent / SECRET_YAML)
+                main_config_dir = CORE.config_path.parent
+                main_secret_yml = main_config_dir / SECRET_YAML
+                secrets = self.yaml_loader(main_secret_yml)
            except EsphomeError as er:
                raise EsphomeError(f"{e}\n{er}") from er

@@ -1355,20 +1357,6 @@ def find_files(directory: Path, pattern: str) -> Iterator[Path]:
                yield filename


-def secrets_path_for(document: Path) -> Path:
-    """The secrets.yaml a ``!secret`` in ``document`` resolves against: the
-    one beside the document when it loads, else the main config's, the
-    way construct_secret falls back."""
-    beside = document.parent / SECRET_YAML
-    if document == CORE.config_path:
-        return beside
-    try:
-        load_yaml(beside, clear_secrets=False, track_document_range=False)
-    except EsphomeError:
-        return CORE.config_path.parent / SECRET_YAML
-    return beside
-
-
 def is_secret(value):
    try:
        return _SECRET_VALUES[str(value)]
@@ -57,6 +57,7 @@ PLATFORM_BUILD_SECONDS = {
    "esp32-idf": 55,
    "esp32-p4-idf": 55,
    "ln882x-ard": 55,
+    "rtl8720c-ard": 55,
    "esp32-s3-idf": 60,
    "bk72xx-ard": 70,
    "bk7238-ard": 70,
@@ -0,0 +1,5 @@
+esphome:
+  name: test
+
+rtl87xx:
+  board: generic-rtl8710bn-2mb-788k
@@ -0,0 +1,5 @@
+esphome:
+  name: test
+
+rtl87xx:
+  board: bw15
@@ -1,4 +1,4 @@
-"""Only BK7238 and BK7231N, the BK chips with an ITCM linker region, get the crash handler."""
+"""The crash handler is built for BK7238, BK7231N (both with an ITCM linker region) and RTL8720C."""

 from __future__ import annotations

@@ -12,12 +12,14 @@ from esphome.core import CORE


@pytest.mark.parametrize(
-    ("config", "enabled"),
+    ("config", "enabled", "bk_hooks"),
    [
-        ("bk7238.yaml", True),
-        ("bk72xx.yaml", True),  # BK7231N
-        ("bk7231t.yaml", False),
-        ("ln882x.yaml", False),
+        ("bk7238.yaml", True, True),
+        ("bk72xx.yaml", True, True),  # BK7231N
+        ("bk7231t.yaml", False, False),
+        ("rtl8720c.yaml", True, False),  # installed at run time
+        ("rtl8710b.yaml", False, False),
+        ("ln882x.yaml", False, False),
    ],
 )
 def test_crash_handler_family_gate(
@@ -25,13 +27,14 @@ def test_crash_handler_family_gate(
    component_config_path: Callable[[str], Path],
    config: str,
    enabled: bool,
+    bk_hooks: bool,
 ) -> None:
    generate_main(component_config_path(config))
    defines = {define.name for define in CORE.defines}
    assert ("USE_LIBRETINY_CRASH_HANDLER" in defines) is enabled
    assert ("USE_CRASH_HANDLER" in defines) is enabled
    for trap in BK_TRAP_HANDLERS:
-        assert (f"-Wl,--wrap={trap}" in CORE.build_flags) is enabled
+        assert (f"-Wl,--wrap={trap}" in CORE.build_flags) is bk_hooks
    assert (
        "pre:patch_noinit.py" in CORE.platformio_options["extra_scripts"]
-    ) is enabled
+    ) is bk_hooks
@@ -0,0 +1,51 @@
+#include <gtest/gtest.h>
+
+#include <string>
+#include <vector>
+
+#include "esphome/core/gpio.h"
+#include "esphome/core/hal.h"
+
+namespace esphome::core::testing {
+
+class RecordingPin : public GPIOPin {
+ public:
+  void setup() override { this->events += 'S'; }
+  void pin_mode(gpio::Flags /*flags*/) override {}
+  gpio::Flags get_flags() const override { return gpio::FLAG_OUTPUT; }
+  bool digital_read() override { return false; }
+  void digital_write(bool value) override {
+    this->events += value ? 'H' : 'L';
+    this->write_ms.push_back(millis());
+  }
+
+  std::string events;
+  std::vector<uint32_t> write_ms;
+};
+
+TEST(PulseReset, NullPinIsIgnored) { pulse_reset(nullptr, {.low_ms = 1}); }
+
+TEST(PulseReset, LowThenHigh) {
+  RecordingPin pin;
+  pulse_reset(&pin, {.low_ms = 1});
+  EXPECT_EQ(pin.events, "SLH");
+}
+
+TEST(PulseReset, PreHighDrivesHighFirst) {
+  RecordingPin pin;
+  pulse_reset(&pin, {.high_before_ms = 1, .low_ms = 1, .high_after_ms = 1});
+  EXPECT_EQ(pin.events, "SHLH");
+}
+
+// Distinct waits, so dropping or swapping one shows up; only lower bounds, a busy host can run long
+TEST(PulseReset, HoldsEachLevelForItsTime) {
+  RecordingPin pin;
+  pulse_reset(&pin, {.high_before_ms = 10, .low_ms = 20, .high_after_ms = 30});
+  const uint32_t end = millis();
+  ASSERT_EQ(pin.events, "SHLH");
+  EXPECT_GE(pin.write_ms[1] - pin.write_ms[0], 10u);
+  EXPECT_GE(pin.write_ms[2] - pin.write_ms[1], 20u);
+  EXPECT_GE(end - pin.write_ms[2], 30u);
+}
+
+}  // namespace esphome::core::testing
@@ -0,0 +1,14 @@
+import esphome.codegen as cg
+from esphome.types import ConfigType
+from tests.testing_helpers import ComponentManifestOverride
+
+
+def override_manifest(manifest: ComponentManifestOverride) -> None:
+    # One empty config entry so to_code runs and sets the define that keeps
+    # core/line_reader.cpp in the test build.
+    manifest.multi_conf = False
+
+    async def to_code(config: ConfigType) -> None:
+        cg.add_define("USE_LINE_READER")
+
+    manifest.to_code = to_code
@@ -0,0 +1,13 @@
+logger:
+  level: VERBOSE
+
+esphome:
+  on_boot:
+    - lambda: |-
+        int x = 100;
+        x = clamp(x, 50, 90);
+        assert(x == 90);
+        x = clamp_at_least(x, 95);
+        assert(x == 95);
+        x = clamp_at_most(x, 40);
+        assert(x == 40);
@@ -0,0 +1,70 @@
+esphome:
+  name: uart-mock-dfrobot-sen0395
+
+host:
+api:
+  batch_delay: 0ms  # Disable batching to receive all state updates
+logger:
+  level: VERBOSE
+
+external_components:
+  - source:
+      type: local
+      path: EXTERNAL_COMPONENT_PATH
+
+# Dummy uart entry to satisfy DEPENDENCIES = ["uart"]
+uart:
+  baud_rate: 115200
+  port: /dev/null
+
+uart_mock:
+  id: mock_uart
+  baud_rate: 115200
+  # A small model of the module's command line: echo, answer and the prompt (no line end)
+  on_tx:
+    - then:
+        - lambda: |-
+            std::string cmd(data.begin(), data.end());
+            std::string reply = cmd + "\r\n";
+            if (cmd == "resetSystem") {
+              // The restart only answers with the prompt
+            } else {
+              reply += "Done\r\n";
+            }
+            reply += "leapMMW:/>";
+            id(mock_uart).inject_to_rx_buffer(std::vector<uint8_t>(reply.begin(), reply.end()));
+
+dfrobot_sen0395:
+  id: sen0395
+  uart_id: mock_uart
+
+binary_sensor:
+  - platform: dfrobot_sen0395
+    name: Detected
+
+switch:
+  - platform: dfrobot_sen0395
+    type: sensor_active
+    name: Sensor Active
+
+button:
+  - platform: template
+    name: Inject Present
+    on_press:
+      - lambda: |-
+          // '$' is kept apart so it is not read as a substitution
+          std::string frame = std::string(1, '$') + "JYBSS,1, , , *\r\n";
+          id(mock_uart).inject_to_rx_buffer(std::vector<uint8_t>(frame.begin(), frame.end()));
+  - platform: template
+    name: Inject Overlong Then Absent
+    on_press:
+      - lambda: |-
+          // An overlong line is dropped up to its line end; the next frame still parses
+          std::string data(300, 'x');
+          data += "\r\n" + std::string(1, '$') + "JYBSS,0, , , *\r\n";
+          id(mock_uart).inject_to_rx_buffer(std::vector<uint8_t>(data.begin(), data.end()));
+  - platform: template
+    name: Reset Sensor
+    on_press:
+      - dfrobot_sen0395.reset:
+          id: sen0395
@@ -0,0 +1,93 @@
+"""Integration test for the DFRobot SEN0395 component against a model of the module's command line.
+
+Checks presence frames, recovery after an overlong line, a command answered by "Done" and the
+reset command, which the module answers only with its "leapMMW:/>" prompt (no line end).
+"""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Callable
+from pathlib import Path
+
+from aioesphomeapi import (
+    BinarySensorInfo,
+    ButtonInfo,
+    EntityInfo,
+    EntityState,
+    SwitchInfo,
+)
+import pytest
+
+from .state_utils import require_entity
+from .types import APIClientConnectedFactory, RunCompiledFunction
+
+
+@pytest.mark.asyncio
+async def test_uart_mock_dfrobot_sen0395(
+    yaml_config: str,
+    run_compiled: RunCompiledFunction,
+    api_client_connected: APIClientConnectedFactory,
+) -> None:
+    """Frames parse, an overlong line is dropped and the prompt completes the reset."""
+    external_components_path = str(
+        Path(__file__).parent / "fixtures" / "external_components"
+    )
+    yaml_config = yaml_config.replace(
+        "EXTERNAL_COMPONENT_PATH", external_components_path
+    )
+
+    loop = asyncio.get_running_loop()
+    dropped = loop.create_future()
+    restarted = loop.create_future()
+
+    def line_callback(line: str) -> None:
+        if "Line over 252 chars dropped" in line and not dropped.done():
+            dropped.set_result(True)
+        if "Restarted sensor." in line and not restarted.done():
+            restarted.set_result(True)
+
+    states: dict[int, EntityState] = {}
+
+    async def wait_for(condition: Callable[[], bool], what: str) -> None:
+        try:
+            async with asyncio.timeout(10.0):
+                while not condition():
+                    await asyncio.sleep(0.05)
+        except TimeoutError:
+            pytest.fail(f"Timeout waiting for {what}; states: {states}")
+
+    async with (
+        run_compiled(yaml_config, line_callback=line_callback),
+        api_client_connected() as client,
+    ):
+        entities, _ = await client.list_entities_services()
+        detected = require_entity(entities, "detected", BinarySensorInfo)
+        active = require_entity(entities, "sensor_active", SwitchInfo)
+        present = require_entity(entities, "inject_present", ButtonInfo)
+        overlong = require_entity(entities, "inject_overlong_then_absent", ButtonInfo)
+        reset = require_entity(entities, "reset_sensor", ButtonInfo)
+
+        client.subscribe_states(lambda state: states.__setitem__(state.key, state))
+
+        def state_of(entity: EntityInfo) -> object | None:
+            state = states.get(entity.key)
+            return None if state is None or state.missing_state else state.state
+
+        client.button_command(present.key)
+        await wait_for(lambda: state_of(detected) is True, "presence")
+
+        client.button_command(overlong.key)
+        await wait_for(
+            lambda: state_of(detected) is False, "absence after overlong line"
+        )
+        await asyncio.wait_for(dropped, timeout=5.0)
+
+        client.switch_command(active.key, False)
+        await wait_for(lambda: state_of(active) is False, "sensor stopped")
+
+        client.button_command(reset.key)
+        try:
+            await asyncio.wait_for(restarted, timeout=10.0)
+        except TimeoutError:
+            pytest.fail("The reset was not completed by the prompt")
@@ -0,0 +1,15 @@
+esphome:
+  name: componenttestesprtl8720c
+  friendly_name: $component_name
+
+rtl87xx:
+  board: bw15
+
+logger:
+  level: VERY_VERBOSE
+
+packages:
+  component_under_test: !include
+    file: $component_test_file
+    vars:
+      component_test_file: $component_test_file
@@ -1,494 +0,0 @@
-"""Tests for ``esphome rotate-key``."""
-
-from __future__ import annotations
-
-from collections.abc import Generator
-from dataclasses import dataclass
-from pathlib import Path
-from typing import Any
-from unittest.mock import ANY, Mock, patch
-
-import pytest
-
-from esphome import yaml_util
-from esphome.cli.rotate_key import PRECHECK_ADVICE, command_rotate_key
-from esphome.const import (
-    CONF_API,
-    CONF_ENCRYPTION,
-    CONF_ESPHOME,
-    CONF_KEY,
-    CONF_OTA,
-    CONF_PLATFORM,
-    CONF_PORT,
-    CONF_WEB_SERVER,
-    KEY_CORE,
-    KEY_TARGET_PLATFORM,
-    PLATFORM_ESP32,
-)
-from esphome.core import CORE, EsphomeError, Toolchain
-from esphome.espota2 import ProbeResult
-from esphome.ota_key_edit import KeyEdit
-from esphome.upload_targets import PortType
-from esphome.util import ESPHOME_COMMAND
-from esphome.yaml_edit import RestoreError
-
-
-@dataclass
-class MockArgs:
-    device: list[str] | None = None
-    prompt_new_key: bool = False
-    yes: bool = False
-    substitution: list[list[str]] | None = None
-    toolchain: Toolchain | None = None
-
-
-def _interrupt_after(writes: int) -> Any:
-    """A write_keeping_mode that lets ``writes`` calls through, then Ctrl-C."""
-    from esphome.yaml_edit import write_keeping_mode
-
-    calls = 0
-
-    def write(*args: Any, **kwargs: Any) -> None:
-        nonlocal calls
-        calls += 1
-        if calls > writes:
-            raise KeyboardInterrupt
-        write_keeping_mode(*args, **kwargs)
-
-    return write
-
-
-def setup_core(tmp_path: Path) -> None:
-    CORE.reset()
-    CORE.config_path = tmp_path / "test.yaml"
-    CORE.name = "test"
-    CORE.build_path = tmp_path / ".esphome" / "build" / "test"
-    CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: PLATFORM_ESP32}
-
-
-OLD_KEY = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
-NEW_KEY = "AQIDBAUGBwgJCgsMDQ4PEBESExQVFhcYGRobHB0eHyA="
-API_YAML = f"""esphome:
-  name: test
-
-api:
-  encryption:
-    key: "{OLD_KEY}"
-
-ota:
-  - platform: esphome
-    encryption:
-"""
-OTA_CONF = {
-    CONF_PLATFORM: CONF_ESPHOME,
-    CONF_PORT: 3232,
-    CONF_ENCRYPTION: {CONF_KEY: OLD_KEY},
-}
-
-
-@pytest.fixture
-def env(tmp_path: Path) -> Generator[dict[str, Mock]]:
-    """A config with an inline api key, loaded the way read_config does so
-    the key node carries its source range, and every network step mocked."""
-    setup_core(tmp_path)
-    CORE.config_path.write_text(API_YAML, encoding="utf-8")
-    CORE.raw_config = yaml_util.load_yaml(CORE.config_path)
-    CORE.config = {
-        CONF_API: {CONF_ENCRYPTION: {CONF_KEY: OLD_KEY}},
-        CONF_OTA: [OTA_CONF],
-    }
-    with (
-        patch("esphome.__main__.choose_upload_log_host", return_value=["dev.local"]),
-        patch("esphome.__main__._resolve_network_devices", return_value=["dev.local"]),
-        patch("esphome.cli.rotate_key.get_port_type", return_value=PortType.NETWORK),
-        patch(
-            "esphome.espota2.probe_ota_key", return_value=ProbeResult.ACCEPTED
-        ) as probe,
-        patch(
-            "esphome.cli.rotate_key.run_external_process", return_value=0
-        ) as compile_,
-        patch("esphome.espota2.run_ota") as upload,
-        patch("esphome.cli.rotate_key.safe_input", return_value="y") as confirm,
-        patch("esphome.cli.rotate_key.sys.stdin") as stdin,
-        patch(
-            "esphome.components.noise.secrets.token_bytes",
-            return_value=bytes(range(1, 33)),
-        ),
-    ):
-        stdin.isatty.return_value = True
-        # The real upload reports the connection before its result
-        upload.return_value = (0, "dev.local")
-
-        def upload_after_connecting(
-            *_args: Any, **kwargs: Any
-        ) -> tuple[int, str | None]:
-            kwargs["on_upload_start"]()
-            return upload.return_value
-
-        upload.side_effect = upload_after_connecting
-        yield {
-            "probe": probe,
-            "compile": compile_,
-            "upload": upload,
-            "confirm": confirm,
-            "stdin": stdin,
-        }
-
-
-def test_success(env: dict[str, Mock], capfd: pytest.CaptureFixture[str]) -> None:
-    """Generate, write, compile in a child, upload with the old key, confirm
-    with the new one, print it."""
-    args = MockArgs()
-    assert command_rotate_key(args, CORE.config) == 0
-
-    assert (
-        CORE.config_path.read_text()
-        == API_YAML.replace(OLD_KEY, NEW_KEY) + f'      old_key: "{OLD_KEY}"\n'
-    )
-    env["confirm"].assert_called_once()
-    precheck, confirm = env["probe"].call_args_list
-    assert precheck.args[2] == OLD_KEY
-    assert precheck.kwargs["retry_rejected"] is False
-    assert confirm.args[2] == NEW_KEY
-    compile_args = env["compile"].call_args.args
-    assert compile_args[-2:] == ("compile", str(CORE.config_path))
-    env["upload"].assert_called_once_with(
-        ["dev.local"],
-        3232,
-        None,
-        CORE.firmware_bin,
-        noise_psk=OLD_KEY,
-        plaintext_fallback=False,
-        on_upload_start=ANY,
-    )
-    out = capfd.readouterr().out
-    assert NEW_KEY in out
-    assert "SUCCESS" in out
-
-
-def test_yes_skips_confirmation(
-    env: dict[str, Mock],
-) -> None:
-    assert command_rotate_key(MockArgs(yes=True), CORE.config) == 0
-    env["confirm"].assert_not_called()
-
-
-def test_declined(env: dict[str, Mock]) -> None:
-    env["confirm"].return_value = "n"
-    assert command_rotate_key(MockArgs(), CORE.config) == 1
-    assert CORE.config_path.read_text() == API_YAML
-    env["probe"].assert_not_called()
-
-
-def test_warns_about_a_shared_secret(
-    env: dict[str, Mock], capfd: pytest.CaptureFixture[str]
-) -> None:
-    """Other configurations using the rewritten secret are named before the
-    confirmation."""
-
-    def scan(edits: list[KeyEdit]) -> list[str]:
-        edits[0].secret = "device_key"
-        edits[0].shared_with = [CORE.config_dir / "b.yaml"]
-        return ["c.yaml is a link"]
-
-    with patch("esphome.cli.rotate_key.with_sharers", side_effect=scan):
-        env["confirm"].return_value = "n"
-        assert command_rotate_key(MockArgs(), CORE.config) == 1
-    out = capfd.readouterr().out
-    assert "b.yaml (secret 'device_key')" in out
-    assert "Could not check every configuration for shared use: c.yaml is a link" in out
-
-
-def test_no_terminal_needs_yes(
-    env: dict[str, Mock],
-) -> None:
-    env["stdin"].isatty.return_value = False
-    assert command_rotate_key(MockArgs(), CORE.config) == 1
-    assert CORE.config_path.read_text() == API_YAML
-
-
-def test_own_ota_key_asks_nothing(
-    env: dict[str, Mock],
-) -> None:
-    """Without an api key nothing changes for Home Assistant."""
-    config = {CONF_OTA: [OTA_CONF]}
-    assert command_rotate_key(MockArgs(), config) == 0
-    env["confirm"].assert_not_called()
-    assert NEW_KEY in CORE.config_path.read_text()
-
-
-def test_prompted_new_key(env: dict[str, Mock]) -> None:
-    key = "AgMEBQYHCAkKCwwNDg8QERITFBUWFxgZGhscHR4fICE="
-    with patch("esphome.cli.rotate_key.read_secret_line", return_value=key):
-        assert command_rotate_key(MockArgs(prompt_new_key=True), CORE.config) == 0
-    assert key in CORE.config_path.read_text()
-    assert env["probe"].call_args_list[1].args[2] == key
-
-
-@pytest.mark.parametrize("key", ["not-base64", OLD_KEY], ids=["invalid", "same"])
-def test_rejects_bad_new_key(env: dict[str, Mock], key: str) -> None:
-    with patch("esphome.cli.rotate_key.read_secret_line", return_value=key):
-        assert command_rotate_key(MockArgs(prompt_new_key=True), CORE.config) == 1
-    assert CORE.config_path.read_text() == API_YAML
-    env["probe"].assert_not_called()
-
-
-def test_precheck_fails_writes_nothing(
-    env: dict[str, Mock], capfd: pytest.CaptureFixture[str]
-) -> None:
-    env["probe"].return_value = ProbeResult.REJECTED
-    assert command_rotate_key(MockArgs(), CORE.config) == 1
-    assert CORE.config_path.read_text() == API_YAML
-    env["compile"].assert_not_called()
-    assert "runs a different key than the yaml" in capfd.readouterr().out
-    for found in (
-        ProbeResult.UNREACHABLE,
-        ProbeResult.NOT_OFFERED,
-        ProbeResult.DEVICE_ERROR,
-    ):
-        env["probe"].return_value = found
-        assert command_rotate_key(MockArgs(), CORE.config) == 1
-        assert PRECHECK_ADVICE[found] in capfd.readouterr().out
-
-
-@pytest.mark.parametrize("step", ["compile", "interrupt"])
-def test_restores_before_the_upload(
-    env: dict[str, Mock], capfd: pytest.CaptureFixture[str], step: str
-) -> None:
-    if step == "compile":
-        env["compile"].return_value = 1
-    else:
-        env["compile"].side_effect = KeyboardInterrupt
-    image = CORE.firmware_bin
-    image.parent.mkdir(parents=True)
-    factory = image.with_name("firmware.factory.bin")
-    for built in (image, factory):
-        built.write_bytes(b"built with the new key")
-    assert command_rotate_key(MockArgs(), CORE.config) == 1
-    assert CORE.config_path.read_text() == API_YAML
-    assert env["probe"].call_count == 1
-    out = capfd.readouterr().out
-    assert ("Interrupted before the upload" in out) is (step == "interrupt")
-    # The images may hold the new key, so no upload can install them
-    assert not image.exists() and not factory.exists()
-    assert "compile again before the next upload" in out
-    assert NEW_KEY not in out
-
-
-def test_a_build_that_cannot_be_removed_shows_the_new_key(
-    env: dict[str, Mock], capfd: pytest.CaptureFixture[str]
-) -> None:
-    """Every image is tried; the ones that stay are all named."""
-    env["compile"].return_value = 1
-    image = CORE.firmware_bin
-    image.mkdir(parents=True)  # unlink fails on a directory
-    stuck = image.with_name("firmware.factory.bin")
-    stuck.mkdir()
-    removable = image.with_name("firmware.ota.bin")
-    removable.write_bytes(b"")
-    assert command_rotate_key(MockArgs(), CORE.config) == 1
-    out = capfd.readouterr().out
-    assert str(image) in out and str(stuck) in out
-    assert not removable.exists()
-    assert OLD_KEY in out and NEW_KEY in out
-
-
-def test_an_interrupt_during_the_image_cleanup_shows_the_keys(
-    env: dict[str, Mock], capfd: pytest.CaptureFixture[str]
-) -> None:
-    env["compile"].return_value = 1
-    CORE.firmware_bin.parent.mkdir(parents=True)
-    CORE.firmware_bin.write_bytes(b"")
-    with patch("pathlib.Path.glob", side_effect=KeyboardInterrupt):
-        assert command_rotate_key(MockArgs(), CORE.config) == 1
-    out = capfd.readouterr().out
-    assert f"{CORE.firmware_bin.parent}: interrupted" in out
-    assert OLD_KEY in out and NEW_KEY in out
-
-
-def test_a_build_from_before_the_rotation_is_kept(
-    env: dict[str, Mock], capfd: pytest.CaptureFixture[str]
-) -> None:
-    env["probe"].return_value = ProbeResult.UNREACHABLE
-    image = CORE.firmware_bin
-    image.parent.mkdir(parents=True)
-    image.write_bytes(b"built with the current key")
-    assert command_rotate_key(MockArgs(), CORE.config) == 1
-    assert image.exists()
-    assert "compile again" not in capfd.readouterr().out
-
-
-@pytest.mark.parametrize("rollback", ["succeeded", "failed"])
-def test_an_interrupt_while_writing_reports_a_failed_rollback(
-    env: dict[str, Mock], capfd: pytest.CaptureFixture[str], rollback: str
-) -> None:
-    """apply_line_edits keeps an interrupt an interrupt and notes a rollback
-    that failed too; then a file still holds the new key, so both are shown."""
-    interrupt = KeyboardInterrupt()
-    if rollback == "failed":
-        interrupt.add_note("Could not restore test.yaml: disk")
-    with patch("esphome.cli.rotate_key.apply_line_edits", side_effect=interrupt):
-        assert command_rotate_key(MockArgs(), CORE.config) == 1
-    out = capfd.readouterr().out
-    assert ("Could not restore test.yaml: disk" in out) is (rollback == "failed")
-    assert (NEW_KEY in out) is (rollback == "failed")
-    assert ("Interrupted before the upload" in out) is (rollback == "succeeded")
-
-
-def test_a_second_interrupt_during_the_restore_shows_the_keys(
-    env: dict[str, Mock], capfd: pytest.CaptureFixture[str]
-) -> None:
-    env["compile"].return_value = 1
-    with patch("esphome.yaml_edit.write_keeping_mode", side_effect=_interrupt_after(1)):
-        assert command_rotate_key(MockArgs(), CORE.config) == 1
-    out = capfd.readouterr().out
-    assert f"Could not restore {CORE.config_path}: interrupted" in out
-    assert OLD_KEY in out and NEW_KEY in out
-
-
-def test_restores_when_the_device_was_never_reached(
-    env: dict[str, Mock], capfd: pytest.CaptureFixture[str]
-) -> None:
-    """An upload that fails before the image is sent cannot have committed."""
-    env["upload"].side_effect = None
-    env["upload"].return_value = (1, None)
-    assert command_rotate_key(MockArgs(), CORE.config) == 1
-    assert CORE.config_path.read_text() == API_YAML
-    out = capfd.readouterr().out
-    assert "failed before the image was sent" in out
-    assert "put back" not in out
-    assert "Restored the previous key" in out
-
-
-def test_keeps_the_edit_after_a_failed_upload(
-    env: dict[str, Mock], capfd: pytest.CaptureFixture[str]
-) -> None:
-    """An attempted upload may have committed, so the config keeps both keys
-    and the new one is printed."""
-    env["upload"].return_value = (1, None)
-    assert command_rotate_key(MockArgs(), CORE.config) == 1
-    text = CORE.config_path.read_text()
-    assert NEW_KEY in text and f'old_key: "{OLD_KEY}"' in text
-    assert NEW_KEY in capfd.readouterr().out
-    assert env["probe"].call_count == 1
-
-
-def test_shows_both_keys_only_when_a_file_stayed_rewritten(
-    env: dict[str, Mock], capfd: pytest.CaptureFixture[str]
-) -> None:
-    with patch(
-        "esphome.cli.rotate_key.apply_line_edits", side_effect=RestoreError("stuck")
-    ):
-        assert command_rotate_key(MockArgs(), CORE.config) == 1
-    out = capfd.readouterr().out
-    assert "stuck" in out and OLD_KEY in out and NEW_KEY in out
-
-
-def test_reports_a_failed_restore(
-    env: dict[str, Mock], capfd: pytest.CaptureFixture[str]
-) -> None:
-    env["compile"].return_value = 1
-    with patch(
-        "esphome.cli.rotate_key.restore_files",
-        side_effect=RestoreError("Could not restore x"),
-    ):
-        assert command_rotate_key(MockArgs(), CORE.config) == 1
-    out = capfd.readouterr().out
-    assert "Could not restore x" in out
-    assert OLD_KEY in out and NEW_KEY in out
-
-
-@pytest.mark.parametrize(
-    "outcome", ["unreachable", "rejected", "not_offered", "device_error", "interrupted"]
-)
-def test_keeps_the_new_key_after_the_upload(
-    env: dict[str, Mock], capfd: pytest.CaptureFixture[str], outcome: str
-) -> None:
-    """Once the upload went through the device most likely runs the new key
-    and old_key covers the other case, so the files stay and the key is
-    printed."""
-    env["probe"].side_effect = [
-        ProbeResult.ACCEPTED,
-        {
-            "interrupted": KeyboardInterrupt,
-            "unreachable": ProbeResult.UNREACHABLE,
-            "rejected": ProbeResult.REJECTED,
-            "not_offered": ProbeResult.NOT_OFFERED,
-            "device_error": ProbeResult.DEVICE_ERROR,
-        }[outcome],
-    ]
-    assert command_rotate_key(MockArgs(), CORE.config) == 1
-    assert NEW_KEY in CORE.config_path.read_text()
-    assert f'old_key: "{OLD_KEY}"' in CORE.config_path.read_text()
-    out = capfd.readouterr().out
-    assert NEW_KEY in out
-    assert ("Interrupted" in out) is (outcome == "interrupted")
-    assert ("still answers with the previous key" in out) is (outcome == "rejected")
-    assert ("did not come back" in out) is (outcome == "unreachable")
-    assert ("came back without encryption" in out) is (outcome == "not_offered")
-    assert ("answered with an error" in out) is (outcome == "device_error")
-
-
-def test_child_compile_gets_global_options_first(
-    env: dict[str, Mock],
-) -> None:
-    """The compile parser is strict, so -s and --toolchain precede it."""
-    args = MockArgs(substitution=[["name", "kitchen"]])
-    CORE.dashboard = True
-    args.toolchain = Toolchain.PLATFORMIO
-    assert command_rotate_key(args, CORE.config) == 0
-    assert env["compile"].call_args.args[len(ESPHOME_COMMAND) :] == (
-        "--dashboard",
-        "--toolchain",
-        "platformio",
-        "-s",
-        "name",
-        "kitchen",
-        "compile",
-        str(CORE.config_path),
-    )
-
-
-@pytest.mark.parametrize(
-    ("config", "port_type", "match"),
-    [
-        (
-            {CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME, CONF_PORT: 3232}]},
-            "NETWORK",
-            "needs 'encryption:'",
-        ),
-        ({CONF_OTA: [{CONF_PLATFORM: CONF_WEB_SERVER}]}, "NETWORK", "esphome OTA"),
-        ({CONF_OTA: [OTA_CONF]}, PortType.SERIAL, "over the air"),
-    ],
-    ids=["no_encryption", "web_server_only", "serial"],
-)
-def test_refuses(
-    env: dict[str, Mock],
-    capfd: pytest.CaptureFixture[str],
-    config: dict[str, Any],
-    port_type: str,
-    match: str,
-) -> None:
-    with patch("esphome.cli.rotate_key.get_port_type", return_value=port_type):
-        assert command_rotate_key(MockArgs(), config) == 1
-    assert match in capfd.readouterr().out
-    env["probe"].assert_not_called()
-
-
-def test_refuses_host(env: dict[str, Mock]) -> None:
-    CORE.data[KEY_CORE][KEY_TARGET_PLATFORM] = "host"
-    assert command_rotate_key(MockArgs(), CORE.config) == 1
-    env["probe"].assert_not_called()
-
-
-@pytest.mark.parametrize("helper", ["locate_key_edits", "apply_line_edits"])
-def test_reports_edit_errors(
-    env: dict[str, Mock], capfd: pytest.CaptureFixture[str], helper: str
-) -> None:
-    with patch(f"esphome.cli.rotate_key.{helper}", side_effect=EsphomeError("nope")):
-        assert command_rotate_key(MockArgs(), CORE.config) == 1
-    out = capfd.readouterr().out
-    assert "nope" in out
-    # A refusal, or an edit rolled back cleanly, shows no key
-    assert NEW_KEY not in out
-    assert CORE.config_path.read_text() == API_YAML
@@ -0,0 +1,125 @@
+"""Tests for trimming a LibreTiny UF2 to the images a device writes over OTA."""
+
+from pathlib import Path
+import struct
+
+import pytest
+
+from esphome.components.libretiny import device_ota_image
+from esphome.components.libretiny.ota_uf2 import UF2_TAG_OTA_PART_INFO, device_ota_uf2
+from esphome.core import CORE
+
+# Written from the UF2 and LibreTiny formats, not from the module under test
+MAGIC_1, MAGIC_2, MAGIC_3 = 0x0A324655, 0x9E5D5157, 0x0AB16F30
+FLAG_FAMILY, FLAG_TAGS = 0x2000, 0x8000
+FAMILY = 0x7B3EF230
+TAG_OTA_FORMAT_2 = 0x6C8492
+TAG_BINPATCH = 0xB948DE
+# OTA_PART_INFO index nibbles: device single, dual 1, dual 2, flasher single, dual 1, dual 2
+DEVICE_SINGLE = b"\x10\x00\x00"
+FLASHER_SINGLE = b"\x00\x01\x00"
+DEVICE_DUAL_1 = b"\x01\x00\x00"
+DEVICE_DUAL_2 = b"\x00\x10\x00"
+
+
+def _tag(tag: int, value: bytes) -> bytes:
+    size = 4 + len(value)
+    return bytes([size]) + tag.to_bytes(3, "little") + value + bytes(-size % 4)
+
+
+def _block(payload: bytes = b"", tags: dict[int, bytes] | None = None) -> bytes:
+    """A block numbered 0 of 0; _package numbers it."""
+    flags = FLAG_FAMILY | (FLAG_TAGS if tags else 0)
+    header = struct.pack("<8I", MAGIC_1, MAGIC_2, flags, 0, len(payload), 0, 0, FAMILY)
+    body = payload + b"".join(_tag(t, v) for t, v in (tags or {}).items())
+    return header + body + bytes(512 - 32 - 4 - len(body)) + struct.pack("<I", MAGIC_3)
+
+
+def _image(part_info: bytes, blocks: int, fill: int) -> list[bytes]:
+    """An image: its first block names the partitions, the rest carry data."""
+    data = bytes([fill]) * 256
+    first = _block(data, {UF2_TAG_OTA_PART_INFO: part_info + b"part\x00"})
+    return [first] + [_block(data) for _ in range(blocks - 1)]
+
+
+def _package(*blocks: bytes) -> bytes:
+    """The blocks numbered in order, as a UF2 file."""
+    return b"".join(
+        block[:20] + struct.pack("<2I", seq, len(blocks)) + block[28:]
+        for seq, block in enumerate(blocks)
+    )
+
+
+HEADER = _block(tags={TAG_OTA_FORMAT_2: b"\x02"})
+FLASHER = _image(FLASHER_SINGLE, 5, 0xAA)
+DEVICE = _image(DEVICE_SINGLE, 3, 0xBB)
+SLOT_1 = _image(DEVICE_DUAL_1, 2, 0x11)
+SLOT_2 = _image(DEVICE_DUAL_2, 2, 0x22)
+PATCHED = _block(bytes([0xAA]) * 256, {TAG_BINPATCH: b"\x01\x02\x03\x04"})
+
+
+@pytest.mark.parametrize(
+    ("blocks", "expected"),
+    [
+        pytest.param([HEADER, *FLASHER, *DEVICE], [HEADER, *DEVICE], id="single_ota"),
+        pytest.param(
+            [HEADER, *SLOT_1, *FLASHER, *SLOT_2],
+            [HEADER, *SLOT_1, *SLOT_2],
+            id="dual_ota_keeps_both_slots",
+        ),
+        pytest.param(
+            [HEADER, *FLASHER, PATCHED, *DEVICE],
+            None,
+            id="dropped_image_carries_a_tag_the_device_acts_on",
+        ),
+        pytest.param([HEADER, *DEVICE], None, id="nothing_to_drop"),
+    ],
+)
+def test_device_ota_uf2(blocks: list[bytes], expected: list[bytes] | None) -> None:
+    trimmed = device_ota_uf2(_package(*blocks))
+    assert trimmed == (_package(*expected) if expected is not None else None)
+
+
+def test_data_that_is_not_a_well_formed_uf2_is_left_alone() -> None:
+    package = _package(HEADER, *FLASHER, *DEVICE)
+    bad_magic = package[:512] + b"\x00" + package[513:]
+    # A tag that runs past the data area of the first flasher block
+    at = 512 + 32 + 256
+    bad_tag = package[:at] + bytes([250]) + package[at + 1 :]
+
+    # A whole block missing, or repeated, keeps the original numbering
+    missing = package[:1024] + package[1536:]
+    repeated = package[:1536] + package[1024:]
+    # OTA_PART_INFO too short to hold the scheme map, which uf2ota rejects
+    short_selectors = [
+        _package(HEADER, _block(bytes(256), {UF2_TAG_OTA_PART_INFO: bytes(n)}), *DEVICE)
+        for n in range(3)
+    ]
+
+    for data in (
+        b"",
+        bytes(range(256)) * 4,
+        package[:-1],
+        bad_magic,
+        bad_tag,
+        missing,
+        repeated,
+        *short_selectors,
+    ):
+        assert device_ota_uf2(data) is None
+
+
+def test_device_ota_image(setup_core: Path) -> None:
+    tmp_path = setup_core
+    CORE.build_path = tmp_path / "build"
+    CORE.name = "test"
+    firmware = tmp_path / "firmware.uf2"
+
+    firmware.write_bytes(_package(HEADER, *FLASHER, *DEVICE))
+    image = device_ota_image(firmware)
+    assert image == CORE.relative_pioenvs_path("test", "firmware.ota.uf2")
+    assert image.read_bytes() == _package(HEADER, *DEVICE)
+
+    # Nothing to drop: the firmware itself is sent
+    firmware.write_bytes(_package(HEADER, *DEVICE))
+    assert device_ota_image(firmware) == firmware
@@ -222,15 +222,9 @@ def test_api_client_does_not_import_heavy_modules() -> None:

 # Packages a domain needs for its own schema; cheap and never an integration
 _ENTITY_COMPONENT_DEPENDENCIES = {
+    "infrared": ("esphome.components.ir_rf_base",),
    "light": ("esphome.components.power_supply", "esphome.components.const.css_colors"),
-}
-
-
-# Entity domains that still import integrations at top level
-_ENTITY_COMPONENT_KNOWN_LEAKS = {
-    "infrared": "imports remote_base and api at top level",
-    "media_player": "audio pulls in the esp32 chain",
-    "radio_frequency": "imports remote_base and api at top level",
+    "radio_frequency": ("esphome.components.ir_rf_base",),
 }


@@ -248,26 +242,11 @@ def _entity_components() -> list[str]:
 def test_entity_component_tables_match_discovery() -> None:
    """Catch stale table entries and a discovery that silently finds nothing."""
    found = set(_entity_components())
-    stale = (
-        _ENTITY_COMPONENT_KNOWN_LEAKS.keys() | _ENTITY_COMPONENT_DEPENDENCIES.keys()
-    ) - found
+    stale = _ENTITY_COMPONENT_DEPENDENCIES.keys() - found
    assert not stale, f"not discovered as entity components: {sorted(stale)}"


-@pytest.mark.parametrize(
-    "component",
-    [
-        pytest.param(
-            c,
-            marks=pytest.mark.xfail(
-                reason=_ENTITY_COMPONENT_KNOWN_LEAKS[c], strict=True
-            ),
-        )
-        if c in _ENTITY_COMPONENT_KNOWN_LEAKS
-        else c
-        for c in _entity_components()
-    ],
-)
+@pytest.mark.parametrize("component", _entity_components())
 def test_entity_component_does_not_import_integrations(component: str) -> None:
    """An entity component must not drag in its optional integrations.

@@ -117,9 +117,9 @@ from esphome.const import (
    PLATFORM_ESP32,
    PLATFORM_ESP8266,
    PLATFORM_HOST,
+    PLATFORM_LN882X,
    PLATFORM_NRF52,
    PLATFORM_RP2,
-    PLATFORM_RTL87XX,
    ErrorFormat,
    Toolchain,
 )
@@ -2135,6 +2135,118 @@ def test_upload_program_ota_success(
    )


+@pytest.mark.parametrize(
+    ("config", "uploader", "image_arg"),
+    [
+        pytest.param(
+            {CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME, CONF_PORT: 8892}]},
+            "mock_run_ota",
+            3,
+            id="native",
+        ),
+        pytest.param(
+            {
+                CONF_OTA: [{CONF_PLATFORM: CONF_WEB_SERVER}],
+                CONF_WEB_SERVER: {CONF_PORT: 80},
+            },
+            "mock_run_web_server_ota",
+            4,
+            id="web_server",
+        ),
+    ],
+)
+def test_upload_program_ota_sends_the_platform_ota_image(
+    config: ConfigType,
+    uploader: str,
+    image_arg: int,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+    request: pytest.FixtureRequest,
+) -> None:
+    """An app update sends the file the platform's ota_image hook picks."""
+    setup_core(platform=PLATFORM_BK72XX, tmp_path=tmp_path)
+    mock_get_port_type.return_value = "NETWORK"
+    mock_upload = request.getfixturevalue(uploader)
+    mock_upload.return_value = (0, "192.168.1.100")
+    image = tmp_path / "firmware.ota.uf2"
+
+    with patch(
+        "esphome.components.bk72xx.ota_image", return_value=image
+    ) as mock_ota_image:
+        exit_code, _ = upload_program(config, MockArgs(), ["192.168.1.100"])
+
+    assert exit_code == 0
+    mock_ota_image.assert_called_once_with(CORE.firmware_bin)
+    assert mock_upload.call_args.args[image_arg] == image
+
+
+def test_upload_program_ota_file_goes_through_the_ota_image_hook(
+    mock_run_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """--file is the firmware the platform's ota_image hook starts from."""
+    setup_core(platform=PLATFORM_BK72XX, tmp_path=tmp_path)
+    mock_get_port_type.return_value = "NETWORK"
+    mock_run_ota.return_value = (0, "192.168.1.100")
+    other = tmp_path / "other.uf2"
+    image = tmp_path / "firmware.ota.uf2"
+
+    config = {CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME, CONF_PORT: 8892}]}
+    with patch(
+        "esphome.components.bk72xx.ota_image", return_value=image
+    ) as mock_ota_image:
+        upload_program(config, MockArgs(file=str(other)), ["192.168.1.100"])
+
+    mock_ota_image.assert_called_once_with(other)
+    assert mock_run_ota.call_args.args[3] == image
+
+
+def test_upload_program_ota_partition_table_skips_the_ota_image_hook(
+    mock_run_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """Only app updates go through the ota_image hook."""
+    setup_core(platform=PLATFORM_BK72XX, tmp_path=tmp_path)
+    mock_get_port_type.return_value = "NETWORK"
+    mock_run_ota.return_value = (0, "192.168.1.100")
+    partition_file = tmp_path / "partitions.bin"
+    partition_file.write_bytes(_make_partition_table_bytes())
+
+    config = {
+        CONF_OTA: [
+            {
+                CONF_PLATFORM: CONF_ESPHOME,
+                CONF_PORT: 8892,
+                "allow_partition_access": True,
+            }
+        ]
+    }
+    args = MockArgs(file=str(partition_file), partition_table=True)
+    with patch("esphome.components.bk72xx.ota_image") as mock_ota_image:
+        upload_program(config, args, ["192.168.1.100"])
+
+    mock_ota_image.assert_not_called()
+    assert mock_run_ota.call_args.args[3] == partition_file
+
+
+def test_upload_program_ota_without_a_hook_sends_the_firmware(
+    mock_run_ota: Mock,
+    mock_get_port_type: Mock,
+    tmp_path: Path,
+) -> None:
+    """A platform without an ota_image hook sends its build output."""
+    setup_core(platform=PLATFORM_ESP32, tmp_path=tmp_path)
+    mock_get_port_type.return_value = "NETWORK"
+    mock_run_ota.return_value = (0, "192.168.1.100")
+
+    config = {CONF_OTA: [{CONF_PLATFORM: CONF_ESPHOME, CONF_PORT: 3232}]}
+    upload_program(config, MockArgs(), ["192.168.1.100"])
+
+    assert mock_run_ota.call_args.args[3].name == "firmware.bin"
+
+
@pytest.mark.parametrize(
    "old_key",
    [None, "AQIDBAUGBwgJCgsMDQ4PEBESExQVFhcYGRobHB0eHyA="],
@@ -6192,7 +6304,7 @@ def test_run_miniterm_no_stacktrace_analyzer(
    """Platforms without a stacktrace analyzer log an info and stream anyway."""
    mock_serial = MockSerial([b"[I][app:100]: Line 1\r\n", MOCK_SERIAL_END])

-    CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: PLATFORM_RTL87XX}
+    CORE.data[KEY_CORE] = {KEY_TARGET_PLATFORM: PLATFORM_LN882X}
    config = {
        CONF_LOGGER: {
            CONF_BAUD_RATE: 115200,
@@ -8136,32 +8248,3 @@ def test_command_rename_is_dispatched_to_the_cli_module() -> None:
    with patch("esphome.cli.rename.command_rename", return_value=7) as run:
        assert main.command_rename(args, {}) == 7
    run.assert_called_once_with(args, {})
-
-
-def test_command_rotate_key_is_dispatched_to_the_cli_module() -> None:
-    args = MockArgs()
-    with patch("esphome.cli.rotate_key.command_rotate_key", return_value=7) as run:
-        assert main.command_rotate_key(args, {}) == 7
-    run.assert_called_once_with(args, {})
-
-
-def test_parse_args_rotate_key() -> None:
-    args = parse_args(
-        [
-            "esphome",
-            "rotate-key",
-            "--device",
-            "dev.local",
-            "--prompt-new-key",
-            "-y",
-            "--username",
-            "mqtt-user",
-            "device.yaml",
-        ]
-    )
-    assert args.command == "rotate-key"
-    assert args.device == ["dev.local"]
-    assert args.prompt_new_key is True
-    assert args.yes is True
-    # MQTT device resolution reads these like upload does
-    assert args.username == "mqtt-user"
@@ -15,7 +15,7 @@ from unittest.mock import Mock
 import pytest

 from esphome import platform_hooks
-from esphome.const import PLATFORM_ESP32, PLATFORM_RTL87XX, Platform
+from esphome.const import PLATFORM_ESP32, PLATFORM_LN882X, Platform


 def test_no_unregistered_platform_exposes_a_hook() -> None:
@@ -165,7 +165,7 @@ def test_get_stacktrace_handler_reports_missing_analyzer(
    caplog: pytest.LogCaptureFixture,
 ) -> None:
    caplog.set_level("INFO", logger="esphome.platform_hooks")
-    assert platform_hooks.get_stacktrace_handler(PLATFORM_RTL87XX) is None
+    assert platform_hooks.get_stacktrace_handler(PLATFORM_LN882X) is None
    assert "no compatible analyzer" in caplog.text
    # A capability gap is ordinary; it must not warn.
    assert not any(r.levelno >= logging.WARNING for r in caplog.records)
@@ -18,6 +18,7 @@ from esphome.const import (
    PLATFORM_BK72XX,
    PLATFORM_ESP32,
    PLATFORM_ESP8266,
+    PLATFORM_LN882X,
    PLATFORM_NRF52,
    PLATFORM_RP2,
    PLATFORM_RTL87XX,
@@ -66,6 +67,10 @@ CRASH_SAMPLES: dict[str, dict[str, list[str]]] = {
        "state_markers": ["CRASH DETECTED ON PREVIOUS BOOT"],
        "addresses": ["PC:  0x0001614A  (fault location)"],
    },
+    PLATFORM_RTL87XX: {
+        "state_markers": ["CRASH DETECTED ON PREVIOUS BOOT"],
+        "addresses": ["PC:  0x9B0073A4  (fault location)"],
+    },
    PLATFORM_NRF52: {
        "state_markers": ["Last crash:"],
        "addresses": [
@@ -231,6 +236,7 @@ DECODER_PATTERNS: dict[str, list[str]] = {
    ],
    PLATFORM_RP2: ["CRASH_HANDLER_RE", "CRASH_HANDLER_ADDR_RE"],
    PLATFORM_BK72XX: ["CRASH_HANDLER_RE", "CRASH_HANDLER_ADDR_RE"],
+    PLATFORM_RTL87XX: ["CRASH_HANDLER_RE", "CRASH_HANDLER_ADDR_RE"],
    PLATFORM_NRF52: ["STACKTRACE_NRF52_PC_LR_RE"],
 }

@@ -548,7 +554,7 @@ def test_processor_registry_miss_disables_at_construction(
    caplog.set_level("INFO", logger="esphome.platform_hooks")

    with patch.object(stacktrace.platform_hooks, "import_module") as mock_import:
-        processor = stacktrace.LogLineProcessor(CONFIG, PLATFORM_RTL87XX)
+        processor = stacktrace.LogLineProcessor(CONFIG, PLATFORM_LN882X)
        processor.process_line("PC: 0x40104960")

    mock_import.assert_not_called()
Author	SHA1	Message	Date
J. Nick Koston	1f01ab32fa	[libretiny] Keep the RTL8720C frame bounds check from wrapping	2026-10-26 06:31:45 -10:00
J. Nick Koston	3ca116aff5	Merge remote-tracking branch 'origin/dev' into rtl87xx-crash-handler	2026-10-26 06:30:01 -10:00
J. Nick Koston	f29745408a	[libretiny] Explain the RTL8720C fault entry stub	2026-10-26 06:30:00 -10:00
J. Nick Koston	9c54f96520	[libretiny] Send only the UF2 images the device writes over OTA (#21642 )	2026-10-26 06:28:54 -10:00
J. Nick Kostonandpre-commit-ci-lite[bot]	0cc5b372fb	[core] Defer integration imports in ir_rf_base and media_player (#21601 ) Co-authored-by: pre-commit-ci-lite[bot] <117423508+pre-commit-ci-lite[bot]@users.noreply.github.com>	2026-10-26 06:26:00 -10:00
J. Nick Koston	ea28b66e8a	Merge branch 'dev' into rtl87xx-crash-handler	2026-10-26 06:22:25 -10:00
J. Nick Koston	f7c8cba8b7	[components] Use pulse_reset for blocking reset pin pulses (#21472 )	2026-10-26 06:22:11 -10:00
J. Nick Koston	79a22e9a9a	[core] Add pulse_reset for active-low reset pins (#21470 )	2026-10-26 06:22:10 -10:00
J. Nick Koston	103023737d	[openthread] Resolve IPv6 literal addresses when the DNS64 client is enabled (#19269 )	2026-10-26 16:18:12 +00:00
J. Nick Koston	6b6b70a8b6	[dfrobot_sen0395] Collect lines with LineReader (#20979 )	2026-10-26 12:16:51 -04:00
J. Nick Koston	cd06a02ff8	[libretiny] Bump the crash record version, gate the RTL capture on RTL8720C The record gained the fault status and address, so its magic is now V2. The AmebaZ2 capture and its install call are built only for the RTL8720C variant, and a family without a capture fails the build.	2026-10-26 06:04:26 -10:00
J. Nick Koston	fcb52d286c	Merge branch 'dev' into rtl87xx-crash-handler	2026-10-26 05:51:43 -10:00
J. Nick Koston	cb943c4492	[libretiny] Report the previous boot's crash on RTL8720C Put a recorder in front of the AmebaZ2 ROM fault handlers (vectors 3 to 7 of the RAM vector table). It saves PC, LR, SP, CFSR, the fault address and a stack scan into .ram.bss.noinit, re-arms the watchdog for a 1 s reset (the ROM handler spins after its dump) and continues into the ROM handler. The next boot logs the record and the CLI decodes it, as on BK72xx.	2026-10-26 05:08:34 -10:00