From fbdc2fb8d7c7e5592c364c29662b06e8882bd995 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 7 Nov 2026 00:07:28 +0800 Subject: [PATCH] fix(benchmark): isolate native inspection from the trial profile Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- benchmark/runtime/codex_readiness.py | 68 +++++++++ benchmark/runtime/sforge.py | 10 +- benchmark/tests/test_codex_readiness.py | 136 ++++++++++++++++++ benchmark/tests/test_sforge_runtime.py | 2 + benchmark/tests/test_shared_codex_runtime.py | 1 + .../control_plane/turn_driver/codex_stdio.py | 15 ++ loopx/hosts/codex_checkpoint.py | 18 +-- 7 files changed, 233 insertions(+), 17 deletions(-) create mode 100644 benchmark/runtime/codex_readiness.py create mode 100644 benchmark/tests/test_codex_readiness.py diff --git a/benchmark/runtime/codex_readiness.py b/benchmark/runtime/codex_readiness.py new file mode 100644 index 0000000000..63662f4789 --- /dev/null +++ b/benchmark/runtime/codex_readiness.py @@ -0,0 +1,68 @@ +"""Check a receiver's native login route without starting a model Turn. + +This startup-only transport check owns neither quota policy nor recovery of an +already dispatched request. A failed observer leaves the original worker, +session and scheduler records untouched. +""" +from __future__ import annotations + +import json +import os +import time +import shutil +import tempfile +from pathlib import Path + +from loopx.control_plane.turn_driver.codex_stdio import native_checkpoint_reader + + +def check_native_receiver(env: dict[str, str]) -> None: + # Match the account owner's existing observation budget. This limits only + # this disposable observer, never the solver or an existing native Turn. + deadline = time.monotonic() + 25 + # Native app-server can install bundled skills even for account-only reads. + # Probe the selected login/config/workspace in a disposable inspection home; + # never create those files in the receiver before its fixed-input owner runs. + # This checks account transport, not a future model request or session store. + home = Path(env["LOOPX_CODEX_HOME"]) + with tempfile.TemporaryDirectory(prefix="loopx-native-inspection-") as temporary: + observer = Path(temporary) + for name in ("auth.json", "config.toml"): + source = home / name + if source.is_file(): + shutil.copyfile(source, observer / name) + (observer / name).chmod(0o600) + native_env = {**env, "CODEX_HOME": str(observer)} + with native_checkpoint_reader(native_env, timeout=25) as reader: + def request(method, params): + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError("Native receiver observation expired") + reader.response_timeout_sec = remaining + return reader.request(method, params) + + # Exercise the selected native binary with the original workspace + # and copied login/config, rather than trusting login-file presence. + account = request("account/read", {"refreshToken": False}).get("account") + if not isinstance(account, dict) or account.get("type") != "chatgpt": + raise ValueError("Native receiver has no ChatGPT account") + usage = request("account/rateLimits/read", {}) + if not isinstance(usage, dict) or not usage: + raise ValueError("Native receiver usage route is unavailable") + + +def main() -> int: + try: + check_native_receiver(dict(os.environ)) + except Exception: + # Native transport errors can contain credentials, account identifiers + # or private routes. Do not copy their messages into runner output. + print(json.dumps({"ok": False, "code": "native_receiver_unavailable", + "inference_invoked": False}), flush=True) + return 1 + print(json.dumps({"ok": True, "inference_invoked": False}), flush=True) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/benchmark/runtime/sforge.py b/benchmark/runtime/sforge.py index cdbf3048bc..5758465132 100644 --- a/benchmark/runtime/sforge.py +++ b/benchmark/runtime/sforge.py @@ -500,8 +500,16 @@ def format_run_cmd(self, prompt_path, *, model=None, cwd="", internet=True, resu # environment. Do not duplicate it in the logged shell command. env.pop("OPENAI_API_KEY", None) exports = " ".join(f"{key}={shlex.quote(value)}" for key, value in env.items()) + # Run once at receiver execution entry, before its phase clock and + # scheduler. Do not turn route failures into an unknown native request + # or count setup failure against solver time. Custom API providers keep + # their own admission path; no observer runs on every heartbeat wake. + readiness = "" + if not self._config.agent_api_base_url: + readiness = (f"env {exports} {shlex.quote(f'{_PYTHON}/bin/python3')} " + "-m benchmark.runtime.codex_readiness; ") return self._execution_command( - f"set -eu; {clock}" + f"set -eu; {readiness}{clock}" f"export LOOPX_PHASE_DEADLINE_EPOCH=$(cat {deadline}); " f"remaining=$(( LOOPX_PHASE_DEADLINE_EPOCH - $(date +%s) )); " f'if test "$remaining" -le 0; then test -f {started}; exit $?; fi; ' diff --git a/benchmark/tests/test_codex_readiness.py b/benchmark/tests/test_codex_readiness.py new file mode 100644 index 0000000000..e788d76892 --- /dev/null +++ b/benchmark/tests/test_codex_readiness.py @@ -0,0 +1,136 @@ +"""Receiver startup checks never dispatch a model or rewrite failed work.""" +import json +import os +import subprocess +import sys + +import pytest + +from benchmark.runtime.codex_readiness import check_native_receiver + + +def native_fixture(tmp_path, mode): + home = tmp_path / "native-home" + home.mkdir() + (home / "auth.json").write_text("private synthetic login") + pending = tmp_path / "scheduler.json.host-attempt" + pending.write_text('{"pending":true}') + trace = tmp_path / "requests.jsonl" + binary = tmp_path / "codex" + binary.write_text(f"#!{sys.executable}\n" + ''' +import json, os, sys +for line in sys.stdin: + packet = json.loads(line) + pathlib = __import__("pathlib") + pathlib.Path(os.environ["CODEX_HOME"], "skills", ".system").mkdir(parents=True, exist_ok=True) + with open(os.environ["PROBE_TRACE"], "a") as stream: + stream.write(json.dumps(packet) + "\\n") + if "id" not in packet: + continue + method = packet["method"] + result = {} + if method == "account/read": + result = {"account": {"type": "chatgpt", "email": "private@example.invalid"}} + if method == "account/rateLimits/read": + result = {"rateLimits": {"primary": {"usedPercent": 97}}} + response = {"id": packet["id"], "result": result} + if os.environ["PROBE_MODE"] == method: + response = {"id": packet["id"], "error": {"code": -32603, + "message": "private login and internal route must not escape"}} + print(json.dumps(response), flush=True) +''') + binary.chmod(0o700) + env = {**os.environ, "CODEX_BIN": str(binary), "LOOPX_CODEX_HOME": str(home), + "CODEX_HOME": str(tmp_path / "wrong-home"), "LOOPX_PROJECT": str(tmp_path), + "PROBE_TRACE": str(trace), "PROBE_MODE": mode} + return env, trace, home, pending + + +@pytest.mark.parametrize("failure", [None, "initialize", "account/read", "account/rateLimits/read"]) +def test_cli_checks_receiver_without_model_or_state_mutation(tmp_path, failure): + env, trace, home, pending = native_fixture(tmp_path, failure or "success") + original = {p: p.read_bytes() for p in (home / "auth.json", pending)} + result = subprocess.run([sys.executable, "-m", "benchmark.runtime.codex_readiness"], + env=env, text=True, capture_output=True, timeout=10) + assert result.returncode == (1 if failure else 0) + receipt = json.loads(result.stdout) + assert receipt["ok"] is (failure is None) + assert receipt["inference_invoked"] is False + assert "private" not in result.stdout + result.stderr + if failure: + assert receipt["code"] == "native_receiver_unavailable" + requests = [json.loads(line) for line in trace.read_text().splitlines()] + methods = [r["method"] for r in requests] + expected = ["initialize", "initialized", "account/read", "account/rateLimits/read"] + if failure: + expected = expected[:expected.index(failure) + 1] + assert methods == expected + if "account/read" in methods: + assert requests[methods.index("account/read")]["params"] == {"refreshToken": False} + assert {p: p.read_bytes() for p in original} == original + assert not (home / "skills").exists() + assert not (tmp_path / "phase-deadline").exists() + + +def test_observer_uses_receiver_home_and_one_total_io_budget(tmp_path, monkeypatch): + from contextlib import contextmanager + from benchmark.runtime import codex_readiness + env, _, home, _ = native_fixture(tmp_path, "success") + times = iter([100, 104, 109]) + monkeypatch.setattr(codex_readiness.time, "monotonic", lambda: next(times)) + budgets = [] + + class Reader: + def request(self, method, params): + budgets.append(self.response_timeout_sec) + return {"account": {"type": "chatgpt"}} if method == "account/read" else {"rateLimits": {}} + + @contextmanager + def observer(native_env, *, timeout): + observer_home = __import__("pathlib").Path(native_env["CODEX_HOME"]) + assert observer_home != home + assert (observer_home / "auth.json").read_bytes() == (home / "auth.json").read_bytes() + assert timeout == 25 + yield Reader() + + monkeypatch.setattr(codex_readiness, "native_checkpoint_reader", observer) + check_native_receiver(env) + assert budgets == [21, 16] + + +@pytest.mark.parametrize("profile", ["native-goal", "heartbeat-resume", "heartbeat-explore"]) +def test_sforge_rejects_bad_receiver_before_clock_or_scheduler(tmp_path, monkeypatch, profile): + pytest.importorskip("sforge") + pytest.importorskip("harbor") + from sforge.harness.config import SForgeConfig + from benchmark.runtime.sforge import SForgeWorker, BenchmarkCodex, _SRC, _PYTHON, _CODEX_HOME + from pathlib import Path + + env, trace, home, pending = native_fixture(tmp_path, "account/read") + monkeypatch.setenv("CODEX_AUTH_JSON_PATH", str(home / "auth.json")) + + async def installed(self, environment): + pass + + monkeypatch.setattr(BenchmarkCodex, "install", installed) + worker = SForgeWorker(SForgeConfig(agent_model="fixture", agent_effort="xhigh"), + profile=profile, cwd=str(tmp_path)) + worker.install_stop_hook(None, None, tmp_path, None) + worker.prepared = True + command = worker.format_run_cmd("/task.md") + checkout = str(Path(__file__).resolve().parents[2]) + runtime = tmp_path / "runtime" + command = (command.replace(_SRC, checkout) + .replace(f"{_PYTHON}/bin/python3", sys.executable) + .replace(_CODEX_HOME, str(home)) + .replace("/usr/local/bin/codex", env["CODEX_BIN"]) + .replace("/opt/loopx-benchmark", str(runtime))) + result = subprocess.run(["/bin/bash", "-c", command], env=env, text=True, + capture_output=True, timeout=10) + assert result.returncode == 1 + assert json.loads(result.stdout)["code"] == "native_receiver_unavailable" + assert not (runtime / "control/phase-deadline").exists() + assert not (runtime / "control/execution-started").exists() + assert pending.read_text() == '{"pending":true}' + assert [json.loads(line)["method"] for line in trace.read_text().splitlines()] == [ + "initialize", "initialized", "account/read"] diff --git a/benchmark/tests/test_sforge_runtime.py b/benchmark/tests/test_sforge_runtime.py index e35ec3374c..770973e41e 100644 --- a/benchmark/tests/test_sforge_runtime.py +++ b/benchmark/tests/test_sforge_runtime.py @@ -677,6 +677,7 @@ async def installed(self, environment): assert first == resumed assert ("benchmark.runtime.sforge_entry" in first) is (entry == "loopx-planned") assert "test -f /opt/loopx-benchmark/control/phase-deadline ||" in first + assert first.index("-m benchmark.runtime.codex_readiness") < first.index("test -f /opt/loopx-benchmark/control/phase-deadline ||") assert first.index("export LOOPX_PHASE_DEADLINE_EPOCH") < first.index("timeout --signal") profile = json.loads((tmp_path / "worker-profile.json").read_text()) assert profile["task_entry"] == entry @@ -1211,6 +1212,7 @@ def copy(handle, source, target): worker.prepared = True command = worker.format_run_cmd('/task/prompt.md') assert config.agent_api_key not in command + assert "benchmark.runtime.codex_readiness" not in command assert all(config.agent_api_key not in contents for contents in copied.values()) config.agent_api_key = None with pytest.raises(ValueError, match='API credential'): diff --git a/benchmark/tests/test_shared_codex_runtime.py b/benchmark/tests/test_shared_codex_runtime.py index e82124dd98..257d4de1b0 100644 --- a/benchmark/tests/test_shared_codex_runtime.py +++ b/benchmark/tests/test_shared_codex_runtime.py @@ -198,6 +198,7 @@ def test_native_goal_staged_source_is_importable_without_loopx_control_plane(tmp f"import sys; sys.path.insert(0, {str(staged)!r}); " "import loopx.capabilities.benchmark_toolkit.native_codex_goal as goal; " "import loopx.capabilities.benchmark_toolkit.native_codex_process as process; " + "import benchmark.runtime.codex_readiness; " "print(goal.__name__, process.__name__)"], cwd=tmp_path, env={"PATH": os.environ["PATH"]}, capture_output=True, text=True, ) diff --git a/loopx/control_plane/turn_driver/codex_stdio.py b/loopx/control_plane/turn_driver/codex_stdio.py index e59d1ae71c..f727704918 100644 --- a/loopx/control_plane/turn_driver/codex_stdio.py +++ b/loopx/control_plane/turn_driver/codex_stdio.py @@ -6,6 +6,7 @@ import subprocess import threading import time +from contextlib import contextmanager from collections import deque from collections.abc import Mapping, Sequence from typing import Any, Self, TextIO @@ -160,3 +161,17 @@ def close(self) -> None: except subprocess.TimeoutExpired: self.process.kill() self.process.wait(timeout=2) + + +@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 diff --git a/loopx/hosts/codex_checkpoint.py b/loopx/hosts/codex_checkpoint.py index c033ccf537..0f2e7b9df8 100644 --- a/loopx/hosts/codex_checkpoint.py +++ b/loopx/hosts/codex_checkpoint.py @@ -19,10 +19,10 @@ import time import tomllib from pathlib import Path -from contextlib import contextmanager, nullcontext, redirect_stdout, redirect_stderr +from contextlib import nullcontext, redirect_stdout, redirect_stderr from typing import Any -from loopx.control_plane.turn_driver.codex_stdio import StdioNativeGoalTransport +from loopx.control_plane.turn_driver.codex_stdio import StdioNativeGoalTransport, native_checkpoint_reader def _complete_history(request, session_id: str, *, interrupted_guard: dict | None = None) -> list[dict[str, Any]]: @@ -143,20 +143,6 @@ 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_thread_readback(request, env: dict[str, str], session_id: str) -> None: readback = request("thread/read", {"threadId": session_id, "includeTurns": False}) if readback.get("thread", {}).get("id") != session_id: