fix(windows): residual encoding gaps — MCP stdio, gateway update file-I/O, STT/TTS, desktop spawn, bootstrap ver-stub

apps/desktop/electron/backend-env.test.ts

@@ -68,6 +68,24 @@
   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 }),

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,

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.

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:

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",

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")

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)
 

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.

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

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",

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(

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(

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.")

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"))