From 05b40ecd8e5f3d70947f1a263228ad4602f54af7 Mon Sep 17 00:00:00 2001 From: luckeyfaraday Date: Sat, 27 Jun 2026 14:07:35 +0200 Subject: [PATCH] Add Codex OAuth routing for openai/* panel models Route openai/* slugs through a ChatGPT subscription via Codex OAuth instead of OpenRouter when FUSION_CODEX_OAUTH is enabled. Reuses tokens from ~/.codex/auth.json (codex login), translates chat completions to the Codex Responses API, and relaxes the OpenRouter key requirement when every model in a run can be served on the Codex path. Includes codex_auth transport module, engine integration, env docs, and unit tests for payload translation, SSE accumulation, and routing logic. --- .env.example | 11 +- codex_auth.py | 432 +++++++++++++++++++++++++++++++++++++++ fusion.py | 195 ++++++++++++++++-- pyproject.toml | 2 +- tests/test_codex_auth.py | 204 ++++++++++++++++++ 5 files changed, 830 insertions(+), 14 deletions(-) create mode 100644 codex_auth.py create mode 100644 tests/test_codex_auth.py diff --git a/.env.example b/.env.example index 30c8c37..6b59b52 100644 --- a/.env.example +++ b/.env.example @@ -3,9 +3,18 @@ # cp .env.example .env # `.env` is git-ignored; never commit real keys. -# REQUIRED. OpenRouter API key. Create one at https://openrouter.ai/keys +# REQUIRED (unless every model is served via Codex OAuth — see below). +# OpenRouter API key. Create one at https://openrouter.ai/keys OPENROUTER_API_KEY=sk-or-v1-... +# Optional. Serve `openai/*` models through a ChatGPT subscription via Codex +# OAuth instead of OpenRouter (no per-token billing for those calls). Requires a +# prior `codex login` (the Codex CLI) so tokens exist at ~/.codex/auth.json. +# Set to 1/true/yes to enable. Non-OpenAI models still go through OpenRouter. +FUSION_CODEX_OAUTH= +# Optional override for the Codex token file (default: ~/.codex/auth.json). +CODEX_AUTH_FILE= + # Panel used when --panel/-p is not passed on the CLI (budget | quality | code) FUSION_DEFAULT_PANEL=budget diff --git a/codex_auth.py b/codex_auth.py new file mode 100644 index 0000000..f7e89da --- /dev/null +++ b/codex_auth.py @@ -0,0 +1,432 @@ +"""Codex (ChatGPT) OAuth transport for routing OpenAI models off OpenRouter. + +OpenRouter bills every panel call against API credits. OpenAI's Codex CLI, by +contrast, signs in with a *ChatGPT* account (Plus/Pro/Team) and calls OpenAI +directly against the subscription. This module lets Fusion Engine reuse that +same login so ``openai/*`` panel members can be served through the ChatGPT +subscription instead of OpenRouter. + +It does **not** implement the browser login flow. It reads the tokens that the +Codex CLI already wrote to ``~/.codex/auth.json`` (run ``codex login`` once), +refreshes them when expired, and translates between the OpenAI-style +``chat/completions`` shape the rest of Fusion Engine speaks and OpenAI's +*Responses* API, which is what the Codex backend actually accepts. + +Caveats worth knowing before relying on this: +- The Codex backend protocol (endpoint, headers, streaming event names) is + reverse-engineered from the Codex CLI and can change without notice. +- Driving a ChatGPT subscription programmatically is outside OpenAI's normal API + terms. This is a convenience for personal use, not a supported integration. + +The translation helpers (:func:`build_responses_payload`, +:class:`ResponsesAccumulator`) are pure functions of their inputs and are unit +tested without any network access; only :class:`CodexAuth` and the streaming +loop in ``fusion.py`` touch the wire. +""" + +from __future__ import annotations + +import asyncio +import base64 +import json +import logging +import os +import time +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Optional + +import httpx + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- # +# Protocol constants (mirror the Codex CLI) +# --------------------------------------------------------------------------- # + +# Public OAuth client id baked into the Codex CLI. Not a secret. +CODEX_CLIENT_ID = "app_EMoamEEZ73f0CkXaXp7hrann" + +# OpenAI OAuth token endpoint, used here only for the refresh_token grant. +OPENAI_TOKEN_URL = "https://auth.openai.com/oauth/token" + +# The Codex backend's Responses endpoint, served against ChatGPT-subscription +# auth (distinct from api.openai.com, which expects an API key). +CODEX_RESPONSES_URL = "https://chatgpt.com/backend-api/codex/responses" + +# Default location of the Codex CLI's stored session. +DEFAULT_AUTH_FILE = Path.home() / ".codex" / "auth.json" + +# Refresh a little before the access token's real expiry to avoid races. +REFRESH_LEEWAY_SECONDS = 60 + + +class CodexAuthError(RuntimeError): + """Raised when Codex OAuth credentials are missing or cannot be refreshed. + + The message is intended to be surfaced to the user (e.g. "run ``codex + login``"), so callers can put it straight into a ``PanelResponse.error``. + """ + + +# --------------------------------------------------------------------------- # +# JWT helpers (claims only — we never verify signatures) +# --------------------------------------------------------------------------- # + +def decode_jwt_claims(token: str) -> dict[str, Any]: + """Return the claims payload of a JWT without verifying its signature. + + We only ever read tokens we obtained from the user's own local auth file or + straight from OpenAI's token endpoint, so there is nothing to verify against + an attacker here — we just need ``exp`` and the account-id claim. Returns an + empty dict if the token is not a well-formed three-segment JWT. + """ + try: + payload_b64 = token.split(".")[1] + except (AttributeError, IndexError): + return {} + # Base64url payloads omit padding; restore it before decoding. + padding = "=" * (-len(payload_b64) % 4) + try: + raw = base64.urlsafe_b64decode(payload_b64 + padding) + claims = json.loads(raw) + except (ValueError, json.JSONDecodeError): + return {} + return claims if isinstance(claims, dict) else {} + + +# --------------------------------------------------------------------------- # +# Credential store +# --------------------------------------------------------------------------- # + +class CodexAuth: + """Reads, refreshes, and serves Codex OAuth tokens from ``~/.codex/auth.json``. + + A single instance is safe to share across concurrent panel calls: refreshes + are serialized with an :class:`asyncio.Lock` so a burst of expired-token + requests triggers exactly one refresh. + """ + + def __init__( + self, + auth_file: Optional[str | Path] = None, + client_id: str = CODEX_CLIENT_ID, + refresh_leeway: int = REFRESH_LEEWAY_SECONDS, + ) -> None: + self.auth_file = Path( + auth_file + or os.environ.get("CODEX_AUTH_FILE") + or DEFAULT_AUTH_FILE + ) + self.client_id = client_id + self.refresh_leeway = refresh_leeway + self._lock = asyncio.Lock() + # Parsed auth.json, cached after first read and updated on refresh. + self._data: Optional[dict[str, Any]] = None + + # ----------------------------- file I/O ------------------------------- # + + def _read_file(self) -> dict[str, Any]: + try: + data = json.loads(self.auth_file.read_text(encoding="utf-8")) + except FileNotFoundError as exc: + raise CodexAuthError( + f"Codex auth file not found at {self.auth_file}. " + "Run `codex login` (with the Codex CLI) first, or set " + "CODEX_AUTH_FILE." + ) from exc + except (OSError, json.JSONDecodeError) as exc: + raise CodexAuthError( + f"Could not read Codex auth file {self.auth_file}: {exc}" + ) from exc + if not isinstance(data, dict): + raise CodexAuthError( + f"Codex auth file {self.auth_file} is not a JSON object." + ) + return data + + def _write_file(self, data: dict[str, Any]) -> None: + """Persist refreshed tokens, best-effort, with owner-only permissions.""" + try: + self.auth_file.write_text( + json.dumps(data, indent=2), encoding="utf-8" + ) + os.chmod(self.auth_file, 0o600) + except OSError as exc: + # A failed write is non-fatal: the in-memory token still works for + # this process; we just won't have persisted it for the next one. + logger.warning( + "Could not write refreshed Codex tokens to %s: %s", + self.auth_file, + exc, + ) + + @staticmethod + def _tokens(data: dict[str, Any]) -> dict[str, Any]: + tokens = data.get("tokens") + if not isinstance(tokens, dict) or not tokens.get("access_token"): + raise CodexAuthError( + "Codex auth file has no access token; run `codex login`." + ) + return tokens + + # --------------------------- token logic ------------------------------ # + + def _expired(self, access_token: str) -> bool: + """True if the access token's ``exp`` is within the refresh leeway. + + If the token carries no ``exp`` claim we cannot tell, so we treat it as + valid and let the server reject it rather than refresh-looping. + """ + exp = decode_jwt_claims(access_token).get("exp") + if not isinstance(exp, (int, float)): + return False + return time.time() >= (exp - self.refresh_leeway) + + @staticmethod + def _account_id(tokens: dict[str, Any]) -> Optional[str]: + """Resolve the ChatGPT account id from the tokens or the id_token claims.""" + if tokens.get("account_id"): + return str(tokens["account_id"]) + id_token = tokens.get("id_token") + if isinstance(id_token, str): + auth_claim = decode_jwt_claims(id_token).get( + "https://api.openai.com/auth" + ) + if isinstance(auth_claim, dict): + account_id = auth_claim.get("chatgpt_account_id") + if account_id: + return str(account_id) + return None + + async def _refresh( + self, client: httpx.AsyncClient, data: dict[str, Any] + ) -> dict[str, Any]: + tokens = self._tokens(data) + refresh_token = tokens.get("refresh_token") + if not refresh_token: + raise CodexAuthError( + "Codex access token expired and no refresh_token is available; " + "run `codex login` again." + ) + logger.info("Refreshing Codex OAuth access token") + try: + resp = await client.post( + OPENAI_TOKEN_URL, + json={ + "client_id": self.client_id, + "grant_type": "refresh_token", + "refresh_token": refresh_token, + "scope": "openid profile email", + }, + headers={"Content-Type": "application/json"}, + ) + resp.raise_for_status() + body = resp.json() + except (httpx.HTTPError, ValueError) as exc: + raise CodexAuthError(f"Codex token refresh failed: {exc}") from exc + + new_tokens = dict(tokens) + new_tokens["access_token"] = body["access_token"] + # OpenAI may rotate the refresh token and id_token; keep whatever it sent. + if body.get("refresh_token"): + new_tokens["refresh_token"] = body["refresh_token"] + if body.get("id_token"): + new_tokens["id_token"] = body["id_token"] + data["tokens"] = new_tokens + data["last_refresh"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) + self._write_file(data) + return data + + async def get_auth(self, client: httpx.AsyncClient) -> tuple[str, Optional[str]]: + """Return a valid ``(access_token, account_id)``, refreshing if needed. + + Raises: + CodexAuthError: If no usable credentials exist and cannot be + refreshed. The message is safe to show the user. + """ + async with self._lock: + data = self._data or self._read_file() + self._data = data + access_token = self._tokens(data)["access_token"] + if self._expired(access_token): + data = await self._refresh(client, data) + self._data = data + access_token = self._tokens(data)["access_token"] + return access_token, self._account_id(self._tokens(data)) + + +# --------------------------------------------------------------------------- # +# chat/completions <-> Responses API translation +# --------------------------------------------------------------------------- # + +def _tool_to_responses(tool: dict[str, Any]) -> dict[str, Any]: + """Flatten a chat-style function tool into the Responses ``tools`` shape. + + chat: ``{"type":"function","function":{"name","description","parameters"}}`` + responses: ``{"type":"function","name","description","parameters"}`` + Non-function tools are passed through untouched. + """ + if tool.get("type") != "function" or "function" not in tool: + return tool + fn = tool["function"] + flattened: dict[str, Any] = {"type": "function", "name": fn.get("name")} + if "description" in fn: + flattened["description"] = fn["description"] + if "parameters" in fn: + flattened["parameters"] = fn["parameters"] + return flattened + + +def _message_text(content: Any) -> str: + """Coerce a chat ``content`` (string or content-part list) into plain text.""" + if isinstance(content, str): + return content + if isinstance(content, list): + parts = [ + part.get("text", "") + for part in content + if isinstance(part, dict) and part.get("type") in ("text", "input_text") + ] + return "".join(parts) + return "" if content is None else str(content) + + +def build_responses_payload( + model: str, + messages: list[dict[str, Any]], + tools: Optional[list[dict[str, Any]]] = None, + tool_choice: Optional[Any] = None, + max_tokens: Optional[int] = None, + stream: bool = True, +) -> dict[str, Any]: + """Translate a chat/completions request into a Codex Responses payload. + + ``system`` messages become the top-level ``instructions`` string; everything + else becomes an ``input`` item. Assistant tool calls and ``tool`` results are + mapped to ``function_call`` / ``function_call_output`` items so multi-turn + tool conversations round-trip. + """ + instructions: list[str] = [] + input_items: list[dict[str, Any]] = [] + + for msg in messages: + role = msg.get("role") + content = msg.get("content") + if role == "system": + instructions.append(_message_text(content)) + elif role == "tool": + input_items.append( + { + "type": "function_call_output", + "call_id": msg.get("tool_call_id"), + "output": _message_text(content), + } + ) + elif role == "assistant" and msg.get("tool_calls"): + for tc in msg["tool_calls"]: + fn = tc.get("function", {}) if isinstance(tc, dict) else {} + input_items.append( + { + "type": "function_call", + "call_id": tc.get("id"), + "name": fn.get("name"), + "arguments": fn.get("arguments", ""), + } + ) + if content: + input_items.append(_text_message_item("assistant", content)) + else: + input_items.append(_text_message_item(role or "user", content)) + + payload: dict[str, Any] = { + "model": model, + "instructions": "\n\n".join(p for p in instructions if p), + "input": input_items, + "stream": stream, + "store": False, + } + if tools: + payload["tools"] = [_tool_to_responses(t) for t in tools] + if tool_choice is not None: + payload["tool_choice"] = tool_choice + if max_tokens is not None: + payload["max_output_tokens"] = max_tokens + return payload + + +def _text_message_item(role: str, content: Any) -> dict[str, Any]: + """Build a Responses ``message`` input item with the role-correct text type.""" + text_type = "input_text" if role == "user" else "output_text" + return { + "type": "message", + "role": role, + "content": [{"type": text_type, "text": _message_text(content)}], + } + + +@dataclass +class ResponsesAccumulator: + """Folds a Codex Responses SSE event stream into a chat-style result. + + Feed each decoded ``data:`` event to :meth:`handle`; then read :meth:`result` + for the ``content`` / ``tool_calls`` / ``finish_reason`` / token counts that + ``fusion.py`` expects, in the same shape ``chat/completions`` would yield. + """ + + text_parts: list[str] = field(default_factory=list) + # Keyed by call_id so ``added`` then ``done`` events for one call collapse. + _tool_calls: dict[str, dict[str, Any]] = field(default_factory=dict) + tokens_in: int = 0 + tokens_out: int = 0 + error: Optional[str] = None + + def handle(self, event: dict[str, Any]) -> None: + event_type = event.get("type") + if event_type == "response.output_text.delta": + self.text_parts.append(event.get("delta", "")) + elif event_type == "response.output_item.done": + self._record_item(event.get("item") or {}) + elif event_type == "response.completed": + self._finalize(event.get("response") or {}) + elif event_type in ("response.failed", "error"): + self.error = json.dumps(event)[:500] + + def _record_item(self, item: dict[str, Any]) -> None: + if item.get("type") != "function_call": + return + call_id = item.get("call_id") or item.get("id") or str(len(self._tool_calls)) + self._tool_calls[call_id] = { + "id": item.get("call_id") or item.get("id"), + "type": "function", + "function": { + "name": item.get("name"), + "arguments": item.get("arguments", ""), + }, + } + + def _finalize(self, response: dict[str, Any]) -> None: + usage = response.get("usage") or {} + self.tokens_in = int(usage.get("input_tokens", 0) or 0) + self.tokens_out = int(usage.get("output_tokens", 0) or 0) + # If we never saw streaming text deltas, recover text from the final + # output items (a non-streaming-looking completion still lands here). + if not self.text_parts: + for item in response.get("output", []): + if item.get("type") == "message": + for part in item.get("content", []): + if part.get("type") == "output_text": + self.text_parts.append(part.get("text", "")) + elif item.get("type") == "function_call": + self._record_item(item) + + def result(self) -> dict[str, Any]: + tool_calls = list(self._tool_calls.values()) or None + return { + "content": "".join(self.text_parts), + "tool_calls": tool_calls, + "finish_reason": "tool_calls" if tool_calls else "stop", + "tokens_in": self.tokens_in, + "tokens_out": self.tokens_out, + "error": self.error, + } diff --git a/fusion.py b/fusion.py index bf46d8d..b60e412 100644 --- a/fusion.py +++ b/fusion.py @@ -26,15 +26,34 @@ from __future__ import annotations import asyncio +import json import logging import os import time from dataclasses import dataclass from pathlib import Path from typing import Any, Optional +from uuid import uuid4 import httpx +try: # Package import (``fusion_engine``) and bare-module import both work. + from .codex_auth import ( + CODEX_RESPONSES_URL, + CodexAuth, + CodexAuthError, + ResponsesAccumulator, + build_responses_payload, + ) +except ImportError: + from codex_auth import ( + CODEX_RESPONSES_URL, + CodexAuth, + CodexAuthError, + ResponsesAccumulator, + build_responses_payload, + ) + logger = logging.getLogger(__name__) logger.addHandler(logging.NullHandler()) @@ -44,6 +63,10 @@ OPENROUTER_API_URL = "https://openrouter.ai/api/v1/chat/completions" +# Slug prefix for models that Codex OAuth can serve through the ChatGPT +# subscription. When OAuth is enabled, these bypass OpenRouter entirely. +CODEX_MODEL_PREFIX = "openai/" + # Default per-request timeout (seconds). Web-search-enabled calls can be slow, # so this is generous on the read side. DEFAULT_TIMEOUT = httpx.Timeout(connect=10.0, read=180.0, write=30.0, pool=10.0) @@ -234,6 +257,8 @@ def __init__( timeout: httpx.Timeout = DEFAULT_TIMEOUT, http_referer: Optional[str] = None, app_title: Optional[str] = None, + codex_oauth: Optional[bool] = None, + codex_auth_file: Optional[str | Path] = None, ) -> None: """Create an engine. @@ -249,8 +274,26 @@ def __init__( attribution). Falls back to ``OPENROUTER_HTTP_REFERER``. app_title: Optional ``X-Title`` header. Falls back to ``OPENROUTER_APP_TITLE``. + codex_oauth: If True, serve ``openai/*`` models through a ChatGPT + subscription via Codex OAuth instead of OpenRouter (see + :mod:`codex_auth`). Defaults to the ``FUSION_CODEX_OAUTH`` env + var (``1``/``true``/``yes`` enable it). When enabled, an + OpenRouter key is still only required if the panel also contains + non-OpenAI models. + codex_auth_file: Path to the Codex ``auth.json``. Falls back to + ``CODEX_AUTH_FILE`` then ``~/.codex/auth.json``. """ self._api_key = api_key or os.environ.get("OPENROUTER_API_KEY") + if codex_oauth is None: + codex_oauth = os.environ.get("FUSION_CODEX_OAUTH", "").strip().lower() in ( + "1", + "true", + "yes", + ) + self.codex_oauth = codex_oauth + self._codex_auth = ( + CodexAuth(auth_file=codex_auth_file) if codex_oauth else None + ) project_root = Path(__file__).resolve().parent self.judge_template_path = Path( judge_template_path @@ -265,6 +308,25 @@ def __init__( # ----------------------------- helpers -------------------------------- # + def _routes_via_codex(self, model: str) -> bool: + """True if ``model`` should be served via Codex OAuth, not OpenRouter.""" + return self._codex_auth is not None and model.startswith(CODEX_MODEL_PREFIX) + + def _require_openrouter_key(self, models: list[str]) -> None: + """Raise unless every non-Codex model has an OpenRouter key available. + + With Codex OAuth on, a panel of only ``openai/*`` models needs no + OpenRouter key; one is required the moment any other model is involved. + """ + if self._api_key: + return + needs = sorted({m for m in models if not self._routes_via_codex(m)}) + if needs: + raise RuntimeError( + "No OpenRouter API key. Set OPENROUTER_API_KEY or pass api_key " + f"(required for: {', '.join(needs)})." + ) + def _headers(self) -> dict[str, str]: """Build request headers, including optional OpenRouter attribution.""" headers = { @@ -400,7 +462,21 @@ async def _complete( When ``tools`` is given it is forwarded to OpenRouter (OpenAI function calling), and any ``tool_calls`` the model emits are parsed onto the returned :class:`PanelResponse`. + + ``openai/*`` models are transparently routed through Codex OAuth (the + ChatGPT subscription) instead of OpenRouter when that mode is enabled. """ + if self._routes_via_codex(model): + return await self._complete_codex( + client, + model, + messages, + tools=tools, + tool_choice=tool_choice, + max_tokens=max_tokens, + web_search=bool(plugins), + ) + payload: dict[str, object] = {"model": model, "messages": messages} if plugins: payload["plugins"] = plugins @@ -472,6 +548,110 @@ async def _complete( finish_reason=finish_reason, ) + async def _complete_codex( + self, + client: httpx.AsyncClient, + model: str, + messages: list[dict[str, object]], + tools: Optional[list[dict[str, object]]] = None, + tool_choice: Optional[object] = None, + max_tokens: Optional[int] = None, + web_search: bool = False, + ) -> PanelResponse: + """Run one completion via Codex OAuth (ChatGPT) instead of OpenRouter. + + Translates the chat request into a Responses payload, streams the SSE + result, and folds it back into the same :class:`PanelResponse` shape as + :meth:`_complete`. Cost is reported as ``0.0`` because the call is billed + against the ChatGPT subscription, not per token. Like :meth:`_complete`, + this never raises: failures land in ``PanelResponse.error``. + + ``web_search`` is accepted for signature parity but ignored — OpenRouter's + web plugin has no Codex-backend equivalent. + """ + if web_search: + logger.debug("Web search is not supported on the Codex path; ignoring for %s", model) + + start = time.perf_counter() + # Strip the ``openai/`` routing prefix; the Codex backend wants the bare + # model name (e.g. ``gpt-5.5``). + backend_model = model[len(CODEX_MODEL_PREFIX):] + try: + access_token, account_id = await self._codex_auth.get_auth(client) + except CodexAuthError as exc: + latency_ms = (time.perf_counter() - start) * 1000 + logger.error("Codex auth failed for %s: %s", model, exc) + return PanelResponse(model=model, content="", latency_ms=latency_ms, error=str(exc)) + + payload = build_responses_payload( + backend_model, messages, tools=tools, tool_choice=tool_choice, + max_tokens=max_tokens, stream=True, + ) + headers = { + "Authorization": f"Bearer {access_token}", + "Content-Type": "application/json", + "Accept": "text/event-stream", + "OpenAI-Beta": "responses=experimental", + "originator": "codex_cli_rs", + "session_id": str(uuid4()), + } + if account_id: + headers["chatgpt-account-id"] = account_id + + acc = ResponsesAccumulator() + try: + async with client.stream( + "POST", CODEX_RESPONSES_URL, headers=headers, json=payload, + timeout=self.timeout, + ) as resp: + if resp.status_code >= 400: + body = (await resp.aread()).decode(errors="replace")[:500] + latency_ms = (time.perf_counter() - start) * 1000 + msg = f"HTTP {resp.status_code}: {body}" + logger.error("Codex model %s failed (%.0f ms): %s", model, latency_ms, msg) + return PanelResponse(model=model, content="", latency_ms=latency_ms, error=msg) + async for line in resp.aiter_lines(): + if not line.startswith("data:"): + continue + data = line[len("data:"):].strip() + if not data or data == "[DONE]": + continue + try: + acc.handle(json.loads(data)) + except ValueError: + # Skip keep-alive comments / malformed event lines. + continue + except httpx.HTTPError as exc: + latency_ms = (time.perf_counter() - start) * 1000 + msg = f"{type(exc).__name__}: {exc}" + logger.error("Codex model %s failed (%.0f ms): %s", model, latency_ms, msg) + return PanelResponse(model=model, content="", latency_ms=latency_ms, error=msg) + + latency_ms = (time.perf_counter() - start) * 1000 + out = acc.result() + if out["error"]: + logger.error("Codex model %s stream error: %s", model, out["error"]) + return PanelResponse(model=model, content="", latency_ms=latency_ms, error=out["error"]) + + logger.info( + "Codex model %s ok: %d in / %d out tokens, %.0f ms, $0 (subscription)%s", + model, + out["tokens_in"], + out["tokens_out"], + latency_ms, + f", {len(out['tool_calls'])} tool call(s)" if out["tool_calls"] else "", + ) + return PanelResponse( + model=model, + content=out["content"], + tokens_in=out["tokens_in"], + tokens_out=out["tokens_out"], + latency_ms=latency_ms, + cost_usd=0.0, + tool_calls=out["tool_calls"], + finish_reason=out["finish_reason"], + ) + async def _dispatch_panel_member( self, client: httpx.AsyncClient, @@ -549,13 +729,10 @@ async def fuse( RuntimeError: If no OpenRouter API key is available. ValueError: If ``panel`` is empty. """ - if not self._api_key: - raise RuntimeError( - "No OpenRouter API key. Set OPENROUTER_API_KEY or pass api_key." - ) if not panel: raise ValueError("panel must contain at least one model slug") panel_specs = [self._model_spec(member) for member in panel] + self._require_openrouter_key([s for s, _ in panel_specs] + [judge_model]) run_start = time.perf_counter() logger.info( @@ -661,13 +838,10 @@ async def fuse_chat( RuntimeError: If no OpenRouter API key is available. ValueError: If ``panel`` is empty. """ - if not self._api_key: - raise RuntimeError( - "No OpenRouter API key. Set OPENROUTER_API_KEY or pass api_key." - ) if not panel: raise ValueError("panel must contain at least one model slug") panel_specs = [self._model_spec(member) for member in panel] + self._require_openrouter_key([s for s, _ in panel_specs] + [judge_model]) run_start = time.perf_counter() plugins = [{"id": "web"}] if web_search else None @@ -763,10 +937,7 @@ async def complete_one( Raises: RuntimeError: If no OpenRouter API key is available. """ - if not self._api_key: - raise RuntimeError( - "No OpenRouter API key. Set OPENROUTER_API_KEY or pass api_key." - ) + self._require_openrouter_key([model]) plugins = [{"id": "web"}] if web_search else None messages = [{"role": "user", "content": prompt}] async with httpx.AsyncClient(timeout=self.timeout) as client: diff --git a/pyproject.toml b/pyproject.toml index 4aa8c59..88bde8a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -38,4 +38,4 @@ fusion = "cli:main" fusion-engine = "cli:main" [tool.setuptools] -py-modules = ["cli", "fusion", "panels", "server"] +py-modules = ["cli", "codex_auth", "fusion", "panels", "server"] diff --git a/tests/test_codex_auth.py b/tests/test_codex_auth.py new file mode 100644 index 0000000..8caf292 --- /dev/null +++ b/tests/test_codex_auth.py @@ -0,0 +1,204 @@ +"""Tests for the Codex OAuth transport — translation and routing, no network.""" + +from __future__ import annotations + +import base64 +import json + +import pytest + +import codex_auth +from codex_auth import ( + CodexAuth, + ResponsesAccumulator, + build_responses_payload, + decode_jwt_claims, +) +from fusion import FusionEngine + + +def _fake_jwt(claims: dict) -> str: + """Build an unsigned JWT-shaped token carrying ``claims`` in its payload.""" + payload = base64.urlsafe_b64encode(json.dumps(claims).encode()).rstrip(b"=") + return f"header.{payload.decode()}.sig" + + +# --------------------------- JWT claim decoding --------------------------- # + +def test_decode_jwt_claims_reads_payload() -> None: + token = _fake_jwt({"exp": 123, "sub": "abc"}) + assert decode_jwt_claims(token) == {"exp": 123, "sub": "abc"} + + +def test_decode_jwt_claims_handles_garbage() -> None: + assert decode_jwt_claims("not-a-jwt") == {} + assert decode_jwt_claims("") == {} + + +# ------------------------- chat -> Responses payload ---------------------- # + +def test_build_responses_payload_maps_system_and_user() -> None: + payload = build_responses_payload( + "gpt-5.5", + [ + {"role": "system", "content": "Be terse."}, + {"role": "user", "content": "Hello"}, + ], + ) + assert payload["model"] == "gpt-5.5" + assert payload["instructions"] == "Be terse." + assert payload["stream"] is True + assert payload["input"] == [ + { + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": "Hello"}], + } + ] + + +def test_build_responses_payload_maps_tool_roundtrip() -> None: + payload = build_responses_payload( + "gpt-5.5", + [ + {"role": "user", "content": "weather?"}, + { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": "call_1", + "function": {"name": "get_weather", "arguments": '{"city":"NYC"}'}, + } + ], + }, + {"role": "tool", "tool_call_id": "call_1", "content": "72F"}, + ], + tools=[ + { + "type": "function", + "function": { + "name": "get_weather", + "description": "look up weather", + "parameters": {"type": "object"}, + }, + } + ], + ) + # Function tool is flattened into the Responses shape. + assert payload["tools"] == [ + { + "type": "function", + "name": "get_weather", + "description": "look up weather", + "parameters": {"type": "object"}, + } + ] + # The assistant call and the tool result both survive as input items. + assert {"type": "function_call", "call_id": "call_1", "name": "get_weather", + "arguments": '{"city":"NYC"}'} in payload["input"] + assert {"type": "function_call_output", "call_id": "call_1", + "output": "72F"} in payload["input"] + + +def test_build_responses_payload_max_tokens() -> None: + payload = build_responses_payload( + "gpt-5.5", [{"role": "user", "content": "hi"}], max_tokens=256 + ) + assert payload["max_output_tokens"] == 256 + + +# ---------------------------- SSE accumulation ---------------------------- # + +def test_accumulator_collects_text_and_usage() -> None: + acc = ResponsesAccumulator() + acc.handle({"type": "response.output_text.delta", "delta": "Hel"}) + acc.handle({"type": "response.output_text.delta", "delta": "lo"}) + acc.handle({"type": "response.completed", "response": { + "usage": {"input_tokens": 10, "output_tokens": 5}}}) + result = acc.result() + assert result["content"] == "Hello" + assert result["tool_calls"] is None + assert result["finish_reason"] == "stop" + assert result["tokens_in"] == 10 + assert result["tokens_out"] == 5 + + +def test_accumulator_collects_tool_calls() -> None: + acc = ResponsesAccumulator() + acc.handle({ + "type": "response.output_item.done", + "item": { + "type": "function_call", + "call_id": "call_9", + "name": "search", + "arguments": '{"q":"x"}', + }, + }) + acc.handle({"type": "response.completed", "response": {"usage": {}}}) + result = acc.result() + assert result["finish_reason"] == "tool_calls" + assert result["tool_calls"] == [ + {"id": "call_9", "type": "function", + "function": {"name": "search", "arguments": '{"q":"x"}'}} + ] + + +def test_accumulator_recovers_text_without_deltas() -> None: + acc = ResponsesAccumulator() + acc.handle({"type": "response.completed", "response": { + "usage": {}, + "output": [ + {"type": "message", "content": [ + {"type": "output_text", "text": "final answer"}]} + ], + }}) + assert acc.result()["content"] == "final answer" + + +# ------------------------------- routing ---------------------------------- # + +def test_codex_disabled_by_default() -> None: + engine = FusionEngine(api_key="test", codex_oauth=False) + assert engine._routes_via_codex("openai/gpt-5.5") is False + + +def test_codex_routes_only_openai_slugs() -> None: + engine = FusionEngine(api_key="test", codex_oauth=True) + assert engine._routes_via_codex("openai/gpt-5.5") is True + assert engine._routes_via_codex("anthropic/claude-opus-4") is False + + +def test_openrouter_key_not_required_for_all_codex_panel() -> None: + engine = FusionEngine(api_key=None, codex_oauth=True) + # All-OpenAI panel + judge: no OpenRouter key needed. + engine._require_openrouter_key(["openai/gpt-5.5", "openai/codex"]) + # A non-OpenAI model reintroduces the requirement. + with pytest.raises(RuntimeError, match="anthropic/claude-opus-4"): + engine._require_openrouter_key(["openai/gpt-5.5", "anthropic/claude-opus-4"]) + + +def test_env_var_enables_codex(monkeypatch) -> None: + monkeypatch.setenv("FUSION_CODEX_OAUTH", "true") + engine = FusionEngine(api_key="test") + assert engine.codex_oauth is True + assert engine._routes_via_codex("openai/gpt-5.5") is True + + +# ---------------------------- token expiry -------------------------------- # + +def test_expired_detects_past_exp() -> None: + auth = CodexAuth(auth_file="/nonexistent") + assert auth._expired(_fake_jwt({"exp": 0})) is True + assert auth._expired(_fake_jwt({"exp": 9999999999})) is False + # No exp claim -> treat as valid, let the server decide. + assert auth._expired(_fake_jwt({"sub": "x"})) is False + + +def test_account_id_from_id_token() -> None: + id_token = _fake_jwt( + {"https://api.openai.com/auth": {"chatgpt_account_id": "acct_42"}} + ) + assert CodexAuth._account_id({"id_token": id_token}) == "acct_42" + # Explicit account_id wins. + assert CodexAuth._account_id({"account_id": "acct_1", "id_token": id_token}) == "acct_1"