Compare commits

...

1 Commits

Author SHA1 Message Date
kshitijk4poor
965d2fec98 feat(provider): add codex-cli external-process provider
Add an external-process inference provider that shells out to the
Codex CLI (codex exec --json) for inference.  This lets users
delegate Hermes requests to their local Codex CLI installation,
leveraging Codex's agent loop while keeping Hermes as the driver.

Key design:
- Text-in/text-out MVP — Hermes tools are disabled (Codex handles its
  own tool calling internally).
- Streaming is disabled (subprocess stdio returns a single
  SimpleNamespace, not an iterable generator).
- Follows the copilot-acp external-process pattern for routing,
  streaming exclusion, and credential resolution.

Files:
- agent/codex_cli_client.py  — Client facade, parses JSONL events
- hermes_cli/auth.py  — ProviderConfig, status helper, cred resolver
- hermes_cli/runtime_provider.py  — Runtime resolution
- run_agent.py  — Client routing, tool disable, streaming exclusion
- hermes_cli/models.py  — Provider entry, aliases, model list
- hermes_cli/main.py  — --provider choices

Env var support: HERMES_CODEX_CLI_COMMAND, CODEX_CLI_PATH,
HERMES_CODEX_CLI_ARGS.
2026-05-09 21:02:32 +05:30
7 changed files with 615 additions and 44 deletions

334
agent/codex_cli_client.py Normal file
View File

