mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-25 17:18:11 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

fix(cli): add explicit encoding to read_text/write_text calls

Browse source 
Path.read_text() and Path.write_text() without explicit encoding
default to the system locale encoding. On Windows this is typically
cp1252, which causes UnicodeDecodeError for UTF-8 content (JSON
configs, user data, service scripts).

Add encoding="utf-8" to all read_text() and write_text() calls
across 8 CLI files, matching the pattern established in PR #50534
(security_audit_startup.py) and ruff rule PLW1514.

Fixed files:
- main.py: 4 read_text calls
- auth.py: 3 read_text calls
- banner.py: 1 read_text + 1 write_text
- service_manager.py: 1 read_text + 4 write_text
- container_boot.py: 1 read_text + 4 write_text
- doctor.py: 3 read_text calls
- uninstall.py: 2 read_text calls
- gateway.py: 1 write_text call
This commit is contained in:
annguyenNous 2026-06-22 13:40:21 +07:00 • committed by Teknium
parent 9f6b2a64e0
commit efaba061fb
8 changed files with 29 additions and 28 deletions
Download patch file Download diff file Expand all files Collapse all files

6
hermes_cli/auth.py
View file

@ -1123,7 +1123,7 @@ def _load_auth_store(auth_file: Optional[Path] = None) -> Dict[str, Any]:
return {"version": AUTH_STORE_VERSION, "providers": {}}
try:
raw = json.loads(auth_file.read_text())
raw = json.loads(auth_file.read_text(encoding="utf-8"))
except Exception as exc:
corrupt_path = auth_file.with_suffix(".json.corrupt")
try:
@ -3826,7 +3826,7 @@ def _import_codex_cli_tokens() -> Optional[Dict[str, str]]:
if not auth_path.is_file():
return None
try:
payload = json.loads(auth_path.read_text())
payload = json.loads(auth_path.read_text(encoding="utf-8"))
tokens = payload.get("tokens")
if not isinstance(tokens, dict):
return None
@ -5261,7 +5261,7 @@ def _read_shared_nous_state() -> Optional[Dict[str, Any]]:
if not path.is_file():
return None
try:
payload = json.loads(path.read_text())
payload = json.loads(path.read_text(encoding="utf-8"))
except (OSError, ValueError) as exc:
logger.debug("Shared Nous auth store at %s is unreadable: %s", path, exc)
return None

5
hermes_cli/banner.py
View file

