158 lines
5.0 KiB
Python
158 lines
5.0 KiB
Python
#!/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}],
|
|
"approvalPolicy": "never",
|
|
"sandboxPolicy": {"type": "dangerFullAccess"},
|
|
}
|
|
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())
|