Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 12 additions & 2 deletions agent/copilot_acp_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -703,7 +703,11 @@ def _handle_server_message(
if block_error:
raise PermissionError(block_error)
try:
content = path.read_text()
# Force UTF-8: read_text() with no encoding falls back to
# the system locale (cp1252/GBK on Windows) and raises
# UnicodeDecodeError on any non-ASCII source/doc the peer
# asked to read. The codebase treats all text files as UTF-8.
content = path.read_text(encoding="utf-8")
except FileNotFoundError:
content = ""
line = params.get("line")
Expand Down Expand Up @@ -732,7 +736,13 @@ def _handle_server_message(
f"Write denied: '{path}' is a protected system/credential file."
)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(str(params.get("content") or ""))
# Force UTF-8: write_text() with no encoding falls back to the
# system locale (cp1252/GBK on Windows) and raises
# UnicodeEncodeError on any non-ASCII content the peer asked to
# write (unicode in source, docs, emoji, CJK).
path.write_text(
str(params.get("content") or ""), encoding="utf-8"
)
response = {
"jsonrpc": "2.0",
"id": message_id,
Expand Down
47 changes: 47 additions & 0 deletions tests/agent/test_copilot_acp_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -226,6 +226,53 @@ def test_write_text_file_reuses_write_denylist(self) -> None:
self.assertIn("error", response)
self.assertFalse(target.exists())

def test_write_text_file_writes_utf8(self) -> None:
"""Non-ASCII content must persist as UTF-8, not the system locale
(cp1252/GBK on Windows would raise UnicodeEncodeError)."""
with tempfile.TemporaryDirectory() as tmpdir:
root = Path(tmpdir)
target = root / "note.md"
payload = "# Заголовок ✅\nステータス: 完了\n"

response = self._dispatch(
{
"jsonrpc": "2.0",
"id": 6,
"method": "fs/write_text_file",
"params": {"path": str(target), "content": payload},
},
cwd=str(root),
)

self.assertNotIn("error", response)
# Decode as UTF-8 and normalise platform newline translation
# (text-mode write turns \n into \r\n on Windows) — the point is
# that the non-ASCII bytes survived, not the line ending.
written = target.read_bytes().decode("utf-8").replace("\r\n", "\n")
self.assertEqual(written, payload)

def test_read_text_file_reads_utf8(self) -> None:
"""A UTF-8 file with non-ASCII content must decode regardless of the
host locale."""
with tempfile.TemporaryDirectory() as tmpdir:
root = Path(tmpdir)
src = root / "src.txt"
payload = "переменная = '完了' # ✅\n"
src.write_bytes(payload.encode("utf-8"))

response = self._dispatch(
{
"jsonrpc": "2.0",
"id": 7,
"method": "fs/read_text_file",
"params": {"path": str(src)},
},
cwd=str(root),
)

content = ((response.get("result") or {}).get("content") or "")
self.assertEqual(content, payload)

def test_write_text_file_respects_safe_root(self) -> None:
with tempfile.TemporaryDirectory() as tmpdir:
root = Path(tmpdir)
Expand Down