741 lines
30 KiB
Python
741 lines
30 KiB
Python
#!/usr/bin/env python3
|
|
"""Credential-isolating Codex Responses proxy for Hermes chat tenants."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import binascii
|
|
import fcntl
|
|
import hmac
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import time
|
|
from datetime import datetime, timezone
|
|
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"))
|
|
|
|
|
|
def _relay_token() -> str:
|
|
"""Read relay auth without exporting it to Codex subprocesses."""
|
|
path = Path(
|
|
os.environ.get(
|
|
"HERMES_CODEX_BROKER_KEY_FILE", "/runtime-access/chat-relay-key"
|
|
)
|
|
)
|
|
try:
|
|
return path.read_text(encoding="utf-8").strip()
|
|
except OSError:
|
|
return ""
|
|
|
|
|
|
TOKEN = _relay_token()
|
|
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/"
|
|
TOKEN_REFRESH_SKEW_SECONDS = int(
|
|
os.environ.get("HERMES_CODEX_BROKER_REFRESH_SKEW_SECONDS", "300")
|
|
)
|
|
HEALTH_PATH = Path(
|
|
os.environ.get("HERMES_CODEX_HEALTH_PATH", "/opt/data/provider-health/codex.json")
|
|
)
|
|
|
|
|
|
def _previous_health() -> dict[str, Any]:
|
|
"""Read the previous non-secret native transport health snapshot."""
|
|
try:
|
|
value = json.loads(HEALTH_PATH.read_text(encoding="utf-8"))
|
|
except (OSError, TypeError, ValueError, json.JSONDecodeError):
|
|
return {}
|
|
return value if isinstance(value, dict) else {}
|
|
|
|
|
|
def _record_health(state: str, **updates: Any) -> None:
|
|
"""Atomically publish current first-party Codex health without credentials."""
|
|
value = _previous_health()
|
|
value.update(
|
|
{
|
|
"transport": "codex-chatgpt-subscription",
|
|
"state": state,
|
|
"checked_at": datetime.now(timezone.utc).isoformat(),
|
|
"authenticated": state != "unavailable",
|
|
**updates,
|
|
}
|
|
)
|
|
try:
|
|
HEALTH_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = HEALTH_PATH.with_name(f".{HEALTH_PATH.name}.{os.getpid()}.tmp")
|
|
temporary.write_text(
|
|
json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8"
|
|
)
|
|
os.replace(temporary, HEALTH_PATH)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def _requested_effort(payload: dict[str, Any]) -> str:
|
|
"""Return the bounded Codex reasoning effort carried by a routed request."""
|
|
reasoning = payload.get("reasoning")
|
|
if isinstance(reasoning, dict) and reasoning.get("effort") in {
|
|
"low",
|
|
"medium",
|
|
"high",
|
|
"xhigh",
|
|
}:
|
|
return str(reasoning["effort"])
|
|
return "medium"
|
|
|
|
|
|
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 _token_expiry(token: str) -> float:
|
|
"""Return a JWT expiry timestamp, or zero for an opaque token."""
|
|
try:
|
|
encoded = token.split(".")[1]
|
|
encoded += "=" * (-len(encoded) % 4)
|
|
return float(json.loads(base64.urlsafe_b64decode(encoded)).get("exp", 0))
|
|
except (IndexError, ValueError, TypeError, json.JSONDecodeError, binascii.Error):
|
|
return 0.0
|
|
|
|
|
|
def _write_codex_auth(path: Path, payload: dict[str, Any]) -> None:
|
|
"""Atomically persist refreshed first-party credentials for every CLI lane."""
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
fd, temporary = tempfile.mkstemp(prefix=".auth.", suffix=".json", dir=path.parent)
|
|
try:
|
|
with os.fdopen(fd, "w", encoding="utf-8") as stream:
|
|
json.dump(payload, stream, separators=(",", ":"))
|
|
stream.write("\n")
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
os.chmod(temporary, 0o600)
|
|
os.replace(temporary, path)
|
|
finally:
|
|
try:
|
|
os.unlink(temporary)
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
|
|
def _access_token(*, force_refresh: bool = False) -> str:
|
|
"""Return a live ChatGPT OAuth token, refreshing the canonical CLI store."""
|
|
codex_home = Path(
|
|
os.environ.get("CODEX_HOME", str(Path.home() / ".codex"))
|
|
).expanduser()
|
|
auth_path = codex_home / "auth.json"
|
|
lock_path = codex_home / "hermes-codex-broker.lock"
|
|
codex_home.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
with lock_path.open("a+", encoding="utf-8") as lock:
|
|
fcntl.flock(lock.fileno(), fcntl.LOCK_EX)
|
|
payload = json.loads(auth_path.read_text(encoding="utf-8"))
|
|
if not isinstance(payload, dict):
|
|
raise RuntimeError("owner Codex authentication is invalid")
|
|
tokens = payload.get("tokens") or {}
|
|
if not isinstance(tokens, dict):
|
|
raise RuntimeError("owner Codex authentication is invalid")
|
|
token = str(tokens.get("access_token") or "").strip()
|
|
if not token:
|
|
raise RuntimeError("owner Codex access token is unavailable")
|
|
expires_at = _token_expiry(token)
|
|
should_refresh = force_refresh or (
|
|
bool(expires_at)
|
|
and expires_at <= time.time() + TOKEN_REFRESH_SKEW_SECONDS
|
|
)
|
|
if should_refresh:
|
|
refresh_token = str(tokens.get("refresh_token") or "").strip()
|
|
if not refresh_token:
|
|
raise RuntimeError("owner Codex refresh token is unavailable")
|
|
# Use the same first-party ChatGPT OAuth refresh as Codex CLI.
|
|
# This never introduces an OpenAI API key or metered billing.
|
|
from hermes_cli.auth import refresh_codex_oauth_pure
|
|
|
|
refreshed = refresh_codex_oauth_pure(
|
|
token,
|
|
refresh_token,
|
|
timeout_seconds=30.0,
|
|
)
|
|
tokens["access_token"] = refreshed["access_token"]
|
|
tokens["refresh_token"] = refreshed.get(
|
|
"refresh_token", refresh_token
|
|
)
|
|
payload["tokens"] = tokens
|
|
payload["last_refresh"] = refreshed.get("last_refresh")
|
|
_write_codex_auth(auth_path, payload)
|
|
token = str(tokens["access_token"]).strip()
|
|
return token
|
|
except RuntimeError:
|
|
raise
|
|
except (OSError, ValueError) as exc:
|
|
raise RuntimeError("owner Codex authentication is unavailable") from exc
|
|
|
|
|
|
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 _normalize_input_images(response_input: list[Any]) -> None:
|
|
"""Normalize Chat-Completions image parts for the Codex Responses API."""
|
|
|
|
def image_url_value(part: dict[str, Any]) -> tuple[str | None, str | None]:
|
|
"""Return a Responses URL from URL-object or base64-source forms."""
|
|
|
|
def search(candidate: Any, depth: int = 0) -> tuple[str | None, str | None]:
|
|
if isinstance(candidate, str):
|
|
if candidate.startswith(("data:image/", "https://", "http://")):
|
|
return candidate, None
|
|
return None, None
|
|
if not isinstance(candidate, dict) or depth > 4:
|
|
return None, None
|
|
detail = candidate.get("detail")
|
|
url = candidate.get("url")
|
|
if isinstance(url, str) and url.strip():
|
|
return url, detail if isinstance(detail, str) else None
|
|
data = candidate.get("data")
|
|
media_type = candidate.get("media_type") or candidate.get("mime_type")
|
|
if (
|
|
isinstance(data, str)
|
|
and data
|
|
and isinstance(media_type, str)
|
|
and media_type.startswith("image/")
|
|
and all(character not in media_type for character in "\r\n;, ")
|
|
):
|
|
return f"data:{media_type};base64,{data}", (
|
|
detail if isinstance(detail, str) else None
|
|
)
|
|
for nested in candidate.values():
|
|
nested_url, nested_detail = search(nested, depth + 1)
|
|
if nested_url:
|
|
return nested_url, (
|
|
detail if isinstance(detail, str) else nested_detail
|
|
)
|
|
return None, None
|
|
|
|
for candidate in (part.get("image_url"), part.get("source"), part):
|
|
image_url, detail = search(candidate)
|
|
if image_url:
|
|
return image_url, detail
|
|
return None, None
|
|
|
|
def image_shape(part: dict[str, Any]) -> str:
|
|
"""Describe only structural types when an upstream image is malformed."""
|
|
|
|
def describe(value: Any, depth: int = 0) -> Any:
|
|
if depth > 3:
|
|
return type(value).__name__
|
|
if isinstance(value, dict):
|
|
return {
|
|
str(key)[:40]: describe(nested, depth + 1)
|
|
for key, nested in list(value.items())[:12]
|
|
}
|
|
if isinstance(value, list):
|
|
return [describe(nested, depth + 1) for nested in value[:4]]
|
|
if isinstance(value, str):
|
|
return f"str[{len(value)}]"
|
|
return type(value).__name__
|
|
|
|
return json.dumps(describe(part), sort_keys=True, separators=(",", ":"))
|
|
|
|
for item in response_input:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
content = item.get("content")
|
|
if not isinstance(content, list):
|
|
continue
|
|
for part in content:
|
|
if not isinstance(part, dict) or part.get("type") not in {
|
|
"image",
|
|
"image_url",
|
|
"input_image",
|
|
}:
|
|
continue
|
|
image_url, detail = image_url_value(part)
|
|
if isinstance(detail, str) and detail and "detail" not in part:
|
|
part["detail"] = detail
|
|
if not isinstance(image_url, str) or not image_url.strip():
|
|
raise ValueError(
|
|
"non-empty Responses image URL required; shape="
|
|
+ image_shape(part)
|
|
)
|
|
part["type"] = "input_image"
|
|
part["image_url"] = image_url
|
|
part.pop("source", None)
|
|
|
|
|
|
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")
|
|
_normalize_input_images(payload["input"])
|
|
# 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:
|
|
token = _access_token()
|
|
except RuntimeError as exc:
|
|
_record_health(
|
|
"unavailable",
|
|
authenticated=False,
|
|
last_error_at=datetime.now(timezone.utc).isoformat(),
|
|
error_type=type(exc).__name__,
|
|
)
|
|
self._json(503, {"ok": False, "error": str(exc)})
|
|
return
|
|
_record_health(
|
|
"available",
|
|
authenticated=True,
|
|
token_expires_at=_token_expiry(token) or None,
|
|
)
|
|
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)
|
|
model = str(payload["model"])
|
|
effort = _requested_effort(payload)
|
|
started = time.monotonic()
|
|
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()
|
|
state = (
|
|
"capacity-limited"
|
|
if response.status_code == 429
|
|
else "unavailable"
|
|
if response.status_code in {401, 403}
|
|
else "degraded"
|
|
)
|
|
_record_health(
|
|
state,
|
|
last_error_at=datetime.now(timezone.utc).isoformat(),
|
|
status_code=response.status_code,
|
|
model=model,
|
|
effort=effort,
|
|
)
|
|
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()
|
|
)
|
|
_record_health(
|
|
"available",
|
|
last_success_at=datetime.now(timezone.utc).isoformat(),
|
|
latency_ms=int((time.monotonic() - started) * 1000),
|
|
model=model,
|
|
effort=effort,
|
|
token_expires_at=_token_expiry(token) or None,
|
|
)
|
|
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
|
|
_record_health(
|
|
"degraded",
|
|
last_error_at=datetime.now(timezone.utc).isoformat(),
|
|
error_type=type(exc).__name__,
|
|
)
|
|
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("runtime relay key is unavailable")
|
|
try:
|
|
token = _access_token()
|
|
except RuntimeError as exc:
|
|
_record_health(
|
|
"unavailable",
|
|
authenticated=False,
|
|
last_error_at=datetime.now(timezone.utc).isoformat(),
|
|
error_type=type(exc).__name__,
|
|
)
|
|
else:
|
|
_record_health(
|
|
"available",
|
|
authenticated=True,
|
|
token_expires_at=_token_expiry(token) or None,
|
|
)
|
|
ThreadingHTTPServer((HOST, PORT), Handler).serve_forever()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|