@ -292,7 +292,7 @@ def check_for_updates() -> Optional[int]:
now = time.time()
try:
if cache_file.exists():
cached = json.loads(cache_file.read_text())
cached = json.loads(cache_file.read_text(encoding="utf-8"))
if (
now - cached.get("ts", 0) < _UPDATE_CHECK_CACHE_SECONDS
and cached.get("rev") == embedded_rev
@ -321,7 +321,8 @@ def check_for_updates() -> Optional[int]:
try:
cache_file.write_text(
json.dumps({"ts": now, "behind": behind, "rev": embedded_rev, "ver": VERSION})
json.dumps({"ts": now, "behind": behind, "rev": embedded_rev, "ver": VERSION}),
encoding="utf-8",
)
except Exception:
pass

10
hermes_cli/container_boot.py
View file

@ -236,7 +236,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"
@ -382,7 +382,7 @@ def _read_desired_state(profile_dir: Path) -> str | None:
if not state_file.exists():
return None
try:
data = json.loads(state_file.read_text())
data = json.loads(state_file.read_text(encoding="utf-8"))
desired_state = data.get("desired_state")
if desired_state is not None:
return desired_state
@ -450,7 +450,7 @@ 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
@ -458,7 +458,7 @@ def _register_service(scandir: Path, profile: str, *, start: bool) -> None:
# 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"
@ -469,7 +469,7 @@ def _register_service(scandir: Path, profile: str, *, start: bool) -> None:
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

4
hermes_cli/doctor.py
View file

@ -2443,7 +2443,7 @@ def run_doctor(args):
if lock_file.exists():
try:
import json
lock_data = json.loads(lock_file.read_text())
lock_data = json.loads(lock_file.read_text(encoding="utf-8"))
count = len(lock_data.get("installed", {}))
check_ok(f"Lock file OK ({count} hub-installed skill(s))")
except Exception:
@ -2609,7 +2609,7 @@ def run_doctor(args):
if not wrapper.is_file():
continue
try:
content = wrapper.read_text()
content = wrapper.read_text(encoding="utf-8")
if "hermes -p" in content:
_m = _re.search(r"hermes -p (\S+)", content)
if _m and not profile_exists(_m.group(1)):

10
hermes_cli/gateway.py
View file

@ -3066,7 +3066,7 @@ def refresh_systemd_unit_if_needed(system: bool = False) -> bool:
if _refuse_temp_home_service_write(new_unit, "systemd unit"):
return False
unit_path.write_text(new_unit, encoding="utf-8")
unit_path.write_text(new_unit, encoding="utf-8", encoding="utf-8")
_run_systemctl(["daemon-reload"], system=system, check=True, timeout=30)
print(
f"↻ Updated gateway {_service_scope_label(system)} service definition to match the current Hermes install"
@ -3248,7 +3248,7 @@ def systemd_install(
if _refuse_temp_home_service_write(new_unit, "systemd unit"):
return
print(f"Installing {_service_scope_label(system)} systemd service to: {unit_path}")
unit_path.write_text(new_unit, encoding="utf-8")
unit_path.write_text(new_unit, encoding="utf-8", encoding="utf-8")
_run_systemctl(["daemon-reload"], system=system, check=True, timeout=30)
if enable_on_startup:
@ -4082,7 +4082,7 @@ def refresh_launchd_plist_if_needed() -> bool:
if _refuse_temp_home_service_write(new_plist, "launchd plist"):
return False
plist_path.write_text(new_plist, encoding="utf-8")
plist_path.write_text(new_plist, encoding="utf-8", encoding="utf-8")
label = get_launchd_label()
domain = _launchd_domain()
target = f"{domain}/{label}"
@ -4251,7 +4251,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(
@ -4301,7 +4301,7 @@ def launchd_start():
sys.exit(1)
print("↻ launchd plist missing; regenerating service definition")
plist_path.parent.mkdir(parents=True, exist_ok=True)
plist_path.write_text(new_plist, encoding="utf-8")
plist_path.write_text(new_plist, encoding="utf-8", encoding="utf-8")
try:
_launchctl_bootstrap(_launchd_domain(), plist_path, label, timeout=30)
subprocess.run(

8
hermes_cli/main.py
View file

@ -639,7 +639,7 @@ def _apply_profile_override() -> None:
active_path = get_default_hermes_root() / "active_profile"
if active_path.exists():
name = active_path.read_text().strip()
name = active_path.read_text(encoding="utf-8").strip()
if name and name != "default":
profile_name = name
consume = 0 # don't strip anything from argv
@ -1021,7 +1021,7 @@ def _has_any_provider_configured() -> bool:
try:
import json
auth = json.loads(auth_file.read_text())
auth = json.loads(auth_file.read_text(encoding="utf-8"))
active = auth.get("active_provider")
if active:
status = get_auth_status(active)
@ -4814,7 +4814,7 @@ def _gateway_prompt(prompt_text: str, default: str = "", timeout: float = 300.0)
while _time.monotonic() < deadline:
if response_path.exists():
try:
answer = response_path.read_text().strip()
answer = response_path.read_text(encoding="utf-8").strip()
response_path.unlink(missing_ok=True)
prompt_path.unlink(missing_ok=True)
return answer if answer else default
@ -13357,7 +13357,7 @@ def _render_distribution_plan(plan) -> None:
env_path = plan.target_dir / ".env"
if env_path.is_file():
try:
for raw in env_path.read_text().splitlines():
for raw in env_path.read_text(encoding="utf-8").splitlines():
line = raw.strip()
if not line or line.startswith("#"):
continue

10
hermes_cli/service_manager.py
View file

@ -376,7 +376,7 @@ def _write_gateway_desired_state(name: str, desired_state: str) -> None:
if not profile_dir.exists():
return
try:
data = json.loads(state_file.read_text()) if state_file.exists() else {}
data = json.loads(state_file.read_text(encoding="utf-8")) if state_file.exists() else {}
if not isinstance(data, dict):
data = {}
except (OSError, json.JSONDecodeError):
@ -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,11 +987,11 @@ class S6ServiceManager:
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"
@ -1002,7 +1002,7 @@ class S6ServiceManager:
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

4
hermes_cli/uninstall.py
View file

@ -57,7 +57,7 @@ def remove_path_from_shell_configs():
for config_path in configs:
try:
content = config_path.read_text()
content = config_path.read_text(encoding="utf-8")
original_content = content
# Remove lines containing hermes-agent or hermes PATH entries
@ -108,7 +108,7 @@ def remove_wrapper_script():
if wrapper.exists():
try:
# Check if it's our wrapper (contains hermes_cli reference)
content = wrapper.read_text()
content = wrapper.read_text(encoding="utf-8")
if 'hermes_cli' in content or 'hermes-agent' in content:
wrapper.unlink()
removed.append(wrapper)