diff --git a/hermes_cli/process_identity.py b/hermes_cli/process_identity.py index 731ba061a6..d2ad83aad6 100644 --- a/hermes_cli/process_identity.py +++ b/hermes_cli/process_identity.py @@ -267,7 +267,9 @@ def _append_entry(entry: LedgerEntry) -> bool: pruned.append(asdict(entry)) try: path.parent.mkdir(parents=True, exist_ok=True) - atomic_json_write(path, pruned, mode=0o600) + # argv may carry surrogate-escaped bytes (non-UTF-8 paths); ensure_ascii keeps the + # utf-8 text handle from raising UnicodeEncodeError (a ValueError, not an OSError). + atomic_json_write(path, pruned, mode=0o600, ensure_ascii=True) return True except OSError: logger.debug("spawn ledger write failed", exc_info=True) diff --git a/tests/hermes_cli/test_process_identity.py b/tests/hermes_cli/test_process_identity.py index 5f9a85dfbe..dc627ab9ee 100644 --- a/tests/hermes_cli/test_process_identity.py +++ b/tests/hermes_cli/test_process_identity.py @@ -134,6 +134,19 @@ def test_register_self_writes_and_prunes_dead(tmp_path): assert me["create_time"] == pytest.approx(50.0, abs=0.01) +def test_register_self_survives_non_utf8_argv(tmp_path): + ledger = tmp_path / "spawn-ledger.json" + fake = _fake_psutil({999: 50.0}) + bad_argv = ["hermes", "serve", os.fsdecode(b"/tmp/project-\xff")] # surrogate-escaped path + with patch.dict(sys.modules, {"psutil": fake}), \ + patch.object(pi, "_ledger_path", return_value=ledger), \ + patch.object(pi.os, "getpid", return_value=999), \ + patch.object(sys, "argv", bad_argv): + assert pi.register_self("serve", project_root=Path("/x/install")) is True + me = next(e for e in json.loads(ledger.read_text(encoding="utf-8")) if e["pid"] == 999) + assert me["argv"] == " ".join(bad_argv) + + @pytest.mark.skipif(os.name == "nt", reason="POSIX mode bits are platform-specific") def test_register_self_writes_ledger_with_0600(tmp_path): ledger = tmp_path / "spawn-ledger.json" diff --git a/utils.py b/utils.py index ed9e35f37a..bbc2da7d81 100644 --- a/utils.py +++ b/utils.py @@ -217,10 +217,17 @@ def atomic_write_text(path: Union[str, Path], content: str, *, encoding: str = " mode=_mode_for_write(path, create_mode, preserve=preserve_mode), preserve_owner=preserve_mode) -def atomic_json_write(path: Union[str, Path], data: Any, *, indent: int = 2, mode: int | None = None, **dump_kwargs: Any) -> None: - """Write JSON to *path* atomically (temp file + fsync + replace).""" +def atomic_json_write( + path: Union[str, Path], data: Any, *, indent: int = 2, mode: int | None = None, + ensure_ascii: bool = False, **dump_kwargs: Any, +) -> None: + """Write JSON to *path* atomically (temp file + fsync + replace). + + ``ensure_ascii=True`` lets callers persist surrogate-escaped strings (non-UTF-8 argv/paths) + that a utf-8 text handle would otherwise reject with ``UnicodeEncodeError``. + """ path = Path(path) - _atomic_write(path, lambda f: json.dump(data, f, indent=indent, ensure_ascii=False, **dump_kwargs), + _atomic_write(path, lambda f: json.dump(data, f, indent=indent, ensure_ascii=ensure_ascii, **dump_kwargs), prefix=f".{path.stem}_", mode=mode if mode is not None else _preserve_file_mode(path))