"""Official Codex CLI harness. This module deliberately does not implement an HTTP model adapter. It starts the official ``codex exec`` binary installed in the container. The CLI reads its provider and MCP configuration from ``CODEX_HOME/config.toml``. """ from __future__ import annotations import json import os import shutil import subprocess import uuid from dataclasses import dataclass from typing import Any @dataclass class CodexResult: final_response: str finish_reason: str = "completed" def _text_from_value(value: Any) -> str: if isinstance(value, str): return value.strip() if isinstance(value, list): parts = [_text_from_value(item) for item in value] return "".join(part for part in parts if part) if isinstance(value, dict): for key in ("output_text", "text", "value", "content"): text = _text_from_value(value.get(key)) if text: return text return "" def _event_text(event: dict[str, Any]) -> str: """Extract only assistant/final text from a Codex JSONL event.""" event_type = str(event.get("type") or "") if event_type in {"item.completed", "item.updated"}: item = event.get("item") if isinstance(item, dict): item_type = str(item.get("type") or "") if item_type in {"agent_message", "assistant_message", "message", "output_text"}: return _text_from_value(item) if event_type in {"response.completed", "response.done", "message", "assistant_message"}: return _text_from_value(event.get("response") or event) if event_type in {"final", "final_response"}: return _text_from_value(event) return "" class CodexNativeHarness: """Run the official Codex CLI with the configured MCP server.""" def __init__(self, model=None, api_key=None, base_url=None, session_root=None, system_prompt=None, **kwargs): self.model = model or os.environ.get("CODEX_MODEL", "gpt-5.6-luna") self.api_key = ( api_key or os.environ.get("OPENAI_API_KEY") or os.environ.get("ZAI_API_KEY") or os.environ.get("GLM_API_KEY") or os.environ.get("ZHIPU_API_KEY") or "" ).strip() self.base_url = (base_url or os.environ.get("OPENAI_BASE_URL") or "").rstrip("/") # /tmp is deliberately avoided: the official CLI refuses to create # helper aliases there. The home directory is writable in HF Spaces # and keeps the CLI state separate from the application source tree. self.codex_home = os.environ.get("CODEX_HOME", "/home/user/.codex-home") self.timeout = int(os.environ.get("CODEX_EXEC_TIMEOUT_SECONDS", "900")) self.system_prompt = system_prompt or "" self.binary = ( os.environ.get("CODEX_BIN") or shutil.which("codex") or "/usr/local/bin/codex" ) if not os.path.isfile(self.binary) and shutil.which(self.binary) is None: raise RuntimeError("官方 Codex CLI 不存在: " + self.binary) def run(self, prompt, session_id=None): if not self.api_key: raise RuntimeError( "没有配置 ZAI API key (ZAI_API_KEY/GLM_API_KEY/ZHIPU_API_KEY)" ) full_prompt = (self.system_prompt + "\n\n" + str(prompt)).strip() env = os.environ.copy() env["CODEX_HOME"] = self.codex_home env["OPENAI_API_KEY"] = self.api_key if self.base_url: env["OPENAI_BASE_URL"] = self.base_url env.setdefault("CODEX_THREAD_ID", str(session_id or uuid.uuid4())) command = [ self.binary, "exec", ] # Use the HTTPS Responses transport for deterministic startup. disable_ws = os.environ.get("CODEX_DISABLE_WEBSOCKETS", "1") if disable_ws.strip().lower() not in {"0", "false", "no", "off"}: command.extend( ["--disable", "responses_websockets", "--disable", "responses_websockets_v2"] ) # Never append the CLI's sandbox/approval bypass. This public # application intentionally exposes only the narrow Marine MCP # allow-list; a deployment environment must not be able to turn a # user chat into arbitrary server control by setting one variable. command.extend( [ "--skip-git-repo-check", "--json", "--model", self.model, full_prompt, ] ) try: completed = subprocess.run( command, cwd=(os.environ.get("APP_ROOT") or "/home/user/app") if os.path.isdir(os.environ.get("APP_ROOT") or "/home/user/app") else None, env=env, # `codex exec` can wait for a second prompt when its stdin is # inherited from the parent process. The harness already # supplies the prompt as the final positional argument, so # close stdin explicitly for non-interactive execution. stdin=subprocess.DEVNULL, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=self.timeout, check=False, ) except subprocess.TimeoutExpired as exc: raise RuntimeError(f"官方 Codex CLI 超时（>{self.timeout}s）") from exc except OSError as exc: raise RuntimeError(f"无法启动官方 Codex CLI: {exc}") from exc candidates: list[str] = [] for line in (completed.stdout or "").splitlines(): raw = line.strip() if not raw: continue try: event = json.loads(raw) except json.JSONDecodeError: continue if isinstance(event, dict): text = _event_text(event) if text: candidates.append(text) if completed.returncode != 0: stderr_detail = (completed.stderr or "").strip()[-1200:] stdout_detail = (completed.stdout or "").strip()[-1600:] detail_parts = [] if stderr_detail: detail_parts.append("stderr: " + stderr_detail) if stdout_detail: detail_parts.append("stdout: " + stdout_detail) detail = "\n".join(detail_parts) raise RuntimeError( f"官方 Codex CLI 退出码 {completed.returncode}" + (f": {detail}" if detail else "") ) answer = candidates[-1].strip() if candidates else "" if not answer: raise RuntimeError("官方 Codex CLI 没有返回最终文本") return CodexResult(final_response=answer)