fix(tools): preserve non-UTF-8 bytes in EditFileTool

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]:

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