From 8280dff842c5adab220376aca9060466a6e0fb36 Mon Sep 17 00:00:00 2001 From: scwf Date: Thu, 23 Jul 2026 14:54:12 +0800 Subject: [PATCH] fix(windows): prevent Unicode output crashes on legacy code pages Legacy Windows code pages could crash CLI and host operations when user-facing output contained unsupported Unicode glyphs. Make CLI stdio configuration aware of terminals versus redirected streams so console output uses UTF-8 without breaking CP936 subprocess capture. Restore the attached console's previous output code page on process exit so short Omnigent commands do not affect later programs in the parent shell. Route host lifecycle and authentication messages through safe console output so encoding failures cannot interrupt tunnel handling after a runner has started or stopped. Initialize stdio in standalone CLI and host-daemon module entry points, and force daemon stdout and stderr to UTF-8 with replacement so managed subprocess paths receive the same protection. Add regression coverage for GBK consoles, redirected streams, code-page restoration, Unicode runner paths, module entry points, daemon startup, and daemon log encoding. Signed-off-by: scwf --- omnigent/_platform.py | 97 +++++++++++++ omnigent/cli.py | 12 +- omnigent/host/_daemon_entry.py | 4 + omnigent/host/connect.py | 37 +++-- tests/cli/test_backend.py | 13 ++ tests/cli/test_cli.py | 22 +++ tests/host/test_connect.py | 66 +++++++++ tests/host/test_daemon_entry.py | 24 ++++ tests/inner/test_proc_and_platform.py | 196 ++++++++++++++++++++++++++ 9 files changed, 450 insertions(+), 21 deletions(-) create mode 100644 tests/host/test_daemon_entry.py diff --git a/omnigent/_platform.py b/omnigent/_platform.py index 5117825a32..dd080c5d64 100644 --- a/omnigent/_platform.py +++ b/omnigent/_platform.py @@ -120,6 +120,103 @@ def resolve_cli_binary(name: str, *, env_var: str | None = None) -> str | None: #: True on macOS specifically (the seatbelt sandbox platform). IS_DARWIN = sys.platform == "darwin" +#: Windows console code page for UTF-8 (``chcp 65001``). +_WINDOWS_UTF8_CP = 65001 + + +def _stream_is_tty(stream: object) -> bool: + """Return whether *stream* is attached to an interactive terminal.""" + try: + return bool(stream.isatty()) # type: ignore[attr-defined] + except (AttributeError, OSError, ValueError): + return False + + +def _set_windows_console_output_utf8() -> None: + """Switch the attached Windows console output code page to UTF-8 temporarily.""" + import atexit + import ctypes + + kernel32 = ctypes.windll.kernel32 # type: ignore[attr-defined] + previous_cp = int(kernel32.GetConsoleOutputCP()) + if previous_cp <= 0 or previous_cp == _WINDOWS_UTF8_CP: + return + if kernel32.SetConsoleOutputCP(_WINDOWS_UTF8_CP): + atexit.register(_restore_windows_console_output_cp, previous_cp) + + +def _restore_windows_console_output_cp(previous_cp: int) -> None: + """Restore *previous_cp* when Omnigent still owns the UTF-8 console setting.""" + import ctypes + + with suppress(Exception): + kernel32 = ctypes.windll.kernel32 # type: ignore[attr-defined] + if int(kernel32.GetConsoleOutputCP()) == _WINDOWS_UTF8_CP: + kernel32.SetConsoleOutputCP(previous_cp) + + +def configure_unicode_safe_stdio() -> bool: + """Make CLI stdout/stderr safe for Unicode on legacy Windows code pages. + + Chinese/Japanese Windows consoles often default to GBK (``cp936``). Writing + emoji or ``✓``/``✗`` then raises :exc:`UnicodeEncodeError`, which can crash + ``omnigent setup`` / ``config list`` and tear down the host tunnel reconnect + loop. On native Windows this reconfigures terminal streams to UTF-8 and + redirected streams to replacement-safe output without changing their + encoding, and switches the console output code page when possible. + + Idempotent and best-effort: failures are swallowed so a weird redirected + stream never blocks CLI startup. No-op on POSIX (including WSL). + + :returns: ``True`` when running on native Windows (configuration attempted), + ``False`` otherwise. + """ + if not IS_WINDOWS: + return False + has_console_stream = False + for stream in (sys.stdout, sys.stderr): + reconfigure = getattr(stream, "reconfigure", None) + if reconfigure is None: + continue + is_console = _stream_is_tty(stream) + has_console_stream = has_console_stream or is_console + with suppress(Exception): + if is_console: + reconfigure(encoding="utf-8", errors="replace") + else: + reconfigure(errors="replace") + if has_console_stream: + with suppress(Exception): + _set_windows_console_output_utf8() + return True + + +def safe_console_print( + message: str, + *, + file: object | None = None, + flush: bool = True, +) -> None: + """Print *message* without letting :exc:`UnicodeEncodeError` escape. + + Long-lived loops (the host tunnel) must not reconnect-loop because a + success/failure banner used a glyph the active code page cannot encode. + Falls back to a replacement-encoded line when the primary write fails. + + :param message: Text to print (may include emoji / box-drawing glyphs). + :param file: Stream to write to; defaults to :data:`sys.stdout`. + :param flush: Forwarded to :func:`print`. + """ + stream = sys.stdout if file is None else file + try: + print(message, file=stream, flush=flush) + except UnicodeEncodeError: + encoding = getattr(stream, "encoding", None) or "ascii" + fallback = message.encode(encoding, errors="replace").decode(encoding, errors="replace") + with suppress(Exception): + print(fallback, file=stream, flush=flush) + + #: Non-sensitive Windows environment variables that a spawned omnigent #: subprocess needs to function, for env-passthrough allowlists that otherwise #: assume POSIX names. Python uppercases env keys on Windows, so these match diff --git a/omnigent/cli.py b/omnigent/cli.py index ace1cd9972..683d314df9 100644 --- a/omnigent/cli.py +++ b/omnigent/cli.py @@ -28,7 +28,7 @@ from rich.console import Console from rich.table import Table -from omnigent._platform import IS_WINDOWS, resolve_repo_symlink +from omnigent._platform import IS_WINDOWS, configure_unicode_safe_stdio, resolve_repo_symlink from omnigent.cli_common import ( RESUME_PICKER_SENTINEL as _RESUME_PICKER_SENTINEL, ) @@ -1556,6 +1556,11 @@ def main() -> None: so unhandled exceptions are captured even when the user didn't enable ``--log`` or ``--debug-events``. """ + # Windows GBK consoles raise UnicodeEncodeError on emoji / ✓ / ✗ and can + # tear down setup/config/host. Do this before any user-facing print + # (crash handler, Rich, click). No-op on POSIX. + configure_unicode_safe_stdio() + # Friendly crash handler: replaces Python's raw traceback with a # calm, branded crash screen + a one-tap path to file a GitHub issue # (browser opens the repo's pre-filled bug-report template with the @@ -2650,6 +2655,9 @@ def _build_host_daemon_env( for key, value in os.environ.items() if key in _RUNNER_ENV_ALLOWLIST or key.startswith(daemon_env_prefixes) } + # The daemon owns this binary log file. Keep raw stdout/stderr aligned with + # the UTF-8 logging handler instead of mixing locale-encoded print output. + env["PYTHONIOENCODING"] = "utf-8:replace" return env @@ -10113,4 +10121,4 @@ def _run_bundled_agent(name: str, run_args: tuple[str, ...]) -> None: if __name__ == "__main__": - cli() + main() diff --git a/omnigent/host/_daemon_entry.py b/omnigent/host/_daemon_entry.py index 662dad6872..ee858aad41 100644 --- a/omnigent/host/_daemon_entry.py +++ b/omnigent/host/_daemon_entry.py @@ -28,6 +28,10 @@ def main() -> None: :raises SystemExit: If neither / both of ``--server`` and ``--local`` are provided. """ + from omnigent._platform import configure_unicode_safe_stdio + + configure_unicode_safe_stdio() + parser = argparse.ArgumentParser( description="Background host daemon", ) diff --git a/omnigent/host/connect.py b/omnigent/host/connect.py index a3553cd6e8..35a4bd6e09 100644 --- a/omnigent/host/connect.py +++ b/omnigent/host/connect.py @@ -22,7 +22,7 @@ import websockets.asyncio.client from websockets.exceptions import InvalidStatus, InvalidURI -from omnigent._platform import WINDOWS_ENV_PASSTHROUGH +from omnigent._platform import WINDOWS_ENV_PASSTHROUGH, safe_console_print from omnigent.env_credentials import env_names_with_omnigent_prefix from omnigent.harness_aliases import canonicalize_harness from omnigent.harness_availability import HARNESS_BINARY_MISSING, HarnessAvailability @@ -1018,11 +1018,10 @@ def _fatal_upgrade_error(self, exc: InvalidURI | InvalidStatus) -> HostConnectEr # terminal — print once per redirect streak so a foreground # `omnigent host` shows the auth problem and its fix instead # of sitting silent while it retries. - print( + safe_console_print( f"⚠ {cause} Retrying — this also happens briefly while " f"the server restarts. {self._credentials_fix_hint()}", file=sys.stderr, - flush=True, ) return None return self._classify_http_status(exc.response.status_code) @@ -1194,11 +1193,10 @@ async def _handle_launch( # host's own terminal shows lifecycle lines, but the runner's real # output — the agent turn, tracebacks — lands only in this file. session_line = f"\n session: {frame.session_id}" if frame.session_id else "" - print( + safe_console_print( f" ↑ Runner started: {runner_id} (pid={proc.pid})\n" f" log: {_display_log_path(log_path)}" f"{session_line}", - flush=True, ) return HostLaunchRunnerResultFrame( request_id=frame.request_id, @@ -1232,9 +1230,8 @@ def _handle_stop( handle.proc.kill() handle.proc.wait() _logger.info("Stopped runner %s", frame.runner_id) - print( + safe_console_print( f" ↓ Runner stopped: {frame.runner_id}", - flush=True, ) return HostStopRunnerResultFrame( request_id=frame.request_id, @@ -2187,15 +2184,14 @@ async def _serve_frames(self, ws: websockets.asyncio.client.ClientConnection) -> for runner_id, error in list(self._unreported_exits.items()): del self._unreported_exits[runner_id] await self._report_runner_exit(runner_id, error) - # ``print`` (not ``_logger.warning``) so the user always sees the - # success line after the noisy ``databricks.sdk`` warnings — - # otherwise the terminal goes silent after auth and there's no - # signal the WS handshake actually completed. - print( + # User-facing banner (not ``_logger``) so the line is visible after + # noisy ``databricks.sdk`` warnings. ``safe_console_print`` so a + # legacy Windows code page cannot UnicodeEncodeError-out of the + # tunnel loop and trigger reconnect flaps. + safe_console_print( f"✓ Connected as {self._identity.name!r} " f"({self._identity.host_id}), {len(hello.runners)} live runner(s). " "Listening for sessions — Ctrl-C to disconnect.", - flush=True, ) loop = asyncio.get_running_loop() @@ -2356,20 +2352,20 @@ def run_host_process( path = config_path or CONFIG_PATH identity = load_or_create_host_identity(path) if not path.exists(): - print(f"Auto-generated {path} ({identity.host_id}, name: {identity.name})") - print(f"Connecting to {server_url} as {identity.name!r} ({identity.host_id})") + safe_console_print(f"Auto-generated {path} ({identity.host_id}, name: {identity.name})") + safe_console_print(f"Connecting to {server_url} as {identity.name!r} ({identity.host_id})") # Tell the user where logs land up front — `omnigent host` used to run # silently, so a stuck/quiet host gave no hint where to look. Session # work goes to per-runner files under the runner dir (the exact # file is printed when each runner launches). The host process's # own diagnostics go to the host destination. - print(f"Session logs: {_display_log_path(_runner_log_dir())}/") - print(f"This host's log: {_display_log_path(host_log_path)}") + safe_console_print(f"Session logs: {_display_log_path(_runner_log_dir())}/") + safe_console_print(f"This host's log: {_display_log_path(host_log_path)}") from omnigent.cli_diagnostics import current_cli_log_path _cli_log = current_cli_log_path() if _cli_log is not None and _cli_log != host_log_path: - print(f"CLI diagnostics: {_display_log_path(_cli_log)}") + safe_console_print(f"CLI diagnostics: {_display_log_path(_cli_log)}") host = HostProcess(identity, server_url) try: @@ -2378,5 +2374,8 @@ def run_host_process( # Fail loud: a permanent connection failure must not look like the # process is still working. Print the cause + fix, then exit non-zero # instead of the old behavior of reconnecting silently forever. - print(f"\n✗ Could not connect to {server_url}.\n{exc}", file=sys.stderr, flush=True) + safe_console_print( + f"\n✗ Could not connect to {server_url}.\n{exc}", + file=sys.stderr, + ) raise SystemExit(1) from exc diff --git a/tests/cli/test_backend.py b/tests/cli/test_backend.py index 683893a493..650ea2cbfe 100644 --- a/tests/cli/test_backend.py +++ b/tests/cli/test_backend.py @@ -204,6 +204,19 @@ def test_ensure_host_daemon_local_inherits_data_dir_and_db_uri( assert env["OMNIGENT_DATABASE_URI"] == "postgresql://u:pw@h/db" +@pytest.mark.parametrize("server_url", [None, "https://example.databricksapps.com"]) +def test_build_host_daemon_env_forces_utf8_stdio( + monkeypatch: pytest.MonkeyPatch, + server_url: str | None, +) -> None: + """Daemon print output uses the same UTF-8 encoding as its log handler.""" + monkeypatch.setenv("PYTHONIOENCODING", "cp936") + + env = _build_host_daemon_env(server_url=server_url) + + assert env["PYTHONIOENCODING"] == "utf-8:replace" + + def test_build_host_daemon_env_local_preserves_server_credentials( monkeypatch: pytest.MonkeyPatch, ) -> None: diff --git a/tests/cli/test_cli.py b/tests/cli/test_cli.py index 75a1ae2c04..0de7fb4ad8 100644 --- a/tests/cli/test_cli.py +++ b/tests/cli/test_cli.py @@ -139,6 +139,28 @@ def test_python_module_entrypoint_uses_unified_click_cli() -> None: assert "Omnigent quick chat" not in result.stdout +def test_cli_module_entrypoint_configures_stdio() -> None: + """``python -m omnigent.cli`` must use the hardened console entry point.""" + marker = "__omnigent_stdio_configured__" + probe = ( + "import runpy,sys;" + "from omnigent import _platform;" + f"_platform.configure_unicode_safe_stdio=lambda:print({marker!r});" + "sys.argv=['python -m omnigent.cli','--help'];" + "runpy.run_module('omnigent.cli',run_name='__main__')" + ) + + result = subprocess.run( + [sys.executable, "-c", probe], + check=True, + capture_output=True, + text=True, + timeout=20, + ) + + assert marker in result.stdout.splitlines() + + @pytest.mark.parametrize( ("argv", "expected"), [ diff --git a/tests/host/test_connect.py b/tests/host/test_connect.py index d8ee4f3ed1..a9648543e9 100644 --- a/tests/host/test_connect.py +++ b/tests/host/test_connect.py @@ -3,8 +3,10 @@ from __future__ import annotations import asyncio +import io import logging import subprocess +import sys import time from pathlib import Path from unittest.mock import patch @@ -481,6 +483,54 @@ def _fake_popen(args: list[str], **kwargs: object) -> subprocess.Popen[bytes]: _cleanup_host(host) +async def test_handle_launch_returns_result_when_banner_is_not_gbk_encodable( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A Unicode log path cannot interrupt launch after the runner is tracked.""" + + class _LiveProcess: + pid = 4242 + returncode: int | None = None + + def poll(self) -> int | None: + return self.returncode + + def terminate(self) -> None: + self.returncode = 0 + + def wait(self, timeout: float | None = None) -> int: + return self.returncode or 0 + + def kill(self) -> None: + self.returncode = -9 + + unicode_home = tmp_path / "home-😀" + unicode_home.mkdir() + workspace = tmp_path / "project" + workspace.mkdir() + monkeypatch.setattr(Path, "home", classmethod(lambda _cls: unicode_home)) + output = io.TextIOWrapper(io.BytesIO(), encoding="gbk", errors="strict", write_through=True) + monkeypatch.setattr(sys, "stdout", output) + monkeypatch.setattr( + "omnigent.host.connect.subprocess.Popen", + lambda *args, **kwargs: _LiveProcess(), + ) + host = _make_host_process() + + result = await host._handle_launch( + HostLaunchRunnerFrame( + request_id="req_unicode_log", + binding_token="tok_unicode_log", + workspace=str(workspace), + ) + ) + + assert result.status == "launched" + assert result.runner_id in host._runners + _cleanup_host(host) + + class _FakeTunnel: """In-memory stand-in for the host's WebSocket tunnel connection. @@ -2566,6 +2616,22 @@ async def test_login_redirect_prints_warning_to_terminal( assert "omnigent login https://app.example.databricks.com" in err +async def test_login_redirect_warning_survives_gbk_stderr( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The warning glyph cannot turn an authentication retry into a disconnect.""" + stderr = io.TextIOWrapper(io.BytesIO(), encoding="gbk", errors="strict", write_through=True) + monkeypatch.setattr(sys, "stderr", stderr) + host = _host() + + result = host._fatal_upgrade_error( + InvalidURI("https://w/oidc/authorize", "scheme isn't ws or wss") + ) + + assert result is None + assert host._login_redirect_streak == 1 + + async def test_fresh_host_fails_loud_after_persistent_login_redirects( monkeypatch: pytest.MonkeyPatch, ) -> None: diff --git a/tests/host/test_daemon_entry.py b/tests/host/test_daemon_entry.py new file mode 100644 index 0000000000..dd3df6a47c --- /dev/null +++ b/tests/host/test_daemon_entry.py @@ -0,0 +1,24 @@ +"""Tests for the background host daemon process entry point.""" + +from __future__ import annotations + +import sys + +import pytest + +from omnigent import _platform +from omnigent.host import _daemon_entry + + +def test_daemon_configures_stdio_before_argument_validation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The standalone daemon receives the same Windows stdio hardening as the CLI.""" + calls: list[None] = [] + monkeypatch.setattr(_platform, "configure_unicode_safe_stdio", lambda: calls.append(None)) + monkeypatch.setattr(sys, "argv", ["omnigent.host._daemon_entry"]) + + with pytest.raises(SystemExit): + _daemon_entry.main() + + assert calls == [None] diff --git a/tests/inner/test_proc_and_platform.py b/tests/inner/test_proc_and_platform.py index b1e5c25b9f..e71ff53415 100644 --- a/tests/inner/test_proc_and_platform.py +++ b/tests/inner/test_proc_and_platform.py @@ -8,11 +8,17 @@ from __future__ import annotations +import atexit import contextlib +import ctypes +import io import os import subprocess +import sys import time +from collections.abc import Callable from pathlib import Path +from types import SimpleNamespace import psutil import pytest @@ -21,6 +27,44 @@ from omnigent.inner import _proc +class _TTYTextIOWrapper(io.TextIOWrapper): + """Text stream stand-in that reports an attached terminal.""" + + def isatty(self) -> bool: + return True + + +class _FakeKernel32: + """Minimal Windows console API stand-in for code-page lifecycle tests.""" + + def __init__(self, output_cp: int, *, set_succeeds: bool = True) -> None: + self.output_cp = output_cp + self.set_succeeds = set_succeeds + self.set_calls: list[int] = [] + + def GetConsoleOutputCP(self) -> int: + return self.output_cp + + def SetConsoleOutputCP(self, code_page: int) -> int: + self.set_calls.append(code_page) + if not self.set_succeeds: + return 0 + self.output_cp = code_page + return 1 + + +def _install_fake_kernel32( + monkeypatch: pytest.MonkeyPatch, + kernel32: _FakeKernel32, +) -> None: + monkeypatch.setattr( + ctypes, + "windll", + SimpleNamespace(kernel32=kernel32), + raising=False, + ) + + def _spin_cmd() -> list[str]: """A short-lived child process that does nothing but sleep.""" if os.name == "nt": @@ -40,6 +84,158 @@ def test_platform_flags_are_mutually_consistent() -> None: assert _platform.IS_WINDOWS != _platform.IS_POSIX +def test_configure_unicode_safe_stdio_noop_off_windows( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """POSIX (and forced-off Windows) must not touch streams.""" + monkeypatch.setattr(_platform, "IS_WINDOWS", False) + assert _platform.configure_unicode_safe_stdio() is False + + +def test_windows_console_output_cp_is_restored_at_exit( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A legacy console code page is restored after Omnigent exits.""" + kernel32 = _FakeKernel32(936) + _install_fake_kernel32(monkeypatch, kernel32) + registrations: list[tuple[Callable[..., object], tuple[object, ...]]] = [] + monkeypatch.setattr( + atexit, + "register", + lambda callback, *args: registrations.append((callback, args)), + ) + + _platform._set_windows_console_output_utf8() + + assert kernel32.set_calls == [_platform._WINDOWS_UTF8_CP] + assert len(registrations) == 1 + callback, args = registrations[0] + callback(*args) + assert kernel32.set_calls == [_platform._WINDOWS_UTF8_CP, 936] + + +@pytest.mark.parametrize("initial_cp", [0, _platform._WINDOWS_UTF8_CP]) +def test_windows_console_output_cp_skips_unneeded_switch( + monkeypatch: pytest.MonkeyPatch, + initial_cp: int, +) -> None: + """Unavailable and already-UTF-8 consoles need no mutation or cleanup.""" + kernel32 = _FakeKernel32(initial_cp) + _install_fake_kernel32(monkeypatch, kernel32) + registrations: list[object] = [] + monkeypatch.setattr(atexit, "register", lambda *args: registrations.append(args)) + + _platform._set_windows_console_output_utf8() + + assert kernel32.set_calls == [] + assert registrations == [] + + +def test_windows_console_output_cp_failed_switch_registers_no_restore( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A failed UTF-8 switch must not schedule a misleading restore.""" + kernel32 = _FakeKernel32(932, set_succeeds=False) + _install_fake_kernel32(monkeypatch, kernel32) + registrations: list[object] = [] + monkeypatch.setattr(atexit, "register", lambda *args: registrations.append(args)) + + _platform._set_windows_console_output_utf8() + + assert kernel32.set_calls == [_platform._WINDOWS_UTF8_CP] + assert registrations == [] + + +def test_windows_console_output_cp_restore_preserves_later_change( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Exit cleanup must not overwrite a code page changed by another owner.""" + kernel32 = _FakeKernel32(936) + _install_fake_kernel32(monkeypatch, kernel32) + registrations: list[tuple[Callable[..., object], tuple[object, ...]]] = [] + monkeypatch.setattr( + atexit, + "register", + lambda callback, *args: registrations.append((callback, args)), + ) + + _platform._set_windows_console_output_utf8() + kernel32.output_cp = 437 + callback, args = registrations[0] + callback(*args) + + assert kernel32.set_calls == [_platform._WINDOWS_UTF8_CP] + + +def test_configure_unicode_safe_stdio_reconfigures_gbk_console_streams( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Legacy GBK TextIOWrappers become UTF-8 so emoji / ✓ are printable.""" + monkeypatch.setattr(_platform, "IS_WINDOWS", True) + output_cp_calls: list[None] = [] + monkeypatch.setattr( + _platform, + "_set_windows_console_output_utf8", + lambda: output_cp_calls.append(None), + ) + out = _TTYTextIOWrapper(io.BytesIO(), encoding="gbk", errors="strict", write_through=True) + err = _TTYTextIOWrapper(io.BytesIO(), encoding="gbk", errors="strict", write_through=True) + monkeypatch.setattr(sys, "stdout", out) + monkeypatch.setattr(sys, "stderr", err) + # Probe: the pre-fix stream rejects the host banner glyph. + with pytest.raises(UnicodeEncodeError): + out.write("✓") + out.seek(0) + out.truncate() + + assert _platform.configure_unicode_safe_stdio() is True + assert out.encoding == "utf-8" + assert err.encoding == "utf-8" + assert output_cp_calls == [None] + # Must not raise — this is the Phase 0 CP936 failure mode. + print("✓ Connected 🖥️", file=out, flush=True) + out.seek(0) + assert "✓" in out.read() + + +def test_configure_unicode_safe_stdio_preserves_redirected_stream_encoding( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Pipes stay locale-decodable while replacing unsupported glyphs.""" + monkeypatch.setattr(_platform, "IS_WINDOWS", True) + output_cp_calls: list[None] = [] + monkeypatch.setattr( + _platform, + "_set_windows_console_output_utf8", + lambda: output_cp_calls.append(None), + ) + buf = io.BytesIO() + out = io.TextIOWrapper(buf, encoding="gbk", errors="strict", write_through=True) + monkeypatch.setattr(sys, "stdout", out) + monkeypatch.setattr(sys, "stderr", out) + + assert _platform.configure_unicode_safe_stdio() is True + assert out.encoding == "gbk" + assert out.errors == "replace" + assert output_cp_calls == [] + + print("✓ Connected", file=out, flush=True) + assert buf.getvalue().decode("gbk").splitlines() == ["? Connected"] + + +def test_safe_console_print_survives_gbk_strict_stream( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A glyph the stream cannot encode must not raise (host-tunnel safety).""" + buf = io.BytesIO() + gbk = io.TextIOWrapper(buf, encoding="gbk", errors="strict", write_through=True) + monkeypatch.setattr(sys, "stdout", gbk) + _platform.safe_console_print("✓ Connected as 'host'") + gbk.flush() + # Replacement bytes were written; no exception escaped. + assert buf.getvalue() + + def test_default_shell_argv_runs_an_echo() -> None: argv = _platform.default_shell_argv("echo omnigent-shell-ok") out = subprocess.run(argv, capture_output=True, text=True, check=True)