diff --git a/loopx/agy_goal_mode/host_runtime.py b/loopx/agy_goal_mode/host_runtime.py index 1e34082547..cd49631876 100644 --- a/loopx/agy_goal_mode/host_runtime.py +++ b/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") diff --git a/loopx/hosts/managed_service_runtime.py b/loopx/hosts/managed_service_runtime.py index 67fc70bfcf..9515c973a5 100644 --- a/loopx/hosts/managed_service_runtime.py +++ b/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") diff --git a/tests/test_antigravity_host_runtime.py b/tests/test_antigravity_host_runtime.py index fe79173ac0..04d362fce8 100644 --- a/tests/test_antigravity_host_runtime.py +++ b/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, diff --git a/tests/test_managed_service_runtime.py b/tests/test_managed_service_runtime.py index 85602e670f..cac39a2c04 100644 --- a/tests/test_managed_service_runtime.py +++ b/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")}