diff --git a/loopx/control_plane/turn_driver/codex_agent_session.py b/loopx/control_plane/turn_driver/codex_agent_session.py index aa7feb8ddb..07680bf078 100644 --- a/loopx/control_plane/turn_driver/codex_agent_session.py +++ b/loopx/control_plane/turn_driver/codex_agent_session.py @@ -92,7 +92,8 @@ def observe(self, path: Path) -> None: if not path.exists(): return # No process was launched. ids = set() - with path.open() as stream: + # The Codex CLI event stream is UTF-8 JSONL; the host default codec is not. + with path.open(encoding="utf-8") as stream: for line in stream: try: event = json.loads(line) @@ -169,7 +170,7 @@ def prepare_checkpoint_refresh(self, review_basis: dict | None, attempt: Path, s return attempt.parent.mkdir(parents=True, exist_ok=True) descriptor = os.open(attempt, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) - with os.fdopen(descriptor, "w") as stream: + with os.fdopen(descriptor, "w", encoding="utf-8") as stream: json.dump({"basis": basis, "stdout": str(stdout)}, stream) stream.flush() os.fsync(stream.fileno()) diff --git a/loopx/hosts/codex_checkpoint.py b/loopx/hosts/codex_checkpoint.py index 2acab8852b..4692ab1027 100644 --- a/loopx/hosts/codex_checkpoint.py +++ b/loopx/hosts/codex_checkpoint.py @@ -412,7 +412,8 @@ def request(method, params): attempt = {"basis": basis, "source_thread_ids": sorted(inventory)} # Persist before dispatch. Partial preparation is not a license to # fork again; a missing outcome stays explicitly unresolved. - with os.fdopen(os.open(attempt_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), "w") as file: + with os.fdopen(os.open(attempt_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), "w", + encoding="utf-8") as file: json.dump(attempt, file, sort_keys=True) file.flush() os.fsync(file.fileno()) diff --git a/tests/control_plane/test_codex_session_utf8_evidence.py b/tests/control_plane/test_codex_session_utf8_evidence.py new file mode 100644 index 0000000000..d87acff902 --- /dev/null +++ b/tests/control_plane/test_codex_session_utf8_evidence.py @@ -0,0 +1,198 @@ +"""Codex session evidence must survive a host default encoding that is not UTF-8. + +The shipped Turn driver reads the Codex CLI's UTF-8 JSONL event stream and +persists its checkpoint-refresh attempt record as UTF-8 JSON. Both go through +`Path.open()` / `os.fdopen()`, which the shipped file gate +(`tests/test_loopx_text_io_utf8.py`) does not inspect, so nothing stopped the +host locale codec from rewriting them. + +Each case drives one real owner in a child interpreter whose default encoding is +not UTF-8 (the host ANSI code page on Windows, `ascii` under `LC_ALL=C` on +POSIX). The child's workspace stays ASCII: the forced filesystem encoding must be +able to decode the `str` paths the owner hands to file I/O, so a non-ASCII +workspace path would fail on the surrogate escape first and hide the codec +hazard this suite exists to reproduce. The non-ASCII payload is the CLI's own +UTF-8 event bytes and the attempt record, which is what the codecs under test +decode. + +The program travels on stdin rather than as a `-c` argument: `-c` text is +decoded with the filesystem encoding, so under `LC_ALL=C` CPython rejects the +program before it imports LoopX ("Unable to decode the command from the command +line"), while source read from stdin is decoded as UTF-8. `-` keeps the working +directory on `sys.path` the way `-c` does. + +The child also proves the case is not vacuous: it decodes the exact bytes the +CLI writes with its resolved default codec and reports whether they are +rejected. A single-byte codec such as `latin-1` or `cp1251` accepts those bytes, +so on such a host the read succeeds with mojibake instead of raising and this +regression cannot be shown; the case skips there rather than claim coverage. +""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +import subprocess +import sys + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[2] + +_CHILD = r''' +import json, locale, sys +from pathlib import Path +from types import SimpleNamespace + + +def events_bytes(identity): + # The Codex CLI writes its event stream as UTF-8 whatever the host codec is. + return b"".join([ + json.dumps({"type": "item.completed", "text": "任务已完成"}, + ensure_ascii=False).encode("utf-8") + b"\n", + json.dumps({"type": "thread.started", "thread_id": identity}).encode("utf-8") + b"\n", + ]) + + +locale.setlocale(locale.LC_CTYPE, "C") +_codec = locale.getpreferredencoding(False) +assert sys.flags.utf8_mode == 0 +assert _codec.lower().replace("-", "") != "utf8" +try: + events_bytes("locale-probe").decode(_codec) +except UnicodeDecodeError: + locale_hazard = True +else: + # A single-byte codec decodes the sample without raising, so the defective + # default-codec read would return mojibake instead of failing. Say so + # instead of letting the case pass for the wrong reason. + locale_hazard = False + +from loopx.control_plane.turn_driver.codex_agent_session import CodexAgentSession +from loopx.control_plane.turn_driver.codex_sessions import ( + _store_codex_cli_session, load_codex_cli_session, +) + +base, mode = Path(sys.argv[1]), sys.argv[2] +review = {"identity": {"effect_id": "current-review"}, "decision_sha256": "c" * 64} + + +def session(root): + subject = CodexAgentSession.__new__(CodexAgentSession) + subject.root = root + subject.lineage = {"goal_id": "destination", "agent_id": "worker"} + subject.goal_ref = {"goal_id": "destination", "instance_id": "current"} + subject.digest = "b" * 64 + _store_codex_cli_session(root, lineage=subject.lineage, session_scope="agent", + session_id="original", goal_ref=subject.goal_ref, session_profile_digest=subject.digest, + checkpoint_artifact_sha256="a" * 64, checkpoint_artifact_root=str(root)) + subject.admission = SimpleNamespace( + require_current=lambda: None, accept_result=lambda callback: callback()) + subject.binding = load_codex_cli_session(root, lineage=subject.lineage, session_scope="agent") + subject.session_id = "original" + subject.receipt = {"session": {"session_id": "original"}} + subject.checkpoint_refresh = None + return subject + + +def events(path, identity): + path.write_bytes(events_bytes(identity)) + + +root = base / mode / "checkout" +root.mkdir(parents=True) +subject = session(root) + +if mode == "observe": + # The refresh is prepared first, then the CLI's own UTF-8 events are read. + subject.prepare_checkpoint_refresh(review, root / "attempt.json", root / "stdout.jsonl") + events(root / "stdout.jsonl", "fresh") + subject.observe(root / "stdout.jsonl") + payload = {"session_id": subject.session_id} +else: + # The durable record carries the workspace and stdout paths; the shipped + # recovery read is `json.loads(attempt.read_bytes())`. + attempt, retry = root / "attempt.json", root / "retry.jsonl" + subject.prepare_checkpoint_refresh(review, attempt, root / "stdout.jsonl") + events(root / "stdout.jsonl", "resumed") + subject.prepare_checkpoint_refresh(review, attempt, retry) + payload = {"session_id": subject.session_id, + "attempt_stdout": json.loads(attempt.read_bytes())["stdout"], + "retry_launched": retry.exists()} + +print(json.dumps({"encoding": _codec, "locale_hazard": locale_hazard, **payload}, + ensure_ascii=True)) +''' + + +def _drive(base: Path, mode: str) -> subprocess.CompletedProcess[bytes]: + environment = { + **os.environ, + "LC_ALL": "C", + "PYTHONCOERCECLOCALE": "0", + "PYTHONIOENCODING": "utf-8", + "PYTHONUTF8": "0", + } + return subprocess.run( + [sys.executable, "-X", "utf8=0", "-", str(base), mode], + cwd=REPO_ROOT, + env=environment, + # `-c` arguments are decoded with the ASCII filesystem encoding this + # environment forces, which rejects a non-ASCII program before LoopX is + # imported; stdin source is decoded as UTF-8 and keeps the working + # directory on `sys.path`, as `-c` would. + input=_CHILD.encode("utf-8"), + capture_output=True, + timeout=180, + ) + + +def _payload(base: Path, mode: str) -> dict: + result = _drive(base, mode) + assert result.returncode == 0, result.stderr.decode("utf-8", "replace") + payload = json.loads(result.stdout.decode("utf-8")) + # A vacuous pass would hide the codec this test exists to exercise. The host + # default must not be UTF-8, and it must reject the exact bytes the CLI + # writes; a permissive single-byte codec decodes them without raising, so the + # defective read would return mojibake and the assertions below would hold + # with the fix reverted. + assert payload["encoding"].lower().replace("-", "") != "utf8" + if not payload["locale_hazard"]: + pytest.skip( + f"host default codec {payload['encoding']} decodes the UTF-8 event bytes; " + "the default-codec read hazard cannot be reproduced on this host" + ) + return payload + + +def test_codex_events_are_read_back_with_a_locale_codec(tmp_path: Path) -> None: + """The CLI's UTF-8 event stream survives a non-UTF-8 default codec. + + Covers the `CodexAgentSession.observe` read only: reverting its codec pin + makes this case fail with the reproduced `UnicodeDecodeError`. + """ + + base = tmp_path + payload = _payload(base, "observe") + + assert payload["session_id"] == "fresh" + + +def test_checkpoint_refresh_attempt_record_is_readable_again(tmp_path: Path) -> None: + """The recovery wake re-reads its original output and its attempt record. + + Covers the recovery path's reads — `prepare_checkpoint_refresh` re-runs the + `observe` read on the recorded stdout path, and the record itself is re-read + as UTF-8 bytes. It does not cover the two `os.fdopen` write pins: their + payload is ASCII today because `json.dump` escapes non-ASCII by default, so + reverting those pins leaves this case green. The gate rule in + `tests/test_loopx_text_io_utf8.py` is what protects them. + """ + + base = tmp_path + payload = _payload(base, "recover") + + assert payload["session_id"] == "resumed" + assert payload["attempt_stdout"] == str(base / "recover" / "checkout" / "stdout.jsonl") + assert payload["retry_launched"] is False diff --git a/tests/test_loopx_text_io_utf8.py b/tests/test_loopx_text_io_utf8.py index 1983f37aa2..6012a0a861 100644 --- a/tests/test_loopx_text_io_utf8.py +++ b/tests/test_loopx_text_io_utf8.py @@ -7,6 +7,10 @@ character either raises `UnicodeDecodeError` on read or writes bytes the next UTF-8 reader cannot decode. +`os.fdopen()` is the same text-mode API reached through a descriptor, and the +shipped runtime already pins UTF-8 on the sites that use it deliberately, so it +is checked here too. + #4338, #4942 and #4997 fixed the reported sites. This guard keeps the class from returning, next to `test_runtime_subprocess_utf8.py`, which does the same for text-mode subprocess reads. @@ -30,6 +34,10 @@ # call that does pin UTF-8. _PATH_HELPER_ENCODING_INDEX = {"read_text": 0, "write_text": 1} +# `os.fdopen(fd, mode, buffering, encoding, ...)` puts `encoding` third, while +# `open`/`os.fdopen` both take `mode` second. +_FDOPEN_ENCODING_INDEX = 3 + def _is_codec_name(value: str) -> bool: """Whether a string literal is a real codec name, as `pathlib` would accept it.""" @@ -69,6 +77,8 @@ def _pinned_encoding(node: ast.Call, attribute: str | None) -> str | None: return None if attribute is not None: index = _PATH_HELPER_ENCODING_INDEX.get(attribute) + if index is None and attribute == "fdopen": + index = _FDOPEN_ENCODING_INDEX if index is not None and len(node.args) > index: argument = node.args[index] if isinstance(argument, ast.Constant) and isinstance(argument.value, str): @@ -76,8 +86,18 @@ def _pinned_encoding(node: ast.Call, attribute: str | None) -> str | None: return None +def _is_fdopen(node: ast.Call) -> bool: + """Whether the call is `os.fdopen(...)`, the descriptor form of text `open()`.""" + + if isinstance(node.func, ast.Attribute): + return (node.func.attr == "fdopen" + and isinstance(node.func.value, ast.Name) + and node.func.value.id == "os") + return isinstance(node.func, ast.Name) and node.func.id == "fdopen" + + def _opens_in_text_mode(node: ast.Call) -> bool: - """Whether a builtin `open()` call is not provably binary.""" + """Whether an `open()`/`fdopen()` call is not provably binary.""" mode: str | None = None if len(node.args) >= 2: @@ -114,11 +134,12 @@ def _text_io_calls_without_utf8_encoding(package_root: Path | None = None) -> li attribute = node.func.attr if isinstance(node.func, ast.Attribute) else None is_path_helper = attribute in _PATH_HELPER_ENCODING_INDEX and _is_path_helper(node, attribute) is_builtin_open = isinstance(node.func, ast.Name) and node.func.id == "open" - if not (is_path_helper or is_builtin_open): + is_fdopen = _is_fdopen(node) + if not (is_path_helper or is_builtin_open or is_fdopen): continue - if is_builtin_open and not _opens_in_text_mode(node): + if (is_builtin_open or is_fdopen) and not _opens_in_text_mode(node): continue - if _pinned_encoding(node, attribute) in ALLOWED_ENCODINGS: + if _pinned_encoding(node, "fdopen" if is_fdopen else attribute) in ALLOWED_ENCODINGS: continue offenders.append(f"{path.relative_to(root.parent)}:{node.lineno}") return offenders @@ -131,7 +152,7 @@ def test_shipped_runtime_pins_utf8_for_every_text_file_read_and_write() -> None: def test_guard_reports_missing_and_non_utf8_codecs(tmp_path: Path) -> None: - """The guard covers a wrong explicit codec and builtin `open`, not only a missing keyword.""" + """The guard covers a wrong explicit codec and `open`/`fdopen`, not only a missing keyword.""" package = tmp_path / "loopx" package.mkdir() @@ -149,11 +170,15 @@ def test_guard_reports_missing_and_non_utf8_codecs(tmp_path: Path) -> None: " open(path, encoding='utf-8')\n" " open(path, 'rb')\n" " open(path)\n" - " installed.read_text('INSTALLER')\n", + " installed.read_text('INSTALLER')\n" + " os.fdopen(fd, 'w')\n" + " os.fdopen(fd, 'wb')\n" + " os.fdopen(fd, 'w', encoding='utf-8')\n" + " os.fdopen(fd, 'w', 0, 'utf-8')\n", encoding="utf-8", ) offenders = _text_io_calls_without_utf8_encoding(package) # Line 14 is a keyed lookup on a non-path object: same method name, not file I/O. - assert [entry.rsplit(":", 1)[-1] for entry in offenders] == ["5", "6", "7", "13"] + assert [entry.rsplit(":", 1)[-1] for entry in offenders] == ["5", "6", "7", "13", "15"]