[core] Let the yaml editor insert lines, write with a rollback and restore

apply_line_edits rewrites the located lines in place, reloads every touched
file and drops the validated caches, undoing everything on any failure;
restore_files puts the originals back only over what was written. A key in
a mapping merge_config rebuilt is placed by its value's line.

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/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,23 @@ 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; the restore only puts
+    ``original`` back over ``written``, never over the user's own edit."""
+
+    original: str
+    written: str
 
 
 def field_line_re(
@@ -63,17 +79,56 @@ 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."""
+    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
+    documents.update(
+        map(str, yaml_util.discover_user_yaml_files(CORE.config_path).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
+        rng = getattr(mapping.get(name), "esp_range", None)
+        if rng is None or Path(rng.start_mark.document).resolve() not in (
+            own_documents()
+        ):
+            return None
+        return Path(rng.start_mark.document), rng.start_mark.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,6 +137,32 @@ def source_of(mapping: ConfigType, name: str) -> tuple[Path, int] | None:
     return Path(rng.start_mark.document), rng.start_mark.line
 
 
+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 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."""
@@ -103,12 +184,74 @@ 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
+    for edit in sorted(edits, key=lambda e: (e.line, e.insert_after), reverse=True):
         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]] = {}
+    for edit in edits:
+        by_path.setdefault(editable_file(edit.path), []).append(edit)
+    originals = {}
+    for path, own in by_path.items():
+        original = read_text(path)
+        originals[path] = Snapshot(original, rewritten_text(original, own))
+    try:
+        for path, snapshot in originals.items():
+            write_keeping_mode(path, snapshot.written)
+        for path in originals:
+            try:
+                yaml_util.load_yaml(path, 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; a file that failed or changed
+    meanwhile is left alone and reported. An interrupt is reported the same
+    way, so the other files are still put back."""
+    failed = []
+    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)
+        except (EsphomeError, KeyboardInterrupt) as err:
+            failed.append(f"{path}: {str(err) or 'interrupted'}")
+    try:
+        compiled_config.invalidate_compiled_config()
+    except EsphomeError as err:
+        failed.append(str(err))
+    if failed:
+        raise RestoreError("Could not restore " + "; ".join(failed))

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_yaml_edit.py

@@ -4,20 +4,30 @@ from __future__ import annotations
 
 from pathlib import Path
 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 +108,42 @@ 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
+
+
+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 +191,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 +217,207 @@ 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)
+    assert rewritten_text(read_text(path), [added, edit]) == (
+        text.replace("name: kitchen", "name: garage") + "\r\n  password: x"
+    )
+    edit.old_line = "changed"
+    with pytest.raises(EsphomeError, match="changed since it was read"):
+        rewritten_text(read_text(path), [added, edit])
+
+
+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("{}")
+    originals = apply_line_edits([_name_edit("garage")])
+    assert originals == {path: Snapshot(YAML, YAML.replace("kitchen  #", "garage  #"))}
+    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_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 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) -> None:
+            calls.append(target)
+            if len(calls) == n:
+                raise KeyboardInterrupt
+            write_keeping_mode(target, text)
+
+        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(EsphomeError, match="Could not restore .*: interrupted"),
+    ):
+        restore_files(originals)
+    assert calls == [path, other] and other.read_text() == "x: 1\n"
+    # Inside apply the failed rollback is reported with the reason it ran
+    _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(RestoreError, match="broken; Could not restore .*interrupted"),
+    ):
+        apply_line_edits([_name_edit("garage")])