"""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, "").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 \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']}" )