diff --git a/README.md b/README.md index 81686b4f..f184e263 100644 --- a/README.md +++ b/README.md @@ -233,6 +233,7 @@ clawcodex # Interactive Ink TUI (default) clawcodex tui # Interactive Ink TUI (explicit) clawcodex login # Configure API keys (interactive) clawcodex logout anthropic # Remove Claude Pro/Max OAuth credentials +clawcodex logout openai # Remove ChatGPT subscription OAuth credentials clawcodex config # Show ~/.clawcodex/config.json-backed settings clawcodex --version # Version string @@ -268,6 +269,28 @@ An explicit `ANTHROPIC_API_KEY` or configured Anthropic API key takes precedence Use `clawcodex config` to check connection status and `clawcodex logout anthropic` to delete the stored OAuth credentials. +### ChatGPT subscription login (OpenAI) + +Run `clawcodex login`, choose `openai`, then choose `subscription`. Three +login methods are offered: `browser` (opens ChatGPT's PKCE authorization page +and completes via a localhost:1455 redirect — the same flow the Codex CLI +uses), `device-code` (for headless/SSH machines: enter a short code at +auth.openai.com/codex/device), and `import-codex-cli` (shown when a Codex CLI +ChatGPT login exists in `~/.codex/auth.json` — copies it). Tokens are stored +in `~/.clawcodex/openai-oauth.json` with user-only permissions and refresh +automatically. With a subscription connected and no OpenAI API key set, +requests go to the ChatGPT Codex backend (Responses API) and draw on your +plan's allowance — the cost display shows $0. An explicit `OPENAI_API_KEY` or +configured OpenAI API key takes precedence. Subscription models: +`gpt-5.5`, `gpt-5.4`, `gpt-5.4-mini`, `gpt-5.3-codex-spark`. + +> **Important:** using a ChatGPT plan from third-party clients rides the +> Codex CLI's official OAuth app, remains subject to OpenAI's terms, and may +> stop working. The login flow requires explicit confirmation. + +Use `clawcodex config` to check connection status and +`clawcodex logout openai` to delete the stored OAuth credentials. + > **`--dangerously-skip-permissions`** disables every tool permission check > for the session. Recommended only inside sandboxed containers/VMs with no > internet access. The flag is refused when the process is running as diff --git a/src/auth/openai_subscription.py b/src/auth/openai_subscription.py new file mode 100644 index 00000000..2bc7c838 --- /dev/null +++ b/src/auth/openai_subscription.py @@ -0,0 +1,470 @@ +"""ChatGPT Plus/Pro OAuth credentials for the OpenAI provider. + +This implements the "Sign in with ChatGPT" flow the Codex CLI uses +(authorization-code + PKCE against ``auth.openai.com`` with a localhost +callback), as reimplemented by OpenCode's ``openai`` plugin +(``reference_projects/opencode/packages/opencode/src/plugin/openai/codex.ts``). +Requests then go to the ChatGPT Codex backend instead of the metered +platform API — see ``src/providers/openai_responses.py``. + +Subscription use from third-party clients rides Codex CLI's official OAuth +app; it is subject to OpenAI's terms and may stop working. Callers must +surface that before starting login. +""" + +from __future__ import annotations + +import base64 +import hashlib +import http.server +import json +import os +import secrets +import threading +import time +import urllib.error +import urllib.parse +import urllib.request +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any + +CLIENT_ID = "app_EMoamEEZ73f0CkXaXp7hrann" +ISSUER = "https://auth.openai.com" +AUTHORIZE_URL = f"{ISSUER}/oauth/authorize" +TOKEN_URL = f"{ISSUER}/oauth/token" +DEVICE_CODE_URL = f"{ISSUER}/api/accounts/deviceauth/usercode" +DEVICE_TOKEN_URL = f"{ISSUER}/api/accounts/deviceauth/token" +DEVICE_VERIFY_URL = f"{ISSUER}/codex/device" +CALLBACK_PORT = 1455 +REDIRECT_URI = f"http://localhost:{CALLBACK_PORT}/auth/callback" +SCOPES = "openid profile email offline_access" +# Wire identity: match Codex CLI exactly (OpenCode ships its own name and +# works, but codex_cli_rs is the value the backend is guaranteed to accept). +ORIGINATOR = "codex_cli_rs" +CODEX_API_ENDPOINT = "https://chatgpt.com/backend-api/codex/responses" + +_refresh_lock = threading.Lock() + + +def credentials_path() -> Path: + root = Path(os.environ.get("CLAWCODEX_CONFIG_DIR", Path.home() / ".clawcodex")) + return root / "openai-oauth.json" + + +@dataclass +class SubscriptionCredentials: + access_token: str + refresh_token: str + expires_at: float + account_id: str = "" + id_token: str = "" + + @property + def needs_refresh(self) -> bool: + return self.expires_at <= time.time() + 60 + + +def load_credentials() -> SubscriptionCredentials | None: + path = credentials_path() + try: + value = json.loads(path.read_text(encoding="utf-8")) + return SubscriptionCredentials( + access_token=str(value["access_token"]), + refresh_token=str(value["refresh_token"]), + expires_at=float(value["expires_at"]), + account_id=str(value.get("account_id", "")), + id_token=str(value.get("id_token", "")), + ) + except (OSError, ValueError, KeyError, TypeError, json.JSONDecodeError): + return None + + +def save_credentials(credentials: SubscriptionCredentials) -> None: + path = credentials_path() + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(f".tmp-{os.getpid()}-{secrets.token_hex(4)}") + try: + fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as stream: + json.dump(asdict(credentials), stream, indent=2) + stream.write("\n") + os.replace(tmp, path) + if os.name != "nt": + os.chmod(path, 0o600) + finally: + try: + tmp.unlink() + except OSError: + pass + + +def remove_credentials() -> bool: + try: + credentials_path().unlink() + return True + except FileNotFoundError: + return False + + +# --- token endpoint --------------------------------------------------------- + + +def _post_form(url: str, payload: dict[str, str]) -> dict[str, Any]: + """POST form-urlencoded (the OpenAI token endpoint rejects JSON bodies).""" + request = urllib.request.Request( + url, + data=urllib.parse.urlencode(payload).encode("utf-8"), + headers={"Content-Type": "application/x-www-form-urlencoded"}, + method="POST", + ) + try: + with urllib.request.urlopen(request, timeout=30) as response: + result = json.loads(response.read().decode("utf-8")) + except urllib.error.HTTPError as exc: + detail = exc.read().decode("utf-8", "replace") + raise RuntimeError(f"ChatGPT OAuth request failed ({exc.code}): {detail}") from exc + if not isinstance(result, dict): + raise RuntimeError("ChatGPT OAuth endpoint returned an invalid response") + return result + + +def _post_json(url: str, payload: dict[str, Any]) -> tuple[int, dict[str, Any]]: + """POST JSON, returning ``(status, body)`` — device-auth endpoints signal + "keep polling" via 403/404, so HTTP errors are data here, not exceptions.""" + request = urllib.request.Request( + url, + data=json.dumps(payload).encode("utf-8"), + headers={"Content-Type": "application/json"}, + method="POST", + ) + try: + with urllib.request.urlopen(request, timeout=30) as response: + body = json.loads(response.read().decode("utf-8")) + return response.status, body if isinstance(body, dict) else {} + except urllib.error.HTTPError as exc: + try: + body = json.loads(exc.read().decode("utf-8", "replace")) + except (ValueError, json.JSONDecodeError): + body = {} + return exc.code, body if isinstance(body, dict) else {} + + +# --- JWT claims (unverified — client-side routing hint only) ---------------- + + +def parse_jwt_claims(token: str) -> dict[str, Any]: + parts = token.split(".") + if len(parts) != 3: + return {} + payload = parts[1] + "=" * (-len(parts[1]) % 4) + try: + claims = json.loads(base64.urlsafe_b64decode(payload)) + except (ValueError, json.JSONDecodeError): + return {} + return claims if isinstance(claims, dict) else {} + + +def _account_id_from_claims(claims: dict[str, Any]) -> str: + direct = claims.get("chatgpt_account_id") + if isinstance(direct, str) and direct: + return direct + auth_claim = claims.get("https://api.openai.com/auth") + if isinstance(auth_claim, dict): + nested = auth_claim.get("chatgpt_account_id") + if isinstance(nested, str) and nested: + return nested + organizations = claims.get("organizations") + if isinstance(organizations, list) and organizations: + first = organizations[0] + if isinstance(first, dict) and isinstance(first.get("id"), str): + return first["id"] + return "" + + +def extract_account_id(id_token: str, access_token: str = "") -> str: + """ChatGPT account id, preferring the id_token's claims (OpenCode order).""" + for token in (id_token, access_token): + if token: + account_id = _account_id_from_claims(parse_jwt_claims(token)) + if account_id: + return account_id + return "" + + +# --- authorization-code + PKCE flow ------------------------------------------ + + +def begin_login() -> tuple[str, str, str]: + """Build the authorize URL. Returns ``(url, verifier, state)``.""" + verifier = secrets.token_urlsafe(64)[:128] + digest = hashlib.sha256(verifier.encode("ascii")).digest() + challenge = base64.urlsafe_b64encode(digest).rstrip(b"=").decode("ascii") + state = secrets.token_urlsafe(32) + query = urllib.parse.urlencode({ + "response_type": "code", + "client_id": CLIENT_ID, + "redirect_uri": REDIRECT_URI, + "scope": SCOPES, + "code_challenge": challenge, + "code_challenge_method": "S256", + # Codex-flow switches (mirrors OpenCode's authorize URL verbatim). + "id_token_add_organizations": "true", + "codex_cli_simplified_flow": "true", + "state": state, + "originator": ORIGINATOR, + }) + return f"{AUTHORIZE_URL}?{query}", verifier, state + + +_SUCCESS_PAGE = ( + "ClawCodex" + "

Authorization successful

You can close this window and return " + "to ClawCodex.

" +) + + +def _error_page(message: str) -> str: + safe = ( + message.replace("&", "&").replace("<", "<").replace(">", ">") + ) + return ( + "ClawCodex" + f"

Authorization failed

{safe}

