diff --git a/loopx/capabilities/benchmark_toolkit/trajectory_assessment.py b/loopx/capabilities/benchmark_toolkit/trajectory_assessment.py
index d1b1523a41..d7c7d38498 100644
--- a/loopx/capabilities/benchmark_toolkit/trajectory_assessment.py
+++ b/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,7 +225,9 @@ 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}
             )
@@ -228,7 +235,9 @@ def execute(
             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:
diff --git a/loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py b/loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py
index 0894d93043..aa430d5323 100644
--- a/loopx/capabilities/benchmark_toolkit/trajectory_assessment_store.py
+++ b/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"
+                            )
                         ),
                     },
                 )
diff --git a/loopx/control_plane/capabilities/trajectory_assessment.ts b/loopx/control_plane/capabilities/trajectory_assessment.ts
index 96f0e01e62..49a5abae5f 100644
--- a/loopx/control_plane/capabilities/trajectory_assessment.ts
+++ b/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')};
   });
diff --git a/tests/architecture/test_content_digest_single_owner.py b/tests/architecture/test_content_digest_single_owner.py
index e0a1f61faa..7c6a8d20ad 100644
--- a/tests/architecture/test_content_digest_single_owner.py
+++ b/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",
diff --git a/tests/capabilities/test_trajectory_assessment.py b/tests/capabilities/test_trajectory_assessment.py
index 3b40389f88..e94c0fc8fb 100644
--- a/tests/capabilities/test_trajectory_assessment.py
+++ b/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,13 +55,17 @@ 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": [
@@ -69,7 +73,7 @@ def provider(tmp_path: Path, *, delay=0, code=0, invalid=False) -> str:
                 episode_id="episode",
                 metric="benefit_test",
                 verdict="unknown",
-                reason="Measurement is unavailable",
+                reason="Measurement is unavailable · 未测量",
                 citations=[],
             )
         ],
@@ -77,15 +81,24 @@ def provider(tmp_path: Path, *, delay=0, code=0, invalid=False) -> str:
     }
     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"
diff --git a/tests/capabilities/test_trajectory_assessment_store.py b/tests/capabilities/test_trajectory_assessment_store.py
index 7bdfa4b0fe..130fe6eaf4 100644
--- a/tests/capabilities/test_trajectory_assessment_store.py
+++ b/tests/capabilities/test_trajectory_assessment_store.py
@@ -4,6 +4,8 @@
 
 import hashlib
 import json
+import os
+import subprocess
 import sys
 
 import pytest
@@ -23,7 +25,9 @@
 def pack(tmp_path):
     sources = tmp_path / "sources"
     sources.mkdir()
-    evidence = b"Prior question\nThe native measurement did not improve utility.\n"
+    evidence = "Prior question\nThe native measurement did not improve utility · 原始测量。\n".encode(
+        "utf-8"
+    )
     (sources / "result.txt").write_bytes(evidence)
     data = {
         "schema_version": "trajectory_assessment_input_v1",
@@ -32,9 +36,9 @@ def pack(tmp_path):
             case_id="case",
             run_id="run",
             revision="fixed",
-            window="one episode",
+            window="one episode · 观察窗口",
         ),
-        "objective": "Improve native utility",
+        "objective": "Improve native utility · 原始目标",
         "selection": dict(
             rule="synthetic complete population", population_episodes=1, exhaustive=True
         ),
@@ -57,10 +61,10 @@ def pack(tmp_path):
         ],
     }
     input_path = tmp_path / "input.json"
-    input_path.write_text(json.dumps(data))
+    input_path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
     scope = dict(runtime_root=tmp_path / "runtime", owner_goal_id="research-owner")
     registered = register_pack(
-        **scope, input_path=input_path, source_root=sources, title="Case · one episode"
+        **scope, input_path=input_path, source_root=sources, title="Case · 原始目标"
     )
     operation = {
         **scope,
@@ -73,20 +77,24 @@ def pack(tmp_path):
                 episode_id="episode",
                 metric="benefit_test",
                 verdict="yes",
-                reason="A negative native test still tests benefit.",
+                reason="A negative native test still tests benefit · 负向测量。",
                 citations=[dict(source_id="source", first_line=2, last_line=2)],
             )
         ],
         "limitations": ["Synthetic process; not model quality evidence"],
     }
     executable = tmp_path / "codex-fixture"
