diff --git a/tests/tools/test_computer_use.py b/tests/tools/test_computer_use.py index 9f4a6a5ca0513..1ea2f20a25877 100644 --- a/tests/tools/test_computer_use.py +++ b/tests/tools/test_computer_use.py @@ -1572,8 +1572,11 @@ class FakeProc: returncode = 0 stderr = "" # Daemon returns a path, not inline base64. - stdout = ('{"element_count": 7, "tree_markdown": "- [0] AXButton",' - ' "screenshot_file_path": "%s"}' % str(shot)) + stdout = json.dumps({ + "element_count": 7, + "tree_markdown": "- [0] AXButton", + "screenshot_file_path": str(shot), + }) import subprocess as _sp orig_run = _sp.run @@ -1687,6 +1690,44 @@ def cli_call(name, args, timeout): assert cap.app == "Finder" +class TestCliFallbackUtf8Decoding: + """Bug A: `_call_tool_via_cli` must force UTF-8 decoding for cua-driver + stdout so non-cp932 window titles do not break Japanese Windows hosts.""" + + def test_call_tool_via_cli_uses_utf8_with_replace(self): + from typing import Any, cast + from tools.computer_use.cua_backend import _CuaDriverSession + + session = cast(Any, _CuaDriverSession.__new__(_CuaDriverSession)) + captured_kw = {} + + class FakeProc: + returncode = 0 + stderr = "" + stdout = ('{"element_count": 2, "tree_markdown": ' + '"- [0] AXWindow \\"⛂ Claude Code\\""}') + + import subprocess as _sp + orig_run = _sp.run + + def fake_run(cmd, **kw): + captured_kw.update(kw) + return FakeProc() + + _sp.run = fake_run + try: + out = session._call_tool_via_cli( + "get_window_state", {"pid": 1, "window_id": 2}, 30.0 + ) + finally: + _sp.run = orig_run + + assert captured_kw.get("encoding") == "utf-8" + assert captured_kw.get("errors") == "replace" + assert "2 elements" in out["data"] + assert out["isError"] is False + + class TestCaptureAppFilterNoMatch: """capture(app=X) must not silently fall back to the frontmost window when X matches nothing — on a non-English macOS, list_windows returns diff --git a/tools/computer_use/cua_backend.py b/tools/computer_use/cua_backend.py index 32dae8b7b8611..9eed89b8112ff 100644 --- a/tools/computer_use/cua_backend.py +++ b/tools/computer_use/cua_backend.py @@ -163,7 +163,9 @@ def _resolve_mcp_invocation( from tools.environments.local import _sanitize_subprocess_env proc = subprocess.run( [driver_cmd, "manifest"], - capture_output=True, text=True, timeout=timeout, + capture_output=True, text=True, + encoding="utf-8", errors="replace", + timeout=timeout, stdin=subprocess.DEVNULL, # cua-driver is a third-party binary — never hand it provider # API keys via inherited env (same policy as the MCP and CLI @@ -254,7 +256,9 @@ def cua_driver_update_check(*, timeout: float = 8.0) -> Optional[Dict[str, Any]] from tools.environments.local import _sanitize_subprocess_env proc = subprocess.run( [_CUA_DRIVER_CMD, "check-update", "--json"], - capture_output=True, text=True, timeout=timeout, + capture_output=True, text=True, + encoding="utf-8", errors="replace", + timeout=timeout, # Some older drivers don't have the verb and fall through to a # stdin-reading mode rather than erroring — DEVNULL gives them EOF # so they exit fast instead of blocking until the timeout. @@ -920,7 +924,9 @@ def _call_tool_via_cli(self, name: str, args: Dict[str, Any], timeout: float) -> for attempt in range(attempts): try: proc = _subprocess.run( - cmd, capture_output=True, text=True, timeout=max(15.0, timeout), + cmd, capture_output=True, text=True, + encoding="utf-8", errors="replace", + timeout=max(15.0, timeout), env=_sanitize_subprocess_env(cua_driver_child_env()), ) except Exception as e: # pragma: no cover - subprocess spawn failure