diff --git a/apps/desktop/electron/backend-env.test.ts b/apps/desktop/electron/backend-env.test.ts
index e24a66ee396e..c42595a82472 100644
--- a/apps/desktop/electron/backend-env.test.ts
+++ b/apps/desktop/electron/backend-env.test.ts
@@ -68,6 +68,24 @@ test('buildDesktopBackendEnv extends PYTHONPATH and backend PATH together', () =
   assert.ok(env.PATH.includes('/opt/homebrew/bin'))
 })
 
+test('buildDesktopBackendEnv forces PYTHONUTF8 unless the user set it explicitly', () => {
+  const defaulted = buildDesktopBackendEnv({
+    hermesHome: '/Users/test/.hermes',
+    currentEnv: { PATH: '/usr/bin' },
+    platform: 'darwin',
+    pathModule: path.posix
+  })
+  assert.equal(defaulted.PYTHONUTF8, '1')
+
+  const optedOut = buildDesktopBackendEnv({
+    hermesHome: '/Users/test/.hermes',
+    currentEnv: { PATH: '/usr/bin', PYTHONUTF8: '0' },
+    platform: 'darwin',
+    pathModule: path.posix
+  })
+  assert.equal(optedOut.PYTHONUTF8, '0')
+})
+
 test('normalizeHermesHomeRoot maps profile homes back to the global Hermes root', () => {
   assert.equal(
     normalizeHermesHomeRoot('/Users/test/.hermes/profiles/oracle', { pathModule: path.posix }),
diff --git a/apps/desktop/electron/backend-env.ts b/apps/desktop/electron/backend-env.ts
index a225e2381776..3db4a19d0341 100644
--- a/apps/desktop/electron/backend-env.ts
+++ b/apps/desktop/electron/backend-env.ts
@@ -104,6 +104,13 @@ function buildDesktopBackendEnv({
 
   return {
     PYTHONPATH: appendUniquePathEntries([...pythonPathEntries, currentPythonPath], { delimiter }),
+    // Force PEP 540 UTF-8 mode in the spawned Python backend so its stdio and
+    // subprocess defaults are UTF-8 even on non-UTF-8 Windows locales (GBK,
+    // cp1252, ...). hermes_bootstrap sets this inside the child too, but only
+    // after import — anything emitted earlier (interpreter startup errors,
+    // pre-bootstrap tracebacks) still decodes with the locale default without
+    // this. User's explicit setting wins. Re-port of PR #56499 (echoriver89).
+    PYTHONUTF8: currentEnv?.PYTHONUTF8 ?? '1',
     [key]: buildDesktopBackendPath({
       hermesHome,
       venvRoot,
diff --git a/gateway/dead_targets.py b/gateway/dead_targets.py
index 66a9247f213e..15f9b908794f 100644
--- a/gateway/dead_targets.py
+++ b/gateway/dead_targets.py
@@ -67,7 +67,7 @@ def __init__(self, path: Optional[Path] = None) -> None:
     def _load(self) -> None:
         try:
             if self._path.exists():
-                raw = json.loads(self._path.read_text())
+                raw = json.loads(self._path.read_text(encoding="utf-8"))
                 if isinstance(raw, dict):
                     # Only keep well-shaped entries.
                     self._dead = {
@@ -82,7 +82,7 @@ def _flush_locked(self) -> None:
         try:
             self._path.parent.mkdir(parents=True, exist_ok=True)
             tmp = self._path.with_suffix(self._path.suffix + ".tmp")
-            tmp.write_text(json.dumps(self._dead, indent=2))
+            tmp.write_text(json.dumps(self._dead, indent=2), encoding="utf-8")
             tmp.replace(self._path)
         except OSError as exc:
             # Best-effort: keep the in-memory state, don't break delivery.
diff --git a/gateway/delivery.py b/gateway/delivery.py
index 3672a7c49e3a..fa43db6d0f92 100644
--- a/gateway/delivery.py
+++ b/gateway/delivery.py
@@ -429,7 +429,7 @@ def _deliver_local(
         lines.append("")
         lines.append(content)
         
-        output_path.write_text("\n".join(lines))
+        output_path.write_text("\n".join(lines), encoding="utf-8")
         
         return {
             "path": str(output_path),
@@ -442,7 +442,7 @@ def _save_full_output(self, content: str, job_id: str) -> Path:
         out_dir = get_hermes_home() / "cron" / "output"
         out_dir.mkdir(parents=True, exist_ok=True)
         path = out_dir / f"{job_id}_{timestamp}.txt"
-        path.write_text(content)
+        path.write_text(content, encoding="utf-8")
         return path
 
     def _filter_silence_narration_enabled(self) -> bool:
diff --git a/gateway/platforms/qqbot/adapter.py b/gateway/platforms/qqbot/adapter.py
index cb9a64e3bd48..2816326efb3f 100644
--- a/gateway/platforms/qqbot/adapter.py
+++ b/gateway/platforms/qqbot/adapter.py
@@ -1202,7 +1202,7 @@ def _write_update_response(answer: str, operator: str = "") -> None:
             home = get_hermes_home()
             response_path = home / ".update_response"
             tmp = response_path.with_suffix(".tmp")
-            tmp.write_text(answer)
+            tmp.write_text(answer, encoding="utf-8")
             tmp.replace(response_path)
             logger.info(
                 "QQ update prompt answered %r by %s",
diff --git a/gateway/run.py b/gateway/run.py
index 72f2b2626ada..5f9219bcd643 100644
--- a/gateway/run.py
+++ b/gateway/run.py
@@ -3778,7 +3778,7 @@ def _voice_key(self, platform: Platform, chat_id: str) -> str:
 
     def _load_voice_modes(self) -> Dict[str, str]:
         try:
-            data = json.loads(self._VOICE_MODE_PATH.read_text())
+            data = json.loads(self._VOICE_MODE_PATH.read_text(encoding="utf-8"))
         except (FileNotFoundError, json.JSONDecodeError, OSError):
             return {}
 
@@ -3806,7 +3806,7 @@ def _save_voice_modes(self) -> None:
         try:
             self._VOICE_MODE_PATH.parent.mkdir(parents=True, exist_ok=True)
             self._VOICE_MODE_PATH.write_text(
-                json.dumps(self._voice_mode, indent=2)
+                json.dumps(self._voice_mode, indent=2), encoding="utf-8"
             )
         except OSError as e:
             logger.warning("Failed to save voice modes: %s", e)
@@ -6956,7 +6956,7 @@ def _increment_restart_failure_counts(self, active_session_keys: set) -> None:
 
         path = _hermes_home / self._STUCK_LOOP_FILE
         try:
-            counts = json.loads(path.read_text()) if path.exists() else {}
+            counts = json.loads(path.read_text(encoding="utf-8")) if path.exists() else {}
         except Exception:
             counts = {}
 
@@ -6986,7 +6986,7 @@ def _suspend_stuck_loop_sessions(self) -> int:
             return 0
 
         try:
-            counts = json.loads(path.read_text())
+            counts = json.loads(path.read_text(encoding="utf-8"))
         except Exception:
             return 0
 
@@ -7032,7 +7032,7 @@ def _clear_restart_failure_count(self, session_key: str) -> None:
         if not path.exists():
             return
         try:
-            counts = json.loads(path.read_text())
+            counts = json.loads(path.read_text(encoding="utf-8"))
             if session_key in counts:
                 del counts[session_key]
                 if counts:
@@ -10774,7 +10774,7 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]:
                 prompt_path = _hermes_home / ".update_prompt.json"
                 try:
                     tmp = response_path.with_suffix(".tmp")
-                    tmp.write_text(response_text)
+                    tmp.write_text(response_text, encoding="utf-8")
                     tmp.replace(response_path)
                     prompt_path.unlink(missing_ok=True)
                 except OSError as e:
@@ -10794,7 +10794,7 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]:
                 prompt_path = _hermes_home / ".update_prompt.json"
                 try:
                     tmp = response_path.with_suffix(".tmp")
-                    tmp.write_text("")
+                    tmp.write_text("", encoding="utf-8")
                     tmp.replace(response_path)
                     prompt_path.unlink(missing_ok=True)
                     logger.info(
@@ -14743,7 +14743,7 @@ def _is_stale_restart_redelivery(self, event: MessageEvent) -> bool:
                     self._booted_from_restart = False
                     return True
                 return False
-            data = json.loads(marker_path.read_text())
+            data = json.loads(marker_path.read_text(encoding="utf-8"))
         except Exception:
             return False
 
@@ -16697,7 +16697,7 @@ async def _watch_update_progress(
         for path in (claimed_path, pending_path):
             if path.exists():
                 try:
-                    pending = json.loads(path.read_text())
+                    pending = json.loads(path.read_text(encoding="utf-8"))
                     platform_str = pending.get("platform")
                     chat_id = pending.get("chat_id")
                     chat_type = pending.get("chat_type")
@@ -16736,7 +16736,7 @@ async def _watch_update_progress(
                     return
                 await asyncio.sleep(poll_interval)
             if (pending_path.exists() or claimed_path.exists()) and not exit_code_path.exists():
-                exit_code_path.write_text("124")
+                exit_code_path.write_text("124", encoding="utf-8")
                 await self._send_update_notification()
             return
 
@@ -16778,7 +16778,7 @@ async def _flush_buffer() -> None:
                 # Read any remaining output
                 if output_path.exists():
                     try:
-                        content = output_path.read_text()
+                        content = output_path.read_text(encoding="utf-8")
                         if len(content) > bytes_sent:
                             buffer += content[bytes_sent:]
                             bytes_sent = len(content)
@@ -16788,7 +16788,7 @@ async def _flush_buffer() -> None:
 
                 # Send final status
                 try:
-                    exit_code_raw = exit_code_path.read_text().strip() or "1"
+                    exit_code_raw = exit_code_path.read_text(encoding="utf-8").strip() or "1"
                     exit_code = int(exit_code_raw)
                     if exit_code == 0:
                         await adapter.send(
@@ -16817,7 +16817,7 @@ async def _flush_buffer() -> None:
             # Check for new output
             if output_path.exists():
                 try:
-                    content = output_path.read_text()
+                    content = output_path.read_text(encoding="utf-8")
                     if len(content) > bytes_sent:
                         buffer += content[bytes_sent:]
                         bytes_sent = len(content)
@@ -16835,7 +16835,7 @@ async def _flush_buffer() -> None:
             if (prompt_path.exists() and session_key
                     and not self._update_prompt_pending.get(session_key)):
                 try:
-                    prompt_data = json.loads(prompt_path.read_text())
+                    prompt_data = json.loads(prompt_path.read_text(encoding="utf-8"))
                     prompt_text = prompt_data.get("prompt", "")
                     default = prompt_data.get("default", "")
                     if prompt_text:
@@ -16883,7 +16883,7 @@ async def _flush_buffer() -> None:
         # Timeout
         if not exit_code_path.exists():
             logger.warning("Update watcher timed out after %.0fs", timeout)
-            exit_code_path.write_text("124")
+            exit_code_path.write_text("124", encoding="utf-8")
             await _flush_buffer()
             try:
                 await adapter.send(
@@ -16929,7 +16929,7 @@ async def _send_update_notification(self) -> bool:
             elif not claimed_path.exists():
                 return True
 
-            pending = json.loads(claimed_path.read_text())
+            pending = json.loads(claimed_path.read_text(encoding="utf-8"))
             platform_str = pending.get("platform")
             chat_id = pending.get("chat_id")
             chat_type = pending.get("chat_type")
@@ -16943,13 +16943,13 @@ async def _send_update_notification(self) -> bool:
                 claimed_path.replace(pending_path)
                 return False
 
-            exit_code_raw = exit_code_path.read_text().strip() or "1"
+            exit_code_raw = exit_code_path.read_text(encoding="utf-8").strip() or "1"
             exit_code = int(exit_code_raw)
 
             # Read the captured update output
             output = ""
             if output_path.exists():
-                output = output_path.read_text()
+                output = output_path.read_text(encoding="utf-8")
 
             # Resolve adapter
             platform = Platform(platform_str)
@@ -17024,7 +17024,7 @@ async def _send_restart_notification(self) -> Optional[tuple[str, str, Optional[
             return None
 
         try:
-            data = json.loads(notify_path.read_text())
+            data = json.loads(notify_path.read_text(encoding="utf-8"))
             platform_str = data.get("platform")
             chat_id = data.get("chat_id")
             chat_type = data.get("chat_type")
diff --git a/gateway/slash_commands.py b/gateway/slash_commands.py
index e70cc25d72d3..58c365ddb3e7 100644
--- a/gateway/slash_commands.py
+++ b/gateway/slash_commands.py
@@ -4963,7 +4963,7 @@ async def _handle_update_command(self, event: MessageEvent) -> str:
         if event.message_id:
             pending["message_id"] = event.message_id
         _tmp_pending = pending_path.with_suffix(".tmp")
-        _tmp_pending.write_text(json.dumps(pending))
+        _tmp_pending.write_text(json.dumps(pending), encoding="utf-8")
         _tmp_pending.replace(pending_path)
         exit_code_path.unlink(missing_ok=True)
 
diff --git a/gateway/status.py b/gateway/status.py
index d2d3c1509188..365518e26e90 100644
--- a/gateway/status.py
+++ b/gateway/status.py
@@ -617,7 +617,7 @@ def _read_pid_record(pid_path: Optional[Path] = None) -> Optional[dict]:
         return None
 
     try:
-        raw = pid_path.read_text().strip()
+        raw = pid_path.read_text(encoding="utf-8").strip()
     except (OSError, UnicodeDecodeError):
         # File was deleted between exists() and read_text(), permission
         # flipped, or it holds non-UTF-8 / binary garbage.
diff --git a/hermes_bootstrap.py b/hermes_bootstrap.py
index ae23cc976296..c0622bb0d961 100644
--- a/hermes_bootstrap.py
+++ b/hermes_bootstrap.py
@@ -122,6 +122,49 @@ def apply_windows_utf8_bootstrap() -> bool:
     return True
 
 
+def suppress_platform_ver_console() -> None:
+    """Stub ``platform._syscmd_ver`` on Windows — decode-crash + flash guard.
+
+    CPython's ``platform.win32_ver()`` (reached via ``platform.uname()`` /
+    ``platform.platform()``, which the OpenAI SDK touches for its
+    platform headers) shells out ``cmd /c ver``. Two failure modes:
+
+    - **Console flash**: the ``check_output(..., shell=True)`` call has no
+      ``CREATE_NO_WINDOW``, so a windowless parent (pythonw gateway, slash
+      workers, kanban workers) flashes a visible console per call.
+    - **UnicodeDecodeError on Python 3.11.0/3.11.1**: those micros lack
+      CPython's ``encoding="locale"`` fix (added 3.11.2), so under PEP 540
+      UTF-8 mode (which we enable above) the ``ver`` output — OEM code page
+      bytes on localized Windows — is strict-utf-8 decoded and raises,
+      crashing ``platform.platform()`` in any process that inherits
+      ``PYTHONUTF8=1`` (issue #69413).
+
+    Stubbing ``_syscmd_ver`` to return its inputs makes ``win32_ver()`` hit
+    its documented fallback and read the version from
+    ``sys.getwindowsversion()`` — same data, in-process, no subprocess.
+    Mirrors ``hermes_cli._subprocess_compat.suppress_platform_ver_console``
+    (kept there for callers that don't import bootstrap); double
+    application is harmless. Lives here so EVERY entry point gets it —
+    ``tui_gateway/slash_worker.py``, ``tui_gateway/entry.py``,
+    ``run_agent.py``, ``batch_runner.py``, and ``cli.py`` import only
+    ``hermes_bootstrap``, never ``hermes_cli.main``.
+    """
+    if not _IS_WINDOWS:
+        return
+    try:
+        import platform
+
+        if hasattr(platform, "_syscmd_ver"):
+            def _quiet_syscmd_ver(system="", release="", version="",
+                                  supported_platforms=("win32", "win16", "dos")):
+                return system, release, version
+
+            platform._syscmd_ver = _quiet_syscmd_ver
+    except Exception:
+        # Hardening only — never let it break an entry point.
+        pass
+
+
 def harden_import_path(src_root: str | None = None) -> None:
     """Stop a package in the current directory from shadowing Hermes modules.
 
@@ -188,6 +231,7 @@ def activate_durable_lazy_target() -> None:
 # the very top of their module, before importing anything else.  The
 # import side effect does the right thing.
 apply_windows_utf8_bootstrap()
+suppress_platform_ver_console()
 
 # Activate the durable lazy-install target (immutable Docker images) so
 # packages installed into the data volume on a previous run are importable
diff --git a/plugins/platforms/discord/adapter.py b/plugins/platforms/discord/adapter.py
index 161669a8a6ec..6e66a2c269a4 100644
--- a/plugins/platforms/discord/adapter.py
+++ b/plugins/platforms/discord/adapter.py
@@ -8164,7 +8164,7 @@ async def _respond(
                 home = get_hermes_home()
                 response_path = home / ".update_response"
                 tmp = response_path.with_suffix(".tmp")
-                tmp.write_text(answer)
+                tmp.write_text(answer, encoding="utf-8")
                 tmp.replace(response_path)
                 logger.info(
                     "Discord update prompt answered '%s' by %s",
diff --git a/plugins/platforms/feishu/adapter.py b/plugins/platforms/feishu/adapter.py
index 93571de06199..e1df2a05e5e5 100644
--- a/plugins/platforms/feishu/adapter.py
+++ b/plugins/platforms/feishu/adapter.py
@@ -2180,7 +2180,7 @@ def _build_resolved_update_prompt_card(*, answer: str, user_name: str) -> Dict[s
     def _write_update_prompt_response(answer: str) -> None:
         response_path = get_hermes_home() / ".update_response"
         tmp_path = response_path.with_suffix(".tmp")
-        tmp_path.write_text(answer)
+        tmp_path.write_text(answer, encoding="utf-8")
         tmp_path.replace(response_path)
 
     async def send_voice(
diff --git a/plugins/platforms/google_chat/adapter.py b/plugins/platforms/google_chat/adapter.py
index 540f9bf84566..0b73968d92fa 100644
--- a/plugins/platforms/google_chat/adapter.py
+++ b/plugins/platforms/google_chat/adapter.py
@@ -558,7 +558,7 @@ def load(self) -> None:
             self._counts = {}
             return
         try:
-            raw = self._path.read_text()
+            raw = self._path.read_text(encoding="utf-8")
             data = json.loads(raw) if raw.strip() else {}
         except json.JSONDecodeError as exc:
             logger.warning(
@@ -613,7 +613,7 @@ def _save(self) -> None:
         try:
             self._path.parent.mkdir(parents=True, exist_ok=True)
             tmp = self._path.with_suffix(self._path.suffix + ".tmp")
-            tmp.write_text(json.dumps(self._counts, separators=(",", ":")))
+            tmp.write_text(json.dumps(self._counts, separators=(",", ":")), encoding="utf-8")
             os.replace(tmp, self._path)
         except OSError as exc:
             logger.warning(
diff --git a/plugins/platforms/google_chat/oauth.py b/plugins/platforms/google_chat/oauth.py
index b66b0829f29d..1b8c89cb2e4b 100644
--- a/plugins/platforms/google_chat/oauth.py
+++ b/plugins/platforms/google_chat/oauth.py
@@ -427,7 +427,7 @@ def store_client_secret(path: str) -> None:
         sys.exit(1)
 
     try:
-        data = json.loads(src.read_text())
+        data = json.loads(src.read_text(encoding="utf-8"))
     except json.JSONDecodeError:
         print("ERROR: File is not valid JSON.")
         sys.exit(1)
@@ -467,7 +467,7 @@ def _load_pending_auth(email: Optional[str] = None) -> dict:
         print("ERROR: No pending OAuth session found. Run --auth-url first.")
         sys.exit(1)
     try:
-        data = json.loads(pending.read_text())
+        data = json.loads(pending.read_text(encoding="utf-8"))
     except Exception as exc:
         print(f"ERROR: Could not read pending OAuth session: {exc}")
         print("Run --auth-url again to start a fresh session.")
diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py
index 77eb7dcb42a4..60e2d7bb9ee1 100644
--- a/plugins/platforms/telegram/adapter.py
+++ b/plugins/platforms/telegram/adapter.py
@@ -6352,7 +6352,7 @@ async def _handle_callback_query(
             home = get_hermes_home()
             response_path = home / ".update_response"
             tmp = response_path.with_suffix(".tmp")
-            tmp.write_text(answer)
+            tmp.write_text(answer, encoding="utf-8")
             tmp.replace(response_path)
             logger.info("Telegram update prompt answered '%s' by user %s",
                         answer, getattr(query.from_user, "id", "unknown"))
diff --git a/plugins/platforms/whatsapp/adapter.py b/plugins/platforms/whatsapp/adapter.py
index 8736274a76cc..8d654197e04a 100644
--- a/plugins/platforms/whatsapp/adapter.py
+++ b/plugins/platforms/whatsapp/adapter.py
@@ -167,7 +167,7 @@ def _kill_stale_bridge_by_pidfile(session_path: Path) -> None:
     try:
         # Format: line 1 = pid, optional line 2 = kernel start time. Legacy
         # files written before the guard existed have only the pid.
-        lines = pid_file.read_text().split("\n")
+        lines = pid_file.read_text(encoding="utf-8").split("\n")
         pid = int(lines[0].strip())
         if len(lines) > 1 and lines[1].strip():
             recorded_start = int(lines[1].strip())
@@ -208,7 +208,7 @@ def _write_bridge_pidfile(session_path: Path, pid: int) -> None:
         from gateway.status import get_process_start_time
         start = get_process_start_time(pid)
         text = str(pid) if start is None else "{}\n{}".format(pid, start)
-        (session_path / "bridge.pid").write_text(text)
+        (session_path / "bridge.pid").write_text(text, encoding="utf-8")
     except OSError:
         pass
 
@@ -531,7 +531,9 @@ async def connect(self, *, is_reconnect: bool = False) -> bool:
             _deps_fresh = False
             if (bridge_dir / "node_modules").exists():
                 try:
-                    _deps_fresh = (_dep_stamp.read_text().strip() == _pkg_hash) and bool(_pkg_hash)
+                    _deps_fresh = (
+                        _dep_stamp.read_text(encoding="utf-8").strip() == _pkg_hash
+                    ) and bool(_pkg_hash)
                 except OSError:
                     _deps_fresh = False
             if not _deps_fresh:
@@ -557,7 +559,7 @@ async def connect(self, *, is_reconnect: bool = False) -> bool:
                     print(f"[{self.name}] Dependencies installed")
                     if _pkg_hash:
                         try:
-                            _dep_stamp.write_text(_pkg_hash)
+                            _dep_stamp.write_text(_pkg_hash, encoding="utf-8")
                         except OSError:
                             pass  # Stamp is an optimization; install still succeeded
                 except Exception as e:
diff --git a/tests/gateway/test_gateway_utf8_encoding.py b/tests/gateway/test_gateway_utf8_encoding.py
new file mode 100644
index 000000000000..2ecc559e6ad1
--- /dev/null
+++ b/tests/gateway/test_gateway_utf8_encoding.py
@@ -0,0 +1,60 @@
+"""Static guard: every ``read_text`` / ``write_text`` call in the gateway and
+bundled update-response adapters must pass an explicit ``encoding=`` keyword
+argument so non-UTF-8 Windows locales don't corrupt file IPC.  Mirrors the
+AST-based guard pattern in
+``tests/tools/test_windows_compat.py``.
+"""
+
+import ast
+import pathlib
+import pytest
+
+REPO_ROOT = pathlib.Path(__file__).resolve().parents[2]
+GATEWAY_DIR = REPO_ROOT / "gateway"
+UPDATE_RESPONSE_FILES = (
+    REPO_ROOT / "plugins/platforms/discord/adapter.py",
+    REPO_ROOT / "plugins/platforms/telegram/adapter.py",
+    REPO_ROOT / "plugins/platforms/feishu/adapter.py",
+    REPO_ROOT / "plugins/platforms/whatsapp/adapter.py",
+    REPO_ROOT / "plugins/platforms/google_chat/adapter.py",
+    REPO_ROOT / "plugins/platforms/google_chat/oauth.py",
+)
+METHODS = {"read_text", "write_text"}
+SUPPRESSION = "# gateway-utf8: ok"
+
+
+def _find_violations():
+    violations = []
+    py_files = list(GATEWAY_DIR.rglob("*.py")) + list(UPDATE_RESPONSE_FILES)
+    for py_file in sorted(py_files):
+        source = py_file.read_text(encoding="utf-8")
+        source_lines = source.splitlines()
+        try:
+            tree = ast.parse(source, filename=str(py_file))
+        except SyntaxError:
+            continue
+        for node in ast.walk(tree):
+            if not isinstance(node, ast.Call):
+                continue
+            func = node.func
+            if not isinstance(func, ast.Attribute):
+                continue
+            if func.attr not in METHODS:
+                continue
+            if any(kw.arg == "encoding" for kw in node.keywords):
+                continue
+            lineno = node.lineno
+            if lineno <= len(source_lines) and SUPPRESSION in source_lines[lineno - 1]:
+                continue
+            rel = py_file.relative_to(REPO_ROOT)
+            violations.append(f"{rel}:{lineno}")
+    return violations
+
+
+def test_all_read_write_text_pass_encoding():
+    violations = _find_violations()
+    assert not violations, (
+        "Bare read_text()/write_text() calls found (missing encoding= kwarg).\n"
+        "Add encoding=\"utf-8\" or suppress with '# gateway-utf8: ok':\n"
+        + "\n".join(f"  {v}" for v in violations)
+    )
diff --git a/tests/test_hermes_bootstrap.py b/tests/test_hermes_bootstrap.py
index 50a582bf998a..6c2ecaf425ee 100644
--- a/tests/test_hermes_bootstrap.py
+++ b/tests/test_hermes_bootstrap.py
@@ -396,3 +396,45 @@ def test_defaults_to_module_dir(self):
             sys.path[:] = original
             if original_env is not None:
                 os.environ["HERMES_PYTHON_SRC_ROOT"] = original_env
+
+
+class TestSuppressPlatformVerConsole:
+    """suppress_platform_ver_console: stub applied on Windows, no-op on POSIX."""
+
+    def test_noop_on_posix(self, monkeypatch):
+        import platform
+        hb = _fresh_import()
+        original = getattr(platform, "_syscmd_ver", None)
+        monkeypatch.setattr(hb, "_IS_WINDOWS", False)
+        hb.suppress_platform_ver_console()
+        assert getattr(platform, "_syscmd_ver", None) is original
+
+    def test_stub_applied_when_windows(self, monkeypatch):
+        import platform
+        hb = _fresh_import()
+        original = getattr(platform, "_syscmd_ver", None)
+        try:
+            monkeypatch.setattr(hb, "_IS_WINDOWS", True)
+            hb.suppress_platform_ver_console()
+            stubbed = platform._syscmd_ver
+            assert stubbed is not original
+            # Stub returns its inputs — win32_ver()'s documented fallback path.
+            assert stubbed("s", "r", "v") == ("s", "r", "v")
+            # No-arg call (how Lib/platform.py invokes it in the fallback
+            # probe) must not raise — the rejected PR #69522 wrapper
+            # TypeError'd here.
+            assert stubbed() == ("", "", "")
+        finally:
+            if original is not None:
+                platform._syscmd_ver = original
+
+    def test_never_raises(self, monkeypatch):
+        hb = _fresh_import()
+        monkeypatch.setattr(hb, "_IS_WINDOWS", True)
+        import platform
+        original = platform._syscmd_ver
+        try:
+            monkeypatch.delattr(platform, "_syscmd_ver")
+            hb.suppress_platform_ver_console()  # hasattr guard → silent no-op
+        finally:
+            platform._syscmd_ver = original
diff --git a/tests/tools/test_mcp_stdio_encoding_handler.py b/tests/tools/test_mcp_stdio_encoding_handler.py
new file mode 100644
index 000000000000..3a84ea92a776
--- /dev/null
+++ b/tests/tools/test_mcp_stdio_encoding_handler.py
@@ -0,0 +1,103 @@
+"""Tests for MCP stdio encoding error handler fix (issue #46099).
+
+On Windows, pipe I/O can produce non-UTF-8 bytes at chunk boundaries,
+causing UnicodeDecodeError when the MCP SDK's TextReceiveStream uses
+errors="strict". This test verifies that StdioServerParameters is created
+with encoding_error_handler="replace".
+"""
+
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from tools.mcp_tool import MCPServerTask, _MCP_AVAILABLE
+
+pytestmark = pytest.mark.skipif(not _MCP_AVAILABLE, reason="MCP SDK not installed")
+
+
+class TestStdioEncodingErrorHandler:
+    """Verify that _run_stdio passes encoding_error_handler='replace'."""
+
+    def test_stdio_server_params_uses_replace_encoding_handler(self):
+        """StdioServerParameters must use encoding_error_handler='replace'.
+
+        On Windows, pipe chunk boundaries can split multi-byte UTF-8 sequences,
+        producing bytes that fail with errors='strict'. Using 'replace' ensures
+        undecodable bytes become U+FFFD instead of crashing.
+        """
+        mock_session = MagicMock()
+        mock_session.initialize = AsyncMock()
+        mock_session.list_tools = AsyncMock(return_value=SimpleNamespace(tools=[]))
+
+        mock_stdio_cm = MagicMock()
+        mock_stdio_cm.__aenter__ = AsyncMock(return_value=(object(), object()))
+        mock_stdio_cm.__aexit__ = AsyncMock(return_value=False)
+
+        mock_session_cm = MagicMock()
+        mock_session_cm.__aenter__ = AsyncMock(return_value=mock_session)
+        mock_session_cm.__aexit__ = AsyncMock(return_value=False)
+
+        async def _test():
+            with (
+                patch("tools.mcp_tool.StdioServerParameters") as mock_params,
+                patch("tools.mcp_tool.stdio_client", return_value=mock_stdio_cm),
+                patch("tools.mcp_tool.ClientSession", return_value=mock_session_cm),
+                patch("tools.mcp_tool._snapshot_child_pids", return_value=set()),
+                patch("tools.mcp_tool._write_stderr_log_header"),
+                patch("tools.mcp_tool._get_mcp_stderr_log", return_value=None),
+            ):
+                server = MCPServerTask("test-encoding")
+                await server.start({
+                    "command": "echo",
+                    "args": ["hello"],
+                })
+
+                call_kwargs = mock_params.call_args.kwargs
+                assert call_kwargs["encoding_error_handler"] == "replace", (
+                    f"Expected encoding_error_handler='replace', "
+                    f"got '{call_kwargs.get('encoding_error_handler')}'"
+                )
+
+                await server.shutdown()
+
+        asyncio.run(_test())
+
+    def test_stdio_server_params_defaults_encoding_utf8(self):
+        """Verify that the default encoding (utf-8) is not overridden."""
+        mock_session = MagicMock()
+        mock_session.initialize = AsyncMock()
+        mock_session.list_tools = AsyncMock(return_value=SimpleNamespace(tools=[]))
+
+        mock_stdio_cm = MagicMock()
+        mock_stdio_cm.__aenter__ = AsyncMock(return_value=(object(), object()))
+        mock_stdio_cm.__aexit__ = AsyncMock(return_value=False)
+
+        mock_session_cm = MagicMock()
+        mock_session_cm.__aenter__ = AsyncMock(return_value=mock_session)
+        mock_session_cm.__aexit__ = AsyncMock(return_value=False)
+
+        async def _test():
+            with (
+                patch("tools.mcp_tool.StdioServerParameters") as mock_params,
+                patch("tools.mcp_tool.stdio_client", return_value=mock_stdio_cm),
+                patch("tools.mcp_tool.ClientSession", return_value=mock_session_cm),
+                patch("tools.mcp_tool._snapshot_child_pids", return_value=set()),
+                patch("tools.mcp_tool._write_stderr_log_header"),
+                patch("tools.mcp_tool._get_mcp_stderr_log", return_value=None),
+            ):
+                server = MCPServerTask("test-encoding")
+                await server.start({
+                    "command": "echo",
+                    "args": ["hello"],
+                })
+
+                call_kwargs = mock_params.call_args.kwargs
+                # encoding is not explicitly set — StdioServerParameters defaults to utf-8
+                # We just verify we don't override it
+                assert "encoding" not in call_kwargs or call_kwargs.get("encoding") == "utf-8"
+
+                await server.shutdown()
+
+        asyncio.run(_test())
diff --git a/tools/mcp_tool.py b/tools/mcp_tool.py
index 35382471a143..3a1964b0cf08 100644
--- a/tools/mcp_tool.py
+++ b/tools/mcp_tool.py
@@ -2411,6 +2411,10 @@ async def _run_stdio(self, config: dict):
             command=command,
             args=args,
             env=safe_env if safe_env else None,
+            # On Windows, pipe I/O can deliver non-UTF-8 bytes at chunk
+            # boundaries.  Use "replace" to substitute undecodable bytes
+            # with U+FFFD instead of crashing with UnicodeDecodeError.
+            encoding_error_handler="replace",
         )
 
         sampling_kwargs = self._sampling.session_kwargs() if self._sampling else {}
diff --git a/tools/transcription_tools.py b/tools/transcription_tools.py
index 1198f53d133a..dd8c60275ad6 100644
--- a/tools/transcription_tools.py
+++ b/tools/transcription_tools.py
@@ -556,6 +556,10 @@ def _run_command_stt(command: str, timeout: float) -> subprocess.CompletedProces
         "stdout": subprocess.PIPE,
         "stderr": subprocess.PIPE,
         "text": True,
+        # Lossy UTF-8 decode — locale-mismatched bytes from the STT command
+        # must not raise in the reader threads on non-UTF-8 Windows (#45099).
+        "encoding": "utf-8",
+        "errors": "replace",
         "env": delegated_child_subprocess_env(),
     }
     if os.name == "nt":
diff --git a/tools/tts_tool.py b/tools/tts_tool.py
index 27c761e17cfc..446a04aa58cc 100644
--- a/tools/tts_tool.py
+++ b/tools/tts_tool.py
@@ -782,6 +782,10 @@ def _run_command_tts(command: str, timeout: float) -> subprocess.CompletedProces
         "stdout": subprocess.PIPE,
         "stderr": subprocess.PIPE,
         "text": True,
+        # Lossy UTF-8 decode — locale-mismatched bytes from the TTS command
+        # must not raise in the reader threads on non-UTF-8 Windows (#45099).
+        "encoding": "utf-8",
+        "errors": "replace",
         "env": delegated_child_subprocess_env(),
     }
     if os.name == "nt":