472 lines
20 KiB
Python
472 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
"""Credential-isolating Codex Responses proxy for Hermes chat tenants."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import binascii
|
|
import hmac
|
|
import json
|
|
import os
|
|
import time
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from typing import Any, Iterable
|
|
|
|
import httpx
|
|
|
|
from routing_catalog import resolve_route
|
|
|
|
|
|
HOST = os.environ.get("HERMES_CODEX_BROKER_LISTEN_HOST", "0.0.0.0")
|
|
PORT = int(os.environ.get("HERMES_CODEX_BROKER_LISTEN_PORT", "9003"))
|
|
TOKEN = os.environ.get("HERMES_IMAGE_BROKER_KEY", "").strip()
|
|
UPSTREAM = os.environ.get(
|
|
"HERMES_CODEX_BROKER_UPSTREAM",
|
|
"https://chatgpt.com/backend-api/codex",
|
|
).rstrip("/")
|
|
MAX_BODY_BYTES = int(os.environ.get("HERMES_CODEX_BROKER_MAX_BODY", str(64 << 20)))
|
|
MAX_RESPONSE_BYTES = int(
|
|
os.environ.get("HERMES_CODEX_BROKER_MAX_RESPONSE", str(64 << 20))
|
|
)
|
|
READ_TIMEOUT_SECONDS = float(os.environ.get("HERMES_CODEX_BROKER_READ_TIMEOUT", "900"))
|
|
FALLBACK_ALLOWED_MODELS = {
|
|
value.strip()
|
|
for value in os.environ.get(
|
|
"HERMES_CODEX_BROKER_MODELS",
|
|
"gpt-5.6-luna,gpt-5.6-sol,gpt-5.6-terra",
|
|
).split(",")
|
|
if value.strip()
|
|
}
|
|
ROUTED_MODEL_PREFIX = "route/codex/"
|
|
|
|
|
|
def _real_model(model: str) -> str:
|
|
"""Translate a Switchyard effort-qualified target into a Codex model."""
|
|
if model.startswith(ROUTED_MODEL_PREFIX):
|
|
model = resolve_route(model)
|
|
return model
|
|
|
|
|
|
def _authorized(header: str | None) -> bool:
|
|
"""Authenticate a tenant without exposing the relay secret."""
|
|
if not TOKEN or not header or not header.startswith("Bearer "):
|
|
return False
|
|
return hmac.compare_digest(header[7:].strip(), TOKEN)
|
|
|
|
|
|
def _access_token() -> str:
|
|
"""Read the current owner token; the Codex CLI remains refresh owner."""
|
|
codex_home = Path(
|
|
os.environ.get("CODEX_HOME", str(Path.home() / ".codex"))
|
|
).expanduser()
|
|
try:
|
|
payload = json.loads((codex_home / "auth.json").read_text(encoding="utf-8"))
|
|
except (OSError, ValueError) as exc:
|
|
raise RuntimeError("owner Codex authentication is unavailable") from exc
|
|
if not isinstance(payload, dict):
|
|
raise RuntimeError("owner Codex authentication is invalid")
|
|
tokens = payload.get("tokens") or {}
|
|
token = tokens.get("access_token") if isinstance(tokens, dict) else None
|
|
if not isinstance(token, str) or not token.strip():
|
|
raise RuntimeError("owner Codex access token is unavailable")
|
|
token = token.strip()
|
|
try:
|
|
encoded = token.split(".")[1]
|
|
encoded += "=" * (-len(encoded) % 4)
|
|
expires_at = json.loads(base64.urlsafe_b64decode(encoded)).get("exp", 0)
|
|
except (IndexError, ValueError, TypeError, json.JSONDecodeError, binascii.Error):
|
|
expires_at = 0
|
|
if expires_at and time.time() >= float(expires_at):
|
|
raise RuntimeError("owner Codex access token is expired")
|
|
return token
|
|
|
|
|
|
def _upstream_headers(token: str) -> dict[str, str]:
|
|
"""Build the first-party headers expected by the Codex backend."""
|
|
from agent.auxiliary_client import _codex_cloudflare_headers
|
|
|
|
headers = _codex_cloudflare_headers(token)
|
|
headers.update(
|
|
{
|
|
"Accept": "text/event-stream",
|
|
"Authorization": f"Bearer {token}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
)
|
|
return headers
|
|
|
|
|
|
def _validate_payload(payload: Any) -> dict[str, Any]:
|
|
"""Allow only bounded Responses requests for the approved model catalog."""
|
|
if not isinstance(payload, dict):
|
|
raise ValueError("JSON object required")
|
|
model = payload.get("model")
|
|
if not isinstance(model, str):
|
|
raise ValueError("unsupported Codex model")
|
|
model = _real_model(model)
|
|
if not model.startswith("gpt-"):
|
|
raise ValueError("unsupported Codex model")
|
|
payload["model"] = model
|
|
response_input = payload.get("input")
|
|
if isinstance(response_input, str):
|
|
if not response_input.strip():
|
|
raise ValueError("non-empty Responses input required")
|
|
# Switchyard accepts OpenAI Chat Completions requests and translates
|
|
# their final text to a scalar Responses input. The first-party Codex
|
|
# endpoint is stricter than the public Responses API and only accepts
|
|
# a list of typed input items.
|
|
payload["input"] = [
|
|
{
|
|
"type": "message",
|
|
"role": "user",
|
|
"content": [{"type": "input_text", "text": response_input}],
|
|
}
|
|
]
|
|
elif isinstance(response_input, dict):
|
|
payload["input"] = [response_input]
|
|
elif not isinstance(response_input, list) or not response_input:
|
|
raise ValueError("non-empty Responses input list required")
|
|
# Switchyard uses ``max_output_tokens`` to bound the tiny classifier call,
|
|
# but its Responses translation can also copy that internal option onto the
|
|
# selected provider request. The first-party subscription Codex endpoint
|
|
# does not accept any of the public API token-budget aliases. Let Codex use
|
|
# its own response budget instead of turning a healthy fallback into a 400.
|
|
for unsupported_key in (
|
|
"max_output_tokens",
|
|
"max_completion_tokens",
|
|
"max_tokens",
|
|
# The subscription Codex backend owns sampling. OpenAI-compatible
|
|
# clients may add these public API fields during a retry; forwarding
|
|
# them turns an otherwise healthy fallback into HTTP 400.
|
|
"temperature",
|
|
"top_p",
|
|
"frequency_penalty",
|
|
"presence_penalty",
|
|
"logprobs",
|
|
"top_logprobs",
|
|
"seed",
|
|
"n",
|
|
"stop",
|
|
):
|
|
payload.pop(unsupported_key, None)
|
|
# Tenant conversations must not enter the owner's server-side history.
|
|
payload["store"] = False
|
|
payload["stream"] = True
|
|
return payload
|
|
|
|
|
|
def _completed_response(lines: Iterable[str]) -> dict[str, Any]:
|
|
"""Collapse a Codex SSE stream for a non-streaming Responses caller."""
|
|
terminal_response: dict[str, Any] | None = None
|
|
output_items: dict[int, dict[str, Any]] = {}
|
|
function_argument_deltas: dict[str, list[str]] = {}
|
|
completed_function_arguments: set[str] = set()
|
|
upstream_error = ""
|
|
|
|
def function_key(event: dict[str, Any]) -> str:
|
|
"""Identify one streamed function call without retaining its content."""
|
|
item_id = str(event.get("item_id") or "").strip()
|
|
if item_id:
|
|
return item_id
|
|
output_index = event.get("output_index")
|
|
return f"output:{output_index}" if isinstance(output_index, int) else ""
|
|
|
|
def validate_arguments(arguments: Any) -> None:
|
|
"""Reject an incomplete JSON object so Switchyard can fail over."""
|
|
if not isinstance(arguments, str):
|
|
raise RuntimeError("Codex returned malformed function arguments")
|
|
try:
|
|
parsed_arguments = json.loads(arguments)
|
|
except (TypeError, ValueError, json.JSONDecodeError) as exc:
|
|
raise RuntimeError(
|
|
"Codex returned retryable malformed function arguments"
|
|
) from exc
|
|
if not isinstance(parsed_arguments, dict):
|
|
raise RuntimeError("Codex returned malformed function arguments")
|
|
|
|
for line in lines:
|
|
if not line.startswith("data:"):
|
|
continue
|
|
value = line[5:].strip()
|
|
if not value or value == "[DONE]":
|
|
continue
|
|
try:
|
|
event = json.loads(value)
|
|
except (TypeError, ValueError, json.JSONDecodeError):
|
|
continue
|
|
if not isinstance(event, dict):
|
|
continue
|
|
event_type = str(event.get("type") or "")
|
|
response = event.get("response")
|
|
if event_type == "response.function_call_arguments.delta":
|
|
key = function_key(event)
|
|
delta = event.get("delta")
|
|
if key and isinstance(delta, str):
|
|
function_argument_deltas.setdefault(key, []).append(delta)
|
|
elif event_type == "response.function_call_arguments.done":
|
|
key = function_key(event)
|
|
arguments = event.get("arguments")
|
|
if not isinstance(arguments, str) and key:
|
|
arguments = "".join(function_argument_deltas.get(key, []))
|
|
validate_arguments(arguments)
|
|
if key:
|
|
completed_function_arguments.add(key)
|
|
if event_type == "response.output_item.done":
|
|
item = event.get("item")
|
|
output_index = event.get("output_index")
|
|
if isinstance(item, dict) and isinstance(output_index, int):
|
|
output_items[output_index] = item
|
|
if event_type in {
|
|
"response.completed",
|
|
"response.failed",
|
|
"response.incomplete",
|
|
} and isinstance(response, dict):
|
|
terminal_response = response
|
|
elif event_type == "error":
|
|
error = event.get("error")
|
|
if isinstance(error, dict):
|
|
upstream_error = str(error.get("message") or error.get("type") or "")
|
|
else:
|
|
upstream_error = str(error or "")
|
|
for key, deltas in function_argument_deltas.items():
|
|
if key not in completed_function_arguments:
|
|
validate_arguments("".join(deltas))
|
|
if terminal_response is not None:
|
|
status = str(terminal_response.get("status") or "").lower()
|
|
if status != "completed":
|
|
details = terminal_response.get("incomplete_details") or {}
|
|
reason = (
|
|
str(details.get("reason") or "")
|
|
if isinstance(details, dict)
|
|
else ""
|
|
)
|
|
raise RuntimeError(
|
|
"Codex returned a retryable incomplete response"
|
|
+ (f": {reason}" if reason else "")
|
|
)
|
|
# The subscription Codex endpoint streams complete output items but
|
|
# currently leaves the terminal response's output array empty. Public
|
|
# Responses clients, including Switchyard, expect those items there.
|
|
if output_items and not terminal_response.get("output"):
|
|
terminal_response["output"] = [
|
|
output_items[index] for index in sorted(output_items)
|
|
]
|
|
# A Responses stream can be marked ``completed`` even when a function
|
|
# call was cut off at the provider's output boundary. Passing that
|
|
# downstream as HTTP 200 makes Hermes retry the same broken turn until
|
|
# it emits the unhelpful "Response truncated" message. Reject malformed
|
|
# terminal tool calls here so Switchyard can fail over to another
|
|
# eligible provider/model for this boundary.
|
|
for item in terminal_response.get("output") or []:
|
|
if not isinstance(item, dict) or item.get("type") != "function_call":
|
|
continue
|
|
if str(item.get("status") or "completed").lower() == "incomplete":
|
|
raise RuntimeError("Codex returned a retryable incomplete tool call")
|
|
arguments = item.get("arguments")
|
|
validate_arguments(arguments)
|
|
return terminal_response
|
|
raise RuntimeError(upstream_error or "Codex stream ended without a terminal response")
|
|
|
|
|
|
def _normalized_stream(body: bytes, completed: dict[str, Any]) -> bytes:
|
|
"""Return Responses SSE with the reconstructed terminal response attached."""
|
|
normalized: list[str] = []
|
|
replaced_terminal = False
|
|
pending_event_line = ""
|
|
for line in body.decode("utf-8", errors="replace").splitlines():
|
|
if line.startswith("event:"):
|
|
pending_event_line = line
|
|
continue
|
|
if line.startswith("data:"):
|
|
value = line[5:].strip()
|
|
try:
|
|
event = json.loads(value)
|
|
except (TypeError, ValueError, json.JSONDecodeError):
|
|
event = None
|
|
event_type = event.get("type") if isinstance(event, dict) else ""
|
|
item = event.get("item") if isinstance(event, dict) else None
|
|
if event_type == "response.function_call_arguments.done" or (
|
|
event_type == "response.output_item.done"
|
|
and isinstance(item, dict)
|
|
and item.get("type") == "function_call"
|
|
):
|
|
# Switchyard translates each terminal function-call snapshot
|
|
# into another Chat Completions argument delta. Retain the
|
|
# already-validated delta sequence, not its duplicate copies.
|
|
pending_event_line = ""
|
|
continue
|
|
if isinstance(event, dict) and event_type == "response.completed":
|
|
stream_completed = dict(completed)
|
|
stream_completed["output"] = [
|
|
output_item
|
|
for output_item in stream_completed.get("output") or []
|
|
if not isinstance(output_item, dict)
|
|
or output_item.get("type") != "function_call"
|
|
]
|
|
event["response"] = stream_completed
|
|
line = "data: " + json.dumps(event, separators=(",", ":"))
|
|
replaced_terminal = True
|
|
if pending_event_line:
|
|
normalized.append(pending_event_line)
|
|
pending_event_line = ""
|
|
elif pending_event_line and line:
|
|
normalized.append(pending_event_line)
|
|
pending_event_line = ""
|
|
normalized.append(line)
|
|
if pending_event_line:
|
|
normalized.append(pending_event_line)
|
|
if not replaced_terminal:
|
|
raise RuntimeError("Codex stream ended without a completed event")
|
|
# Preserve the blank event terminator required by SSE clients. Responses
|
|
# streams end at response.completed; they do not require a Chat
|
|
# Completions-style [DONE] sentinel.
|
|
return ("\n".join(normalized).rstrip("\n") + "\n\n").encode("utf-8")
|
|
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
"""Authenticated streaming proxy; request bodies and tokens are never logged."""
|
|
|
|
server_version = "HermesCodexBroker/1"
|
|
|
|
def _json(self, status: int, value: dict[str, Any]) -> None:
|
|
body = json.dumps(value, separators=(",", ":")).encode("utf-8")
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.send_header("Cache-Control", "no-store")
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def _check_auth(self) -> bool:
|
|
if _authorized(self.headers.get("Authorization")):
|
|
return True
|
|
self._json(401, {"error": {"message": "unauthorized", "type": "auth_error"}})
|
|
return False
|
|
|
|
def do_GET(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API
|
|
if not self._check_auth():
|
|
return
|
|
if self.path == "/health":
|
|
try:
|
|
_access_token()
|
|
except RuntimeError as exc:
|
|
self._json(503, {"ok": False, "error": str(exc)})
|
|
return
|
|
self._json(200, {"ok": True, "provider": "openai-codex"})
|
|
return
|
|
if self.path in {"/models", "/v1/models"}:
|
|
self._json(
|
|
200,
|
|
{
|
|
"object": "list",
|
|
"data": [
|
|
{"id": model, "object": "model", "owned_by": "openai-codex"}
|
|
for model in sorted(FALLBACK_ALLOWED_MODELS)
|
|
],
|
|
},
|
|
)
|
|
return
|
|
self._json(404, {"error": {"message": "not found", "type": "not_found"}})
|
|
|
|
def do_POST(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API
|
|
if self.path not in {"/responses", "/v1/responses"}:
|
|
self._json(404, {"error": {"message": "not found", "type": "not_found"}})
|
|
return
|
|
if not self._check_auth():
|
|
return
|
|
response_started = False
|
|
try:
|
|
length = int(self.headers.get("Content-Length", "0"))
|
|
except ValueError:
|
|
length = 0
|
|
if length <= 0 or length > MAX_BODY_BYTES:
|
|
self._json(413, {"error": {"message": "invalid request size", "type": "invalid_request_error"}})
|
|
return
|
|
|
|
try:
|
|
payload = json.loads(self.rfile.read(length))
|
|
requested_stream = payload.get("stream") is True
|
|
payload = _validate_payload(payload)
|
|
token = _access_token()
|
|
timeout = httpx.Timeout(
|
|
READ_TIMEOUT_SECONDS,
|
|
connect=30.0,
|
|
read=READ_TIMEOUT_SECONDS,
|
|
write=60.0,
|
|
pool=30.0,
|
|
)
|
|
with httpx.Client(timeout=timeout, headers=_upstream_headers(token)) as client:
|
|
with client.stream("POST", f"{UPSTREAM}/responses", json=payload) as response:
|
|
if response.status_code >= 400:
|
|
body = response.read()
|
|
self.send_response(response.status_code)
|
|
self.send_header("Content-Type", response.headers.get("Content-Type", "application/json"))
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.send_header("Cache-Control", "no-store")
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
return
|
|
|
|
# Validate the terminal Responses event before committing
|
|
# HTTP 200 downstream. The Codex subscription endpoint can
|
|
# end a tool call with ``response.incomplete`` while still
|
|
# returning HTTP 200. If that is streamed through, Hermes
|
|
# sees malformed JSON and eventually exposes "Response
|
|
# truncated" to the user. Buffering one model boundary
|
|
# lets Switchyard receive a retryable 502 and choose another
|
|
# provider/model instead. Tool execution remains streamed
|
|
# by Hermes after this short model boundary completes.
|
|
body = response.read()
|
|
if len(body) > MAX_RESPONSE_BYTES:
|
|
raise RuntimeError("Codex response exceeded broker limit")
|
|
completed = _completed_response(
|
|
body.decode("utf-8", errors="replace").splitlines()
|
|
)
|
|
if not requested_stream:
|
|
self._json(200, completed)
|
|
return
|
|
|
|
body = _normalized_stream(body, completed)
|
|
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "text/event-stream")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.send_header("Cache-Control", "no-store")
|
|
self.send_header("X-Accel-Buffering", "no")
|
|
self.end_headers()
|
|
response_started = True
|
|
self.wfile.write(body)
|
|
self.wfile.flush()
|
|
except ValueError as exc:
|
|
self._json(400, {"error": {"message": str(exc), "type": "invalid_request_error"}})
|
|
except (BrokenPipeError, ConnectionResetError):
|
|
return
|
|
except Exception as exc:
|
|
if response_started:
|
|
return
|
|
self._json(
|
|
502,
|
|
{
|
|
"error": {
|
|
"message": f"Codex broker failed: {type(exc).__name__}: {exc}",
|
|
"type": "upstream_error",
|
|
}
|
|
},
|
|
)
|
|
|
|
def log_message(self, format: str, *args: Any) -> None:
|
|
"""Log only method/path/status metadata, never bodies or headers."""
|
|
print(f"codex-broker {self.address_string()} {format % args}", flush=True)
|
|
|
|
|
|
def main() -> None:
|
|
"""Serve until Kubernetes terminates the sidecar."""
|
|
if not TOKEN:
|
|
raise SystemExit("HERMES_IMAGE_BROKER_KEY is required")
|
|
ThreadingHTTPServer((HOST, PORT), Handler).serve_forever()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|