fix: add encoding="utf-8" to Path.write_text() calls (P1)

agent/copilot_acp_client.py

@@ -732,7 +732,7 @@ def _handle_server_message(
                         f"Write denied: '{path}' is a protected system/credential file."
                     )
                 path.parent.mkdir(parents=True, exist_ok=True)
-                path.write_text(str(params.get("content") or ""))
+                path.write_text(str(params.get("content") or ""), encoding="utf-8")
                 response = {
                     "jsonrpc": "2.0",
                     "id": message_id,

gateway/dead_targets.py

@@ -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

@@ -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:

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

gateway/run.py

@@ -3049,7 +3049,7 @@ def _save_voice_modes(self) -> None:
             self._VOICE_MODE_PATH.parent.mkdir(parents=True, exist_ok=True)
             self._VOICE_MODE_PATH.write_text(
                 json.dumps(self._voice_mode, indent=2)
-            )
+, encoding="utf-8")
         except OSError as e:
             logger.warning("Failed to save voice modes: %s", e)
 
@@ -8408,7 +8408,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:
@@ -8428,7 +8428,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(
@@ -13524,7 +13524,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
 
@@ -13671,7 +13671,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(

gateway/slash_commands.py

@@ -4105,7 +4105,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)
 

hermes_cli/banner.py

@@ -363,7 +363,7 @@ def check_for_updates() -> Optional[int]:
     try:
         cache_file.write_text(
             json.dumps({"ts": now, "behind": behind, "rev": embedded_rev, "ver": VERSION})
-        )
+, encoding="utf-8")
     except Exception:
         pass
 

hermes_cli/container_boot.py

@@ -224,7 +224,7 @@ def _maybe_migrate_legacy_gateway_run_state(
             "desired_state": "running",
             "timestamp": int(time.time()),
             "migrated_from": "legacy-container-cmd",
-        }) + "\n")
+        }) + "\n", encoding="utf-8")
     return "running"
 
 
@@ -438,26 +438,26 @@ def _register_service(scandir: Path, profile: str, *, start: bool) -> None:
     tmp_dir.mkdir(parents=True)
 
     try:
-        (tmp_dir / "type").write_text("longrun\n")
+        (tmp_dir / "type").write_text("longrun\n", encoding="utf-8")
 
         # Reuse the manager's run-script rendering — single source of
         # truth so register_profile_gateway and reconcile_profile_gateways
         # stay consistent. extra_env is empty here; users who need
         # per-profile env can set it via the profile's config.yaml
         # (which the gateway itself loads).
         run = tmp_dir / "run"
-        run.write_text(S6ServiceManager._render_run_script(profile, extra_env={}))
+        run.write_text(S6ServiceManager._render_run_script(profile, extra_env={}), encoding="utf-8")
         run.chmod(0o755)
 
         finish = tmp_dir / "finish"
-        finish.write_text(S6ServiceManager._render_finish_script())
+        finish.write_text(S6ServiceManager._render_finish_script(), encoding="utf-8")
         finish.chmod(0o755)
 
         # Persistent log rotation (OQ8-C).
         log_subdir = tmp_dir / "log"
         log_subdir.mkdir()
         log_run = log_subdir / "run"
-        log_run.write_text(S6ServiceManager._render_log_run(profile))
+        log_run.write_text(S6ServiceManager._render_log_run(profile), encoding="utf-8")
         log_run.chmod(0o755)
 
         # The presence of a `down` file tells s6-supervise to NOT

hermes_cli/gateway.py

@@ -4045,7 +4045,7 @@ def launchd_install(force: bool = False):
     if _refuse_temp_home_service_write(new_plist, "launchd plist"):
         return
     print(f"Installing launchd service to: {plist_path}")
-    plist_path.write_text(new_plist)
+    plist_path.write_text(new_plist, encoding="utf-8")
 
     try:
         _launchctl_bootstrap(

hermes_cli/main.py

@@ -4539,7 +4539,7 @@ def _gateway_prompt(prompt_text: str, default: str = "", timeout: float = 300.0)
         "id": str(_uuid.uuid4()),
     }
     tmp = prompt_path.with_suffix(".tmp")
