atlas-iac/services/hermes/scripts/patch_codex_runtime.py

#!/usr/bin/env python3
"""Let Hermes use the authenticated Codex app-server with routed settings."""

from __future__ import annotations

import argparse
from pathlib import Path


PROVIDER_BEFORE = '''    if provider == "openai-codex":
        try:
            creds = resolve_codex_runtime_credentials()
'''
PROVIDER_AFTER = '''    if provider == "openai-codex":
        if (
            str(model_cfg.get("openai_runtime") or "").strip().lower()
            == "codex_app_server"
        ):
            return {
                "provider": "openai-codex",
                "api_mode": "codex_app_server",
                "base_url": DEFAULT_CODEX_BASE_URL,
                "api_key": "codex-cli-runtime",
                "source": "codex-cli",
                "requested_provider": requested_provider,
            }
        try:
            creds = resolve_codex_runtime_credentials()
'''

SESSION_SIGNATURE_BEFORE = '''    def run_turn(
        self,
        user_input: Any,
        *,
        turn_timeout: float = 600.0,
'''
SESSION_SIGNATURE_AFTER = '''    def run_turn(
        self,
        user_input: Any,
        *,
        model: Optional[str] = None,
        effort: Optional[str] = None,
        turn_timeout: float = 600.0,
'''

SESSION_REQUEST_BEFORE = '''            ts = self._client.request(
                "turn/start",
                {
                    "threadId": self._thread_id,
                    "input": [{"type": "text", "text": user_input_text}],
                },
                timeout=10,
            )
'''
SESSION_REQUEST_AFTER = '''            turn_params: dict[str, Any] = {
                "threadId": self._thread_id,
                "input": [{"type": "text", "text": user_input_text}],
            }
            if model:
                turn_params["model"] = model
            if effort:
                turn_params["effort"] = effort
            ts = self._client.request(
                "turn/start",
                turn_params,
                timeout=10,
            )
'''

TURN_BEFORE = '''    try:
        turn = agent._codex_session.run_turn(user_input=user_message)
'''
TURN_AFTER = '''    reasoning = getattr(agent, "reasoning_config", None)
    effort = (
        str(reasoning.get("effort") or "").strip()
        if isinstance(reasoning, dict)
        else ""
    )
    try:
        turn = agent._codex_session.run_turn(
            user_input=user_message,
            model=str(getattr(agent, "model", "") or "").strip() or None,
            effort=effort or None,
        )
'''


def _replace_once(content: str, before: str, after: str, label: str) -> str:
    """Apply one exact replacement and fail closed when upstream drifts."""
    if content.count(before) != 1:
        raise RuntimeError(f"Hermes {label} patch context changed")
    return content.replace(before, after, 1)


def patch_provider(source: Path, destination: Path) -> None:
    """Allow the explicit Codex app-server runtime without duplicate OAuth."""
    content = source.read_text(encoding="utf-8")
    destination.parent.mkdir(parents=True, exist_ok=True)
    destination.write_text(
        _replace_once(
            content,
            PROVIDER_BEFORE,
            PROVIDER_AFTER,
            "runtime provider",
        ),
        encoding="utf-8",
    )


def patch_session(source: Path, destination: Path) -> None:
    """Pass the router's selected model and effort to each Codex turn."""
    content = source.read_text(encoding="utf-8")
    content = _replace_once(
        content,
        SESSION_SIGNATURE_BEFORE,
        SESSION_SIGNATURE_AFTER,
        "app-server session signature",
    )
    content = _replace_once(
        content,
        SESSION_REQUEST_BEFORE,
        SESSION_REQUEST_AFTER,
        "app-server turn request",
    )
    destination.parent.mkdir(parents=True, exist_ok=True)
    destination.write_text(content, encoding="utf-8")


def patch_turn(source: Path, destination: Path) -> None:
    """Forward live Hermes route metadata into the Codex session adapter."""
    content = source.read_text(encoding="utf-8")
    destination.parent.mkdir(parents=True, exist_ok=True)
    destination.write_text(
        _replace_once(content, TURN_BEFORE, TURN_AFTER, "Codex turn"),
        encoding="utf-8",
    )


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("provider_source", type=Path)
    parser.add_argument("provider_destination", type=Path)
    parser.add_argument("session_source", type=Path)
    parser.add_argument("session_destination", type=Path)
    parser.add_argument("turn_source", type=Path)
    parser.add_argument("turn_destination", type=Path)
    args = parser.parse_args()
    patch_provider(args.provider_source, args.provider_destination)
    patch_session(args.session_source, args.session_destination)
    patch_turn(args.turn_source, args.turn_destination)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())