fix(runtime): pin UTF-8 on host recovery I/O

loopx/agy_goal_mode/host_runtime.py

@@ -17,6 +17,7 @@
 import time
 from typing import Any
 
+from ..entrypoint import pin_utf8_stdio
 from ..file_lock import exclusive_file_lock
 
 
@@ -66,6 +67,7 @@ def _admitted(args: argparse.Namespace) -> bool:
         _core_args(args, ["quota", "should-run"]),
         capture_output=True,
         text=True,
+        encoding="utf-8",
         timeout=10,
         check=True,
         env={key: value for key, value in os.environ.items() if key != "PYTHONPATH"},
@@ -94,7 +96,7 @@ def _state(args: argparse.Namespace):
     path = Path(args.state_dir).expanduser() / (key + ".json")
     path.parent.mkdir(parents=True, exist_ok=True)
     with exclusive_file_lock(path, timeout_seconds=2):
-        data = json.loads(path.read_text()) if path.exists() else {}
+        data = json.loads(path.read_text(encoding="utf-8")) if path.exists() else {}
         yield data
         temporary = path.with_suffix(".tmp")
         with temporary.open("w", encoding="utf-8") as stream:
@@ -200,6 +202,7 @@ def stop_hook(args: argparse.Namespace, event: dict[str, Any]) -> dict[str, Any]
 
 
 def main() -> int:
+    pin_utf8_stdio()
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("operation", choices=("stop-hook",))
     parser.add_argument("--cli-bin", default="loopx")

loopx/hosts/managed_service_runtime.py

@@ -203,7 +203,7 @@ def align_managed_runtime(executable: str) -> dict:
 def _recover_alignment(journal: Path) -> None:
     if not journal.exists():
         return
-    saved = json.loads(journal.read_text())
+    saved = json.loads(journal.read_text(encoding="utf-8"))
     if not isinstance(saved, dict) or saved.get("schema_version") != "loopx_managed_runtime_handoff_v1":
         raise ValueError("service_runtime_route_unverified")
     entries = saved.get("services")
@@ -256,7 +256,7 @@ def _align_managed_runtime(executable: str, journal: Path) -> dict:
     environment = {k: v for k, v in os.environ.items()
         if k not in {"LOOPX_PYTHON", "PYTHONPATH", "LOOPX_RELEASE_ROOT"}}
     doctor = subprocess.run([str(target), "--format", "json", "doctor", "--installation-only"],
-        env=environment, capture_output=True, text=True, timeout=15)
+        env=environment, capture_output=True, text=True, encoding="utf-8", timeout=15)
     qualified = json.loads(doctor.stdout)
     if not isinstance(qualified, dict):
         raise ValueError("service_runtime_unverified")

tests/test_antigravity_host_runtime.py

@@ -2,6 +2,7 @@
 
 from argparse import Namespace
 import json
+import os
 from pathlib import Path
 import shlex
 import subprocess
