From 0039c365de56fccc4afbe094daa0d776677694b5 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 15 Oct 2026 14:45:43 +0800 Subject: [PATCH 1/4] fix(canary): retire all owned runtime revisions before fixture cleanup Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- docs/development/testing-and-quality.md | 16 +++ loopx/canary/runner.py | 95 +++++++++----- loopx/control_plane/effect_runtime.py | 43 ++++++- tests/canary/test_telemetry_isolation.py | 118 ++++++++++++++++++ .../test_effect_runtime_integration.py | 39 ++++++ 5 files changed, 277 insertions(+), 34 deletions(-) diff --git a/docs/development/testing-and-quality.md b/docs/development/testing-and-quality.md index e8e808a3b8..988e71918d 100644 --- a/docs/development/testing-and-quality.md +++ b/docs/development/testing-and-quality.md @@ -406,6 +406,22 @@ Effect 服务先排空已接纳的 handler，再退出。启用编译缓存时 继续普通清理。一秒额度不涵盖定位文件锁、存储等待或操作系统调度；故障注入不能 代替原生平台资格验证，V8 CPU 样本也不测量内核 IO。 +Smoke cleanup owns every valid runtime locator in its newly created private +temporary namespace, including runtimes loaded by temporary base checkouts. +It uses the existing authenticated shutdown protocol, then observes actual +process exit before deleting the namespace; serving-locator turnover alone +does not establish that Node has finished its exit-time cache writes. Namespace +and locator symlinks are not followed. A cleanup failure retains the fixture +and reports a failed check with the original command output. Operator restart +keeps its existing serving-handoff semantics. This fixture repair does not +qualify native intermittent failures or sustained operation. + +Smoke 清理覆盖新建私有临时目录中所有合法 runtime locator，包括临时基线 checkout +启动的进程。它复用既有认证 shutdown，等实际进程退出后再删目录；定位文件退休不代表 +Node 的退出缓存写入已经结束。清理不跟随 namespace 或 locator 符号链接；无法证明 +退出时保留 fixture，以失败结果报告并保留原命令输出。操作员 restart 仍沿用既有服务 +交接语义；该 fixture 修复不认证原生间歇故障或持续运行。 + Keep a deterministic, public-safe production-scale fixture beside the focused cases. Its envelope should cover realistic role/status distributions, multi-agent claims, user gates and standing decisions, current and retired diff --git a/loopx/canary/runner.py b/loopx/canary/runner.py index f376a64b9a..febd2b8951 100644 --- a/loopx/canary/runner.py +++ b/loopx/canary/runner.py @@ -3,6 +3,7 @@ import os import re import shlex +import shutil import subprocess import sys import tempfile @@ -258,6 +259,35 @@ def _restore_tracked_paths(paths: list[str], repo_root: Path | None = None) -> d } +def _retire_fixture_runtimes(root: Path, *, timeout: float = 15) -> None: + """Stop all source revisions in this newly created private smoke root.""" + from ..control_plane.effect_runtime import _read_info, _stop_effect_runtime + + deadline = time.monotonic() + timeout + for namespace in sorted(root.glob("loopx-effect-runtime-*")): + if namespace.is_symlink() or not namespace.is_dir(): + continue + for path in sorted(namespace.glob("runtime-*.json")): + # Never follow fixture links into an unrelated runtime namespace. + if path.is_symlink() or not path.is_file(): + continue + # The namespace can contain a temporary base checkout's revision. + # Read and validate one snapshot before retirement removes it. + info = _read_info(path, fingerprint=None) + if info is None: + continue + if path.name != f"runtime-{info['fingerprint'][:16]}.json": + raise RuntimeError(f"invalid managed runtime fingerprint: {path.name}") + remaining = (deadline - time.monotonic()) / 2 + if remaining <= 0: + raise RuntimeError("managed runtime cleanup deadline exceeded") + result = _stop_effect_runtime( + path, info, timeout=min(5, remaining), wait_for_process_exit=True, + ) + if result["status"] != "stopped": + raise RuntimeError(f"managed runtime still exiting: {path.name}") + + def _run_check( check: dict[str, Any], *, @@ -280,9 +310,10 @@ def _run_check( return result started = time.monotonic() + cleanup_error: str | None = None + root = Path(tempfile.mkdtemp(prefix="loopx-smoke-")) try: - with tempfile.TemporaryDirectory(prefix="loopx-smoke-") as temporary: - root = Path(temporary) + try: home = root / "home" home.mkdir() env = {**os.environ, "LOOPX_USAGE_PING": "0", "HOME": str(home), @@ -297,35 +328,31 @@ def _run_check( ("OPENCODE_CONFIG_DIR", ".config/opencode"), ): env[name] = str(home / directory) + with subprocess.Popen( + normalized["argv"], cwd=repo_root, env=env, + text=True, encoding="utf-8", errors="replace", + stdout=subprocess.PIPE, stderr=subprocess.PIPE, + start_new_session=os.name == "posix", + ) as process: + try: + stdout, stderr = process.communicate(timeout=timeout_seconds) + except BaseException: + # Deadline/cancellation must stop the owned children + # before deleting their fixture, even after leader exit. + # Reuse OS transport; keep the caller's existing budget. + terminate_process_tree(process, grace_seconds=0) + raise + completed = subprocess.CompletedProcess( + normalized["argv"], process.returncode, stdout, stderr, + ) + finally: try: - with subprocess.Popen( - normalized["argv"], cwd=repo_root, env=env, - text=True, encoding="utf-8", errors="replace", - stdout=subprocess.PIPE, stderr=subprocess.PIPE, - start_new_session=os.name == "posix", - ) as process: - try: - stdout, stderr = process.communicate(timeout=timeout_seconds) - except BaseException: - # Deadline/cancellation must stop the owned children - # before deleting their fixture, even after leader exit. - # Reuse OS transport; keep the caller's existing budget. - terminate_process_tree(process, grace_seconds=0) - raise - completed = subprocess.CompletedProcess( - normalized["argv"], process.returncode, stdout, stderr, - ) - finally: - # Each fixture owns its Effect process as well as its data. - # Use the existing shutdown owner in the same temporary scope. - if any(root.glob("loopx-effect-runtime-*/runtime-*.json")): - subprocess.run( - [sys.executable, "-c", "from loopx.control_plane.effect_runtime " - "import restart_effect_runtime; r=restart_effect_runtime(); " - "raise SystemExit(r['status'] == 'shutdown_pending')"], - cwd=repo_root, env=env, check=True, timeout=15, - stdout=subprocess.PIPE, stderr=subprocess.PIPE, - ) + _retire_fixture_runtimes(root) + shutil.rmtree(root) + except (OSError, RuntimeError, ValueError) as error: + # Retain the owned namespace if shutdown cannot be proved. + # Do not mask the check with unconditional second deletion. + cleanup_error = f"fixture cleanup failed; retained {root}: {error}" except subprocess.TimeoutExpired as exc: result.update( { @@ -337,18 +364,22 @@ def _run_check( "stderr_tail": (exc.stderr or "")[-800:] if isinstance(exc.stderr, str) else "", } ) + if cleanup_error: + result["reason"] = cleanup_error return result result.update( { - "status": "passed" if completed.returncode == 0 else "failed", - "ok": completed.returncode == 0, + "status": "passed" if completed.returncode == 0 and not cleanup_error else "failed", + "ok": completed.returncode == 0 and not cleanup_error, "returncode": completed.returncode, "duration_seconds": round(time.monotonic() - started, 3), "stdout_tail": completed.stdout[-800:], "stderr_tail": completed.stderr[-800:], } ) + if cleanup_error: + result["reason"] = cleanup_error return result diff --git a/loopx/control_plane/effect_runtime.py b/loopx/control_plane/effect_runtime.py index 2d5d1ee104..a8a78ee68f 100644 --- a/loopx/control_plane/effect_runtime.py +++ b/loopx/control_plane/effect_runtime.py @@ -20,6 +20,7 @@ from typing import IO, Any from ..file_lock import process_is_alive +from ..extensions.process_runtime import _posix_owned_group_has_exited from .runtime.file_reads import iter_binary_file_reads from .runtime.node_probe import ( HOST_PERMISSION_RECOMMENDATION, @@ -447,7 +448,7 @@ def _start_lock_holder_pid(path: Path) -> int | None: return value if value > 0 else None -def _read_info(path: Path, *, fingerprint: str) -> dict[str, Any] | None: +def _read_info(path: Path, *, fingerprint: str | None) -> dict[str, Any] | None: for attempt in range(RUNTIME_LOCATOR_PERMISSION_RETRIES + 1): try: raw = path.read_text(encoding="utf-8") @@ -474,7 +475,11 @@ def _read_info(path: Path, *, fingerprint: str) -> dict[str, Any] | None: return None if ( payload.get("schema_version") != EFFECT_RUNTIME_INFO_SCHEMA_VERSION - or payload.get("fingerprint") != fingerprint + or (fingerprint is not None and payload.get("fingerprint") != fingerprint) + or (fingerprint is None and ( + not isinstance(payload.get("fingerprint"), str) + or not BARE_SHA256_PATTERN.fullmatch(payload["fingerprint"]) + )) or payload.get("host") != "127.0.0.1" or not isinstance(payload.get("port"), int) or not isinstance(payload.get("token"), str) @@ -598,6 +603,22 @@ def restart_effect_runtime(*, timeout: float = 5.0) -> dict[str, Any]: fingerprint = _runtime_fingerprint() info_path = _runtime_info_path(fingerprint) info = _read_info(info_path, fingerprint=fingerprint) + return _stop_effect_runtime(info_path, info, timeout=timeout) + + +def _stop_effect_runtime( + info_path: Path, + info: Mapping[str, Any] | None, + *, + timeout: float, + wait_for_process_exit: bool = False, +) -> dict[str, Any]: + """Transport shutdown for a validated locator, optionally awaiting OS exit. + + Operator restart keeps its existing serving-locator turnover semantics. + An isolated fixture instead needs the original process to stop writing + before deleting its namespace. TS still owns admission, drain and retirement. + """ identity = runtime_identity_from_info(info) if info is None: return { @@ -629,6 +650,24 @@ def restart_effect_runtime(*, timeout: float = 5.0) -> dict[str, Any]: deadline = time.monotonic() + timeout stopped = False while time.monotonic() < deadline: + if wait_for_process_exit: + _reap_exited_runtime_child(info) + if not _pid_is_alive(pid): + stopped = True + break + if os.name == "posix": + # A non-child zombie cannot write, but kill(pid, 0) still sees + # it. Managed runtimes start in their own session/process group. + try: + stopped = _posix_owned_group_has_exited( + pid, max(.001, deadline - time.monotonic()), + ) + except (OSError, subprocess.TimeoutExpired, UnicodeError, RuntimeError): + stopped = False + if stopped: + break + time.sleep(.05) + continue published, published_token = _serving_token(info_path) if published and published_token is not None and published_token != serving_token: stopped = True diff --git a/tests/canary/test_telemetry_isolation.py b/tests/canary/test_telemetry_isolation.py index 3acc3f9b30..7defb90a61 100644 --- a/tests/canary/test_telemetry_isolation.py +++ b/tests/canary/test_telemetry_isolation.py @@ -4,6 +4,7 @@ import subprocess import os import signal +import shutil import sys import time from pathlib import Path @@ -203,3 +204,120 @@ def test_runtime_started_by_smoke_is_stopped_before_fixture_cleanup(tmp_path, mo "from loopx.control_plane.effect_runtime import restart_effect_runtime; " "r=restart_effect_runtime(); assert r['status'] != 'shutdown_pending', r"], env=env,check=True,capture_output=True,text=True) + + +@pytest.mark.skipif(os.name != "posix", reason="POSIX process exit observation") +@pytest.mark.parametrize("outcome", ["passed", "failed", "timed_out"]) +def test_smoke_retires_every_owned_runtime_revision(tmp_path, monkeypatch, outcome): + from loopx.control_plane import effect_runtime + + marker = tmp_path / "owned-runtimes.json" + examples = tmp_path / "examples" + examples.mkdir() + script = examples / "revisions.py" + script.write_text( + "import json,os,time\nfrom pathlib import Path\n" + "from loopx.control_plane import effect_runtime as r\n" + "infos=[]\n" + "for fingerprint in (r._runtime_fingerprint(), 'f'*64):\n" + " infos.append(r._start_runtime(fingerprint=fingerprint, info_path=r._runtime_info_path(fingerprint)))\n" + f"Path({str(marker)!r}).write_text(json.dumps({{'root':os.environ['TMPDIR'],'infos':infos}}))\n" + + ("time.sleep(60)\n" if outcome == "timed_out" else "") + + f"raise SystemExit({3 if outcome == 'failed' else 0})\n", + encoding="utf-8", + ) + monkeypatch.setattr(runner, "REPO_ROOT", tmp_path) + real_popen = subprocess.Popen + + def spawn(*args, **kwargs): + process = real_popen(*args, **kwargs) + if str(script) in args[0] and outcome == "timed_out": + deadline = time.monotonic() + 20 + while not marker.exists() and time.monotonic() < deadline: + time.sleep(.01) + assert marker.exists(), "both owned runtimes must be ready before the deadline" + return process + + monkeypatch.setattr(subprocess, "Popen", spawn) + try: + result = runner._run_check( + {"command": "python examples/revisions.py"}, + timeout_seconds=.1 if outcome == "timed_out" else 30, + ) + assert result["status"] == outcome, result + recorded = json.loads(marker.read_text()) + assert not Path(recorded["root"]).exists() + for info in recorded["infos"]: + status = subprocess.run( + ["ps", "-p", str(info["pid"]), "-o", "stat="], + capture_output=True, text=True, check=False, + ).stdout.strip() + assert not status or status.startswith("Z"), "runtime survived its fixture" + finally: + # Baseline failures must not leave the regression's own daemons alive. + if marker.exists(): + recorded = json.loads(marker.read_text()) + for info in recorded["infos"]: + try: + effect_runtime._request_with_info( + info, request_id="fixture-cleanup", method="runtime.shutdown", + params={}, timeout=1, + ) + except (effect_runtime.EffectRuntimeRemoteError, + effect_runtime.EffectRuntimeResponseAmbiguous, OSError): + pass + deadline = time.monotonic() + 8 + while time.monotonic() < deadline: + states = [subprocess.run( + ["ps", "-p", str(info["pid"]), "-o", "stat="], + capture_output=True, text=True, check=False, + ).stdout.strip() for info in recorded["infos"]] + if all(not state or state.startswith("Z") for state in states): + break + time.sleep(.05) + else: + pytest.fail("owned regression runtime did not exit") + if Path(recorded["root"]).exists(): + shutil.rmtree(recorded["root"]) + + +def test_smoke_reports_cleanup_failure_and_retains_the_owned_fixture(tmp_path, monkeypatch): + examples = tmp_path / "examples" + examples.mkdir() + (examples / "output.py").write_text("print('original check output')\n") + monkeypatch.setattr(runner, "REPO_ROOT", tmp_path) + retained = [] + + def pending(root): + retained.append(root) + raise RuntimeError("runtime shutdown remains pending") + + monkeypatch.setattr(runner, "_retire_fixture_runtimes", pending) + try: + result = runner._run_check({"command": "python examples/output.py"}, timeout_seconds=5) + assert not result["ok"] and result["status"] == "failed" + assert result["returncode"] == 0 and "original check output" in result["stdout_tail"] + assert "runtime shutdown remains pending" in result["reason"] + assert len(retained) == 1 and retained[0].is_dir() + finally: + for root in retained: + shutil.rmtree(root) + + +@pytest.mark.skipif(os.name != "posix", reason="POSIX symlink fixture") +def test_fixture_retirement_does_not_follow_runtime_namespace_links(tmp_path, monkeypatch): + from loopx.control_plane import effect_runtime + + external = tmp_path / "outside" + external.mkdir() + locator = external / ("runtime-" + "f" * 16 + ".json") + locator.write_text("external runtime metadata") + root = tmp_path / "fixture" + root.mkdir() + (root / "loopx-effect-runtime-linked").symlink_to(external, target_is_directory=True) + namespace = root / "loopx-effect-runtime-local" + namespace.mkdir() + (namespace / locator.name).symlink_to(locator) + monkeypatch.setattr(effect_runtime, "_read_info", lambda *a, **kw: pytest.fail("followed external locator")) + runner._retire_fixture_runtimes(root) + assert locator.read_text() == "external runtime metadata" diff --git a/tests/control_plane/test_effect_runtime_integration.py b/tests/control_plane/test_effect_runtime_integration.py index 4fd5ecc676..830594738e 100644 --- a/tests/control_plane/test_effect_runtime_integration.py +++ b/tests/control_plane/test_effect_runtime_integration.py @@ -5,6 +5,8 @@ import shutil import signal import socket +import subprocess +import sys import time from concurrent.futures import ThreadPoolExecutor from pathlib import Path @@ -30,6 +32,43 @@ _TODO_ID = "todo_fixture0001" +@pytest.mark.parametrize("replacement", [False, True]) +def test_fixture_shutdown_waits_for_process_exit_after_locator_turnover( + tmp_path, monkeypatch, replacement, +): + locator = tmp_path / "runtime.json" + if replacement: + locator.write_text(json.dumps({"token": "replacement"})) + monkeypatch.setattr(effect_runtime, "_request_with_info", lambda *a, **kw: {}) + with subprocess.Popen( + [sys.executable, "-c", "import time; print('ready',flush=True); time.sleep(60)"], + stdout=subprocess.PIPE, text=True, start_new_session=os.name == "posix", + ) as process: + try: + assert process.stdout.readline().strip() == "ready" + info = {"pid": process.pid, "token": "original"} + # Serving handoff remains enough for the operator's existing action. + assert effect_runtime._stop_effect_runtime( + locator, info, timeout=.05, + )["status"] == "stopped" + # Namespace disposal needs stronger evidence than locator turnover. + assert effect_runtime._stop_effect_runtime( + locator, info, timeout=.05, wait_for_process_exit=True, + )["status"] == "shutdown_pending" + assert process.poll() is None + process.terminate() + process.wait(timeout=5) + assert effect_runtime._stop_effect_runtime( + locator, info, timeout=.05, wait_for_process_exit=True, + )["status"] == "stopped" + if replacement: + assert json.loads(locator.read_text()) == {"token": "replacement"} + finally: + if process.poll() is None: + process.kill() + process.wait() + + def _effect_id(label: str) -> str: return f"fixture-goal:fixture-agent:{_TODO_ID}:{label}" From 31e33376bff9059ac61317fa4e5cf30cc7b73f86 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 15 Oct 2026 15:46:36 +0800 Subject: [PATCH 2/4] fix(checkpoint): keep host composition outside the control plane Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- benchmark/README.md | 6 +- benchmark/runtime/checkpoint.py | 4 +- benchmark/runtime/checkpoint_continuation.py | 2 +- benchmark/runtime/checkpoint_wake.py | 2 +- benchmark/runtime/sforge_warm_start.py | 4 +- .../tests/test_checkpoint_joint_receiver.py | 4 +- benchmark/tests/test_native_checkpoint.py | 46 +++++++++---- .../tests/test_native_checkpoint_handoff.py | 4 +- loopx/cli_commands/research_checkpoint.py | 4 +- loopx/help_surface.py | 4 ++ .../turn_driver => hosts}/codex_checkpoint.py | 47 ++++++++------ .../runtime => hosts}/research_checkpoint.py | 64 ++++++++++--------- man/loopx.1 | 3 + .../test_content_digest_single_owner.py | 2 + .../test_research_checkpoint_capture.py | 6 +- .../test_research_checkpoint_cli.py | 8 ++- .../test_research_context_adoption.py | 8 +-- 17 files changed, 132 insertions(+), 86 deletions(-) rename loopx/{control_plane/turn_driver => hosts}/codex_checkpoint.py (95%) rename loopx/{control_plane/runtime => hosts}/research_checkpoint.py (91%) diff --git a/benchmark/README.md b/benchmark/README.md index 74a59aba03..05d3c0629a 100644 --- a/benchmark/README.md +++ b/benchmark/README.md @@ -285,7 +285,7 @@ effective configuration digest and Host version; tools/skills and credentials ne their existing installation/authentication owners. Native state is opaque and its storage layout remains a pinned provider qualification. -`loopx.control_plane.turn_driver.codex_checkpoint.fork_native_checkpoint` verifies a unit and +`loopx.hosts.codex_checkpoint.fork_native_checkpoint` verifies a unit and reconstructs a distinct native session in a **new isolated environment preserving the captured guest paths**, with matching configuration and Host version. It checks the full history digest and cursor, starts no model and registers no LoopX binding. @@ -303,7 +303,7 @@ it does not remove historical authority. The controller environment and explicit configuration are not exported as authority or credentials. ```sh -python -m loopx.control_plane.turn_driver.codex_checkpoint \ +python -m loopx.hosts.codex_checkpoint \ --root /private/captured-point --expected-sha256 CAPTURE_DIGEST \ --config-path /private/qualified-config.toml --sandbox read-only ``` @@ -464,7 +464,7 @@ adopt this retained context on an **already claimed and leased destination Todo* whose note is empty: ```sh -python -m loopx.control_plane.runtime.research_checkpoint \ +python -m loopx.hosts.research_checkpoint \ --root /private/checkpoints/point --expected-sha256 CHECKPOINT_SHA256 \ --source-todo-id SOURCE_TODO --todo-id DESTINATION_TODO \ --operation-id CONTEXT_OPERATION --expected-revision CURRENT_REVISION \ diff --git a/benchmark/runtime/checkpoint.py b/benchmark/runtime/checkpoint.py index 7c7b9ab4d1..69171521f0 100644 --- a/benchmark/runtime/checkpoint.py +++ b/benchmark/runtime/checkpoint.py @@ -17,7 +17,7 @@ from loopx.control_plane.effect_runtime import effect_runtime_result from loopx.control_plane.quota.settlement import read_heartbeat_settlement -from loopx.control_plane.runtime.research_checkpoint import ( +from loopx.hosts.research_checkpoint import ( _participant_id, native_binding_participant, research_participants, verify_research_participants, ) @@ -324,7 +324,7 @@ def capture_after_wake(env: dict[str, str], receipt: dict[str, Any], *, metadata["evaluation_opportunity"] = {"component_id": opportunity_id, "incumbent_component_id": incumbent_id} if request.get("native_session"): - from loopx.control_plane.turn_driver.codex_checkpoint import native_session_checkpoint + from loopx.hosts.codex_checkpoint import native_session_checkpoint from .codex import Execution from .sessions import BenchmarkSessionWake diff --git a/benchmark/runtime/checkpoint_continuation.py b/benchmark/runtime/checkpoint_continuation.py index fbed865ab5..1fa6e3424e 100644 --- a/benchmark/runtime/checkpoint_continuation.py +++ b/benchmark/runtime/checkpoint_continuation.py @@ -22,7 +22,7 @@ def restored_capture_gate(env, current_obligation_id=None): return {"capture_allowed": True} from .codex import Execution from .sessions import BenchmarkSessionWake - from loopx.control_plane.turn_driver.codex_checkpoint import native_session_checkpoint + from loopx.hosts.codex_checkpoint import native_session_checkpoint wake = BenchmarkSessionWake(env, Execution(context="resume", sandbox=env.get("LOOPX_CODEX_SANDBOX", "danger-full-access")), {}) diff --git a/benchmark/runtime/checkpoint_wake.py b/benchmark/runtime/checkpoint_wake.py index 2b35940e80..48387cafbd 100644 --- a/benchmark/runtime/checkpoint_wake.py +++ b/benchmark/runtime/checkpoint_wake.py @@ -115,7 +115,7 @@ def _capture(self, held, deadline): if research is None: raise ValueError("controller checkpoint requires complete canonical research context") if self.request.get("native_session"): - from loopx.control_plane.turn_driver.codex_checkpoint import native_session_checkpoint + from loopx.hosts.codex_checkpoint import native_session_checkpoint from .checkpoint import native_binding_participant components, source_wake = native_binding_participant(self.env, components, metadata) guard = {"cli": self.env["LOOPX_CLI"], "workspace": self.env["LOOPX_PROJECT"], diff --git a/benchmark/runtime/sforge_warm_start.py b/benchmark/runtime/sforge_warm_start.py index 06ca8c61bc..cc52a41dd2 100644 --- a/benchmark/runtime/sforge_warm_start.py +++ b/benchmark/runtime/sforge_warm_start.py @@ -145,8 +145,8 @@ def _prepare_native_skills(home: Path, installed: Path) -> None: def prepare_context(env: dict[str, str], declaration: dict) -> dict: """Join actual session + research owners on one freshly bootstrapped Goal.""" from benchmark.runtime.codex import Execution - from loopx.control_plane.turn_driver.codex_checkpoint import admit_native_checkpoint_session - from loopx.control_plane.runtime.research_checkpoint import adopt_todo_context + from loopx.hosts.codex_checkpoint import admit_native_checkpoint_session + from loopx.hosts.research_checkpoint import adopt_todo_context from benchmark.runtime.worker import loopx_command import subprocess diff --git a/benchmark/tests/test_checkpoint_joint_receiver.py b/benchmark/tests/test_checkpoint_joint_receiver.py index baa21f68c3..72c7e0723e 100644 --- a/benchmark/tests/test_checkpoint_joint_receiver.py +++ b/benchmark/tests/test_checkpoint_joint_receiver.py @@ -32,11 +32,11 @@ def test_captured_point_continues_with_fresh_authority( assert os.getuid() != 0, "Workspace continuation must qualify the actual non-root writer" from benchmark.edgebench.checkpoint_controller import restore from benchmark.runtime.codex import Execution - from loopx.control_plane.turn_driver.codex_checkpoint import ( + from loopx.hosts.codex_checkpoint import ( admit_native_checkpoint_session, native_session_checkpoint, ) - from loopx.control_plane.runtime.research_checkpoint import adopt_todo_context + from loopx.hosts.research_checkpoint import adopt_todo_context from benchmark.runtime.sessions import BenchmarkSessionWake from benchmark.runtime.worker import native_command from loopx.cli import main diff --git a/benchmark/tests/test_native_checkpoint.py b/benchmark/tests/test_native_checkpoint.py index be96a6563e..6c6b6bfaea 100644 --- a/benchmark/tests/test_native_checkpoint.py +++ b/benchmark/tests/test_native_checkpoint.py @@ -9,7 +9,7 @@ from tests.control_plane.codex_checkpoint_fixture import real_native_source -from loopx.control_plane.turn_driver.codex_checkpoint import native_session_checkpoint +from loopx.hosts.codex_checkpoint import native_session_checkpoint from benchmark.tests.test_checkpoint_controller import controllers # noqa: F401 -- shared real HTTP fixture @@ -43,8 +43,8 @@ def request(self, method, params): assert method == "thread/turns/list" # No model or fork is requested. return pages.pop(0) - monkeypatch.setattr("loopx.control_plane.turn_driver.codex_checkpoint.StdioNativeGoalTransport.spawn", lambda *a, **k: Native()) - monkeypatch.setattr("loopx.control_plane.turn_driver.codex_checkpoint.subprocess.check_output", lambda *a, **k: "codex-cli fixture") + monkeypatch.setattr("loopx.hosts.codex_checkpoint.StdioNativeGoalTransport.spawn", lambda *a, **k: Native()) + monkeypatch.setattr("loopx.hosts.codex_checkpoint.subprocess.check_output", lambda *a, **k: "codex-cli fixture") return {"CODEX_HOME": str(home), "CODEX_BIN": "/fixture/codex", "LOOPX_PROJECT": str(tmp_path)}, calls @@ -208,7 +208,7 @@ def test_real_fork_admission_and_native_resume_use_retained_state(tmp_path, caps import sys from pathlib import Path from benchmark.runtime.codex import Execution - from loopx.control_plane.turn_driver.codex_checkpoint import admit_native_checkpoint_session + from loopx.hosts.codex_checkpoint import admit_native_checkpoint_session from benchmark.runtime.sessions import BenchmarkSessionWake from benchmark.runtime.worker import native_command from loopx.cli import main @@ -251,7 +251,7 @@ def register(name, agent): assert main([*prefix, "quota", "should-run", "--runtime-profile", "generic_cli", *binding]) == 0 guard = json.loads(capsys.readouterr().out) assert guard["interaction_contract"]["agent_channel"]["delivery_allowed"] is True - from loopx.control_plane.runtime.research_checkpoint import canonical_context, research_context, adopt_todo_context, _encoded + from loopx.hosts.research_checkpoint import canonical_context, research_context, adopt_todo_context, _encoded source_prefix = ["--format", "json", "--registry", str(source_registry), "--runtime-root", str(source_runtime)] def invoke(arguments): code = main(arguments) @@ -406,7 +406,7 @@ def retained_source(): with monkeypatch.context() as automatic: def forbidden_automatic_fork(*args, **kwargs): raise AssertionError("an unsupported automatic Goal profile must fail before native reconstruction") - automatic.setattr("loopx.control_plane.turn_driver.codex_checkpoint.fork_native_checkpoint", forbidden_automatic_fork) + automatic.setattr("loopx.hosts.codex_checkpoint.fork_native_checkpoint", forbidden_automatic_fork) with pytest.raises(ValueError, match="automatic Goals explicitly disabled"): admit_native_checkpoint_session(env, unit, digest, unsupported_config, sandbox=(execution).sandbox, timeout=30) with pytest.raises(ValueError, match="distinct destination"): @@ -414,7 +414,7 @@ def forbidden_automatic_fork(*args, **kwargs): config, sandbox=(execution).sandbox, timeout=30) assert not Path(env["CODEX_HOME"]).exists() and len(calls) == 1 if case == "scope_change": - from benchmark.runtime import native_checkpoint + from loopx.hosts import codex_checkpoint as native_checkpoint original_fork = native_checkpoint.fork_native_checkpoint def change_scope_after_real_fork(*args, **kwargs): result = original_fork(*args, **kwargs) @@ -441,7 +441,7 @@ def change_scope_after_real_fork(*args, **kwargs): return recovered_fork = None if case in {"fork_reply_loss", "fork_missing", "fork_ambiguous"}: - from loopx.control_plane.turn_driver.codex_checkpoint import StdioNativeGoalTransport + from loopx.hosts.codex_checkpoint import StdioNativeGoalTransport original_request = StdioNativeGoalTransport.request forks = [] def lose_fork_reply(self, method, params): @@ -474,7 +474,7 @@ def no_second_fork(self, method, params): recovered_fork = forks[0] # The isolated attempt is bound to this destination, not merely an # equivalent visible history or another Goal's transport directory. - from loopx.control_plane.turn_driver.codex_checkpoint import fork_native_checkpoint + from loopx.hosts.codex_checkpoint import fork_native_checkpoint with pytest.raises(ValueError, match="attempt differs"): fork_native_checkpoint({**env, "LOOPX_AGENT_ID": "unrelated-destination"}, unit, digest, config, timeout=30) @@ -491,7 +491,7 @@ def lose_binding_reply(self, *args): assert load_codex_cli_session(runtime, lineage={"goal_id": destination_goal, "agent_id": destination_agent}, session_scope="agent")["checkpoint_artifact_sha256"] == digest assert len(calls) == 1 - admitted_cli = subprocess.run([sys.executable, "-m", "loopx.control_plane.turn_driver.codex_checkpoint", + admitted_cli = subprocess.run([sys.executable, "-m", "loopx.hosts.codex_checkpoint", "--root", str(unit), "--expected-sha256", digest, "--config-path", str(config), "--sandbox", execution.sandbox], env=env, text=True, capture_output=True, timeout=40) assert admitted_cli.returncode == 0, admitted_cli.stderr @@ -508,7 +508,7 @@ def lose_binding_reply(self, *args): with monkeypatch.context() as replay: def forbidden_fork(*args, **kwargs): raise AssertionError("a repeated admission must not create a second native fork") - replay.setattr("loopx.control_plane.turn_driver.codex_checkpoint.fork_native_checkpoint", forbidden_fork) + replay.setattr("loopx.hosts.codex_checkpoint.fork_native_checkpoint", forbidden_fork) recovered = admit_native_checkpoint_session(env, unit, digest, config, sandbox=(execution).sandbox, timeout=30) assert recovered == admitted assert bound["checkpoint_artifact_sha256"] == digest @@ -549,11 +549,12 @@ def forbidden_fork(*args, **kwargs): lease = invoke([*prefix, "task-lease", "acquire", "--goal-id", destination_goal, "--todo-id", destination_todo, "--owner", destination_agent, "--idempotency-key", "branch-context"]) current = invoke([*prefix, "todo", "list", "--goal-id", destination_goal, "--todo-id", destination_todo]) - context_command = [sys.executable, "-m", "loopx.control_plane.runtime.research_checkpoint", + context_command = [sys.executable, "-m", "loopx.hosts.research_checkpoint", "--root", str(unit), "--expected-sha256", digest, "--source-todo-id", source_todo, "--todo-id", destination_todo, "--operation-id", "branch-context", "--expected-revision", current["authority_read"]["provider_revision"], "--lease-key", "branch-context", "--lease-version", str(lease["lease"]["version"])] + assert not explore_result_log_path(runtime, destination_goal).exists(), "session admission must not adopt knowledge" adopted_cli = subprocess.run(context_command, env=env, text=True, capture_output=True, timeout=60) assert adopted_cli.returncode == 0, adopted_cli.stderr context_result = json.loads(adopted_cli.stdout) @@ -644,7 +645,24 @@ def forbidden_fork(*args, **kwargs): assert "fresh samples reversed its pilot gain" in json.dumps(requests[2]) assert "Complete a fresh-sample experiment" in json.dumps(requests[2]) assert source_context["research_evidence"]["explore_events"][0]["status"] == "dead_end" - assert not explore_result_log_path(runtime, destination_goal).exists() + # Explicit context adoption seeds verified knowledge, never historical + # execution observations. Native session admission alone above did not. + from loopx.capabilities.explore.result_log import EVENT_KIND_EDGE + adopted_events = [json.loads(line) for line in + explore_result_log_path(runtime, destination_goal).read_text().splitlines()] + captured_events = source_context["research_evidence"]["explore_events"] + assert len(adopted_events) == len(captured_events) + for captured, adopted_event in zip(captured_events, adopted_events, strict=True): + assert adopted_event["goal_id"] == destination_goal + assert adopted_event["run_id"] == "checkpoint:" + digest + ":" + captured["event_id"] + assert adopted_event["event_id"] != captured["event_id"] + assert "research_observation" not in adopted_event + changed_identity = {"goal_id", "run_id", "event_id", "research_observation"} + if captured["event_kind"] == EVENT_KIND_EDGE: + changed_identity.add("result_id") + assert all(adopted_event.get(key) == value for key, value in captured.items() + if key not in changed_identity) + assert "Candidate 999" not in json.dumps(requests[2]) assert "Future route outside this checkpoint" not in json.dumps(requests[2]) assert canonical_context({**env, "LOOPX_GOAL_ID": source_goal, @@ -703,7 +721,7 @@ def test_interrupted_cursor_requires_exact_unreleased_final_guard(tmp_path, monk """A stopped command is eligible only at the declared pre-output boundary.""" import copy import shlex - from loopx.control_plane.turn_driver.codex_checkpoint import _verify_interrupted_guard + from loopx.hosts.codex_checkpoint import _verify_interrupted_guard guard = {"cli": "/fixture/loopx", "workspace": str(tmp_path), "registry": "/fixture/registry.json", "runtime_root": "/fixture/runtime", "goal_id": "research", "agent_id": "worker", diff --git a/benchmark/tests/test_native_checkpoint_handoff.py b/benchmark/tests/test_native_checkpoint_handoff.py index f2a9e7f254..c80e67860f 100644 --- a/benchmark/tests/test_native_checkpoint_handoff.py +++ b/benchmark/tests/test_native_checkpoint_handoff.py @@ -12,7 +12,7 @@ import pytest -from loopx.control_plane.turn_driver.codex_checkpoint import native_session_checkpoint +from loopx.hosts.codex_checkpoint import native_session_checkpoint from tests.control_plane.codex_checkpoint_fixture import real_native_source from loopx.capabilities.benchmark_toolkit.native_codex_goal import ( NativeGoalProtocolError, StdioNativeGoalTransport, @@ -740,7 +740,7 @@ def response(payload, index, message): # Reconstruct the earlier captured store after the source has # already continued. Fixed guest paths require an isolated home; # retain and restore the source bytes around this synthetic test. - from loopx.control_plane.turn_driver.codex_checkpoint import fork_native_checkpoint, admit_native_checkpoint_session + from loopx.hosts.codex_checkpoint import fork_native_checkpoint, admit_native_checkpoint_session from benchmark.runtime.codex import Execution home = Path(native["guest_home"]) saved_home = home.with_name(home.name + "-retained") diff --git a/loopx/cli_commands/research_checkpoint.py b/loopx/cli_commands/research_checkpoint.py index f9a9de50bf..3ae2993ecb 100644 --- a/loopx/cli_commands/research_checkpoint.py +++ b/loopx/cli_commands/research_checkpoint.py @@ -65,12 +65,12 @@ def required(*names): missing = [key for key in ("CODEX_BIN", "CODEX_HOME", "MODEL_NAME", "REASONING_EFFORT") if not env.get(key)] if missing: raise ValueError("Explicit Codex Host environment is missing: " + ", ".join(missing)) - from ..control_plane.runtime.research_checkpoint import capture_research_checkpoint, adopt_todo_context + from ..hosts.research_checkpoint import capture_research_checkpoint, adopt_todo_context if args.operation == "capture": result = capture_research_checkpoint(env, json.loads(args.request_file.read_bytes()), turn_instance_id=args.turn_instance_id, timeout=args.timeout_seconds) elif args.operation == "admit-session": - from ..control_plane.turn_driver.codex_checkpoint import admit_native_checkpoint_session + from ..hosts.codex_checkpoint import admit_native_checkpoint_session result = admit_native_checkpoint_session(env, args.root, args.expected_sha256, args.config_path, sandbox=args.sandbox, timeout=args.timeout_seconds) else: diff --git a/loopx/help_surface.py b/loopx/help_surface.py index c95de588ea..55abccdf52 100644 --- a/loopx/help_surface.py +++ b/loopx/help_surface.py @@ -338,6 +338,10 @@ def register_command_reference(subparsers: argparse._SubParsersAction) -> None: "command": "loopx zcode-goal --help", "purpose": "Explicitly bind and inspect one managed ZCode CLI native Goal session.", }, + { + "command": "loopx research-checkpoint --help", + "purpose": "Explicitly capture, verify, or adopt a sealed checkpoint through current Goal and native Host owners.", + }, { "command": "loopx promotion-readiness", "purpose": "Record release-scoped canary readiness evidence in the shared runtime ledger.", diff --git a/loopx/control_plane/turn_driver/codex_checkpoint.py b/loopx/hosts/codex_checkpoint.py similarity index 95% rename from loopx/control_plane/turn_driver/codex_checkpoint.py rename to loopx/hosts/codex_checkpoint.py index cc012288ef..06ddf0e22c 100644 --- a/loopx/control_plane/turn_driver/codex_checkpoint.py +++ b/loopx/hosts/codex_checkpoint.py @@ -19,10 +19,10 @@ import time import tomllib from pathlib import Path -from contextlib import redirect_stdout, redirect_stderr +from contextlib import contextmanager, nullcontext, redirect_stdout, redirect_stderr from typing import Any -from .codex_stdio import StdioNativeGoalTransport +from loopx.control_plane.turn_driver.codex_stdio import StdioNativeGoalTransport def _complete_history(request, session_id: str, *, interrupted_guard: dict | None = None) -> list[dict[str, Any]]: @@ -130,31 +130,40 @@ def arguments(command): raise ValueError("native pending guard call identity differs") +@contextmanager +def native_checkpoint_reader(env: dict[str, str], *, timeout: float): + """Keep one native inspection process open across a sealed capture window.""" + with StdioNativeGoalTransport.spawn( + [env["CODEX_BIN"], "-c", "features.goals=false", "app-server", "--listen", "stdio://"], + cwd=env["LOOPX_PROJECT"], env=env, stderr=subprocess.DEVNULL, + response_timeout_sec=timeout, + ) as transport: + transport.request("initialize", {"clientInfo": {"name": "loopx_checkpoint", "version": "0.1.0"}, + "capabilities": {"experimentalApi": True}}) + transport.notify("initialized", {}) + yield transport + + def native_session_checkpoint(env: dict[str, str], session_id: str, *, timeout: float, interrupted_guard: dict | None = None, - retained_history: dict | None = None) -> dict[str, Any]: + retained_history: dict | None = None, + transport: StdioNativeGoalTransport | None = None) -> dict[str, Any]: home = Path(env["CODEX_HOME"]) if not session_id or timeout <= 0: raise ValueError("native checkpoint requires a session and an open capture window") deadline = time.monotonic() + timeout - version = subprocess.check_output([env["CODEX_BIN"], "--version"], env=env, text=True, + version = subprocess.check_output([env["CODEX_BIN"], "--version"], env=env, text=True, encoding="utf-8", timeout=timeout).strip() # Native storage is opaque. Retain both rollout and paginated/index stores; # auth, logs, memories and the symlinked skill installation are not members. - with StdioNativeGoalTransport.spawn( - [env["CODEX_BIN"], "-c", "features.goals=false", "app-server", "--listen", "stdio://"], - cwd=env["LOOPX_PROJECT"], env=env, stderr=subprocess.DEVNULL, - response_timeout_sec=timeout, - ) as transport: + reader = nullcontext(transport) if transport is not None else native_checkpoint_reader(env, timeout=timeout) + with reader as transport: def request(method, params): remaining = deadline - time.monotonic() if remaining <= 0: raise TimeoutError("native checkpoint window closed") transport.response_timeout_sec = remaining return transport.request(method, params) - request("initialize", {"clientInfo": {"name": "loopx_checkpoint", "version": "0.1.0"}, - "capabilities": {"experimentalApi": True}}) - transport.notify("initialized", {}) readback = request("thread/read", {"threadId": session_id, "includeTurns": False}) if readback.get("thread", {}).get("id") != session_id: raise ValueError("native checkpoint session identity mismatch") @@ -245,7 +254,7 @@ def _fork_native_checkpoint(env: dict[str, str], unit: Path, expected_sha256: st effect_runtime_result("runtime.checkpoint_artifacts", { "action": "verify", "root": str(unit), "expected_sha256": expected_sha256, }, timeout=timeout) - manifest = json.loads((unit / "checkpoint.json").read_text()) + manifest = json.loads((unit / "checkpoint.json").read_text(encoding="utf-8")) native = manifest["point_metadata"]["native_session"] if native.get("interrupted_guard") is not None: component_id = manifest["point_metadata"].get("held_decision_component_id") @@ -271,7 +280,7 @@ def _fork_native_checkpoint(env: dict[str, str], unit: Path, expected_sha256: st config = config_path.read_bytes() if hashlib.sha256(config).hexdigest() != native["config_sha256"]: raise ValueError("native checkpoint effective configuration differs") - version = subprocess.check_output([env["CODEX_BIN"], "--version"], env=env, text=True, + version = subprocess.check_output([env["CODEX_BIN"], "--version"], env=env, text=True, encoding="utf-8", timeout=max(.1, deadline - time.monotonic())).strip() if version != native["host_version"]: raise ValueError("native checkpoint Host version differs") @@ -397,7 +406,7 @@ def admit_native_checkpoint_session(env: dict[str, str], unit: Path, expected_sh project_registry_transaction, ) from loopx.control_plane.turn_driver.codex_sessions import select_codex_cli_session - from .codex_agent_session import CodexAgentSession + from loopx.control_plane.turn_driver.codex_agent_session import CodexAgentSession if not math.isfinite(timeout) or timeout <= 0: raise ValueError("checkpoint session admission timeout must be finite and positive") @@ -407,9 +416,11 @@ def admit_native_checkpoint_session(env: dict[str, str], unit: Path, expected_sh }, timeout=timeout) metadata = json.loads((unit / "checkpoint.json").read_bytes())["point_metadata"] source_profile = (metadata.get("session") or {}).get("session_profile_digest") - if not isinstance(source_profile, str) or not re.fullmatch(r"[0-9a-f]{64}", source_profile): + from loopx.control_plane.content_digest import BARE_SHA256_PATTERN + + if not isinstance(source_profile, str) or not BARE_SHA256_PATTERN.fullmatch(source_profile): raise ValueError("checkpoint session admission requires its captured source session profile") - if tomllib.loads(config_path.read_text()).get("features", {}).get("goals") is not False: + if tomllib.loads(config_path.read_text(encoding="utf-8")).get("features", {}).get("goals") is not False: raise ValueError("checkpoint session admission requires native automatic Goals explicitly disabled") if hashlib.sha256(config_path.read_bytes()).hexdigest() != metadata["native_session"]["config_sha256"]: raise ValueError("native checkpoint effective configuration differs") @@ -475,7 +486,7 @@ def admit_native_checkpoint_session(env: dict[str, str], unit: Path, expected_sh def main(argv=None) -> None: """Controller entrypoint inside an explicitly prepared isolated guest.""" import os - from .codex_cli import CODEX_CLI_SANDBOXES + from loopx.control_plane.turn_driver.codex_cli import CODEX_CLI_SANDBOXES parser = argparse.ArgumentParser(description="Admit a checkpoint's native session participant; no model starts.") parser.add_argument("--root", type=Path, required=True) diff --git a/loopx/control_plane/runtime/research_checkpoint.py b/loopx/hosts/research_checkpoint.py similarity index 91% rename from loopx/control_plane/runtime/research_checkpoint.py rename to loopx/hosts/research_checkpoint.py index 153c7558d7..2ce24b8058 100644 --- a/loopx/control_plane/runtime/research_checkpoint.py +++ b/loopx/hosts/research_checkpoint.py @@ -136,8 +136,8 @@ def _handoff(env, payload): def _context_session(env, metadata): """Read the admitted native branch before publishing captured Todo context.""" - from ..turn_driver.codex_agent_session import CodexAgentSession - from ..turn_driver.codex_checkpoint import native_session_checkpoint + from loopx.control_plane.turn_driver.codex_agent_session import CodexAgentSession + from loopx.hosts.codex_checkpoint import native_session_checkpoint wake = CodexAgentSession(env, context="resume", sandbox=env["LOOPX_CODEX_SANDBOX"], receipt={}) native = metadata["native_session"] @@ -384,7 +384,7 @@ def _loopx_state_components(env, components, output_dir): def native_binding_participant(env, components, metadata): - from ..turn_driver.codex_agent_session import CodexAgentSession + from loopx.control_plane.turn_driver.codex_agent_session import CodexAgentSession source_wake = CodexAgentSession(env, context="resume", sandbox=env.get("LOOPX_CODEX_SANDBOX", "danger-full-access"), receipt={}) if source_wake.session_id != metadata["session"].get("session_id"): @@ -452,9 +452,9 @@ def capture_research_checkpoint(env: dict[str, str], request: dict, *, """ import math import tempfile - from ..quota.settlement import read_heartbeat_settlement - from ..turn_driver.codex_checkpoint import native_session_checkpoint - from ..turn_driver.codex_agent_session import CodexAgentSession + from loopx.control_plane.quota.settlement import read_heartbeat_settlement + from loopx.hosts.codex_checkpoint import native_checkpoint_reader, native_session_checkpoint + from loopx.control_plane.turn_driver.codex_agent_session import CodexAgentSession if (not isinstance(request, dict) or set(request) != {"destination", "components", "point_metadata"} or @@ -489,30 +489,34 @@ def settled(): if research is None: raise ValueError("research capture requires canonical source coordination") components, observed_session = native_binding_participant(env, components, metadata) - native = native_session_checkpoint(env, session.session_id, timeout=max(.1, deadline-time.time())) - allocated = {} - for entry in native["components"]: - name = _participant_id(components, entry["id"]) - allocated[entry["id"]] = name - components = [*components, {**entry, "id": name}] - metadata["native_session"] = {**native["metadata"], "store_members": [ - {**entry, "id": allocated[entry["id"]]} for entry in native["metadata"]["store_members"]]} - probe = effect_runtime_result("runtime.checkpoint_artifacts", { - "action": "probe", "components": components, "deadline_epoch_ms": deadline*1000}, - timeout=max(.1, deadline-time.time()), retry_safe=False) - if native_session_checkpoint(env, session.session_id, timeout=max(.1, deadline-time.time())) != native: - raise ValueError("native checkpoint source moved before sealing") - _, confirmed = native_binding_participant(env, components, metadata) - if ((confirmed.session_id, confirmed.goal_ref, confirmed.digest) != - (observed_session.session_id, observed_session.goal_ref, observed_session.digest)): - raise ValueError("research checkpoint session binding moved before sealing") - verify_research_participants(env, components, metadata, research, probe["source_witness"], deadline) - if settled().identity.value != original.identity.value: - raise ValueError("research checkpoint source settlement moved") - result = effect_runtime_result("runtime.checkpoint_artifacts", { - "action": "capture", "destination": str(destination), "components": components, - "expected_source_sha256": probe["source_sha256"], "point_metadata": metadata, - "deadline_epoch_ms": deadline*1000}, timeout=max(.1, deadline-time.time()), retry_safe=False) + # Reopening the native reader after probing can checkpoint its history + # database and invalidate our own byte witness. Keep one inspection + # process open; the original logical reread and byte fence both remain. + with native_checkpoint_reader(env, timeout=max(.1, deadline-time.time())) as reader: + native = native_session_checkpoint(env, session.session_id, timeout=max(.1, deadline-time.time()), transport=reader) + allocated = {} + for entry in native["components"]: + name = _participant_id(components, entry["id"]) + allocated[entry["id"]] = name + components = [*components, {**entry, "id": name}] + metadata["native_session"] = {**native["metadata"], "store_members": [ + {**entry, "id": allocated[entry["id"]]} for entry in native["metadata"]["store_members"]]} + probe = effect_runtime_result("runtime.checkpoint_artifacts", { + "action": "probe", "components": components, "deadline_epoch_ms": deadline*1000}, + timeout=max(.1, deadline-time.time()), retry_safe=False) + if native_session_checkpoint(env, session.session_id, timeout=max(.1, deadline-time.time()), transport=reader) != native: + raise ValueError("native checkpoint source moved before sealing") + _, confirmed = native_binding_participant(env, components, metadata) + if ((confirmed.session_id, confirmed.goal_ref, confirmed.digest) != + (observed_session.session_id, observed_session.goal_ref, observed_session.digest)): + raise ValueError("research checkpoint session binding moved before sealing") + verify_research_participants(env, components, metadata, research, probe["source_witness"], deadline) + if settled().identity.value != original.identity.value: + raise ValueError("research checkpoint source settlement moved") + result = effect_runtime_result("runtime.checkpoint_artifacts", { + "action": "capture", "destination": str(destination), "components": components, + "expected_source_sha256": probe["source_sha256"], "point_metadata": metadata, + "deadline_epoch_ms": deadline*1000}, timeout=max(.1, deadline-time.time()), retry_safe=False) return result diff --git a/man/loopx.1 b/man/loopx.1 index dad2415da1..148a65183d 100644 --- a/man/loopx.1 +++ b/man/loopx.1 @@ -279,6 +279,9 @@ Inspect or explicitly configure a Goal material basis; writes require \-\-execut \fBloopx zcode\-goal \-\-help\fR Explicitly bind and inspect one managed ZCode CLI native Goal session. .TP +\fBloopx research\-checkpoint \-\-help\fR +Explicitly capture, verify, or adopt a sealed checkpoint through current Goal and native Host owners. +.TP \fBloopx promotion\-readiness\fR Record release\-scoped canary readiness evidence in the shared runtime ledger. .TP diff --git a/tests/architecture/test_content_digest_single_owner.py b/tests/architecture/test_content_digest_single_owner.py index cb9571e58e..a850876e36 100644 --- a/tests/architecture/test_content_digest_single_owner.py +++ b/tests/architecture/test_content_digest_single_owner.py @@ -182,6 +182,7 @@ "loopx.control_plane.collaboration.inbox", "loopx.control_plane.collaboration.links", "loopx.control_plane.collaboration.peers", + "loopx.control_plane.collaboration.request_control_source", "loopx.control_plane.collaboration.result_files", "loopx.control_plane.coordination.local_authority_shadow_history", "loopx.control_plane.coordination.shadow_management", @@ -206,6 +207,7 @@ "loopx.extensions.lark.document_comment_provider", "loopx.extensions.openviking_semantic_preference.history_export", "loopx.extensions.presentation", + "loopx.hosts.codex_checkpoint", "loopx.presentation.chat_bundle", "loopx.presentation.static_site", ) diff --git a/tests/control_plane/test_research_checkpoint_capture.py b/tests/control_plane/test_research_checkpoint_capture.py index 7db471527d..86ea23ae7b 100644 --- a/tests/control_plane/test_research_checkpoint_capture.py +++ b/tests/control_plane/test_research_checkpoint_capture.py @@ -320,7 +320,7 @@ def between_read_and_probe(method, params, **kwargs): # provider fork entry, retaining the original home for byte comparison. # Fixed paths are swapped in this fixture; this is not guest isolation. import shutil - from loopx.control_plane.turn_driver.codex_checkpoint import fork_native_checkpoint + from loopx.hosts.codex_checkpoint import fork_native_checkpoint native = manifest["point_metadata"]["native_session"] assert all(member["id"] != caller_id for member in native["store_members"]) @@ -636,7 +636,7 @@ def test_research_participants_preserve_legal_caller_ids(tmp_path, monkeypatch, @pytest.mark.parametrize('gap', ['shared_registry','wrong_actor','wrong_runtime','missing_state','symlink_state','overlap']) def test_loopx_participant_rejects_wrong_or_ambiguous_routes(tmp_path, gap): - from loopx.control_plane.runtime.research_checkpoint import _loopx_state_components + from loopx.hosts.research_checkpoint import _loopx_state_components project, runtime, registry = cli._write_fixture(tmp_path) runtime.mkdir() (project/'solver.py').write_text('candidate = 7\n') @@ -667,7 +667,7 @@ def test_loopx_participant_rejects_wrong_or_ambiguous_routes(tmp_path, gap): def test_loopx_participant_preserves_legacy_and_current_owned_goal_directories(tmp_path): - from loopx.control_plane.runtime.research_checkpoint import _loopx_state_components + from loopx.hosts.research_checkpoint import _loopx_state_components from benchmark.runtime.checkpoint import parse_checkpoint_request project, runtime, registry = cli._write_fixture(tmp_path) # These are legitimate controller partitions, not two Goal registrations. diff --git a/tests/control_plane/test_research_checkpoint_cli.py b/tests/control_plane/test_research_checkpoint_cli.py index c701c02d56..f76942d9b4 100644 --- a/tests/control_plane/test_research_checkpoint_cli.py +++ b/tests/control_plane/test_research_checkpoint_cli.py @@ -96,7 +96,7 @@ def response(body, index, message): session.admission.accept_result(lambda: _store_codex_cli_session(runtime, lineage=session.lineage, session_scope='agent', session_id=source_session, goal_ref=session.goal_ref, session_profile_digest=session.digest)) observations = [] - from loopx.control_plane.turn_driver.codex_checkpoint import native_session_checkpoint + from loopx.hosts.codex_checkpoint import native_session_checkpoint def native(env, session_id, **kwargs): observations.append(session_id) if real_host: @@ -104,7 +104,11 @@ def native(env, session_id, **kwargs): return {'components': [{'id': 'native_history', 'source_path': str(history)}], 'metadata': {'session_id': session_id, 'turn_ids': ['completed-work'], 'completed_turn_id': 'completed-work', 'store_members': [{'id': 'native_history', 'name': history.name}], 'execution_authority_granted': False}} - monkeypatch.setattr('loopx.control_plane.turn_driver.codex_checkpoint.native_session_checkpoint', native) + monkeypatch.setattr('loopx.hosts.codex_checkpoint.native_session_checkpoint', native) + if not real_host: + from contextlib import nullcontext + monkeypatch.setattr('loopx.hosts.codex_checkpoint.native_checkpoint_reader', + lambda *args, **kwargs: nullcontext(None)) destination = tmp_path / 'points' / 'captured' capture_request = tmp_path / 'capture.json' capture_request.write_text(json.dumps({'destination': str(destination), 'components': [{'id': 'code', 'source_path': str(code)}], diff --git a/tests/control_plane/test_research_context_adoption.py b/tests/control_plane/test_research_context_adoption.py index df5e2f32d7..a3dfffcaa3 100644 --- a/tests/control_plane/test_research_context_adoption.py +++ b/tests/control_plane/test_research_context_adoption.py @@ -12,7 +12,7 @@ from loopx.cli import main from loopx.control_plane.effect_runtime import effect_runtime_result -from loopx.control_plane.runtime.research_checkpoint import canonical_context, research_context, verify_context_sources, adopt_todo_context, _encoded +from loopx.hosts.research_checkpoint import canonical_context, research_context, verify_context_sources, adopt_todo_context, _encoded from canonical_authority_fixture import isolate_sqlite_runtime from loopx.control_plane.turn_driver.codex_sessions import _store_codex_cli_session @@ -96,7 +96,7 @@ def run(*args): session_scope='agent',session_id='branch-session',session_profile_digest='synthetic-profile') # This owner-focused fixture substitutes only native history verification. # The real Host + CLI + canonical context journey lives in test_native_checkpoint. - monkeypatch.setattr('loopx.control_plane.runtime.research_checkpoint._context_session', lambda *_: 'branch-session') + monkeypatch.setattr('loopx.hosts.research_checkpoint._context_session', lambda *_: 'branch-session') before=run('todo','list','--goal-id','branch','--todo-id',destination_todo) args=dict(source_todo_id=source_todo,todo_id=destination_todo,operation_id='context-adoption', expected_revision=before['authority_read']['provider_revision'],lease_key='branch-lease',lease_version=1) @@ -138,7 +138,7 @@ def lost_append(*args, **kwargs): raise OSError("synthetic committed knowledge response lost") interrupted.setattr(result_log, "append_explore_result_events", lost_append) else: - from loopx.control_plane.runtime import research_checkpoint + from loopx.hosts import research_checkpoint original_handoff = research_checkpoint._handoff def lost_handoff(env, payload): result = original_handoff(env, payload) @@ -369,7 +369,7 @@ def test_research_history_is_complete_and_bound_to_original_source(tmp_path, mon with pytest.raises(ValueError, match="appeared before sealing"): verify_context_sources(absent, components, appeared["source_witness"]) # An append during the canonical read is also rejected, not silently mixed. - from loopx.control_plane.runtime import research_checkpoint + from loopx.hosts import research_checkpoint original_read = research_checkpoint.canonical_context def racing_read(*args, **kwargs): result = original_read(*args, **kwargs) From fa56a95d180daad2c363ffa137f77d7110262b14 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 15 Oct 2026 15:46:36 +0800 Subject: [PATCH 3/4] fix(extensions): expose the shared presentation path classifier Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../rfcs/monorepo-distribution-split-v0.md | 18 ++++++++++++++++++ .../monorepo-distribution-split-v0.zh-CN.md | 16 ++++++++++++++++ docs/reference/extensions.md | 9 +++++++++ loopx/extensions/presentation.py | 2 ++ .../presentation_compat.py | 12 +----------- .../presentation_validation.py | 2 +- .../test_finance_value_discovery_extension.py | 3 ++- 7 files changed, 49 insertions(+), 13 deletions(-) diff --git a/docs/architecture/rfcs/monorepo-distribution-split-v0.md b/docs/architecture/rfcs/monorepo-distribution-split-v0.md index 9cce2d0b2e..1facd043f4 100644 --- a/docs/architecture/rfcs/monorepo-distribution-split-v0.md +++ b/docs/architecture/rfcs/monorepo-distribution-split-v0.md @@ -362,6 +362,24 @@ qualification or performance claim is made by this RFC. the evidence column gains an executable interim no-growth check beside the manual review. +### 2026-10-15 — Checkpoint host composition and presentation SDK boundary + +- **Baseline:** `77b404b92`. +- **Delivered:** native checkpoint inspection/reconstruction and research context + composition move from `control_plane` to the existing `hosts` owner in + Section 5. All internal CLI, benchmark and test callers move together; neither + old module shipped in v1.4.0, so no public compatibility shim is retained. + Typed state/effect authority remains in the existing control plane. +- **Boundary repair:** Finance consumes the shared local-path classifier through + its existing `extensions.presentation` SDK import. It adds no package-to-kernel + edge or copied privacy matcher. The pinned 13 legacy imports remain unchanged. +- **Evidence:** architecture/import ratchets, Finance's minimal public SDK fixture, + real native checkpoint capture/fork/adoption and negative consistency cases. + Capture keeps one inspection process across both logical reads and physical + sealing, retaining the byte witness rather than reopening its history writer. +- **Remaining:** the wider M1–M5 distribution and zero-private-import acceptance + is not closed; D1–D4 stay open. No packaging or authority decision changes. + ## Appendix B: Decision log | Date | Decision | Owner / approval | Alternatives | Normative sections changed | diff --git a/docs/architecture/rfcs/monorepo-distribution-split-v0.zh-CN.md b/docs/architecture/rfcs/monorepo-distribution-split-v0.zh-CN.md index ca89334bd2..35f8d7fbdc 100644 --- a/docs/architecture/rfcs/monorepo-distribution-split-v0.zh-CN.md +++ b/docs/architecture/rfcs/monorepo-distribution-split-v0.zh-CN.md @@ -246,6 +246,22 @@ capability 包直接采用现有 `extension.toml` 契约，不做扩展。唯一 - **对规范设计的影响：** 无；必需结果不变，证据列在人工审查之外增加一项可执行的 过渡性不增长检查。 +### 2026-10-15 — Checkpoint Host 组合与 presentation SDK 边界 + +- **基线：** `77b404b92`。 +- **交付：** native checkpoint 检查、重建及研究上下文组合从 `control_plane` + 移到第 5 节已有的 `hosts` owner。CLI、benchmark 与测试内部调用方一起迁移； + 两个旧模块均未随 v1.4.0 发布，因此不保留公开兼容 shim。类型化状态与 effect + 权威仍由现有控制面持有。 +- **边界修复：** Finance 经已有 `extensions.presentation` SDK import 使用共享 + 本地路径分类器，不新增包到内核的依赖边，也不复制隐私 matcher。既有 13 条 + 历史 import pin 保持不变。 +- **证据：** 架构/import ratchet、Finance 最小公开 SDK fixture、真实 native + checkpoint 捕获/分叉/采用及一致性反例。捕获在两次逻辑读取与物理封存之间复用 + 一个检查进程，保留字节 witness，不重新打开历史库写入进程。 +- **剩余：** 不关闭更广的 M1–M5 发行物与零私有 import 验收；D1–D4 仍未决。 + 不改变打包方案或权威决策。 + ## 附录 B：决策日志 | 日期 | 决策 | 负责人 / 批准 | 备选 | 变更的规范章节 | diff --git a/docs/reference/extensions.md b/docs/reference/extensions.md index f4bd11ddb4..b184d45114 100644 --- a/docs/reference/extensions.md +++ b/docs/reference/extensions.md @@ -601,6 +601,15 @@ into core. Dashboard's generic status parser consumes only that compact contract. A built-in renderer may separately own a provider view schema, as the Finance renderer does for `decision_research_dashboard_v0`. +The public `loopx.extensions.presentation` API also exports +`find_public_safe_local_path(text)`. Providers use this shared path classifier +through their existing presentation API rather than importing its internal Core +owner. It does not certify that text is secret-free or authorize publication; +providers retain their complete view validation and the normal publication gate. +A provider that requires this utility must reject presentation when it is +unavailable, while unrelated non-presentation operations may retain their +explicit legacy validation path. + The validator executes inside the active runtime interpreter - the extension's own environment, not the LoopX process - and LoopX launches it isolated (`python -I`), so `PYTHONPATH`, `PYTHONIOENCODING`, and `PYTHONUTF8` are ignored diff --git a/loopx/extensions/presentation.py b/loopx/extensions/presentation.py index d5c61ca803..91268b5e88 100644 --- a/loopx/extensions/presentation.py +++ b/loopx/extensions/presentation.py @@ -16,6 +16,8 @@ from urllib.parse import parse_qsl, urlparse from ..file_lock import exclusive_file_lock +# Public extension presentation utility; keep path classification single-owned. +from ..public_safe_text import find_public_safe_local_path as find_public_safe_local_path from .runtime import ( MAX_EXTENSION_RESPONSE_BYTES, _resolved_active_extension, diff --git a/packages/loopx-finance-value-discovery/src/loopx_finance_value_discovery/presentation_compat.py b/packages/loopx-finance-value-discovery/src/loopx_finance_value_discovery/presentation_compat.py index 24d35d53e6..b6ea0ca3bd 100644 --- a/packages/loopx-finance-value-discovery/src/loopx_finance_value_discovery/presentation_compat.py +++ b/packages/loopx-finance-value-discovery/src/loopx_finance_value_discovery/presentation_compat.py @@ -9,7 +9,6 @@ _PRESENTATION_ATTRIBUTES = ( "load_presentation_view_validator", ) -_PUBLIC_PATH_MODULE = "loopx.public_safe_text" _PUBLIC_PATH_ATTRIBUTE = "find_public_safe_local_path" @@ -31,16 +30,7 @@ def presentation_api_error() -> str | None: "the installed loopx version has an incomplete extension presentation " f"API (missing {', '.join(missing)})" ) - try: - public_path_module: ModuleType = import_module(_PUBLIC_PATH_MODULE) - except ModuleNotFoundError as exc: - if exc.name in {_PUBLIC_PATH_MODULE, "loopx"}: - return ( - "the installed loopx version does not provide the shared " - "public-safe local-path classifier required for Finance research dashboards" - ) - raise - if not callable(getattr(public_path_module, _PUBLIC_PATH_ATTRIBUTE, None)): + if not callable(getattr(module, _PUBLIC_PATH_ATTRIBUTE, None)): return ( "the installed loopx version does not provide the shared " "public-safe local-path classifier required for Finance research dashboards" diff --git a/packages/loopx-finance-value-discovery/src/loopx_finance_value_discovery/presentation_validation.py b/packages/loopx-finance-value-discovery/src/loopx_finance_value_discovery/presentation_validation.py index 22fd789281..98db33cc88 100644 --- a/packages/loopx-finance-value-discovery/src/loopx_finance_value_discovery/presentation_validation.py +++ b/packages/loopx-finance-value-discovery/src/loopx_finance_value_discovery/presentation_validation.py @@ -15,7 +15,7 @@ _find_public_safe_local_path: Callable[[str | None], re.Pattern[str] | None] | None try: - from loopx.public_safe_text import ( + from loopx.extensions.presentation import ( find_public_safe_local_path as _find_public_safe_local_path, ) except (ImportError, ModuleNotFoundError): diff --git a/tests/extensions/test_finance_value_discovery_extension.py b/tests/extensions/test_finance_value_discovery_extension.py index dbc9b254d1..528971a6b3 100644 --- a/tests/extensions/test_finance_value_discovery_extension.py +++ b/tests/extensions/test_finance_value_discovery_extension.py @@ -1227,7 +1227,7 @@ def test_declared_minimum_core_without_presentation_api_can_import_and_doctor( ) -def test_dashboard_requires_only_public_presentation_validator_api( +def test_dashboard_requires_only_public_presentation_api( tmp_path: Path, ) -> None: """A Core with the public resolver need not preserve Core-private helpers.""" @@ -1238,6 +1238,7 @@ def test_dashboard_requires_only_public_presentation_validator_api( (core_package / "__init__.py").write_text("") (core_package / "presentation.py").write_text( "def load_presentation_view_validator(_surface):\n return lambda value: value\n" + "def find_public_safe_local_path(_text):\n return None\n" ) environment = dict(os.environ) environment["PYTHONPATH"] = os.pathsep.join( From 93e510af60e77926e617f5f71aa698133a4fba86 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 15 Oct 2026 15:46:36 +0800 Subject: [PATCH 4/4] test(release): align portable runtime and readiness fixtures Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- apps/presentation/dashboard/package.json | 2 +- examples/loopx-chat-runtime-smoke.py | 3 +++ .../public_safe_outbound/tests/test_scan_selection.py | 2 ++ tests/control_plane/test_chat_cold_source_import.py | 4 +++- 4 files changed, 9 insertions(+), 2 deletions(-) diff --git a/apps/presentation/dashboard/package.json b/apps/presentation/dashboard/package.json index d0790b9993..cb9394aebe 100644 --- a/apps/presentation/dashboard/package.json +++ b/apps/presentation/dashboard/package.json @@ -56,7 +56,7 @@ "smoke:status-source-switch-packaged": "LOOPX_STATUS_SOURCE_SWITCH_PACKAGED=1 LOOPX_STATUS_SOURCE_SWITCH_PORT=5198 node ../../../examples/status-source-switch-browser-smoke.mjs", "smoke:status-sources": "rm -rf /tmp/loopx-status-source-smoke && tsc --ignoreConfig --target ES2022 --module CommonJS --moduleResolution Node --ignoreDeprecations 6.0 --skipLibCheck --strict --resolveJsonModule --esModuleInterop --outDir /tmp/loopx-status-source-smoke smoke/status-source-catalog-smoke.ts src/data/ssh-host-catalog.ts src/data/status-source-catalog.ts src/data/local-status-query.ts src/data/status.ts && NODE_PATH=\"$PWD/node_modules\" node /tmp/loopx-status-source-smoke/apps/presentation/dashboard/smoke/status-source-catalog-smoke.js", "smoke:task-board-scroll": "node smoke/task-board-scroll-smoke.mjs", - "smoke:usage-progress": "rm -rf /tmp/loopx-usage-progress-smoke && tsc --ignoreConfig --target ES2022 --module CommonJS --moduleResolution Node --ignoreDeprecations 6.0 --resolveJsonModule --esModuleInterop --jsx react-jsx --skipLibCheck --strict --rewriteRelativeImportExtensions --outDir /tmp/loopx-usage-progress-smoke smoke/usage-progress-smoke.ts src/features/personal-workspace/personal-workspace-model.ts && node /tmp/loopx-usage-progress-smoke/apps/presentation/dashboard/smoke/usage-progress-smoke.js", + "smoke:usage-progress": "tsc --ignoreConfig --target ES2022 --module ESNext --moduleResolution Bundler --allowImportingTsExtensions --types vite/client --resolveJsonModule --esModuleInterop --jsx react-jsx --skipLibCheck --strict --noEmit smoke/usage-progress-smoke.ts src/features/personal-workspace/personal-workspace-model.ts && vite build --ssr smoke/usage-progress-smoke.ts --outDir node_modules/.cache/loopx-usage-progress-smoke --emptyOutDir && node node_modules/.cache/loopx-usage-progress-smoke/usage-progress-smoke.js", "smoke:action-review-plan": "tsc --noEmit --ignoreConfig --target ES2022 --module ES2022 --moduleResolution Bundler --skipLibCheck --strict smoke/action-review-plan-smoke.ts src/vite-env.d.ts && vite build --ssr smoke/action-review-plan-smoke.ts --outDir node_modules/.cache/loopx-action-review-plan --emptyOutDir && node node_modules/.cache/loopx-action-review-plan/action-review-plan-smoke.js", "smoke:attention-details": "tsc --ignoreConfig --target ES2022 --module CommonJS --moduleResolution Node --ignoreDeprecations 6.0 --resolveJsonModule --esModuleInterop --jsx react-jsx --skipLibCheck --strict --outDir /tmp/loopx-attention-details-smoke smoke/attention-details-smoke.ts && NODE_PATH=\"$PWD/node_modules\" node /tmp/loopx-attention-details-smoke/apps/presentation/dashboard/smoke/attention-details-smoke.js", "smoke:delivery-review": "tsc --ignoreConfig --target ES2022 --module NodeNext --moduleResolution NodeNext --skipLibCheck --strict --rootDir src --outDir node_modules/.cache/delivery-review src/data/delivery-review.ts src/data/goal-work-map.ts src/data/goal-acceptance-observation.ts src/features/personal-workspace/delivery-review-copy.ts && node smoke/delivery-review-smoke.mjs", diff --git a/examples/loopx-chat-runtime-smoke.py b/examples/loopx-chat-runtime-smoke.py index 8ff5d65488..62321087e0 100644 --- a/examples/loopx-chat-runtime-smoke.py +++ b/examples/loopx-chat-runtime-smoke.py @@ -37,6 +37,9 @@ result = {"serverInfo": {"name": "fake-runtime-codex"}} elif method in {"thread/start", "thread/resume"}: result = {"thread": {"id": "durable-thread"}} + settings = (request.get("params", {}).get("config") or {}).get("sandbox_workspace_write") + if settings is not None: + result["sandbox"] = {"type": "workspaceWrite", "writableRoots": settings["writable_roots"]} elif method in {"thread/goal/set", "thread/goal/get"}: print(json.dumps({ "id": request_id, diff --git a/loopx/capabilities/public_safe_outbound/tests/test_scan_selection.py b/loopx/capabilities/public_safe_outbound/tests/test_scan_selection.py index fa116e4b62..b202a5a611 100644 --- a/loopx/capabilities/public_safe_outbound/tests/test_scan_selection.py +++ b/loopx/capabilities/public_safe_outbound/tests/test_scan_selection.py @@ -199,6 +199,7 @@ def test_scan_root_cli_reports_a_hit_without_content_or_absolute_paths(tmp_path: ], capture_output=True, text=True, + encoding="utf-8", check=False, # Only the generic rules: a marker file from the caller's shell must # not change what this test observes. @@ -226,6 +227,7 @@ def _scan_root_cli(root: Path) -> subprocess.CompletedProcess[str]: ], capture_output=True, text=True, + encoding="utf-8", check=False, env={key: value for key, value in os.environ.items() if key != "PUBLIC_SAFE_CONFIG"}, ) diff --git a/tests/control_plane/test_chat_cold_source_import.py b/tests/control_plane/test_chat_cold_source_import.py index bab0eca6f0..b58658484a 100644 --- a/tests/control_plane/test_chat_cold_source_import.py +++ b/tests/control_plane/test_chat_cold_source_import.py @@ -48,7 +48,9 @@ def cold_api(tmp_path, monkeypatch, request, capture_present): from loopx.chat_server import serve_chat registry, runtime, ready = map(Path, sys.argv[1:4]) port = int(sys.argv[4]) -ready.write_text(json.dumps({'port': port, 'package': loopx.__file__})) +staged_ready = ready.with_suffix('.tmp') +staged_ready.write_text(json.dumps({'port': port, 'package': loopx.__file__}), encoding='utf-8') +staged_ready.replace(ready) serve_chat(registry_path=registry, runtime_root_override=runtime, scan_roots=[Path(json.loads(registry.read_text())['goals'][0]['repo'])], host='127.0.0.1', port=port, goal_id='cold', open_browser=False, verbose=False)