@@ -0,0 +1,334 @@
"""OpenAI-compatible shim that forwards Hermes requests to ``codex exec --json``.
This adapter lets Hermes treat the OpenAI Codex CLI as a chat-style backend.
Each request spawns ``codex exec --json --ephemeral --dangerously-bypass-approvals-and-sandbox``,
parses the JSONL event stream, extracts the agent message text and token usage,
and converts the result into the minimal shape Hermes expects from an OpenAI client.
"""
from __future__ import annotations
import json
import logging
import os
import subprocess
import threading
import time
from pathlib import Path
from types import SimpleNamespace
from typing import Any
logger = logging.getLogger(__name__)
_CODEX_CLI_BASE_URL = "codex-cli://local"
_DEFAULT_TIMEOUT_SECONDS = 900.0
def _resolve_command() -> str:
return (
os.getenv("HERMES_CODEX_CLI_COMMAND", "").strip()
or os.getenv("CODEX_CLI_PATH", "").strip()
or "codex"
)
def _resolve_args() -> list[str]:
raw = os.getenv("HERMES_CODEX_CLI_ARGS", "").strip()
if not raw:
return [
"exec",
"--json",
"--ephemeral",
"--dangerously-bypass-approvals-and-sandbox",
"--skip-git-repo-check",
]
import shlex
return shlex.split(raw)
def _build_subprocess_env() -> dict[str, str]:
env = os.environ.copy()
# Preserve HOME so codex can find ~/.codex/auth.json
home = os.environ.get("HOME", "")
if not home:
home = os.path.expanduser("~")
if home and home != "~":
env["HOME"] = home
return env
def _parse_turn_completed_usage(event: dict[str, Any]) -> SimpleNamespace:
usage = event.get("usage") or {}
input_tokens = int(usage.get("input_tokens") or 0)
cached_tokens = int(usage.get("cached_input_tokens") or 0)
output_tokens = int(usage.get("output_tokens") or 0)
reasoning_tokens = int(usage.get("reasoning_output_tokens") or 0)
return SimpleNamespace(
prompt_tokens=input_tokens,
completion_tokens=output_tokens + reasoning_tokens,
total_tokens=input_tokens + output_tokens + reasoning_tokens,
prompt_tokens_details=SimpleNamespace(cached_tokens=cached_tokens),
)
class _CodexCLIChatCompletions:
def __init__(self, client: "CodexCLIClient"):
self._client = client
def create(self, **kwargs: Any) -> Any:
return self._client._create_chat_completion(**kwargs)
class _CodexCLIChatNamespace:
def __init__(self, client: "CodexCLIClient"):
self.completions = _CodexCLIChatCompletions(client)
class CodexCLIClient:
"""Minimal OpenAI-client-compatible facade for Codex CLI."""
def __init__(
self,
*,
api_key: str | None = None,
base_url: str | None = None,
default_headers: dict[str, str] | None = None,
command: str | None = None,
args: list[str] | None = None,
**_: Any,
):
self.api_key = api_key or "codex-cli"
self.base_url = base_url or _CODEX_CLI_BASE_URL
self._default_headers = dict(default_headers or {})
self._command = command or _resolve_command()
self._args = list(args or _resolve_args())
self.chat = _CodexCLIChatNamespace(self)
self.is_closed = False
self._active_process: subprocess.Popen[str] | None = None
self._active_process_lock = threading.Lock()
def close(self) -> None:
proc: subprocess.Popen[str] | None
with self._active_process_lock:
proc = self._active_process
self._active_process = None
self.is_closed = True
if proc is None:
return
try:
proc.terminate()
proc.wait(timeout=2)
except Exception:
try:
proc.kill()
except Exception:
pass
def _build_prompt(self, messages: list[dict[str, Any]], model: str | None = None) -> str:
sections: list[str] = [
"You are being used as the active Codex CLI agent backend for Hermes.",
"Respond to the user's request directly. Do NOT call tools — Hermes handles tools.",
]
if model:
sections.append(f"Hermes requested model hint: {model}")
transcript: list[str] = []
for message in messages:
if not isinstance(message, dict):
continue
role = str(message.get("role") or "unknown").strip().lower()
content = message.get("content")
if content is None:
continue
if isinstance(content, list):
parts = []
for item in content:
if isinstance(item, str):
parts.append(item)
elif isinstance(item, dict) and "text" in item:
parts.append(str(item["text"]))
content = "\n".join(parts).strip()
if not content:
continue
label = {
"system": "System",
"user": "User",
"assistant": "Assistant",
"tool": "Tool",
}.get(role, role.title())
transcript.append(f"{label}:\n{content}")
if transcript:
sections.append("Conversation transcript:\n\n" + "\n\n".join(transcript))
sections.append("Continue the conversation from the latest user request.")
return "\n\n".join(s.strip() for s in sections if s and s.strip())
def _create_chat_completion(
self,
*,
model: str | None = None,
messages: list[dict[str, Any]] | None = None,
timeout: float | None = None,
tools: list[dict[str, Any]] | None = None,
tool_choice: Any = None,
**_: Any,
) -> Any:
prompt_text = self._build_prompt(messages or [], model=model)
# Normalise timeout: run_agent.py may pass an httpx.Timeout object
if timeout is None:
effective_timeout = _DEFAULT_TIMEOUT_SECONDS
elif isinstance(timeout, (int, float)):
effective_timeout = float(timeout)
else:
candidates = [
getattr(timeout, attr, None)
for attr in ("read", "write", "connect", "pool", "timeout")
]
numeric = [float(v) for v in candidates if isinstance(v, (int, float))]
effective_timeout = max(numeric) if numeric else _DEFAULT_TIMEOUT_SECONDS
response_text, usage = self._run_prompt(prompt_text, timeout_seconds=effective_timeout)
assistant_message = SimpleNamespace(
content=response_text,
tool_calls=[],
reasoning=None,
reasoning_content=None,
reasoning_details=None,
)
choice = SimpleNamespace(message=assistant_message, finish_reason="stop")
return SimpleNamespace(
choices=[choice],
usage=usage,
model=model or "codex-cli",
)
def _run_prompt(self, prompt_text: str, *, timeout_seconds: float) -> tuple[str, SimpleNamespace]:
cmd = [self._command] + self._args
# The prompt is a positional arg — pass it via stdin with pipe
try:
proc = subprocess.Popen(
cmd,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1,
env=_build_subprocess_env(),
)
except FileNotFoundError as exc:
raise RuntimeError(
f"Could not start Codex CLI command '{self._command}'. "
"Install Codex CLI (npm install -g @openai/codex) or set "
f"HERMES_CODEX_CLI_COMMAND / CODEX_CLI_PATH."
) from exc
if proc.stdin is None or proc.stdout is None:
proc.kill()
raise RuntimeError("Codex CLI process did not expose stdin/stdout pipes.")
self.is_closed = False
with self._active_process_lock:
self._active_process = proc
response_parts: list[str] = []
usage = SimpleNamespace(
prompt_tokens=0,
completion_tokens=0,
total_tokens=0,
prompt_tokens_details=SimpleNamespace(cached_tokens=0),
)
stderr_lines: list[str] = []
try:
# Write prompt to stdin and close it to signal end of input
proc.stdin.write(prompt_text)
proc.stdin.close()
deadline = time.monotonic() + timeout_seconds
stdout_thread = threading.Thread(target=lambda: None, daemon=True)
# Collect stdout lines
stdout_lines: list[str] = []
def _read_stdout():
if proc.stdout is None:
return
for line in proc.stdout:
stdout_lines.append(line.rstrip("\n"))
stdout_thread = threading.Thread(target=_read_stdout, daemon=True)
stdout_thread.start()
# We'll also collect stderr
stderr_output: list[str] = []
def _read_stderr():
if proc.stderr is None:
return
for line in proc.stderr:
stderr_output.append(line.rstrip("\n"))
stderr_thread = threading.Thread(target=_read_stderr, daemon=True)
stderr_thread.start()
# Wait for process to complete or timeout
remaining = deadline - time.monotonic()
while remaining > 0:
if proc.poll() is not None:
break
time.sleep(0.1)
remaining = deadline - time.monotonic()
if proc.poll() is None:
proc.kill()
raise TimeoutError("Timed out waiting for Codex CLI response.")
# Wait for threads to finish reading
stdout_thread.join(timeout=5)
stderr_thread.join(timeout=5)
# Parse JSONL output
agent_text = ""
for line in stdout_lines:
try:
event = json.loads(line)
except Exception:
# Non-JSON line (banner, status) — skip
continue
event_type = event.get("type", "")
if event_type == "item.completed":
item = event.get("item") or {}
if item.get("type") == "agent_message":
text = item.get("text") or ""
if text:
agent_text += text
elif event_type == "turn.completed":
usage = _parse_turn_completed_usage(event)
if agent_text:
response_parts.append(agent_text)
# Stderr with useful diagnostics
for line in stderr_output:
if line.strip():
stderr_lines.append(line)
if stderr_lines and not agent_text:
raise RuntimeError(
"Codex CLI produced no agent message. "
f"stderr: {'; '.join(stderr_lines[-5:])}"
)
return "\n".join(response_parts).strip(), usage
finally:
if proc.poll() is None:
try:
proc.kill()
except Exception:
pass
with self._active_process_lock:
if self._active_process is proc:
self._active_process = None