-    executable.write_text(f"""#!{sys.executable}
+    executable.write_text(
+        f"""#!{sys.executable}
 import json,sys
 from pathlib import Path
-sys.stdin.read()
-Path(sys.argv[sys.argv.index('-o')+1]).write_text({json.dumps(json.dumps(output))})
-print(json.dumps({{'type':'turn.completed','usage':{{'input_tokens':10,'output_tokens':2}}}}))
-""")
+assert "\\u539f\\u59cb\\u76ee\\u6807" in sys.stdin.buffer.read().decode("utf-8")
+Path(sys.argv[sys.argv.index('-o')+1]).write_bytes({json.dumps(output, ensure_ascii=False).encode("utf-8")!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':10,'output_tokens':2}}}},ensure_ascii=False)+'\\n').encode("utf-8"))
+""",
+        encoding="utf-8",
+    )
     executable.chmod(0o700)
     return scope, operation, registered, input_path, sources, str(executable)
 
@@ -112,10 +120,13 @@ def test_scope_discovery_execute_read_citation_and_changed_source_history(pack):
     assert "directory" not in view["job"] and "process_id" not in view["job"]
     assert execute_pack(**operation, codex="must-not-run")["reused"] is True
     citation = read_citation(**operation, source_id="source", first_line=2, last_line=2)
-    assert citation["text"] == "The native measurement did not improve utility."
+    assert (
+        citation["text"]
+        == "The native measurement did not improve utility · 原始测量。"
+    )
     with pytest.raises(PackUnavailable, match="citation_not_in_result"):
         read_citation(**operation, source_id="source", first_line=1, last_line=2)
-    (sources / "result.txt").write_text("changed")
+    (sources / "result.txt").write_text("changed", encoding="utf-8")
     assert discover(**scope)["packs"][0]["reason_code"] == "source_changed"
     with pytest.raises(PackUnavailable, match="source_changed"):
         execute_pack(**operation, codex="must-not-run")
@@ -132,6 +143,84 @@ def test_scope_discovery_execute_read_citation_and_changed_source_history(pack):
     )
 
 
+@pytest.mark.skipif(
+    os.name != "posix", reason="C locale coercion is a POSIX regression"
+)
+def test_utf8_pack_and_child_roundtrip_with_ascii_default_locale(pack):
+    scope, _, registered, input_path, sources, executable = pack
+    script = r"""
+import codecs, locale, sys
+from pathlib import Path
+from loopx.capabilities.benchmark_toolkit.trajectory_assessment import prepare, read_job
+from loopx.capabilities.benchmark_toolkit.trajectory_assessment_store import (
+    PackUnavailable, discover, execute_pack, inspect_pack, read_citation, register_pack,
+)
+assert sys.flags.utf8_mode == 0
+assert codecs.lookup(locale.getencoding()).name == "ascii"
+input_path, sources, runtime = map(Path, sys.argv[1:4])
+scope = dict(runtime_root=runtime, owner_goal_id="research-owner")
+prepared = prepare(input_path, model="gpt-6.1-sol", effort="max", service_tier="fast")
+assert prepared["input"]["objective"] == "Improve native utility · 原始目标"
+registered = register_pack(**scope, input_path=input_path, source_root=sources, title="Case · 原始目标")
+assert registered["pack_id"] == sys.argv[4]
+assert registered["analysis_key"] == prepared["analysis_key"]
+assert discover(**scope)["packs"][0]["title"] == "Case · 原始目标"
+assert inspect_pack(**scope)["job"] is None
+operation = dict(**scope, pack_id=registered["pack_id"], analysis_key=registered["analysis_key"])
+first = execute_pack(**operation, codex=sys.argv[5])
+assert first["status"] == "succeeded" and first["usage"] == dict(input_tokens=10, output_tokens=2)
+view = inspect_pack(**scope)
+assert view["result"]["assessments"][0]["reason"] == "A negative native test still tests benefit · 负向测量。"
+assert view["result"]["metrics"][1]["yes"] == 1
+assert "directory" not in view["job"] and "process_id" not in view["job"]
+assert execute_pack(**operation, codex="must-not-run")["reused"] is True
+directory = next(runtime.glob("benchmark/trajectory-assessments/*/jobs/*"))
+assert read_job(directory)["scope"]["window"] == "one episode · 观察窗口"
+assert "原始目标" in (directory / "prompt.txt").read_text(encoding="utf-8")
+assert "分析完成" in (directory / "events.jsonl").read_text(encoding="utf-8")
+assert (directory / "stderr.txt").read_text(encoding="utf-8") == "合成诊断"
+citation = dict(source_id="source", first_line=2, last_line=2)
+assert read_citation(**operation, **citation)["text"] == "The native measurement did not improve utility · 原始测量。"
+(sources / "result.txt").write_bytes(b"changed")
+assert discover(**scope)["packs"][0]["reason_code"] == "source_changed"
+try:
+    execute_pack(**operation, codex="must-not-run")
+except PackUnavailable as error:
+    assert str(error) == "source_changed"
+else:
+    raise AssertionError("Changed sources must not run a new analysis")
+assert inspect_pack(**scope)["result"] == view["result"]
+assert read_citation(**operation, **citation)["applicability"] == "historical"
+"""
+    result = subprocess.run(
+        [
+            sys.executable,
+            "-X",
+            "utf8=0",
+            "-c",
+            script.encode("ascii", "backslashreplace").decode("ascii"),
+            str(input_path),
+            str(sources),
+            str(scope["runtime_root"]),
+            registered["pack_id"],
+            executable,
+        ],
+        env={
+            **os.environ,
+            "LC_ALL": "C",
+            "LANG": "C",
+            "PYTHONUTF8": "0",
+            "PYTHONCOERCECLOCALE": "0",
+            "PYTHONIOENCODING": "ascii",
+        },
+        capture_output=True,
+        text=True,
+        encoding="utf-8",
+        timeout=60,
+    )
+    assert result.returncode == 0, result.stdout + result.stderr
+
+
 def test_owner_isolation_unknown_ids_and_keys_are_rejected(pack):
     scope, operation, _, _, _, _ = pack
     other = {**scope, "owner_goal_id": "other-owner"}
