From e65b0c74147bb43ac2fc120aea318cb34627212a Mon Sep 17 00:00:00 2001 From: JasonBuildAI Date: Sat, 17 Oct 2026 12:12:15 +0800 Subject: [PATCH 1/3] fix(turn-driver): pin UTF-8 on the Codex session evidence I/O The managed Codex session observed the CLI exec JSONL through Path.open() without a codec, so a host whose default encoding is not UTF-8 (cp936 on zh-CN Windows, or an ASCII locale on POSIX) decoded the UTF-8 event stream with the locale codec. The read raised UnicodeDecodeError at the first non-ASCII event, the thread.started identity was never observed, and the Turn failed with "Codex did not confirm the expected session identity", which the class contract refuses to repair with a fresh conversation. Pin UTF-8 on that read and on the two descriptor writes whose JSON is read back with json.loads(read_bytes()): the checkpoint-refresh attempt record and the native fork attempt record, whose basis carries the guest workspace and home paths. The shipped file gate did not inspect os.fdopen(), so those writes were invisible to it; the gate now reports text-mode os.fdopen() without a UTF-8 codec, matching the sites that already pin it. Fixes #6708 Signed-off-by: JasonBuildAI --- .../turn_driver/codex_agent_session.py | 5 +- loopx/hosts/codex_checkpoint.py | 3 +- .../test_codex_session_utf8_evidence.py | 136 ++++++++++++++++++ tests/test_loopx_text_io_utf8.py | 39 ++++- 4 files changed, 173 insertions(+), 10 deletions(-) create mode 100644 tests/control_plane/test_codex_session_utf8_evidence.py 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..ac80804bcb --- /dev/null +++ b/tests/control_plane/test_codex_session_utf8_evidence.py @@ -0,0 +1,136 @@ +"""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 +ASCII, with a workspace whose path carries a non-ASCII component: the refresh +record keeps the observed stdout path and the binding's checkpoint root, so that +workspace shape is the ordinary case rather than a synthetic one. +""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +import subprocess +import sys + +REPO_ROOT = Path(__file__).resolve().parents[2] + +WORKSPACE_COMPONENT = "项目" + +_CHILD = r''' +import json, locale, sys +from pathlib import Path +from types import SimpleNamespace + +locale.setlocale(locale.LC_CTYPE, "C") +assert sys.flags.utf8_mode == 0 +assert locale.getpreferredencoding(False).lower().replace("-", "") != "utf8" + +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): + # The Codex CLI writes its event stream as UTF-8 whatever the host codec is. + path.write_bytes(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", + ])) + + +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": locale.getpreferredencoding(False), **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", "-c", _CHILD, str(base), mode], + cwd=REPO_ROOT, + env=environment, + 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. + assert payload["encoding"].lower().replace("-", "") != "utf8" + return payload + + +def test_codex_events_are_read_back_with_a_locale_codec(tmp_path: Path) -> None: + base = tmp_path / WORKSPACE_COMPONENT + payload = _payload(base, "observe") + + assert payload["session_id"] == "fresh" + + +def test_checkpoint_refresh_attempt_record_is_readable_again(tmp_path: Path) -> None: + base = tmp_path / WORKSPACE_COMPONENT + 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"] From 38371c513bf3406ad0bed9fbb0dcd68d42a38e79 Mon Sep 17 00:00:00 2001 From: JasonBuildAI Date: Sat, 17 Oct 2026 12:44:56 +0800 Subject: [PATCH 2/3] test(turn-driver): refuse a vacuous pass in the locale evidence case The non-UTF-8 default-codec case asserted only that the host codec is not UTF-8, which does not prove the CLI's UTF-8 bytes are rejected: a single-byte codec such as latin-1 or cp1251 decodes the sample without raising, so the defective read returns mojibake and both cases would pass with the codec pin reverted. The child now decodes the exact bytes it writes and reports the result, and the case skips with the resolved codec named where the hazard cannot be shown here. Also state per-case coverage: the recovery case exercises the observe read and the attempt record's byte read, while the two os.fdopen write pins are protected by the gate rule alone, because json.dump escapes non-ASCII by default and their payload is ASCII today. Verified by reverting both pins: this suite stays green and the gate reports exactly codex_agent_session.py:173 and codex_checkpoint.py:415. Signed-off-by: JasonBuildAI --- .../test_codex_session_utf8_evidence.py | 70 ++++++++++++++++--- 1 file changed, 60 insertions(+), 10 deletions(-) diff --git a/tests/control_plane/test_codex_session_utf8_evidence.py b/tests/control_plane/test_codex_session_utf8_evidence.py index ac80804bcb..647713755d 100644 --- a/tests/control_plane/test_codex_session_utf8_evidence.py +++ b/tests/control_plane/test_codex_session_utf8_evidence.py @@ -7,9 +7,16 @@ host locale codec from rewriting them. Each case drives one real owner in a child interpreter whose default encoding is -ASCII, with a workspace whose path carries a non-ASCII component: the refresh +not UTF-8 (the host ANSI code page on Windows, `ascii` under `LC_ALL=C` on +POSIX), with a workspace whose path carries a non-ASCII component: the refresh record keeps the observed stdout path and the binding's checkpoint root, so that workspace shape is the ordinary case rather than a synthetic one. + +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 @@ -20,6 +27,8 @@ import subprocess import sys +import pytest + REPO_ROOT = Path(__file__).resolve().parents[2] WORKSPACE_COMPONENT = "项目" @@ -29,9 +38,29 @@ 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 locale.getpreferredencoding(False).lower().replace("-", "") != "utf8" +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 ( @@ -61,12 +90,7 @@ def session(root): def events(path, identity): - # The Codex CLI writes its event stream as UTF-8 whatever the host codec is. - path.write_bytes(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", - ])) + path.write_bytes(events_bytes(identity)) root = base / mode / "checkout" @@ -90,7 +114,8 @@ def events(path, identity): "attempt_stdout": json.loads(attempt.read_bytes())["stdout"], "retry_launched": retry.exists()} -print(json.dumps({"encoding": locale.getpreferredencoding(False), **payload}, ensure_ascii=True)) +print(json.dumps({"encoding": _codec, "locale_hazard": locale_hazard, **payload}, + ensure_ascii=True)) ''' @@ -115,12 +140,27 @@ 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. + # 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 / WORKSPACE_COMPONENT payload = _payload(base, "observe") @@ -128,6 +168,16 @@ def test_codex_events_are_read_back_with_a_locale_codec(tmp_path: Path) -> None: 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 / WORKSPACE_COMPONENT payload = _payload(base, "recover") From e1da32c1cb9409ece520e5b8d8265a10338dace9 Mon Sep 17 00:00:00 2001 From: JasonBuildAI Date: Sat, 17 Oct 2026 15:49:54 +0800 Subject: [PATCH 3/3] test(turn-driver): run the locale evidence child from stdin Both cases passed the child program as a `-c` argument. CPython decodes `-c` text with the filesystem encoding, which this fixture forces to ASCII (`LC_ALL=C`, `-X utf8=0`, `PYTHONCOERCECLOCALE=0`), so on Linux the child died before importing LoopX: `Unable to decode the command from the command line` / `UnicodeEncodeError: ... surrogates not allowed`. The Windows lane passed only because its filesystem encoding is UTF-8. The program now travels on stdin, which CPython decodes as UTF-8 and which keeps the working directory on `sys.path` the way `-c` does. The workspace is the pytest temporary directory rather than a path with a non-ASCII component: the child addresses it as `str` under that ASCII filesystem encoding, so such a path would fail on the surrogate escape before the codec hazard this suite exists to reproduce. The non-ASCII payload stays in the CLI's UTF-8 event bytes and in the attempt record, so the codec assertions are unchanged. Reported by the independent review of a786ed7cb. Reproduced on Linux Python 3.12: a786ed7cb fails both cases at the command-line decode; this head passes both, and reverting the observe codec pin still fails both with `UnicodeDecodeError: 'ascii' codec can't decode byte 0xe4 in position 36`. Signed-off-by: JasonBuildAI --- .../test_codex_session_utf8_evidence.py | 28 +++++++++++++------ 1 file changed, 20 insertions(+), 8 deletions(-) diff --git a/tests/control_plane/test_codex_session_utf8_evidence.py b/tests/control_plane/test_codex_session_utf8_evidence.py index 647713755d..d87acff902 100644 --- a/tests/control_plane/test_codex_session_utf8_evidence.py +++ b/tests/control_plane/test_codex_session_utf8_evidence.py @@ -8,9 +8,18 @@ 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), with a workspace whose path carries a non-ASCII component: the refresh -record keeps the observed stdout path and the binding's checkpoint root, so that -workspace shape is the ordinary case rather than a synthetic one. +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 @@ -31,8 +40,6 @@ REPO_ROOT = Path(__file__).resolve().parents[2] -WORKSPACE_COMPONENT = "项目" - _CHILD = r''' import json, locale, sys from pathlib import Path @@ -128,9 +135,14 @@ def _drive(base: Path, mode: str) -> subprocess.CompletedProcess[bytes]: "PYTHONUTF8": "0", } return subprocess.run( - [sys.executable, "-X", "utf8=0", "-c", _CHILD, str(base), mode], + [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, ) @@ -161,7 +173,7 @@ def test_codex_events_are_read_back_with_a_locale_codec(tmp_path: Path) -> None: makes this case fail with the reproduced `UnicodeDecodeError`. """ - base = tmp_path / WORKSPACE_COMPONENT + base = tmp_path payload = _payload(base, "observe") assert payload["session_id"] == "fresh" @@ -178,7 +190,7 @@ def test_checkpoint_refresh_attempt_record_is_readable_again(tmp_path: Path) -> `tests/test_loopx_text_io_utf8.py` is what protects them. """ - base = tmp_path / WORKSPACE_COMPONENT + base = tmp_path payload = _payload(base, "recover") assert payload["session_id"] == "resumed"