Isolate native receiver preflight before starting benchmark solver time

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())

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; '

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"]

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'):

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,
     )

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

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: