diff --git a/src/nooa/tools/shell_tools.py b/src/nooa/tools/shell_tools.py index f0ea8ce9b..424045c72 100644 --- a/src/nooa/tools/shell_tools.py +++ b/src/nooa/tools/shell_tools.py @@ -318,6 +318,11 @@ class ShellTools(Skill): replace(match_or_path, ...) — edit at a Match anchor, or by unique string write_file(path, content) — create/overwrite a file + File reads, writes, and search anchors use UTF-8 by default. Set + ``encoding="cp1252"`` (or another codec) for legacy files, or + ``encoding=None`` to retain the platform's default file encoding. + This setting does not change the shell session's command-output encoding. + Grep that you can edit from directly. When run() executes a plain search (grep/rg/egrep), the result still prints the EXACT bytes your command produced — and it also carries ``.matches``, a list of Match objects you can @@ -347,9 +352,17 @@ class ShellTools(Skill): """ - def __init__(self, cwd: str = ".", init_command: str | None = None, **kwargs: Any): + def __init__( + self, + cwd: str = ".", + init_command: str | None = None, + *, + encoding: str | None = "utf-8", + **kwargs: Any, + ): super().__init__(**kwargs) self.cwd = Path(cwd).resolve() + self.encoding = encoding # Construct the session eagerly (it starts lazily on first run) so a # consumer wired at construction time — e.g. RepoTools(session=shell.session) # in the TUI — shares this shell's bash session instead of capturing None. @@ -585,7 +598,7 @@ async def _harvest_matches(self, command: str, displayed_stdout: str) -> list[Ma if mpath not in file_cache: try: resolved = self._resolve_path(mpath) - lines = resolved.read_text().splitlines(keepends=True) + lines = resolved.read_text(encoding=self.encoding).splitlines(keepends=True) file_cache[mpath] = (resolved, lines) except (OSError, ValueError): return None @@ -711,7 +724,7 @@ async def read( Match with .text, .numbered, .path, .start, .end. """ resolved = self._resolve_path(path) - content = resolved.read_text() + content = resolved.read_text(encoding=self.encoding) all_lines = content.splitlines(keepends=True) total = len(all_lines) @@ -753,7 +766,7 @@ async def replace( if isinstance(target, Match): new_text = old_or_new resolved = Path(target.resolved_path) - content = resolved.read_text() + content = resolved.read_text(encoding=self.encoding) all_lines = content.splitlines(keepends=True) before = all_lines[: target.start - 1] @@ -761,7 +774,7 @@ async def replace( if new_text and not new_text.endswith("\n") and after: new_text += "\n" new_content = "".join(before) + new_text + "".join(after) - resolved.write_text(new_content) + resolved.write_text(new_content, encoding=self.encoding) diff = f"--- a/{target.path}\n+++ b/{target.path}\n" diff += f"@@ -{target.start},{target.end - target.start + 1} @@\n" @@ -780,7 +793,7 @@ async def replace( ) old_text = old_or_new resolved = self._resolve_path(target) - content = resolved.read_text() + content = resolved.read_text(encoding=self.encoding) count = content.count(old_text) if count == 0: @@ -795,7 +808,7 @@ async def replace( ) new_content = content.replace(old_text, new, 1) - resolved.write_text(new_content) + resolved.write_text(new_content, encoding=self.encoding) return FileWrite( path=target, @@ -820,7 +833,7 @@ async def write_file( """ resolved = self._resolve_path(path) resolved.parent.mkdir(parents=True, exist_ok=True) - resolved.write_text(content) + resolved.write_text(content, encoding=self.encoding) line_count = content.count("\n") + (1 if content and not content.endswith("\n") else 0) return FileWrite( path=path, diff --git a/tests/tools/test_shell_file_encoding.py b/tests/tools/test_shell_file_encoding.py new file mode 100644 index 000000000..0fabeebcc --- /dev/null +++ b/tests/tools/test_shell_file_encoding.py @@ -0,0 +1,77 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +"""Consistent file encodings across ShellTools reads and edits, without Bash.""" + +import json +from unittest.mock import AsyncMock + +import pytest + +from nooa.tools.shell_tools import ShellTools + + +async def test_utf8_file_operations_with_legacy_default(tmp_path, monkeypatch): + monkeypatch.setattr("io.text_encoding", lambda encoding, stacklevel=2: encoding or "cp1252") + shell = ShellTools(cwd=str(tmp_path)) + try: + text = "coffee ☕\n中文\n" + await shell.write_file("note.txt", text) + assert (tmp_path / "note.txt").read_text(encoding="utf-8") == text + assert (await shell.read("note.txt")).text == text + await shell.replace("note.txt", "☕", "🚀") + region = await shell.read("note.txt", lines=(2, 2)) + await shell.replace(region, "café €\n") + expected = "coffee 🚀\ncafé €\n" + assert (tmp_path / "note.txt").read_text(encoding="utf-8") == expected + assert (await shell.read("note.txt")).text == expected + finally: + await shell.close() + + +async def test_read_existing_utf8_file_with_legacy_default(tmp_path, monkeypatch): + text = "café €\n" + (tmp_path / "note.txt").write_text(text, encoding="utf-8") + monkeypatch.setattr("io.text_encoding", lambda encoding, stacklevel=2: encoding or "cp1252") + shell = ShellTools(cwd=str(tmp_path)) + try: + assert (await shell.read("note.txt")).text == text + finally: + await shell.close() + + +async def test_search_anchors_use_file_encoding(tmp_path, monkeypatch): + text = "coffee café\n" + (tmp_path / "note.txt").write_text(text, encoding="utf-8") + monkeypatch.setattr("io.text_encoding", lambda encoding, stacklevel=2: encoding or "cp1252") + shell = ShellTools(cwd=str(tmp_path)) + session = AsyncMock() + session.run_with_timeout_flag.return_value = ( + json.dumps({"type": "match", "data": {"path": {"text": "note.txt"}, "line_number": 1}}), + "", + 0, + False, + ) + monkeypatch.setattr(shell, "_get_session", AsyncMock(return_value=session)) + try: + matches = await shell._harvest_matches("rg -n coffee note.txt", "1:" + text) + assert matches is not None + assert len(matches) == 1 + assert matches[0].text == text + await shell.replace(matches[0], "tea 中文\n") + assert (tmp_path / "note.txt").read_text(encoding="utf-8") == "tea 中文\n" + finally: + await shell.close() + + +@pytest.mark.parametrize("encoding", ["cp1252", None]) +async def test_explicit_legacy_encoding_preserves_existing_files(tmp_path, monkeypatch, encoding): + path = tmp_path / "legacy.txt" + path.write_bytes("café €\n".encode("cp1252")) + monkeypatch.setattr("io.text_encoding", lambda encoding, stacklevel=2: encoding or "cp1252") + shell = ShellTools(cwd=str(tmp_path), encoding=encoding) + try: + assert (await shell.read("legacy.txt")).text == "café €\n" + await shell.replace("legacy.txt", "café", "thé") + assert path.read_text(encoding="cp1252") == "thé €\n" + finally: + await shell.close()