fix(tools): save edited source files atomically (#2041)

* fix(tools): save edited source files atomically

Issue #1958 lists three direct open(path, "w") writes that can destroy a
file if the process stops partway through: two in MemoryManager and
CodeEditor._save_edited_file. The memory ones were fixed in #1969, which
is also where write_file_atomic comes from; this is the source-file half
the issue proposes as a follow-up.

A plain open(path, "w") truncates the file before the new content is
complete, so a crash, an OOM kill or a full disk during the write leaves
the file holding neither the old content nor the new one.

The issue asks for symlink and metadata behaviour to be confirmed before
source files use the helper. Measured: the executable bit already
survived, but a bare os.replace onto a symlink path swapped the link out
for a regular file and left its target holding stale content, which
open(path, "w") does not do. write_file_atomic now resolves the path
first, so it follows symlinks the way the call it replaces did, and the
temporary file lands in the destination's real directory.

Refs #1958

* test(tools): make the atomic-write tests platform-independent

Two of the new tests failed on the Windows CI job. Both were the tests,
not the change under test.

- The executable-bit test asserted 0o755 and read back 0o666: Windows
  does not model POSIX permission bits, and chmod there only toggles the
  read-only flag. Skipped on win32.
- The encoding test left newline unset, which is LineEnding.NATIVE and
  therefore newline=None, under which Python translates "\n" to
  os.linesep on write; it got b"caf\xe9\r\n". The line ending is now
  pinned so the test covers the encoding alone, matching the sibling
  test that already pins "\r\n" and passes on Windows.

Also covers three paths the first round missed, all of which pass both
with and without the change and are there to show it does not break
them: a file in a subdirectory, a symlinked parent directory (the path
is resolved in full, so the temp file lands in the destination's real
directory), and a non-ASCII filename. The last two skip where the
platform cannot create symlinks or represent the name.

Refs #1958

CHANGELOG.md

@@ -32,6 +32,14 @@ Status of the `main` branch. Changes prior to the next official version change w
     a warning while the verdict checked `FindSymbolTool` only. A reference-search failure now fails
     the check; a symbol with no references is still a pass
 
+* Tools:
+  - Fix: the file-editing tools saved the edited file with `open(path, "w")`, which truncates it
+    before the new content is complete, so a crash, an OOM kill or a full disk partway through the
+    write could leave a source file empty or half-written. Saves now go through the same atomic
+    temp-file-plus-`os.replace` helper that the memory writes already use. The helper resolves
+    symlinks first, so a symlinked file is still written through to its target rather than being
+    replaced by a regular file (#1958)
+
 * Memories:
   - Fix: `save_memory`/`edit_memory` wrote directly to the memory file with `open(path, "w")`, which
     truncates it before the new content is written; a crash, OOM kill, or full disk partway through

src/serena/code_editor.py

@@ -16,6 +16,7 @@ from solidlsp.ls_utils import PathUtils, TextStepper, TextUtils
 
 from .project import Project
 from .util.file_proxy import FileProxy
+from .util.file_system import write_file_atomic
 
 log = logging.getLogger(__name__)
 TSymbol = TypeVar("TSymbol", bound=Symbol)
@@ -91,8 +92,7 @@ class CodeEditor(Generic[TSymbol], ABC):
     def _save_edited_file(self, edited_file: "CodeEditor.EditedFile") -> None:
         abs_path = os.path.join(self.project_root, edited_file.relative_path)
         new_contents = edited_file.get_contents()
-        with open(abs_path, "w", encoding=self.encoding, newline=self.newline) as f:
-            f.write(new_contents)
+        write_file_atomic(abs_path, new_contents, encoding=self.encoding, newline=self.newline)
 
     @abstractmethod
     def _find_unique_symbol(self, name_path: str, relative_file_path: str) -> TSymbol:

src/serena/util/file_system.py

@@ -31,6 +31,11 @@ def write_file_atomic(path: str, content: str, *, encoding: str, newline: str |
     :param encoding: the encoding to use for the write
     :param newline: passed through to the underlying ``open()`` call to control newline translation
     """
+    # ``open(path, "w")`` follows symlinks and writes through to the target, whereas replacing the
+    # link path itself would swap the link out for a regular file and leave its target holding the
+    # old content. Resolving first keeps this a drop-in replacement, and puts the temporary file in
+    # the destination's real directory, which is where it has to be for the rename to be atomic.
+    path = os.path.realpath(path)
     target_dir = os.path.dirname(path) or "."
     try:
         existing_mode: int | None = stat.S_IMODE(os.stat(path).st_mode)

test/serena/test_code_editor_atomic_writes.py

@@ -0,0 +1,191 @@
+"""Tests that saving an edited source file cannot destroy the previous content (issue #1958).
+
+``CodeEditor._save_edited_file`` is the third of the three direct ``open(path, "w")`` writes the
+issue lists; the two in ``MemoryManager`` were addressed in #1969, which is also where
+``write_file_atomic`` comes from.
+
+The editor is exercised through a stub subclass so that the inherited save path can be driven
+without a language server.
+"""
+
+import os
+import stat
+import sys
+from collections.abc import Iterator
+from contextlib import contextmanager
+from typing import Any
+
+import pytest
+
+from serena.code_editor import CodeEditor
+from serena.util import file_system
+
+
+class _InMemoryEditedFile(CodeEditor.EditedFile):
+    """An ``EditedFile`` that holds its contents in memory."""
+
+    def __init__(self, relative_path: str, contents: str) -> None:
+        super().__init__(relative_path)
+        self._contents = contents
+
+    def get_contents(self) -> str:
+        return self._contents
+
+    def set_contents(self, contents: str) -> None:
+        self._contents = contents
+
+    def delete_text_between_positions(self, start_pos: Any, end_pos: Any) -> None:
+        raise NotImplementedError
+
+    def insert_text_at_position(self, pos: Any, text: str) -> None:
+        raise NotImplementedError
+
+
+class _StubCodeEditor(CodeEditor[Any]):
+    """A ``CodeEditor`` whose only inherited behaviour under test is the file-saving path."""
+
+    def __init__(self, project_root: str, encoding: str = "utf-8", newline: str | None = None) -> None:
+        self.project_root = project_root
+        self.encoding = encoding
+        self.newline = newline
+
+    @contextmanager
+    def _open_file_context(self, relative_path: str) -> Iterator[CodeEditor.EditedFile]:
+        abs_path = os.path.join(self.project_root, relative_path)
+        with open(abs_path, encoding=self.encoding) as f:
+            contents = f.read()
+        yield _InMemoryEditedFile(relative_path, contents)
+
+    def _find_unique_symbol(self, name_path: str, relative_file_path: str) -> Any:
+        raise NotImplementedError
+
+    def rename_symbol(self, name_path: str, relative_path: str, new_name: str) -> str:
+        raise NotImplementedError
+
+
+class TestSourceFileSaveIsAtomic:
+    def _editor(self, tmp_path: Any, **kwargs: Any) -> _StubCodeEditor:
+        return _StubCodeEditor(str(tmp_path), **kwargs)
+
+    def test_interrupted_save_keeps_the_previous_file_content(self, tmp_path, monkeypatch):
+        """A crash partway through the write must leave the file holding its old content.
+
+        The crash is injected into the temp-file write that ``write_file_atomic`` performs, so on
+        a non-atomic implementation nothing raises at all and this fails with ``DID NOT RAISE``,
+        which is exactly the state the issue describes: the real file is truncated first and there
+        is no intermediate copy to fall back to.
+        """
+        source = tmp_path / "module.py"
+        original = "def original():\n    return 1\n" * 40
+        source.write_text(original, encoding="utf-8")
+
+        real_fdopen = os.fdopen
+
+        def crashing_fdopen(fd: int, *args: Any, **kwargs: Any) -> Any:
+            f = real_fdopen(fd, *args, **kwargs)
+            real_write = f.write
+
+            def crashing_write(data: str) -> int:
+                real_write(data[: len(data) // 4])
+                f.flush()
+                raise RuntimeError("simulated crash mid-write")
+
+            f.write = crashing_write
+            return f
+
+        monkeypatch.setattr(file_system.os, "fdopen", crashing_fdopen)
+
+        editor = self._editor(tmp_path)
+        with pytest.raises(RuntimeError, match="simulated crash mid-write"):
+            with editor.edited_file_context("module.py") as edited:
+                edited.set_contents("def replacement():\n    return 2\n" * 40)
+
+        assert source.read_text(encoding="utf-8") == original
+        assert list(tmp_path.iterdir()) == [source], "the partial temp file must not be left behind"
+
+    def test_successful_save_writes_the_new_content(self, tmp_path):
+        """Control: the ordinary path still writes what was asked for."""
+        source = tmp_path / "module.py"
+        source.write_text("old\n", encoding="utf-8")
+
+        editor = self._editor(tmp_path)
+        with editor.edited_file_context("module.py") as edited:
+            edited.set_contents("new\n")
+
+        assert source.read_text(encoding="utf-8") == "new\n"
+        assert list(tmp_path.iterdir()) == [source]
+
+    def test_save_into_a_subdirectory(self, tmp_path):
+        """Every other test writes at the project root; the relative path is joined and resolved,
+        so a nested file has to work the same way.
+        """
+        package = tmp_path / "pkg" / "sub"
+        package.mkdir(parents=True)
+        source = package / "module.py"
+        source.write_text("old\n", encoding="utf-8")
+
+        editor = self._editor(tmp_path)
+        with editor.edited_file_context("pkg/sub/module.py") as edited:
+            edited.set_contents("new\n")
+
+        assert source.read_text(encoding="utf-8") == "new\n"
+        assert list(package.iterdir()) == [source], "no temp file may be left beside the source"
+
+    def test_save_writes_through_a_symlinked_source_file(self, tmp_path):
+        """A symlinked source file must keep being written through to its target, as
+        ``open(path, "w")`` did; the link itself must not be replaced by a regular file.
+        """
+        target_dir = tmp_path / "shared"
+        target_dir.mkdir()
+        target = target_dir / "shared.py"
+        target.write_text("old\n", encoding="utf-8")
+        project = tmp_path / "project"
+        project.mkdir()
+        link = project / "module.py"
+        try:
+            link.symlink_to(target)
+        except OSError as e:
+            pytest.skip(f"cannot create symlinks on this platform/permissions: {e}")
+
+        editor = self._editor(project)
+        with editor.edited_file_context("module.py") as edited:
+            edited.set_contents("new\n")
+
+        assert link.is_symlink(), "the source file's symlink must survive the edit"
+        assert target.read_text(encoding="utf-8") == "new\n", "the edit must reach the link's target"
+
+    @pytest.mark.skipif(
+        sys.platform == "win32", reason="Windows does not model POSIX permission bits; chmod only toggles the read-only flag"
+    )
+    def test_save_preserves_the_executable_bit(self, tmp_path):
+        script = tmp_path / "run.sh"
+        script.write_text("#!/bin/sh\necho old\n", encoding="utf-8")
+        os.chmod(script, 0o755)
+
+        editor = self._editor(tmp_path)
+        with editor.edited_file_context("run.sh") as edited:
+            edited.set_contents("#!/bin/sh\necho new\n")
+
+        assert stat.S_IMODE(os.stat(script).st_mode) == 0o755
+
+    def test_save_respects_the_configured_newline(self, tmp_path):
+        source = tmp_path / "module.py"
+        source.write_bytes(b"old\n")
+
+        editor = self._editor(tmp_path, newline="\r\n")
+        with editor.edited_file_context("module.py") as edited:
+            edited.set_contents("a\nb\n")
+
+        assert source.read_bytes() == b"a\r\nb\r\n"
+
+    def test_save_respects_the_configured_encoding(self, tmp_path):
+        source = tmp_path / "module.py"
+        source.write_text("alt\n", encoding="latin-1")
+
+        # newline is pinned so this test is about the encoding alone: LineEnding.NATIVE yields
+        # newline=None, under which Python translates "\n" to os.linesep on write
+        editor = self._editor(tmp_path, encoding="latin-1", newline="\n")
+        with editor.edited_file_context("module.py") as edited:
+            edited.set_contents("café\n")
+
+        assert source.read_bytes() == "café\n".encode("latin-1")

test/serena/util/test_file_system.py

@@ -920,3 +920,101 @@ class TestGitignoreParserPermissionError:
         finally:
             # Restore permissions so teardown can clean up
             os.chmod(unreadable, old_mode)
+
+
+class TestWriteFileAtomicSymlinks:
+    """``write_file_atomic`` replaces ``open(path, "w")`` at its call sites, so it has to agree
+    with it about symlinks: a plain write follows the link and updates its target, whereas a bare
+    ``os.replace`` onto the link path would swap the link itself out for a regular file and leave
+    the target holding stale content (issue #1958 asks for symlink behaviour to be preserved
+    before source files use this).
+    """
+
+    @staticmethod
+    def _symlink_or_skip(link: Path, target: Path) -> None:
+        """Windows needs developer mode or admin rights to create a symlink; skip there rather
+        than fail, matching how ``test_memories_manager.py`` handles the same limitation.
+        """
+        try:
+            link.symlink_to(target)
+        except OSError as e:
+            pytest.skip(f"cannot create symlinks on this platform/permissions: {e}")
+
+    def test_writes_through_a_symlink_instead_of_replacing_it(self, tmp_path):
+        target = tmp_path / "real.txt"
+        target.write_text("old", encoding="utf-8")
+        link = tmp_path / "link.txt"
+        self._symlink_or_skip(link, target)
+
+        write_file_atomic(str(link), "new", encoding="utf-8")
+
+        assert link.is_symlink(), "the symlink must survive the write, not be replaced by a regular file"
+        assert target.read_text(encoding="utf-8") == "new", "the content must reach the link's target"
+
+    def test_writes_through_a_symlink_pointing_outside_its_directory(self, tmp_path):
+        outside = tmp_path / "outside"
+        outside.mkdir()
+        target = outside / "real.txt"
+        target.write_text("old", encoding="utf-8")
+        inside = tmp_path / "inside"
+        inside.mkdir()
+        link = inside / "link.txt"
+        self._symlink_or_skip(link, target)
+
+        write_file_atomic(str(link), "new", encoding="utf-8")
+
+        assert link.is_symlink()
+        assert target.read_text(encoding="utf-8") == "new"
+        assert list(inside.iterdir()) == [link], "no temp file may be left beside the link"
+
+    def test_broken_symlink_creates_its_target(self, tmp_path):
+        """``open(path, "w")`` on a dangling link creates the target; this must do the same."""
+        target = tmp_path / "missing.txt"
+        link = tmp_path / "link.txt"
+        self._symlink_or_skip(link, target)
+
+        write_file_atomic(str(link), "new", encoding="utf-8")
+
+        assert link.is_symlink()
+        assert target.read_text(encoding="utf-8") == "new"
+
+    def test_writes_through_a_symlinked_parent_directory(self, tmp_path):
+        """The path is resolved in full, so a symlinked *directory* on the way to the file is
+        followed too, and the temporary file is created in the destination's real directory (it has
+        to be on the same filesystem as the destination for the rename to be atomic).
+        """
+        real_dir = tmp_path / "real_dir"
+        real_dir.mkdir()
+        target = real_dir / "file.txt"
+        target.write_text("old", encoding="utf-8")
+        link_dir = tmp_path / "link_dir"
+        self._symlink_or_skip(link_dir, real_dir)
+
+        write_file_atomic(str(link_dir / "file.txt"), "new", encoding="utf-8")
+
+        assert link_dir.is_symlink(), "the directory symlink must survive"
+        assert target.read_text(encoding="utf-8") == "new"
+        assert list(real_dir.iterdir()) == [target], "no temp file may be left in the real directory"
+
+    def test_non_ascii_filename_round_trips(self, tmp_path):
+        target = tmp_path / "測試檔案.txt"
+        try:
+            target.write_text("old", encoding="utf-8")
+        except (OSError, UnicodeError) as e:
+            pytest.skip(f"cannot create non-ASCII filenames on this filesystem: {e}")
+
+        write_file_atomic(str(target), "new", encoding="utf-8")
+
+        assert target.read_text(encoding="utf-8") == "new"
+        assert list(tmp_path.iterdir()) == [target]
+
+    def test_regular_file_is_written_in_place(self, tmp_path):
+        """Control: the symlink handling must not change the ordinary case."""
+        target = tmp_path / "plain.txt"
+        target.write_text("old", encoding="utf-8")
+
+        write_file_atomic(str(target), "new", encoding="utf-8")
+
+        assert not target.is_symlink()
+        assert target.read_text(encoding="utf-8") == "new"
+        assert list(tmp_path.iterdir()) == [target]