" + ) + + +def wait_for_callback(state: str, timeout: float = 300.0) -> str: + """Serve ``http://localhost:1455/auth/callback`` until the browser + redirect delivers the authorization code (or ``timeout`` expires). + + Binds to localhost only. A ``state`` mismatch is rejected (CSRF guard, + same as OpenCode's server). Raises ``RuntimeError`` on OAuth errors, + timeout, or if the port is taken (typically a concurrent Codex CLI + login — the caller should suggest the device-code flow). + """ + result: dict[str, str] = {} + done = threading.Event() + + class Handler(http.server.BaseHTTPRequestHandler): + def do_GET(self) -> None: # noqa: N802 — BaseHTTPRequestHandler API + parsed = urllib.parse.urlparse(self.path) + if parsed.path != "/auth/callback": + self.send_response(404) + self.end_headers() + self.wfile.write(b"Not found") + return + params = urllib.parse.parse_qs(parsed.query) + error = (params.get("error_description") or params.get("error") or [""])[0] + code = (params.get("code") or [""])[0] + returned_state = (params.get("state") or [""])[0] + if error: + result["error"] = error + elif not code: + result["error"] = "Missing authorization code" + elif returned_state != state: + result["error"] = "Invalid OAuth state (possible CSRF)" + else: + result["code"] = code + failed = "error" in result + self.send_response(400 if failed else 200) + self.send_header("Content-Type", "text/html; charset=utf-8") + self.end_headers() + page = _error_page(result["error"]) if failed else _SUCCESS_PAGE + self.wfile.write(page.encode("utf-8")) + done.set() + + def log_message(self, *args: Any) -> None: # silence request logging + return + + try: + server = http.server.HTTPServer(("127.0.0.1", CALLBACK_PORT), Handler) + except OSError as exc: + raise RuntimeError( + f"Could not listen on localhost:{CALLBACK_PORT} ({exc}). " + "Close the application using that port (often another CLI's login " + "flow) or use the device-code login instead." + ) from exc + server.timeout = 1.0 + try: + deadline = time.time() + timeout + while not done.is_set(): + if time.time() > deadline: + raise RuntimeError("Timed out waiting for the browser authorization") + server.handle_request() + finally: + server.server_close() + if "error" in result: + raise RuntimeError(f"ChatGPT authorization failed: {result['error']}") + return result["code"] + + +def complete_login(code: str, verifier: str) -> SubscriptionCredentials: + result = _post_form(TOKEN_URL, { + "grant_type": "authorization_code", + "code": code.strip(), + "redirect_uri": REDIRECT_URI, + "client_id": CLIENT_ID, + "code_verifier": verifier, + }) + credentials = _credentials_from_response(result) + save_credentials(credentials) + return credentials + + +# --- device-code flow (headless boxes; OpenCode's second method) ------------- + + +def begin_device_login() -> dict[str, str]: + """Start device authorization. Returns ``{device_auth_id, user_code, + interval, verify_url}`` — show ``user_code`` and ``verify_url`` to the user.""" + status, body = _post_json(DEVICE_CODE_URL, {"client_id": CLIENT_ID}) + if status != 200 or "user_code" not in body: + raise RuntimeError(f"Failed to initiate device authorization ({status})") + return { + "device_auth_id": str(body.get("device_auth_id", "")), + "user_code": str(body.get("user_code", "")), + "interval": str(body.get("interval", "5")), + "verify_url": DEVICE_VERIFY_URL, + } + + +def poll_device_login( + device: dict[str, str], timeout: float = 900.0 +) -> SubscriptionCredentials: + """Poll until the user enters the code, then exchange and persist tokens.""" + try: + interval = max(int(device.get("interval", "5") or "5"), 1) + except ValueError: + interval = 5 + deadline = time.time() + timeout + while time.time() < deadline: + status, body = _post_json(DEVICE_TOKEN_URL, { + "device_auth_id": device["device_auth_id"], + "user_code": device["user_code"], + }) + if status == 200 and body.get("authorization_code"): + result = _post_form(TOKEN_URL, { + "grant_type": "authorization_code", + "code": str(body["authorization_code"]), + "redirect_uri": f"{ISSUER}/deviceauth/callback", + "client_id": CLIENT_ID, + "code_verifier": str(body.get("code_verifier", "")), + }) + credentials = _credentials_from_response(result) + save_credentials(credentials) + return credentials + if status not in (403, 404): + raise RuntimeError(f"Device authorization failed ({status})") + # 403/404 = "not approved yet" — keep polling (+3s safety margin, + # matching OpenCode's OAUTH_POLLING_SAFETY_MARGIN_MS). + time.sleep(interval + 3) + raise RuntimeError("Timed out waiting for device authorization") + + +# --- import from Codex CLI ---------------------------------------------------- + + +def codex_cli_auth_path() -> Path: + return Path(os.environ.get("CODEX_HOME", Path.home() / ".codex")) / "auth.json" + + +def has_codex_cli_credentials() -> bool: + try: + value = json.loads(codex_cli_auth_path().read_text(encoding="utf-8")) + except (OSError, ValueError, json.JSONDecodeError): + return False + tokens = value.get("tokens") if isinstance(value, dict) else None + return bool( + isinstance(tokens, dict) + and tokens.get("access_token") + and tokens.get("refresh_token") + ) + + +def import_codex_cli_credentials() -> SubscriptionCredentials: + """Copy an existing Codex CLI ChatGPT login into the clawcodex store. + + The file has no expiry field, so it is derived from the access token's + ``exp`` claim. The copies then refresh independently; if OpenAI ever + rotates refresh tokens exclusively, the Codex CLI copy may need a + re-login — the CLI warns about this at import time. + """ + path = codex_cli_auth_path() + value = json.loads(path.read_text(encoding="utf-8")) + tokens = value.get("tokens") or {} + access = str(tokens.get("access_token") or "") + refresh = str(tokens.get("refresh_token") or "") + if not access or not refresh: + raise RuntimeError(f"No ChatGPT tokens found in {path}") + id_token = str(tokens.get("id_token") or "") + claims = parse_jwt_claims(access) + expires_at = float(claims.get("exp") or 0.0) + if expires_at <= 0: + # Unknown expiry — treat as already stale so first use refreshes. + expires_at = time.time() - 1 + account_id = str(tokens.get("account_id") or "") or extract_account_id( + id_token, access + ) + credentials = SubscriptionCredentials( + access_token=access, + refresh_token=refresh, + expires_at=expires_at, + account_id=account_id, + id_token=id_token, + ) + save_credentials(credentials) + return credentials + + +# --- refresh ------------------------------------------------------------------ + + +def _credentials_from_response( + result: dict[str, Any], *, old: SubscriptionCredentials | None = None +) -> SubscriptionCredentials: + try: + access = str(result["access_token"]) + refresh = str(result.get("refresh_token") or (old.refresh_token if old else "")) + expires = time.time() + float(result.get("expires_in") or 3600) + except (KeyError, TypeError, ValueError) as exc: + raise RuntimeError("ChatGPT OAuth response omitted required token fields") from exc + if not access or not refresh: + raise RuntimeError("ChatGPT OAuth response omitted required token fields") + id_token = str(result.get("id_token") or (old.id_token if old else "")) + account_id = extract_account_id(id_token, access) or (old.account_id if old else "") + return SubscriptionCredentials(access, refresh, expires, account_id, id_token) + + +def get_valid_credentials() -> SubscriptionCredentials | None: + credentials = load_credentials() + if credentials is None or not credentials.needs_refresh: + return credentials + with _refresh_lock: + credentials = load_credentials() + if credentials is None or not credentials.needs_refresh: + return credentials + result = _post_form(TOKEN_URL, { + "grant_type": "refresh_token", + "refresh_token": credentials.refresh_token, + "client_id": CLIENT_ID, + }) + refreshed = _credentials_from_response(result, old=credentials) + save_credentials(refreshed) + return refreshed + + +def force_refresh() -> SubscriptionCredentials | None: + """Refresh regardless of local expiry (reactive 401 handling).""" + with _refresh_lock: + credentials = load_credentials() + if credentials is None: + return None + result = _post_form(TOKEN_URL, { + "grant_type": "refresh_token", + "refresh_token": credentials.refresh_token, + "client_id": CLIENT_ID, + }) + refreshed = _credentials_from_response(result, old=credentials) + save_credentials(refreshed) + return refreshed diff --git a/src/cli.py b/src/cli.py index 9c3d5c95..6a4e029e 100644 --- a/src/cli.py +++ b/src/cli.py @@ -725,6 +725,15 @@ def handle_login(): if auth_method == "subscription": return _handle_anthropic_subscription_login(console, Prompt) + if provider == "openai": + auth_method = Prompt.ask( + "Authentication method", + choices=["subscription", "api-key"], + default="subscription", + ) + if auth_method == "subscription": + return _handle_openai_subscription_login(console, Prompt) + api_key = Prompt.ask( f"Enter {provider.upper()} API Key", password=True @@ -794,20 +803,99 @@ def _handle_anthropic_subscription_login(console, Prompt): return 0 +def _handle_openai_subscription_login(console, Prompt): + """Run the ChatGPT "Sign in with ChatGPT" OAuth flow (Codex backend).""" + import webbrowser + + from src.auth import openai_subscription as oai + from src.config import set_api_key, set_default_provider + from src.providers import PROVIDER_INFO + + console.print( + "\n[yellow]Important:[/yellow] this signs in with your ChatGPT " + "Plus/Pro plan via the same OAuth flow the Codex CLI uses. Usage " + "draws on your plan's allowance, is subject to OpenAI's terms, " + "and may stop working for third-party clients." + ) + consent = Prompt.ask("Continue?", choices=["yes", "no"], default="no") + if consent != "yes": + console.print("Login cancelled.") + return 1 + + methods = ["browser", "device-code"] + if oai.has_codex_cli_credentials(): + methods.append("import-codex-cli") + method = Prompt.ask("Login method", choices=methods, default=methods[0]) + + try: + if method == "import-codex-cli": + console.print( + "[dim]Copying the ChatGPT login stored by Codex CLI " + f"({oai.codex_cli_auth_path()}). Both clients refresh " + "independently; if Codex CLI later asks you to sign in " + "again, that's why.[/dim]" + ) + oai.import_codex_cli_credentials() + elif method == "device-code": + device = oai.begin_device_login() + console.print( + f"\nOpen [bold]{device['verify_url']}[/bold] on any device " + f"and enter code: [bold]{device['user_code']}[/bold]" + ) + console.print("Waiting for authorization...") + oai.poll_device_login(device) + else: + url, verifier, state = oai.begin_login() + console.print("\nOpening ChatGPT authorization in your browser...") + if not webbrowser.open(url): + console.print(f"Open this URL manually:\n{url}") + console.print("Waiting for the browser redirect on localhost:1455...") + code = oai.wait_for_callback(state) + oai.complete_login(code, verifier) + except Exception as exc: + console.print(f"\n[red]ChatGPT subscription login failed:[/red] {exc}") + return 1 + + from src.providers.openai_responses import SUBSCRIPTION_MODELS + + info = PROVIDER_INFO["openai"] + set_api_key( + "openai", api_key="", base_url=info["default_base_url"], + default_model=info["default_model"], + ) + set_default_provider("openai") + console.print("\n[green]✓ ChatGPT subscription connected.[/green]") + console.print( + "[dim]Subscription models: " + ", ".join(SUBSCRIPTION_MODELS) + "[/dim]" + ) + import os as _os + if (_os.environ.get("OPENAI_API_KEY") or "").strip(): + console.print( + "[yellow]Note:[/yellow] OPENAI_API_KEY is set in your environment " + "and takes precedence — unset it to use the subscription." + ) + return 0 + + def handle_logout(args: list[str]) -> int: """Remove locally stored provider OAuth credentials.""" from rich.console import Console console = Console() provider = args[0] if args else "anthropic" - if provider != "anthropic": + if provider == "anthropic": + from src.auth.anthropic_subscription import remove_credentials + label = "Claude Pro/Max" + elif provider == "openai": + from src.auth.openai_subscription import remove_credentials + label = "ChatGPT" + else: console.print(f"[red]OAuth logout is not supported for provider '{provider}'.[/red]") return 2 - from src.auth.anthropic_subscription import remove_credentials if remove_credentials(): - console.print("[green]✓ Claude Pro/Max subscription disconnected.[/green]") + console.print(f"[green]✓ {label} subscription disconnected.[/green]") else: - console.print("No Claude Pro/Max subscription login was stored.") + console.print(f"No {label} subscription login was stored.") return 0 @@ -843,6 +931,13 @@ def show_config(): "\n[cyan]Claude Pro/Max:[/cyan] " + ("Connected" if load_credentials() is not None else "Not connected") ) + from src.auth.openai_subscription import ( + load_credentials as load_openai_credentials, + ) + console.print( + "[cyan]ChatGPT subscription:[/cyan] " + + ("Connected" if load_openai_credentials() is not None else "Not connected") + ) # Stored API keys (config "env" block, e.g. TAVILY_API_KEY). Values are # masked — only the name and a hint are shown. diff --git a/src/entrypoints/provider_validation.py b/src/entrypoints/provider_validation.py index 568c97fb..77b728a3 100644 --- a/src/entrypoints/provider_validation.py +++ b/src/entrypoints/provider_validation.py @@ -38,7 +38,7 @@ def get_provider_validation_error(provider_name: str | None) -> str | None: paths use (``options.provider_name or get_default_provider()``). """ from src.config import get_default_provider, get_provider_config - from src.providers import provider_requires_api_key, resolve_api_key + from src.providers import provider_has_credentials, resolve_api_key name = provider_name or get_default_provider() try: @@ -48,13 +48,11 @@ def get_provider_validation_error(provider_name: str | None) -> str | None: # Config api_key wins; fall back to the provider's known env vars (e.g. # ``DEEPSEEK_API_KEY``). Local providers (Ollama / vLLM / SGLang) need - # no key. Same check the headless path ran inline pre-ENTRY-2. + # no key; anthropic/openai subscription OAuth logins also pass. Same + # check the headless path ran inline pre-ENTRY-2; the shared helper + # keeps this gate consistent with the agent-server's two gates. api_key = resolve_api_key(name, provider_cfg) - if name == "anthropic" and not api_key: - from src.auth.anthropic_subscription import load_credentials - if load_credentials() is not None: - return None - if not api_key and provider_requires_api_key(name): + if not provider_has_credentials(name, api_key): return ( f"error: API key for provider '{name}' is not configured. " "Run `clawcodex login` to set it up." diff --git a/src/models/configs.py b/src/models/configs.py index 3eb47dd6..3a9f1edb 100644 --- a/src/models/configs.py +++ b/src/models/configs.py @@ -232,6 +232,74 @@ class ModelConfig: max_output_tokens=16_384, supports_cache=True, ), + + # OpenAI GPT-5 family. 272K input / 128K output is the GPT-5 window + # (400K total); it matches what the ChatGPT-subscription backend reports + # for gpt-5.5 / gpt-5.4 / gpt-5.4-mini (Codex CLI models cache; OpenCode + # pins the same numbers for gpt-5.5, plugin/openai/codex.ts:387). Until + # now every gpt model silently fell back to DEFAULT_CONTEXT_WINDOW + # (200K), making auto-compact fire ~70K tokens early on these. NOTE on + # the prefix fallback in ``get_model_config``: a gpt id with no exact + # entry resolves to the FIRST gpt entry below (base "gpt"), i.e. 272K — + # the safe direction for compaction (under-estimating compacts early; + # over-estimating overflows); the smaller legacy windows get exact + # entries so they never take that path. As with the Meta entry above, + # max_output_tokens is not sent on the wire for OpenAI providers — its + # live effect is the auto-compact output reservation (clamped at 20K). + "gpt-5.5": ModelConfig( + model_id="gpt-5.5", + display_name="GPT-5.5", + context_window=272_000, + max_output_tokens=128_000, + ), + "gpt-5.4": ModelConfig( + model_id="gpt-5.4", + display_name="GPT-5.4", + context_window=272_000, + max_output_tokens=128_000, + ), + "gpt-5.4-mini": ModelConfig( + model_id="gpt-5.4-mini", + display_name="GPT-5.4 Mini", + context_window=272_000, + max_output_tokens=128_000, + ), + "gpt-5.3-codex-spark": ModelConfig( + model_id="gpt-5.3-codex-spark", + display_name="GPT-5.3 Codex Spark", + context_window=128_000, + max_output_tokens=64_000, + ), + "gpt-4o": ModelConfig( + model_id="gpt-4o", + display_name="GPT-4o", + context_window=128_000, + max_output_tokens=16_384, + ), + "gpt-4o-mini": ModelConfig( + model_id="gpt-4o-mini", + display_name="GPT-4o Mini", + context_window=128_000, + max_output_tokens=16_384, + ), + "gpt-4-turbo": ModelConfig( + model_id="gpt-4-turbo", + display_name="GPT-4 Turbo", + context_window=128_000, + max_output_tokens=4_096, + ), + "gpt-4": ModelConfig( + model_id="gpt-4", + display_name="GPT-4", + context_window=8_192, + max_output_tokens=8_192, + ), + "gpt-3.5-turbo": ModelConfig( + model_id="gpt-3.5-turbo", + display_name="GPT-3.5 Turbo", + context_window=16_385, + max_output_tokens=4_096, + ), } diff --git a/src/providers/__init__.py b/src/providers/__init__.py index 385caa61..3e9811d4 100644 --- a/src/providers/__init__.py +++ b/src/providers/__init__.py @@ -49,7 +49,9 @@ class ProviderInfo(TypedDict): "default_base_url": "https://api.openai.com/v1", "default_model": "gpt-5.4", "available_models": [ - # GPT-5.4 series (latest flagship) + # GPT-5.5 (flagship; also served by the ChatGPT subscription) + "gpt-5.5", + # GPT-5.4 series "gpt-5.4", "gpt-5.4-pro", "gpt-5.4-mini", @@ -61,6 +63,7 @@ class ProviderInfo(TypedDict): "gpt-5.2-nano", # GPT-5.3-Codex (coding-specialized) "gpt-5.3-codex", + "gpt-5.3-codex-spark", # Legacy GPT-4 series "gpt-4o", "gpt-4o-mini", @@ -300,6 +303,32 @@ def provider_requires_api_key(provider_name: str) -> bool: return spec.requires_api_key if spec else True +def provider_has_credentials(provider_name: str, api_key: str) -> bool: + """Whether requests to ``provider_name`` can authenticate. + + True when an API key is present, the provider needs none (local + servers), or the user has a stored subscription OAuth login — Claude + Pro/Max for ``anthropic`` (#697) or a ChatGPT plan for ``openai``. + Every "no API key configured" fatality gate must go through this + helper (startup validation, agent-server session init, agent-server + provider switch) so subscription logins work on all of them — + gating on ``provider_requires_api_key`` alone bricked subscription + sessions in the TUI. + """ + if api_key or not provider_requires_api_key(provider_name): + return True + canonical = _canonical_provider_name(provider_name) + if canonical == "anthropic": + from src.auth.anthropic_subscription import load_credentials + + return load_credentials() is not None + if canonical == "openai": + from src.auth.openai_subscription import load_credentials + + return load_credentials() is not None + return False + + def resolve_api_key( provider_name: str, provider_cfg: dict[str, Any] | None = None ) -> str: @@ -347,6 +376,7 @@ def resolve_api_key( "get_provider_info", "canonical_provider_name", "provider_env_vars", + "provider_has_credentials", "provider_requires_api_key", "resolve_api_key", "PROVIDER_INFO", diff --git a/src/providers/anthropic_provider.py b/src/providers/anthropic_provider.py index ec4648ef..69f2b4bb 100644 --- a/src/providers/anthropic_provider.py +++ b/src/providers/anthropic_provider.py @@ -228,6 +228,18 @@ def _ensure_client(self): self.client = mod.anthropic.Anthropic(**self._client_kwargs) return self.client + def _prepare_messages(self, messages: list[Any]) -> list[dict[str, Any]]: + """Base preparation + removal of foreign passthrough blocks. + + A mid-session ``/model`` switch away from the ChatGPT-subscription + provider leaves ``openai_responses_item`` blocks (encrypted + reasoning replay state) in assistant history; the Anthropic API + rejects unknown content-block types, so they are stripped here. + """ + prepared = super()._prepare_messages(messages) + from .openai_responses import strip_responses_item_blocks + return strip_responses_item_blocks(prepared) + def _effective_base_url(self) -> str | None: """The base URL the SDK will actually use. diff --git a/src/providers/minimax_provider.py b/src/providers/minimax_provider.py index 37a69d98..79a67a20 100644 --- a/src/providers/minimax_provider.py +++ b/src/providers/minimax_provider.py @@ -52,6 +52,19 @@ def _ensure_client(self): self.client = anthropic.Anthropic(**self._client_kwargs) return self.client + def _prepare_messages(self, messages: list[Any]) -> list[dict[str, Any]]: + """Base preparation + removal of foreign passthrough blocks. + + Minimax speaks the Anthropic wire format, which rejects unknown + content-block types — a mid-session ``/model`` switch away from + the ChatGPT-subscription provider must not leak its + ``openai_responses_item`` replay blocks here. Same strip as + ``AnthropicProvider._prepare_messages``. + """ + prepared = super()._prepare_messages(messages) + from .openai_responses import strip_responses_item_blocks + return strip_responses_item_blocks(prepared) + def _build_chat_response(self, response: Any) -> ChatResponse: content_text = "" tool_uses: list[dict[str, Any]] = [] diff --git a/src/providers/openai_compatible.py b/src/providers/openai_compatible.py index 8508a2b0..3bf922d7 100644 --- a/src/providers/openai_compatible.py +++ b/src/providers/openai_compatible.py @@ -166,6 +166,14 @@ def _convert_anthropic_messages_to_openai( ever appears it lands in the closest OpenAI shape rather than passing through as an unrecognised Anthropic block. """ + # ChatGPT-subscription passthrough items (encrypted reasoning et al.) + # have no Chat Completions representation — they exist only for the + # OpenAI Responses replay path. Strip them BEFORE conversion so a + # mid-session model switch never leaks them to a server that rejects + # unknown block types. See openai_responses.RESPONSES_ITEM_BLOCK_TYPE. + from .openai_responses import strip_responses_item_blocks + messages = strip_responses_item_blocks(messages) + result: list[dict[str, Any]] = [] # Pre-scan every assistant message for the tool_use ids it declares. diff --git a/src/providers/openai_provider.py b/src/providers/openai_provider.py index 4479b2cc..eb865e24 100644 --- a/src/providers/openai_provider.py +++ b/src/providers/openai_provider.py @@ -1,19 +1,93 @@ -"""OpenAI provider implementation.""" +"""OpenAI provider implementation. + +Two request paths: + +- **API key** (default): OpenAI SDK against ``/v1/chat/completions`` via + :class:`OpenAICompatibleProvider` — unchanged behaviour. +- **ChatGPT subscription**: when no API key is configured but the user has + connected a ChatGPT Plus/Pro plan (``clawcodex login`` → + ``src/auth/openai_subscription.py``), requests go to the ChatGPT Codex + backend (``https://chatgpt.com/backend-api/codex/responses``) speaking the + Responses API — the mechanism OpenCode's ``openai`` plugin uses + (reference_projects/opencode/packages/opencode/src/plugin/openai/codex.ts). + Wire-format conversion lives in ``src/providers/openai_responses.py``. +""" from __future__ import annotations -from typing import Any, Optional +import logging +import os +import queue +import threading +from typing import Any, Generator, Optional +from urllib.parse import urlparse +from uuid import uuid4 try: from openai import OpenAI # type: ignore except ModuleNotFoundError: # pragma: no cover OpenAI = None -from .openai_compatible import OpenAICompatibleProvider +from .base import BaseProvider, ChatResponse, MessageInput, TextChunkCallback +from .openai_compatible import ( + _CHUNK_QUEUE_MAXSIZE, + OpenAICompatibleProvider, + _parse_tool_call_arguments, +) +from .openai_responses import ( + RESPONSES_ITEM_BLOCK_TYPE, + INCLUDE_ENCRYPTED_REASONING, + SUBSCRIPTION_MODELS, + build_usage_dict, + convert_messages_to_responses_input, + convert_tools_to_responses_format, + parse_sse_line, + strip_item_for_replay, + supports_verbosity, +) + +logger = logging.getLogger(__name__) + +_REASONING_EFFORTS = ("minimal", "low", "medium", "high", "xhigh") + + +def _subscription_reasoning_effort(requested: str | None = None) -> str: + """Reasoning effort for subscription requests. + + Precedence: the session's ``/effort`` setting (arrives as + ``extra_body.reasoning_effort`` via the agent-server's + ``_EffortProvider`` wrapper) → ``CLAWCODEX_OPENAI_REASONING_EFFORT`` + → ``medium`` (OpenCode's default, transform.ts:1176, and the + backend's own default_reasoning_level). ``xhigh``/``max`` clamp to + ``high`` — the general gpt-5.x models advertise low/medium/high and + reject higher tiers. + """ + for candidate in (requested, os.environ.get("CLAWCODEX_OPENAI_REASONING_EFFORT")): + effort = (candidate or "").strip().lower() + if effort in ("xhigh", "max"): + return "high" + if effort in _REASONING_EFFORTS: + return effort + return "medium" + + +class _HttpxStreamHolder: + """Adapter so ``StreamAbortGuard.attach`` can close an httpx response. + + The guard closes ``stream.response`` on abort — the SDK stream objects + expose that attribute; for the raw-httpx subscription path this shim + provides it. + """ + + __slots__ = ("response",) + + def __init__(self, response: Any) -> None: + self.response = response class OpenAIProvider(OpenAICompatibleProvider): - """OpenAI provider using OpenAI SDK.""" + """OpenAI provider using OpenAI SDK (API key) or the ChatGPT Codex + backend (subscription OAuth).""" def __init__( self, api_key: str, base_url: Optional[str] = None, model: Optional[str] = None @@ -27,8 +101,33 @@ def __init__( """ super().__init__(api_key, base_url, model or "gpt-5.4") + self._subscription_active = False + self._subscription_account_id = "" + # One id per provider instance ≈ one per session: rides the + # ``session_id`` header and ``prompt_cache_key`` so the backend can + # route consecutive requests to the same prompt cache. + self._subscription_session_id = str(uuid4()) + # A configured API key deliberately wins. With no key, fall back to + # the user's explicitly stored ChatGPT OAuth login — but only against + # the first-party endpoint (custom base URLs mean a proxy/gateway + # that expects the configured key semantics). Same policy as the + # Anthropic provider's Claude-subscription fallback. + oauth_eligible = not base_url or urlparse(base_url).hostname == "api.openai.com" + if not api_key and oauth_eligible: + # Presence check only — deliberately NOT ``get_valid_credentials``: + # that can perform a blocking token refresh (30 s urllib timeout), + # and providers are constructed at startup and on every /model + # switch. The request path refreshes right before each call, so + # freshness at construction time buys nothing. + from src.auth.openai_subscription import load_credentials + + credentials = load_credentials() + if credentials is not None: + self._subscription_active = True + self._subscription_account_id = credentials.account_id + def _create_client(self) -> Any: - """Create OpenAI SDK client. + """Create OpenAI SDK client (API-key path only). The read timeout that prevents a stalled stream from freezing the event loop is applied centrally by ``OpenAICompatibleProvider.client`` (via @@ -42,7 +141,6 @@ def _create_client(self) -> Any: if self.base_url: kwargs["base_url"] = self.base_url # Support SSL verification bypass for corporate/internal endpoints. - import os if os.environ.get("CLAWCODEX_SSL_VERIFY", "").lower() in ("0", "false", "no"): import httpx kwargs["http_client"] = httpx.Client(verify=False) @@ -54,8 +152,12 @@ def get_available_models(self) -> list[str]: Returns: List of model names """ + if self._subscription_active: + return list(SUBSCRIPTION_MODELS) return [ - # GPT-5.4 series (latest flagship) + # GPT-5.5 (flagship; also the ChatGPT-subscription default family) + "gpt-5.5", + # GPT-5.4 series "gpt-5.4", "gpt-5.4-pro", "gpt-5.4-mini", @@ -65,8 +167,9 @@ def get_available_models(self) -> list[str]: "gpt-5.2-pro", "gpt-5.2-mini", "gpt-5.2-nano", - # GPT-5.3-Codex (coding-specialized) + # Codex (coding-specialized) "gpt-5.3-codex", + "gpt-5.3-codex-spark", # Legacy GPT-4 series "gpt-4o", "gpt-4o-mini", @@ -74,3 +177,374 @@ def get_available_models(self) -> list[str]: "gpt-4", "gpt-3.5-turbo", ] + + # ------------------------------------------------------------------ + # ChatGPT-subscription path (Responses API against the Codex backend) + # ------------------------------------------------------------------ + + def chat( + self, + messages: list[MessageInput], + tools: Optional[list[dict[str, Any]]] = None, + **kwargs, + ) -> ChatResponse: + if self._subscription_active: + return self._subscription_stream_request(messages, tools, **kwargs) + return super().chat(messages, tools, **kwargs) + + def chat_stream( + self, + messages: list[MessageInput], + tools: Optional[list[dict[str, Any]]] = None, + **kwargs, + ) -> Generator[str, None, None]: + if not self._subscription_active: + yield from super().chat_stream(messages, tools, **kwargs) + return + # Callback → generator adaptation: run the request on a worker and + # relay text deltas through a bounded queue. + chunk_queue: queue.Queue = queue.Queue(maxsize=_CHUNK_QUEUE_MAXSIZE) + _DONE = object() + + def _run() -> None: + try: + self._subscription_stream_request( + messages, tools, + on_text_chunk=lambda piece: chunk_queue.put(piece), + **kwargs, + ) + except BaseException as exc: # noqa: BLE001 — surface to consumer + chunk_queue.put(exc) + finally: + chunk_queue.put(_DONE) + + worker = threading.Thread(target=_run, daemon=True, name="openai-subscription-stream") + worker.start() + while True: + item = chunk_queue.get() + if item is _DONE: + return + if isinstance(item, BaseException): + raise item + yield item + + def chat_stream_response( + self, + messages: list[MessageInput], + tools: Optional[list[dict[str, Any]]] = None, + on_text_chunk: TextChunkCallback | None = None, + abort_signal: Any = None, + on_thinking_chunk: TextChunkCallback | None = None, + **kwargs, + ) -> ChatResponse: + if self._subscription_active: + return self._subscription_stream_request( + messages, + tools, + on_text_chunk=on_text_chunk, + on_thinking_chunk=on_thinking_chunk, + abort_signal=abort_signal, + **kwargs, + ) + return super().chat_stream_response( + messages, + tools, + on_text_chunk=on_text_chunk, + abort_signal=abort_signal, + on_thinking_chunk=on_thinking_chunk, + **kwargs, + ) + + def _subscription_request_body( + self, + messages: list[MessageInput], + tools: Optional[list[dict[str, Any]]], + **kwargs, + ) -> dict[str, Any]: + model = self._get_model(**kwargs) + # RAW Anthropic-shape messages (dict conversion + image validation + # only) — deliberately NOT the base class's Chat Completions + # conversion; the Responses converter owns the translation. + prepared = BaseProvider._prepare_messages(self, messages) + input_items, instructions = convert_messages_to_responses_input(prepared) + # Side paths (compaction, agent hooks, memdir selector) pass an + # Anthropic-style ``system`` kwarg instead of a system message. + system_kwarg = kwargs.get("system") + if system_kwarg: + from .openai_responses import _system_text + + system_text = _system_text(system_kwarg) + if system_text: + instructions = ( + f"{system_text}\n\n{instructions}" if instructions else system_text + ) + + body: dict[str, Any] = { + "model": model, + "input": input_items, + # Stateless mode: the backend stores nothing; encrypted reasoning + # comes back inline and is replayed by the converter next turn. + "store": False, + "stream": True, + "include": list(INCLUDE_ENCRYPTED_REASONING), + "reasoning": { + # /effort arrives as extra_body.reasoning_effort via the + # agent-server's _EffortProvider wrapper (agent_server.py). + "effort": _subscription_reasoning_effort( + (kwargs.get("extra_body") or {}).get("reasoning_effort") + ), + "summary": "auto", + }, + "prompt_cache_key": self._subscription_session_id, + } + if instructions: + body["instructions"] = instructions + if tools: + converted = convert_tools_to_responses_format(tools) + if converted: + body["tools"] = converted + if supports_verbosity(model): + # OpenCode sends verbosity=low for gpt-5.x non-codex non-chat + # (transform.ts:1189); matches the backend's own default. + body["text"] = {"verbosity": "low"} + # NOTE: remaining kwargs (max_tokens, temperature, …) are + # intentionally NOT forwarded. The Codex backend rejects sampler + # params on reasoning models, and OpenCode explicitly forces + # maxOutputTokens off for this provider ("Match codex cli", + # plugin/openai/codex.ts:637-641). + return body + + def _subscription_headers(self, access_token: str) -> dict[str, str]: + from src.auth.openai_subscription import ORIGINATOR + + headers = { + "Authorization": f"Bearer {access_token}", + "OpenAI-Beta": "responses=experimental", + "originator": ORIGINATOR, + "session_id": self._subscription_session_id, + "Accept": "text/event-stream", + } + if self._subscription_account_id: + headers["chatgpt-account-id"] = self._subscription_account_id + return headers + + def _subscription_stream_request( + self, + messages: list[MessageInput], + tools: Optional[list[dict[str, Any]]] = None, + on_text_chunk: TextChunkCallback | None = None, + on_thinking_chunk: TextChunkCallback | None = None, + abort_signal: Any = None, + **kwargs, + ) -> ChatResponse: + """POST to the Codex backend and rebuild a ChatResponse from SSE. + + Same ESC-abort architecture as + ``OpenAICompatibleProvider.chat_stream_response``: the blocking + socket reads run on a daemon worker pushing lines into a bounded + queue; the main thread polls with a 100 ms tick and re-checks the + abort signal between ticks, so the user's prompt returns promptly + regardless of socket state. See that method's docstring for the + full rationale. + """ + import httpx + + from src.auth.openai_subscription import ( + CODEX_API_ENDPOINT, + force_refresh, + get_valid_credentials, + ) + from ._stream_abort import StreamAbortGuard + + guard = StreamAbortGuard(abort_signal) + guard.raise_if_pre_aborted() + + credentials = get_valid_credentials() + if credentials is None: + raise RuntimeError( + "ChatGPT subscription login was removed; run `clawcodex login`" + ) + self._subscription_account_id = ( + credentials.account_id or self._subscription_account_id + ) + + body = self._subscription_request_body(messages, tools, **kwargs) + + read = float(os.environ.get("CLAWCODEX_LLM_READ_TIMEOUT", "120")) + connect = float(os.environ.get("CLAWCODEX_LLM_CONNECT_TIMEOUT", "15")) + timeout = httpx.Timeout(connect=connect, read=read, write=30.0, pool=15.0) + verify = os.environ.get("CLAWCODEX_SSL_VERIFY", "").lower() not in ( + "0", "false", "no", + ) + + client = httpx.Client(timeout=timeout, verify=verify) + response: Any = None + try: + response = client.send( + client.build_request( + "POST", + CODEX_API_ENDPOINT, + headers=self._subscription_headers(credentials.access_token), + json=body, + ), + stream=True, + ) + if response.status_code == 401: + # Server-side invalidation ahead of local expiry — refresh + # once and retry. + response.close() + refreshed = force_refresh() + if refreshed is None: + raise RuntimeError( + "ChatGPT subscription login expired; run `clawcodex login`" + ) + self._subscription_account_id = ( + refreshed.account_id or self._subscription_account_id + ) + response = client.send( + client.build_request( + "POST", + CODEX_API_ENDPOINT, + headers=self._subscription_headers(refreshed.access_token), + json=body, + ), + stream=True, + ) + if response.status_code != 200: + detail = response.read().decode("utf-8", "replace") + raise RuntimeError( + f"ChatGPT backend error ({response.status_code}): {detail[:600]}" + ) + return self._consume_subscription_stream( + response, guard, on_text_chunk, on_thinking_chunk, + request_model=str(body.get("model", "")), + ) + except Exception as exc: + guard.reraise_if_aborted(exc) + raise + finally: + if response is not None: + try: + response.close() + except Exception: + pass + client.close() + + def _consume_subscription_stream( + self, + response: Any, + guard: Any, + on_text_chunk: TextChunkCallback | None, + on_thinking_chunk: TextChunkCallback | None, + request_model: str, + ) -> ChatResponse: + content_parts: list[str] = [] + reasoning_parts: list[str] = [] + items: list[dict[str, Any]] = [] + tool_uses: list[dict[str, Any]] = [] + usage: dict[str, Any] = {"billing_mode": "subscription"} + response_model = request_model + finish_reason = "stop" + failure: str | None = None + + _DONE = object() + line_queue: queue.Queue = queue.Queue(maxsize=_CHUNK_QUEUE_MAXSIZE) + + def _drain() -> None: + try: + for line in response.iter_lines(): + line_queue.put(line) + except BaseException as exc: # noqa: BLE001 — surface to consumer + line_queue.put(exc) + finally: + line_queue.put(_DONE) + + worker = threading.Thread( + target=_drain, daemon=True, name=f"openai-subscription-{id(response)}" + ) + + with guard.attach(_HttpxStreamHolder(response)): + worker.start() + while True: + try: + item = line_queue.get(timeout=0.1) + except queue.Empty: + if guard.aborted: + guard.raise_if_post_aborted() + continue + if item is _DONE: + break + if isinstance(item, BaseException): + if isinstance(item, Exception): + guard.reraise_if_aborted(item) + raise item + raise item + + event = parse_sse_line(str(item)) + if event is None: + continue + etype = event.get("type", "") + + if etype == "response.output_text.delta": + delta = str(event.get("delta", "") or "") + if delta: + content_parts.append(delta) + if on_text_chunk is not None: + on_text_chunk(delta) + elif etype == "response.reasoning_summary_text.delta": + delta = str(event.get("delta", "") or "") + if delta: + reasoning_parts.append(delta) + if on_thinking_chunk is not None: + on_thinking_chunk(delta) + elif etype == "response.output_item.done": + raw_item = event.get("item") + if isinstance(raw_item, dict): + stripped = strip_item_for_replay(raw_item) + items.append(stripped) + if stripped.get("type") == "function_call": + tool_uses.append({ + "id": str(stripped.get("call_id", "")), + "name": str(stripped.get("name", "")), + "input": _parse_tool_call_arguments( + stripped.get("arguments") + ), + }) + elif etype == "response.completed": + payload = event.get("response") or {} + usage = build_usage_dict(payload.get("usage")) + response_model = str(payload.get("model") or response_model) + elif etype == "response.incomplete": + payload = event.get("response") or {} + details = payload.get("incomplete_details") or {} + if "max_output_tokens" in str(details.get("reason", "")): + finish_reason = "max_tokens" + usage = build_usage_dict(payload.get("usage")) + elif etype in ("response.failed", "error"): + if etype == "error": + failure = str(event.get("message") or event) + else: + error = (event.get("response") or {}).get("error") or {} + failure = str(error.get("message") or error or event) + + if guard.aborted: + guard.raise_if_post_aborted() + + guard.raise_if_post_aborted() + if failure: + raise RuntimeError(f"ChatGPT backend request failed: {failure}") + + if tool_uses and finish_reason == "stop": + finish_reason = "tool_calls" + raw_blocks = [ + {"type": RESPONSES_ITEM_BLOCK_TYPE, "item": item} for item in items + ] + return ChatResponse( + content="".join(content_parts), + model=response_model, + usage=usage, + finish_reason=finish_reason, + reasoning_content="".join(reasoning_parts) or None, + tool_uses=tool_uses or None, + raw_content_blocks=raw_blocks or None, + ) diff --git a/src/providers/openai_responses.py b/src/providers/openai_responses.py new file mode 100644 index 00000000..2e8f917f --- /dev/null +++ b/src/providers/openai_responses.py @@ -0,0 +1,471 @@ +"""Wire-format helpers for the OpenAI Responses API (ChatGPT Codex backend). + +The ChatGPT-subscription request path (``src/auth/openai_subscription.py``) +speaks the *Responses* API — a different wire format from the Chat +Completions shape the rest of the OpenAI-compatible providers use: + +- tools are FLAT ``{"type": "function", "name", ...}`` (no nested + ``function`` object), +- the conversation is a single ``input`` item list mixing ``message``, + ``function_call``, ``function_call_output`` and ``reasoning`` items, +- with ``store: false`` the client replays reasoning items (with + ``encrypted_content``) itself, and item ``id``s are stripped — the + Codex CLI convention (OpenCode does the same, transform.ts:462). + +Shapes ported from OpenCode's vendored @ai-sdk/openai responses model +(``reference_projects/opencode/packages/core/src/github-copilot/responses/ +convert-to-openai-responses-input.ts``) and validated live against +``https://chatgpt.com/backend-api/codex/responses`` (2026-07-12): custom +``instructions`` are accepted, id-stripped replay works, and reasoning +items are optional on replay. + +Pure functions only — no I/O. The HTTP/streaming half lives in +``OpenAIProvider`` (``src/providers/openai_provider.py``). +""" + +from __future__ import annotations + +import json +import logging +from typing import Any + +logger = logging.getLogger(__name__) + +# Assistant-history passthrough block carrying one raw Responses output item +# (ids stripped). Emitted via ``ChatResponse.raw_content_blocks`` — the query +# loop appends these to assistant history verbatim (query.py:1090) so the next +# subscription request can replay the exact item sequence (reasoning +# interleaving included). Foreign converters must skip this type: +# ``_convert_anthropic_messages_to_openai`` and the Anthropic provider both +# filter it; Gemini's converter drops unknown block types by construction. +RESPONSES_ITEM_BLOCK_TYPE = "openai_responses_item" + +# Models served by the ChatGPT-subscription backend. Source of truth: +# OpenCode's ALLOWED_MODELS (plugin/openai/codex.ts:15) — which matches the +# live list Codex CLI caches from the backend (~/.codex/models_cache.json). +SUBSCRIPTION_MODELS = [ + "gpt-5.5", + "gpt-5.4", + "gpt-5.4-mini", + "gpt-5.3-codex-spark", +] + +INCLUDE_ENCRYPTED_REASONING = ["reasoning.encrypted_content"] + +# Item keys never replayed: ``id`` per the Codex convention (server-assigned +# ``rs_``/``fc_``/``msg_`` ids are meaningless with ``store: false``) and +# ``status`` (response-lifecycle metadata, not content). +_STRIPPED_ITEM_KEYS = ("id", "status") + + +def strip_item_for_replay(item: dict[str, Any]) -> dict[str, Any]: + """Drop server-assigned bookkeeping fields from a Responses output item.""" + return {k: v for k, v in item.items() if k not in _STRIPPED_ITEM_KEYS} + + +def strip_responses_item_blocks( + messages: list[dict[str, Any]], +) -> list[dict[str, Any]]: + """Remove ``openai_responses_item`` passthrough blocks for foreign providers. + + A mid-session ``/model`` switch away from the ChatGPT-subscription path + leaves these blocks in assistant history; Anthropic and Chat-Completions + APIs reject unknown content-block types, so foreign converters call this + first. A message left with no content after filtering is dropped + entirely (a reasoning-only turn has no foreign representation; it never + carries tool_use blocks, so no pairing is broken). + """ + result: list[dict[str, Any]] = [] + for msg in messages: + content = msg.get("content") + if isinstance(content, list) and any( + isinstance(block, dict) + and block.get("type") == RESPONSES_ITEM_BLOCK_TYPE + for block in content + ): + filtered = [ + block + for block in content + if not ( + isinstance(block, dict) + and block.get("type") == RESPONSES_ITEM_BLOCK_TYPE + ) + ] + if not filtered: + continue + msg = {**msg, "content": filtered} + result.append(msg) + return result + + +def supports_verbosity(model: str) -> bool: + """gpt-5.x general models accept ``text.verbosity``; codex/chat variants + don't (OpenCode transform.ts:1189-1196).""" + return ( + model.startswith("gpt-5") + and "codex" not in model + and "-chat" not in model + ) + + +# --- tools --------------------------------------------------------------- + + +def convert_tools_to_responses_format( + tools: list[dict[str, Any]] | None, +) -> list[dict[str, Any]]: + """Anthropic tool schemas → flat Responses function-tool format. + + Same invalid-schema guards as the Chat Completions converter + (``_convert_to_openai_tool_schema``): tools with a missing/None schema + type are skipped, bare object schemas gain empty ``properties``. + """ + converted: list[dict[str, Any]] = [] + for tool in tools or []: + input_schema = tool.get("input_schema") + if not input_schema or not isinstance(input_schema, dict): + continue + schema_type = input_schema.get("type") + if schema_type is None or schema_type == "None": + continue + if ( + schema_type == "object" + and "properties" not in input_schema + and "anyOf" not in input_schema + and "oneOf" not in input_schema + ): + input_schema = {**input_schema, "properties": {}} + converted.append({ + "type": "function", + "name": tool["name"], + "description": tool.get("description", ""), + "parameters": input_schema, + "strict": False, + }) + return converted + + +# --- input items ---------------------------------------------------------- + + +def _system_text(content: Any) -> str: + """Flatten a system message's content (str or text-block list) to text.""" + if isinstance(content, str): + return content + if isinstance(content, list): + parts = [] + for block in content: + if isinstance(block, str): + parts.append(block) + elif isinstance(block, dict): + parts.append(str(block.get("text", ""))) + return "\n".join(p for p in parts if p) + return str(content or "") + + +def _image_block_to_input_image(block: dict[str, Any]) -> dict[str, Any] | None: + """Anthropic base64 image block → Responses ``input_image`` part.""" + source = block.get("source") + if not isinstance(source, dict) or source.get("type") != "base64": + return None + data = source.get("data") + if not data or not isinstance(data, str): + return None + media_type = source.get("media_type") or "image/png" + return { + "type": "input_image", + "image_url": f"data:{media_type};base64,{data}", + } + + +def _document_block_to_input_file(block: dict[str, Any]) -> dict[str, Any] | None: + """Anthropic base64 document (PDF) block → Responses ``input_file`` part.""" + source = block.get("source") + if not isinstance(source, dict) or source.get("type") != "base64": + return None + data = source.get("data") + if not data or not isinstance(data, str): + return None + media_type = source.get("media_type") or "application/pdf" + return { + "type": "input_file", + "filename": "document.pdf", + "file_data": f"data:{media_type};base64,{data}", + } + + +def _user_block_to_part(block: Any) -> dict[str, Any] | None: + """One user-content block → Responses input part (None = drop).""" + if isinstance(block, str): + return {"type": "input_text", "text": block} if block else None + if not isinstance(block, dict): + return None + btype = block.get("type") + if btype == "text": + text = block.get("text", "") + return {"type": "input_text", "text": str(text)} if text else None + if btype == "image": + return _image_block_to_input_image(block) + if btype == "document": + return _document_block_to_input_file(block) + # Unknown block shapes (e.g. stale provider-specific passthroughs after a + # mid-session model switch) are dropped — the Responses API rejects + # unrecognised part types outright, unlike Chat Completions servers that + # merely 400 with a pointer. + logger.debug("Dropping unsupported user block type %r for Responses input", btype) + return None + + +def _flatten_tool_result_content( + raw_content: Any, +) -> tuple[str, list[dict[str, Any]]]: + """tool_result content → (text, multimodal input parts). + + Mirrors the Chat Completions converter's flattening: text blocks join by + newline, unknown dict blocks JSON-dump, images/documents split out for a + follow-up user message (``function_call_output.output`` is text-only). + """ + multimodal: list[dict[str, Any]] = [] + if isinstance(raw_content, list): + parts: list[str] = [] + for item in raw_content: + if isinstance(item, dict) and item.get("type") == "text": + parts.append(item.get("text", "")) + elif isinstance(item, dict) and item.get("type") in ("image", "document"): + translated = ( + _image_block_to_input_image(item) + if item.get("type") == "image" + else _document_block_to_input_file(item) + ) + if translated is not None: + multimodal.append(translated) + elif isinstance(item, str): + parts.append(item) + else: + parts.append(json.dumps(item) if isinstance(item, dict) else str(item)) + return "\n".join(parts) if parts else "", multimodal + if isinstance(raw_content, str): + return raw_content, multimodal + return str(raw_content), multimodal + + +def convert_messages_to_responses_input( + messages: list[dict[str, Any]], +) -> tuple[list[dict[str, Any]], str]: + """Anthropic-format messages → Responses ``input`` items + instructions. + + Returns ``(input_items, instructions)`` where ``instructions`` is the + flattened text of every ``role=system`` message (the query loop injects + the system prompt as a leading system message for non-Anthropic + providers, query.py:875). It rides the top-level ``instructions`` field — + Codex-CLI parity; the backend accepts arbitrary instructions (validated + live). + + Assistant messages that carry ``openai_responses_item`` passthrough + blocks replay those raw items verbatim in order and skip the projected + text/tool_use blocks (which are projections of the same items — see + ``OpenAIProvider`` response assembly). Anthropic-native assistant + messages (API-key sessions, model switches) are reconstructed from their + text/tool_use blocks instead. + """ + input_items: list[dict[str, Any]] = [] + instructions_parts: list[str] = [] + + # Orphan guard, ported from ``_convert_anthropic_messages_to_openai``: + # a ``function_call_output`` whose call_id was never sent as a + # ``function_call`` gets the request rejected. Collect known ids from + # both projections and passthrough items. + known_call_ids: set[str] = set() + for msg in messages: + if msg.get("role") != "assistant": + continue + content = msg.get("content") + if not isinstance(content, list): + continue + for block in content: + if not isinstance(block, dict): + continue + if block.get("type") == "tool_use": + known_call_ids.add(str(block.get("id", ""))) + elif block.get("type") == RESPONSES_ITEM_BLOCK_TYPE: + item = block.get("item") + if isinstance(item, dict) and item.get("type") == "function_call": + known_call_ids.add(str(item.get("call_id", ""))) + + for msg in messages: + role = msg.get("role", "user") + content = msg.get("content", "") + + if role == "system": + text = _system_text(content) + if text: + instructions_parts.append(text) + continue + + if role == "user": + if isinstance(content, str): + if content: + input_items.append({ + "role": "user", + "content": [{"type": "input_text", "text": content}], + }) + continue + if not isinstance(content, list): + continue + + user_parts: list[dict[str, Any]] = [] + deferred_multimodal: list[dict[str, Any]] = [] + # Tool outputs are emitted before the remaining user content, + # preserving the Chat Completions converter's ordering policy + # (outputs bind to the immediately preceding function_call turn). + for block in content: + if isinstance(block, dict) and block.get("type") == "tool_result": + call_id = str(block.get("tool_use_id", "")) + if call_id not in known_call_ids: + logger.debug( + "Dropping orphan tool_result for call_id=%r " + "(no matching function_call in history)", + call_id, + ) + continue + flat, multimodal = _flatten_tool_result_content( + block.get("content", "") + ) + if multimodal: + correlation = ( + f"[multimodal content for tool_use_id={call_id} " + "delivered in the following message]" + ) + flat = f"{flat}\n\n{correlation}" if flat else correlation + deferred_multimodal.append({ + "role": "user", + "content": [ + { + "type": "input_text", + "text": f"[content for tool_use_id={call_id}]", + }, + *multimodal, + ], + }) + if not flat: + flat = "[empty tool result]" + input_items.append({ + "type": "function_call_output", + "call_id": call_id, + "output": flat, + }) + else: + part = _user_block_to_part(block) + if part is not None: + user_parts.append(part) + + input_items.extend(deferred_multimodal) + if user_parts: + input_items.append({"role": "user", "content": user_parts}) + continue + + if role == "assistant": + if isinstance(content, str): + if content: + input_items.append({ + "role": "assistant", + "content": [{"type": "output_text", "text": content}], + }) + continue + if not isinstance(content, list): + continue + + passthrough_items = [ + block["item"] + for block in content + if isinstance(block, dict) + and block.get("type") == RESPONSES_ITEM_BLOCK_TYPE + and isinstance(block.get("item"), dict) + ] + if passthrough_items: + # Raw items carry the full generated sequence (reasoning / + # message / function_call, in order); text and tool_use + # blocks in the same message are projections of these items + # and must not be double-sent. + input_items.extend( + strip_item_for_replay(item) for item in passthrough_items + ) + continue + + text_parts: list[str] = [] + tool_calls: list[dict[str, Any]] = [] + for block in content: + if not isinstance(block, dict): + continue + btype = block.get("type") + if btype == "text": + text = block.get("text", "") + if text: + text_parts.append(str(text)) + elif btype == "tool_use": + tool_input = block.get("input", {}) + tool_calls.append({ + "type": "function_call", + "call_id": str(block.get("id", "")), + "name": str(block.get("name", "")), + "arguments": ( + json.dumps(tool_input) + if isinstance(tool_input, dict) + else str(tool_input) + ), + }) + # Other block types (thinking from an Anthropic turn, advisor + # passthroughs, …) have no Responses representation — drop. + if text_parts: + input_items.append({ + "role": "assistant", + "content": [ + {"type": "output_text", "text": "\n".join(text_parts)} + ], + }) + input_items.extend(tool_calls) + continue + + # Unknown role — drop (mirrors the tolerance of the CC converter's + # fallthrough, which the Responses API does not share). + logger.debug("Dropping message with unsupported role %r", role) + + return input_items, "\n\n".join(instructions_parts) + + +# --- responses -------------------------------------------------------------- + + +def build_usage_dict(usage: dict[str, Any] | None) -> dict[str, Any]: + """Responses usage JSON → the ChatResponse usage shape. + + ``billing_mode: subscription`` zeroes the cost in ``record_api_usage`` + (cost_tracker.py — the #697 mechanism); token counts still feed the + context-left display. + """ + usage = usage or {} + input_details = usage.get("input_tokens_details") or {} + result: dict[str, Any] = { + "input_tokens": int(usage.get("input_tokens", 0) or 0), + "output_tokens": int(usage.get("output_tokens", 0) or 0), + "total_tokens": int(usage.get("total_tokens", 0) or 0), + "billing_mode": "subscription", + } + cached = input_details.get("cached_tokens") + if cached: + result["cache_read_input_tokens"] = int(cached) + return result + + +def parse_sse_line(line: str) -> dict[str, Any] | None: + """One SSE line → event dict (None for non-data/keepalive/DONE lines).""" + if not line.startswith("data:"): + return None + data = line[5:].strip() + if not data or data == "[DONE]": + return None + try: + event = json.loads(data) + except (ValueError, json.JSONDecodeError): + return None + return event if isinstance(event, dict) else None diff --git a/src/server/agent_server.py b/src/server/agent_server.py index b14211b3..86f30b77 100644 --- a/src/server/agent_server.py +++ b/src/server/agent_server.py @@ -1039,12 +1039,12 @@ def _do_set_provider(self, request_id: object, name: object) -> None: self._reply(request_id, {"ok": False, "error": "missing provider"}) return from src.config import get_provider_config - from src.providers import get_provider_class, provider_requires_api_key, resolve_api_key + from src.providers import get_provider_class, provider_has_credentials, resolve_api_key from src.tool_system.defaults import build_default_registry provider_cfg = get_provider_config(name) api_key = resolve_api_key(name, provider_cfg) - if not api_key and provider_requires_api_key(name): + if not provider_has_credentials(name, api_key): self._reply(request_id, {"ok": False, "error": f"provider '{name}' is not configured (no API key)"}) return provider_cls = get_provider_class(name) @@ -3594,7 +3594,7 @@ def _build_runtime(sess: _AgentSession, perm_mode: str | None) -> None: from src.permissions.setup import setup_permissions from src.providers import ( get_provider_class, - provider_requires_api_key, + provider_has_credentials, resolve_api_key, ) from src.agent import Session @@ -3681,7 +3681,7 @@ def _build_runtime(sess: _AgentSession, perm_mode: str | None) -> None: provider_name = cfg.provider_name or get_default_provider() provider_cfg = get_provider_config(provider_name) api_key = resolve_api_key(provider_name, provider_cfg) - if not api_key and provider_requires_api_key(provider_name): + if not provider_has_credentials(provider_name, api_key): sess.init_error = ( f"API key for provider '{provider_name}' is not configured. " "Run `clawcodex login` to set it up." diff --git a/tests/test_deepseek_prefix_cache.py b/tests/test_deepseek_prefix_cache.py index 4fb4e0a0..11563110 100644 --- a/tests/test_deepseek_prefix_cache.py +++ b/tests/test_deepseek_prefix_cache.py @@ -55,7 +55,9 @@ def test_deepseek_v4_context_windows_registered(): def test_other_providers_context_window_unchanged(): # OpenRouter-DeepSeek (decision #1: out of scope) keeps the 200K default. assert get_context_window_for_model("deepseek/deepseek-v4-pro") == 200_000 - assert get_context_window_for_model("gpt-5.4") == 200_000 + # gpt-5.4 gained a registered 272K window with the ChatGPT-subscription + # work (models/configs.py) — no longer the 200K unknown-model default. + assert get_context_window_for_model("gpt-5.4") == 272_000 assert get_context_window_for_model("some-unknown-model") == 200_000 # Legacy aliases intentionally NOT registered (broad prefix-match risk). assert get_context_window_for_model("deepseek-chat") == 200_000 diff --git a/tests/test_model_system.py b/tests/test_model_system.py index 8d39fdf3..636c2fbc 100644 --- a/tests/test_model_system.py +++ b/tests/test_model_system.py @@ -61,7 +61,9 @@ def test_haiku_config(self): assert cfg.cost_input_per_mtok == 1.0 def test_unknown_returns_none(self): - assert get_model_config("gpt-4o") is None + # gpt-4o gained a real config with the ChatGPT-subscription work; + # use a genuinely unregistered id (no key shares its prefix base). + assert get_model_config("totally-unknown-model") is None def test_prefix_match(self): cfg = get_model_config("claude-sonnet-4-20250514-v2") diff --git a/tests/test_openai_subscription.py b/tests/test_openai_subscription.py new file mode 100644 index 00000000..e11e89c1 --- /dev/null +++ b/tests/test_openai_subscription.py @@ -0,0 +1,649 @@ +from __future__ import annotations + +import base64 +import hashlib +import json +import time +import urllib.parse +from pathlib import Path +from unittest.mock import patch + +from src.auth import openai_subscription as auth +from src.providers.openai_provider import OpenAIProvider +from src.providers.openai_responses import ( + RESPONSES_ITEM_BLOCK_TYPE, + build_usage_dict, + convert_messages_to_responses_input, + convert_tools_to_responses_format, + strip_responses_item_blocks, +) + + +def _credentials(expires_at: float | None = None) -> auth.SubscriptionCredentials: + return auth.SubscriptionCredentials( + "access", "refresh", expires_at or time.time() + 3600, "acct-123", "idtok" + ) + + +def _fake_jwt(claims: dict) -> str: + payload = base64.urlsafe_b64encode(json.dumps(claims).encode()).rstrip(b"=") + return f"h.{payload.decode()}.s" + + +# --- credential store --------------------------------------------------- + + +def test_credentials_are_private_and_round_trip(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setenv("CLAWCODEX_CONFIG_DIR", str(tmp_path)) + saved = _credentials() + auth.save_credentials(saved) + assert auth.load_credentials() == saved + assert auth.credentials_path().stat().st_mode & 0o777 == 0o600 + + +def test_refresh_posts_form_and_rotates_tokens(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setenv("CLAWCODEX_CONFIG_DIR", str(tmp_path)) + auth.save_credentials(_credentials(time.time() - 1)) + with patch.object(auth, "_post_form", return_value={ + "access_token": "new-access", "refresh_token": "new-refresh", "expires_in": 7200, + }) as post: + result = auth.get_valid_credentials() + assert result and result.access_token == "new-access" + stored = auth.load_credentials() + assert stored and stored.refresh_token == "new-refresh" + # Account id survives a refresh response that has no id_token. + assert stored.account_id == "acct-123" + assert post.call_args.args[0] == auth.TOKEN_URL + assert post.call_args.args[1]["grant_type"] == "refresh_token" + + +def test_refresh_keeps_old_refresh_token_when_absent(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setenv("CLAWCODEX_CONFIG_DIR", str(tmp_path)) + auth.save_credentials(_credentials(time.time() - 1)) + with patch.object(auth, "_post_form", return_value={ + "access_token": "new-access", "expires_in": 3600, + }): + result = auth.get_valid_credentials() + assert result and result.refresh_token == "refresh" + + +# --- login flows --------------------------------------------------------- + + +def test_begin_login_builds_pkce_authorize_url() -> None: + url, verifier, state = auth.begin_login() + parsed = urllib.parse.urlparse(url) + params = dict(urllib.parse.parse_qsl(parsed.query)) + assert url.startswith(auth.AUTHORIZE_URL) + assert params["client_id"] == auth.CLIENT_ID + assert params["redirect_uri"] == auth.REDIRECT_URI + assert params["state"] == state + assert params["codex_cli_simplified_flow"] == "true" + expected_challenge = ( + base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()) + .rstrip(b"=") + .decode() + ) + assert params["code_challenge"] == expected_challenge + + +def test_complete_login_exchanges_code_and_extracts_account( + tmp_path: Path, monkeypatch +) -> None: + monkeypatch.setenv("CLAWCODEX_CONFIG_DIR", str(tmp_path)) + id_token = _fake_jwt({ + "https://api.openai.com/auth": {"chatgpt_account_id": "acct-jwt"}, + }) + with patch.object(auth, "_post_form", return_value={ + "access_token": "a", "refresh_token": "r", "expires_in": 3600, + "id_token": id_token, + }) as post: + credentials = auth.complete_login("auth-code", "verifier") + payload = post.call_args.args[1] + assert payload["code"] == "auth-code" + assert payload["code_verifier"] == "verifier" + assert payload["grant_type"] == "authorization_code" + assert credentials.account_id == "acct-jwt" + assert auth.load_credentials() == credentials + + +def test_extract_account_id_claim_precedence() -> None: + direct = _fake_jwt({"chatgpt_account_id": "direct"}) + nested = _fake_jwt({"https://api.openai.com/auth": {"chatgpt_account_id": "nested"}}) + orgs = _fake_jwt({"organizations": [{"id": "org-1"}]}) + assert auth.extract_account_id(direct) == "direct" + assert auth.extract_account_id(nested) == "nested" + assert auth.extract_account_id(orgs) == "org-1" + # id_token wins over access_token; fall back when id_token has none. + assert auth.extract_account_id(_fake_jwt({}), nested) == "nested" + + +def test_import_codex_cli_credentials(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setenv("CLAWCODEX_CONFIG_DIR", str(tmp_path)) + codex_home = tmp_path / "codex" + codex_home.mkdir() + monkeypatch.setenv("CODEX_HOME", str(codex_home)) + access = _fake_jwt({"exp": time.time() + 1000}) + (codex_home / "auth.json").write_text(json.dumps({ + "auth_mode": "chatgpt", + "tokens": { + "id_token": "id", "access_token": access, + "refresh_token": "ref", "account_id": "acct-codex", + }, + })) + assert auth.has_codex_cli_credentials() + imported = auth.import_codex_cli_credentials() + assert imported.account_id == "acct-codex" + assert imported.refresh_token == "ref" + assert not imported.needs_refresh + assert auth.load_credentials() == imported + + +# --- Responses wire format ------------------------------------------------ + + +def test_converter_maps_system_tools_and_history() -> None: + messages = [ + {"role": "system", "content": "SYSTEM PROMPT"}, + {"role": "user", "content": "hello"}, + { + "role": "assistant", + "content": [ + {"type": "text", "text": "calling"}, + {"type": "tool_use", "id": "call_1", "name": "Bash", "input": {"c": 1}}, + ], + }, + { + "role": "user", + "content": [ + {"type": "tool_result", "tool_use_id": "call_1", "content": "ok"}, + {"type": "text", "text": "continue"}, + ], + }, + ] + items, instructions = convert_messages_to_responses_input(messages) + assert instructions == "SYSTEM PROMPT" + assert items[0] == { + "role": "user", "content": [{"type": "input_text", "text": "hello"}], + } + assert items[1]["content"] == [{"type": "output_text", "text": "calling"}] + assert items[2] == { + "type": "function_call", "call_id": "call_1", "name": "Bash", + "arguments": json.dumps({"c": 1}), + } + # Tool output precedes the remaining user content. + assert items[3] == { + "type": "function_call_output", "call_id": "call_1", "output": "ok", + } + assert items[4]["content"] == [{"type": "input_text", "text": "continue"}] + + +def test_converter_drops_orphan_tool_results() -> None: + messages = [ + {"role": "user", "content": [ + {"type": "tool_result", "tool_use_id": "never-issued", "content": "x"}, + ]}, + ] + items, _ = convert_messages_to_responses_input(messages) + assert items == [] + + +def test_converter_replays_passthrough_items_and_skips_projections() -> None: + reasoning_item = { + "type": "reasoning", "encrypted_content": "gAAA", "summary": [], + } + fcall_item = { + "type": "function_call", "call_id": "call_9", "name": "Bash", + "arguments": "{}", + } + messages = [ + { + "role": "assistant", + "content": [ + {"type": "text", "text": "projected text"}, + {"type": "tool_use", "id": "call_9", "name": "Bash", "input": {}}, + {"type": RESPONSES_ITEM_BLOCK_TYPE, "item": reasoning_item}, + {"type": RESPONSES_ITEM_BLOCK_TYPE, "item": fcall_item}, + ], + }, + {"role": "user", "content": [ + {"type": "tool_result", "tool_use_id": "call_9", "content": "done"}, + ]}, + ] + items, _ = convert_messages_to_responses_input(messages) + # Raw items only — the projected text/tool_use blocks must not double-send. + assert items[0] == reasoning_item + assert items[1] == fcall_item + assert items[2]["type"] == "function_call_output" + assert len(items) == 3 + + +def test_converter_translates_images_and_multimodal_tool_results() -> None: + image = {"type": "image", "source": { + "type": "base64", "media_type": "image/png", "data": "AAA", + }} + messages = [ + {"role": "assistant", "content": [ + {"type": "tool_use", "id": "c1", "name": "Read", "input": {}}, + ]}, + {"role": "user", "content": [ + {"type": "tool_result", "tool_use_id": "c1", "content": [image]}, + image, + ]}, + ] + items, _ = convert_messages_to_responses_input(messages) + output_item = items[1] + assert output_item["type"] == "function_call_output" + assert "delivered in the following message" in output_item["output"] + follow_up = items[2] + assert follow_up["role"] == "user" + assert follow_up["content"][1]["type"] == "input_image" + assert follow_up["content"][1]["image_url"] == "data:image/png;base64,AAA" + direct = items[3] + assert direct["content"][0]["type"] == "input_image" + + +def test_tools_convert_to_flat_function_format() -> None: + tools = [ + {"name": "Bash", "description": "run", "input_schema": {"type": "object"}}, + {"name": "Broken", "description": "x", "input_schema": {"type": None}}, + ] + converted = convert_tools_to_responses_format(tools) + assert converted == [{ + "type": "function", "name": "Bash", "description": "run", + "parameters": {"type": "object", "properties": {}}, "strict": False, + }] + + +def test_strip_responses_item_blocks_for_foreign_providers() -> None: + messages = [ + {"role": "assistant", "content": [ + {"type": "text", "text": "keep"}, + {"type": RESPONSES_ITEM_BLOCK_TYPE, "item": {"type": "reasoning"}}, + ]}, + {"role": "assistant", "content": [ + {"type": RESPONSES_ITEM_BLOCK_TYPE, "item": {"type": "reasoning"}}, + ]}, + {"role": "user", "content": "untouched"}, + ] + stripped = strip_responses_item_blocks(messages) + assert stripped[0]["content"] == [{"type": "text", "text": "keep"}] + # A passthrough-only assistant message is dropped entirely. + assert len(stripped) == 2 + assert stripped[1]["content"] == "untouched" + + +def test_chat_completions_converter_strips_passthrough_blocks() -> None: + from src.providers.openai_compatible import _convert_anthropic_messages_to_openai + + messages = [ + {"role": "assistant", "content": [ + {"type": "text", "text": "hi"}, + {"type": RESPONSES_ITEM_BLOCK_TYPE, "item": {"type": "reasoning"}}, + ]}, + ] + converted = _convert_anthropic_messages_to_openai(messages) + assert json.dumps(converted).find(RESPONSES_ITEM_BLOCK_TYPE) == -1 + + +def test_usage_dict_marks_subscription_billing() -> None: + usage = build_usage_dict({ + "input_tokens": 10, "output_tokens": 5, "total_tokens": 15, + "input_tokens_details": {"cached_tokens": 4}, + }) + assert usage == { + "input_tokens": 10, "output_tokens": 5, "total_tokens": 15, + "billing_mode": "subscription", "cache_read_input_tokens": 4, + } + + +# --- provider --------------------------------------------------------------- + + +class _FakeStreamResponse: + """Stand-in for a streamed httpx response emitting canned SSE lines.""" + + def __init__(self, lines: list[str], status_code: int = 200) -> None: + self._lines = lines + self.status_code = status_code + self.closed = False + + def iter_lines(self): + yield from self._lines + + def read(self) -> bytes: + return b"error-body" + + def close(self) -> None: + self.closed = True + + +def _sse(event: dict) -> str: + return "data: " + json.dumps(event) + + +def _subscription_provider(monkeypatch) -> OpenAIProvider: + monkeypatch.delenv("OPENAI_API_KEY", raising=False) + # Construction checks credential PRESENCE only (load_credentials); + # request-time freshness goes through get_valid_credentials. + with patch( + "src.auth.openai_subscription.load_credentials", + return_value=_credentials(), + ): + provider = OpenAIProvider(api_key="") + assert provider._subscription_active + return provider + + +def test_provider_streams_responses_and_builds_chat_response(monkeypatch) -> None: + provider = _subscription_provider(monkeypatch) + + reasoning_item = { + "id": "rs_1", "type": "reasoning", "encrypted_content": "enc", + "summary": [], + } + fcall_item = { + "id": "fc_1", "type": "function_call", "status": "completed", + "call_id": "call_7", "name": "Bash", "arguments": '{"cmd":"ls"}', + } + lines = [ + _sse({"type": "response.created", "response": {"id": "resp_1"}}), + _sse({"type": "response.output_text.delta", "delta": "hel"}), + _sse({"type": "response.reasoning_summary_text.delta", "delta": "think"}), + _sse({"type": "response.output_text.delta", "delta": "lo"}), + _sse({"type": "response.output_item.done", "item": reasoning_item}), + _sse({"type": "response.output_item.done", "item": fcall_item}), + _sse({"type": "response.completed", "response": { + "model": "gpt-5.4", + "usage": {"input_tokens": 7, "output_tokens": 3, "total_tokens": 10}, + }}), + ] + fake_response = _FakeStreamResponse(lines) + captured: dict = {} + + class _FakeClient: + def __init__(self, **kwargs): + captured["client_kwargs"] = kwargs + + def build_request(self, method, url, headers=None, json=None): + captured["url"] = url + captured["headers"] = headers + captured["body"] = json + return "request" + + def send(self, request, stream=False): + return fake_response + + def close(self): + pass + + text_chunks: list[str] = [] + thinking_chunks: list[str] = [] + with patch( + "src.auth.openai_subscription.get_valid_credentials", + return_value=_credentials(), + ), patch("httpx.Client", _FakeClient): + result = provider.chat_stream_response( + [ + {"role": "system", "content": "SYS"}, + {"role": "user", "content": "hi"}, + ], + tools=[{"name": "Bash", "description": "run", + "input_schema": {"type": "object", "properties": {}}}], + on_text_chunk=text_chunks.append, + on_thinking_chunk=thinking_chunks.append, + system="EXTRA", + ) + + assert captured["url"] == auth.CODEX_API_ENDPOINT + assert captured["headers"]["Authorization"] == "Bearer access" + assert captured["headers"]["chatgpt-account-id"] == "acct-123" + assert captured["headers"]["originator"] == auth.ORIGINATOR + body = captured["body"] + assert body["store"] is False and body["stream"] is True + assert body["include"] == ["reasoning.encrypted_content"] + assert body["instructions"] == "EXTRA\n\nSYS" + assert body["tools"][0]["name"] == "Bash" + assert "max_tokens" not in body and "max_output_tokens" not in body + + assert result.content == "hello" + assert text_chunks == ["hel", "lo"] + assert thinking_chunks == ["think"] + assert result.reasoning_content == "think" + assert result.tool_uses == [{"id": "call_7", "name": "Bash", "input": {"cmd": "ls"}}] + assert result.finish_reason == "tool_calls" + assert result.usage["billing_mode"] == "subscription" + assert result.usage["input_tokens"] == 7 + # Raw items round-trip with server ids stripped, ready for replay. + assert result.raw_content_blocks == [ + {"type": RESPONSES_ITEM_BLOCK_TYPE, + "item": {"type": "reasoning", "encrypted_content": "enc", "summary": []}}, + {"type": RESPONSES_ITEM_BLOCK_TYPE, + "item": {"type": "function_call", "call_id": "call_7", "name": "Bash", + "arguments": '{"cmd":"ls"}'}}, + ] + + +def test_chat_stream_yields_text_deltas(monkeypatch) -> None: + provider = _subscription_provider(monkeypatch) + lines = [ + _sse({"type": "response.output_text.delta", "delta": "hel"}), + _sse({"type": "response.output_text.delta", "delta": "lo"}), + _sse({"type": "response.completed", "response": { + "model": "gpt-5.4", "usage": {"input_tokens": 1, "output_tokens": 1}, + }}), + ] + + class _FakeClient: + def __init__(self, **kwargs): + pass + + def build_request(self, method, url, headers=None, json=None): + return "request" + + def send(self, request, stream=False): + return _FakeStreamResponse(lines) + + def close(self): + pass + + with patch( + "src.auth.openai_subscription.get_valid_credentials", + return_value=_credentials(), + ), patch("httpx.Client", _FakeClient): + assert list(provider.chat_stream([{"role": "user", "content": "hi"}])) == [ + "hel", "lo", + ] + + +def test_provider_refreshes_once_on_401(monkeypatch) -> None: + provider = _subscription_provider(monkeypatch) + + ok_lines = [ + _sse({"type": "response.output_text.delta", "delta": "ok"}), + _sse({"type": "response.completed", "response": { + "model": "gpt-5.4", "usage": {"input_tokens": 1, "output_tokens": 1}, + }}), + ] + responses = [ + _FakeStreamResponse([], status_code=401), + _FakeStreamResponse(ok_lines), + ] + sent_headers: list[dict] = [] + + class _FakeClient: + def __init__(self, **kwargs): + pass + + def build_request(self, method, url, headers=None, json=None): + sent_headers.append(headers) + return "request" + + def send(self, request, stream=False): + return responses.pop(0) + + def close(self): + pass + + refreshed = auth.SubscriptionCredentials( + "fresh-access", "fresh-refresh", time.time() + 3600, "acct-123", "" + ) + with patch( + "src.auth.openai_subscription.get_valid_credentials", + return_value=_credentials(), + ), patch( + "src.auth.openai_subscription.force_refresh", return_value=refreshed, + ) as force, patch("httpx.Client", _FakeClient): + result = provider.chat_stream_response([{"role": "user", "content": "hi"}]) + + assert force.call_count == 1 + assert sent_headers[0]["Authorization"] == "Bearer access" + assert sent_headers[1]["Authorization"] == "Bearer fresh-access" + assert result.content == "ok" + + +def test_provider_surfaces_backend_failure_events(monkeypatch) -> None: + provider = _subscription_provider(monkeypatch) + lines = [ + _sse({"type": "response.failed", "response": { + "error": {"message": "usage limit reached"}, + }}), + ] + + class _FakeClient: + def __init__(self, **kwargs): + pass + + def build_request(self, method, url, headers=None, json=None): + return "request" + + def send(self, request, stream=False): + return _FakeStreamResponse(lines) + + def close(self): + pass + + with patch( + "src.auth.openai_subscription.get_valid_credentials", + return_value=_credentials(), + ), patch("httpx.Client", _FakeClient): + try: + provider.chat_stream_response([{"role": "user", "content": "hi"}]) + raise AssertionError("expected RuntimeError") + except RuntimeError as exc: + assert "usage limit reached" in str(exc) + + +def test_abort_mid_stream_raises_and_closes_response(monkeypatch) -> None: + """ESC during generation: AbortError surfaces promptly and the guard's + close-on-abort listener actually closes the underlying httpx response + (via the _HttpxStreamHolder.response slot).""" + import threading + + from src.utils.abort_controller import AbortController, AbortError + + provider = _subscription_provider(monkeypatch) + controller = AbortController() + started = threading.Event() + + class _HangingResponse(_FakeStreamResponse): + def iter_lines(self): + yield _sse({"type": "response.output_text.delta", "delta": "par"}) + started.set() + # Block like a stalled socket until the abort closes us. + while not self.closed: + time.sleep(0.02) + raise RuntimeError("connection closed") + + fake_response = _HangingResponse([]) + + class _FakeClient: + def __init__(self, **kwargs): + pass + + def build_request(self, method, url, headers=None, json=None): + return "request" + + def send(self, request, stream=False): + return fake_response + + def close(self): + pass + + def _abort_after_first_chunk(): + started.wait(timeout=5) + controller.abort("user_interrupt") + + aborter = threading.Thread(target=_abort_after_first_chunk, daemon=True) + aborter.start() + with patch( + "src.auth.openai_subscription.get_valid_credentials", + return_value=_credentials(), + ), patch("httpx.Client", _FakeClient): + try: + provider.chat_stream_response( + [{"role": "user", "content": "hi"}], + abort_signal=controller.signal, + ) + raise AssertionError("expected AbortError") + except AbortError: + pass + aborter.join(timeout=5) + assert fake_response.closed, "abort must close the underlying response" + + +def test_effort_setting_reaches_reasoning_body(monkeypatch) -> None: + """/effort (injected as extra_body.reasoning_effort by the agent-server + wrapper) wins over the default; xhigh clamps to high.""" + provider = _subscription_provider(monkeypatch) + body = provider._subscription_request_body( + [{"role": "user", "content": "hi"}], None, + extra_body={"reasoning_effort": "low"}, + ) + assert body["reasoning"]["effort"] == "low" + body = provider._subscription_request_body( + [{"role": "user", "content": "hi"}], None, + extra_body={"reasoning_effort": "xhigh"}, + ) + assert body["reasoning"]["effort"] == "high" + body = provider._subscription_request_body( + [{"role": "user", "content": "hi"}], None, + ) + assert body["reasoning"]["effort"] == "medium" + + +def test_provider_lists_subscription_models(monkeypatch) -> None: + provider = _subscription_provider(monkeypatch) + models = provider.get_available_models() + assert "gpt-5.5" in models and "gpt-5.3-codex-spark" in models + + +def test_api_key_wins_over_subscription(monkeypatch) -> None: + with patch( + "src.auth.openai_subscription.load_credentials", + return_value=_credentials(), + ): + provider = OpenAIProvider(api_key="sk-live") + assert not provider._subscription_active + + +def test_provider_validation_accepts_subscription(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setenv("CLAWCODEX_CONFIG_DIR", str(tmp_path)) + monkeypatch.delenv("OPENAI_API_KEY", raising=False) + auth.save_credentials(_credentials()) + from src.providers import provider_has_credentials + + assert provider_has_credentials("openai", "") + auth.remove_credentials() + assert not provider_has_credentials("openai", "") + # The shared gate helper covers the Claude subscription too (the #697 + # flow was blocked by the agent-server's inline api-key checks before + # this helper existed — see agent_server.py session init / set_provider). + from src.auth import anthropic_subscription as anth + + anth.save_credentials( + anth.SubscriptionCredentials("a", "r", time.time() + 3600) + ) + assert provider_has_credentials("anthropic", "") + anth.remove_credentials() + assert not provider_has_credentials("anthropic", "")