View File

@@ -197,6 +197,13 @@ PROVIDER_REGISTRY: Dict[str, ProviderConfig] = {
inference_base_url=DEFAULT_COPILOT_ACP_BASE_URL,
base_url_env_var="COPILOT_ACP_BASE_URL",
),
"codex-cli": ProviderConfig(
id="codex-cli",
name="OpenAI Codex CLI",
auth_type="external_process",
inference_base_url="codex-cli://local",
base_url_env_var="CODEX_CLI_BASE_URL",
),
"gemini": ProviderConfig(
id="gemini",
name="Google AI Studio",
@@ -1377,6 +1384,7 @@ def resolve_provider(
"github": "copilot", "github-copilot": "copilot",
"github-models": "copilot", "github-model": "copilot",
"github-copilot-acp": "copilot-acp", "copilot-acp-agent": "copilot-acp",
"codexcli": "codex-cli", "openai-codex-cli": "codex-cli",
"aigateway": "ai-gateway", "vercel": "ai-gateway", "vercel-ai-gateway": "ai-gateway",
"opencode": "opencode-zen", "zen": "opencode-zen",
"qwen-portal": "qwen-oauth", "qwen-cli": "qwen-oauth", "qwen-oauth": "qwen-oauth", "google-gemini-cli": "google-gemini-cli", "gemini-cli": "google-gemini-cli", "gemini-oauth": "google-gemini-cli",
@@ -4009,28 +4017,60 @@ def get_external_process_provider_status(provider_id: str) -> Dict[str, Any]:
if not pconfig or pconfig.auth_type != "external_process":
return {"configured": False}
command = (
os.getenv("HERMES_COPILOT_ACP_COMMAND", "").strip()
or os.getenv("COPILOT_CLI_PATH", "").strip()
or "copilot"
)
raw_args = os.getenv("HERMES_COPILOT_ACP_ARGS", "").strip()
args = shlex.split(raw_args) if raw_args else ["--acp", "--stdio"]
base_url = os.getenv(pconfig.base_url_env_var, "").strip() if pconfig.base_url_env_var else ""
if not base_url:
base_url = pconfig.inference_base_url
if provider_id == "copilot-acp":
command = (
os.getenv("HERMES_COPILOT_ACP_COMMAND", "").strip()
or os.getenv("COPILOT_CLI_PATH", "").strip()
or "copilot"
)
raw_args = os.getenv("HERMES_COPILOT_ACP_ARGS", "").strip()
args = shlex.split(raw_args) if raw_args else ["--acp", "--stdio"]
base_url = os.getenv(pconfig.base_url_env_var, "").strip() if pconfig.base_url_env_var else ""
if not base_url:
base_url = pconfig.inference_base_url
resolved_command = shutil.which(command) if command else None
return {
"configured": bool(resolved_command or base_url.startswith("acp+tcp://")),
"provider": provider_id,
"name": pconfig.name,
"command": command,
"args": args,
"resolved_command": resolved_command,
"base_url": base_url,
"logged_in": bool(resolved_command or base_url.startswith("acp+tcp://")),
}
resolved_command = shutil.which(command) if command else None
return {
"configured": bool(resolved_command or base_url.startswith("acp+tcp://")),
"provider": provider_id,
"name": pconfig.name,
"command": command,
"args": args,
"resolved_command": resolved_command,
"base_url": base_url,
"logged_in": bool(resolved_command or base_url.startswith("acp+tcp://")),
}
if provider_id == "codex-cli":
command = (
os.getenv("HERMES_CODEX_CLI_COMMAND", "").strip()
or os.getenv("CODEX_CLI_PATH", "").strip()
or "codex"
)
raw_args = os.getenv("HERMES_CODEX_CLI_ARGS", "").strip()
default_args = [
"exec",
"--json",
"--ephemeral",
"--dangerously-bypass-approvals-and-sandbox",
"--skip-git-repo-check",
]
args = shlex.split(raw_args) if raw_args else default_args
base_url = os.getenv(pconfig.base_url_env_var, "").strip() if pconfig.base_url_env_var else ""
if not base_url:
base_url = pconfig.inference_base_url
resolved_command = shutil.which(command) if command else None
return {
"configured": bool(resolved_command),
"provider": provider_id,
"name": pconfig.name,
"command": command,
"args": args,
"resolved_command": resolved_command,
"base_url": base_url,
"logged_in": bool(resolved_command),
}
return {"configured": False}
def get_auth_status(provider_id: Optional[str] = None) -> Dict[str, Any]:
@@ -4048,6 +4088,8 @@ def get_auth_status(provider_id: Optional[str] = None) -> Dict[str, Any]:
return get_gemini_oauth_auth_status()
if target == "copilot-acp":
return get_external_process_provider_status(target)
if target == "codex-cli":
return get_external_process_provider_status(target)
# API-key providers
pconfig = PROVIDER_REGISTRY.get(target)
if pconfig and pconfig.auth_type == "api_key":
@@ -4121,30 +4163,69 @@ def resolve_external_process_provider_credentials(provider_id: str) -> Dict[str,
if not base_url:
base_url = pconfig.inference_base_url
command = (
os.getenv("HERMES_COPILOT_ACP_COMMAND", "").strip()
or os.getenv("COPILOT_CLI_PATH", "").strip()
or "copilot"
)
raw_args = os.getenv("HERMES_COPILOT_ACP_ARGS", "").strip()
args = shlex.split(raw_args) if raw_args else ["--acp", "--stdio"]
resolved_command = shutil.which(command) if command else None
if not resolved_command and not base_url.startswith("acp+tcp://"):
raise AuthError(
f"Could not find the Copilot CLI command '{command}'. "
"Install GitHub Copilot CLI or set HERMES_COPILOT_ACP_COMMAND/COPILOT_CLI_PATH.",
provider=provider_id,
code="missing_copilot_cli",
if provider_id == "copilot-acp":
command = (
os.getenv("HERMES_COPILOT_ACP_COMMAND", "").strip()
or os.getenv("COPILOT_CLI_PATH", "").strip()
or "copilot"
)
raw_args = os.getenv("HERMES_COPILOT_ACP_ARGS", "").strip()
args = shlex.split(raw_args) if raw_args else ["--acp", "--stdio"]
resolved_command = shutil.which(command) if command else None
if not resolved_command and not base_url.startswith("acp+tcp://"):
raise AuthError(
f"Could not find the Copilot CLI command '{command}'. "
"Install GitHub Copilot CLI or set HERMES_COPILOT_ACP_COMMAND/COPILOT_CLI_PATH.",
provider=provider_id,
code="missing_copilot_cli",
)
return {
"provider": provider_id,
"api_key": "copilot-acp",
"base_url": base_url.rstrip("/"),
"command": resolved_command or command,
"args": args,
"source": "process",
}
return {
"provider": provider_id,
"api_key": "copilot-acp",
"base_url": base_url.rstrip("/"),
"command": resolved_command or command,
"args": args,
"source": "process",
}
if provider_id == "codex-cli":
command = (
os.getenv("HERMES_CODEX_CLI_COMMAND", "").strip()
or os.getenv("CODEX_CLI_PATH", "").strip()
or "codex"
)
raw_args = os.getenv("HERMES_CODEX_CLI_ARGS", "").strip()
default_args = [
"exec",
"--json",
"--ephemeral",
"--dangerously-bypass-approvals-and-sandbox",
"--skip-git-repo-check",
]
args = shlex.split(raw_args) if raw_args else default_args
resolved_command = shutil.which(command) if command else None
if not resolved_command:
raise AuthError(
f"Could not find the Codex CLI command '{command}'. "
"Install Codex CLI (npm install -g @openai/codex) or set "
"HERMES_CODEX_CLI_COMMAND / CODEX_CLI_PATH.",
provider=provider_id,
code="missing_codex_cli",
)
return {
"provider": provider_id,
"api_key": "codex-cli",
"base_url": base_url.rstrip("/"),
"command": resolved_command or command,
"args": args,
"source": "process",
}
raise AuthError(
f"Unknown external-process provider '{provider_id}'.",
provider=provider_id,
code="unknown_external_process_provider",
)
# =============================================================================

View File

@@ -8858,7 +8858,7 @@ def _build_provider_choices() -> list[str]:
except Exception:
# Fallback: static list guarantees the CLI always works
return [
"auto", "openrouter", "nous", "openai-codex", "copilot-acp", "copilot",
"auto", "openrouter", "nous", "openai-codex", "copilot-acp", "codex-cli", "copilot",
"anthropic", "gemini", "google-gemini-cli", "xai", "bedrock", "azure-foundry",
"ollama-cloud", "huggingface", "zai", "kimi-coding", "kimi-coding-cn",
"stepfun", "minimax", "minimax-cn", "kilocode", "xiaomi", "arcee",

View File

@@ -207,6 +207,17 @@ _PROVIDER_MODELS: dict[str, list[str]] = {
"copilot-acp": [
"copilot-acp",
],
"codex-cli": [
"gpt-5.5",
"gpt-5.4",
"gpt-5.4-mini",
"gpt-5.3-codex",
"gpt-5.2-codex",
"gpt-5.1-codex-max",
"gpt-5.1-codex-mini",
"o3",
"o4-mini",
],
"copilot": [
"gpt-5.4",
"gpt-5.4-mini",
@@ -799,6 +810,7 @@ CANONICAL_PROVIDERS: list[ProviderEntry] = [
ProviderEntry("qwen-oauth", "Qwen OAuth (Portal)", "Qwen OAuth (reuses local Qwen CLI login)"),
ProviderEntry("copilot", "GitHub Copilot", "GitHub Copilot (uses GITHUB_TOKEN or gh auth token)"),
ProviderEntry("copilot-acp", "GitHub Copilot ACP", "GitHub Copilot ACP (spawns `copilot --acp --stdio`)"),
ProviderEntry("codex-cli", "OpenAI Codex CLI", "OpenAI Codex CLI (spawns `codex exec --json` — text-only MVP, Hermes tools disabled)"),
ProviderEntry("huggingface", "Hugging Face", "Hugging Face Inference Providers (20+ open models)"),
ProviderEntry("gemini", "Google AI Studio", "Google AI Studio (Gemini models — native Gemini API)"),
ProviderEntry("google-gemini-cli", "Google Gemini (OAuth)", "Google Gemini via OAuth + Code Assist (free tier supported; no API key needed)"),
@@ -858,6 +870,8 @@ _PROVIDER_ALIASES = {
"github-model": "copilot",
"github-copilot-acp": "copilot-acp",
"copilot-acp-agent": "copilot-acp",
"codexcli": "codex-cli",
"openai-codex-cli": "codex-cli",
"google": "gemini",
"google-gemini": "gemini",
"google-ai-studio": "gemini",

View File

@@ -1137,6 +1137,19 @@ def resolve_runtime_provider(
"requested_provider": requested_provider,
}
if provider == "codex-cli":
creds = resolve_external_process_provider_credentials(provider)
return {
"provider": "codex-cli",
"api_mode": "chat_completions",
"base_url": creds.get("base_url", "").rstrip("/"),
"api_key": creds.get("api_key", ""),
"command": creds.get("command", ""),
"args": list(creds.get("args") or []),
"source": creds.get("source", "process"),
"requested_provider": requested_provider,
}
# Anthropic (native Messages API)
if provider == "anthropic":
# Allow base URL override from config.yaml model.base_url, but only

View File

@@ -1264,6 +1264,7 @@ class AIAgent:
api_mode is None
and self.api_mode == "chat_completions"
and self.provider != "copilot-acp"
and self.provider != "codex-cli"
and not str(self.base_url or "").lower().startswith("acp://copilot")
and not str(self.base_url or "").lower().startswith("acp+tcp://")
and not self._is_azure_openai_url()
@@ -1587,6 +1588,9 @@ class AIAgent:
if self.provider == "copilot-acp":
client_kwargs["command"] = self.acp_command
client_kwargs["args"] = self.acp_args
if self.provider == "codex-cli":
client_kwargs["command"] = self.acp_command
client_kwargs["args"] = self.acp_args
effective_base = base_url
if base_url_host_matches(effective_base, "openrouter.ai"):
from agent.auxiliary_client import build_or_headers
@@ -1761,6 +1765,11 @@ class AIAgent:
disabled_toolsets=disabled_toolsets,
quiet_mode=self.quiet_mode,
)
# Codex CLI provider is text-in/text-out MVP — Hermes tools are disabled
# because Codex handles its own tool calling internally via `codex exec`.
if self.provider == "codex-cli":
self.tools = []
# Show tool configuration and store valid tool names for validation
self.valid_tool_names = set()
@@ -5959,6 +5968,17 @@ class AIAgent:
self._client_log_context(),
)
return client
if self.provider == "codex-cli" or str(client_kwargs.get("base_url", "")).startswith("codex-cli://"):
from agent.codex_cli_client import CodexCLIClient
client = CodexCLIClient(**client_kwargs)
logger.info(
"Codex CLI client created (%s, shared=%s) %s",
reason,
shared,
self._client_log_context(),
)
return client
if self.provider == "google-gemini-cli" or str(client_kwargs.get("base_url", "")).startswith("cloudcode-pa://"):
from agent.gemini_cloudcode_adapter import GeminiCloudCodeClient
@@ -11809,8 +11829,10 @@ class AIAgent:
# API upgrade (lines ~1083-1085).
elif (
self.provider == "copilot-acp"
or self.provider == "codex-cli"
or str(self.base_url or "").lower().startswith("acp://copilot")
or str(self.base_url or "").lower().startswith("acp+tcp://")
or str(self.base_url or "").lower().startswith("codex-cli://")
):
_use_streaming = False
elif not self._has_stream_consumers():

View File

@@ -0,0 +1,107 @@
"""Tests for the codex-cli external-process provider."""
from __future__ import annotations
import os
from unittest.mock import patch
import pytest
# CRITICAL: import directly from the module to avoid module-level side effects
from hermes_cli.auth import (
PROVIDER_REGISTRY,
get_external_process_provider_status,
get_auth_status,
resolve_external_process_provider_credentials,
)
class TestCodexCLIProviderRegistry:
"""Test that the codex-cli provider is correctly registered."""
def test_provider_registered(self):
assert "codex-cli" in PROVIDER_REGISTRY
pconfig = PROVIDER_REGISTRY["codex-cli"]
assert pconfig.name == "OpenAI Codex CLI"
assert pconfig.auth_type == "external_process"
assert pconfig.inference_base_url == "codex-cli://local"
assert pconfig.base_url_env_var == "CODEX_CLI_BASE_URL"
def test_aliases_resolve(self):
from hermes_cli.auth import resolve_provider
assert resolve_provider("codexcli") == "codex-cli"
assert resolve_provider("openai-codex-cli") == "codex-cli"
class TestCodexCLIStatus:
"""Test the external-process status helper for codex-cli."""
def test_status_not_configured_when_codex_missing(self):
with patch.dict(os.environ, {}, clear=True):
status = get_external_process_provider_status("codex-cli")
assert status["configured"] is False
assert status["provider"] == "codex-cli"
def test_status_configured_when_codex_exists(self):
with patch.dict(os.environ, {"PATH": "/usr/bin:/bin"}):
with patch("shutil.which", return_value="/opt/homebrew/bin/codex"):
status = get_external_process_provider_status("codex-cli")
assert status["configured"] is True
assert status["provider"] == "codex-cli"
assert status["resolved_command"] == "/opt/homebrew/bin/codex"
assert status["command"] == "codex"
def test_auth_status_dispatches(self):
with patch.dict(os.environ, {}, clear=True):
status = get_auth_status("codex-cli")
# Should not throw, returns a dict even when not configured
assert isinstance(status, dict)
assert "configured" in status or "logged_in" in status
def test_status_with_custom_command_env(self):
with patch.dict(os.environ, {"HERMES_CODEX_CLI_COMMAND": "/usr/local/bin/my-codex"}, clear=False):
status = get_external_process_provider_status("codex-cli")
assert status["command"] == "/usr/local/bin/my-codex"
assert status["command"] == "/usr/local/bin/my-codex"
def test_status_with_custom_args_env(self):
with patch.dict(os.environ, {
"HERMES_CODEX_CLI_ARGS": "exec --json --model gpt-5.5",
}, clear=False):
status = get_external_process_provider_status("codex-cli")
assert "exec" in status["args"]
assert "--json" in status["args"]
assert "--model" in status["args"]
def test_status_unknown_provider(self):
status = get_external_process_provider_status("nonexistent")
assert status == {"configured": False}
class TestCodexCLICredentials:
"""Test the credential resolver for codex-cli."""
def test_resolves_command_path_when_available(self):
with patch.dict(os.environ, {}, clear=True):
with patch("shutil.which", return_value="/opt/homebrew/bin/codex"):
creds = resolve_external_process_provider_credentials("codex-cli")
assert creds["provider"] == "codex-cli"
assert creds["command"] == "/opt/homebrew/bin/codex"
assert creds["api_key"] == "codex-cli"
assert creds["base_url"] == "codex-cli://local"
assert "--json" in creds["args"]
assert "--ephemeral" in creds["args"]
def test_raises_when_command_missing(self):
with patch.dict(os.environ, {}, clear=True):
with patch("shutil.which", return_value=None):
with pytest.raises(Exception) as exc_info:
resolve_external_process_provider_credentials("codex-cli")
assert "codex-cli" in str(exc_info.value).lower() or "codex" in str(exc_info.value).lower()
def test_custom_command_from_env(self):
with patch.dict(os.environ, {"HERMES_CODEX_CLI_COMMAND": "/usr/local/bin/custom-codex"}, clear=False):
with patch("shutil.which", return_value="/usr/local/bin/custom-codex"):
creds = resolve_external_process_provider_credentials("codex-cli")
assert creds["command"] == "/usr/local/bin/custom-codex"