File size: 11,423 Bytes
34c3ee8 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 | """Shared logic for the /codex-runtime slash command.
Toggles `model.openai_runtime` between "auto" (= chat_completions, Hermes'
default) and "codex_app_server" (= hand turns to a codex subprocess).
Both CLI (cli.py) and gateway (gateway/run.py) call into this module so the
behavior stays identical across surfaces.
The actual runtime resolution happens in hermes_cli.runtime_provider's
_maybe_apply_codex_app_server_runtime() helper, which reads the persisted
config value. This module just persists the value and reports the change.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass
from typing import Optional
logger = logging.getLogger(__name__)
VALID_RUNTIMES = ("auto", "codex_app_server")
@dataclass
class CodexRuntimeStatus:
"""Result of a /codex-runtime invocation. Callers render this however
suits their surface (CLI uses Rich panels, gateway sends a text message)."""
success: bool
new_value: Optional[str] = None
old_value: Optional[str] = None
message: str = ""
requires_new_session: bool = False
codex_binary_ok: bool = True
codex_version: Optional[str] = None
def parse_args(arg_string: str) -> tuple[Optional[str], list[str]]:
"""Parse the slash-command argument string. Returns (value, errors).
No args → return current state (value=None)
'auto' / 'codex_app_server' / 'on' / 'off' → return that value
anything else → error
"""
raw = (arg_string or "").strip().lower()
if not raw:
return None, []
# Accept human-friendly synonyms
if raw in {"on", "codex", "enable"}:
return "codex_app_server", []
if raw in {"off", "default", "disable", "hermes"}:
return "auto", []
if raw in VALID_RUNTIMES:
return raw, []
return None, [
f"Unknown runtime {raw!r}. Use one of: auto, codex_app_server, on, off"
]
def get_current_runtime(config: dict) -> str:
"""Read the current `model.openai_runtime` value from a config dict.
Returns 'auto' for unset / empty / unrecognized values."""
if not isinstance(config, dict):
return "auto"
model_cfg = config.get("model") or {}
if not isinstance(model_cfg, dict):
return "auto"
value = str(model_cfg.get("openai_runtime") or "").strip().lower()
if value in VALID_RUNTIMES:
return value
return "auto"
def set_runtime(config: dict, new_value: str) -> str:
"""Mutate the config dict in place to persist the new runtime value.
Returns the previous value for callers that want to report a delta."""
if new_value not in VALID_RUNTIMES:
raise ValueError(
f"invalid runtime {new_value!r}; must be one of {VALID_RUNTIMES}"
)
old = get_current_runtime(config)
if not isinstance(config.get("model"), dict):
config["model"] = {}
config["model"]["openai_runtime"] = new_value
return old
def check_codex_binary_ok() -> tuple[bool, Optional[str]]:
"""Best-effort verification that codex CLI is installed at acceptable
version. Returns (ok, version_or_message)."""
try:
from agent.transports.codex_app_server import check_codex_binary
return check_codex_binary()
except Exception as exc: # pragma: no cover
return False, f"codex check failed: {exc}"
def apply(
config: dict,
new_value: Optional[str],
*,
persist_callback=None,
) -> CodexRuntimeStatus:
"""Top-level entry point used by both CLI and gateway handlers.
Args:
config: in-memory config dict (will be mutated when new_value is set)
new_value: desired runtime; None means "show current state only"
persist_callback: optional callable taking the mutated config dict
and persisting it to disk. Skipped when None (used by tests).
Returns: CodexRuntimeStatus describing the outcome.
"""
current = get_current_runtime(config)
# Cache the codex binary check for this apply() call. Subprocess spawn
# is cheap (~50ms for `codex --version`), but we'd otherwise call it up
# to 3 times in the enable path (read-only/state, gate, success message).
# None = not yet checked; (bool, str) = result.
_binary_check: Optional[tuple[bool, Optional[str]]] = None
def _check_binary_cached() -> tuple[bool, Optional[str]]:
nonlocal _binary_check
if _binary_check is None:
_binary_check = check_codex_binary_ok()
return _binary_check
# Read-only call: just report state
if new_value is None:
ok, ver = _check_binary_cached()
msg = (
f"openai_runtime: {current}\n"
f"codex CLI: {'OK ' + ver if ok else 'not available — ' + (ver or 'install with `npm i -g @openai/codex`')}"
)
return CodexRuntimeStatus(
success=True,
new_value=current,
old_value=current,
message=msg,
codex_binary_ok=ok,
codex_version=ver if ok else None,
)
# No-config-change paths. For `auto` we return immediately — disabling
# doesn't touch ~/.codex/. For `codex_app_server`, we fall through to
# the migration block below: the config value is already correct, but
# the world state (managed block in ~/.codex/config.toml, hermes-tools
# MCP callback, plugin discovery) may be stale or missing — common
# footgun when users pre-set `openai_runtime: codex_app_server` in
# config.yaml without ever running the slash command. The migration is
# idempotent by design (it replaces its own managed block in place), so
# re-running is cheap and safe.
reapplying_enable = new_value == current == "codex_app_server"
if new_value == current and not reapplying_enable:
return CodexRuntimeStatus(
success=True,
new_value=current,
old_value=current,
message=f"openai_runtime already set to {current}",
)
# If switching ON, verify codex CLI is installed before persisting —
# an opt-in toggle that silently fails on the first turn is the
# worst possible UX. Block here with a clear install hint.
if new_value == "codex_app_server":
ok, ver_or_msg = _check_binary_cached()
if not ok:
return CodexRuntimeStatus(
success=False,
new_value=None,
old_value=current,
message=(
"Cannot enable codex_app_server runtime: "
f"{ver_or_msg or 'codex CLI not available'}\n"
"Install with: npm i -g @openai/codex"
),
codex_binary_ok=False,
codex_version=None,
)
if not reapplying_enable:
set_runtime(config, new_value)
if persist_callback is not None:
try:
persist_callback(config)
except Exception as exc:
logger.exception("failed to persist openai_runtime change")
return CodexRuntimeStatus(
success=False,
new_value=new_value,
old_value=current,
message=f"updated config in memory but persist failed: {exc}",
)
if reapplying_enable:
msg_lines = [
f"openai_runtime already set to {current} — re-applying migration"
]
else:
msg_lines = [f"openai_runtime: {current} → {new_value}"]
if new_value == "codex_app_server":
ok, ver = _check_binary_cached()
if ok:
msg_lines.append(f"codex CLI: {ver}")
# Auto-migrate Hermes' MCP servers + Codex's installed curated
# plugins into ~/.codex/config.toml so the spawned codex subprocess
# sees the same tool surface AND can call back into Hermes for
# browser/web/delegate_task/vision/memory tools (#7 fix).
# Failures are non-fatal — the runtime change still proceeds.
try:
from hermes_cli.codex_runtime_plugin_migration import migrate
mig_report = migrate(config)
# Tools/MCP servers (excluding the hermes-tools callback,
# which is internal plumbing — surface separately).
user_servers = [
s for s in mig_report.migrated if s != "hermes-tools"
]
if user_servers:
msg_lines.append(
f"Migrated {len(user_servers)} MCP server(s): "
f"{', '.join(user_servers)}"
)
# Native Codex plugin migration (Linear, GitHub, etc.)
if mig_report.migrated_plugins:
msg_lines.append(
f"Migrated {len(mig_report.migrated_plugins)} native "
f"Codex plugin(s): {', '.join(mig_report.migrated_plugins)}"
)
elif mig_report.plugin_query_error:
msg_lines.append(
f"Codex plugin discovery skipped: "
f"{mig_report.plugin_query_error}"
)
# Permissions + Hermes tool callback are always-on production
# bits the user benefits from knowing about.
if mig_report.wrote_permissions_default:
msg_lines.append(
f"Default sandbox: {mig_report.wrote_permissions_default} "
f"(no approval prompt on every write)"
)
if "hermes-tools" in mig_report.migrated:
msg_lines.append(
"Hermes tool callback registered: codex can now use "
"web_search, web_extract, browser_*, vision_analyze, "
"image_generate, skill_view, skills_list, text_to_speech, "
"kanban_* (worker + orchestrator) via MCP."
)
msg_lines.append(
" (delegate_task, memory, session_search, todo run "
"only on the default Hermes runtime — they need the "
"agent loop context.)"
)
msg_lines.append(f" (config: {mig_report.target_path})")
for err in mig_report.errors:
msg_lines.append(f"⚠ MCP migration: {err}")
except Exception as exc:
msg_lines.append(f"⚠ MCP migration skipped: {exc}")
msg_lines.append(
"OpenAI/Codex turns now run through `codex app-server` "
"(terminal/file ops/patching inside Codex; "
"Hermes tools available via MCP callback)."
)
msg_lines.append(
"Effective on next session — current cached agent keeps "
"the prior runtime to preserve prompt cache."
)
else:
msg_lines.append("OpenAI/Codex turns will use the default Hermes runtime.")
msg_lines.append("Effective on next session.")
return CodexRuntimeStatus(
success=True,
new_value=new_value,
old_value=current,
message="\n".join(msg_lines),
requires_new_session=True,
)
|