fix: decode/encode owned config, state and template assets as UTF-8

Headroom reads and writes its own dashboard template, JSON deployment/sync
state and provider config files using the platform default text codec. On
systems whose default encoding is not UTF-8 (e.g. Windows cp949/cp1252
locales) this raises UnicodeDecodeError when the file contains non-ASCII
bytes.

The dashboard template ships with non-ASCII UTF-8 content, so loading the
dashboard crashes on a Korean Windows locale at byte 20523 (fixes #533).
The same latent bug exists in the sibling JSON state/config I/O; since these
files are owned by Headroom and JSON is UTF-8 by spec (RFC 8259 §8.1), read
and write them with an explicit encoding="utf-8" so they round-trip on every
platform.

Add a regression test covering the dashboard load and a non-ASCII JSON
state round-trip.
This commit is contained in:
supermario_leo 2026-06-03 03:25:42 +08:00
parent 9f8b621eb1
commit 2f1538a641
7 changed files with 77 additions and 16 deletions

View file

@ -9,4 +9,4 @@ TEMPLATES_DIR = DASHBOARD_DIR / "templates"
def get_dashboard_html() -> str:
"""Load the dashboard HTML template."""
template_path = TEMPLATES_DIR / "dashboard.html"
return template_path.read_text()
return template_path.read_text(encoding="utf-8")

View file

@ -24,7 +24,7 @@ def save_manifest(manifest: DeploymentManifest) -> None:
root.mkdir(parents=True, exist_ok=True)
manifest.updated_at = iso_utc_now()
path = manifest_path(manifest.profile)
path.write_text(json.dumps(asdict(manifest), indent=2) + "\n")
path.write_text(json.dumps(asdict(manifest), indent=2) + "\n", encoding="utf-8")
except OSError as e:
logger.warning("Cannot save deployment manifest: %s — continuing without persistence", e)
@ -35,7 +35,7 @@ def load_manifest(profile: str = "default") -> DeploymentManifest | None:
path = manifest_path(profile)
if not path.exists():
return None
payload = json.loads(path.read_text())
payload = json.loads(path.read_text(encoding="utf-8"))
payload["mutations"] = [ManagedMutation(**item) for item in payload.get("mutations", [])]
payload["artifacts"] = [ArtifactRecord(**item) for item in payload.get("artifacts", [])]
return DeploymentManifest(**payload)
@ -51,7 +51,7 @@ def list_manifests() -> list[DeploymentManifest]:
manifests: list[DeploymentManifest] = []
for candidate in sorted(root.glob("*/manifest.json")):
try:
payload = json.loads(candidate.read_text())
payload = json.loads(candidate.read_text(encoding="utf-8"))
payload["mutations"] = [
ManagedMutation(**item) for item in payload.get("mutations", [])
]

View file

@ -124,7 +124,7 @@ def _load_sync_state(state_path: Path) -> dict[str, Any]:
"""Load sync state from disk."""
if state_path.exists():
try:
result: dict[str, Any] = json.loads(state_path.read_text())
result: dict[str, Any] = json.loads(state_path.read_text(encoding="utf-8"))
return result
except (json.JSONDecodeError, OSError):
pass
@ -134,7 +134,7 @@ def _load_sync_state(state_path: Path) -> dict[str, Any]:
def _save_sync_state(state_path: Path, state: dict[str, Any]) -> None:
"""Save sync state to disk."""
state_path.parent.mkdir(parents=True, exist_ok=True)
state_path.write_text(json.dumps(state, indent=2))
state_path.write_text(json.dumps(state, indent=2), encoding="utf-8")
def _db_fingerprint(memories: list[Any]) -> str:

View file

@ -26,14 +26,14 @@ def apply_provider_scope(manifest: DeploymentManifest) -> ManagedMutation | None
path.parent.mkdir(parents=True, exist_ok=True)
payload: dict[str, object] = {}
if path.exists():
payload = json.loads(path.read_text())
payload = json.loads(path.read_text(encoding="utf-8"))
env = payload.get("env")
env_map = dict(env) if isinstance(env, dict) else {}
values = manifest.tool_envs.get(ToolTarget.CLAUDE.value, {})
previous = {name: env_map.get(name) for name in values}
env_map.update(values)
payload["env"] = env_map
path.write_text(json.dumps(payload, indent=2) + "\n")
path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
return ManagedMutation(
target=ToolTarget.CLAUDE.value,
kind="json-env",
@ -49,7 +49,7 @@ def revert_provider_scope(mutation: ManagedMutation, manifest: DeploymentManifes
path = Path(mutation.path)
if not path.exists():
return
payload = json.loads(path.read_text())
payload = json.loads(path.read_text(encoding="utf-8"))
env = payload.get("env")
env_map = dict(env) if isinstance(env, dict) else {}
previous: dict[str, object] = mutation.data.get("previous", {})
@ -60,4 +60,4 @@ def revert_provider_scope(mutation: ManagedMutation, manifest: DeploymentManifes
else:
env_map[name] = previous[name]
payload["env"] = env_map
path.write_text(json.dumps(payload, indent=2) + "\n")
path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")

View file

@ -80,14 +80,14 @@ def apply_provider_scope(manifest: DeploymentManifest) -> ManagedMutation | None
+ f"{_CODEX_MARKER_END}\n"
)
if path.exists():
existing = path.read_text()
existing = path.read_text(encoding="utf-8")
if _CODEX_MARKER_START in existing:
merged = _CODEX_PATTERN.sub(section, existing)
else:
merged = existing.rstrip() + "\n\n" + section + "\n"
else:
merged = section + "\n"
path.write_text(merged)
path.write_text(merged, encoding="utf-8")
return ManagedMutation(target=ToolTarget.CODEX.value, kind="toml-block", path=str(path))
@ -99,7 +99,7 @@ def revert_provider_scope(mutation: ManagedMutation, manifest: DeploymentManifes
path = Path(mutation.path)
if not path.exists():
return
content = path.read_text()
content = path.read_text(encoding="utf-8")
# Remove the managed marker block.
if _CODEX_MARKER_START in content:
content = _CODEX_PATTERN.sub("", content)
@ -108,4 +108,4 @@ def revert_provider_scope(mutation: ManagedMutation, manifest: DeploymentManifes
content = _ORPHAN_MODEL_PROVIDER.sub("", content)
content = _ORPHAN_OPENAI_BASE_URL.sub("", content)
content = _ORPHAN_HEADROOM_TABLE.sub("", content)
path.write_text(content.strip() + "\n")
path.write_text(content.strip() + "\n", encoding="utf-8")

View file

@ -357,7 +357,9 @@ class UsageReporter:
return
try:
self._cache_path.parent.mkdir(parents=True, exist_ok=True)
self._cache_path.write_text(json.dumps(self._license_info.to_dict(), indent=2))
self._cache_path.write_text(
json.dumps(self._license_info.to_dict(), indent=2), encoding="utf-8"
)
except OSError:
logger.warning("Could not save license cache to %s", self._cache_path)
@ -365,7 +367,7 @@ class UsageReporter:
"""Load cached license info, or return a default if expired/missing."""
try:
if self._cache_path.exists():
data = json.loads(self._cache_path.read_text())
data = json.loads(self._cache_path.read_text(encoding="utf-8"))
cached = LicenseInfo.from_dict(data)
age = (datetime.now(timezone.utc) - cached.validated_at).total_seconds()
if age < GRACE_PERIOD_SECONDS:

View file

@ -0,0 +1,59 @@
"""Regression tests for UTF-8 decoding/encoding of headroom-owned assets.
These guard against ``UnicodeDecodeError`` on systems whose default text
encoding is not UTF-8 (e.g. Windows ``cp949``/``cp1252`` locales). Headroom
ships and writes its own templates, JSON state and config files as UTF-8, so
they must be read and written with an explicit ``encoding="utf-8"`` rather than
relying on the platform default codec. See issue #533.
"""
from __future__ import annotations
from pathlib import Path
from headroom.dashboard import TEMPLATES_DIR, get_dashboard_html
from headroom.memory.sync import _load_sync_state, _save_sync_state
def test_dashboard_template_contains_non_ascii() -> None:
"""The bundled template has non-ASCII bytes, so the bug is reproducible."""
raw = (TEMPLATES_DIR / "dashboard.html").read_bytes()
assert any(byte > 0x7F for byte in raw), "template expected to contain non-ASCII bytes"
def test_get_dashboard_html_reads_as_utf8(monkeypatch) -> None:
"""get_dashboard_html must decode the template as UTF-8, not the OS default.
Before the fix, ``read_text()`` used the platform default codec and raised
``UnicodeDecodeError`` on non-UTF-8 locales. We assert the explicit encoding
is passed so the regression cannot silently return (a utf-8 CI host would
otherwise mask it).
"""
captured: dict[str, object] = {}
original = Path.read_text
def _spy(self: Path, *args: object, **kwargs: object) -> str:
captured["encoding"] = kwargs.get("encoding")
return original(self, *args, **kwargs) # type: ignore[arg-type]
monkeypatch.setattr(Path, "read_text", _spy)
html = get_dashboard_html()
assert captured["encoding"] == "utf-8"
assert html # non-empty
# Content must equal an explicit UTF-8 decode of the raw template.
expected = (TEMPLATES_DIR / "dashboard.html").read_bytes().decode("utf-8")
assert html == expected
def test_sync_state_round_trips_non_ascii(tmp_path) -> None:
"""JSON sync state with non-ASCII values must survive a save/load round-trip."""
state_path = tmp_path / "nested" / "sync_state.json"
state = {"agent": "café", "note": "한국어 메모", "emoji": "🚀"}
_save_sync_state(state_path, state)
# Persisted bytes must be valid UTF-8 regardless of the platform default.
assert state_path.read_bytes().decode("utf-8")
assert _load_sync_state(state_path) == state