fix(environment): prevent truncation on invalid file content

src/google/adk/environment/_local_environment.py

@@ -236,8 +236,10 @@ def _sync_read(path: Path) -> bytes:
   def _sync_write(path: Path, content: str | bytes) -> None:
     os.makedirs(path.parent, exist_ok=True)
     if isinstance(content, str):
-      with open(path, 'w', encoding='utf-8', newline='') as f:
-        f.write(content)
+      data = content.encode('utf-8')
+    elif isinstance(content, bytes):
+      data = content
     else:
-      with open(path, 'wb') as f:
-        f.write(content)
+      raise TypeError('content must be str or bytes')
+    with open(path, 'wb') as f:
+      f.write(data)

tests/unittests/environment/test_local_environment.py

@@ -80,6 +80,32 @@ async def test_write_preserves_explicit_crlf(self, env: LocalEnvironment):
 
     assert data == b"first\r\nsecond\r\n"
 
+  @pytest.mark.asyncio
+  async def test_unencodable_text_does_not_truncate_existing_file(
+      self, env: LocalEnvironment
+  ):
+    """An encoding error leaves existing file bytes intact."""
+    original = b"keep these bytes\n"
+    await env.write_file("existing.txt", original)
+
+    with pytest.raises(UnicodeEncodeError):
+      await env.write_file("existing.txt", "replacement \ud800")
+
+    assert await env.read_file("existing.txt") == original
+
+  @pytest.mark.asyncio
+  async def test_invalid_content_does_not_truncate_existing_file(
+      self, env: LocalEnvironment
+  ):
+    """An invalid content type leaves existing file bytes intact."""
+    original = b"keep these bytes\n"
+    await env.write_file("existing.txt", original)
+
+    with pytest.raises(TypeError, match="content must be str or bytes"):
+      await env.write_file("existing.txt", 123)
+
+    assert await env.read_file("existing.txt") == original
+
   @pytest.mark.asyncio
   async def test_write_creates_parent_dirs(self, env: LocalEnvironment):
     """Parent directories are created automatically."""