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 77b245d291c3..97472bd9d852 100644
--- a/gateway/delivery.py
+++ b/gateway/delivery.py
@@ -357,7 +357,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),
@@ -370,7 +370,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 2639ab52fdd3..bb6bd32f5a1f 100644
--- a/gateway/platforms/qqbot/adapter.py
+++ b/gateway/platforms/qqbot/adapter.py
@@ -1193,7 +1193,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 8256c283d4cf..2aeb1c977ee0 100644
--- a/gateway/run.py
+++ b/gateway/run.py
@@ -3177,7 +3177,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 {}
 
@@ -3205,7 +3205,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)
@@ -5923,7 +5923,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 = {}
 
@@ -5953,7 +5953,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
 
@@ -5999,7 +5999,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:
@@ -8832,7 +8832,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:
@@ -8852,7 +8852,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(
@@ -12184,7 +12184,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
 
@@ -14019,7 +14019,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")
@@ -14058,7 +14058,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
 
@@ -14100,7 +14100,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)
@@ -14110,7 +14110,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(
@@ -14139,7 +14139,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)
@@ -14157,7 +14157,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:
@@ -14205,7 +14205,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(
@@ -14251,7 +14251,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")
@@ -14265,13 +14265,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)
@@ -14346,7 +14346,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 14687e07dde0..44fd05fc92d7 100644
--- a/gateway/slash_commands.py
+++ b/gateway/slash_commands.py
@@ -4523,7 +4523,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 9b8a1b6f83c2..47b2bab84b6d 100644
--- a/gateway/status.py
+++ b/gateway/status.py
@@ -444,7 +444,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/plugins/platforms/discord/adapter.py b/plugins/platforms/discord/adapter.py
index 92955e631a03..322a28fd1800 100644
--- a/plugins/platforms/discord/adapter.py
+++ b/plugins/platforms/discord/adapter.py
@@ -6804,7 +6804,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 7dd7e238937c..9a628cab584c 100644
--- a/plugins/platforms/feishu/adapter.py
+++ b/plugins/platforms/feishu/adapter.py
@@ -2152,7 +2152,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/telegram/adapter.py b/plugins/platforms/telegram/adapter.py
index 40f1ab094238..a01645200ba8 100644
--- a/plugins/platforms/telegram/adapter.py
+++ b/plugins/platforms/telegram/adapter.py
@@ -5565,7 +5565,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 82f08199631e..c2deeaf9488b 100644
--- a/plugins/platforms/whatsapp/adapter.py
+++ b/plugins/platforms/whatsapp/adapter.py
@@ -166,7 +166,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())
@@ -207,7 +207,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
 
@@ -530,7 +530,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:
@@ -556,7 +558,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..c900d09d9319
--- /dev/null
+++ b/tests/gateway/test_gateway_utf8_encoding.py
@@ -0,0 +1,57 @@
+"""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",
+)
+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)
+    )