@@ -147,11 +236,48 @@ def test_owner_isolation_unknown_ids_and_keys_are_rejected(pack):
         cancel_pack(**{**operation, "analysis_key": "a" * 64})
 
 
+def test_digest_rejections_preserve_registration_and_create_no_job(pack):
+    scope, operation, _, _, _, _ = pack
+    root = scope["runtime_root"]
+    before = {
+        p.relative_to(root): p.read_bytes() for p in root.rglob("*") if p.is_file()
+    }
+    malformed = (
+        "a" * 63,
+        "a" * 65,
+        "A" * 64,
+        "g" * 64,
+        "sha256:" + "a" * 64,
+        " " + "a" * 64,
+        "a" * 64 + " ",
+        "\n" + "a" * 64,
+        "a" * 64 + "\n",
+        "a" * 64 + "\r\n",
+        "a" * 64 + "\0",
+        True,
+        64,
+        {"id": "a" * 64},
+        ["a" * 64],
+    )
+    for value in malformed:
+        with pytest.raises(PackUnavailable, match="^invalid_identifier$"):
+            inspect_pack(**scope, pack_id=value)
+    for value in (*malformed, "", None, False, 0, {}, []):
+        with pytest.raises(PackUnavailable, match="^invalid_identifier$"):
+            execute_pack(**{**operation, "analysis_key": value}, codex="must-not-run")
+    for value in ("0" * 64, "0123456789abcdef" * 4, "f" * 64):
+        with pytest.raises(PackUnavailable, match="^pack_not_registered$"):
+            inspect_pack(**scope, pack_id=value)
+    assert {
+        p.relative_to(root): p.read_bytes() for p in root.rglob("*") if p.is_file()
+    } == before
+
+
 def test_registration_requires_real_hash_literal_excerpt_and_contained_path(
     pack, tmp_path
 ):
     scope, _, _, input_path, sources, _ = pack
-    original = json.loads(input_path.read_text())
+    original = json.loads(input_path.read_text(encoding="utf-8"))
     for change, message in (
         ({"text": "invented"}, "excerpt_mismatch"),
         ({"artifact_sha256": "a" * 64}, "source_changed"),
@@ -159,16 +285,16 @@ def test_registration_requires_real_hash_literal_excerpt_and_contained_path(
     ):
         data = json.loads(json.dumps(original))
         data["sources"][0].update(change)
-        input_path.write_text(json.dumps(data))
+        input_path.write_text(json.dumps(data), encoding="utf-8")
         with pytest.raises(PackUnavailable, match=message):
             register_pack(
                 **scope, input_path=input_path, source_root=sources, title="bad"
             )
     outside = tmp_path / "outside.txt"
-    outside.write_text("outside")
+    outside.write_text("outside", encoding="utf-8")
     (sources / "link.txt").symlink_to(outside)
     original["sources"][0]["artifact"] = "link.txt"
-    input_path.write_text(json.dumps(original))
+    input_path.write_text(json.dumps(original), encoding="utf-8")
     with pytest.raises(PackUnavailable, match="invalid_source_path"):
         register_pack(**scope, input_path=input_path, source_root=sources, title="bad")
     assert len(discover(**scope)["packs"]) == 1
@@ -224,6 +350,6 @@ def upgraded_method(record):
             "benchmark/trajectory-assessments/*/jobs/*/assessment.json"
         )
     )
-    result_file.write_text("{}")
+    result_file.write_text("{}", encoding="utf-8")
     with pytest.raises(PackUnavailable, match="historical_result_unverified"):
         inspect_pack(**scope)
diff --git a/tests/control_plane_ts/content_digest_single_owner.test.ts b/tests/control_plane_ts/content_digest_single_owner.test.ts
index 89abeb040f..f14229be3c 100644
--- a/tests/control_plane_ts/content_digest_single_owner.test.ts
+++ b/tests/control_plane_ts/content_digest_single_owner.test.ts
@@ -95,6 +95,7 @@ const CANONICAL_CONSUMERS = [
   "control_plane/capabilities/explore_result_page.ts",
   "control_plane/capabilities/external_evidence.ts",
   "control_plane/capabilities/goal_capability_organization.ts",
+  "control_plane/capabilities/trajectory_assessment.ts",
   "control_plane/collaboration/chat_mode.ts",
   "control_plane/collaboration/delegation.ts",
   "control_plane/collaboration/delegation_result_use.ts",
diff --git a/tests/control_plane_ts/trajectory_assessment.test.ts b/tests/control_plane_ts/trajectory_assessment.test.ts
index aa47416727..9c04537640 100644
--- a/tests/control_plane_ts/trajectory_assessment.test.ts
+++ b/tests/control_plane_ts/trajectory_assessment.test.ts
@@ -14,6 +14,25 @@ const output = () => ({assessments:[
   {episode_id:'e2',metric:'information_gain',verdict:'unknown',reason:'No valid measurement',citations:[]}],limitations:['Synthetic contract test, not evaluator calibration']});
 const prepared = () => prepareTrajectoryAssessment({input:input(), config});
 
+test('artifact digests retain the exact bare SHA-256 boundary', () => {
+  for (const hash of ['0'.repeat(64), '0123456789abcdef'.repeat(4), 'f'.repeat(64)]) {
+    const x = input(); x.sources[0]!.artifact_sha256 = hash;
+    assert.equal(prepareTrajectoryAssessment({input:x, config}).input.sources[0]!.artifact_sha256, hash);
+  }
+  for (const hash of ['a'.repeat(63), 'a'.repeat(65), 'A'.repeat(64), 'g'.repeat(64),
+    'sha256:' + 'a'.repeat(64), ' ' + 'a'.repeat(64), 'a'.repeat(64) + ' ',
+    '\n' + 'a'.repeat(64), 'a'.repeat(64) + '\n', 'a'.repeat(64) + '\r\n']) {
+    const x = input(); x.sources[0]!.artifact_sha256 = hash;
+    assert.throws(() => prepareTrajectoryAssessment({input:x, config}),
+      {message:'artifact_sha256 must be a sha256 hex digest'});
+  }
+  for (const hash of [null, undefined, true, 64, {}, [], '', ' ', 'a'.repeat(64) + '\0']) {
+    const x = input(); Object.assign(x.sources[0]!, {artifact_sha256:hash});
+    assert.throws(() => prepareTrajectoryAssessment({input:x, config}),
+      {message:'artifact_sha256 must be nonempty text'});
+  }
+});
+
 test('known denominator retains unknown and produces missing-data bounds, not a confidence interval', () => {
   const result=qualifyTrajectoryAssessment({...prepared(),output:output()});
   assert.deepEqual(result.metrics.find(m=>m.metric==='information_gain'),{metric:'information_gain',yes:1,no:0,unknown:1,eligible:2,assessed_fraction:.5,missing_data_bounds:[.5,1]});