Single-agent baseline that gives one Codex CLI session the complete search loop inside the task's own image: Codex edits and evaluates, while Galapagos independently scores every observed checkpoint.
"""Codex CLI — a single-agent baseline that delegates search to ``codex exec``.
The execution topology deliberately matches :mod:`galapagos.scaffolds.claude_code`: in Docker mode
the CLI is installed into the task's own image, edits a bind-mounted ``solution`` file as a non-root
user, and shares the container with the framework's root-owned trusted evaluator. The framework
polls and independently scores changed snapshots; Codex's own prose is never trusted as a score.
Codex differs at the CLI boundary. ``codex exec --json`` emits JSONL events rather than Claude's
stream-json envelope, has no native ``--max-turns`` flag, and persists rollouts below
``$CODEX_HOME/sessions``. This scaffold therefore enforces a cap over completed actionable JSONL
items (commands, file changes, tool calls, plan updates, and agent messages), records token usage from
``turn.completed``, and preserves both the native rollout and raw event log.
Billing is structurally ChatGPT-managed, matching ``claude_code``: an official
``CODEX_ACCESS_TOKEN`` is consumed first (the Codex equivalent of a headless setup token), with a
cached ChatGPT browser login from ``~/.codex/auth.json`` as the fallback. API-key environment
variables are removed before the CLI starts. Browser-login credentials are copied into the isolated
run home, allowed to refresh there, written back atomically, and then deleted from run artifacts.
"""
from __future__ import annotations
import json
import logging
import os
import shlex
import shutil
import subprocess
import tempfile
import threading
import time
import uuid
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from ..._logging import STATUS
from ...components.evaluator import ContainerEvaluator, _RUNNER, _default_hf_cache_runtime
from ...environments import (
EVAL_DIR,
TaskContainer,
build_agent_image,
dood,
provision_task_dir,
resolve_image,
)
from ..claude_code.scaffold import (
WORKSPACE_DIR,
ClaudeCodeScaffold,
_DEFAULT_TURN_OVERHEAD_S,
_MIN_WALL_TIMEOUT_S,
_POLL_INTERVAL_S,
_kill_proc_group,
)
from ..registry import register_scaffold
log = logging.getLogger(__name__)
# Newer model catalogs extend the original range with ``max`` and ``ultra``; support remains
# model-dependent, so Codex itself performs the final model/effort compatibility check.
_CODEX_EFFORT_LEVELS = ("minimal", "low", "medium", "high", "xhigh", "max", "ultra")
_CODEX_REASONING_SUMMARIES = ("none", "auto", "concise", "detailed")
_CODEX_WEB_SEARCH_MODES = ("disabled", "cached", "indexed", "live")
_ACTIONABLE_ITEM_TYPES = {
"agent_message",
"command_execution",
"file_change",
"mcp_tool_call",
"web_search",
"plan_update",
}
_CODEX_AGENT_PID_FILE = f"{WORKSPACE_DIR}/.agent.pid"
_CODEX_SESSIONS_DIRNAME = "codex_sessions"
@dataclass(slots=True)
class _CodexAuthState:
"""Controller-owned authentication state; never serialize or log ``access_token``."""
source: str
access_token: str | None = None
credential_source: Path | None = None
initial_auth: str | None = None
finalized: bool = False
def _prepare_codex_home(path: Path) -> Path:
"""Create a subscription-only ``CODEX_HOME`` that can persist native rollout JSONL."""
path.mkdir(parents=True, exist_ok=True)
(path / "sessions").mkdir(exist_ok=True)
# The directory can survive a failed/retried run. Reset stale provider/API configuration from
# the earlier mixed-auth implementation before Codex reads this scaffold-owned home. Account
# apps and remote plugins stay off so an untrusted benchmark cannot inherit personal connector
# access from the ChatGPT login. Internal delegation is off because this is a single-agent
# baseline; native session history remains on for Galapagos trace retention.
(path / "config.toml").write_text(
'cli_auth_credentials_store = "file"\n'
"check_for_update_on_startup = false\n"
'history.persistence = "save-all"\n'
"\n"
"[features]\n"
"apps = false\n"
"memories = false\n"
"multi_agent = false\n"
"remote_plugin = false\n",
encoding="utf-8",
)
return path
def _validate_subscription_auth(payload: dict[str, Any], source: Path) -> None:
"""Reject API-key artifacts that are not a cached ChatGPT subscription login."""
mode = str(payload.get("auth_mode") or "").lower()
api_key = payload.get("OPENAI_API_KEY")
tokens = payload.get("tokens")
token_map = tokens if isinstance(tokens, dict) else {}
access_token = token_map.get("access_token") or token_map.get("accessToken")
refresh_token = token_map.get("refresh_token") or token_map.get("refreshToken")
if (
(isinstance(api_key, str) and api_key.strip())
or "api" in mode
or (mode and "chatgpt" not in mode)
or not isinstance(access_token, str)
or not access_token.strip()
or not isinstance(refresh_token, str)
or not refresh_token.strip()
):
raise RuntimeError(
"codex is subscription-only, but the saved credential at "
f"{source} is not a ChatGPT browser-login session. Configure "
'cli_auth_credentials_store = "file", run codex login without '
"--with-api-key, and retry. API-key billing is not supported."
)
def _setup_subscription_auth(env: dict[str, str], codex_home: Path) -> _CodexAuthState:
"""Select managed auth and scrub every usage-billed/provider path.
Priority deliberately mirrors the Claude scaffold:
1. ``CODEX_ACCESS_TOKEN`` — an official Business/Enterprise Codex automation token. The
controller consumes it with ``codex login --with-access-token`` inside the isolated
``CODEX_HOME`` before the agent starts; the token is not left in the agent environment.
2. ``~/.codex/auth.json`` — a normal ChatGPT browser-login cache containing both access and
refresh tokens. Codex may refresh the staged copy, so finalization writes changes back to the
source before deleting the run copy.
``GALAPAGOS_CODEX_CREDENTIALS_FILE`` can name a different file-backed browser login. API keys
and custom provider routing are never accepted by this scaffold.
"""
source_env = dict(env)
for key in (
"CODEX_API_KEY",
"OPENAI_API_KEY",
"CODEX_ACCESS_TOKEN",
"OPENAI_BASE_URL",
"GALAPAGOS_CODEX_CREDENTIALS_FILE",
):
env.pop(key, None)
access_token = (source_env.get("CODEX_ACCESS_TOKEN") or "").strip()
if access_token:
log.info(
"using CODEX_ACCESS_TOKEN via isolated `codex login --with-access-token`; "
"API-key billing is disabled"
)
return _CodexAuthState(source="access-token", access_token=access_token)
source_raw = (source_env.get("GALAPAGOS_CODEX_CREDENTIALS_FILE") or "").strip()
source = (
Path(source_raw).expanduser()
if source_raw
else Path.home() / ".codex" / "auth.json"
)
if not source.is_file():
raise RuntimeError(
"codex uses ChatGPT-managed auth, but neither CODEX_ACCESS_TOKEN nor a cached ChatGPT "
f"login was found at {source}. Business/Enterprise users may create a Codex access "
'token; otherwise configure cli_auth_credentials_store = "file" in '
"~/.codex/config.toml and run codex login. API-key billing is not supported."
)
source = source.resolve()
try:
raw_auth = source.read_text(encoding="utf-8")
payload = json.loads(raw_auth)
except (OSError, json.JSONDecodeError) as exc:
raise RuntimeError(
f"could not read Codex subscription login at {source}: {exc}"
) from exc
if not isinstance(payload, dict):
raise RuntimeError(
f"Codex subscription login at {source} must contain a JSON object"
)
_validate_subscription_auth(payload, source)
target = codex_home / "auth.json"
if source == target.resolve():
raise RuntimeError(
"Codex credential source must be outside the isolated run CODEX_HOME"
)
target.write_text(raw_auth, encoding="utf-8")
target.chmod(0o600)
log.info(
"using %s (copied into the run's isolated CODEX_HOME); API-key billing is disabled",
source,
)
return _CodexAuthState(
source="login-credentials",
credential_source=source,
initial_auth=raw_auth,
)
def _auth_command_error(stdout: Any, stderr: Any, token: str) -> str:
"""Return a bounded, token-redacted diagnostic from ``codex login``."""
chunks: list[str] = []
for value in (stderr, stdout):
if isinstance(value, bytes):
chunks.append(value.decode("utf-8", errors="replace"))
elif value:
chunks.append(str(value))
detail = "\n".join(chunks).replace(token, "<redacted>").strip()
return detail[-1200:] or "no diagnostic output"
def _bootstrap_access_token_host(
state: _CodexAuthState,
codex_home: Path,
cli: str,
env: dict[str, str],
) -> None:
"""Consume a Codex access token over stdin into the host run's isolated auth store."""
token = state.access_token
if token is None:
return
try:
try:
result = subprocess.run(
[cli, "login", "--with-access-token"],
input=token,
capture_output=True,
text=True,
env=env,
timeout=120,
check=False,
)
except Exception as exc: # noqa: BLE001 — normalize timeout/launch failures without the token
raise RuntimeError(
f"could not initialize Codex access-token auth: {exc}"
) from exc
if result.returncode != 0:
detail = _auth_command_error(result.stdout, result.stderr, token)
raise RuntimeError(
"codex login --with-access-token failed. Use a Codex access token created by a "
f"supported ChatGPT workspace, not a browser-session token.\n{detail}"
)
finally:
state.access_token = None
if not (codex_home / "auth.json").is_file():
raise RuntimeError(
"codex login --with-access-token succeeded but created no file auth cache"
)
def _bootstrap_access_token_container(
state: _CodexAuthState,
codex_home: Path,
container: TaskContainer,
) -> None:
"""Consume a Codex access token over ``docker exec -i`` stdin, never Docker ``-e``."""
token = state.access_token
if token is None:
return
try:
try:
result = container.exec(
["codex", "login", "--with-access-token"],
input=token.encode("utf-8"),
text=False,
timeout=120,
)
except Exception as exc: # noqa: BLE001 — normalize timeout/daemon failures without the token
raise RuntimeError(
f"could not initialize container Codex access-token auth: {exc}"
) from exc
if result.returncode != 0:
detail = _auth_command_error(result.stdout, result.stderr, token)
raise RuntimeError(
"container codex login --with-access-token failed. Use a Codex access token created "
f"by a supported ChatGPT workspace, not a browser-session token.\n{detail}"
)
finally:
state.access_token = None
if not (codex_home / "auth.json").is_file():
raise RuntimeError(
"container Codex login succeeded but created no file auth cache"
)
def _write_auth_atomically(path: Path, content: str) -> None:
"""Replace a credential file without exposing partial contents or permissive modes."""
fd, tmp_raw = tempfile.mkstemp(prefix=f".{path.name}.galapagos-", dir=path.parent)
tmp = Path(tmp_raw)
try:
os.fchmod(fd, 0o600)
with os.fdopen(fd, "w", encoding="utf-8") as handle:
fd = -1
handle.write(content)
handle.flush()
os.fsync(handle.fileno())
os.replace(tmp, path)
path.chmod(0o600)
finally:
if fd >= 0:
os.close(fd)
try:
tmp.unlink()
except FileNotFoundError:
pass
def _finalize_subscription_auth(
state: _CodexAuthState | None, codex_home: Path
) -> None:
"""Persist a Codex-managed refresh, then remove all run-artifact credentials."""
staged = codex_home / "auth.json"
if state is None:
try:
staged.unlink()
except FileNotFoundError:
pass
return
if state.finalized:
return
state.finalized = True
state.access_token = None
writeback_error: Exception | None = None
try:
if state.credential_source is not None and staged.is_file():
refreshed = staged.read_text(encoding="utf-8")
if refreshed != state.initial_auth:
try:
payload = json.loads(refreshed)
except json.JSONDecodeError as exc:
raise RuntimeError(
"Codex rewrote its auth cache as invalid JSON"
) from exc
if not isinstance(payload, dict):
raise RuntimeError(
"Codex rewrote its auth cache with a non-object payload"
)
_validate_subscription_auth(payload, staged)
# Another Codex process may have refreshed the source while this run was active. Never
# clobber that newer bundle; official account-auth automation requires serialized use.
current = state.credential_source.read_text(encoding="utf-8")
if current != state.initial_auth:
raise RuntimeError(
"the Codex credential source changed during this run; refusing to overwrite "
"a concurrent token refresh. Serialize runs that share one auth.json"
)
_write_auth_atomically(state.credential_source, refreshed)
log.info(
"persisted Codex's refreshed ChatGPT login to %s",
state.credential_source,
)
except Exception as exc: # noqa: BLE001 — cleanup must still remove the staged secret
writeback_error = exc
finally:
try:
staged.unlink()
except FileNotFoundError:
pass
if writeback_error is not None:
raise RuntimeError(
"Codex refreshed the staged ChatGPT login but Galapagos could not safely persist it; "
"the run copy was deleted to keep credentials out of artifacts. Run `codex login` again "
f"before the next job. Cause: {writeback_error}"
) from writeback_error
def _native_codex_session_files(
codex_home: Path, root: Path | None = None
) -> list[str]:
sessions = codex_home / "sessions"
if not sessions.is_dir():
return []
root = root or codex_home.parent
files: list[str] = []
for path in sessions.rglob("*.jsonl"):
if not path.is_file():
continue
try:
files.append(str(path.relative_to(root)))
except ValueError:
files.append(str(path))
return sorted(files)
def _describe_codex_item(item: dict[str, Any], limit: int = 96) -> str:
"""Render one machine-readable Codex item as a compact live progress line."""
kind = str(item.get("type") or "item")
target: Any = ""
if kind == "command_execution":
target = item.get("command")
elif kind == "file_change":
changes = item.get("changes")
if isinstance(changes, list):
paths = [
str(c.get("path"))
for c in changes
if isinstance(c, dict) and c.get("path")
]
target = ", ".join(paths)
target = target or item.get("path")
elif kind == "mcp_tool_call":
target = item.get("tool") or item.get("name")
elif kind == "web_search":
target = item.get("query")
elif kind == "plan_update":
target = item.get("text") or item.get("plan")
elif kind == "agent_message":
target = item.get("text")
text = " ".join(str(target or "").split())
if len(text) > limit:
text = text[:limit] + "…"
return f"{kind}: {text}" if text else kind
def _assert_codex_session_did_something(
stats: dict[str, Any], ckpt_count: int, returncode: int | None, log_path: Path
) -> None:
"""Reject auth/startup failures that would otherwise masquerade as a seed-only result."""
if ckpt_count or stats["steps"] or stats["input_tokens"] or stats["output_tokens"]:
return
failure = stats.get("error") or (
f"the CLI exited with code {returncode}" if returncode else None
)
if failure is None:
return
try:
tail = log_path.read_text(errors="replace")[-2000:]
except OSError:
tail = ""
raise RuntimeError(
"the Codex session produced nothing — no actionable event, token usage, or candidate "
f"({failure}). Is the cached ChatGPT subscription login still valid? Run `codex login` "
f"again if needed. Log tail:\n{tail}"
)
@register_scaffold("codex")
class CodexScaffold(ClaudeCodeScaffold):
"""One Codex CLI session driving the task's complete optimization loop."""
name = "codex"
status_unit = "step"
agent_metadata_kind_key = "codex"
agent_metadata_position_key = "codex_step"
def _model_display_name(self) -> str:
return self.config.proposer.codex_model or "Codex CLI default"
def _validate_config(
self,
) -> tuple[int, int, str | None, str | None, str | None, str]:
cfg = self.config.proposer
prompt_budget = int(self.general.max_iterations)
if prompt_budget <= 0:
raise ValueError("codex requires general.max_iterations to be positive")
cli_max_steps = (
int(cfg.codex_cli_max_steps)
if cfg.codex_cli_max_steps is not None
else prompt_budget * 2
)
if cli_max_steps < prompt_budget:
raise ValueError(
"proposer.codex_cli_max_steps must be >= general.max_iterations "
f"(got {cli_max_steps} < {prompt_budget})"
)
if (
cfg.codex_wall_timeout_seconds is not None
and int(cfg.codex_wall_timeout_seconds) <= 0
):
raise ValueError(
"proposer.codex_wall_timeout_seconds must be positive when set"
)
effort = cfg.reasoning_effort
if effort is not None and effort not in _CODEX_EFFORT_LEVELS:
raise ValueError(
f"codex: invalid proposer.reasoning_effort {effort!r}; use one of "
f"{', '.join(_CODEX_EFFORT_LEVELS)} (or null)"
)
reasoning_summary = cfg.reasoning_summary
if (
reasoning_summary is not None
and reasoning_summary not in _CODEX_REASONING_SUMMARIES
):
raise ValueError(
f"codex: invalid proposer.reasoning_summary {reasoning_summary!r}; use one of "
f"{', '.join(_CODEX_REASONING_SUMMARIES)} (or null)"
)
web_search = str(cfg.codex_web_search or "disabled")
if web_search not in _CODEX_WEB_SEARCH_MODES:
raise ValueError(
f"codex: invalid proposer.codex_web_search {web_search!r}; use one of "
f"{', '.join(_CODEX_WEB_SEARCH_MODES)}"
)
return (
prompt_budget,
cli_max_steps,
cfg.codex_model,
effort,
reasoning_summary,
web_search,
)
def _loop(self, task) -> None:
if getattr(self, "_container_mode", None) is None:
self._container_mode = self._wants_container(task)
started = time.monotonic()
session_event = self._emit_event(
event_type="agent",
action="start",
status="completed",
component_role="proposer",
component_name=type(self).__name__,
evolution_track="solution",
details={
"agent": "codex",
"model": self.config.proposer.codex_model,
"execution_mode": "container" if self._container_mode else "host",
"step_budget": int(self.general.max_iterations),
},
)
self._active_agent_event_id = session_event
try:
if self._container_mode:
self._loop_container(task)
else:
self._loop_host(task)
except KeyboardInterrupt as exc:
self._emit_event(
event_type="agent",
action="interrupt",
status="interrupted",
component_role="proposer",
component_name=type(self).__name__,
evolution_track="solution",
caused_by_event_ids=[session_event] if session_event else [],
duration_seconds=time.monotonic() - started,
error=exc,
details={"agent": "codex", **self._agent_run_summary()},
)
raise
except Exception as exc:
self._emit_event(
event_type="agent",
action="fail",
status="failed",
component_role="proposer",
component_name=type(self).__name__,
evolution_track="solution",
caused_by_event_ids=[session_event] if session_event else [],
duration_seconds=time.monotonic() - started,
error=exc,
details={"agent": "codex", **self._agent_run_summary()},
)
raise
else:
self._emit_event(
event_type="agent",
action="complete",
status="completed",
component_role="proposer",
component_name=type(self).__name__,
evolution_track="solution",
caused_by_event_ids=[session_event] if session_event else [],
duration_seconds=time.monotonic() - started,
details={
"agent": "codex",
"model": self.config.proposer.codex_model,
"steps": int(self.ctx.iteration),
**self._agent_run_summary(),
},
)
finally:
self._active_agent_event_id = None
def _bring_up(self, task) -> None:
cfg = self.config.proposer
self._validate_config()
run_dir = Path(self.ctx.run_dir).resolve() if self.ctx.run_dir else None
if run_dir:
self._out_dir, self._cleanup_out_dir = run_dir, False
else:
self._out_dir = Path(tempfile.mkdtemp(prefix="galapagos-codex-docker-"))
self._cleanup_out_dir = True
workspace = self._out_dir / "codex_workspace"
workspace.mkdir(parents=True, exist_ok=True)
self._codex_home = _prepare_codex_home(workspace / ".codex")
auth_env = dict(os.environ)
self._auth_state = _setup_subscription_auth(auth_env, self._codex_home)
self._auth_source = self._auth_state.source
spec = task.image_spec()
private_judge = bool(
getattr(task, "is_frontiercs2", False)
or getattr(self.evaluator, "private_judge", False)
)
if private_judge and cfg.codex_docker_network and cfg.codex_docker_network != spec.network:
raise ValueError(
"Frontier-CS 2.0 creates a private per-run Docker network for its judge; "
"proposer.codex_docker_network cannot override it"
)
dood_mounts, dood_env, group_add = (
dood.runtime() if spec.docker_access else ([], {}, [])
)
task_hf_runtime = getattr(task, "hf_cache_runtime", None)
hf_mounts, hf_env = (
task_hf_runtime() if callable(task_hf_runtime) else _default_hf_cache_runtime()
)
task_cache_runtime = getattr(task, "evaluator_cache_runtime", None)
cache_mounts, cache_env = (
task_cache_runtime() if callable(task_cache_runtime) else ([], {})
)
task_image = cfg.codex_docker_image or resolve_image(spec)
agent_image = build_agent_image(
task_image,
cli_version=cfg.codex_docker_cli_version,
agent="codex",
)
self._task_image, self._agent_image = task_image, agent_image
env = {
**dict(spec.env),
**dood_env,
# Task-specific HF settings are a security boundary: an arbitrary card env must not turn
# an allowlisted read-only repository mount back into an online/token-bearing cache.
**hf_env,
**cache_env,
"HOME": WORKSPACE_DIR,
"CODEX_HOME": f"{WORKSPACE_DIR}/.codex",
}
if spec.docker_access:
env["ALE_BENCH_CACHE"] = str(dood.parity_dir() / ".cache" / "ale-bench")
self._agent = TaskContainer.start(
agent_image,
name=f"galapagos-codex-{uuid.uuid4().hex[:12]}",
user=f"{os.getuid()}:{os.getgid()}",
env=env,
mounts=[
(str(workspace), WORKSPACE_DIR, "rw"),
*hf_mounts,
*cache_mounts,
*dood_mounts,
],
group_add=group_add,
gpus=spec.gpus,
network=cfg.codex_docker_network or spec.network,
memory=cfg.codex_docker_memory or spec.memory,
cpus=cfg.codex_docker_cpus or spec.cpus,
workdir=WORKSPACE_DIR,
)
try:
# Docker accepts a bare numeric `--user`, but libraries used by Codex and its login flow may
# resolve the current identity through NSS. Register it before invoking the CLI; the attached
# evaluator repeats this idempotently before scoring.
self._agent.ensure_posix_user(os.getuid(), os.getgid(), home=WORKSPACE_DIR)
_bootstrap_access_token_container(
self._auth_state,
self._codex_home,
self._agent,
)
if not isinstance(self.evaluator, ContainerEvaluator):
self.evaluator = task.container_evaluator(
attach_to=None if private_judge else self._agent
)
if private_judge:
attach = getattr(self.evaluator, "attach_agent", None)
if not callable(attach):
raise RuntimeError("Frontier-CS 2.0 evaluator cannot attach its public agent")
attach(self._agent)
self._keep_agent_for_final = True
else:
provision_task_dir(self._agent, str(task.root), _RUNNER)
except Exception:
self._agent.stop()
raise
def _loop_host(self, task) -> None:
self._run_codex_session(task, container=None)
def _loop_container(self, task) -> None:
if getattr(self, "_agent", None) is None:
self._bring_up(task)
self._apply_task_metadata(task)
self._run_codex_session(task, container=self._agent)
def run(self, task=None, **kwargs):
"""Run with refresh persistence and a credential-cleanup backstop."""
run_failed = True
try:
result = super().run(task=task, **kwargs)
run_failed = False
return result
finally:
finalize_error: Exception | None = None
codex_home = getattr(self, "_codex_home", None)
if codex_home is not None:
try:
_finalize_subscription_auth(
getattr(self, "_auth_state", None),
Path(codex_home),
)
except Exception as exc: # noqa: BLE001 — preserve original run error when both fail
if run_failed:
log.exception(
"failed to finalize Codex authentication after a run error"
)
else:
finalize_error = exc
agent = getattr(self, "_agent", None)
if agent is not None:
agent.stop()
ephemeral_workspace = getattr(self, "_ephemeral_codex_workspace", None)
if ephemeral_workspace is not None:
shutil.rmtree(ephemeral_workspace, ignore_errors=True)
self._ephemeral_codex_workspace = None
out_dir = getattr(self, "_out_dir", None)
if out_dir is not None and getattr(self, "_cleanup_out_dir", False):
shutil.rmtree(out_dir, ignore_errors=True)
if finalize_error is not None:
raise finalize_error
@staticmethod
def _command(
cli: str,
workspace: str,
model: str | None,
effort: str | None,
reasoning_summary: str | None,
web_search: str,
) -> list[str]:
cmd = [
cli,
"exec",
"--json",
"--skip-git-repo-check",
"--dangerously-bypass-approvals-and-sandbox",
"-C",
workspace,
]
if model:
cmd += ["--model", model.rsplit("/", 1)[-1]]
if effort:
cmd += ["-c", f'model_reasoning_effort="{effort}"']
if reasoning_summary:
cmd += ["-c", f'model_reasoning_summary="{reasoning_summary}"']
cmd += ["-c", f'web_search="{web_search}"', "-"]
return cmd
def _run_codex_session(self, task, *, container: TaskContainer | None) -> None:
cfg = self.config.proposer
(
prompt_budget,
cli_max_steps,
model,
effort,
reasoning_summary,
web_search,
) = self._validate_config()
seed = self.ctx.best
eval_timeout = int(getattr(self.evaluator, "timeout", 120))
suffix = getattr(self.evaluator, "suffix", ".py")
container_mode = container is not None
host_cli = None if container_mode else shutil.which("codex")
if not container_mode and not host_cli:
raise RuntimeError(
"the Codex CLI ('codex') was not found on PATH — install it with "
"`npm install -g @openai/codex`"
)
if not container_mode and not getattr(self.evaluator, "evaluator_path", None):
raise RuntimeError(
f"task {task.name!r} exposes no evaluator path for run_eval.sh"
)
if container_mode:
out_dir = self._out_dir
workspace = out_dir / "codex_workspace"
codex_home = self._codex_home
auth_source = self._auth_source
runtime_env = None
cli = "codex"
else:
run_dir = Path(self.ctx.run_dir).resolve() if self.ctx.run_dir else None
if run_dir:
out_dir, workspace = run_dir, run_dir / "codex_workspace"
workspace.mkdir(parents=True, exist_ok=True)
else:
workspace = Path(tempfile.mkdtemp(prefix="galapagos-codex-"))
out_dir = workspace
self._ephemeral_codex_workspace = workspace
codex_home = _prepare_codex_home(out_dir / _CODEX_SESSIONS_DIRNAME)
self._codex_home = codex_home
runtime_env = dict(os.environ)
runtime_env["CODEX_HOME"] = str(codex_home)
self._auth_state = _setup_subscription_auth(runtime_env, codex_home)
auth_source = self._auth_state.source
_bootstrap_access_token_host(
self._auth_state, codex_home, host_cli or "codex", runtime_env
)
cli = host_cli or "codex"
self._codex_home = codex_home
billing = "subscription"
solution_path = workspace / f"solution{suffix}"
solution_path.write_text(seed.content if seed else "", encoding="utf-8")
if container_mode:
self._write_eval_script_container(workspace, eval_timeout)
solution_display = f"{WORKSPACE_DIR}/solution{suffix}"
eval_command = f"bash {WORKSPACE_DIR}/run_eval.sh {solution_display}"
else:
ev_path = self.evaluator.evaluator_path
self._write_eval_script_host(workspace, str(ev_path), eval_timeout)
solution_display = str(solution_path)
eval_command = (
f"bash {shlex.quote(str(workspace / 'run_eval.sh'))} "
f"{shlex.quote(str(solution_path))}"
)
prompt = self._write_task_prompt(
workspace,
solution_display,
eval_command,
prompt_budget,
cli_max_steps,
eval_timeout,
container_mode,
)
prompt_path = workspace / ".prompt.txt"
prompt_path.write_text(prompt, encoding="utf-8")
command = self._command(
cli,
WORKSPACE_DIR if container_mode else str(workspace),
model,
effort,
reasoning_summary,
web_search,
)
if container_mode:
self._write_run_script(workspace, command)
wall_timeout = (
int(cfg.codex_wall_timeout_seconds)
if cfg.codex_wall_timeout_seconds is not None
else max(
cli_max_steps * (_DEFAULT_TURN_OVERHEAD_S + eval_timeout),
_MIN_WALL_TIMEOUT_S,
)
)
log_path, progress_path = out_dir / "codex.log", out_dir / "progress.log"
plock = threading.Lock()
def progress(line: str) -> None:
log.info("%s", line)
with plock, open(progress_path, "a", encoding="utf-8") as handle:
handle.write(f"[{time.strftime('%H:%M:%S')}] {line}\n")
progress(
f"Codex ({'Docker' if container_mode else 'host'}) run started — "
f"model={model or 'CLI default'}, effort={effort or 'CLI default'}, "
f"prompt_step_budget={prompt_budget}, cli_max_steps={cli_max_steps}, "
f"billing={billing} ({auth_source}), native traces → {codex_home / 'sessions'}, "
f"wall_timeout={wall_timeout}s"
)
stats: dict[str, Any] = {
"steps": 0,
"turns": 0,
"input_tokens": 0,
"cached_input_tokens": 0,
"output_tokens": 0,
"reasoning_output_tokens": 0,
"error": None,
"session_id": None,
"timed_out": False,
"budget_exhausted": False,
"returncode": None,
}
last_content = seed.content if seed else ""
ckpt_count = 0
run_start = time.monotonic()
proc = None
usage_recorded = False
kill_fn = None
try:
with open(log_path, "w", encoding="utf-8") as log_fh:
if container_mode:
proc = container.popen(
["bash", f"{WORKSPACE_DIR}/.run.sh"],
stdout=subprocess.PIPE,
stderr=log_fh,
)
kill_fn = self._kill_agent_session
else:
stdin_fh = open(prompt_path, encoding="utf-8")
try:
proc = subprocess.Popen(
command,
stdin=stdin_fh,
stdout=subprocess.PIPE,
stderr=log_fh,
cwd=str(workspace),
env=runtime_env,
start_new_session=True,
)
finally:
stdin_fh.close()
def kill_host_session() -> None:
_kill_proc_group(proc)
kill_fn = kill_host_session
reader = threading.Thread(
target=self._pump_events,
args=(proc, log_fh, stats, cli_max_steps, progress, kill_fn),
daemon=True,
)
reader.start()
while reader.is_alive():
remaining = wall_timeout - (time.monotonic() - run_start)
if remaining <= 0:
stats["timed_out"] = True
progress(f"Wall timeout ({wall_timeout}s) exceeded — stopping")
kill_fn()
reader.join(timeout=5)
else:
reader.join(timeout=min(_POLL_INTERVAL_S, remaining))
try:
current = solution_path.read_text(encoding="utf-8")
except OSError:
continue
if current == last_content or not current.strip():
continue
last_content = current
ckpt_count += 1
observed_steps = stats["steps"]
self._admit(
current,
seed,
iteration=max(observed_steps, ckpt_count),
kind="checkpoint",
turn=observed_steps,
)
self._maybe_checkpoint()
actual_steps = int(stats["steps"])
self.ctx.record_cost(0.0, stats["input_tokens"], stats["output_tokens"])
self._cached_tokens += int(stats["cached_input_tokens"])
usage_recorded = True
_assert_codex_session_did_something(
stats, ckpt_count, proc.returncode, log_path
)
try:
final = solution_path.read_text(encoding="utf-8")
except OSError:
final = ""
if final.strip() and final != last_content:
ckpt_count += 1
self._admit(
final,
seed,
iteration=max(actual_steps, ckpt_count),
kind="final",
turn=actual_steps,
)
self.ctx.iteration = max(self.ctx.iteration, actual_steps, 1)
best = self.ctx.best
native_files = _native_codex_session_files(codex_home, out_dir)
summary = {
"agent_session_format": "codex-native",
"model": model,
"effort": effort,
"reasoning_summary": reasoning_summary,
"web_search": web_search,
"billing": billing,
"auth_source": auth_source,
"session_id": stats["session_id"],
"native_session_files": native_files,
"event_log": str(log_path.relative_to(out_dir)),
"prompt_step_budget": prompt_budget,
"cli_max_steps": cli_max_steps,
"actual_steps": actual_steps,
"turns_completed": stats["turns"],
"checkpoints_scored": ckpt_count,
"timed_out": stats["timed_out"],
"budget_exhausted": stats["budget_exhausted"],
"return_code": stats["returncode"],
"tokens": {
"input": stats["input_tokens"],
"cached_input": stats["cached_input_tokens"],
"output": stats["output_tokens"],
"reasoning_output": stats["reasoning_output_tokens"],
},
"cost_usd": None,
"wall_seconds": round(time.monotonic() - run_start, 1),
"baseline_score": seed.scores.get("combined_score") if seed else None,
"final_score": best.scores.get("combined_score") if best else None,
}
if container_mode:
summary.update(
task_image=self._task_image, agent_image=self._agent_image
)
(out_dir / "run_summary.json").write_text(
json.dumps(summary, indent=2, default=str) + "\n", encoding="utf-8"
)
progress(
f"Run complete: steps={actual_steps}/{prompt_budget} prompt budget "
f"({cli_max_steps} CLI cap), tokens={stats['input_tokens']}↑/"
f"{stats['output_tokens']}↓, checkpoints={ckpt_count}, "
f"score={summary['final_score']}"
)
finally:
if proc is not None and proc.poll() is None:
(kill_fn or (lambda: None))()
proc.wait()
if not usage_recorded and (stats["input_tokens"] or stats["output_tokens"]):
self.ctx.record_cost(0.0, stats["input_tokens"], stats["output_tokens"])
self._cached_tokens += int(stats["cached_input_tokens"])
native_files = _native_codex_session_files(codex_home, out_dir)
if native_files:
progress(f"Native Codex trace(s) preserved: {', '.join(native_files)}")
else:
progress(
f"Warning: no native Codex session found under {codex_home / 'sessions'}"
)
if container_mode and not getattr(self, "_keep_agent_for_final", False):
container.stop()
# Ephemeral directories are removed by run() only after refreshed credentials have been
# written back. Removing them here would discard a Docker-side refresh before finalization.
def _write_task_prompt(
self,
workspace: Path,
solution_path: str,
eval_command: str,
prompt_budget: int,
cli_max_steps: int,
eval_timeout: int,
container_mode: bool,
) -> str:
environment = (
"You are running inside the task's isolated container."
if container_mode
else "You are running directly on the host; keep all scratch work in the workspace."
)
if container_mode and bool(getattr(self.evaluator, "private_judge", False)):
submission_path = str(getattr(self.evaluator, "submission_path", "/app/solution.py"))
submission_kind = str(getattr(self.evaluator, "submission_kind", "file"))
workspace_text = (
"the complete public project under `/app`"
if submission_kind == "directory"
else f"the public source tree under `/app` and final artifact `{submission_path}`"
)
content = (
"You are an AI coding agent solving a Frontier-CS 2.0 open-ended optimization "
f"problem. Aim for at most **{prompt_budget} improvement iterations**; the harness "
f"also enforces **{cli_max_steps} completed agent actions**. You are running inside "
"the public agent container; the evaluator and hidden benchmark data are in a "
"separate private judge.\n\n"
"## Public workspace and final artifact\n\n"
f"Work on {workspace_text}. Read `/app/readme`, `/app/AGENT.md`, and any public "
"README or test scripts before editing.\n\n"
"## How to evaluate\n\n"
"Submissions are asynchronous. Snapshot your current official artifact with:\n\n"
"```bash\n"
"bash /app/submit.sh\n"
"bash /app/submissions.sh\n"
"bash /app/wait_submission.sh <uuid>\n"
"```\n\n"
f"A judge run can take up to **{eval_timeout}s**. Submit early, continue improving "
"while work is queued, and use the returned score/feedback to choose what to keep. "
"The final verifier reruns both the artifact left at the official path and the best "
"completed iterative submission.\n\n"
"## Task description\n\n"
f"{self.ctx.task_context}\n\n"
"## Instructions\n\n"
f"- Always leave your best final artifact at `{submission_path}`.\n"
"- For patch tasks, edit the clean checkout in `/app`, then run the provided "
"`make_submission.sh` before `submit.sh`.\n"
"- Try several distinct approaches; do not wait idly for a queued evaluation.\n"
f"- Keep scratch files in `{WORKSPACE_DIR}` and public solution work in `/app`.\n"
"- Do not search the network for benchmark answers.\n"
"- You cannot access the evaluator, final-role token, or hidden data.\n"
)
(workspace / "TASK.md").write_text(content, encoding="utf-8")
return content
content = (
"You are an AI coding agent iteratively improving a program to maximize its evaluation "
f"score. Aim for at most **{prompt_budget} improvement iterations**; the harness also "
f"enforces a safety cap of **{cli_max_steps} completed agent actions**. {environment}\n\n"
"## Current solution\n\n"
f"`{solution_path}` -- read it, understand it, and modify it freely.\n\n"
"## How to evaluate\n\n"
"```bash\n"
f"{eval_command}\n"
"```\n\n"
"Output is JSON. Maximize `combined_score` (higher is better). The evaluator has a "
f"**{eval_timeout}s timeout**.\n\n"
"## Task description\n\n"
f"{self.ctx.task_context}\n\n"
"## Instructions\n\n"
"- Run the evaluator once to confirm the baseline, then start improving.\n"
"- After each change, evaluate it and decide whether to keep or revert.\n"
f"- Always leave `{solution_path}` containing your best solution.\n"
"- Try several distinct approaches within the budget; do not stop after the first idea.\n"
f"- Create scratch files only inside `{workspace if not container_mode else WORKSPACE_DIR}`.\n"
"- Do not use the network to look up benchmark answers.\n"
)
if container_mode:
content += (
f"- `{EVAL_DIR}` contains the task scorer and data. Do not modify it; the framework "
"recomputes your score independently.\n"
)
(workspace / "TASK.md").write_text(content, encoding="utf-8")
return content
@staticmethod
def _write_run_script(workspace: Path, command: list[str]) -> None:
lines = [
"#!/bin/bash",
"set -euo pipefail",
f"echo $$ > {_CODEX_AGENT_PID_FILE}",
"exec env -u CODEX_API_KEY -u OPENAI_API_KEY -u CODEX_ACCESS_TOKEN "
"-u OPENAI_BASE_URL -u GALAPAGOS_CODEX_CREDENTIALS_FILE "
+ " ".join(shlex.quote(part) for part in command)
+ f" < {WORKSPACE_DIR}/.prompt.txt",
]
path = workspace / ".run.sh"
path.write_text("\n".join(lines) + "\n", encoding="utf-8")
path.chmod(0o755)
def _kill_agent_session(self) -> None:
for sig in ("TERM", "KILL"):
if self._agent.kill_process(_CODEX_AGENT_PID_FILE, sig=sig):
return
if self._agent.exec(["pkill", f"-{sig}", "-f", "codex"]).returncode == 0:
return
log.warning(
"could not signal the Codex session via %s or pkill; not killing the shared evaluator "
"container",
_CODEX_AGENT_PID_FILE,
)
@staticmethod
def _pump_events(
proc, log_fh, stats: dict[str, Any], cli_max_steps: int, progress, kill_fn
) -> None:
"""Tee ``codex exec --json`` and maintain live step/token/session accounting."""
seen_items: set[str] = set()
start = time.monotonic()
def record(raw) -> dict[str, Any] | None:
decoded = (
raw if isinstance(raw, str) else raw.decode("utf-8", errors="replace")
)
log_fh.write(decoded)
log_fh.flush()
try:
event = json.loads(decoded)
except (json.JSONDecodeError, ValueError):
return None
return event if isinstance(event, dict) else None
def consume(event: dict[str, Any], *, count_items: bool = True) -> None:
event_type = event.get("type")
if event_type == "thread.started" and event.get("thread_id"):
stats["session_id"] = event["thread_id"]
elif event_type == "turn.completed":
stats["turns"] += 1
usage = (
event.get("usage") if isinstance(event.get("usage"), dict) else {}
)
stats["input_tokens"] += int(usage.get("input_tokens", 0) or 0)
stats["cached_input_tokens"] += int(
usage.get("cached_input_tokens", 0) or 0
)
stats["output_tokens"] += int(usage.get("output_tokens", 0) or 0)
stats["reasoning_output_tokens"] += int(
usage.get("reasoning_output_tokens", 0) or 0
)
STATUS.update(
prompt_tokens=stats["input_tokens"],
completion_tokens=stats["output_tokens"],
)
elif event_type in {"turn.failed", "error"}:
error = event.get("error")
if isinstance(error, dict):
error = error.get("message") or json.dumps(error, default=str)
stats["error"] = str(error or event.get("message") or event_type)
elif event_type == "item.completed" and count_items:
item = event.get("item") if isinstance(event.get("item"), dict) else {}
item_id = str(item.get("id") or "")
if item_id and item_id in seen_items:
return
if item_id:
seen_items.add(item_id)
if item.get("type") not in _ACTIONABLE_ITEM_TYPES:
return
stats["steps"] += 1
STATUS.update(iteration=stats["steps"])
progress(
f"step {stats['steps']}/{cli_max_steps} {_describe_codex_item(item)} "
f"[{time.monotonic() - start:.0f}s]"
)
def starts_another_action(event: dict[str, Any]) -> bool:
if event.get("type") not in {"item.started", "item.completed"}:
return False
item = event.get("item") if isinstance(event.get("item"), dict) else {}
item_id = str(item.get("id") or "")
if item_id and item_id in seen_items:
return False
# Let Codex emit its terminal answer after the last allowed tool action. That normally
# carries the following turn.completed usage event; killing at the exact cap would lose
# those token totals even though no additional command/file/tool action is being allowed.
return item.get("type") in _ACTIONABLE_ITEM_TYPES - {"agent_message"}
try:
cap_reached = False
for raw in proc.stdout:
event = record(raw)
if event is None:
continue
if cap_reached:
if starts_another_action(event):
item = (
event.get("item")
if isinstance(event.get("item"), dict)
else {}
)
progress(
f"Action budget {cli_max_steps} reached; stopping before additional "
f"{item.get('type') or 'action'}"
)
kill_fn()
break
consume(event, count_items=False)
continue
consume(event)
if stats["steps"] >= cli_max_steps:
stats["budget_exhausted"] = True
cap_reached = True
progress(
f"Action budget {cli_max_steps} reached — waiting for the terminal response"
)
finally:
proc.wait()
stats["returncode"] = proc.returncode
try:
for raw in proc.stdout:
event = record(raw)
if event is not None:
consume(event, count_items=False)
except (OSError, ValueError):
pass
progress(
f"CLI exited (code {proc.returncode}), completed steps: {stats['steps']}, "
f"turns: {stats['turns']}"
)