-    tmp.write_text(_json.dumps(payload))
+    tmp.write_text(_json.dumps(payload), encoding="utf-8")
     tmp.replace(prompt_path)
 
     # Poll for response
@@ -10013,7 +10013,7 @@ def _print_items(items, label, key, fallback_key=None):
         if gateway_mode:
             _exit_code_path = get_hermes_home() / ".update_exit_code"
             try:
-                _exit_code_path.write_text("0")
+                _exit_code_path.write_text("0", encoding="utf-8")
             except OSError:
                 pass
 

hermes_cli/profiles.py

@@ -439,7 +439,7 @@ def create_wrapper_script(name: str, target: Optional[str] = None) -> Optional[P
     if is_windows:
         wrapper_path = wrapper_dir / f"{canon}.bat"
         try:
-            wrapper_path.write_text(f"@echo off\r\nhermes -p {profile} %*\r\n")
+            wrapper_path.write_text(f"@echo off\r\nhermes -p {profile} %*\r\n", encoding="utf-8")
             return wrapper_path
         except OSError as e:
             print(f"⚠ Could not create wrapper at {wrapper_path}: {e}")
@@ -448,7 +448,7 @@ def create_wrapper_script(name: str, target: Optional[str] = None) -> Optional[P
         wrapper_path = wrapper_dir / canon
         try:
             hermes_exe = shutil.which("hermes") or "hermes"
-            wrapper_path.write_text(f'#!/bin/sh\nexec {shlex.quote(hermes_exe)} -p {profile} "$@"\n')
+            wrapper_path.write_text(f'#!/bin/sh\nexec {shlex.quote(hermes_exe)} -p {profile} "$@"\n', encoding="utf-8")
             wrapper_path.chmod(wrapper_path.stat().st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH)
             return wrapper_path
         except OSError as e:
@@ -1619,7 +1619,7 @@ def set_active_profile(name: str) -> None:
     else:
         # Atomic write
         tmp = path.with_suffix(".tmp")
-        tmp.write_text(canon + "\n")
+        tmp.write_text(canon + "\n", encoding="utf-8")
         tmp.replace(path)
 
 

hermes_cli/service_manager.py

@@ -384,7 +384,7 @@ def _write_gateway_desired_state(name: str, desired_state: str) -> None:
         data["desired_state"] = desired_state
         data["updated_at"] = int(time.time())
         tmp = state_file.with_suffix(state_file.suffix + ".tmp")
-        tmp.write_text(json.dumps(data, separators=(",", ":")) + "\n")
+        tmp.write_text(json.dumps(data, separators=(",", ":")) + "\n", encoding="utf-8")
         tmp.replace(state_file)
     except OSError:
         return
@@ -987,22 +987,22 @@ def register_profile_gateway(
         tmp_dir.mkdir(parents=True)
 
         try:
-            (tmp_dir / "type").write_text("longrun\n")
+            (tmp_dir / "type").write_text("longrun\n", encoding="utf-8")
 
             run_script = self._render_run_script(profile, extra_env or {})
             run_path = tmp_dir / "run"
-            run_path.write_text(run_script)
+            run_path.write_text(run_script, encoding="utf-8")
             run_path.chmod(0o755)
 
             finish_path = tmp_dir / "finish"
-            finish_path.write_text(self._render_finish_script())
+            finish_path.write_text(self._render_finish_script(), encoding="utf-8")
             finish_path.chmod(0o755)
 
             # Persistent log rotation (OQ8-C).
             log_subdir = tmp_dir / "log"
             log_subdir.mkdir()
             log_run = log_subdir / "run"
-            log_run.write_text(self._render_log_run(profile))
+            log_run.write_text(self._render_log_run(profile), encoding="utf-8")
             log_run.chmod(0o755)
 
             # Pre-create the supervise/ skeleton with hermes ownership

hermes_cli/uninstall.py

@@ -87,7 +87,7 @@ def remove_path_from_shell_configs():
                 new_content = new_content.replace('\n\n\n', '\n\n')
 
             if new_content != original_content:
-                config_path.write_text(new_content)
+                config_path.write_text(new_content, encoding="utf-8")
                 removed_from.append(config_path)
 
         except Exception as e:

tools/skills_hub.py

@@ -1051,7 +1051,7 @@ def _write_cache(self, key: str, data: list) -> None:
         index_cache_dir.mkdir(parents=True, exist_ok=True)
         cache_file = index_cache_dir / f"{key}.json"
         try:
-            cache_file.write_text(json.dumps(data, ensure_ascii=False))
+            cache_file.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
         except OSError as e:
             logger.debug("Could not write cache: %s", e)
 
@@ -3245,12 +3245,12 @@ def _write_index_cache(key: str, data: Any) -> None:
     ignore_file = _hub_dir() / ".ignore"
     if not ignore_file.exists():
         try:
-            ignore_file.write_text("# Exclude hub internals from search tools\n*\n")
+            ignore_file.write_text("# Exclude hub internals from search tools\n*\n", encoding="utf-8")
         except OSError:
             pass
     cache_file = index_cache_dir / f"{key}.json"
     try:
-        cache_file.write_text(json.dumps(data, ensure_ascii=False, default=str))
+        cache_file.write_text(json.dumps(data, ensure_ascii=False, default=str), encoding="utf-8")
     except OSError as e:
         logger.debug("Could not write cache: %s", e)
 
@@ -3290,7 +3290,7 @@ def load(self) -> dict:
 
     def save(self, data: dict) -> None:
         self.path.parent.mkdir(parents=True, exist_ok=True)
-        self.path.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n")
+        self.path.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
 
     def record_install(
         self,
@@ -3363,7 +3363,7 @@ def load(self) -> List[dict]:
 
     def save(self, taps: List[dict]) -> None:
         self.path.parent.mkdir(parents=True, exist_ok=True)
-        self.path.write_text(json.dumps({"taps": taps}, indent=2) + "\n")
+        self.path.write_text(json.dumps({"taps": taps}, indent=2) + "\n", encoding="utf-8")
 
     def add(self, repo: str, path: str = "skills/") -> bool:
         """Add a tap. Returns False if already exists."""
@@ -3422,11 +3422,11 @@ def ensure_hub_dirs() -> None:
     _quarantine_dir().mkdir(exist_ok=True)
     _index_cache_dir().mkdir(exist_ok=True)
     if not lock_file.exists():
-        lock_file.write_text('{"version": 1, "installed": {}}\n')
+        lock_file.write_text('{"version": 1, "installed": {}}\n', encoding="utf-8")
     if not audit_log.exists():
         audit_log.touch()
     if not taps_file.exists():
-        taps_file.write_text('{"taps": []}\n')
+        taps_file.write_text('{"taps": []}\n', encoding="utf-8")
 
 
 def quarantine_bundle(bundle: SkillBundle) -> Path:
@@ -3696,7 +3696,7 @@ def _load_hermes_index() -> Optional[dict]:
     # Cache locally
     try:
         hermes_index_cache_file.parent.mkdir(parents=True, exist_ok=True)
-        hermes_index_cache_file.write_text(json.dumps(data))
+        hermes_index_cache_file.write_text(json.dumps(data), encoding="utf-8")
     except OSError:
         pass
 

tools/web_tools.py

@@ -308,7 +308,7 @@ def _web_requires_env() -> list[str]:
 DEFAULT_EXTRACT_CHAR_LIMIT = 15000
 
 # Hard ceiling on the full-text file written to cache/web. The truncate-store
-# path otherwise calls path.write_text(content) with no upper bound, so a
+# path otherwise calls path.write_text(content, encoding="utf-8") with no upper bound, so a
 # multi-MB page (some backends return very large markdown) writes unbounded
 # bytes to disk on every extract. Cap the stored copy; the model only ever
 # sees char_limit anyway, and a 2MB page is already far more than any single

tools/xai_http.py

@@ -215,7 +215,7 @@ def maybe_mark_xai_storage_notice_seen(section_name: str) -> Optional[str]:
         marker = marker_dir / f"{section_name}_xai_storage_notice_seen"
         if marker.exists():
             return None
-        marker.write_text(datetime.datetime.now(datetime.UTC).isoformat() + "\n")
+        marker.write_text(datetime.datetime.now(datetime.UTC).isoformat() + "\n", encoding="utf-8")
         return notice
     except Exception:
         return notice