diff --git a/skills/skill-comply/scripts/classifier.py b/skills/skill-comply/scripts/classifier.py index 74361c8bf..281792c67 100644 --- a/skills/skill-comply/scripts/classifier.py +++ b/skills/skill-comply/scripts/classifier.py @@ -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, ) diff --git a/skills/skill-comply/scripts/parser.py b/skills/skill-comply/scripts/parser.py index 0b8169d4f..30b99a011 100644 --- a/skills/skill-comply/scripts/parser.py +++ b/skills/skill-comply/scripts/parser.py @@ -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"]: diff --git a/skills/skill-comply/scripts/run.py b/skills/skill-comply/scripts/run.py index 3e4695e51..ede6186af 100644 --- a/skills/skill-comply/scripts/run.py +++ b/skills/skill-comply/scripts/run.py @@ -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 diff --git a/skills/skill-comply/scripts/runner.py b/skills/skill-comply/scripts/runner.py index eeb8fea32..dd75b6a8f 100644 --- a/skills/skill-comply/scripts/runner.py +++ b/skills/skill-comply/scripts/runner.py @@ -59,7 +59,7 @@ def run_scenario( "--verbose", ], capture_output=True, - text=True, + encoding="utf-8", timeout=timeout, cwd=sandbox_dir, ) diff --git a/skills/skill-comply/scripts/scenario_generator.py b/skills/skill-comply/scripts/scenario_generator.py index db8db26a1..aeb262b4b 100644 --- a/skills/skill-comply/scripts/scenario_generator.py +++ b/skills/skill-comply/scripts/scenario_generator.py @@ -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, ) diff --git a/skills/skill-comply/scripts/spec_generator.py b/skills/skill-comply/scripts/spec_generator.py index 3b3661736..8d99036ef 100644 --- a/skills/skill-comply/scripts/spec_generator.py +++ b/skills/skill-comply/scripts/spec_generator.py @@ -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) diff --git a/skills/skill-comply/tests/test_utf8.py b/skills/skill-comply/tests/test_utf8.py new file mode 100644 index 000000000..def3d5246 --- /dev/null +++ b/skills/skill-comply/tests/test_utf8.py @@ -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