A comment needs whitespace before its hash in a secret name, old_key takes its secret from the ota block it lands on

Locates the current key on its yaml or secrets.yaml line, writes the new
one with the editor from #20602, keeps the replaced key as old_key, checks
the device with the probe from #20601, builds, uploads with the current key
and confirms with the new one; anything short of a reached device puts the
files back.

esphome/__main__.py

@@ -2232,6 +2232,12 @@ 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
 
@@ -2258,6 +2264,7 @@ 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,
@@ -2655,6 +2662,37 @@ 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.",

esphome/cli/rotate_key.py

@@ -0,0 +1,285 @@
+"""``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."
+    ),
+}
+
+
+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 did not reach 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(
+                (
+                    "The device still answers with the previous key."
+                    if found is ProbeResult.REJECTED
+                    else "The device did not come back to confirm the new key."
+                )
+                + " 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.",
+            )
+        )

esphome/compiled_config.py

@@ -33,9 +33,35 @@ _CACHE_VERSION = 1
 _LAMBDA_KEY = "__esphome_lambda__"
 
 
+_CACHE_SUFFIX = ".validated.json"
+
+
 def compiled_config_path(config_filename: str) -> Path:
     """Path to the cached validated config alongside the storage sidecar."""
-    return CORE.data_dir / "storage" / f"{config_filename}.validated.json"
+    return CORE.data_dir / "storage" / f"{config_filename}{_CACHE_SUFFIX}"
+
+
+def invalidate_compiled_config() -> None:
+    """Drop every cache in the storage dir after a change to a file the
+    mtime check cannot see, such as secrets.yaml or an include; other
+    configurations may share that file."""
+    storage = compiled_config_path(CORE.config_filename).parent
+    try:
+        caches = [p for p in storage.iterdir() if p.name.endswith(_CACHE_SUFFIX)]
+    except FileNotFoundError:
+        return
+    except OSError as err:
+        raise EsphomeError(
+            f"Could not list the validated config caches in {storage}: {err}"
+        ) from err
+    for path in caches:
+        try:
+            path.unlink(missing_ok=True)
+        except OSError as err:
+            # A stale cache would present the old key on the next upload
+            raise EsphomeError(
+                f"Could not remove the validated config cache {path}: {err}"
+            ) from err
 
 
 def save_compiled_config(config: ConfigType) -> None:

esphome/ota_key_edit.py

@@ -0,0 +1,376 @@
+"""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")
+_COMMENT_RE = re.compile(r"(?:^|\s)#.*$")
+_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, which starts at a `#` after whitespace."""
+    return _COMMENT_RE.sub("", 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]

esphome/yaml_edit.py

@@ -9,7 +9,8 @@ from pathlib import Path
 import re
 import stat
 
-from esphome.core import EsphomeError
+from esphome import compiled_config, yaml_util
+from esphome.core import CORE, EsphomeError
 from esphome.helpers import write_file
 from esphome.types import ConfigType
 
@@ -20,6 +21,10 @@ PLAIN_SCALAR = r"[^\s#\"'>|&*!%@`\[\]{},]+"
 TRAILER = r"(?P<quote>[\"']?){value}(?P=quote)(?P<trail>(?:\s+#.*)?\s*)$"
 
 
+class RestoreError(EsphomeError):
+    """A rollback left at least one file as it was written."""
+
+
 def read_text(path: Path) -> str:
     """The file as written, line endings included; read_file would fold them."""
     try:
@@ -30,12 +35,25 @@ def read_text(path: Path) -> str:
 
 @dataclass
 class LineEdit:
-    """One line to rewrite; ``old_line`` is what it held when located."""
+    """One line to rewrite, or with ``insert_after`` a line to add right
+    after it; ``old_line`` is what the line held when it was located."""
 
     path: Path
     line: int
     old_line: str
     new_line: str
+    insert_after: bool = False
+
+
+@dataclass
+class Snapshot:
+    """A rewritten file's text before and after, and its mode before; the
+    restore only puts ``original`` back over ``written``, never over the
+    user's own edit."""
+
+    original: str
+    written: str
+    mode: int | None = None
 
 
 def field_line_re(
@@ -63,17 +81,66 @@ def line_at(doc: Path, line_no: int) -> str:
     return lines[line_no]
 
 
+_DOMAIN = "yaml_edit"
+
+
+def own_documents() -> set[Path]:
+    """Every file the loader read for this configuration, found once per run.
+    A file behind an include path with a substitution that gave nothing but
+    substitutions is the one kind this misses."""
+    if (documents := CORE.data.get(_DOMAIN)) is None:
+        documents = CORE.data[_DOMAIN] = _own_documents()
+    return documents
+
+
+def _own_documents() -> set[Path]:
+    documents = {str(CORE.config_path)}
+
+    def walk(node: object) -> None:
+        if (rng := getattr(node, "esp_range", None)) is not None:
+            documents.add(rng.start_mark.document)
+        if isinstance(node, dict):
+            for key, value in node.items():
+                walk(key)
+                walk(value)
+        elif isinstance(node, list):
+            for value in node:
+                walk(value)
+
+    walk(CORE.raw_config or {})
+    # A file that only gave substitutions leaves no node behind; the tracker
+    # also lists a secrets file the loader only tried, hence the is_file check
+    found = yaml_util.discover_user_yaml_files(CORE.config_path, clear_secrets=False)
+    if found.load_errors:  # an incomplete set would misplace a key
+        raise EsphomeError(
+            "Could not read every file of the configuration: "
+            + "; ".join(found.load_errors)
+        )
+    documents.update(map(str, found.files))
+    return {p.resolve() for d in documents if (p := Path(d)).is_file()}
+
+
 def source_of(mapping: ConfigType, name: str) -> tuple[Path, int] | None:
-    """The file and line ``name:`` was read from, None when validation added
-    it or a merge key brought it in from an anchor elsewhere. Mapping keys
-    keep their range through validation, values may not."""
+    """The file and line ``name:`` was read from; None when validation added
+    it or a merge key brought it in from an anchor elsewhere."""
     rng = getattr(next((k for k in mapping if k == name), None), "esp_range", None)
     if rng is None:
         return None
-    # An included mapping carries the `!include` line of its parent, so only
-    # a key in the same document can be placed against the mapping
     if (own := getattr(mapping, "esp_range", None)) is None:
-        return None  # a mapping built in code, its keys are not on its lines
+        # merge_config keeps the first mapping's keys with the last one's
+        # values, so only the value's range names the line that won, and
+        # only when the key is on it (a block value starts on a line of its own)
+        rng = getattr(mapping.get(name), "esp_range", None)
+        if rng is None:
+            return None
+        doc, line = Path(rng.start_mark.document), rng.start_mark.line
+        if doc.resolve() not in own_documents() or not field_line_re(name).match(
+            line_at(doc, line)
+        ):
+            return None
+        return doc, line
+    # An included mapping carries its parent's `!include` line, so only a
+    # key in the same document can be checked against the mapping's lines
     if (
         rng.start_mark.document == own.start_mark.document
         and not own.start_mark.line <= rng.start_mark.line <= own.end_mark.line
@@ -82,13 +149,46 @@ def source_of(mapping: ConfigType, name: str) -> tuple[Path, int] | None:
     return Path(rng.start_mark.document), rng.start_mark.line
 
 
-def write_keeping_mode(path: Path, text: str, like: Path | None = None) -> None:
-    """Write with the mode of ``like`` (default: the file itself) rather
-    than write_file's 0644; a 0600 secrets file stays 0600."""
+def editable_file(path: Path) -> Path:
+    """``path`` resolved, so a symlink's target is edited; refuses a file
+    outside the config dir or under the build data."""
+    resolved = path.resolve()
+    if (
+        not path.is_file()
+        or not resolved.is_relative_to(CORE.config_dir.resolve())
+        or resolved.is_relative_to(CORE.data_dir.resolve())
+    ):
+        raise EsphomeError(
+            f"{path} is not an editable file in the configuration directory"
+        )
+    return resolved
+
+
+def source_line(mapping: ConfigType, name: str) -> tuple[Path, int, str]:
+    """The file, line number and text ``name:`` was read from."""
+    if (source := source_of(mapping, name)) is None:
+        raise EsphomeError(f"'{name}' was not read from a file")
+    doc, line_no = source
+    # Returned as the loader saw it: a `!secret` in a symlinked include
+    # resolves beside the link
+    editable_file(doc)
+    return doc, line_no, line_at(doc, line_no)
+
+
+def mode_of(path: Path) -> int:
     try:
-        mode = stat.S_IMODE((like or path).stat().st_mode)
+        return stat.S_IMODE(path.stat().st_mode)
     except OSError as err:
-        raise EsphomeError(f"Could not read the mode of {like or path}: {err}") from err
+        raise EsphomeError(f"Could not read the mode of {path}: {err}") from err
+
+
+def write_keeping_mode(
+    path: Path, text: str, like: Path | None = None, mode: int | None = None
+) -> None:
+    """Write with ``mode``, else the mode of ``like`` (default: the file
+    itself), rather than write_file's 0644; a 0600 secrets file stays 0600."""
+    if mode is None:
+        mode = mode_of(like or path)
     try:
         write_file(path, text, private=True)
     except EsphomeError as err:
@@ -103,12 +203,113 @@ def write_keeping_mode(path: Path, text: str, like: Path | None = None) -> None:
 
 
 def rewritten_text(original: str, edits: list[LineEdit]) -> str:
-    """``original`` with the edits applied; every edit must still find the
-    line it was located on."""
+    """``original`` with the edits applied; a line that changed is refused."""
     lines = original.splitlines(keepends=True)
-    for edit in edits:
+    file_newline = "\r\n" if "\r\n" in original else "\n"
+    # Bottom up, so an insertion never shifts a later edit; insertions under
+    # one line keep their order
+    order = sorted(
+        enumerate(edits), key=lambda ie: (ie[1].line, ie[1].insert_after, ie[0])
+    )
+    for _, edit in reversed(order):
         text = lines[edit.line].rstrip("\r\n") if edit.line < len(lines) else None
         if text != edit.old_line:
             raise EsphomeError(f"{edit.path}:{edit.line + 1} changed since it was read")
-        lines[edit.line] = edit.new_line + lines[edit.line][len(text) :]
+        ending = lines[edit.line][len(text) :]
+        if edit.insert_after:
+            # A last line without a newline gets the file's own kind
+            lines[edit.line] = text + (ending or file_newline)
+            lines.insert(edit.line + 1, edit.new_line + ending)
+        else:
+            lines[edit.line] = edit.new_line + ending
     return "".join(lines)
+
+
+def apply_line_edits(edits: list[LineEdit]) -> dict[Path, Snapshot]:
+    """Rewrite the located lines in place; returns each file's text before
+    and after for restore_files. Nothing is written while any line is stale,
+    and a file that no longer loads is put back here."""
+    by_path: dict[Path, list[LineEdit]] = {}
+    seen_as: dict[Path, Path] = {}  # the path the loader read a file by
+    for edit in edits:
+        resolved = editable_file(edit.path)
+        by_path.setdefault(resolved, []).append(edit)
+        seen_as.setdefault(resolved, edit.path)
+    originals = {}
+    for path, own in by_path.items():
+        original = read_text(path)
+        originals[path] = Snapshot(
+            original, rewritten_text(original, own), mode_of(path)
+        )
+    try:
+        for path, snapshot in originals.items():
+            write_keeping_mode(path, snapshot.written)
+        # Reloaded as the loader saw them: a relative !include inside a
+        # symlinked file resolves beside the link
+        reloads = [seen_as[path] for path in originals]
+        if CORE.config_path.resolve() not in originals:
+            # Its !secret tags register a value written elsewhere
+            reloads.append(CORE.config_path)
+        for path in reloads:
+            try:
+                # The secret registry stays: later output still redacts
+                yaml_util.load_yaml(
+                    path, clear_secrets=False, track_document_range=False
+                )
+            except EsphomeError as err:
+                raise EsphomeError(f"{path} no longer loads: {err}") from err
+        compiled_config.invalidate_compiled_config()
+    except BaseException as err:
+        try:
+            restore_files(originals)
+        except RestoreError as restore_err:
+            if not isinstance(err, Exception):
+                err.add_note(str(restore_err))  # an interrupt stays one
+                raise err from None
+            raise RestoreError(f"{err}; {restore_err}") from err
+        raise
+    return originals
+
+
+def restore_files(originals: dict[Path, Snapshot]) -> None:
+    """Put back what apply_line_edits wrote, mode included; a file that
+    failed or changed meanwhile is left alone and reported. Every file is
+    tried whatever went wrong before; an interrupt or exit is raised again
+    at the end with the report as a note."""
+    failed: list[str] = []
+    stop: BaseException | None = None
+
+    def report(what: str, err: BaseException) -> None:
+        nonlocal stop
+        if isinstance(err, Exception):
+            why = str(err) or type(err).__name__
+        else:  # an interrupt or exit says what it is, not its code
+            why = (
+                "interrupted"
+                if isinstance(err, KeyboardInterrupt)
+                else type(err).__name__
+            )
+            stop = err
+        failed.append(f"{what}: {why}")
+
+    for path, snapshot in originals.items():
+        try:
+            current = read_text(path)
+            if current == snapshot.original:
+                continue  # never written, or already put back
+            if current != snapshot.written:
+                raise EsphomeError("changed since it was written, left as is")
+            write_keeping_mode(path, snapshot.original, mode=snapshot.mode)
+        except BaseException as err:  # noqa: BLE001  # pylint: disable=broad-except
+            report(str(path), err)
+    try:
+        compiled_config.invalidate_compiled_config()
+    except BaseException as err:  # noqa: BLE001  # pylint: disable=broad-except
+        report("the validated caches", err)
+    if not failed:
+        return
+    message = "Could not restore " + "; ".join(failed)
+    if stop is not None:
+        stop.add_note(message)
+        raise stop
+    raise RestoreError(message)

esphome/yaml_util.py

@@ -637,9 +637,13 @@ class DiscoveredYamlFiles:
     load_errors: list[str] = field(default_factory=list)
 
 
-def discover_user_yaml_files(config_path: Path) -> DiscoveredYamlFiles:
+def discover_user_yaml_files(
+    config_path: Path, clear_secrets: bool = True
+) -> DiscoveredYamlFiles:
     """Fresh-re-parse ``config_path`` and report every file the YAML loader
     pulled in, plus which of them came in under a secrets filename.
+    ``clear_secrets`` is passed to the load; False keeps the redaction
+    registry of a configuration loaded earlier in the run.
 
     Does NOT run schema validation, substitutions, or package resolution — so
     component-internal YAML loaded by validators (LVGL helpers, dashboard
@@ -663,7 +667,7 @@ def discover_user_yaml_files(config_path: Path) -> DiscoveredYamlFiles:
         _load_listeners.append(_capture_secret)
         try:
             try:
-                data = load_yaml(config_path)
+                data = load_yaml(config_path, clear_secrets=clear_secrets)
             except EsphomeError as err:
                 return DiscoveredYamlFiles(
                     list(loaded), secrets, load_errors=[f"{config_path}: {err}"]
@@ -922,9 +926,7 @@ class ESPHomeLoaderMixin:
             if self.name == CORE.config_path:
                 raise e
             try:
-                main_config_dir = CORE.config_path.parent
-                main_secret_yml = main_config_dir / SECRET_YAML
-                secrets = self.yaml_loader(main_secret_yml)
+                secrets = self.yaml_loader(CORE.config_path.parent / SECRET_YAML)
             except EsphomeError as er:
                 raise EsphomeError(f"{e}\n{er}") from er
 
@@ -1297,6 +1299,20 @@ 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)]

tests/unit_tests/cli/test_rotate_key.py

@@ -0,0 +1,488 @@
+"""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 a connection opens 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 "did not reach the device" 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", "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,
+        }[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")
+
+
+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

tests/unit_tests/test_compiled_config.py

@@ -18,6 +18,7 @@ from esphome.__main__ import run_esphome
 from esphome.compiled_config import (
     _LAMBDA_KEY,
     compiled_config_path,
+    invalidate_compiled_config,
     load_compiled_config,
     save_compiled_config,
     save_compiled_config_and_sidecar,
@@ -937,3 +938,46 @@ def test_load_compiled_config_rejects_wizard_only_sidecar(
     _set_cache_mtime(cache_path, yaml_path, offset=5)
 
     assert load_compiled_config(yaml_path) is None
+
+
+def test_invalidate_compiled_config_reports_an_unlink_failure(
+    tmp_path: Path,
+) -> None:
+    """A cache that cannot be removed would serve the old key later."""
+    CORE.config_path = tmp_path / "test.yaml"
+    cache = compiled_config_path("test.yaml")
+    cache.parent.mkdir(parents=True)
+    cache.write_text("{}")
+    with (
+        patch("pathlib.Path.unlink", side_effect=OSError("busy")),
+        pytest.raises(EsphomeError, match="validated config cache"),
+    ):
+        invalidate_compiled_config()
+
+
+def test_invalidate_compiled_config_drops_every_cache(tmp_path: Path) -> None:
+    """Another configuration sharing the edited secrets.yaml or include has
+    a cache the mtime check would keep serving."""
+    CORE.config_path = tmp_path / "test.yaml"
+    storage = compiled_config_path("test.yaml").parent
+    storage.mkdir(parents=True)
+    for name in ("test.yaml", "other.yaml"):
+        (storage / f"{name}.validated.json").write_text("{}")
+    (storage / "other.json").write_text("{}")
+    invalidate_compiled_config()
+    assert sorted(p.name for p in storage.iterdir()) == ["other.json"]
+
+
+def test_invalidate_compiled_config_reports_a_storage_dir_it_cannot_list(
+    tmp_path: Path,
+) -> None:
+    CORE.config_path = tmp_path / "test.yaml"
+    storage = compiled_config_path("test.yaml").parent
+    storage.mkdir(parents=True)
+    with (
+        patch("pathlib.Path.iterdir", side_effect=OSError("denied")),
+        pytest.raises(EsphomeError, match="Could not list"),
+    ):
+        invalidate_compiled_config()
+    storage.rmdir()
+    invalidate_compiled_config()  # no storage dir yet is nothing to drop

tests/unit_tests/test_main.py

@@ -8087,3 +8087,32 @@ 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"

tests/unit_tests/test_yaml_edit.py

@@ -3,21 +3,32 @@
 from __future__ import annotations
 
 from pathlib import Path
+import stat
 import sys
+from typing import Any
+from unittest.mock import patch
 
 import pytest
 
 from esphome import yaml_util
+from esphome.compiled_config import compiled_config_path
 from esphome.config import do_substitution_pass
 from esphome.const import CONF_ESPHOME, CONF_NAME
 from esphome.core import CORE, EsphomeError
 from esphome.yaml_edit import (
     LineEdit,
+    RestoreError,
+    Snapshot,
+    apply_line_edits,
+    editable_file,
     field_line_re,
     line_at,
+    own_documents,
     read_text,
+    restore_files,
     rewrite,
     rewritten_text,
+    source_line,
     source_of,
     write_keeping_mode,
 )
@@ -98,13 +109,46 @@ def test_a_comment_needs_whitespace_and_a_scalar_is_not_empty() -> None:
     assert field_line_re("key", "abc#def").match("key: abc#def") is not None
 
 
-def test_source_of_is_none_for_a_value_validation_added(tmp_path: Path) -> None:
-    """Only a key read from a file carries a range, and only a mapping read
-    from a file can place its keys; a mapping built in code cannot."""
+def test_source_of_is_none_for_a_value_validation_added() -> None:
+    """Only a key read from a file carries a range."""
     assert source_of({"name": "kitchen"}, "name") is None
+
+
+def test_a_mapping_built_in_code_places_only_this_configurations_keys(
+    tmp_path: Path,
+) -> None:
+    """merge_config rebuilds a mapping without a range; a value read for this
+    configuration is still placed, one read from elsewhere is not."""
     _setup(tmp_path, YAML)
-    loaded_key = next(iter(CORE.raw_config[CONF_ESPHOME]))
-    assert source_of({loaded_key: "kitchen"}, CONF_NAME) is None
+    own_key, own_value = next(iter(CORE.raw_config[CONF_ESPHOME].items()))
+    assert source_of({own_key: own_value}, CONF_NAME) == (CORE.config_path, 1)
+    # Only the value names the line that won; without a range it is refused
+    assert source_of({own_key: "x"}, CONF_NAME) is None
+    (tmp_path / "elsewhere.yaml").write_bytes(YAML.encode())
+    other = yaml_util.load_yaml(tmp_path / "elsewhere.yaml")
+    other_key, other_value = next(iter(other[CONF_ESPHOME].items()))
+    assert source_of({other_key: other_value}, CONF_NAME) is None
+    # A value on a line of its own does not name the key's line
+    _setup(tmp_path, "esphome:\n  name:\n    kitchen\n")
+    own_key, own_value = next(iter(CORE.raw_config[CONF_ESPHOME].items()))
+    assert source_of({own_key: own_value}, CONF_NAME) is None
+
+
+def test_own_documents_include_a_file_that_only_gives_substitutions(
+    tmp_path: Path,
+) -> None:
+    (tmp_path / "subs.yaml").write_bytes(b"substitutions:\n  x: y\n")
+    (tmp_path / "sensor.yaml").write_bytes(b"platform: template\nname: x\n")
+    _setup(
+        tmp_path,
+        YAML + "packages:\n  subs: !include subs.yaml\nsensor:\n"
+        "  - !include sensor.yaml\n",
+    )
+    assert (tmp_path / "sensor.yaml").resolve() in own_documents()  # via a list
+    CORE.reset()
+    CORE.config_path = tmp_path / "test.yaml"
+    CORE.raw_config = {}  # what is left after substitutions are taken out
+    assert (tmp_path / "subs.yaml").resolve() in own_documents()
 
 
 def test_source_of_is_none_for_a_merged_key(tmp_path: Path) -> None:
@@ -152,8 +196,6 @@ def test_write_keeps_the_mode_of_the_file_or_another(tmp_path: Path) -> None:
 def test_write_failures_say_which_step_and_why(tmp_path: Path) -> None:
     """A missing mode source, a write that fails, and a mode that cannot be
     put back after the write are three different situations."""
-    from unittest.mock import patch
-
     path = _setup(tmp_path, YAML)
     with pytest.raises(EsphomeError, match="Could not read the mode of .*gone.yaml"):
         write_keeping_mode(path, YAML, like=tmp_path / "gone.yaml")
@@ -180,3 +222,349 @@ def test_read_text_reports_a_file_it_cannot_decode(tmp_path: Path) -> None:
     path.write_bytes(b"caf\xe9: 1\n")
     with pytest.raises(EsphomeError, match="Error reading file"):
         read_text(path)
+
+
+def test_source_line_refuses_what_is_not_an_editable_file(tmp_path: Path) -> None:
+    """A remote package is checked out under the build data and is not the
+    user's; a file outside the configuration directory is not either."""
+    (tmp_path / ".esphome").mkdir()
+    (tmp_path / ".esphome" / "base.yaml").write_bytes(b"name: kitchen\n")
+    _setup(tmp_path, "esphome: !include .esphome/base.yaml\n")
+    with pytest.raises(EsphomeError, match="not an editable file"):
+        source_line(CORE.raw_config[CONF_ESPHOME], CONF_NAME)
+    with pytest.raises(EsphomeError, match="was not read from a file"):
+        source_line({"name": "kitchen"}, CONF_NAME)
+    _setup(tmp_path, YAML)
+    assert source_line(CORE.raw_config[CONF_ESPHOME], CONF_NAME) == (
+        CORE.config_path,
+        1,
+        "  name: kitchen  # the device",
+    )
+    outside = tmp_path.parent / "outside.yaml"
+    outside.write_bytes(b"")
+    with pytest.raises(EsphomeError, match="not an editable file"):
+        editable_file(outside)
+    (tmp_path / "link.yaml").symlink_to(CORE.config_path)
+    assert editable_file(tmp_path / "link.yaml") == CORE.config_path.resolve()
+
+
+def test_insert_after_goes_before_the_rewrite_of_the_same_line(
+    tmp_path: Path,
+) -> None:
+    """Edits apply from the bottom up so an insertion never moves a later
+    line; the file's own line ending is used, also on a bare last line."""
+    text = YAML.replace("\n", "\r\n").rstrip("\r\n")
+    path = _setup(tmp_path, text)
+    edit = _name_edit("garage")
+    added = LineEdit(path, 4, "  ssid: kitchen", "  password: x", insert_after=True)
+    more = LineEdit(
+        path, 4, "  ssid: kitchen", "  fast_connect: true", insert_after=True
+    )
+    assert rewritten_text(read_text(path), [added, more, edit]) == (
+        text.replace("name: kitchen", "name: garage")
+        + "\r\n  password: x\r\n  fast_connect: true"
+    )
+    edit.old_line = "changed"
+    with pytest.raises(EsphomeError, match="changed since it was read"):
+        rewritten_text(read_text(path), [added, edit])
+
+
+def test_own_documents_refuses_an_incomplete_set(tmp_path: Path) -> None:
+    _setup(tmp_path, YAML)
+    found = yaml_util.DiscoveredYamlFiles(load_errors=["sub.yaml: broken"])
+    with (
+        patch("esphome.yaml_util.discover_user_yaml_files", return_value=found),
+        pytest.raises(EsphomeError, match="Could not read every file .*sub.yaml"),
+    ):
+        own_documents()
+
+
+def test_apply_keeps_the_secret_registry(tmp_path: Path) -> None:
+    """The reload must not clear what later output redacts as !secret."""
+    (tmp_path / "secrets.yaml").write_bytes(b"ssid: hidden\n")
+    path = _setup(tmp_path, YAML.replace("ssid: kitchen", "ssid: !secret ssid"))
+    assert yaml_util.is_secret("hidden") == "ssid"
+    apply_line_edits([_name_edit("garage")])
+    assert yaml_util.is_secret("hidden") == "ssid"
+    assert "!secret ssid" in path.read_text()
+
+
+def test_apply_writes_reloads_drops_the_cache_and_restores(tmp_path: Path) -> None:
+    path = _setup(tmp_path, YAML)
+    cache = compiled_config_path(CORE.config_filename)
+    cache.parent.mkdir(parents=True, exist_ok=True)
+    cache.write_text("{}")
+    mode = stat.S_IMODE(path.stat().st_mode)
+    originals = apply_line_edits([_name_edit("garage")])
+    assert originals == {
+        path: Snapshot(YAML, YAML.replace("kitchen  #", "garage  #"), mode)
+    }
+    assert path.read_text() == YAML.replace("kitchen  #", "garage  #")
+    assert not cache.exists()
+    cache.write_text("{}")
+    restore_files(originals)
+    assert path.read_text() == YAML
+    assert not cache.exists()
+
+
+def test_apply_reloads_a_symlinked_file_by_the_link(tmp_path: Path) -> None:
+    """A relative !include inside the file resolves beside the link, as the
+    loader read it, not beside the target."""
+    shared = tmp_path / "shared"
+    shared.mkdir()
+    (shared / "device.yaml").write_bytes(
+        b"name: kitchen\nfriendly_name: !include fn.yaml\n"
+    )
+    (tmp_path / "fn.yaml").write_bytes(b"Kitchen\n")  # beside the link only
+    (tmp_path / "device.yaml").symlink_to(shared / "device.yaml")
+    _setup(tmp_path, "esphome: !include device.yaml\n")
+    doc, line_no = source_of(CORE.raw_config[CONF_ESPHOME], CONF_NAME)
+    assert doc == tmp_path / "device.yaml"
+    match = field_line_re(CONF_NAME, "kitchen").match(line_at(doc, line_no))
+    edit = LineEdit(doc, line_no, match.string, rewrite(match, "garage"))
+    originals = apply_line_edits([edit])
+    assert list(originals) == [(shared / "device.yaml").resolve()]
+    assert (shared / "device.yaml").read_text() == (
+        "name: garage\nfriendly_name: !include fn.yaml\n"
+    )
+
+
+def test_apply_rolls_back_a_rewrite_the_loader_refuses(tmp_path: Path) -> None:
+    path = _setup(tmp_path, YAML)
+    edit = _name_edit("garage")
+    edit.new_line = "  name: [unterminated"
+    with pytest.raises(EsphomeError, match="no longer loads"):
+        apply_line_edits([edit])
+    assert path.read_text() == YAML
+    with (
+        patch("esphome.yaml_util.load_yaml", side_effect=KeyboardInterrupt),
+        pytest.raises(KeyboardInterrupt),
+    ):
+        apply_line_edits([_name_edit("garage")])
+    assert path.read_text() == YAML
+
+
+def test_restore_reports_every_file_it_could_not_write(tmp_path: Path) -> None:
+    """A missing file, a failed write and a cache that cannot be dropped are
+    all reported; a file the user changed meanwhile is left alone."""
+    path = _setup(tmp_path, YAML)
+    good = tmp_path / "a.yaml"
+    (tmp_path / "b.yaml").write_bytes(b"")
+    good.write_bytes(b"")
+    with (
+        patch("esphome.yaml_edit.write_file", side_effect=[EsphomeError("disk"), None]),
+        pytest.raises(EsphomeError, match="Could not restore .*b.yaml: disk"),
+    ):
+        restore_files({tmp_path / "b.yaml": Snapshot("x", ""), good: Snapshot("y", "")})
+    with pytest.raises(
+        EsphomeError, match="Could not restore .*gone.yaml: Error reading file"
+    ):
+        restore_files({tmp_path / "gone.yaml": Snapshot("x", "x")})
+    with (
+        patch(
+            "esphome.compiled_config.invalidate_compiled_config",
+            side_effect=EsphomeError("busy"),
+        ),
+        pytest.raises(
+            EsphomeError, match="Could not restore the validated caches: busy"
+        ),
+    ):
+        restore_files({path: Snapshot(YAML, YAML)})
+    originals = apply_line_edits([_name_edit("garage")])
+    edited = path.read_text() + "logger:\n"
+    path.write_bytes(edited.encode())
+    with pytest.raises(EsphomeError, match="changed since it was written, left as is"):
+        restore_files(originals)
+    assert path.read_text() == edited
+
+
+def test_apply_rolls_back_when_the_cache_cannot_be_dropped(tmp_path: Path) -> None:
+    path = _setup(tmp_path, YAML)
+    with (
+        patch(
+            "esphome.compiled_config.invalidate_compiled_config",
+            side_effect=[EsphomeError("busy"), None],
+        ),
+        pytest.raises(EsphomeError, match="busy"),
+    ):
+        apply_line_edits([_name_edit("garage")])
+    assert path.read_text() == YAML
+
+
+def test_apply_reports_a_rollback_that_also_failed(tmp_path: Path) -> None:
+    """The rewrite lands, the reload fails, and the rollback write fails too."""
+    from esphome.yaml_edit import write_file
+
+    _setup(tmp_path, YAML)
+    writes: list[int] = []
+
+    def write_then_fail(*args: Any, **kwargs: Any) -> None:
+        writes.append(1)
+        if len(writes) > 1:
+            raise EsphomeError("disk")
+        write_file(*args, **kwargs)
+
+    with (
+        patch("esphome.yaml_util.load_yaml", side_effect=EsphomeError("broken")),
+        patch("esphome.yaml_edit.write_file", side_effect=write_then_fail),
+        pytest.raises(EsphomeError, match="broken; Could not restore .*disk"),
+    ):
+        apply_line_edits([_name_edit("garage")])
+
+
+def test_a_failed_rollback_keeps_an_interrupt_an_interrupt(tmp_path: Path) -> None:
+    _setup(tmp_path, YAML)
+    with (
+        patch("esphome.yaml_util.load_yaml", side_effect=KeyboardInterrupt),
+        patch("pathlib.Path.chmod", side_effect=[None, OSError("disk")]),
+        pytest.raises(KeyboardInterrupt) as info,
+    ):
+        apply_line_edits([_name_edit("garage")])
+    assert "disk" in "".join(info.value.__notes__)
+
+
+def test_rollback_after_a_failed_second_write_leaves_the_untouched_file_alone(
+    tmp_path: Path,
+) -> None:
+    """The file that was never written is not reported as changed."""
+    path = _setup(tmp_path, YAML)
+    other = tmp_path / "other.yaml"
+    other.write_bytes(b"x: 1\n")
+    edits = [_name_edit("garage"), LineEdit(other, 0, "x: 1", "x: 2")]
+    with (
+        patch(
+            "esphome.yaml_edit.write_keeping_mode",
+            side_effect=[None, EsphomeError("disk")],
+        ),
+        pytest.raises(EsphomeError) as info,
+    ):
+        apply_line_edits(edits)
+    assert "changed since it was written" not in str(info.value)
+    assert path.read_text() == YAML and other.read_text() == "x: 1\n"
+
+
+def test_restore_goes_on_after_an_interrupt(tmp_path: Path) -> None:
+    """A second Ctrl-C must not leave the other files holding the new value."""
+    path = _setup(tmp_path, YAML)
+    other = tmp_path / "other.yaml"
+    other.write_bytes(b"x: 1\n")
+    calls: list[Path] = []
+
+    def interrupt_at(n: int) -> Any:
+        def write(target: Path, text: str, **kwargs: Any) -> None:
+            calls.append(target)
+            if len(calls) == n:
+                raise KeyboardInterrupt
+            write_keeping_mode(target, text, **kwargs)
+
+        return write
+
+    originals = apply_line_edits(
+        [_name_edit("garage"), LineEdit(other, 0, "x: 1", "x: 2")]
+    )
+    with (
+        patch("esphome.yaml_edit.write_keeping_mode", side_effect=interrupt_at(1)),
+        pytest.raises(KeyboardInterrupt) as info,
+    ):
+        restore_files(originals)
+    assert calls == [path, other] and other.read_text() == "x: 1\n"
+    assert f"Could not restore {path}: interrupted" in "".join(info.value.__notes__)
+    # The interrupt stays one inside apply as well
+    _setup(tmp_path, YAML)
+    calls.clear()
+    with (
+        patch("esphome.yaml_util.load_yaml", side_effect=EsphomeError("broken")),
+        patch("esphome.yaml_edit.write_keeping_mode", side_effect=interrupt_at(2)),
+        pytest.raises(KeyboardInterrupt) as info,
+    ):
+        apply_line_edits([_name_edit("garage")])
+    assert "Could not restore" in "".join(info.value.__notes__)
+
+
+def test_restore_finishes_after_any_failure(tmp_path: Path) -> None:
+    """A raw OSError still leaves the other files put back, and an interrupt
+    while dropping the caches carries the report."""
+    path = _setup(tmp_path, YAML)
+    other = tmp_path / "other.yaml"
+    other.write_bytes(b"x: 1\n")
+    originals = apply_line_edits(
+        [_name_edit("garage"), LineEdit(other, 0, "x: 1", "x: 2")]
+    )
+    writes: list[Path] = []
+
+    def fail_first(target: Path, text: str, **kwargs: Any) -> None:
+        writes.append(target)
+        if len(writes) == 1:
+            raise OSError("odd")
+        write_keeping_mode(target, text, **kwargs)
+
+    with (
+        patch("esphome.yaml_edit.write_keeping_mode", side_effect=fail_first),
+        pytest.raises(RestoreError) as info,
+    ):
+        restore_files(originals)
+    assert f"Could not restore {path}: odd" in str(
+        info.value
+    )  # a Windows path is no regex
+    assert "garage" in path.read_text() and other.read_text() == "x: 1\n"
+    with (
+        patch(
+            "esphome.compiled_config.invalidate_compiled_config",
+            side_effect=KeyboardInterrupt,
+        ),
+        pytest.raises(KeyboardInterrupt) as info,
+    ):
+        restore_files(originals)
+    assert path.read_text() == YAML
+    assert "the validated caches: interrupted" in "".join(info.value.__notes__)
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="posix file modes")
+def test_restore_puts_back_the_mode_from_before_the_edit(tmp_path: Path) -> None:
+    """A write whose chmod failed leaves the file 0600; the restore does not
+    copy that, it has the mode from before."""
+    path = _setup(tmp_path, YAML)
+    path.chmod(0o644)
+    originals = apply_line_edits([_name_edit("garage")])
+    path.chmod(0o600)
+    restore_files(originals)
+    assert path.read_text() == YAML
+    assert path.stat().st_mode & 0o777 == 0o644
+
+
+def test_restore_finishes_after_an_exit(tmp_path: Path) -> None:
+    """A SystemExit from a signal handler still leaves the other files put
+    back and comes out again with the report."""
+    path = _setup(tmp_path, YAML)
+    other = tmp_path / "other.yaml"
+    other.write_bytes(b"x: 1\n")
+    originals = apply_line_edits(
+        [_name_edit("garage"), LineEdit(other, 0, "x: 1", "x: 2")]
+    )
+    writes: list[Path] = []
+
+    def exit_first(target: Path, text: str, **kwargs: Any) -> None:
+        writes.append(target)
+        if len(writes) == 1:
+            raise SystemExit(1)
+        write_keeping_mode(target, text, **kwargs)
+
+    with (
+        patch("esphome.yaml_edit.write_keeping_mode", side_effect=exit_first),
+        pytest.raises(SystemExit) as info,
+    ):
+        restore_files(originals)
+    assert other.read_text() == "x: 1\n"
+    assert f"Could not restore {path}: SystemExit" in "".join(info.value.__notes__)
+
+
+def test_a_value_written_to_the_secrets_file_is_registered(tmp_path: Path) -> None:
+    """Only a resolved !secret registers its value, so the main file is
+    loaded again after an edit elsewhere; own_documents keeps the registry."""
+    secrets = tmp_path / "secrets.yaml"
+    secrets.write_bytes(b"ssid: hidden\n")
+    _setup(tmp_path, YAML.replace("ssid: kitchen", "ssid: !secret ssid"))
+    own_documents()
+    assert yaml_util.is_secret("hidden") == "ssid"
+    apply_line_edits([LineEdit(secrets, 0, "ssid: hidden", "ssid: hidden2")])
+    assert yaml_util.is_secret("hidden2") == "ssid"