From 80963767f4eae66bed381c7a14288bd8617c2103 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 7 Nov 2026 04:54:30 +0800 Subject: [PATCH 1/2] fix(benchmark): diagnose and bound native receiver startup retries Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- benchmark/runtime/RUNTIME.md | 12 +- benchmark/runtime/codex_readiness.py | 170 ++++++++++++++++------- benchmark/tests/test_codex_readiness.py | 176 +++++++++++++++++++++++- 3 files changed, 301 insertions(+), 57 deletions(-) diff --git a/benchmark/runtime/RUNTIME.md b/benchmark/runtime/RUNTIME.md index b95b418d65..017f507290 100644 --- a/benchmark/runtime/RUNTIME.md +++ b/benchmark/runtime/RUNTIME.md @@ -7,9 +7,15 @@ 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 -one startup observation reads the account and usage route without requesting a -credential refresh or invoking a model. An unavailable route exits with -`native_receiver_unavailable`; repair the receiver's proxy/login and retry its +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 diff --git a/benchmark/runtime/codex_readiness.py b/benchmark/runtime/codex_readiness.py index 63662f4789..4fd8225981 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) 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..54964004bc 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,162 @@ 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"]) + 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) - check_native_receiver(env) - assert budgets == [21, 16] + 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): + 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) + 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): + 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) + + +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"]) From 1a309ccb3278b68f79d674b66833fadffd8ad476 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 7 Nov 2026 04:56:33 +0800 Subject: [PATCH 2/2] fix(runtime): charge native startup against the original observation deadline Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- benchmark/runtime/codex_readiness.py | 2 +- benchmark/tests/test_codex_readiness.py | 43 +++++++++++++++++-- .../control_plane/turn_driver/codex_stdio.py | 22 ++++++++-- 3 files changed, 60 insertions(+), 7 deletions(-) diff --git a/benchmark/runtime/codex_readiness.py b/benchmark/runtime/codex_readiness.py index 4fd8225981..3d35830f41 100644 --- a/benchmark/runtime/codex_readiness.py +++ b/benchmark/runtime/codex_readiness.py @@ -83,7 +83,7 @@ def _observe_once(env: dict[str, str], deadline: float) -> dict: native_env = {**env, "CODEX_HOME": str(observer)} timeout = _remaining(deadline) stage = "initialize" - with native_checkpoint_reader(native_env, timeout=timeout) as reader: + 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) diff --git a/benchmark/tests/test_codex_readiness.py b/benchmark/tests/test_codex_readiness.py index 54964004bc..02804c4707 100644 --- a/benchmark/tests/test_codex_readiness.py +++ b/benchmark/tests/test_codex_readiness.py @@ -120,7 +120,7 @@ def request(self, method, params): return {"account": {"type": "chatgpt"}} if method == "account/read" else {"rateLimits": {}} @contextmanager - def observer(native_env, *, timeout): + 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() @@ -191,7 +191,7 @@ def request(self, method, params): return {"account": {"type": "chatgpt"}} if method == "account/read" else {"rateLimits": {}} @contextmanager - def observer(native_env, *, timeout): + def observer(native_env, *, timeout, deadline): instances.append(native_env["CODEX_HOME"]) if expire_at == "initialize": now[0] = 126 @@ -222,7 +222,7 @@ def request(self, method, params): return {"account": {"type": "chatgpt"}} if method == "account/read" else {"rateLimits": {}} @contextmanager - def observer(native_env, *, timeout): + def observer(native_env, *, timeout, deadline): instance = Path(native_env["CODEX_HOME"]) assert instance != home assert not (instance / "prior-observer-write").exists() @@ -244,6 +244,43 @@ def observer(native_env, *, timeout): 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") 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