fix(benchmark): preserve trajectory evidence across locales

loopx/capabilities/benchmark_toolkit/trajectory_assessment.py

@@ -33,7 +33,9 @@ def _result_proof(directory: Path, prepared: dict[str, Any]) -> dict[str, str]:
 
 def _write(path: Path, value: Any) -> None:
     temporary = path.with_suffix(".tmp")
-    temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n")
+    temporary.write_text(
+        json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
+    )
     temporary.replace(path)
 
 
@@ -43,14 +45,14 @@ def prepare(
     return effect_runtime_result(
         "benchmark.trajectory.prepare",
         {
-            "input": json.loads(input_path.read_text()),
+            "input": json.loads(input_path.read_text(encoding="utf-8")),
             "config": {"model": model, "effort": effort, "service_tier": service_tier},
         },
     )
 
 
 def read_job(directory: Path) -> dict[str, Any]:
-    receipt = json.loads((directory / "receipt.json").read_text())
+    receipt = json.loads((directory / "receipt.json").read_text(encoding="utf-8"))
     if receipt.get("schema_version") != "trajectory_assessment_job_v1":
         raise ValueError("Unsupported trajectory assessment receipt")
     return receipt
@@ -106,7 +108,9 @@ def execute(
                     {
                         **prepared,
                         "output": json.loads(
-                            (directory / "model-output.json").read_text()
+                            (directory / "model-output.json").read_text(
+                                encoding="utf-8"
+                            )
                         ),
                     },
                 )
@@ -138,7 +142,7 @@ def execute(
         (directory / "cancel.request").unlink(missing_ok=True)
         _write(directory / "input.json", prepared["input"])
         _write(directory / "output-schema.json", prepared["output_schema"])
-        (directory / "prompt.txt").write_text(prepared["prompt"])
+        (directory / "prompt.txt").write_text(prepared["prompt"], encoding="utf-8")
         config = prepared["config"]
         command = [
             codex,
@@ -186,9 +190,9 @@ def execute(
         }
         _write(directory / "receipt.json", receipt)
         with (
-            (directory / "prompt.txt").open() as prompt,
-            (directory / "events.jsonl").open("w") as events,
-            (directory / "stderr.txt").open("w") as errors,
+            (directory / "prompt.txt").open(encoding="utf-8") as prompt,
+            (directory / "events.jsonl").open("w", encoding="utf-8") as events,
+            (directory / "stderr.txt").open("w", encoding="utf-8") as errors,
         ):
             process = subprocess.Popen(
                 command,
@@ -197,6 +201,7 @@ def execute(
                 stdout=events,
                 stderr=errors,
                 text=True,
+                encoding="utf-8",
             )
             receipt["process_id"] = process.pid
             _write(directory / "receipt.json", receipt)
@@ -220,15 +225,19 @@ def execute(
                 raise ValueError(
                     f"Codex exited {process.returncode}; inspect stderr.txt and events.jsonl"
                 )
-            output = json.loads((directory / "model-output.json").read_text())
+            output = json.loads(
+                (directory / "model-output.json").read_text(encoding="utf-8")
+            )
             result = effect_runtime_result(
                 "benchmark.trajectory.qualify", {**prepared, "output": output}
             )
             _write(directory / "assessment.json", result)
             receipt.update(
                 status="succeeded", ok=True, **_result_proof(directory, prepared)
             )
-        for line in (directory / "events.jsonl").read_text().splitlines():
+        for line in (
+            (directory / "events.jsonl").read_text(encoding="utf-8").splitlines()
+        ):
             try:
                 event = json.loads(line)
             except json.JSONDecodeError:

loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py

@@ -10,8 +10,8 @@
 import hashlib
 import json
 from pathlib import Path, PurePosixPath
-import re
 
+from ...control_plane.content_digest import BARE_SHA256_PATTERN
 from ...control_plane.effect_runtime import effect_runtime_result
 from ...file_lock import exclusive_file_lock
 from .trajectory_assessment import _digest, _write, execute, read_job, request_cancel
@@ -28,7 +28,7 @@ def __init__(self, reason_code: str):
 
 
 def _opaque(value: str) -> str:
-    if not isinstance(value, str) or not re.fullmatch(r"[0-9a-f]{64}", value):
+    if not isinstance(value, str) or not BARE_SHA256_PATTERN.fullmatch(value):
         raise PackUnavailable("invalid_identifier")
     return value
 
@@ -88,7 +88,7 @@ def _index(owner: Path) -> dict:
     path = owner / "packs.json"
     if not path.exists():
         return {"default_pack_id": None, "pack_ids": []}
-    value = json.loads(path.read_text())
+    value = json.loads(path.read_text(encoding="utf-8"))
     for pack_id in value["pack_ids"]:
         _opaque(pack_id)
     if value["default_pack_id"] not in value["pack_ids"]:
@@ -107,7 +107,7 @@ def _resolve(
     path = owner / "packs" / f"{pack_id}.json"
     if path.is_symlink():
         raise PackUnavailable("invalid_registration")
-    record = json.loads(path.read_text())
+    record = json.loads(path.read_text(encoding="utf-8"))
     if (
         record["owner_goal_id"] != owner_goal_id
         or _pack_id(owner_goal_id, record) != pack_id
@@ -159,7 +159,7 @@ def register_pack(
     prepared = effect_runtime_result(
         "benchmark.trajectory.prepare",
         {
-            "input": json.loads(input_path.read_text()),
+            "input": json.loads(input_path.read_text(encoding="utf-8")),
             "config": config or DEFAULT_CONFIG,
         },
     )
@@ -180,7 +180,7 @@ def register_pack(
         index = _index(owner)
         path = owner / "packs" / f"{pack_id}.json"
         if path.exists():
-            if json.loads(path.read_text()) != record:
+            if json.loads(path.read_text(encoding="utf-8")) != record:
                 raise PackUnavailable("registration_conflict")
         else:
             _write(path, record)
@@ -286,7 +286,9 @@ def inspect_pack(
                     {
                         **prepared,
                         "output": json.loads(
-                            (directory / "model-output.json").read_text()
+                            (directory / "model-output.json").read_text(
+                                encoding="utf-8"
+                            )
                         ),
                     },
                 )

loopx/control_plane/capabilities/trajectory_assessment.ts

@@ -1,5 +1,6 @@
 /** Benchmark-toolkit's observer sidecar; never a solver or native-score authority. */
 import {createHash} from 'node:crypto';
+import {BARE_SHA256_PATTERN} from '../content_digest.ts';
 import {EffectRuntimeRequestError} from '../effect_runtime_errors.ts';
 
 export const TRAJECTORY_METHOD = 'research-episodes-v2';
@@ -38,7 +39,7 @@ export function parseTrajectoryInput(raw: unknown): TrajectoryInput {
   const sources = list(x.sources, 'sources').map(rawSource => {
     const r = object(rawSource, 'source');
     const hash = str(r.artifact_sha256, 'artifact_sha256');
-    if (!/^[a-f0-9]{64}$/.test(hash)) throw new EffectRuntimeRequestError('artifact_sha256 must be a sha256 hex digest');
+    if (!BARE_SHA256_PATTERN.test(hash)) throw new EffectRuntimeRequestError('artifact_sha256 must be a sha256 hex digest');
     return {id: str(r.id, 'source.id'), artifact: str(r.artifact, 'artifact'), artifact_sha256: hash,
       first_line: integer(r.first_line, 'first_line', 1), text: str(r.text, 'source.text')};
   });

tests/architecture/test_content_digest_single_owner.py

@@ -161,6 +161,7 @@
     "loopx.capabilities.benchmark_toolkit.integrity",
     "loopx.capabilities.benchmark_toolkit.runtime_continuity",
     "loopx.capabilities.benchmark_toolkit.study_projection",
+    "loopx.capabilities.benchmark_toolkit.trajectory_assessment_store",
     "loopx.capabilities.content_ops.item_lifecycle",
     "loopx.capabilities.deep_research.runtime",
     "loopx.capabilities.issue_fix.outcome_projection",

tests/capabilities/test_trajectory_assessment.py

@@ -30,9 +30,9 @@ def prepared(tmp_path):
                     case_id="case",
                     run_id="run",
                     revision="r1",
-                    window="one episode",
+                    window="one episode · 观察窗口",
                 ),
-                "objective": "Improve native utility",
+                "objective": "Improve native utility · 原始目标",
                 "selection": dict(
                     rule="synthetic lifecycle fixture",
                     population_episodes=1,
@@ -44,7 +44,7 @@ def prepared(tmp_path):
                         artifact="result.txt",
                         artifact_sha256="a" * 64,
                         first_line=1,
-                        text="No utility measurement is available.",
+                        text="No utility measurement is available · 未测量。",
                     )
                 ],
                 "episodes": [
@@ -55,37 +55,50 @@ def prepared(tmp_path):
                         eligible_metrics=["benefit_test"],
                     )
                 ],
-            }
-        )
+            },
+            ensure_ascii=False,
+        ),
+        encoding="utf-8",
     )
     return prepare(path, model="fixture", effort="max", service_tier="fast")
 
 
-def provider(tmp_path: Path, *, delay=0, code=0, invalid=False) -> str:
+def provider(
+    tmp_path: Path, *, delay=0, code=0, invalid=False, invalid_utf8=False
+) -> str:
     script = tmp_path / "codex-fixture"
     output = {
         "assessments": [
             dict(
                 episode_id="episode",
                 metric="benefit_test",
                 verdict="unknown",
-                reason="Measurement is unavailable",
+                reason="Measurement is unavailable · 未测量",
                 citations=[],
             )
         ],
         "limitations": ["Synthetic provider; no model qualification"],
     }
     if invalid:
         output["assessments"] = []
-    script.write_text(f"""#!{sys.executable}
+    payload = (
+        b"\xff"
+        if invalid_utf8
+        else json.dumps(output, ensure_ascii=False).encode("utf-8")
+    )
+    script.write_text(
+        f"""#!{sys.executable}
 import json,sys,time
 from pathlib import Path
-sys.stdin.read()
+assert "\\u539f\\u59cb\\u76ee\\u6807" in sys.stdin.buffer.read().decode("utf-8")
 time.sleep({delay})
 if {code}: sys.exit({code})
-Path(sys.argv[sys.argv.index('-o')+1]).write_text({json.dumps(json.dumps(output))})
-print(json.dumps({{'type':'turn.completed','usage':{{'input_tokens':12,'output_tokens':3}}}}))
-""")
+Path(sys.argv[sys.argv.index('-o')+1]).write_bytes({payload!r})
+sys.stderr.buffer.write("\\u5408\\u6210\\u8bca\\u65ad".encode("utf-8"))
+sys.stdout.buffer.write((json.dumps({{'type':'turn.completed','message':'\\u5206\\u6790\\u5b8c\\u6210','usage':{{'input_tokens':12,'output_tokens':3}}}},ensure_ascii=False)+'\\n').encode("utf-8"))
+""",
+        encoding="utf-8",
+    )
     script.chmod(0o700)
     return str(script)
 
@@ -98,11 +111,13 @@ def test_actual_child_reuse_and_repair_of_derived_output(prepared, tmp_path):
     }
     assert first["served_service_tier"] is None
     directory = Path(first["directory"])
-    (directory / "assessment.json").write_text("{}")
+    (directory / "assessment.json").write_text("{}", encoding="utf-8")
     second = execute(prepared, tmp_path / "jobs", codex="must-not-run")
     assert second["reused"] is True
     assert (
-        json.loads((directory / "assessment.json").read_text())["metrics"][1]["unknown"]
+        json.loads((directory / "assessment.json").read_text(encoding="utf-8"))[
+            "metrics"
+        ][1]["unknown"]
         == 1
     )
     assert second["analysis_wall_seconds"] == first["analysis_wall_seconds"]
@@ -120,11 +135,36 @@ def test_failed_model_output_needs_explicit_retry_and_preserves_failure(
     success = execute(prepared, tmp_path / "jobs", retry=True, codex=provider(tmp_path))
     assert success["status"] == "succeeded" and success["attempt"] == 2
     assert (
-        json.loads((directory / "attempt-1" / "receipt.json").read_text())["status"]
+        json.loads(
+            (directory / "attempt-1" / "receipt.json").read_text(encoding="utf-8")
+        )["status"]
         == "failed"
     )
 
 
+def test_invalid_utf8_output_fails_strictly_and_retry_preserves_bytes(
+    prepared, tmp_path
+):
+    with pytest.raises(UnicodeDecodeError):
+        execute(
+            prepared, tmp_path / "jobs", codex=provider(tmp_path, invalid_utf8=True)
+        )
+    directory = tmp_path / "jobs" / prepared["analysis_key"]
+    assert read_job(directory)["status"] == "failed"
+    assert not (directory / "execution.lock").exists()
+    assert not (directory / "assessment.json").exists()
+    assert execute(prepared, tmp_path / "jobs", codex="must-not-run")["ok"] is False
+    assert (
+        execute(prepared, tmp_path / "jobs", retry=True, codex=provider(tmp_path))[
+            "status"
+        ]
+        == "succeeded"
+    )
+    assert (directory / "attempt-1" / "model-output.json").read_bytes() == b"\xff"
+    result = json.loads((directory / "assessment.json").read_text(encoding="utf-8"))
+    assert result["assessments"][0]["reason"] == "Measurement is unavailable · 未测量"
+
+
 def test_concurrent_execution_and_cancel_do_not_signal_stored_pid(prepared, tmp_path):
     results = []
     worker = threading.Thread(
@@ -170,6 +210,7 @@ def test_preview_cli_is_read_only_and_renders_json(prepared, tmp_path):
         ],
         capture_output=True,
         text=True,
+        encoding="utf-8",
     )
     assert result.returncode == 0, result.stderr
     assert json.loads(result.stdout)["status"] == "preview"