From a92d1d5f8cab8fd35b25cf9a20276b10025a266e Mon Sep 17 00:00:00 2001
From: 1wos <1wosomm1@gmail.com>
Date: Sun, 27 Sep 2026 22:46:47 +0900
Subject: [PATCH] fix(tools): preserve non-UTF-8 bytes in EditFileTool

EditFileTool read the target file with errors='replace' and wrote the whole
decoded string back, so editing one ASCII line permanently rewrote every
non-UTF-8 byte elsewhere in the file as U+FFFD while reporting status: ok.

The write-back also truncated the file before encoding, so a new_string that
could not be encoded left the file at zero bytes.

Decode with errors='surrogateescape', encode before calling write_file, and
return the tool's own error dict when the encode fails. write_file already
accepts str | bytes and _sync_write already has a 'wb' branch, so no new
dependency or error path is introduced.

Fixes #7314
---
 .../adk/tools/environment/_edit_file_tool.py  | 15 +++++-
 .../tools/environment/test_edit_file_tool.py  | 50 +++++++++++++++++++
 2 files changed, 63 insertions(+), 2 deletions(-)

diff --git a/src/google/adk/tools/environment/_edit_file_tool.py b/src/google/adk/tools/environment/_edit_file_tool.py
index fa9f3143f99..697159320a4 100644
--- a/src/google/adk/tools/environment/_edit_file_tool.py
+++ b/src/google/adk/tools/environment/_edit_file_tool.py
@@ -101,7 +101,7 @@ async def run_async(
 
     try:
       data_bytes = await self._environment.read_file(path)
-      content = data_bytes.decode('utf-8', errors='replace')
+      content = data_bytes.decode('utf-8', errors='surrogateescape')
     except FileNotFoundError:
       return {'status': 'error', 'error': f'File not found: {path}'}
 
@@ -130,7 +130,18 @@ async def run_async(
       }
 
     new_content = re.sub(pattern, lambda m: new_string, content, count=1)
-    await self._environment.write_file(path, new_content)
+    try:
+      # Encode before opening the file: a failure here must not truncate it.
+      data = new_content.encode('utf-8', errors='surrogateescape')
+    except UnicodeEncodeError:
+      return {
+          'status': 'error',
+          'error': (
+              '`new_string` contains characters that cannot be encoded. '
+              'The file was not modified.'
+          ),
+      }
+    await self._environment.write_file(path, data)
     return {'status': 'ok', 'message': f'Edited {path}'}
 
   def _detect_error_in_response(self, response: Any) -> Optional[str]:
diff --git a/tests/unittests/tools/environment/test_edit_file_tool.py b/tests/unittests/tools/environment/test_edit_file_tool.py
index 42204096eb9..72addb64f44 100644
--- a/tests/unittests/tools/environment/test_edit_file_tool.py
+++ b/tests/unittests/tools/environment/test_edit_file_tool.py
@@ -148,3 +148,53 @@ async def test_edit_file_handles_special_regex_chars(
     assert result["status"] == "ok"
     data = await env.read_file("test.txt")
     assert data == b"replaced\nline2"
+
+  @pytest.mark.asyncio
+  async def test_edit_file_preserves_non_utf8_bytes(
+      self, env: LocalEnvironment
+  ):
+    """Bytes outside the edited region survive the write-back."""
+    # Arrange
+    tool = EditFileTool(env)
+    await env.write_file("notes.py", b"# header \xe9\nOLD\n")
+
+    args = {
+        "path": "notes.py",
+        "old_string": "OLD",
+        "new_string": "NEW",
+    }
+
+    # Act
+    result = await tool.run_async(args=args, tool_context=None)
+
+    # Assert
+    assert result["status"] == "ok"
+    data = await env.read_file("notes.py")
+    assert data == b"# header \xe9\nNEW\n"
+
+  @pytest.mark.asyncio
+  async def test_edit_file_reports_unencodable_new_string(
+      self, env: LocalEnvironment
+  ):
+    """An unencodable `new_string` is refused and the file is left alone.
+
+    U+D800 is outside the U+DC80-U+DCFF range that surrogateescape can encode,
+    and it reaches the tool from ordinary model output because `json` accepts it.
+    """
+    # Arrange
+    tool = EditFileTool(env)
+    original = b"# header \xe9\nOLD\n"
+    await env.write_file("notes.py", original)
+
+    args = {
+        "path": "notes.py",
+        "old_string": "OLD",
+        "new_string": "\ud800",
+    }
+
+    # Act
+    result = await tool.run_async(args=args, tool_context=None)
+
+    # Assert
+    assert result["status"] == "error"
+    assert await env.read_file("notes.py") == original