akadmin/headroom
1
0
Fork 0
mirror of https://github.com/headroomlabs-ai/headroom.git synced 2026-08-27 14:17:10 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix(cli): add explicit UTF-8 encoding to file I/O in wrap commands (#1126) (#1164)

Browse source 
## Description

On Windows, `Path.read_text()` and `open()` default to the system locale
encoding (cp1252, GBK, etc.) instead of UTF-8. This causes
`UnicodeDecodeError` when reading or writing instruction files that
contain multi-byte UTF-8 characters such as smart quotes or em dashes.

The RTK instructions block itself contains an em dash (U+2014, `—`), so
`_inject_rtk_instructions` crashes on any non-UTF-8 Windows locale when
writing to `.copilot_instructions`, `.clinerules`, `AGENTS.md`, or
similar hint files.

Closes #1126

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)
- [ ] New feature (non-breaking change that adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [ ] Documentation update
- [ ] Performance improvement
- [ ] Code refactoring (no functional changes)

## Changes Made

- Add `encoding="utf-8"` to all bare `read_text()`, `write_text()`, and
`open()` calls in `headroom/cli/wrap.py` that handle instruction or
config files (18 call sites)
- Update test assertions in `test_wrap_hintfile_agents.py`,
`test_wrap_copilot.py`, and `test_wrap_bridge.py` to read with
`encoding="utf-8"`
- Add `test_inject_rtk_handles_utf8_content` verifying that existing
hint files with smart quotes and em dashes survive RTK injection without
crashing

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check .`)
- [ ] Type checking passes (`mypy headroom`)
- [x] New tests added for new functionality
- [ ] Manual testing performed

### Test Output

```text
$ python -m pytest tests/test_cli/test_wrap_hintfile_agents.py tests/test_cli/test_wrap_copilot.py tests/test_cli/test_wrap_bridge.py -v
47 passed in 1.28s
```

## Real Behavior Proof

- Environment: Windows 11 China (GBK locale), Python 3.11, headroom main
(f03e77b)
- Exact command / steps: python -m pytest
tests/test_cli/test_wrap_hintfile_agents.py -v (on GBK Windows)
- Observed result: Before fix,
test_prepare_only_injects_rtk_into_hintfile fails with
UnicodeDecodeError 'gbk' at position 283 (the em dash in RTK block).
After fix, all 12 hintfile tests pass including new UTF-8 round-trip
test.
- Not tested: no manual `headroom wrap copilot` run against a real
Copilot installation

## Review Readiness

- [x] I have performed a self-review
- [x] This PR is ready for human review

## Checklist

- [x] My code follows the project's style guidelines
- [x] I have performed a self-review of my code
- [ ] I have commented my code, particularly in hard-to-understand areas
- [ ] I have made corresponding changes to the documentation
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes
- [ ] I have updated the CHANGELOG.md if applicable

## Additional Notes

This is the same class of bug reported in #733 (GBK config.toml
corruption). This PR fixes the `wrap.py` call sites; other modules
(`learn/analyzer.py`, `install/providers.py`) have the same pattern and
could benefit from the same treatment in a follow-up.

---------

Signed-off-by: Yiming Zeng <yzeng424@gmail.com>
Signed-off-by: RTCartist <wangshengb@buaa.edu.cn>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Shengbo_Wang 2026-06-27 01:07:03 +08:00 • committed by GitHub
parent fda4670ef8
commit a0cb7982e3
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 218 additions and 116 deletions
Download patch file Download diff file Expand all files Collapse all files

77
headroom/cli/wrap.py
View file

@ -124,6 +124,23 @@ from headroom.proxy.project_context import with_project_prefix as _with_project_
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")
def _write_text(path: Path, content: str) -> None:
"""Write a text file with explicit UTF-8 encoding."""
path.write_text(content, encoding="utf-8")
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)
_CONTEXT_TOOL_ENV = "HEADROOM_CONTEXT_TOOL"
_CONTEXT_TOOL_RTK = "rtk"
_CONTEXT_TOOL_LEAN_CTX = "lean-ctx"
@ -418,7 +435,7 @@ def _start_proxy(
timeout_seconds = _resolve_wrap_proxy_timeout_seconds()
log_path = _get_log_path()
stdio_log_path = _get_proxy_stdio_log_path()
stdio_log_file = open(stdio_log_path, "a") # noqa: SIM115
stdio_log_file = open(stdio_log_path, "a", encoding="utf-8") # noqa: SIM115
# Ensure proxy subprocess uses UTF-8 (Windows defaults to cp1252)
proxy_env = os.environ.copy()
@ -465,7 +482,7 @@ def _start_proxy(
stdio_log_file.close()
# Read last few lines of log for error context
try:
tail = stdio_log_path.read_text()[-500:]
tail = _read_text(stdio_log_path)[-500:]
except Exception:
tail = "(no log output)"
raise RuntimeError(f"Proxy exited with code {proc.returncode}: {tail}")
@ -579,7 +596,7 @@ def _remove_claude_rtk_hooks(settings_path: Path | None = None) -> bool:
return False
try:
payload = json.loads(path.read_text(encoding="utf-8"))
payload = json.loads(_read_text(path))
except (OSError, json.JSONDecodeError):
return False
if not isinstance(payload, dict):
@ -647,7 +664,7 @@ def _remove_claude_rtk_hooks(settings_path: Path | None = None) -> bool:
if not changed:
return False
path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
_write_text(path, json.dumps(payload, indent=2) + "\n")
return True
@ -700,7 +717,7 @@ def _write_claude_wrap_base_url(
payload: dict[str, Any] = {}
if path.exists():
try:
payload = json.loads(path.read_text(encoding="utf-8"))
payload = json.loads(_read_text(path))
except (OSError, json.JSONDecodeError):
payload = {}
if not isinstance(payload, dict):
@ -711,7 +728,7 @@ def _write_claude_wrap_base_url(
env_map[key] = proxy_url
payload["env"] = env_map
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
_write_text(path, json.dumps(payload, indent=2) + "\n")
return previous
@ -732,7 +749,7 @@ def _restore_claude_wrap_base_url(
if not path.exists():
return
try:
payload = json.loads(path.read_text(encoding="utf-8"))
payload = json.loads(_read_text(path))
except (OSError, json.JSONDecodeError):
return
if not isinstance(payload, dict):
@ -753,7 +770,7 @@ def _restore_claude_wrap_base_url(
env_map[key] = previous
payload["env"] = env_map
if payload:
path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
_write_text(path, json.dumps(payload, indent=2) + "\n")
else:
path.unlink(missing_ok=True)
@ -1362,7 +1379,7 @@ def _snapshot_codex_config_if_unwrapped(config_file: Path, backup_file: Path) ->
if not config_file.exists():
return
try:
content = config_file.read_text()
content = _read_text(config_file)
except OSError:
return
if _codex_config_has_headroom_markers(content):
@ -1535,7 +1552,7 @@ def _inject_codex_provider_config(port: int) -> None:
_snapshot_codex_config_if_unwrapped(config_file, backup_file)
if config_file.exists():
content = config_file.read_text()
content = _read_text(config_file)
# Remove any prior Headroom-managed blocks before re-injecting so
# the operation is idempotent and supports port changes.
content = _strip_codex_headroom_blocks(content)
@ -1576,7 +1593,7 @@ def _inject_codex_provider_config(port: int) -> None:
f"\n{provider_section}"
)
config_file.write_text(content)
_write_text(config_file, content)
click.echo(f" Codex config: injected Headroom provider (WS + HTTP) into {config_file}")
# Pull existing native threads into the headroom-provider menu so Codex's
# history list stays whole once it routes through Headroom. Best-effort.
@ -1608,7 +1625,7 @@ def _restore_codex_provider_config() -> tuple[str, Path]:
# Case 2: no backup, but config file exists and has markers — strip them.
if config_file.exists():
original = config_file.read_text()
original = _read_text(config_file)
if _codex_config_has_headroom_markers(original):
# Without a backup, only remove named MCP blocks when this file
# also carries wrap-owned provider markers from a full wrap.
@ -1631,7 +1648,7 @@ def _restore_codex_provider_config() -> tuple[str, Path]:
# so Codex falls back to its default config.
config_file.unlink()
return "removed", config_file
config_file.write_text(cleaned)
_write_text(config_file, cleaned)
return "cleaned", config_file
# Nothing to undo.
@ -1802,17 +1819,16 @@ def _inject_rtk_instructions(file_path: Path, verbose: bool = False) -> bool:
Returns True if instructions were written.
"""
if file_path.exists():
existing = file_path.read_text()
existing = _read_text(file_path)
if _RTK_MARKER in existing:
if verbose:
click.echo(f" rtk instructions already in {file_path.name}")
return True
# Append to existing file
with open(file_path, "a") as f:
f.write("\n\n" + RTK_INSTRUCTIONS_BLOCK)
_append_text(file_path, "\n\n" + RTK_INSTRUCTIONS_BLOCK)
else:
file_path.parent.mkdir(parents=True, exist_ok=True)
file_path.write_text(RTK_INSTRUCTIONS_BLOCK)
_write_text(file_path, RTK_INSTRUCTIONS_BLOCK)
click.echo(f" rtk instructions injected into {file_path}")
return True
@ -1823,7 +1839,7 @@ def _remove_rtk_instructions(file_path: Path) -> bool:
if not file_path.exists():
return False
content = file_path.read_text(encoding="utf-8")
content = _read_text(file_path)
end_marker = "<!-- /headroom:rtk-instructions -->"
start = content.find(_RTK_MARKER)
if start < 0:
@ -1840,7 +1856,7 @@ def _remove_rtk_instructions(file_path: Path) -> bool:
cleaned = cleaned.rstrip() + "\n"
if cleaned:
file_path.write_text(cleaned, encoding="utf-8")
_write_text(file_path, cleaned)
else:
file_path.unlink()
return True
@ -1879,7 +1895,7 @@ def _inject_memory_mcp_config(user_id: str) -> None:
_snapshot_codex_config_if_unwrapped(config_file, backup_file)
if config_file.exists():
content = config_file.read_text()
content = _read_text(config_file)
if _MEMORY_MCP_MARKER in content:
start = content.index(_MEMORY_MCP_MARKER)
end = content.index(_MEMORY_MCP_END) + len(_MEMORY_MCP_END)
@ -1889,7 +1905,7 @@ def _inject_memory_mcp_config(user_id: str) -> None:
else:
content = mcp_section
config_file.write_text(content)
_write_text(config_file, content)
click.echo(f" Memory MCP: registered in {config_file}")
except Exception as e:
click.echo(f" Warning: could not register memory MCP: {e}")
@ -1913,14 +1929,13 @@ def _inject_memory_agents_md(file_path: Path) -> bool:
)
if file_path.exists():
existing = file_path.read_text()
existing = _read_text(file_path)
if _MEMORY_AGENTS_MARKER in existing:
return True # Already injected
with open(file_path, "a") as f:
f.write("\n\n" + memory_block)
_append_text(file_path, "\n\n" + memory_block)
else:
file_path.parent.mkdir(parents=True, exist_ok=True)
file_path.write_text(memory_block)
_write_text(file_path, memory_block)
click.echo(f" Memory guidance injected into {file_path.name}")
return True
@ -2002,7 +2017,7 @@ def _inject_continue_rtk_systemmessage(config_file: Path, verbose: bool = False)
"""
if config_file.exists():
try:
content = config_file.read_text()
content = _read_text(config_file)
except OSError as exc:
click.echo(f" Warning: could not read {config_file}: {exc}")
return False
@ -2055,7 +2070,7 @@ def _inject_continue_rtk_systemmessage(config_file: Path, verbose: bool = False)
if any_changed:
config_file.parent.mkdir(parents=True, exist_ok=True)
config_file.write_text(json.dumps(data, indent=2) + "\n")
_write_text(config_file, json.dumps(data, indent=2) + "\n")
click.echo(f" rtk instructions injected into {config_file}")
elif all_ok and verbose:
# Idempotent re-run with no refusals — nothing to do.
@ -2748,7 +2763,7 @@ def _register_proxy_client(port: int) -> None:
ident = _proc_identity(os.getpid())
if ident is not None:
payload["start_src"], payload["start_time"] = ident
_client_marker_path(port).write_text(json.dumps(payload))
_write_text(_client_marker_path(port), json.dumps(payload))
except OSError:
pass
@ -2782,7 +2797,7 @@ def _marker_pid_reused(marker: Path, pid: int) -> bool:
mismatched source) returns ``False`` so a real client is never pruned.
"""
try:
rec = json.loads(marker.read_text())
rec = json.loads(_read_text(marker))
except (OSError, ValueError):
return False
src = rec.get("start_src")
@ -5296,11 +5311,11 @@ def unwrap_opencode(port: int, no_stop_proxy: bool) -> None:
f"could not restore OpenCode config from backup: {exc}"
) from exc
elif config_file.exists():
content = config_file.read_text()
content = _read_text(config_file)
if _PROVIDER_MARKER_START in content or _MCP_MARKER_START in content:
cleaned = strip_opencode_headroom_blocks(content)
if cleaned.strip():
config_file.write_text(cleaned + "\n", encoding="utf-8")
_write_text(config_file, cleaned + "\n")
click.echo(f" Removed Headroom block from {config_file}; other content preserved.")
status = "cleaned"
else:

9
tests/test_cli/test_wrap_bridge.py
View file

@ -89,8 +89,9 @@ def test_wrap_codex_prepare_only_updates_config(monkeypatch, tmp_path: Path) ->
assert result.exit_code == 0, result.output
config_file = tmp_path / ".codex" / "config.toml"
assert config_file.exists()
assert 'model_provider = "headroom"' in config_file.read_text()
assert 'base_url = "http://127.0.0.1:8787/v1"' in config_file.read_text()
content = config_file.read_text(encoding="utf-8")
assert 'model_provider = "headroom"' in content
assert 'base_url = "http://127.0.0.1:8787/v1"' in content
def test_wrap_codex_prepare_only_uses_lean_ctx_when_configured(monkeypatch, tmp_path: Path) -> None:
@ -151,7 +152,7 @@ def test_wrap_aider_prepare_only_injects_conventions(monkeypatch, tmp_path: Path
assert result.exit_code == 0, result.output
conventions = Path("CONVENTIONS.md")
assert conventions.exists()
assert "headroom:rtk-instructions" in conventions.read_text()
assert "headroom:rtk-instructions" in conventions.read_text(encoding="utf-8")
def test_wrap_cursor_prepare_only_injects_cursorrules(monkeypatch, tmp_path: Path) -> None:
@ -165,7 +166,7 @@ def test_wrap_cursor_prepare_only_injects_cursorrules(monkeypatch, tmp_path: Pat
assert result.exit_code == 0, result.output
cursorrules = Path(".cursorrules")
assert cursorrules.exists()
assert "headroom:rtk-instructions" in cursorrules.read_text()
assert "headroom:rtk-instructions" in cursorrules.read_text(encoding="utf-8")
def test_wrap_cursor_prepare_only_uses_lean_ctx_when_configured(

126
tests/test_cli/test_wrap_codex.py
View file

@ -138,23 +138,23 @@ class TestSnapshotCodexConfig:
def test_creates_backup_on_first_call(self, tmp_path: Path) -> None:
config_file = tmp_path / "config.toml"
backup_file = tmp_path / "config.toml.headroom-backup"
config_file.write_text('model = "gpt-4o"\n')
config_file.write_text('model = "gpt-4o"\n', encoding="utf-8")
wrap_mod._snapshot_codex_config_if_unwrapped(config_file, backup_file)
assert backup_file.exists()
assert backup_file.read_text() == 'model = "gpt-4o"\n'
assert backup_file.read_text(encoding="utf-8") == 'model = "gpt-4o"\n'
def test_does_not_overwrite_existing_backup(self, tmp_path: Path) -> None:
config_file = tmp_path / "config.toml"
backup_file = tmp_path / "config.toml.headroom-backup"
config_file.write_text("second-wrap content\n")
backup_file.write_text("original-pre-wrap content\n")
config_file.write_text("second-wrap content\n", encoding="utf-8")
backup_file.write_text("original-pre-wrap content\n", encoding="utf-8")
wrap_mod._snapshot_codex_config_if_unwrapped(config_file, backup_file)
# Backup must still contain the *original* pre-wrap content.
assert backup_file.read_text() == "original-pre-wrap content\n"
assert backup_file.read_text(encoding="utf-8") == "original-pre-wrap content\n"
def test_no_backup_when_config_missing(self, tmp_path: Path) -> None:
config_file = tmp_path / "config.toml"
@ -170,7 +170,8 @@ class TestSnapshotCodexConfig:
config_file.write_text(
f"{wrap_mod._CODEX_TOP_LEVEL_MARKER}\n"
'model_provider = "headroom"\n'
f"{wrap_mod._CODEX_END_MARKER}\n"
f"{wrap_mod._CODEX_END_MARKER}\n",
encoding="utf-8",
)
wrap_mod._snapshot_codex_config_if_unwrapped(config_file, backup_file)
@ -249,7 +250,7 @@ class TestInjectAndRestoreRoundTrip:
wrap_mod._inject_codex_provider_config(8787)
assert config_file.exists()
assert 'model_provider = "headroom"' in config_file.read_text()
assert 'model_provider = "headroom"' in config_file.read_text(encoding="utf-8")
status, _ = wrap_mod._restore_codex_provider_config()
# No prior config existed → the injected file is fully removed.
@ -267,7 +268,7 @@ class TestInjectAndRestoreRoundTrip:
wrap_mod._inject_codex_provider_config(8787)
assert config_file.exists()
assert 'model_provider = "headroom"' in config_file.read_text()
assert 'model_provider = "headroom"' in config_file.read_text(encoding="utf-8")
assert not (tmp_path / ".codex" / "config.toml").exists()
status, _ = wrap_mod._restore_codex_provider_config()
@ -288,16 +289,16 @@ class TestInjectAndRestoreRoundTrip:
'name = "OpenAI"\n'
'base_url = "https://api.openai.com/v1"\n'
)
config_file.write_text(original)
config_file.write_text(original, encoding="utf-8")
wrap_mod._inject_codex_provider_config(8787)
wrapped = config_file.read_text()
wrapped = config_file.read_text(encoding="utf-8")
assert 'model_provider = "headroom"' in wrapped
assert "[model_providers.headroom]" in wrapped
status, _ = wrap_mod._restore_codex_provider_config()
assert status == "restored"
assert config_file.read_text() == original
assert config_file.read_text(encoding="utf-8") == original
assert not (config_dir / "config.toml.headroom-backup").exists()
def test_wrap_is_idempotent(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
@ -306,13 +307,13 @@ class TestInjectAndRestoreRoundTrip:
config_dir.mkdir()
config_file = config_dir / "config.toml"
original = '[profiles.default]\nmodel = "gpt-4o"\n'
config_file.write_text(original)
config_file.write_text(original, encoding="utf-8")
wrap_mod._inject_codex_provider_config(8787)
wrap_mod._inject_codex_provider_config(8787)
wrap_mod._inject_codex_provider_config(9999) # port change
content = config_file.read_text()
content = config_file.read_text(encoding="utf-8")
# Exactly two Headroom blocks — a top-level-key block and the
# provider-table block. Re-wrapping must not duplicate them.
assert content.count(wrap_mod._CODEX_TOP_LEVEL_MARKER) == 2
@ -327,7 +328,7 @@ class TestInjectAndRestoreRoundTrip:
status, _ = wrap_mod._restore_codex_provider_config()
assert status == "restored"
assert config_file.read_text() == original
assert config_file.read_text(encoding="utf-8") == original
def test_unwrap_is_noop_when_never_wrapped(
self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
@ -351,12 +352,13 @@ class TestInjectAndRestoreRoundTrip:
'model_provider = "headroom"\n\n'
"[model_providers.headroom]\n"
'base_url = "http://127.0.0.1:8787/v1"\n'
f"{wrap_mod._CODEX_END_MARKER}\n"
f"{wrap_mod._CODEX_END_MARKER}\n",
encoding="utf-8",
)
status, _ = wrap_mod._restore_codex_provider_config()
assert status == "cleaned"
cleaned = config_file.read_text()
cleaned = config_file.read_text(encoding="utf-8")
assert wrap_mod._CODEX_TOP_LEVEL_MARKER not in cleaned
assert wrap_mod._CODEX_END_MARKER not in cleaned
assert 'model_provider = "headroom"' not in cleaned
@ -381,13 +383,14 @@ class TestInjectAndRestoreRoundTrip:
f"{wrap_mod._MEMORY_MCP_MARKER}\n"
"[mcp_servers.headroom_memory]\n"
'command = "python"\n'
f"{wrap_mod._MEMORY_MCP_END}\n"
f"{wrap_mod._MEMORY_MCP_END}\n",
encoding="utf-8",
)
status, _ = wrap_mod._restore_codex_provider_config()
assert status == "cleaned"
cleaned = config_file.read_text()
cleaned = config_file.read_text(encoding="utf-8")
assert 'model = "gpt-4o"' in cleaned
assert "headroom" not in cleaned
@ -446,13 +449,13 @@ class TestInjectAndRestoreRoundTrip:
config_dir.mkdir()
config_file = config_dir / "config.toml"
malformed = 'this is not valid toml ][ "" \x00\n'
config_file.write_text(malformed)
config_file.write_text(malformed, encoding="utf-8")
wrap_mod._inject_codex_provider_config(8787)
status, _ = wrap_mod._restore_codex_provider_config()
assert status == "restored"
assert config_file.read_text() == malformed
assert config_file.read_text(encoding="utf-8") == malformed
# ---------------------------------------------------------------------------
@ -533,7 +536,7 @@ class TestSubscriptionRouting:
wrap_mod._inject_codex_provider_config(8787)
content = (tmp_path / ".codex" / "config.toml").read_text()
content = (tmp_path / ".codex" / "config.toml").read_text(encoding="utf-8")
assert 'openai_base_url = "http://127.0.0.1:8787/v1"' in content
def test_inject_emits_requires_openai_auth_for_chatgpt(
@ -546,7 +549,9 @@ class TestSubscriptionRouting:
wrap_mod._inject_codex_provider_config(8787)
assert "requires_openai_auth = true" in (config_dir / "config.toml").read_text()
assert "requires_openai_auth = true" in (config_dir / "config.toml").read_text(
encoding="utf-8"
)
def test_inject_omits_requires_openai_auth_for_api_key(
self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
@ -558,7 +563,9 @@ class TestSubscriptionRouting:
wrap_mod._inject_codex_provider_config(8787)
assert "requires_openai_auth" not in (config_dir / "config.toml").read_text()
assert "requires_openai_auth" not in (config_dir / "config.toml").read_text(
encoding="utf-8"
)
def test_openai_base_url_port_updates_on_rewrap(
self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
@ -568,7 +575,7 @@ class TestSubscriptionRouting:
wrap_mod._inject_codex_provider_config(8787)
wrap_mod._inject_codex_provider_config(9999)
content = (tmp_path / ".codex" / "config.toml").read_text()
content = (tmp_path / ".codex" / "config.toml").read_text(encoding="utf-8")
assert 'openai_base_url = "http://127.0.0.1:9999/v1"' in content
assert 'openai_base_url = "http://127.0.0.1:8787/v1"' not in content
@ -580,13 +587,15 @@ class TestSubscriptionRouting:
config_dir.mkdir()
config_file = config_dir / "config.toml"
original = '[profiles.default]\nmodel = "gpt-4o"\n'
config_file.write_text(original)
config_file.write_text(original, encoding="utf-8")
wrap_mod._inject_codex_provider_config(8787)
assert 'openai_base_url = "http://127.0.0.1:8787/v1"' in config_file.read_text()
assert 'openai_base_url = "http://127.0.0.1:8787/v1"' in config_file.read_text(
encoding="utf-8"
)
wrap_mod._restore_codex_provider_config()
assert config_file.read_text() == original
assert config_file.read_text(encoding="utf-8") == original
def test_strip_cleans_orphaned_openai_base_url(self) -> None:
"""Safety net: orphaned openai_base_url lines are cleaned up."""
@ -611,7 +620,7 @@ class TestSubscriptionRouting:
wrap_mod._inject_codex_provider_config(8787)
content = (tmp_path / ".codex" / "config.toml").read_text()
content = (tmp_path / ".codex" / "config.toml").read_text(encoding="utf-8")
assert "env_key" not in content
@ -643,12 +652,13 @@ class TestInjectAvoidsDuplicateTopLevelKeys:
"[model_providers.ccswitch]\n"
'name = "OpenAI"\n'
'base_url = "http://llm-gateway-proxy/v1"\n'
'wire_api = "responses"\n'
'wire_api = "responses"\n',
encoding="utf-8",
)
wrap_mod._inject_codex_provider_config(8787)
content = config_file.read_text()
content = config_file.read_text(encoding="utf-8")
# The wrapped file must be TOML-parseable — duplicate keys were
# the failure mode the user reported.
tomllib.loads(content)
@ -674,12 +684,13 @@ class TestInjectAvoidsDuplicateTopLevelKeys:
config_dir.mkdir()
config_file = config_dir / "config.toml"
config_file.write_text(
'model_provider = "ccswitch"\nopenai_base_url = "http://llm-gateway-proxy/v1"\n'
'model_provider = "ccswitch"\nopenai_base_url = "http://llm-gateway-proxy/v1"\n',
encoding="utf-8",
)
wrap_mod._inject_codex_provider_config(8787)
content = config_file.read_text()
content = config_file.read_text(encoding="utf-8")
# Original value kept in a comment so the user can recover it.
# The comment intentionally drops the surrounding quotes — the
# value is a single TOML string and the comment is human-facing.
@ -696,12 +707,12 @@ class TestInjectAvoidsDuplicateTopLevelKeys:
config_dir = tmp_path / ".codex"
config_dir.mkdir()
config_file = config_dir / "config.toml"
config_file.write_text('model_provider = "ccswitch"\n')
config_file.write_text('model_provider = "ccswitch"\n', encoding="utf-8")
wrap_mod._inject_codex_provider_config(8787)
wrap_mod._inject_codex_provider_config(9999) # port change
content = config_file.read_text()
content = config_file.read_text(encoding="utf-8")
tomllib.loads(content)
assert content.count("model_provider =") == 1
assert 'model_provider = "headroom"' in content
@ -716,7 +727,7 @@ class TestInjectAvoidsDuplicateTopLevelKeys:
_set_test_home(monkeypatch, tmp_path)
wrap_mod._inject_codex_provider_config(8787)
content = (tmp_path / ".codex" / "config.toml").read_text()
content = (tmp_path / ".codex" / "config.toml").read_text(encoding="utf-8")
assert wrap_mod._CODEX_TOP_LEVEL_MARKER in content
assert 'model_provider = "headroom"' in content
assert 'openai_base_url = "http://127.0.0.1:8787/v1"' in content
@ -731,13 +742,13 @@ class TestInjectAvoidsDuplicateTopLevelKeys:
config_dir.mkdir()
config_file = config_dir / "config.toml"
original = 'model_provider = "ccswitch"\nopenai_base_url = "http://llm-gateway-proxy/v1"\n'
config_file.write_text(original)
config_file.write_text(original, encoding="utf-8")
wrap_mod._inject_codex_provider_config(8787)
status, _ = wrap_mod._restore_codex_provider_config()
assert status == "restored"
assert config_file.read_text() == original
assert config_file.read_text(encoding="utf-8") == original
# ---------------------------------------------------------------------------
@ -752,16 +763,16 @@ def test_wrap_codex_prepare_only_creates_backup_and_config(
config_file = tmp_path / ".codex" / "config.toml"
config_file.parent.mkdir(parents=True)
original = 'model_provider = "openai"\n'
config_file.write_text(original)
config_file.write_text(original, encoding="utf-8")
with patch("headroom.cli.wrap._ensure_rtk_binary", return_value=None):
result = runner.invoke(main, ["wrap", "codex", "--prepare-only", "--port", "8787"])
assert result.exit_code == 0, result.output
assert 'model_provider = "headroom"' in config_file.read_text()
assert 'model_provider = "headroom"' in config_file.read_text(encoding="utf-8")
backup = tmp_path / ".codex" / "config.toml.headroom-backup"
assert backup.exists()
assert backup.read_text() == original
assert backup.read_text(encoding="utf-8") == original
def test_wrap_codex_prepare_only_respects_codex_home(
@ -781,7 +792,7 @@ def test_wrap_codex_prepare_only_respects_codex_home(
assert result.exit_code == 0, result.output
config_file = codex_home / "config.toml"
assert config_file.exists()
content = config_file.read_text()
content = config_file.read_text(encoding="utf-8")
assert 'model_provider = "headroom"' in content
assert "[mcp_servers.headroom]" in content
assert not (tmp_path / ".codex" / "config.toml").exists()
@ -844,7 +855,7 @@ def test_unwrap_codex_without_codex_home_warns_on_ambiguous_noop(
assert wrap_result.exit_code == 0, wrap_result.output
config_file = codex_home / "config.toml"
assert 'openai_base_url = "http://127.0.0.1:8787/v1"' in config_file.read_text()
assert 'openai_base_url = "http://127.0.0.1:8787/v1"' in config_file.read_text(encoding="utf-8")
monkeypatch.delenv("CODEX_HOME", raising=False)
unwrap_result = runner.invoke(main, ["unwrap", "codex", "--no-stop-proxy"])
@ -856,7 +867,7 @@ def test_unwrap_codex_without_codex_home_warns_on_ambiguous_noop(
assert "If you wrapped Codex with CODEX_HOME" in unwrap_result.output
assert "CODEX_HOME=/path/to/codex-home headroom unwrap codex" in unwrap_result.output
assert "Nothing to undo" in unwrap_result.output
assert 'openai_base_url = "http://127.0.0.1:8787/v1"' in config_file.read_text()
assert 'openai_base_url = "http://127.0.0.1:8787/v1"' in config_file.read_text(encoding="utf-8")
def test_start_proxy_uses_separate_session_for_signal_isolation(
@ -990,14 +1001,15 @@ def test_wrap_codex_prepare_only_updates_stale_mcp_proxy_url(
"\n"
"[mcp_servers.headroom.env]\n"
'HEADROOM_PROXY_URL = "http://127.0.0.1:9000"\n'
"# --- end Headroom MCP server ---\n"
"# --- end Headroom MCP server ---\n",
encoding="utf-8",
)
with patch("headroom.cli.wrap._ensure_rtk_binary", return_value=None):
result = runner.invoke(main, ["wrap", "codex", "--prepare-only", "--port", "8787"])
assert result.exit_code == 0, result.output
content = config_file.read_text()
content = config_file.read_text(encoding="utf-8")
assert "[mcp_servers.headroom]" in content
assert 'command = "headroom"' in content
assert 'args = ["mcp", "serve"]' in content
@ -1086,7 +1098,7 @@ def test_wrap_codex_prepare_only_registers_serena_when_uvx_exists(
result = runner.invoke(main, ["wrap", "codex", "--prepare-only"])
assert result.exit_code == 0, result.output
content = config_file.read_text()
content = config_file.read_text(encoding="utf-8")
assert "[mcp_servers.serena]" in content
assert 'command = "uvx"' in content
assert '"--context", "codex"' in content
@ -1103,7 +1115,7 @@ def test_wrap_codex_prepare_only_no_serena_skips_serena(
result = runner.invoke(main, ["wrap", "codex", "--prepare-only", "--no-serena"])
assert result.exit_code == 0, result.output
assert "[mcp_servers.serena]" not in config_file.read_text()
assert "[mcp_servers.serena]" not in config_file.read_text(encoding="utf-8")
def test_unwrap_codex_restores_prior_config_end_to_end(
@ -1120,12 +1132,12 @@ def test_unwrap_codex_restores_prior_config_end_to_end(
"[model_providers.openai]\n"
'base_url = "https://api.openai.com/v1"\n'
)
config_file.write_text(original)
config_file.write_text(original, encoding="utf-8")
with patch("headroom.cli.wrap._ensure_rtk_binary", return_value=None):
wrap_result = runner.invoke(main, ["wrap", "codex", "--prepare-only", "--port", "8787"])
assert wrap_result.exit_code == 0, wrap_result.output
assert 'model_provider = "headroom"' in config_file.read_text()
assert 'model_provider = "headroom"' in config_file.read_text(encoding="utf-8")
stopped: list[int] = []
@ -1139,8 +1151,8 @@ def test_unwrap_codex_restores_prior_config_end_to_end(
# Config must be byte-for-byte what the user had before wrap, and the
# injected block must be gone — no more "Missing OPENAI_API_KEY" when the
# proxy is stopped.
assert config_file.read_text() == original
assert 'model_provider = "headroom"' not in config_file.read_text()
assert config_file.read_text(encoding="utf-8") == original
assert 'model_provider = "headroom"' not in config_file.read_text(encoding="utf-8")
assert not (tmp_path / ".codex" / "config.toml.headroom-backup").exists()
assert stopped == [9999]
assert "Stopped local Headroom proxy on port 9999" in unwrap_result.output
@ -1337,14 +1349,14 @@ def test_unwrap_codex_preserves_unrelated_sections(
config_file.parent.mkdir(parents=True)
# A config with an MCP server the user configured by hand.
original = '[mcp_servers.local_thing]\ncommand = "/usr/local/bin/thing"\nargs = ["--serve"]\n'
config_file.write_text(original)
config_file.write_text(original, encoding="utf-8")
with patch("headroom.cli.wrap._ensure_rtk_binary", return_value=None):
runner.invoke(main, ["wrap", "codex", "--prepare-only", "--port", "8787"])
result = runner.invoke(main, ["unwrap", "codex"])
assert result.exit_code == 0, result.output
restored = config_file.read_text()
restored = config_file.read_text(encoding="utf-8")
assert restored == original
@ -1368,7 +1380,7 @@ class TestCodexProjectHeaderConfig:
wrap_mod._inject_codex_provider_config(8787)
content = (tmp_path / ".codex" / "config.toml").read_text()
content = (tmp_path / ".codex" / "config.toml").read_text(encoding="utf-8")
assert 'env_http_headers = { "X-Headroom-Project" = "HEADROOM_PROJECT" }' in content
def test_env_http_headers_inside_provider_section(
@ -1380,7 +1392,7 @@ class TestCodexProjectHeaderConfig:
wrap_mod._inject_codex_provider_config(8787)
content = (tmp_path / ".codex" / "config.toml").read_text()
content = (tmp_path / ".codex" / "config.toml").read_text(encoding="utf-8")
section_start = content.index("[model_providers.headroom]")
mapping_pos = content.index("env_http_headers")
end_marker_pos = content.index(wrap_mod._CODEX_END_MARKER, section_start)
@ -1396,10 +1408,10 @@ class TestCodexProjectHeaderConfig:
config_dir.mkdir()
config_file = config_dir / "config.toml"
original = '[profiles.default]\nmodel = "gpt-4o"\n'
config_file.write_text(original)
config_file.write_text(original, encoding="utf-8")
wrap_mod._inject_codex_provider_config(8787)
wrapped = config_file.read_text()
wrapped = config_file.read_text(encoding="utf-8")
assert "env_http_headers" in wrapped
cleaned = wrap_mod._strip_codex_headroom_blocks(wrapped)

2
tests/test_cli/test_wrap_copilot.py
View file

@ -113,7 +113,7 @@ def test_wrap_copilot_auto_anthropic_injects_instructions(
assert result.exit_code == 0, result.output
instructions = tmp_path / ".github" / "copilot-instructions.md"
assert instructions.exists()
content = instructions.read_text()
content = instructions.read_text(encoding="utf-8")
assert wrap_cli._RTK_MARKER in content
assert "RTK (Rust Token Killer)" in content

34
tests/test_cli/test_wrap_hintfile_agents.py
View file

@ -54,7 +54,7 @@ def test_prepare_only_injects_rtk_into_hintfile(
assert result.exit_code == 0, result.output
marker = tmp_path / hintfile
assert marker.exists(), f"{hintfile} should be created"
content = marker.read_text()
content = marker.read_text(encoding="utf-8")
assert wrap_mod._RTK_MARKER in content
assert "RTK (Rust Token Killer)" in content
@ -75,7 +75,7 @@ def test_prepare_only_idempotent_no_duplicate_block(
runner.invoke(main, ["wrap", agent, "--prepare-only"])
runner.invoke(main, ["wrap", agent, "--prepare-only"])
content = (tmp_path / hintfile).read_text()
content = (tmp_path / hintfile).read_text(encoding="utf-8")
assert content.count(wrap_mod._RTK_MARKER) == 1
@ -111,13 +111,13 @@ def test_preserves_existing_hintfile_content(
monkeypatch.delenv("HEADROOM_CONTEXT_TOOL", raising=False)
marker_path = tmp_path / hintfile
original = "# Project conventions\n\nAlways use Python 3.12.\n"
marker_path.write_text(original)
marker_path.write_text(original, encoding="utf-8")
with patch.object(wrap_mod, "_ensure_rtk_binary", return_value=Path("/tmp/rtk")):
result = runner.invoke(main, ["wrap", agent, "--prepare-only"])
assert result.exit_code == 0, result.output
content = marker_path.read_text()
content = marker_path.read_text(encoding="utf-8")
assert "Always use Python 3.12." in content
assert wrap_mod._RTK_MARKER in content
@ -145,7 +145,7 @@ def test_keyboardinterrupt_during_prelude_emits_clear_message(
# hint-file marker but before _ensure_proxy returns. We trigger via
# _ensure_rtk_binary side-effect so the marker exists on disk.
marker_path = tmp_path / hintfile
marker_path.write_text(wrap_mod.RTK_INSTRUCTIONS_BLOCK)
marker_path.write_text(wrap_mod.RTK_INSTRUCTIONS_BLOCK, encoding="utf-8")
raise KeyboardInterrupt
with patch.object(wrap_mod, "_ensure_rtk_binary", side_effect=raise_kbd):
@ -156,3 +156,27 @@ def test_keyboardinterrupt_during_prelude_emits_clear_message(
assert "idempotent" in result.output.lower()
assert (tmp_path / hintfile).exists()
assert hintfile in result.output
@pytest.mark.parametrize("agent,hintfile", HINTFILE_AGENTS)
def test_inject_rtk_handles_utf8_content(
agent: str,
hintfile: str,
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Existing hint files with non-ASCII UTF-8 content must not crash (#1126)."""
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("HEADROOM_CONTEXT_TOOL", raising=False)
marker_path = tmp_path / hintfile
original = "# Instructions\n\nUse “smart quotes” and an em dash — here.\n"
marker_path.write_text(original, encoding="utf-8")
with patch.object(wrap_mod, "_ensure_rtk_binary", return_value=Path("/tmp/rtk")):
result = runner.invoke(main, ["wrap", agent, "--prepare-only"])
assert result.exit_code == 0, result.output
content = marker_path.read_text(encoding="utf-8")
assert "“smart quotes”" in content
assert wrap_mod._RTK_MARKER in content

86
tests/test_cli/test_wrap_opencode.py
View file

@ -132,7 +132,7 @@ def test_wrap_opencode_prepare_only_injects_config(
assert result.exit_code == 0, result.output
config_file = tmp_path / ".config" / "opencode" / "opencode.json"
assert config_file.exists()
config = json.loads(config_file.read_text())
config = json.loads(config_file.read_text(encoding="utf-8"))
assert config["provider"]["headroom"]["options"]["baseURL"] == "http://127.0.0.1:9000/v1"
@ -209,8 +209,8 @@ def test_wrap_opencode_injects_rtk_into_agents_md(
project_agents = tmp_path / "AGENTS.md"
assert global_agents.exists(), "Global AGENTS.md should be created"
assert project_agents.exists(), "Project AGENTS.md should be created"
assert wrap_mod._RTK_MARKER in global_agents.read_text()
assert wrap_mod._RTK_MARKER in project_agents.read_text()
assert wrap_mod._RTK_MARKER in global_agents.read_text(encoding="utf-8")
assert wrap_mod._RTK_MARKER in project_agents.read_text(encoding="utf-8")
def test_wrap_opencode_idempotent_no_duplicate_block(
@ -230,7 +230,7 @@ def test_wrap_opencode_idempotent_no_duplicate_block(
runner.invoke(main, ["wrap", "opencode", "--port", "9000", "--no-mcp"])
project_agents = tmp_path / "AGENTS.md"
content = project_agents.read_text()
content = project_agents.read_text(encoding="utf-8")
assert content.count(wrap_mod._RTK_MARKER) == 1
@ -261,7 +261,7 @@ def test_unwrap_opencode_restores_from_backup(
assert result.exit_code == 0, result.output
assert "Restored prior" in result.output
assert not backup_file.exists()
assert config_file.read_text() == original
assert config_file.read_text(encoding="utf-8") == original
def test_unwrap_opencode_strips_blocks_when_no_backup(
@ -290,8 +290,8 @@ def test_unwrap_opencode_strips_blocks_when_no_backup(
assert result.exit_code == 0, result.output
assert "Removed Headroom block" in result.output
assert user_content in config_file.read_text()
assert wrap_mod._PROVIDER_MARKER_START not in config_file.read_text()
assert user_content in config_file.read_text(encoding="utf-8")
assert wrap_mod._PROVIDER_MARKER_START not in config_file.read_text(encoding="utf-8")
# ---------------------------------------------------------------------------
@ -319,7 +319,7 @@ def test_wrap_opencode_preserves_existing_user_providers(
result = runner.invoke(main, ["wrap", "opencode", "--port", "9000", "--no-mcp"])
assert result.exit_code == 0, result.output
config = json.loads(config_file.read_text())
config = json.loads(config_file.read_text(encoding="utf-8"))
assert "headroom" in config["provider"], "headroom provider not injected"
assert "openai" in config["provider"], "user's openai provider was removed"
@ -341,7 +341,7 @@ def test_wrap_opencode_port_change_updates_existing_config(
runner.invoke(main, ["wrap", "opencode", "--port", "9001", "--no-mcp"])
config_file = tmp_path / ".config" / "opencode" / "opencode.json"
config = json.loads(config_file.read_text())
config = json.loads(config_file.read_text(encoding="utf-8"))
assert config["provider"]["headroom"]["options"]["baseURL"] == "http://127.0.0.1:9001/v1"
@ -368,9 +368,11 @@ def test_wrap_opencode_handles_malformed_config_file(
assert result.exit_code == 0, result.output
assert backup_file.exists(), "backup must be created before overwriting"
assert backup_file.read_text() == malformed, "backup must preserve original byte-for-byte"
assert backup_file.read_text(encoding="utf-8") == malformed, (
"backup must preserve original byte-for-byte"
)
# The config file is now valid JSON with headroom provider.
config = json.loads(config_file.read_text())
config = json.loads(config_file.read_text(encoding="utf-8"))
assert "headroom" in config.get("provider", {})
@ -394,7 +396,7 @@ def test_wrap_opencode_handles_empty_config_file(
result = runner.invoke(main, ["wrap", "opencode", "--port", "9000", "--no-mcp"])
assert result.exit_code == 0, result.output
config = json.loads(config_file.read_text())
config = json.loads(config_file.read_text(encoding="utf-8"))
assert config["provider"]["headroom"]["options"]["baseURL"] == "http://127.0.0.1:9000/v1"
@ -440,7 +442,7 @@ def test_wrap_opencode_rtk_preserves_existing_agents_md(
result = runner.invoke(main, ["wrap", "opencode", "--port", "9000", "--no-mcp"])
assert result.exit_code == 0, result.output
content = (tmp_path / "AGENTS.md").read_text()
content = (tmp_path / "AGENTS.md").read_text(encoding="utf-8")
assert existing_content in content
assert wrap_mod._RTK_MARKER in content
@ -466,7 +468,7 @@ def test_wrap_opencode_no_rtk_leaves_agents_md_untouched(
)
assert result.exit_code == 0, result.output
content = (tmp_path / "AGENTS.md").read_text()
content = (tmp_path / "AGENTS.md").read_text(encoding="utf-8")
assert content == existing_content, "--no-rtk modified AGENTS.md"
assert wrap_mod._RTK_MARKER not in content
@ -571,7 +573,7 @@ def test_wrap_opencode_config_merges_existing_model(
result = runner.invoke(main, ["wrap", "opencode", "--port", "9000", "--no-mcp"])
assert result.exit_code == 0, result.output
config = json.loads(config_file.read_text())
config = json.loads(config_file.read_text(encoding="utf-8"))
assert config["model"] == "openai/gpt-4o"
assert config["provider"]["headroom"]["npm"] == "@ai-sdk/openai-compatible"
@ -639,7 +641,7 @@ def test_unwrap_opencode_noop_when_no_headroom_markers(
assert result.exit_code == 0, result.output
assert "no Headroom wrap markers" in result.output
assert config_file.read_text().strip() == '{"model": "openai/gpt-4o"}'
assert config_file.read_text(encoding="utf-8").strip() == '{"model": "openai/gpt-4o"}'
def test_wrap_unwrap_rewrap_is_idempotent(
@ -668,7 +670,7 @@ def test_wrap_unwrap_rewrap_is_idempotent(
runner.invoke(main, ["unwrap", "opencode"])
# After unwrap, file should match original
after_unwrap = json.loads(config_file.read_text())
after_unwrap = json.loads(config_file.read_text(encoding="utf-8"))
assert after_unwrap["model"] == "openai/gpt-4o"
assert "headroom" not in after_unwrap.get("provider", {})
@ -679,7 +681,7 @@ def test_wrap_unwrap_rewrap_is_idempotent(
runner.invoke(main, ["wrap", "opencode", "--port", "9001", "--no-mcp"])
# After re-wrap, headroom should be back, model unchanged
after_rewrap = json.loads(config_file.read_text())
after_rewrap = json.loads(config_file.read_text(encoding="utf-8"))
assert after_rewrap["model"] == "openai/gpt-4o"
assert "headroom" in after_rewrap.get("provider", {})
@ -844,3 +846,51 @@ def test_wrap_opencode_respects_opencode_home_env(
assert result.exit_code == 0, result.output
agents_md = Path(custom_home) / "AGENTS.md"
assert agents_md.exists()
# ---------------------------------------------------------------------------
# Regression: unwrap must preserve non-ASCII UTF-8 user content (#1126)
# ---------------------------------------------------------------------------
def test_unwrap_opencode_preserves_utf8_user_content(
runner: CliRunner,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Unwrap strips Headroom blocks but preserves non-ASCII UTF-8 user content (#1126)."""
monkeypatch.chdir(tmp_path)
_set_test_home(monkeypatch, tmp_path)
config_file = tmp_path / ".config" / "opencode" / "opencode.json"
config_file.parent.mkdir(parents=True, exist_ok=True)
# User content with smart quotes and em dashes (non-ASCII UTF-8)
user_config = {
"model": "openai/gpt-4o",
"description": "“smart quotes” and an em dash — here",
}
user_json = json.dumps(user_config, ensure_ascii=False)
wrapped_content = (
wrap_mod._PROVIDER_MARKER_START
+ '\n"provider": {},\n'
+ wrap_mod._PROVIDER_MARKER_END
+ "\n"
+ user_json
)
config_file.write_text(wrapped_content, encoding="utf-8")
# Mock out OpencodeRegistrar to avoid its own bare-open encoding issue
# (pre-existing; outside this PR's scope).
fake_registrar = type("FakeRegistrar", (), {"detect": lambda self: False})()
with patch.object(wrap_mod, "_stop_local_proxy_for_unwrap", return_value="stopped"):
with patch("headroom.mcp_registry.OpencodeRegistrar", return_value=fake_registrar):
result = runner.invoke(main, ["unwrap", "opencode"])
assert result.exit_code == 0, result.output
assert "Removed Headroom block" in result.output
content = config_file.read_text(encoding="utf-8")
assert "“smart quotes”" in content
assert "—" in content
assert wrap_mod._PROVIDER_MARKER_START not in content