"""OpenAI Codex CLI adapter (`codex exec`, non-interactive). OpenAI Platform env vars (``OPENAI_API_KEY``, ``OPENAI_ORG_ID``, ``OPENAI_PROJECT_ID``, ``OPENAI_BASE_URL``) are forwarded on invoke when set, so Codex runs work with usage-based API key auth as well as ``codex login`` sessions. """ from __future__ import annotations import os import subprocess from integrations.llm_cli.base import CLIInvocation, CLIProbe from integrations.llm_cli.binary_resolver import ( candidate_binary_names as _candidate_binary_names, ) from integrations.llm_cli.binary_resolver import ( default_cli_fallback_paths as _default_cli_fallback_paths, ) from integrations.llm_cli.binary_resolver import ( resolve_cli_binary, ) from integrations.llm_cli.constants import DEFAULT_EXEC_TIMEOUT_SEC from integrations.llm_cli.env_overrides import ( OPENAI_PLATFORM_ENV_KEYS, nonempty_env_values, ) from integrations.llm_cli.probe_utils import run_version_probe from integrations.llm_cli.semver_utils import parse_semver_three_part, semver_to_tuple _PROBE_TIMEOUT_SEC = 3.0 _READ_ONLY_SANDBOX = "read-only" _AUTH_STATUS_PROBE_ENV = "OPENSRE_CODEX_AUTH_STATUS_PROBE" _TRUTHY_ENV_VALUES = {"1", "true", "yes", "on"} def _classify_codex_auth(returncode: int, stdout: str, stderr: str) -> tuple[bool | None, str]: text = (stdout + "\n" + stderr).lower() # Negative phrases first: "logged in" is a substring of "not logged in". if "not logged in" in text or "no credentials" in text: return False, "Not logged in. Run: codex login" if returncode == 0 and "logged in" in text: return True, (stdout.strip() or stderr.strip() or "Logged in.").splitlines()[0] if "expired" in text or ("invalid" in text and "token" in text): return False, "Session expired. Re-authenticate: codex login" if "rate limit" in text or "quota" in text: return True, "Logged in but rate-limited; try again later." if "network" in text or "unreachable" in text or "dns" in text or "connection refused" in text: return None, "Network error while checking auth; will retry at invocation." if returncode != 0: tail = (stderr or stdout).strip()[:200] return ( None, f"Auth status unclear (exit {returncode}): {tail}" if tail else f"Auth status unclear (exit {returncode}).", ) return None, "Auth status unknown." def _codex_workspace_and_skip_git() -> tuple[str, bool]: try: proc = subprocess.run( ["git", "rev-parse", "--show-toplevel"], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=5.0, check=False, ) root = (proc.stdout or "").strip() if proc.returncode == 0 and root: return root, False except (OSError, subprocess.TimeoutExpired): # git missing, not a repo, or timed out — use cwd and let codex skip repo checks. pass return os.getcwd(), True def _fallback_codex_paths() -> list[str]: return _default_cli_fallback_paths("codex") def _has_openai_api_key() -> bool: return bool(os.environ.get("OPENAI_API_KEY", "").strip()) def _should_probe_codex_login_status() -> bool: return os.environ.get(_AUTH_STATUS_PROBE_ENV, "").strip().lower() in _TRUTHY_ENV_VALUES class CodexAdapter: """Non-interactive Codex CLI (`codex exec` with read-only sandbox).""" name = "codex" binary_env_key = "CODEX_BIN" install_hint = "npm i -g @openai/codex" auth_hint = "Run: codex login" min_version: str | None = None default_exec_timeout_sec = DEFAULT_EXEC_TIMEOUT_SEC def _resolve_binary(self) -> str | None: return resolve_cli_binary( explicit_env_key="CODEX_BIN", binary_names=_candidate_binary_names("codex"), fallback_paths=_fallback_codex_paths, ) def _probe_binary(self, binary_path: str) -> CLIProbe: version_output, version_error = run_version_probe( binary_path, timeout_sec=_PROBE_TIMEOUT_SEC, ) if version_error: return CLIProbe( installed=False, version=None, logged_in=None, bin_path=None, detail=version_error, ) version = parse_semver_three_part(version_output or "") upgrade_note = "" if ( self.min_version and version and semver_to_tuple(version) < semver_to_tuple(self.min_version) ): upgrade_note = ( f" Codex {version} is below tested minimum {self.min_version}; " f"upgrade: {self.install_hint}@latest" ) logged_in: bool | None auth_detail: str if not _should_probe_codex_login_status(): logged_in = None auth_detail = "Codex CLI installed." else: try: auth_proc = subprocess.run( [binary_path, "login", "status"], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=_PROBE_TIMEOUT_SEC, check=False, ) except (OSError, subprocess.TimeoutExpired): logged_in = None auth_detail = "Could not verify login status (timeout or OS error)." else: logged_in, auth_detail = _classify_codex_auth( auth_proc.returncode, auth_proc.stdout, auth_proc.stderr ) if logged_in is not True and _has_openai_api_key(): # Allow API-key auth when ChatGPT/session login is absent or unclear. logged_in = True auth_detail = "Authenticated via OPENAI_API_KEY fallback." detail = auth_detail + upgrade_note return CLIProbe( installed=True, version=version, logged_in=logged_in, bin_path=binary_path, detail=detail.strip(), ) def detect(self) -> CLIProbe: binary = self._resolve_binary() if not binary: return CLIProbe( installed=False, version=None, logged_in=None, bin_path=None, detail="Codex CLI not found on PATH or known install locations.", ) return self._probe_binary(binary) def build( self, *, prompt: str, model: str | None, workspace: str, reasoning_effort: str | None = None, ) -> CLIInvocation: binary = self._resolve_binary() if not binary: raise RuntimeError( "Codex CLI not found. Install with `npm i -g @openai/codex` or set CODEX_BIN." ) ws, skip_git = _codex_workspace_and_skip_git() if workspace: ws = workspace argv: list[str] = [ binary, "exec", "--ephemeral", "-s", _READ_ONLY_SANDBOX, "--color", "never", "-C", ws, ] if skip_git: argv.append("--skip-git-repo-check") resolved_model = (model or "").strip() if resolved_model: argv.extend(["-m", resolved_model]) if reasoning_effort: argv.extend(["-c", f'model_reasoning_effort="{reasoning_effort}"']) argv.append("-") oai = nonempty_env_values(OPENAI_PLATFORM_ENV_KEYS) return CLIInvocation( argv=tuple(argv), stdin=prompt, cwd=ws, env=oai or None, timeout_sec=self.default_exec_timeout_sec, ) def parse(self, *, stdout: str, stderr: str, returncode: int) -> str: result = (stdout or "").strip() if not result: raise RuntimeError( self.explain_failure(stdout=stdout, stderr=stderr, returncode=returncode) + " (empty output)" ) return result def explain_failure(self, *, stdout: str, stderr: str, returncode: int) -> str: from integrations.llm_cli.failure_explain import explain_cli_failure return explain_cli_failure( exit_label="codex exec", stdout=stdout, stderr=stderr, returncode=returncode, )