diff --git a/omnigent/_platform.py b/omnigent/_platform.py
index 5117825a32a..dd080c5d646 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 ace1cd9972c..683d314df98 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 662dad68720..ee858aad418 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 a3553cd6e84..35a4bd6e090 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 683893a493e..650ea2cbfec 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 75a1ae2c048..0de7fb4ad8c 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 d8ee4f3ed1c..a9648543e93 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 00000000000..dd3df6a47cd
--- /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 b1e5c25b9f7..e71ff534150 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)