diff --git a/README.md b/README.md
index 959e8af63..83e609a7d 100644
--- a/README.md
+++ b/README.md
@@ -463,6 +463,33 @@ See the [Copilot CLI quickstart example](examples/copilot_cli_quickstart/) for a
+
+🧩 Codex App Server (No API Key)
+
+Use the [Codex CLI's app-server](https://developers.openai.com/codex/app-server/) as the LLM backend — no API keys needed. Authentication uses your ChatGPT (Codex) subscription: the app-server owns the managed browser/device-code login and refreshes the tokens itself. OpenEvolve never reads the CLI's token files, and before each generation it verifies with the app-server that a ChatGPT-plan login is active (it fails with a clear message pointing at `codex login` otherwise). To bill an API key instead, keep the default OpenAI provider — both paths coexist.
+
+```bash
+# Install and authenticate
+npm install -g @openai/codex
+codex login # browser flow, or: codex login --device-code
+```
+
+```yaml
+# config.yaml
+llm:
+ provider: "codex_app_server"
+ models:
+ - name: "gpt-5.3-codex"
+ weight: 0.8
+ reasoning_effort: "medium"
+ - name: "gpt-5.3-codex-mini"
+ weight: 0.2
+```
+
+Each call speaks newline-delimited JSON-RPC to a short-lived `codex app-server` process over stdio (unsupported on Windows). Threads run non-interactive: tool approvals are `never` and the sandbox is read-only, so responses are generated without mutating your workspace.
+
+
+
## Examples Gallery
diff --git a/openevolve/llm/__init__.py b/openevolve/llm/__init__.py
index 6c5b938fd..8cc00b8cd 100644
--- a/openevolve/llm/__init__.py
+++ b/openevolve/llm/__init__.py
@@ -7,6 +7,7 @@
from openevolve.llm.openai import OpenAILLM
from openevolve.llm.claude_code import ClaudeCodeLLM, init_claude_code_client
from openevolve.llm.copilot_cli import CopilotCLILLM, init_copilot_cli_client
+from openevolve.llm.codex_app_server import CodexAppServerLLM, init_codex_app_server_client
__all__ = [
"LLMInterface",
@@ -15,5 +16,7 @@
"init_claude_code_client",
"CopilotCLILLM",
"init_copilot_cli_client",
+ "CodexAppServerLLM",
+ "init_codex_app_server_client",
"LLMEnsemble",
]
diff --git a/openevolve/llm/codex_app_server.py b/openevolve/llm/codex_app_server.py
new file mode 100644
index 000000000..b0800bdbc
--- /dev/null
+++ b/openevolve/llm/codex_app_server.py
@@ -0,0 +1,365 @@
+"""
+Codex app-server interface for LLMs.
+
+Uses the Codex CLI's app-server (`codex app-server`) as a non-interactive LLM
+backend, enabling OpenEvolve to run with models available through a ChatGPT
+(Codex) subscription without requiring direct API keys.
+
+Authentication is delegated entirely to the app-server's managed ChatGPT login
+(browser or device-code flow): the app-server persists the tokens itself and
+refreshes them automatically. This adapter never reads the CLI's token files.
+Before running a turn it asks the app-server for the current account state
+(`account/read`) and requires a ChatGPT-plan login — if Codex is not
+authenticated, it fails with actionable instructions instead of falling back
+to an API key.
+
+Usage in config.yaml:
+ llm:
+ provider: "codex_app_server"
+ models:
+ - name: "gpt-5.3-codex"
+ weight: 1.0
+ timeout: 300
+ reasoning_effort: "medium"
+
+Or inject programmatically:
+ from openevolve.llm.codex_app_server import init_codex_app_server_client
+ for model_cfg in config.llm.models:
+ model_cfg.init_client = init_codex_app_server_client
+
+Notes:
+- `temperature`, `top_p` and `max_tokens` have no equivalent app-server option
+ in this first pass and are therefore ignored.
+- Each generation call spawns a fresh `codex app-server` process, speaks
+ newline-delimited JSON-RPC over its stdio (the protocol's default transport),
+ and shuts it down again. Threads are non-interactive: tool approvals are set
+ to "never" and the workspace sandbox to read-only, so the model can answer
+ but not mutate the tree while generating a response.
+- The stdio transport is unsupported on Windows; the backend expects a POSIX
+ shell there.
+"""
+
+import asyncio
+import json
+import logging
+import queue
+import subprocess
+import threading
+import time
+from collections import deque
+from typing import Any, Dict, List, Optional
+
+from openevolve.llm.base import LLMInterface
+
+logger = logging.getLogger(__name__)
+
+# Union of reasoning efforts documented for the GPT-5.x / Codex model families.
+VALID_REASONING_EFFORTS = ("none", "minimal", "low", "medium", "high", "xhigh")
+
+_CLIENT_INFO = {"name": "openevolve", "title": "OpenEvolve", "version": "1.0.0"}
+
+
+class CodexAuthError(RuntimeError):
+ """Raised when the Codex app-server reports no usable ChatGPT login.
+
+ The Codex app-server owns the ChatGPT browser/device-code login and
+ refreshes tokens itself; OpenEvolve never reads or writes the CLI's token
+ files, so the only remedy is to log in via the Codex CLI first.
+ """
+
+ def __init__(self, message: str):
+ super().__init__(
+ f"{message} Run `codex login` (browser flow) or `codex login --device-code` "
+ "to authenticate with your ChatGPT account first."
+ )
+
+
+def _cfg_value(model_cfg: Any, attr: str, default: Any) -> Any:
+ value = getattr(model_cfg, attr, None)
+ return default if value is None else value
+
+
+class _AppServerSession:
+ """Client for one `codex app-server` process speaking JSONL JSON-RPC.
+
+ Requests are written as one JSON object per line to the process's stdin;
+ responses (which echo the request id) and notifications (which omit it) are
+ read line by line from stdout. A background thread feeds a queue so reads
+ can honor a wall-clock deadline instead of blocking forever on a hung
+ app-server.
+ """
+
+ def __init__(self, proc: "subprocess.Popen", stderr_tail: int = 500):
+ self.proc = proc
+ self._next_id = 0
+ self._messages: "queue.Queue[Any]" = queue.Queue()
+ self._pending: "deque[Any]" = deque()
+ self._stderr_tail = stderr_tail
+ self._reader = threading.Thread(target=self._read_stdout, daemon=True)
+ self._reader.start()
+
+ def _read_stdout(self) -> None:
+ try:
+ for line in self.proc.stdout:
+ line = line.strip()
+ if not line:
+ continue
+ try:
+ self._messages.put(json.loads(line))
+ except json.JSONDecodeError:
+ logger.debug(f"Non-JSON line from codex app-server: {line[:200]}")
+ except (ValueError, OSError):
+ pass
+ finally:
+ self._messages.put(None) # EOF sentinel
+
+ def stderr_summary(self) -> str:
+ try:
+ raw = (self.proc.stderr.read() or "") if self.proc.stderr else ""
+ except (ValueError, OSError):
+ raw = ""
+ return raw.strip()[-self._stderr_tail :]
+
+ def request(self, method: str, params: Optional[Dict[str, Any]], deadline: float) -> Any:
+ """Send a request and block until its response arrives or time runs out."""
+ self._next_id += 1
+ request_id = self._next_id
+ self._send({"id": request_id, "method": method, "params": params or {}})
+ while True:
+ message = self.next_message(deadline)
+ if message.get("id") != request_id:
+ # A notification (e.g. an early item/completed) or a late
+ # response: park it instead of dropping it.
+ self._pending.append(message)
+ continue
+ if "error" in message:
+ error = message["error"] or {}
+ raise RuntimeError(
+ f"codex app-server rejected {method}: "
+ f"{error.get('code')}: {error.get('message')}"
+ )
+ return message.get("result")
+
+ def notify(self, method: str, params: Optional[Dict[str, Any]] = None) -> None:
+ self._send({"method": method, "params": params or {}})
+
+ def next_message(self, deadline: float) -> Dict[str, Any]:
+ """Return the next raw JSON-RPC message, honoring the wall-clock deadline."""
+ if self._pending:
+ return self._pending.popleft()
+ remaining = deadline - time.monotonic()
+ if remaining <= 0:
+ raise asyncio.TimeoutError("codex app-server did not respond in time")
+ try:
+ message = self._messages.get(timeout=remaining)
+ except queue.Empty:
+ raise asyncio.TimeoutError("codex app-server did not respond in time")
+ if message is None: # EOF: the process exited
+ raise RuntimeError(
+ f"codex app-server exited unexpectedly. stderr: {self.stderr_summary()}"
+ )
+ return message
+
+ def _send(self, payload: Dict[str, Any]) -> None:
+ try:
+ self.proc.stdin.write(json.dumps(payload) + "\n")
+ self.proc.stdin.flush()
+ except (ValueError, OSError, BrokenPipeError) as e:
+ raise RuntimeError(f"Failed to write to codex app-server: {e}. {self.stderr_summary()}")
+
+ def close(self) -> None:
+ for stream in (self.proc.stdin, self.proc.stdout, self.proc.stderr):
+ try:
+ if stream:
+ stream.close()
+ except (OSError, ValueError):
+ pass
+ try:
+ self.proc.terminate()
+ self.proc.wait(timeout=5)
+ except Exception:
+ try:
+ self.proc.kill()
+ except Exception:
+ pass
+
+
+class CodexAppServerLLM(LLMInterface):
+ """LLM interface that generates through the Codex app-server.
+
+ Requires the `codex` CLI to be installed and logged in with a ChatGPT
+ account (the app-server manages that login and token refresh itself).
+ """
+
+ def __init__(self, model_cfg=None):
+ self.model = _cfg_value(model_cfg, "name", "gpt-5.3-codex")
+ self.system_message = getattr(model_cfg, "system_message", None)
+ self.timeout = _cfg_value(model_cfg, "timeout", 300)
+ self.weight = _cfg_value(model_cfg, "weight", 1.0)
+ self.retries = _cfg_value(model_cfg, "retries", 3)
+ self.retry_delay = _cfg_value(model_cfg, "retry_delay", 5)
+ self.reasoning_effort = getattr(model_cfg, "reasoning_effort", None)
+ self.cwd = getattr(model_cfg, "cwd", None)
+ self.codex_bin = _cfg_value(model_cfg, "codex_bin", "codex")
+ if (
+ self.reasoning_effort is not None
+ and self.reasoning_effort not in VALID_REASONING_EFFORTS
+ ):
+ raise ValueError(
+ f"Invalid reasoning_effort: {self.reasoning_effort}. "
+ f"Expected one of {', '.join(VALID_REASONING_EFFORTS)}"
+ )
+ logger.info(f"Initialized CodexAppServerLLM with model: {self.model}")
+
+ async def generate(self, prompt: str, **kwargs) -> str:
+ sys_msg = kwargs.pop("system_message", self.system_message) or ""
+ return await self.generate_with_context(
+ system_message=sys_msg,
+ messages=[{"role": "user", "content": prompt}],
+ **kwargs,
+ )
+
+ async def generate_with_context(
+ self, system_message: str, messages: List[Dict[str, str]], **kwargs
+ ) -> str:
+ user_content = "\n\n".join(
+ m.get("content", "") for m in messages if m.get("role") == "user"
+ )
+ prompt = f"{system_message}\n\n{user_content}" if system_message else user_content
+
+ timeout = kwargs.get("timeout", self.timeout)
+ retries = kwargs.get("retries", self.retries)
+ retry_delay = kwargs.get("retry_delay", self.retry_delay)
+
+ loop = asyncio.get_event_loop()
+ for attempt in range(retries + 1):
+ try:
+ return await asyncio.wait_for(
+ loop.run_in_executor(None, lambda: self._run_app_server(prompt, timeout)),
+ timeout=timeout + 30,
+ )
+ except FileNotFoundError:
+ logger.error(
+ "The `codex` CLI was not found. Install it and run `codex login` first."
+ )
+ raise
+ except CodexAuthError as e:
+ # Retrying cannot help until the user logs in; fail immediately.
+ logger.error(f"Codex app-server authentication error (not retried): {e}")
+ raise
+ except asyncio.TimeoutError:
+ if attempt < retries:
+ logger.warning(
+ f"Codex app-server timeout on attempt {attempt + 1}/{retries + 1}. Retrying..."
+ )
+ await asyncio.sleep(retry_delay)
+ else:
+ logger.error(f"All {retries + 1} attempts failed with timeout")
+ raise
+ except Exception as e:
+ if attempt < retries:
+ logger.warning(
+ f"Codex app-server error on attempt {attempt + 1}/{retries + 1}: {e}. Retrying..."
+ )
+ await asyncio.sleep(retry_delay)
+ else:
+ logger.error(f"All {retries + 1} attempts failed with error: {e}")
+ raise
+
+ def _build_command(self) -> List[str]:
+ return [self.codex_bin, "app-server"]
+
+ def _run_app_server(self, prompt: str, timeout: int) -> str:
+ deadline = time.monotonic() + timeout
+ session = self._start_session()
+ try:
+ # JSON-RPC handshake (any request before this is rejected by the server).
+ session.request("initialize", {"clientInfo": _CLIENT_INFO}, deadline)
+ session.notify("initialized")
+
+ # Auth boundary: the app-server owns the ChatGPT login and token
+ # refresh; verify it has a usable subscription account up front.
+ self._ensure_authenticated(session, deadline)
+
+ thread_params: Dict[str, Any] = {
+ "model": self.model,
+ "approvalPolicy": "never",
+ "sandbox": {"mode": "read-only", "networkAccess": False},
+ }
+ if self.cwd:
+ thread_params["cwd"] = self.cwd
+ thread_result = session.request("thread/start", thread_params, deadline)
+ thread_id = (thread_result or {}).get("thread", {}).get("id")
+ if not thread_id:
+ raise RuntimeError(f"codex app-server returned no thread id: {thread_result}")
+
+ turn_params: Dict[str, Any] = {
+ "threadId": thread_id,
+ "input": [{"type": "text", "text": prompt}],
+ }
+ if self.reasoning_effort is not None:
+ turn_params["effort"] = self.reasoning_effort
+ session.request("turn/start", turn_params, deadline)
+
+ return self._collect_response(session, deadline)
+ finally:
+ session.close()
+
+ def _ensure_authenticated(self, session: _AppServerSession, deadline: float) -> None:
+ result = session.request("account/read", {}, deadline) or {}
+ if result.get("requiresOpenaiAuth") or result.get("account") is None:
+ raise CodexAuthError("Codex app-server reports the account is not authenticated.")
+ account = result.get("account") or {}
+ if account.get("type") != "chatgpt":
+ raise CodexAuthError(
+ "Codex app-server is not signed in with a ChatGPT account "
+ f"(found auth type '{account.get('type')}'). The codex_app_server "
+ "provider bills a ChatGPT subscription; to use an API key instead, "
+ "keep the default OpenAI provider."
+ )
+
+ def _collect_response(self, session: _AppServerSession, deadline: float) -> str:
+ """Stream notifications until the turn finishes; return the agent message.
+
+ The agent text arrives as `item/completed` notifications
+ (`item.type == "agentMessage"`); `turn/completed` closes the turn.
+ """
+ text: Optional[str] = None
+ while True:
+ message = session.next_message(deadline)
+ if message.get("id") is not None:
+ continue # a response, not a notification
+ method = message.get("method")
+ params = message.get("params") or {}
+ if method == "item/completed":
+ item = params.get("item") or {}
+ if item.get("type") == "agentMessage":
+ text = item.get("text") or ""
+ elif method == "turn/completed":
+ status = (params.get("turn") or {}).get("status")
+ if status and status != "completed":
+ raise RuntimeError(f"codex app-server turn did not complete: {status}")
+ break
+ elif method == "error":
+ raise RuntimeError(f"codex app-server reported an error: {params}")
+ if not text or not text.strip():
+ raise RuntimeError(
+ f"Empty response from codex app-server. stderr: {session.stderr_summary()}"
+ )
+ return text
+
+ def _start_session(self) -> _AppServerSession:
+ proc = subprocess.Popen(
+ self._build_command(),
+ stdin=subprocess.PIPE,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ text=True,
+ cwd=self.cwd,
+ )
+ return _AppServerSession(proc)
+
+
+def init_codex_app_server_client(model_cfg):
+ """Factory function compatible with OpenEvolve's init_client config hook."""
+ return CodexAppServerLLM(model_cfg)
diff --git a/openevolve/llm/ensemble.py b/openevolve/llm/ensemble.py
index f03af713c..e62e66e49 100644
--- a/openevolve/llm/ensemble.py
+++ b/openevolve/llm/ensemble.py
@@ -19,6 +19,7 @@
try:
from openevolve.llm.claude_code import ClaudeCodeLLM
+
_PROVIDER_REGISTRY["claude_code"] = lambda cfg: ClaudeCodeLLM(cfg)
except ImportError:
pass
@@ -30,6 +31,13 @@
except ImportError:
pass
+try:
+ from openevolve.llm.codex_app_server import CodexAppServerLLM
+
+ _PROVIDER_REGISTRY["codex_app_server"] = lambda cfg: CodexAppServerLLM(cfg)
+except ImportError:
+ pass
+
def _create_model(model_cfg: LLMModelConfig) -> LLMInterface:
if model_cfg.init_client:
diff --git a/tests/test_codex_app_server_llm.py b/tests/test_codex_app_server_llm.py
new file mode 100644
index 000000000..9324c3aa5
--- /dev/null
+++ b/tests/test_codex_app_server_llm.py
@@ -0,0 +1,415 @@
+"""Tests for the Codex app-server LLM backend.
+
+The auth boundary is exercised against an in-process fake of `codex
+app-server` that speaks the same newline-delimited JSON-RPC over real OS
+pipes. No network access and no real `codex` binary are involved; the fake
+never touches (and the real backend must never read) the CLI's token files.
+"""
+
+import asyncio
+import json
+import os
+import subprocess
+import threading
+import unittest
+from unittest.mock import MagicMock, patch
+
+from openevolve.llm.codex_app_server import (
+ CodexAppServerLLM,
+ CodexAuthError,
+ init_codex_app_server_client,
+)
+
+AUTHENTICATED_ACCOUNT = {
+ "type": "chatgpt",
+ "email": "user@example.com",
+ "planType": "pro",
+}
+
+
+class FakeAppServer:
+ """In-process fake of `codex app-server` speaking JSONL JSON-RPC on real pipes.
+
+ It validates the protocol the same way the real server does: requests
+ before `initialize` are answered with an error, and the requested methods
+ are recorded so tests can assert on the exact conversation.
+ """
+
+ def __init__(self, scenario="ok"):
+ self.scenario = scenario
+ self.requests = []
+ client_to_server_r, client_to_server_w = os.pipe()
+ server_to_client_r, server_to_client_w = os.pipe()
+ self.stdin = os.fdopen(client_to_server_w, "w") # client writes here
+ self.stdout = os.fdopen(server_to_client_r) # client reads here
+ self._server_reader = os.fdopen(client_to_server_r)
+ self._server_writer = os.fdopen(server_to_client_w, "w")
+ self.stderr = None
+ self.pid = os.getpid()
+ self.returncode = None
+ self._exiting = threading.Event()
+ self._thread = threading.Thread(target=self._serve, daemon=True)
+ self._thread.start()
+
+ # -- Popen-compatible surface used by the client -------------------------
+ def terminate(self):
+ self._exiting.set()
+ try:
+ self._server_reader.close()
+ except OSError:
+ pass
+
+ def kill(self):
+ self.terminate()
+
+ def wait(self, timeout=None):
+ return 0
+
+ # -- Fake server side ----------------------------------------------------
+ def _reply(self, request_id, result):
+ self._send({"id": request_id, "result": result})
+
+ def _error(self, request_id, error):
+ self._send({"id": request_id, "error": error})
+
+ def _notify(self, method, params):
+ self._send({"method": method, "params": params})
+
+ def _send(self, payload):
+ try:
+ self._server_writer.write(json.dumps(payload) + "\n")
+ self._server_writer.flush()
+ except (ValueError, OSError):
+ pass
+
+ def _serve(self):
+ try:
+ for line in self._server_reader:
+ line = line.strip()
+ if not line:
+ continue
+ message = json.loads(line)
+ method = message.get("method")
+ if not method:
+ continue
+ params = message.get("params") or {}
+ self.requests.append((method, params))
+ if not self._dispatch(method, message.get("id"), params):
+ break
+ except (ValueError, OSError):
+ pass
+ finally:
+ try:
+ self._server_writer.close()
+ except OSError:
+ pass
+
+ def _dispatch(self, method, request_id, params):
+ """Handle one request; return False to hang up on the client."""
+ if self.scenario == "hang" and method == "account/read":
+ return True # never reply: the client must hit its deadline
+ if method == "initialize":
+ self._reply(request_id, {"model": "gpt-5.3-codex", "authMode": "chatgpt"})
+ elif method == "initialized":
+ pass # notification, no reply
+ elif method == "account/read":
+ if self.scenario == "unauthenticated":
+ self._reply(request_id, {"account": None, "requiresOpenaiAuth": True})
+ elif self.scenario == "apikey_account":
+ self._reply(
+ request_id,
+ {"account": {"type": "apiKey"}, "requiresOpenaiAuth": False},
+ )
+ else:
+ self._reply(
+ request_id,
+ {"account": AUTHENTICATED_ACCOUNT, "requiresOpenaiAuth": False},
+ )
+ elif method == "thread/start":
+ self._reply(
+ request_id,
+ {"thread": {"id": "thread-123", "model": params.get("model")}},
+ )
+ elif self.scenario == "crash_after_thread" and method == "turn/start":
+ return False # hang up: the client must see EOF
+ elif method == "turn/start":
+ self._reply(
+ request_id,
+ {"threadId": "thread-123", "turn": {"id": "turn-1", "status": "inProgress"}},
+ )
+ self._notify(
+ "turn/started",
+ {"threadId": "thread-123", "turn": {"id": "turn-1", "status": "inProgress"}},
+ )
+ self._notify(
+ "item/completed",
+ {
+ "threadId": "thread-123",
+ "item": {"id": "item-0", "type": "agentMessage", "text": "hello world"},
+ },
+ )
+ status = "interrupted" if self.scenario == "interrupted" else "completed"
+ if self.scenario == "empty":
+ self._notify(
+ "item/completed",
+ {
+ "threadId": "thread-123",
+ "item": {"id": "item-0", "type": "agentMessage", "text": ""},
+ },
+ )
+ self._notify(
+ "turn/completed",
+ {"threadId": "thread-123", "turn": {"id": "turn-1", "status": status}},
+ )
+ else:
+ if request_id is not None:
+ self._error(request_id, {"code": -1, "message": f"unknown method {method}"})
+ return True
+
+
+def _make_cfg(**overrides):
+ defaults = {
+ "name": "gpt-5.3-codex",
+ "system_message": None,
+ "timeout": 10,
+ "weight": 1.0,
+ "retries": 0,
+ "retry_delay": 0,
+ "reasoning_effort": None,
+ "cwd": None,
+ "codex_bin": None,
+ "init_client": None,
+ "provider": None,
+ }
+ defaults.update(overrides)
+ cfg = MagicMock()
+ for key, value in defaults.items():
+ setattr(cfg, key, value)
+ return cfg
+
+
+def run_generate(scenario="ok", prompt="test prompt", cfg_overrides=None, kwargs=None):
+ """Drive one generate() call against a FakeAppServer via patched Popen."""
+ fake = FakeAppServer(scenario)
+ captured = {}
+
+ def _popen(cmd, **kwargs_):
+ captured["cmd"] = cmd
+ return fake
+
+ llm = CodexAppServerLLM(_make_cfg(**(cfg_overrides or {})))
+ outcome = {}
+
+ def _execute():
+ try:
+ outcome["result"] = asyncio.run(llm.generate(prompt, **(kwargs or {})))
+ outcome["error"] = None
+ except Exception as e: # noqa: BLE001 - tests assert on the type
+ outcome["error"] = e
+
+ with patch("openevolve.llm.codex_app_server.subprocess.Popen", side_effect=_popen):
+ _execute()
+ try:
+ fake.terminate()
+ except OSError:
+ pass
+ return outcome, fake, captured
+
+
+class TestCodexAppServerLLMBasics(unittest.TestCase):
+ def test_init_defaults(self):
+ llm = CodexAppServerLLM(_make_cfg())
+ self.assertEqual(llm.model, "gpt-5.3-codex")
+ self.assertEqual(llm.timeout, 10)
+ self.assertEqual(llm.weight, 1.0)
+ self.assertEqual(llm.retries, 0)
+
+ def test_init_falls_back_when_config_values_are_none(self):
+ llm = CodexAppServerLLM(_make_cfg(name=None, timeout=None, retries=None, retry_delay=None))
+ self.assertEqual(llm.model, "gpt-5.3-codex")
+ self.assertEqual(llm.timeout, 300)
+ self.assertEqual(llm.retries, 3)
+ self.assertEqual(llm.retry_delay, 5)
+
+ def test_factory_function(self):
+ llm = init_codex_app_server_client(_make_cfg())
+ self.assertIsInstance(llm, CodexAppServerLLM)
+ self.assertEqual(llm.model, "gpt-5.3-codex")
+
+ def test_invalid_reasoning_effort_raises(self):
+ with self.assertRaisesRegex(ValueError, "Invalid reasoning_effort: turbo"):
+ CodexAppServerLLM(_make_cfg(reasoning_effort="turbo"))
+
+
+class TestAuthBoundary(unittest.TestCase):
+ def test_authenticated_chatgpt_account_generates(self):
+ outcome, fake, captured = run_generate(scenario="ok")
+ self.assertIsNone(outcome["error"])
+ self.assertEqual(outcome["result"], "hello world")
+ methods = [m for m, _ in fake.requests]
+ self.assertEqual(
+ methods, ["initialize", "initialized", "account/read", "thread/start", "turn/start"]
+ )
+ self.assertEqual(captured["cmd"], ["codex", "app-server"])
+
+ def test_unauthenticated_account_is_rejected_before_any_turn(self):
+ outcome, fake, _ = run_generate(scenario="unauthenticated")
+ self.assertIsInstance(outcome["error"], CodexAuthError)
+ self.assertIn("codex login", str(outcome["error"]))
+ methods = [m for m, _ in fake.requests]
+ self.assertNotIn("thread/start", methods)
+ self.assertNotIn("turn/start", methods)
+ self.assertIn("account/read", methods)
+
+ def test_apikey_account_is_rejected_with_apikey_guidance(self):
+ outcome, fake, _ = run_generate(scenario="apikey_account")
+ self.assertIsInstance(outcome["error"], CodexAuthError)
+ self.assertIn("OpenAI provider", str(outcome["error"]))
+ self.assertNotIn("turn/start", [m for m, _ in fake.requests])
+
+ def test_auth_error_is_not_retried(self):
+ outcome, fake, _ = run_generate(
+ scenario="unauthenticated", cfg_overrides={"retries": 3, "retry_delay": 0}
+ )
+ self.assertIsInstance(outcome["error"], CodexAuthError)
+ # Exactly one account/read round-trip: auth failures are terminal.
+ self.assertEqual([m for m, _ in fake.requests].count("account/read"), 1)
+
+
+class TestGenerationProtocol(unittest.TestCase):
+ def test_initialize_handshake_carries_client_info(self):
+ _, fake, _ = run_generate()
+ params = next(p for m, p in fake.requests if m == "initialize")
+ self.assertEqual(params["clientInfo"]["name"], "openevolve")
+
+ def test_thread_is_non_interactive_and_read_only(self):
+ _, fake, _ = run_generate()
+ params = next(p for m, p in fake.requests if m == "thread/start")
+ self.assertEqual(params["model"], "gpt-5.3-codex")
+ self.assertEqual(params["approvalPolicy"], "never")
+ self.assertEqual(params["sandbox"]["mode"], "read-only")
+ self.assertFalse(params["sandbox"]["networkAccess"])
+
+ def test_turn_input_and_effort(self):
+ _, fake, _ = run_generate(
+ prompt="write code",
+ cfg_overrides={"reasoning_effort": "high", "system_message": "Be brief."},
+ )
+ params = next(p for m, p in fake.requests if m == "turn/start")
+ self.assertEqual(params["threadId"], "thread-123")
+ self.assertEqual(params["input"], [{"type": "text", "text": "Be brief.\n\nwrite code"}])
+ self.assertEqual(params["effort"], "high")
+
+ def test_no_effort_param_when_unset(self):
+ _, fake, _ = run_generate()
+ params = next(p for m, p in fake.requests if m == "turn/start")
+ self.assertNotIn("effort", params)
+
+ def test_custom_codex_bin_is_used(self):
+ _, _, captured = run_generate(cfg_overrides={"codex_bin": "/opt/codex/bin/codex"})
+ self.assertEqual(captured["cmd"], ["/opt/codex/bin/codex", "app-server"])
+
+ def test_interrupted_turn_raises(self):
+ outcome, _, _ = run_generate(scenario="interrupted")
+ self.assertIsInstance(outcome["error"], RuntimeError)
+ self.assertIn("did not complete", str(outcome["error"]))
+
+ def test_empty_agent_message_raises(self):
+ outcome, _, _ = run_generate(scenario="empty")
+ self.assertIsInstance(outcome["error"], RuntimeError)
+ self.assertIn("Empty response", str(outcome["error"]))
+
+ def test_server_crash_mid_turn_raises(self):
+ outcome, _, _ = run_generate(scenario="crash_after_thread")
+ self.assertIsInstance(outcome["error"], RuntimeError)
+ self.assertIn("exited unexpectedly", str(outcome["error"]))
+
+ def test_hung_server_hits_deadline(self):
+ outcome, _, _ = run_generate(scenario="hang", cfg_overrides={"timeout": 1})
+ self.assertIsInstance(outcome["error"], asyncio.TimeoutError)
+
+ def test_missing_cli_raises_without_retrying(self):
+ llm = CodexAppServerLLM(_make_cfg(retries=3, retry_delay=0))
+ with patch(
+ "openevolve.llm.codex_app_server.subprocess.Popen",
+ side_effect=FileNotFoundError("codex"),
+ ) as mock_popen:
+ with self.assertRaises(FileNotFoundError):
+ asyncio.run(llm.generate("prompt"))
+ self.assertEqual(mock_popen.call_count, 1)
+
+ def test_retry_on_transient_failure(self):
+ llm = CodexAppServerLLM(_make_cfg(retries=1, retry_delay=0))
+ good = FakeAppServer("ok")
+ self.addCleanup(good.terminate)
+ with patch(
+ "openevolve.llm.codex_app_server.subprocess.Popen",
+ side_effect=[subprocess.TimeoutExpired(cmd="codex", timeout=1), good],
+ ):
+ result = asyncio.run(llm.generate("prompt"))
+ self.assertEqual(result, "hello world")
+
+
+class TestCodexAppServerConfig(unittest.TestCase):
+ def test_config_round_trip_without_api_key(self):
+ from openevolve.config import Config
+
+ config = Config.from_dict(
+ {
+ "llm": {
+ "provider": "codex_app_server",
+ "models": [
+ {
+ "name": "gpt-5.3-codex",
+ "weight": 1.0,
+ "reasoning_effort": "medium",
+ }
+ ],
+ }
+ }
+ )
+ model = config.llm.models[0]
+ self.assertEqual(model.provider, "codex_app_server")
+ self.assertIsNone(model.api_key)
+ self.assertEqual(model.reasoning_effort, "medium")
+
+ def test_provider_propagates_from_llm_config(self):
+ from openevolve.config import Config
+
+ config = Config.from_dict(
+ {
+ "llm": {
+ "provider": "codex_app_server",
+ "models": [{"name": "gpt-5.3-codex", "weight": 1.0}],
+ }
+ }
+ )
+ self.assertEqual(config.llm.models[0].provider, "codex_app_server")
+ self.assertEqual(config.llm.evaluator_models[0].provider, "codex_app_server")
+
+
+class TestProviderRegistry(unittest.TestCase):
+ def test_codex_app_server_in_registry(self):
+ from openevolve.llm.ensemble import _PROVIDER_REGISTRY
+
+ self.assertIn("codex_app_server", _PROVIDER_REGISTRY)
+
+ def test_ensemble_creates_codex_app_server(self):
+ from openevolve.llm.ensemble import _create_model
+
+ cfg = _make_cfg()
+ cfg.provider = "codex_app_server"
+ self.assertIsInstance(_create_model(cfg), CodexAppServerLLM)
+
+ def test_apikey_default_path_still_opensai(self):
+ """No provider configured -> the OpenAI (API key) path is untouched."""
+ from openevolve.config import Config
+ from openevolve.llm.ensemble import LLMEnsemble
+
+ config = Config.from_dict({"llm": {"models": [{"name": "gpt-4"}]}})
+ ensemble = LLMEnsemble(config.llm.models)
+ self.assertEqual(type(ensemble.models[0]).__name__, "OpenAILLM")
+
+
+if __name__ == "__main__":
+ unittest.main()