fix: use UTF-8 throughout skill compliance workflows

RED: seven Unicode regressions reproduced under Windows cp1252 and Linux ASCII with UTF-8 mode disabled. GREEN: Windows 91 passed with 5 platform skips; Linux ASCII 96 passed. Nine offline Unicode cases cover file, temporary YAML, prompt and all four CLI capture boundaries without model calls. Changed functions have 84.7% line coverage, each exceeds 80%; all 13 changed IO sites execute. Pi/catalog parity checks pass.
(cherry picked from commit 642cb0d512)
This commit is contained in:
Eneko Ruiz authored and haelyra committed 2026-10-20 17:41:00 -04:00
1 parent 99b6a9bb8f
commit 429737572a
7 files changed
+144 -13

No files matched your search

+2 -2
View File
@@ -40,7 +40,7 @@ def classify_events(
for i, event in enumerate(trace)
)
prompt_template = (PROMPTS_DIR / "classifier.md").read_text()
prompt_template = (PROMPTS_DIR / "classifier.md").read_text(encoding="utf-8")
prompt = (
prompt_template
.replace("{steps_description}", steps_desc)
@@ -50,7 +50,7 @@ def classify_events(
result = subprocess.run(
["claude", "-p", prompt, "--model", model, "--output-format", "text"],
capture_output=True,
text=True,
encoding="utf-8",
timeout=60,
)
+2 -2
View File
@@ -49,7 +49,7 @@ def parse_trace(path: Path) -> list[ObservationEvent]:
if not path.is_file():
raise FileNotFoundError(f"Trace file not found: {path}")
text = path.read_text().strip()
text = path.read_text(encoding="utf-8").strip()
if not text:
return []
@@ -78,7 +78,7 @@ def parse_spec(path: Path) -> ComplianceSpec:
"""Parse a YAML compliance spec file."""
if not path.is_file():
raise FileNotFoundError(f"Spec file not found: {path}")
raw = yaml.safe_load(path.read_text())
raw = yaml.safe_load(path.read_text(encoding="utf-8"))
steps: list[Step] = []
for s in raw["steps"]:
+1 -1
View File
@@ -106,7 +106,7 @@ def main() -> None:
report = generate_report(args.skill, spec, graded_results, scenarios=scenarios)
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_text(report)
output_path.write_text(report, encoding="utf-8")
logger.info(" Report saved to %s", output_path)
# Summary
+1 -1
View File
@@ -59,7 +59,7 @@ def run_scenario(
"--verbose",
],
capture_output=True,
text=True,
encoding="utf-8",
timeout=timeout,
cwd=sandbox_dir,
)
@@ -32,8 +32,8 @@ def generate_scenarios(
Calls claude -p with the scenario_generator prompt, parses YAML output.
"""
skill_content = skill_path.read_text()
prompt_template = (PROMPTS_DIR / "scenario_generator.md").read_text()
skill_content = skill_path.read_text(encoding="utf-8")
prompt_template = (PROMPTS_DIR / "scenario_generator.md").read_text(encoding="utf-8")
prompt = (
prompt_template
.replace("{skill_content}", skill_content)
@@ -43,7 +43,7 @@ def generate_scenarios(
result = subprocess.run(
["claude", "-p", prompt, "--model", model, "--output-format", "text"],
capture_output=True,
text=True,
encoding="utf-8",
timeout=120,
)
@@ -24,8 +24,8 @@ def generate_spec(
Calls claude -p with the spec_generator prompt, parses YAML output.
Retries on YAML parse errors with error feedback.
"""
skill_content = skill_path.read_text()
prompt_template = (PROMPTS_DIR / "spec_generator.md").read_text()
skill_content = skill_path.read_text(encoding="utf-8")
prompt_template = (PROMPTS_DIR / "spec_generator.md").read_text(encoding="utf-8")
base_prompt = prompt_template.replace("{skill_content}", skill_content)
last_error: Exception | None = None
@@ -43,7 +43,7 @@ def generate_spec(
result = subprocess.run(
["claude", "-p", prompt, "--model", model, "--output-format", "text"],
capture_output=True,
text=True,
encoding="utf-8",
timeout=120,
)
@@ -54,7 +54,7 @@ def generate_spec(
tmp_path = None
with tempfile.NamedTemporaryFile(
mode="w", suffix=".yaml", delete=False,
mode="w", suffix=".yaml", delete=False, encoding="utf-8",
) as f:
f.write(raw_yaml)
tmp_path = Path(f.name)
+131
View File
@@ -0,0 +1,131 @@
"""UTF-8 file and CLI boundaries must preserve Unicode under legacy locales."""
import json
import locale
import subprocess
import sys
from pathlib import Path
from typing import Any, Callable
import pytest
import yaml
from scripts import classifier, parser, run, runner, scenario_generator, spec_generator
from scripts.grader import ComplianceResult
from scripts.parser import ComplianceSpec, Detector, ObservationEvent, Step
from scripts.runner import ScenarioRun
from scripts.scenario_generator import Scenario
TEXT = "caf\u00e9 \u65e5\u672c"
PROMPT_TEXT = "consign\u00e9e"
SPEC = {"id": "unicode", "name": TEXT, "source_rule": TEXT, "version": "1",
"steps": [], "scoring": {"threshold_promote_to_hook": 0.6}}
def mock_cli(payload: str) -> Callable[..., subprocess.CompletedProcess[str]]:
"""Emulate decoding UTF-8 CLI bytes using the caller's selected encoding."""
def execute(command: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]:
assert TEXT in command[2]
encoding = kwargs.get("encoding") or locale.getpreferredencoding(False)
return subprocess.CompletedProcess(command, 0, stdout=payload.encode("utf-8").decode(encoding), stderr="")
return execute
@pytest.mark.parametrize("kind", ["trace", "spec"])
def test_parser_reads_utf8(tmp_path: Path, kind: str) -> None:
source = tmp_path / "source.txt"
if kind == "trace":
record = {"timestamp": "T0000", "event": "tool_complete", "tool": "Read",
"session": "unicode", "input": TEXT, "output": TEXT}
source.write_text(json.dumps(record, ensure_ascii=False), encoding="utf-8")
assert parser.parse_trace(source)[0].input == TEXT
else:
source.write_text(yaml.safe_dump(SPEC, allow_unicode=True), encoding="utf-8")
assert parser.parse_spec(source).name == TEXT
def test_invalid_unicode_trace_keeps_line_diagnostic(tmp_path: Path) -> None:
source = tmp_path / "invalid.jsonl"
source.write_text('{"input": "' + TEXT + '", invalid}', encoding="utf-8")
with pytest.raises(ValueError, match="Invalid JSON at line 1"):
parser.parse_trace(source)
@pytest.mark.parametrize("kind", ["spec", "scenarios"])
def test_generators_preserve_utf8(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, kind: str) -> None:
skill = tmp_path / "SKILL.md"
skill.write_text(TEXT, encoding="utf-8")
module = spec_generator if kind == "spec" else scenario_generator
monkeypatch.setattr(module, "PROMPTS_DIR", tmp_path)
(tmp_path / f"{kind[:-1] if kind == 'scenarios' else kind}_generator.md").write_text(
PROMPT_TEXT + " {skill_content} {spec_yaml}", encoding="utf-8",
)
response = SPEC if kind == "spec" else {"scenarios": [
{"id": "one", "level": 1, "level_name": TEXT, "description": TEXT, "prompt": TEXT},
]}
execute = mock_cli(yaml.safe_dump(response, allow_unicode=True))
def checked_execute(command: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]:
assert PROMPT_TEXT in command[2]
return execute(command, **kwargs)
monkeypatch.setattr(module.subprocess, "run", checked_execute)
if kind == "spec":
assert spec_generator.generate_spec(skill, max_retries=0).name == TEXT
else:
assert scenario_generator.generate_scenarios(skill, "steps: []")[0].prompt == TEXT
def test_spec_retry_preserves_utf8(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
skill = tmp_path / "SKILL.md"
skill.write_text(TEXT, encoding="utf-8")
responses = iter(["steps: [invalid", yaml.safe_dump(SPEC, allow_unicode=True)])
prompts: list[str] = []
def execute(command: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]:
prompts.append(command[2])
return mock_cli(next(responses))(command, **kwargs)
monkeypatch.setattr(spec_generator.subprocess, "run", execute)
assert spec_generator.generate_spec(skill, max_retries=1).name == TEXT
assert len(prompts) == 2
assert "PREVIOUS ATTEMPT FAILED" in prompts[1]
def test_classifier_preserves_utf8(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
(tmp_path / "classifier.md").write_text(PROMPT_TEXT + " {steps_description} {tool_calls}", encoding="utf-8")
monkeypatch.setattr(classifier, "PROMPTS_DIR", tmp_path)
execute = mock_cli(json.dumps({TEXT: [0]}, ensure_ascii=False))
def checked_execute(command: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]:
assert PROMPT_TEXT in command[2]
return execute(command, **kwargs)
monkeypatch.setattr(classifier.subprocess, "run", checked_execute)
spec = ComplianceSpec("unicode", TEXT, TEXT, "1", (Step(TEXT, TEXT, True, Detector(TEXT)),), 0.6)
trace = [ObservationEvent("T0000", "tool_complete", "Read", "unicode", TEXT, TEXT)]
assert classifier.classify_events(spec, trace) == {TEXT: [0]}
def test_runner_preserves_utf8_capture(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(runner, "SANDBOX_BASE", tmp_path)
monkeypatch.setattr(runner, "_setup_sandbox", lambda *_args: None)
messages = [
{"type": "assistant", "message": {"content": [{"type": "tool_use", "id": "one", "name": "Read", "input": TEXT}]}},
{"type": "user", "session_id": "unicode", "message": {"content": [{"type": "tool_result", "tool_use_id": "one", "content": TEXT}]}},
]
payload = "\n".join(json.dumps(message, ensure_ascii=False) for message in messages)
monkeypatch.setattr(runner.subprocess, "run", mock_cli(payload))
scenario = Scenario("unicode", 1, "one", TEXT, TEXT, ())
assert runner.run_scenario(scenario).observations[0].output == TEXT
def test_report_is_written_as_utf8(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
skill = tmp_path / "SKILL.md"
skill.write_text(TEXT, encoding="utf-8")
output = tmp_path / "report.md"
monkeypatch.setattr(run, "__file__", str(tmp_path / "scripts/run.py"))
monkeypatch.setattr(sys, "argv", ["skill-comply", str(skill), "--output", str(output)])
monkeypatch.setattr(run, "generate_spec", lambda *_args, **_kwargs: parser.ComplianceSpec("unicode", TEXT, TEXT, "1", (), 0.6))
scenario = Scenario("unicode", 1, TEXT, TEXT, TEXT, ())
observation = ObservationEvent("T0000", "tool_complete", "Read", "unicode", TEXT, TEXT)
monkeypatch.setattr(run, "generate_scenarios", lambda *_args, **_kwargs: [scenario])
monkeypatch.setattr(run, "run_scenario", lambda *_args, **_kwargs: ScenarioRun(scenario, (observation,), tmp_path))
monkeypatch.setattr(run, "grade", lambda *_args, **_kwargs: ComplianceResult("unicode", (), 0.5, True, {}))
monkeypatch.setattr(run, "generate_report", lambda *_args, **_kwargs: TEXT)
run.main()
assert output.read_text(encoding="utf-8") == TEXT