diff --git a/headroom/cli/mcp.py b/headroom/cli/mcp.py index 2f83aea96..b5aede3a5 100644 --- a/headroom/cli/mcp.py +++ b/headroom/cli/mcp.py @@ -35,7 +35,7 @@ def load_mcp_config() -> dict[str, Any]: """Load existing MCP config or return empty structure.""" if MCP_CONFIG_PATH.exists(): try: - with open(MCP_CONFIG_PATH) as f: + with open(MCP_CONFIG_PATH, encoding="utf-8") as f: result: dict[str, Any] = json.load(f) return result except (json.JSONDecodeError, OSError): @@ -46,7 +46,7 @@ def load_mcp_config() -> dict[str, Any]: def save_mcp_config(config: dict) -> None: """Save MCP config, creating directory if needed.""" CLAUDE_CONFIG_DIR.mkdir(parents=True, exist_ok=True) - with open(MCP_CONFIG_PATH, "w") as f: + with open(MCP_CONFIG_PATH, "w", encoding="utf-8") as f: json.dump(config, f, indent=2) f.write("\n") # Trailing newline diff --git a/headroom/cli/memory.py b/headroom/cli/memory.py index c02798c18..91e717c94 100644 --- a/headroom/cli/memory.py +++ b/headroom/cli/memory.py @@ -12,6 +12,7 @@ from typing import Any import click +from .. import fsutil from ..memory.adapters.sqlite import SQLiteMemoryStore from ..memory.models import Memory, ScopeLevel from ..memory.ports import MemoryFilter @@ -874,7 +875,7 @@ def export_memories(ctx: click.Context, db_path: str, output: str | None) -> Non if output: output_path = Path(output) - output_path.write_text(json_output) + fsutil.write_text(output_path, json_output) print_success(f"Exported {len(memories)} memory(ies) to {output_path}") else: click.echo(json_output) @@ -904,7 +905,7 @@ def import_memories(ctx: click.Context, db_path: str, file: str, force: bool) -> try: # Read and parse file - content = file_path.read_text() + content = fsutil.read_text(file_path) memories_data = json.loads(content) if not isinstance(memories_data, list): diff --git a/headroom/cli/wrap.py b/headroom/cli/wrap.py index 51b280faf..b7ff1cd36 100644 --- a/headroom/cli/wrap.py +++ b/headroom/cli/wrap.py @@ -43,6 +43,7 @@ if sys.platform == "win32" and hasattr(sys.stdout, "buffer"): import click +from headroom import fsutil from headroom._version import __version__ as _HEADROOM_VERSION from headroom.agent_savings import ( apply_agent_savings_env_defaults, @@ -128,19 +129,18 @@ from .main import main def _read_text(path: Path) -> str: - """Read a text file with explicit UTF-8 encoding.""" - return path.read_text(encoding="utf-8") + """Read a text file as UTF-8, falling back to the system locale encoding.""" + return fsutil.read_text(path) def _write_text(path: Path, content: str) -> None: - """Write a text file with explicit UTF-8 encoding.""" - path.write_text(content, encoding="utf-8") + """Write a text file as UTF-8 without translating line endings (preserves CRLF).""" + fsutil.write_text(path, content) def _append_text(path: Path, content: str) -> None: - """Append to a text file with explicit UTF-8 encoding.""" - with open(path, "a", encoding="utf-8") as f: - f.write(content) + """Append to a text file as UTF-8 without translating line endings.""" + fsutil.append_text(path, content) _CONTEXT_TOOL_ENV = "HEADROOM_CONTEXT_TOOL" @@ -556,7 +556,7 @@ def _patch_rtk_hook_absolute_path(rtk_path: Path, hook_script_path: Path | None if not hook_script_path.exists(): return False - original = hook_script_path.read_text() + original = _read_text(hook_script_path) # Quote the absolute path safely for POSIX shells. This matters because # paths containing spaces or other shell-special characters (e.g. @@ -576,7 +576,7 @@ def _patch_rtk_hook_absolute_path(rtk_path: Path, hook_script_path: Path | None ) if count and patched != original: - hook_script_path.write_text(patched) + _write_text(hook_script_path, patched) return True return False diff --git a/headroom/fsutil.py b/headroom/fsutil.py new file mode 100644 index 000000000..b88fb4d99 --- /dev/null +++ b/headroom/fsutil.py @@ -0,0 +1,81 @@ +"""Encoding- and newline-safe text file I/O. + +``Path.read_text()`` / ``Path.write_text()`` and the builtin ``open()`` default +to the *system locale* encoding and, in text mode, translate ``\\n`` to +``os.linesep`` on write. On non-UTF-8 Windows locales (e.g. GBK / cp936 on +zh-CN) this corrupts config files two ways: + +1. Reading a UTF-8 file as GBK — or a GBK file as UTF-8 — raises + ``UnicodeDecodeError``. +2. Writing re-translates ``\\n`` to ``\\r\\n``; content that already has + ``\\r\\n`` becomes ``\\r\\r\\n``, which TOML parsers reject with + "carriage return must be followed by newline". + +These helpers always use UTF-8, fall back to the locale encoding when a file +predates the fix (tools may have written it in the locale encoding), and write +with ``newline=""`` so existing line endings pass through unchanged. + +See issue #733. +""" + +from __future__ import annotations + +import locale +import os +from pathlib import Path + +# Sentinel so ``default=None`` can be a real return value if a caller wants it. +_RAISE = object() + + +def read_text(path: str | os.PathLike[str], *, default: object = _RAISE) -> str: + """Read text, preferring UTF-8 and falling back to the locale encoding. + + Decoding order: UTF-8 (strict) → locale preferred encoding (strict) → + UTF-8 with ``errors="replace"`` (never raises on content). If the file + cannot be opened (missing/unreadable) and ``default`` is given, it is + returned; otherwise the ``OSError`` propagates. + + Line endings are normalised to ``\\n`` (universal-newline semantics, + matching the stdlib text-mode default) so callers that search or rewrite + the text work on a single ending, and a later :func:`write_text` cannot + re-double an existing ``\\r\\n``. + """ + try: + raw = Path(path).read_bytes() + except OSError: + if default is not _RAISE: + return default # type: ignore[return-value] + raise + + text: str | None = None + try: + text = raw.decode("utf-8") + except UnicodeDecodeError: + loc = locale.getpreferredencoding(False) + if loc and loc.lower().replace("-", "") != "utf8": + try: + text = raw.decode(loc) + except (UnicodeDecodeError, LookupError): + text = None + if text is None: + text = raw.decode("utf-8", errors="replace") + + return text.replace("\r\n", "\n").replace("\r", "\n") + + +def write_text(path: str | os.PathLike[str], content: str) -> None: + """Write text as UTF-8 without translating line endings. + + ``newline=""`` disables the platform ``\\n`` → ``\\r\\n`` rewrite, so the + bytes written match ``content`` exactly and existing ``\\r\\n`` endings are + never doubled. + """ + with Path(path).open("w", encoding="utf-8", newline="") as f: + f.write(content) + + +def append_text(path: str | os.PathLike[str], content: str) -> None: + """Append text as UTF-8 without translating line endings (see ``write_text``).""" + with Path(path).open("a", encoding="utf-8", newline="") as f: + f.write(content) diff --git a/headroom/install/providers.py b/headroom/install/providers.py index a976411d5..fd1fabfd8 100644 --- a/headroom/install/providers.py +++ b/headroom/install/providers.py @@ -7,6 +7,7 @@ import re import subprocess from pathlib import Path +from headroom import fsutil from headroom._subprocess import run from headroom.providers.install_registry import ( apply_provider_scope_mutations, @@ -29,7 +30,7 @@ _ENV_PATTERN = re.compile( def _merge_marker_block(file_path: Path, block: str, pattern: re.Pattern[str], marker: str) -> str: if file_path.exists(): - existing = file_path.read_text() + existing = fsutil.read_text(file_path) if marker in existing: return pattern.sub(block, existing) return existing.rstrip() + "\n\n" + block + "\n" @@ -66,7 +67,7 @@ def _apply_unix_env_scope(manifest: DeploymentManifest) -> list[ManagedMutation] for path in targets: path.parent.mkdir(parents=True, exist_ok=True) merged = _merge_marker_block(path, block, _ENV_PATTERN, _ENV_MARKER_START) - path.write_text(merged) + fsutil.write_text(path, merged) mutations.append(ManagedMutation(target="env", kind="shell-block", path=str(path))) return mutations @@ -78,10 +79,10 @@ def _remove_unix_env_scope(mutations: list[ManagedMutation]) -> None: path = Path(mutation.path) if not path.exists(): continue - content = path.read_text() + content = fsutil.read_text(path) if _ENV_MARKER_START not in content: continue - path.write_text(_ENV_PATTERN.sub("", content).strip() + "\n") + fsutil.write_text(path, _ENV_PATTERN.sub("", content).strip() + "\n") def _apply_windows_env_scope(manifest: DeploymentManifest) -> list[ManagedMutation]: diff --git a/headroom/mcp_registry/claude.py b/headroom/mcp_registry/claude.py index 939a6c141..c537e679b 100644 --- a/headroom/mcp_registry/claude.py +++ b/headroom/mcp_registry/claude.py @@ -213,7 +213,7 @@ def _read_json(path: Path) -> dict[str, Any]: if not path.exists(): return {} try: - with open(path) as f: + with open(path, encoding="utf-8") as f: data = json.load(f) except (OSError, json.JSONDecodeError): return {} @@ -224,7 +224,7 @@ def _read_json(path: Path) -> dict[str, Any]: def _write_json(path: Path, data: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) - with open(path, "w") as f: + with open(path, "w", encoding="utf-8") as f: json.dump(data, f, indent=2) f.write("\n") diff --git a/headroom/mcp_registry/codex.py b/headroom/mcp_registry/codex.py index e4b675d64..190ad96df 100644 --- a/headroom/mcp_registry/codex.py +++ b/headroom/mcp_registry/codex.py @@ -17,6 +17,8 @@ import sys from pathlib import Path from typing import Any +from headroom import fsutil + from .base import MCPRegistrar, RegisterResult, RegisterStatus, ServerSpec if sys.version_info >= (3, 11): @@ -132,7 +134,7 @@ class CodexRegistrar(MCPRegistrar): else: new_content = (before or after).rstrip("\n") + ("\n" if (before or after) else "") try: - self._config_file.write_text(new_content) + fsutil.write_text(self._config_file, new_content) except OSError: return False return True @@ -145,17 +147,16 @@ class CodexRegistrar(MCPRegistrar): if not self._config_file.exists(): return {} try: - with open(self._config_file, "rb") as f: - data = tomllib.load(f) + # Read via fsutil (UTF-8 with locale fallback) so a config that a + # tool wrote in the system locale (e.g. GBK) still parses instead + # of failing tomllib's UTF-8 requirement. See #733. + data = tomllib.loads(fsutil.read_text(self._config_file)) except (tomllib.TOMLDecodeError, OSError): return {} return data if isinstance(data, dict) else {} def _read_text(self) -> str: - try: - return self._config_file.read_text() - except OSError: - return "" + return fsutil.read_text(self._config_file, default="") def _write_block(self, spec: ServerSpec) -> RegisterResult: block = _render_block(spec) @@ -178,7 +179,7 @@ class CodexRegistrar(MCPRegistrar): content = content.rstrip("\n") + "\n\n" + block + "\n" else: content = block + "\n" - self._config_file.write_text(content) + fsutil.write_text(self._config_file, content) except OSError as exc: return RegisterResult( RegisterStatus.FAILED, f"could not write {self._config_file}: {exc}" diff --git a/headroom/mcp_registry/opencode.py b/headroom/mcp_registry/opencode.py index 93dda3e6d..6ab6cca91 100644 --- a/headroom/mcp_registry/opencode.py +++ b/headroom/mcp_registry/opencode.py @@ -39,7 +39,7 @@ def _read_json(path: Path) -> dict[str, Any]: if not path.exists(): return {} try: - with open(path) as f: + with open(path, encoding="utf-8") as f: data = json.load(f) except (OSError, json.JSONDecodeError): return {} @@ -50,7 +50,7 @@ def _read_json(path: Path) -> dict[str, Any]: def _write_json(path: Path, data: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) - with open(path, "w") as f: + with open(path, "w", encoding="utf-8") as f: json.dump(data, f, indent=2) f.write("\n") diff --git a/headroom/providers/anthropic.py b/headroom/providers/anthropic.py index 93d3d1bd6..e1ff0126c 100644 --- a/headroom/providers/anthropic.py +++ b/headroom/providers/anthropic.py @@ -199,7 +199,7 @@ def _load_custom_model_config() -> dict[str, Any]: try: # Check if it's a file path if os.path.isfile(env_config): - with open(env_config) as f: + with open(env_config, encoding="utf-8") as f: loaded = json.load(f) else: # Try to parse as JSON string @@ -225,7 +225,7 @@ def _load_custom_model_config() -> dict[str, Any]: config_file = legacy_models if config_file.exists(): try: - with open(config_file) as f: + with open(config_file, encoding="utf-8") as f: loaded = json.load(f) # Only load anthropic-specific config diff --git a/headroom/providers/openai.py b/headroom/providers/openai.py index b4c688b15..320fb59e6 100644 --- a/headroom/providers/openai.py +++ b/headroom/providers/openai.py @@ -157,7 +157,7 @@ def _load_custom_model_config() -> dict[str, Any]: try: # Check if it's a file path if os.path.isfile(env_config): - with open(env_config) as f: + with open(env_config, encoding="utf-8") as f: loaded = json.load(f) else: # Try to parse as JSON string @@ -184,7 +184,7 @@ def _load_custom_model_config() -> dict[str, Any]: config_file = legacy_models if config_file.exists(): try: - with open(config_file) as f: + with open(config_file, encoding="utf-8") as f: loaded = json.load(f) openai_config = loaded.get("openai", {}) diff --git a/headroom/providers/opencode/config.py b/headroom/providers/opencode/config.py index 3579fabbf..097022dcc 100644 --- a/headroom/providers/opencode/config.py +++ b/headroom/providers/opencode/config.py @@ -11,6 +11,7 @@ from typing import Any import click +from headroom import fsutil from headroom.install.paths import opencode_config_path from headroom.mcp_registry.install import DEFAULT_PROXY_URL @@ -58,7 +59,7 @@ def snapshot_opencode_config_if_unwrapped(config_file: Path, backup_file: Path) if not config_file.exists(): return try: - content = config_file.read_text() + content = fsutil.read_text(config_file) except OSError: return if _PROVIDER_MARKER_START in content or _MCP_MARKER_START in content: @@ -190,7 +191,7 @@ def inject_opencode_provider_config(port: int) -> None: snapshot_opencode_config_if_unwrapped(config_file, backup_file) if config_file.exists(): - content = config_file.read_text() + content = fsutil.read_text(config_file) data = _parse_json_loose(content) else: content = "" diff --git a/headroom/providers/opencode/install.py b/headroom/providers/opencode/install.py index 1209bc27a..38bc951f9 100644 --- a/headroom/providers/opencode/install.py +++ b/headroom/providers/opencode/install.py @@ -6,6 +6,7 @@ import json import shutil from pathlib import Path +from headroom import fsutil from headroom.install.models import ConfigScope, DeploymentManifest, ManagedMutation, ToolTarget from headroom.install.paths import opencode_config_path @@ -38,7 +39,7 @@ def apply_provider_scope(manifest: DeploymentManifest) -> ManagedMutation | None ) if config_file.exists(): - content = config_file.read_text() + content = fsutil.read_text(config_file) data = _parse_json_loose(content) else: data = {} @@ -80,7 +81,7 @@ def revert_provider_scope(mutation: ManagedMutation, manifest: DeploymentManifes pass if not path.exists(): return - content = path.read_text() + content = fsutil.read_text(path) cleaned = strip_opencode_headroom_blocks(content) if cleaned: path.write_text(cleaned + "\n", encoding="utf-8") diff --git a/tests/test_fsutil.py b/tests/test_fsutil.py new file mode 100644 index 000000000..9bf5dfb1e --- /dev/null +++ b/tests/test_fsutil.py @@ -0,0 +1,73 @@ +"""Tests for headroom.fsutil — encoding- and newline-safe text I/O (#733).""" + +from __future__ import annotations + +import pytest + +from headroom import fsutil + + +def test_write_text_does_not_double_existing_crlf(tmp_path): + """A string containing \\r\\n must be written verbatim, never as \\r\\r\\n.""" + p = tmp_path / "config.toml" + fsutil.write_text(p, 'model = "gpt-5"\r\nport = 8787\r\n') + raw = p.read_bytes() + assert b"\r\r\n" not in raw + assert raw == b'model = "gpt-5"\r\nport = 8787\r\n' + + +def test_write_text_does_not_translate_lf(tmp_path): + """\\n must stay \\n on every platform (no \\r\\n rewrite).""" + p = tmp_path / "hook.sh" + fsutil.write_text(p, "#!/bin/sh\necho hi\n") + assert p.read_bytes() == b"#!/bin/sh\necho hi\n" + + +def test_read_text_normalises_crlf(tmp_path): + """read_text returns universal-newline (\\n) text, so a round trip can't double CRLF.""" + p = tmp_path / "config.toml" + p.write_bytes(b"a = 1\r\nb = 2\r\n") + text = fsutil.read_text(p) + assert text == "a = 1\nb = 2\n" + fsutil.write_text(p, text) + assert b"\r" not in p.read_bytes() + + +def test_read_text_roundtrips_utf8_non_ascii(tmp_path): + p = tmp_path / "config.toml" + fsutil.write_text(p, 'project = "比赛/机器人"\n') + assert fsutil.read_text(p) == 'project = "比赛/机器人"\n' + + +def test_read_text_falls_back_to_locale_encoding(tmp_path, monkeypatch): + """A file a tool wrote in the locale encoding (e.g. GBK) still decodes.""" + monkeypatch.setattr(fsutil.locale, "getpreferredencoding", lambda *_: "gbk") + p = tmp_path / "config.toml" + p.write_bytes('path = "模型"\n'.encode("gbk")) # not valid UTF-8 + assert fsutil.read_text(p) == 'path = "模型"\n' + + +def test_read_text_replace_fallback_never_raises(tmp_path, monkeypatch): + """When neither UTF-8 nor the locale encoding decodes, fall back to replace.""" + monkeypatch.setattr(fsutil.locale, "getpreferredencoding", lambda *_: "ascii") + p = tmp_path / "config.toml" + p.write_bytes(b"\xff\xfe bad bytes") + # Must not raise; returns *something* decodable. + assert isinstance(fsutil.read_text(p), str) + + +def test_read_text_missing_returns_default(tmp_path): + p = tmp_path / "nope.toml" + assert fsutil.read_text(p, default="") == "" + + +def test_read_text_missing_raises_without_default(tmp_path): + with pytest.raises(OSError): + fsutil.read_text(tmp_path / "nope.toml") + + +def test_append_text_preserves_endings(tmp_path): + p = tmp_path / "AGENTS.md" + fsutil.write_text(p, "line1\n") + fsutil.append_text(p, "line2\n") + assert p.read_bytes() == b"line1\nline2\n" diff --git a/tests/test_mcp_registry/test_codex_registrar.py b/tests/test_mcp_registry/test_codex_registrar.py index 6a48b4d2b..937777610 100644 --- a/tests/test_mcp_registry/test_codex_registrar.py +++ b/tests/test_mcp_registry/test_codex_registrar.py @@ -359,3 +359,40 @@ def test_round_trip(tmp_path: Path, spec: ServerSpec) -> None: assert got.command == spec.command assert got.args == spec.args assert got.env == spec.env + + +# --------------------------------------------------------------------------- +# #733 — encoding / line-ending safety on GBK / CRLF Windows configs +# --------------------------------------------------------------------------- + + +def test_register_does_not_double_crlf(tmp_path: Path) -> None: + """A pre-existing CRLF config must not gain ``\\r\\r\\n`` after register.""" + import tomllib + + cfg = _config_path(tmp_path) + cfg.parent.mkdir(parents=True, exist_ok=True) + cfg.write_bytes(b'model = "gpt-5"\r\nworkers = 1\r\n') + + result = _make_registrar(tmp_path).register_server(_spec()) + + assert result.status == RegisterStatus.REGISTERED + raw = cfg.read_bytes() + assert b"\r\r\n" not in raw + tomllib.loads(raw.decode("utf-8")) # still valid TOML + + +def test_register_preserves_non_ascii_values(tmp_path: Path) -> None: + """A config with non-ASCII (Chinese) values survives register and stays parseable.""" + import tomllib + + cfg = _config_path(tmp_path) + cfg.parent.mkdir(parents=True, exist_ok=True) + cfg.write_text('model = "gpt-5"\nproject = "比赛/机器人"\n', encoding="utf-8") + + result = _make_registrar(tmp_path).register_server(_spec()) + + assert result.status == RegisterStatus.REGISTERED + data = tomllib.loads(cfg.read_text(encoding="utf-8")) + assert data["project"] == "比赛/机器人" + assert "headroom" in data.get("mcp_servers", {})