From 39a8849ec23a4384ca1573ee22ece1fb5ba9cf76 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 7 Nov 2026 15:53:21 +0800 Subject: [PATCH] fix(benchmark): bound and diagnose native receiver startup Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- benchmark/runtime/RUNTIME.md | 16 ++ benchmark/runtime/codex_readiness.py | 170 ++++++++++---- benchmark/tests/test_codex_readiness.py | 215 +++++++++++++++++- .../control_plane/turn_driver/codex_stdio.py | 22 +- 4 files changed, 365 insertions(+), 58 deletions(-) diff --git a/benchmark/runtime/RUNTIME.md b/benchmark/runtime/RUNTIME.md index f2aa6517d8..7064e295a3 100644 --- a/benchmark/runtime/RUNTIME.md +++ b/benchmark/runtime/RUNTIME.md @@ -5,6 +5,22 @@ LHTB, SWE-Marathon and other Harbor tasks use `PYTHONPATH`. This research runner is not another installed product package. Native tasks, environment, phases, feedback, verifier and scores stay in Harbor. +SForge LoopX workers using a native ChatGPT login check that login through the +receiver's native app-server before entering the phase clock or scheduler. This +startup observation reads the account and usage route without requesting a +credential refresh or invoking a model. It preserves a single 25-second IO +window across preparation, initialization and reads. An explicit native +transport disconnect or timeout can open at most three disposable observers +within that window; rejected RPCs, invalid accounts and unknown failures do not +retry. Process cleanup may finish after the IO deadline. An unavailable route +exits with `native_receiver_unavailable` and bounded local stage, error category, +attempt and elapsed-time diagnostics; raw errors, native stderr and account +responses are not emitted. Repair the receiver's proxy/login and retry its +ordinary entry. Existing phase deadlines, sessions, pending requests and retry +limits are retained. This check does not recover an already dispatched failed +Turn, extend its budget, select an account or authorize a retry. Official/plain +workers and custom API routes keep their existing entry behavior. + The shared adapter reads `PATH` from the task container before installing its isolated profile. Worker and login shells retain those task toolchain directories; LoopX modes prepend their staged Node and CLI. No operator-host PATH or other diff --git a/benchmark/runtime/codex_readiness.py b/benchmark/runtime/codex_readiness.py index 63662f4789..3d35830f41 100644 --- a/benchmark/runtime/codex_readiness.py +++ b/benchmark/runtime/codex_readiness.py @@ -8,60 +8,136 @@ import json import os -import time import shutil import tempfile +import time 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") +from loopx.control_plane.turn_driver.codex_stdio import ( + NativeGoalProtocolError, + native_checkpoint_reader, +) +# These are exact machine codes emitted by the shared native transport, not +# classifications of remote error messages. Unknown errors remain terminal. +_TRANSPORT_TIMEOUTS = frozenset({ + "app_server_response_timeout:initialize", + "app_server_response_timeout:account/read", + "app_server_response_timeout:account/rateLimits/read", +}) +_TRANSPORT_DISCONNECTS = frozenset({ + "app_server_stream_closed", "app_server_stdin_closed", + "app_server_write_failed", "app_server_exited_before_response", +}) -def main() -> int: + +class _ObservationExpired(TimeoutError): + """The local absolute observation deadline was exhausted.""" + + +class _InvalidObservation(Exception): + """A local validation failure; never populated from a remote response.""" + + +def _failure(exc: Exception) -> tuple[str, bool]: + if isinstance(exc, _ObservationExpired): + return "timeout", True + if isinstance(exc, NativeGoalProtocolError): + if str(exc) in _TRANSPORT_TIMEOUTS: + return "timeout", True + if str(exc) in _TRANSPORT_DISCONNECTS: + return "disconnected", True + return "protocol_rejected", False + if isinstance(exc, _InvalidObservation): + return str(exc), False + if isinstance(exc, OSError): + return "local_io", False + if isinstance(exc, KeyError): + return "configuration", False + return "unexpected", False + + +def _remaining(deadline: float) -> float: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise _ObservationExpired + return remaining + + +def _observe_once(env: dict[str, str], deadline: float) -> dict: + started = time.monotonic() + stage = "inspection_home" 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 + _remaining(deadline) + # Native app-server can install bundled skills during account-only + # reads. Each observer gets fresh copied login/config, never the solver's + # session store; a retry cannot carry the previous observer's writes. + 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)} + timeout = _remaining(deadline) + stage = "initialize" + with native_checkpoint_reader(native_env, timeout=timeout, deadline=deadline) as reader: + def request(method, params): + reader.response_timeout_sec = _remaining(deadline) + return reader.request(method, params) + + stage = "account_read" + account = request("account/read", {"refreshToken": False}).get("account") + if not isinstance(account, dict) or account.get("type") != "chatgpt": + raise _InvalidObservation("invalid_account") + stage = "rate_limits" + usage = request("account/rateLimits/read", {}) + if not isinstance(usage, dict) or not usage: + raise _InvalidObservation("unavailable_rate_limits") + _remaining(deadline) + except Exception as exc: + error_kind, retryable = _failure(exc) + return {"ok": False, "stage": stage, "error_kind": error_kind, + "retryable": retryable, "elapsed_seconds": round(time.monotonic() - started, 3)} + return {"ok": True, "stage": stage, + "elapsed_seconds": round(time.monotonic() - started, 3)} + + +def check_native_receiver(env: dict[str, str]) -> dict: + # Retain the existing total observation budget, including home preparation, + # initialization, reads and retry backoff. Process cleanup may finish later; + # it never grants another request beyond this deadline or starts a model. + started = time.monotonic() + deadline = started + 25 + attempts = [] + for index in range(3): + if index: + remaining = deadline - time.monotonic() + if remaining <= 0: + break + time.sleep(min(0.2, remaining)) + if time.monotonic() >= deadline: + break + observation = _observe_once(env, deadline) + attempts.append(observation) + if observation["ok"] or not observation["retryable"]: + break + ok = attempts[-1]["ok"] + result = {"ok": ok, "inference_invoked": False, "attempts": attempts, + "elapsed_seconds": round(time.monotonic() - started, 3)} + if not ok: + result["code"] = "native_receiver_unavailable" + return result + + +def main() -> int: + # Emit only bounded local stage/category/timing diagnostics. Never include + # raw exception text, native stderr, login values or account/RPC payloads. + result = check_native_receiver(dict(os.environ)) + print(json.dumps(result), flush=True) + return 0 if result["ok"] else 1 if __name__ == "__main__": diff --git a/benchmark/tests/test_codex_readiness.py b/benchmark/tests/test_codex_readiness.py index e788d76892..02804c4707 100644 --- a/benchmark/tests/test_codex_readiness.py +++ b/benchmark/tests/test_codex_readiness.py @@ -3,6 +3,7 @@ import os import subprocess import sys +from pathlib import Path import pytest @@ -19,6 +20,12 @@ def native_fixture(tmp_path, mode): binary = tmp_path / "codex" binary.write_text(f"#!{sys.executable}\n" + ''' import json, os, sys +from pathlib import Path +counter = Path(os.environ["PROBE_ATTEMPTS"]) +attempt = int(counter.read_text()) + 1 if counter.exists() else 1 +counter.write_text(str(attempt)) +mode = os.environ["PROBE_MODE"] +print("private native stderr must not escape", file=sys.stderr) for line in sys.stdin: packet = json.loads(line) pathlib = __import__("pathlib") @@ -28,12 +35,25 @@ def native_fixture(tmp_path, mode): if "id" not in packet: continue method = packet["method"] + if mode in ("disconnect", "disconnect_once") and method == "initialize": + if mode == "disconnect" or attempt == 1: + sys.exit(0) + if mode == "malformed": + print("private malformed frame", flush=True) + continue result = {} if method == "account/read": result = {"account": {"type": "chatgpt", "email": "private@example.invalid"}} if method == "account/rateLimits/read": result = {"rateLimits": {"primary": {"usedPercent": 97}}} + if mode == "invalid_account" and method == "account/read": + result = {"account": {"type": "apiKey"}} + if mode == "empty_usage" and method == "account/rateLimits/read": + result = {} response = {"id": packet["id"], "result": result} + if mode == "forged_timeout" and method == "account/read": + response = {"id": packet["id"], "error": { + "code": -32603, "message": "app_server_response_timeout:account/read"}} if os.environ["PROBE_MODE"] == method: response = {"id": packet["id"], "error": {"code": -32603, "message": "private login and internal route must not escape"}} @@ -42,7 +62,8 @@ def native_fixture(tmp_path, mode): 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} + "PROBE_TRACE": str(trace), "PROBE_MODE": mode, + "PROBE_ATTEMPTS": str(tmp_path / "attempts")} return env, trace, home, pending @@ -59,6 +80,11 @@ def test_cli_checks_receiver_without_model_or_state_mutation(tmp_path, failure): assert "private" not in result.stdout + result.stderr if failure: assert receipt["code"] == "native_receiver_unavailable" + assert receipt["attempts"][0]["stage"] == { + "initialize": "initialize", "account/read": "account_read", + "account/rateLimits/read": "rate_limits"}[failure] + assert receipt["attempts"][0]["error_kind"] == "protocol_rejected" + assert len(receipt["attempts"]) == 1 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"] @@ -76,26 +102,199 @@ def test_observer_uses_receiver_home_and_one_total_io_budget(tmp_path, monkeypat 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)) + now = [100.0] + monkeypatch.setattr(codex_readiness.time, "monotonic", lambda: now[0]) + original_copy = codex_readiness.shutil.copyfile + + def copy(*args): + now[0] += 4 + return original_copy(*args) + + monkeypatch.setattr(codex_readiness.shutil, "copyfile", copy) budgets = [] class Reader: def request(self, method, params): budgets.append(self.response_timeout_sec) + now[0] += 5 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"]) + def observer(native_env, *, timeout, deadline): + observer_home = Path(native_env["CODEX_HOME"]) assert observer_home != home assert (observer_home / "auth.json").read_bytes() == (home / "auth.json").read_bytes() - assert timeout == 25 + assert timeout == 21 # Copy time is charged before initialization. + now[0] += 5 + yield Reader() + + monkeypatch.setattr(codex_readiness, "native_checkpoint_reader", observer) + result = check_native_receiver(env) + assert result["ok"] is True + assert budgets == [16, 11] + assert result["elapsed_seconds"] == 19 + + +@pytest.mark.parametrize("mode,ok,attempts,stage,kind", [ + ("disconnect_once", True, 2, "initialize", "disconnected"), + ("disconnect", False, 3, "initialize", "disconnected"), + ("forged_timeout", False, 1, "account_read", "protocol_rejected"), + ("malformed", False, 1, "initialize", "protocol_rejected"), + ("invalid_account", False, 1, "account_read", "invalid_account"), + ("empty_usage", False, 1, "rate_limits", "unavailable_rate_limits"), +]) +def test_real_native_cli_retry_and_failure_boundaries(tmp_path, mode, ok, attempts, stage, kind): + env, trace, home, pending = native_fixture(tmp_path, mode) + 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) + receipt = json.loads(result.stdout) + assert result.returncode == (0 if ok else 1) + assert receipt["ok"] is ok + assert receipt["inference_invoked"] is False + assert len(receipt["attempts"]) == attempts + assert int(Path(env["PROBE_ATTEMPTS"]).read_text()) == attempts + assert receipt["attempts"][0]["stage"] == stage + assert receipt["attempts"][0]["error_kind"] == kind + assert "private" not in result.stdout + result.stderr + methods = [json.loads(line)["method"] for line in trace.read_text().splitlines()] + assert set(methods) <= {"initialize", "initialized", "account/read", "account/rateLimits/read"} + assert {p: p.read_bytes() for p in original} == original + assert not (home / "skills").exists() + assert not (tmp_path / "phase-deadline").exists() + + +@pytest.mark.parametrize("expire_at", ["copy", "initialize", "account_read", "rate_limits"]) +def test_total_deadline_cannot_be_reset_by_retry(tmp_path, monkeypatch, expire_at): + from contextlib import contextmanager + from benchmark.runtime import codex_readiness + from loopx.control_plane.turn_driver.codex_stdio import NativeGoalProtocolError + env, _, _, _ = native_fixture(tmp_path, "success") + now = [100.0] + monkeypatch.setattr(codex_readiness.time, "monotonic", lambda: now[0]) + instances = [] + original_copy = codex_readiness.shutil.copyfile + + def copy(*args): + if expire_at == "copy": + now[0] = 126 + return original_copy(*args) + + monkeypatch.setattr(codex_readiness.shutil, "copyfile", copy) + + class Reader: + def request(self, method, params): + if (expire_at, method) in (("account_read", "account/read"), + ("rate_limits", "account/rateLimits/read")): + now[0] = 126 + raise NativeGoalProtocolError("app_server_response_timeout:" + method) + return {"account": {"type": "chatgpt"}} if method == "account/read" else {"rateLimits": {}} + + @contextmanager + def observer(native_env, *, timeout, deadline): + instances.append(native_env["CODEX_HOME"]) + if expire_at == "initialize": + now[0] = 126 + raise NativeGoalProtocolError("app_server_response_timeout:initialize") yield Reader() monkeypatch.setattr(codex_readiness, "native_checkpoint_reader", observer) - check_native_receiver(env) - assert budgets == [21, 16] + receipt = check_native_receiver(env) + assert receipt["ok"] is False + assert len(receipt["attempts"]) == 1 + assert receipt["attempts"][0]["error_kind"] == "timeout" + assert receipt["attempts"][0]["stage"] == ("inspection_home" if expire_at == "copy" else expire_at) + assert len(instances) == (0 if expire_at == "copy" else 1) + + +def test_retry_charges_cleanup_and_backoff_and_uses_a_fresh_home(tmp_path, monkeypatch): + from contextlib import contextmanager + from benchmark.runtime import codex_readiness + from loopx.control_plane.turn_driver.codex_stdio import NativeGoalProtocolError + env, _, home, _ = native_fixture(tmp_path, "success") + now = [100.0] + monkeypatch.setattr(codex_readiness.time, "monotonic", lambda: now[0]) + monkeypatch.setattr(codex_readiness.time, "sleep", lambda seconds: now.__setitem__(0, now[0] + seconds)) + instances, budgets = [], [] + + class Reader: + def request(self, method, params): + return {"account": {"type": "chatgpt"}} if method == "account/read" else {"rateLimits": {}} + + @contextmanager + def observer(native_env, *, timeout, deadline): + instance = Path(native_env["CODEX_HOME"]) + assert instance != home + assert not (instance / "prior-observer-write").exists() + assert (instance / "auth.json").read_bytes() == (home / "auth.json").read_bytes() + instances.append(instance) + budgets.append(timeout) + if len(instances) == 1: + (instance / "prior-observer-write").touch() + now[0] += 5 # failed transport plus its cleanup + raise NativeGoalProtocolError("app_server_stream_closed") + yield Reader() + + monkeypatch.setattr(codex_readiness, "native_checkpoint_reader", observer) + receipt = check_native_receiver(env) + assert receipt["ok"] is True + assert len(instances) == 2 and len(set(instances)) == 2 + assert all(not instance.exists() for instance in instances) + assert budgets == pytest.approx([25, 19.8]) + assert receipt["elapsed_seconds"] == pytest.approx(5.2) + + +@pytest.mark.parametrize("expiry", ["spawn", "initialize"]) +def test_native_reader_rechecks_deadline_after_spawn_and_before_notify(tmp_path, monkeypatch, expiry): + from benchmark.runtime import codex_readiness + from loopx.control_plane.turn_driver import codex_stdio + env, _, _, _ = native_fixture(tmp_path, "success") + now = [100.0] + monkeypatch.setattr(codex_readiness.time, "monotonic", lambda: now[0]) + calls = [] + + class Transport: + def __enter__(self): + return self + def __exit__(self, *_): + calls.append("closed") + def request(self, method, params): + calls.append((method, now[0], self.response_timeout_sec)) + now[0] = 126 + return {} + def notify(self, method, params): + calls.append(method) + + def spawn(*args, **kwargs): + calls.append("spawn") + if expiry == "spawn": + now[0] = 126 + return Transport() + + monkeypatch.setattr(codex_stdio.StdioNativeGoalTransport, "spawn", spawn) + result = check_native_receiver(env) + assert result["ok"] is False + assert len(result["attempts"]) == 1 + assert result["attempts"][0]["stage"] == "initialize" + assert result["attempts"][0]["error_kind"] == "timeout" + assert calls == (["spawn", "closed"] if expiry == "spawn" else [ + "spawn", ("initialize", 100, 25), "closed"]) + + +def test_unknown_exception_and_local_configuration_fail_closed(tmp_path, monkeypatch): + from benchmark.runtime import codex_readiness + env, _, _, _ = native_fixture(tmp_path, "success") + def observer(*args, **kwargs): + raise RuntimeError("private login details and app_server_stream_closed") + monkeypatch.setattr(codex_readiness, "native_checkpoint_reader", observer) + receipt = check_native_receiver(env) + assert len(receipt["attempts"]) == 1 + assert receipt["attempts"][0]["error_kind"] == "unexpected" + assert "private" not in json.dumps(receipt) + receipt = check_native_receiver({}) + assert len(receipt["attempts"]) == 1 + assert receipt["attempts"][0]["stage"] == "inspection_home" + assert receipt["attempts"][0]["error_kind"] == "configuration" @pytest.mark.parametrize("profile", ["native-goal", "heartbeat-resume", "heartbeat-explore"]) diff --git a/loopx/control_plane/turn_driver/codex_stdio.py b/loopx/control_plane/turn_driver/codex_stdio.py index f727704918..d35543499e 100644 --- a/loopx/control_plane/turn_driver/codex_stdio.py +++ b/loopx/control_plane/turn_driver/codex_stdio.py @@ -164,14 +164,30 @@ def close(self) -> None: @contextmanager -def native_checkpoint_reader(env: dict[str, str], *, timeout: float): - """Keep one native inspection process open across a sealed capture window.""" +def native_checkpoint_reader( + env: dict[str, str], *, timeout: float, deadline: float | None = None, +): + """Keep one native inspection process open across a sealed capture window. + + Startup observers may supply their existing monotonic deadline so spawning + cannot reset their IO budget. Other inspection callers retain relative timeouts. + """ + def observation_timeout() -> float: + if deadline is None: + return timeout + remaining = deadline - time.monotonic() + if remaining <= 0: + raise NativeGoalProtocolError("app_server_response_timeout:initialize") + return remaining + 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, + response_timeout_sec=observation_timeout(), ) as transport: + transport.response_timeout_sec = observation_timeout() transport.request("initialize", {"clientInfo": {"name": "loopx_checkpoint", "version": "0.1.0"}, "capabilities": {"experimentalApi": True}}) + observation_timeout() transport.notify("initialized", {}) yield transport