@@ -200,6 +201,81 @@ def test_generated_native_configs_execute_the_stdio_hook_from_another_directory(
     assert json.loads(result.stdout) == {}
 
 
+def test_stop_hook_recovers_a_non_ascii_event_under_a_locale_codec(tmp_path):
+    expected = "继续授权的工作"
+    script = f'''
+import sys
+from loopx.agy_goal_mode import host_runtime
+
+def native_input(args, text):
+    return ("frozen-input", "RUNNING", 1) if text == {expected!r} else None
+
+host_runtime._native_input = native_input
+host_runtime._admitted = lambda args: True
+sys.argv = [
+    "host_runtime", "stop-hook",
+    "--goal-id", "work", "--agent-id", "worker",
+    "--conversation-id", "native-chat", "--state-dir", sys.argv[1],
+]
+raise SystemExit(host_runtime.main())
+'''
+    payload = json.dumps(event(lastUserInput=expected), ensure_ascii=False).encode("utf-8")
+    environment = {
+        **os.environ,
+        "PYTHONIOENCODING": "gbk",
+        "PYTHONUTF8": "0",
+    }
+    result = subprocess.run(
+        [sys.executable, "-c", script, str(tmp_path / "recovery")],
+        input=payload,
+        cwd=Path(__file__).resolve().parents[1],
+        env=environment,
+        capture_output=True,
+        timeout=10,
+    )
+
+    assert result.returncode == 0, result.stderr.decode("utf-8", "replace")
+    assert json.loads(result.stdout.decode("utf-8"))["decision"] == "continue"
+    assert result.stderr == b""
+
+
+def test_admission_decodes_utf8_core_output_under_a_locale_codec(tmp_path):
+    fake_cli = tmp_path / "fake_core.py"
+    fake_cli.write_text(
+        "import json, sys\n"
+        "sys.stdout.buffer.write(json.dumps("
+        "{'should_run': True, 'note': '继续授权的工作'}, ensure_ascii=False"
+        ").encode('utf-8'))\n",
+        encoding="utf-8",
+    )
+    script = '''
+import argparse
+import sys
+from loopx.agy_goal_mode import host_runtime
+
+host_runtime._core_args = lambda args, command: [sys.executable, sys.argv[1]]
+assert host_runtime._admitted(argparse.Namespace()) is True
+'''
+    environment = {
+        **os.environ,
+        "PYTHONIOENCODING": "gbk",
+        "PYTHONUTF8": "0",
+        "PYTHONCOERCECLOCALE": "0",
+        "LC_ALL": "C",
+        "LANG": "C",
+    }
+    result = subprocess.run(
+        [sys.executable, "-c", script, str(fake_cli)],
+        cwd=Path(__file__).resolve().parents[1],
+        env=environment,
+        capture_output=True,
+        timeout=10,
+    )
+
+    assert result.returncode == 0, result.stderr.decode("utf-8", "replace")
+    assert result.stdout == b""
+
+
 def test_native_hook_admission_uses_real_core_ready_user_gate_and_zero_quota(tmp_path):
     from examples.control_plane.quota_plan_fixtures import (
         SCOPED_AGENT_ID,

tests/test_managed_service_runtime.py

@@ -165,6 +165,49 @@ def test_recovery_rejects_a_journal_with_changed_role_or_data_route(managed):
     assert path.read_bytes() == original and journal.exists()
 
 
+def test_recovery_reads_a_utf8_journal_under_a_locale_codec(tmp_path):
+    journal = tmp_path / "handoff.json"
+    journal.write_text(
+        json.dumps(
+            {
+                "schema_version": "loopx_managed_runtime_handoff_v1",
+                "note": "继续授权的工作",
+                "services": {},
+            },
+            ensure_ascii=False,
+        ),
+        encoding="utf-8",
+    )
+    script = '''
+import sys
+from pathlib import Path
+from loopx.hosts.managed_service_runtime import _recover_alignment
+
+try:
+    _recover_alignment(Path(sys.argv[1]))
+except ValueError as exc:
+    raise SystemExit(0 if str(exc) == "service_runtime_route_unverified" else 1)
+raise SystemExit(2)
+'''
+    environment = {
+        **os.environ,
+        "PYTHONIOENCODING": "gbk",
+        "PYTHONUTF8": "0",
+        "PYTHONCOERCECLOCALE": "0",
+        "LC_ALL": "C",
+        "LANG": "C",
+    }
+    result = subprocess.run(
+        [sys.executable, "-c", script, str(journal)],
+        cwd=Path(__file__).resolve().parents[1],
+        env=environment,
+        capture_output=True,
+        timeout=10,
+    )
+
+    assert result.returncode == 0, result.stderr.decode("utf-8", "replace")
+
+
 def test_failed_second_bootstrap_restores_original_loaded_pair(managed):
     _, agents, target, loaded, calls, fail = managed
     before = {p: p.read_bytes() for p in agents.glob("*.plist")}