diff --git a/coworker/providers/__init__.py b/coworker/providers/__init__.py
index 5245a15a..d38da557 100644
--- a/coworker/providers/__init__.py
+++ b/coworker/providers/__init__.py
@@ -8,6 +8,7 @@ from .base import (
ToolCall,
)
from .capabilities import capabilities_for
+from .codex_provider import CodexProvider
from .gemini_provider import GeminiProvider
from .openai_provider import OpenAIProvider, resolve_api_key
from .openai_responses import OpenAIResponsesProvider
@@ -33,6 +34,7 @@ __all__ = [
"ToolCall",
"AnthropicProvider",
"BedrockProvider",
+ "CodexProvider",
"GeminiProvider",
"OpenAIProvider",
"OpenAIResponsesProvider",
diff --git a/coworker/providers/codex_auth.py b/coworker/providers/codex_auth.py
new file mode 100644
index 00000000..681a6d9f
--- /dev/null
+++ b/coworker/providers/codex_auth.py
@@ -0,0 +1,472 @@
+"""Subscription sign-in for the `openai-codex` provider (OAuth 2.0 + PKCE).
+
+Instead of an API key, the user signs in with their ChatGPT plan: a browser flow
+against the vendor's auth service using their public subscription client id, with the
+loopback redirect that id is registered for (the port is FIXED — any other port fails
+the redirect-uri check server-side). Tokens land in the SecretStore profile
+`provider:openai-codex` — the same local-only storage every provider profile uses,
+never a plaintext config file — mirroring `mcp/oauth.py` (tokens + `tokens_issued_at`).
+
+The pieces:
+
+ - `sign_in()` — async, explicit-action only: bind the loopback port, open the
+ browser, wait for the redirect, exchange the code, persist tokens + account id.
+ - `CodexTokenStore` — persistence + proactive refresh (JWT `exp`, sync httpx: the
+ provider is called from engine worker threads, `asyncio.to_thread` like its peers).
+ - `verify()` — the Test-button probe: one cheap authenticated request that
+ distinguishes signed-out vs expired vs OK.
+
+The account id rides the token JWTs (the `https://api.openai.com/auth` claim); we
+decode without verification — the backend verifies the token, we only route with it.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+import hashlib
+import json
+import logging
+import secrets as pysecrets
+import time
+import uuid
+from typing import Any, Optional
+from urllib.parse import parse_qs, urlencode, urlsplit
+
+logger = logging.getLogger(__name__)
+
+AUTH_ISSUER = "https://auth.openai.com"
+AUTHORIZE_URL = AUTH_ISSUER + "/oauth/authorize"
+TOKEN_URL = AUTH_ISSUER + "/oauth/token"
+# The public subscription client id (ships in the vendor's own tooling — not a secret).
+CLIENT_ID = "app_EMoamEEZ73f0CkXaXp7hrann"
+CALLBACK_PORT = 1455
+CALLBACK_PATH = "/auth/callback"
+# Registered redirect for CLIENT_ID, verbatim — host and port are not ours to choose.
+REDIRECT_URI = f"http://localhost:{CALLBACK_PORT}{CALLBACK_PATH}"
+SCOPE = "openid profile email offline_access"
+ORIGINATOR = "openworker"
+CODEX_BASE_URL = "https://chatgpt.com/backend-api/codex"
+PROFILE = "provider:openai-codex"
+FLOW_TIMEOUT_SECONDS = 300
+# Refresh this close to the JWT `exp` instead of sending an about-to-die bearer.
+REFRESH_MARGIN_SECONDS = 300
+_ACCOUNT_CLAIM = "https://api.openai.com/auth"
+# Smallest curated model — the verify probe should cost as close to nothing as possible.
+_VERIFY_MODEL = "gpt-5.1-codex-mini"
+
+SIGNED_OUT_ERROR = (
+ "Not signed in to ChatGPT — connect your account in Settings ▸ Models to use "
+ "the subscription provider."
+)
+EXPIRED_ERROR = "ChatGPT session expired — sign in again in Settings ▸ Models."
+PLAN_LIMIT_ERROR = (
+ "ChatGPT plan limit reached — your subscription's rolling usage window (about "
+ "5 hours) is used up. Wait for it to reset, upgrade the plan, or switch to an "
+ "API-key provider."
+)
+PORT_BUSY_ERROR = (
+ f"Port {CALLBACK_PORT} is already in use — the OpenAI Codex CLI is the usual "
+ "holder. Quit it and start the sign-in again."
+)
+
+
+class CodexAuthError(RuntimeError):
+ """A subscription-auth failure with a user-readable message."""
+
+
+class CodexSignInRequired(CodexAuthError):
+ """No usable tokens — the fix is an explicit sign-in, never a silent browser."""
+
+
+# -- PKCE / JWT helpers -----------------------------------------------------------
+
+
+def create_pkce() -> tuple[str, str]:
+ """(verifier, S256 challenge) per RFC 7636."""
+ verifier = pysecrets.token_urlsafe(64)
+ digest = hashlib.sha256(verifier.encode("ascii")).digest()
+ challenge = base64.urlsafe_b64encode(digest).rstrip(b"=").decode("ascii")
+ return verifier, challenge
+
+
+def build_authorize_url(state: str, challenge: str) -> str:
+ params = {
+ "response_type": "code",
+ "client_id": CLIENT_ID,
+ "redirect_uri": REDIRECT_URI,
+ "scope": SCOPE,
+ "state": state,
+ "code_challenge": challenge,
+ "code_challenge_method": "S256",
+ # The simplified-flow switch the subscription client id expects, plus the
+ # client-name tag the backend requires on every call.
+ "codex_cli_simplified_flow": "true",
+ "originator": ORIGINATOR,
+ }
+ return AUTHORIZE_URL + "?" + urlencode(params)
+
+
+def _jwt_claims(token: str) -> dict[str, Any]:
+ """Decode a JWT payload WITHOUT verification — we only read routing claims
+ (`exp`, the account object); the backend is the one verifying signatures."""
+ try:
+ payload = token.split(".")[1]
+ payload += "=" * (-len(payload) % 4)
+ claims = json.loads(base64.urlsafe_b64decode(payload.encode("ascii")))
+ return claims if isinstance(claims, dict) else {}
+ except Exception:
+ return {}
+
+
+def account_id_from(tokens: dict[str, Any]) -> str:
+ """The ChatGPT account id, from the auth claim of the id/access token."""
+ for key in ("id_token", "access_token"):
+ auth = _jwt_claims(tokens.get(key) or "").get(_ACCOUNT_CLAIM) or {}
+ if isinstance(auth, dict):
+ acct = auth.get("chatgpt_account_id") or auth.get("account_id") or ""
+ if acct:
+ return str(acct)
+ return ""
+
+
+def backend_headers(account_id: str, session_id: str) -> dict[str, str]:
+ """The non-auth headers every backend request must carry (auth is the bearer)."""
+ return {
+ "chatgpt-account-id": account_id,
+ "originator": ORIGINATOR,
+ "OpenAI-Beta": "responses=experimental",
+ "session-id": session_id,
+ }
+
+
+# -- token persistence + refresh ----------------------------------------------------
+
+
+def _token_post(data: dict[str, str], timeout: float = 30.0) -> Any:
+ """One POST to the token endpoint (module-level so tests stub the wire here)."""
+ import httpx
+
+ return httpx.post(
+ TOKEN_URL, data=data, headers={"Accept": "application/json"}, timeout=timeout
+ )
+
+
+def exchange_code(code: str, verifier: str, timeout: float = 30.0) -> dict[str, Any]:
+ """authorization_code + PKCE verifier → the token set. Blocking (httpx sync);
+ `sign_in` runs it via `asyncio.to_thread`."""
+ resp = _token_post(
+ {
+ "grant_type": "authorization_code",
+ "code": code,
+ "redirect_uri": REDIRECT_URI,
+ "client_id": CLIENT_ID,
+ "code_verifier": verifier,
+ },
+ timeout,
+ )
+ if resp.status_code >= 300:
+ raise CodexAuthError(
+ f"Sign-in failed — token exchange returned HTTP {resp.status_code}."
+ )
+ return resp.json()
+
+
+class CodexTokenStore:
+ """Token set + account metadata in the `provider:openai-codex` SecretStore profile.
+
+ `access_token()` is what the provider calls per request: it hands back a live
+ bearer, refreshing proactively near the JWT `exp` and clearing the profile to a
+ clean signed-out state when the refresh token is rejected — never a crash loop.
+ """
+
+ def __init__(self, secrets: Any) -> None:
+ self._secrets = secrets
+
+ def _data(self) -> dict[str, Any]:
+ if self._secrets is None:
+ return {}
+ return self._secrets.get(PROFILE) or {}
+
+ def _merge(self, patch: dict[str, Any]) -> None:
+ self._secrets.put(PROFILE, {**self._data(), **patch})
+
+ def signed_in(self) -> bool:
+ return bool(self._data().get("tokens"))
+
+ def account_label(self) -> Optional[str]:
+ data = self._data()
+ return data.get("account_email") or data.get("account_id") or None
+
+ def save(self, tokens: dict[str, Any]) -> None:
+ """Persist a token response, keeping prior values a refresh omitted (the
+ refresh grant often returns no new refresh/id token)."""
+ existing = self._data().get("tokens") or {}
+ merged = {
+ k: (tokens.get(k) or existing.get(k))
+ for k in ("access_token", "refresh_token", "id_token")
+ }
+ merged = {k: v for k, v in merged.items() if v}
+ patch: dict[str, Any] = {
+ "tokens": merged,
+ "tokens_issued_at": int(time.time()),
+ }
+ account_id = account_id_from(merged) or self._data().get("account_id")
+ if account_id:
+ patch["account_id"] = account_id
+ email = _jwt_claims(merged.get("id_token") or "").get("email") or self._data().get(
+ "account_email"
+ )
+ if email:
+ patch["account_email"] = email
+ self._merge(patch)
+
+ def clear(self) -> bool:
+ if self._secrets is None:
+ return False
+ return bool(self._secrets.delete(PROFILE))
+
+ def access_token(self) -> tuple[str, str]:
+ """(live access token, account id) — refreshing first when stale/absent."""
+ data = self._data()
+ tokens = data.get("tokens") or {}
+ access = tokens.get("access_token") or ""
+ if not access and not tokens.get("refresh_token"):
+ raise CodexSignInRequired(SIGNED_OUT_ERROR)
+ exp = _jwt_claims(access).get("exp")
+ stale = not access or (
+ isinstance(exp, (int, float)) and exp - time.time() < REFRESH_MARGIN_SECONDS
+ )
+ if stale:
+ return self.refresh()
+ return access, data.get("account_id") or ""
+
+ def refresh(self) -> tuple[str, str]:
+ """refresh_token grant → fresh (access token, account id). A rejected refresh
+ token blanks the profile — the provider reads as cleanly signed out."""
+ refresh = (self._data().get("tokens") or {}).get("refresh_token") or ""
+ if not refresh:
+ self.clear()
+ raise CodexSignInRequired(EXPIRED_ERROR)
+ try:
+ resp = _token_post(
+ {
+ "grant_type": "refresh_token",
+ "refresh_token": refresh,
+ "client_id": CLIENT_ID,
+ }
+ )
+ except Exception as exc:
+ raise CodexAuthError(
+ "Couldn't reach the sign-in service to refresh the ChatGPT session "
+ f"({exc.__class__.__name__})."
+ ) from exc
+ if 400 <= resp.status_code < 500:
+ self.clear()
+ raise CodexSignInRequired(EXPIRED_ERROR)
+ if resp.status_code >= 300:
+ raise CodexAuthError(
+ f"ChatGPT session refresh failed (HTTP {resp.status_code}) — try again."
+ )
+ self.save(resp.json())
+ data = self._data()
+ return (data.get("tokens") or {}).get("access_token") or "", (
+ data.get("account_id") or ""
+ )
+
+
+# -- interactive sign-in flow -------------------------------------------------------
+
+# The last authorize URL, surfaced over REST so the GUI can offer "reopen sign-in
+# page" if the popup was lost (same affordance as mcp/oauth.py).
+last_authorize_url: Optional[str] = None
+_active_server: Optional[asyncio.AbstractServer] = None
+
+_PAGE = """OpenWorker
+
+{title}
{body}
"""
+
+
+def _http_response(status: str, title: str, body: str) -> bytes:
+ html = _PAGE.format(title=title, body=body).encode("utf-8")
+ head = (
+ f"HTTP/1.1 {status}\r\nContent-Type: text/html; charset=utf-8\r\n"
+ f"Content-Length: {len(html)}\r\nConnection: close\r\n\r\n"
+ )
+ return head.encode("ascii") + html
+
+
+async def _start_callback_server(
+ expected_state: str,
+) -> tuple[asyncio.AbstractServer, "asyncio.Future[str]"]:
+ """Bind the fixed loopback port and resolve the future with the auth code when
+ the redirect (carrying the matching `state`) lands."""
+ loop = asyncio.get_running_loop()
+ future: asyncio.Future[str] = loop.create_future()
+
+ async def handle(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+ try:
+ request_line = await reader.readline()
+ while True: # drain headers; the redirect is a bare GET
+ line = await reader.readline()
+ if line in (b"\r\n", b"\n", b""):
+ break
+ parts = request_line.decode("ascii", errors="replace").split()
+ target = urlsplit(parts[1] if len(parts) > 1 else "/")
+ if target.path != CALLBACK_PATH:
+ writer.write(_http_response("404 Not Found", "Not found", ""))
+ return
+ query = parse_qs(target.query)
+ error = (query.get("error") or [""])[0]
+ code = (query.get("code") or [""])[0]
+ state = (query.get("state") or [""])[0]
+ if error:
+ writer.write(
+ _http_response(
+ "400 Bad Request",
+ "Sign-in failed",
+ "The service reported an error. Return to OpenWorker and try again.",
+ )
+ )
+ if not future.done():
+ future.set_exception(
+ CodexAuthError(f"Sign-in failed — the service returned: {error}")
+ )
+ return
+ # Same loopback gate as mcp/oauth.py: a stray local hit with the wrong
+ # state must not consume the flow — only the genuine redirect resolves it.
+ if not code or not pysecrets.compare_digest(state, expected_state):
+ writer.write(
+ _http_response(
+ "400 Bad Request",
+ "Nothing waiting for this sign-in",
+ "The sign-in may have timed out. Return to OpenWorker and start it again.",
+ )
+ )
+ return
+ writer.write(
+ _http_response(
+ "200 OK",
+ "Signed in",
+ "You can close this tab and return to OpenWorker.",
+ )
+ )
+ if not future.done():
+ future.set_result(code)
+ finally:
+ try:
+ await writer.drain()
+ writer.close()
+ except Exception:
+ pass
+
+ try:
+ server = await asyncio.start_server(handle, "127.0.0.1", CALLBACK_PORT)
+ except OSError as exc:
+ raise CodexAuthError(PORT_BUSY_ERROR) from exc
+ return server, future
+
+
+async def sign_in(
+ secrets: Any,
+ *,
+ timeout: float = FLOW_TIMEOUT_SECONDS,
+ open_browser: bool = True,
+) -> dict[str, Any]:
+ """Run the full interactive flow: loopback server → browser → code → tokens.
+
+ Explicit-action only (a Settings button) — never called from an engine turn, so
+ unlike mcp/oauth.py it needs no non-interactive refusal path.
+ """
+ global last_authorize_url, _active_server
+ if _active_server is not None:
+ # A stale flow lost its browser tab; the new one takes the port.
+ _active_server.close()
+ await _active_server.wait_closed()
+ _active_server = None
+ verifier, challenge = create_pkce()
+ state = pysecrets.token_urlsafe(24)
+ url = build_authorize_url(state, challenge)
+ last_authorize_url = url
+ server, code_future = await _start_callback_server(state)
+ _active_server = server
+ try:
+ if open_browser:
+ import webbrowser
+
+ logger.info("codex auth: opening browser for sign-in")
+ await asyncio.get_running_loop().run_in_executor(None, webbrowser.open, url)
+ try:
+ code = await asyncio.wait_for(code_future, timeout)
+ except asyncio.TimeoutError:
+ raise CodexAuthError(
+ "Sign-in timed out — the browser window was not completed in "
+ f"{int(timeout) // 60} minutes."
+ )
+ finally:
+ server.close()
+ await server.wait_closed()
+ if _active_server is server:
+ _active_server = None
+ tokens = await asyncio.to_thread(exchange_code, code, verifier)
+ store = CodexTokenStore(secrets)
+ store.save(tokens)
+ if not (store._data().get("tokens") or {}).get("access_token"):
+ store.clear()
+ raise CodexAuthError("Sign-in failed — the token response had no access token.")
+ return {"ok": True, "account": store.account_label()}
+
+
+# -- verify probe -------------------------------------------------------------------
+
+
+def verify(secrets: Any, timeout: float = 10.0) -> dict[str, Any]:
+ """Test-button probe: one cheap authenticated request against the backend.
+
+ Distinguishes signed-out (no/rejected tokens) vs expired (401 with a bearer we
+ thought was live) vs OK. Never raises; {ok, error?, state?} like the other
+ provider verifies.
+ """
+ import httpx
+
+ store = CodexTokenStore(secrets)
+ if not store.signed_in():
+ return {"ok": False, "error": SIGNED_OUT_ERROR, "state": "signed_out"}
+ try:
+ token, account = store.access_token()
+ except CodexSignInRequired as exc:
+ return {"ok": False, "error": str(exc), "state": "signed_out"}
+ except CodexAuthError as exc:
+ return {"ok": False, "error": str(exc)}
+ try:
+ resp = httpx.post(
+ CODEX_BASE_URL + "/responses",
+ headers={
+ "Authorization": f"Bearer {token}",
+ **backend_headers(account, str(uuid.uuid4())),
+ },
+ json={
+ "model": _VERIFY_MODEL,
+ "input": "Reply with OK.",
+ "store": False,
+ "stream": True,
+ "max_output_tokens": 16,
+ },
+ timeout=timeout,
+ )
+ except Exception as exc:
+ return {
+ "ok": False,
+ "error": f"Couldn't reach the ChatGPT backend ({exc.__class__.__name__}).",
+ }
+ if resp.status_code < 300:
+ return {"ok": True, "account": store.account_label()}
+ if resp.status_code in (401, 403):
+ return {"ok": False, "error": EXPIRED_ERROR, "state": "expired"}
+ if resp.status_code == 429:
+ # Auth is fine — the plan window is just used up right now.
+ return {"ok": True, "account": store.account_label(), "note": PLAN_LIMIT_ERROR}
+ return {
+ "ok": False,
+ "error": f"The ChatGPT backend returned HTTP {resp.status_code}.",
+ }
diff --git a/coworker/providers/codex_provider.py b/coworker/providers/codex_provider.py
new file mode 100644
index 00000000..e180c1d6
--- /dev/null
+++ b/coworker/providers/codex_provider.py
@@ -0,0 +1,128 @@
+"""`openai-codex` provider — OpenAI models through a ChatGPT subscription.
+
+The backend speaks the same Responses wire as `/v1/responses` (stateless: full
+history each turn, `store: false`, encrypted reasoning in the `_openai` sidecar), so
+all conversion/parsing is inherited from `OpenAIResponsesProvider` — this subclass
+only swaps the credential: a short-lived OAuth bearer from `codex_auth` instead of an
+API key, plus the account/originator/session headers the backend requires.
+
+Differences from the API-key path:
+ - The backend serves streamed responses only, so `complete()` drains `stream()`.
+ - 401 → one refresh-and-retry (the bearer died mid-flight); a rejected refresh
+ token surfaces as a typed sign-in-required error, never a crash loop.
+ - 429 → the plan's rolling usage window, surfaced as a user-readable message.
+"""
+
+from __future__ import annotations
+
+import uuid
+from typing import Any, Optional
+
+from .base import AssistantTurn
+from .codex_auth import (
+ CODEX_BASE_URL,
+ PLAN_LIMIT_ERROR,
+ CodexTokenStore,
+ backend_headers,
+)
+from .openai_responses import OpenAIResponsesProvider
+
+
+def _status_code(exc: Exception) -> Optional[int]:
+ status = getattr(exc, "status_code", None)
+ if isinstance(status, int):
+ return status
+ status = getattr(getattr(exc, "response", None), "status_code", None)
+ return status if isinstance(status, int) else None
+
+
+class CodexProvider(OpenAIResponsesProvider):
+ def __init__(
+ self,
+ client: Any = None,
+ *,
+ secrets: Any = None,
+ default_model: str = "gpt-5.2-codex",
+ reasoning_summary: bool = True,
+ ):
+ super().__init__(
+ client=client,
+ default_model=default_model,
+ base_url=CODEX_BASE_URL,
+ reasoning_summary=reasoning_summary,
+ )
+ self._store = CodexTokenStore(secrets)
+ # One conversation per provider instance in practice (the router caches one
+ # client per provider); a uuid per instance satisfies the per-conversation
+ # session header without threading conversation ids through ProviderClient.
+ self._session_id = str(uuid.uuid4())
+ self._client_token: Optional[str] = None
+ self._injected = client is not None
+
+ def _ensure_client(self) -> Any:
+ if self._injected:
+ return self._client
+ # The bearer is short-lived: fetch per call (refreshes itself near expiry)
+ # and rebuild the SDK client whenever the token rotated.
+ token, account = self._store.access_token()
+ if self._client is None or token != self._client_token:
+ from openai import OpenAI
+
+ self._client = OpenAI(
+ api_key=token,
+ base_url=CODEX_BASE_URL,
+ default_headers=backend_headers(account, self._session_id),
+ )
+ self._client_token = token
+ return self._client
+
+ def _request_kwargs(
+ self,
+ *,
+ model: str,
+ messages: list[dict[str, Any]],
+ tools: Optional[list[dict[str, Any]]],
+ settings: dict[str, Any],
+ ) -> dict[str, Any]:
+ kwargs = super()._request_kwargs(
+ model=model, messages=messages, tools=tools, settings=settings
+ )
+ # Unlike stock /v1/responses, this backend honors a reasoning effort knob.
+ effort = settings.get("reasoning_effort")
+ if isinstance(effort, str) and effort:
+ kwargs["reasoning"] = {**kwargs.get("reasoning", {}), "effort": effort}
+ # The backend rejects requests without instructions; history normally
+ # carries a system prompt — this is only the bare-call fallback.
+ kwargs.setdefault("instructions", "You are a helpful assistant.")
+ return kwargs
+
+ def _create(self, client: Any, kwargs: dict[str, Any]) -> Any:
+ try:
+ return super()._create(client, kwargs)
+ except Exception as exc:
+ status = _status_code(exc)
+ if status == 401 and not self._injected:
+ # The bearer died mid-flight: force one refresh and retry once.
+ # A rejected refresh raises CodexSignInRequired out of the store.
+ self._store.refresh()
+ self._client = None
+ self._client_token = None
+ return super()._create(self._ensure_client(), kwargs)
+ if status == 429:
+ raise RuntimeError(PLAN_LIMIT_ERROR) from exc
+ raise
+
+ def complete(
+ self,
+ *,
+ model: str,
+ messages: list[dict[str, Any]],
+ tools: Optional[list[dict[str, Any]]] = None,
+ **settings: Any,
+ ) -> AssistantTurn:
+ # The backend only serves streamed responses — aggregate the stream.
+ turn: Optional[AssistantTurn] = None
+ for chunk in self.stream(model=model, messages=messages, tools=tools, **settings):
+ if chunk.turn is not None:
+ turn = chunk.turn
+ return turn if turn is not None else AssistantTurn()
diff --git a/coworker/providers/matrix.py b/coworker/providers/matrix.py
index f48b6944..839458a0 100644
--- a/coworker/providers/matrix.py
+++ b/coworker/providers/matrix.py
@@ -57,6 +57,33 @@ MATRIX: dict[str, ModelEntry] = {
"gpt-5.6-terra": ModelEntry("GPT-5.6 Terra · OpenAI", _AGENTIC_VISION, 400_000),
"gpt-5.6-luna": ModelEntry("GPT-5.6 Luna · OpenAI", _AGENTIC_VISION, 400_000),
"gpt-5.5": ModelEntry("GPT-5.5 · OpenAI", _AGENTIC_VISION, 400_000),
+ # ChatGPT-subscription catalog (the `openai-codex` OAuth provider). Curated to the
+ # ids the subscription backend actually serves; vision per the vendor's model docs,
+ # PDF unverified over this backend → local fallback via pdf_support.py.
+ "openai-codex:gpt-5.2-codex": ModelEntry(
+ "GPT-5.2 Codex · ChatGPT plan",
+ ModelCapabilities(
+ tools=True, vision=True, parallel_tool_calls=True, streaming=True
+ ),
+ 400_000,
+ ),
+ "openai-codex:gpt-5.2": ModelEntry(
+ "GPT-5.2 · ChatGPT plan",
+ ModelCapabilities(
+ tools=True, vision=True, parallel_tool_calls=True, streaming=True
+ ),
+ 400_000,
+ ),
+ "openai-codex:gpt-5.1-codex": ModelEntry(
+ "GPT-5.1 Codex · ChatGPT plan",
+ ModelCapabilities(
+ tools=True, vision=True, parallel_tool_calls=True, streaming=True
+ ),
+ 400_000,
+ ),
+ "openai-codex:gpt-5.1-codex-mini": ModelEntry(
+ "GPT-5.1 Codex Mini · ChatGPT plan", _AGENTIC, 400_000
+ ),
# Fable 5 (2026-06-09) is GA; its Mythos 5 sibling is approved-orgs-only, so it
# stays out of a picker meant for the public.
"anthropic:claude-fable-5": ModelEntry(
diff --git a/coworker/providers/registry.py b/coworker/providers/registry.py
index 1f7aea8e..bb91db4e 100644
--- a/coworker/providers/registry.py
+++ b/coworker/providers/registry.py
@@ -86,6 +86,10 @@ class ProviderDescriptor:
)
# One-line note under the provider title (e.g. "Connects through X's OpenAI-compatible API").
blurb: str = ""
+ # "oauth" → no key form at all: the provider is configured by a browser sign-in
+ # (tokens in its `provider:` profile) and the GUI renders connect/sign-out
+ # instead of fields. None → the usual key/field form.
+ auth: Optional[str] = None
def to_dict(self) -> dict[str, Any]:
return {
@@ -95,6 +99,7 @@ class ProviderDescriptor:
"fields": [f.to_dict() for f in self.fields],
"recommended_model": self.recommended_model,
"blurb": self.blurb,
+ "auth": self.auth,
}
@@ -124,6 +129,14 @@ def _build_openai(profile: dict[str, Any], secrets: Any) -> ProviderClient:
return OpenAIResponsesProvider(secrets=secrets)
+def _build_codex(profile: dict[str, Any], secrets: Any) -> ProviderClient:
+ # Credentials come from the OAuth token set in the provider's own profile,
+ # resolved (and refreshed) at call time by the token store — never a key.
+ from .codex_provider import CodexProvider
+
+ return CodexProvider(secrets=secrets)
+
+
def _build_anthropic(profile: dict[str, Any], secrets: Any) -> ProviderClient:
# Key resolution stays in AnthropicProvider/resolve_api_key (explicit → env → SecretStore),
# deferred to first call so the provider can be built before a key exists.
@@ -347,6 +360,17 @@ DESCRIPTORS: list[ProviderDescriptor] = [
recommended_model="gpt-5.6-sol",
env_key="OPENAI_API_KEY",
),
+ ProviderDescriptor(
+ name="openai-codex",
+ title="ChatGPT (OpenAI subscription)",
+ needs_key=False,
+ fields=[],
+ build=_build_codex,
+ recommended_model="gpt-5.2-codex",
+ blurb="Sign in with your ChatGPT plan and run OpenAI models through your "
+ "subscription — no API key. Tokens stay on this machine.",
+ auth="oauth",
+ ),
ProviderDescriptor(
name="anthropic",
title="Claude (Anthropic)",
@@ -693,6 +717,9 @@ def descriptor_configured(d: ProviderDescriptor, profile: dict[str, Any]) -> boo
a stored or env key. Multi-field cloud providers (no `api_key` field, e.g. Bedrock):
every required field present — their actual credentials may be ambient (~/.aws, ADC).
"""
+ if d.auth == "oauth":
+ # A stored token set = signed in (the tokens live in the same profile).
+ return bool((profile or {}).get("tokens"))
if not d.needs_key:
return True # keyless (Ollama) — usable out of the box
profile = profile or {}
@@ -908,6 +935,10 @@ def verify_provider_key(
d = _BY_NAME.get(name) or _BY_NAME["openai"]
key = (api_key or "").strip()
+ if d.auth == "oauth":
+ # OAuth providers verify from their stored tokens (needs the SecretStore),
+ # which only the manager holds — see SessionManager.verify_provider.
+ return {"ok": False, "error": f"{d.title} verifies via its sign-in, not a key."}
if name == "bedrock":
return _verify_bedrock(fields or {}, timeout)
if name == "vertex":
diff --git a/coworker/server/app.py b/coworker/server/app.py
index 0f6c553b..fdd7c64b 100644
--- a/coworker/server/app.py
+++ b/coworker/server/app.py
@@ -1811,6 +1811,24 @@ def create_app(manager: SessionManager) -> FastAPI:
manager.verify_provider, name, (body or {}).get("fields")
)
+ @app.post("/v1/providers/openai-codex/signin")
+ async def codex_signin() -> dict[str, Any]:
+ # Opens the system browser and waits on the loopback callback — that can
+ # take minutes, so it runs as a background task; the GUI polls the status
+ # route for the flip (authorizing → signed_in | last_error). Same shape as
+ # the MCP OAuth connect route.
+ manager.begin_codex_signin()
+ asyncio.create_task(manager.codex_signin())
+ return {"ok": True, "started": True}
+
+ @app.get("/v1/providers/openai-codex/status")
+ def codex_status() -> dict[str, Any]:
+ return manager.codex_status()
+
+ @app.post("/v1/providers/openai-codex/signout")
+ def codex_signout() -> dict[str, Any]:
+ return manager.codex_signout()
+
# -- settings (model API key) -----------------------------------------------
@app.get("/v1/settings")
def settings_get() -> dict[str, Any]:
diff --git a/coworker/server/manager.py b/coworker/server/manager.py
index 6390ecfc..bd02ab2d 100644
--- a/coworker/server/manager.py
+++ b/coworker/server/manager.py
@@ -186,6 +186,10 @@ class SessionManager:
# feeds list_mcp's status so the GUI can show "authorizing…" and failures.
self._mcp_authorizing: set[str] = set()
self._mcp_errors: dict[str, str] = {}
+ # ChatGPT-subscription provider sign-in in flight / its last error — feeds
+ # the providers list + status route so the GUI can show "authorizing…".
+ self._codex_authorizing = False
+ self._codex_error: Optional[str] = None
# http servers whose anonymous connect came back 401/403 — the failure is
# "needs sign-in", so the GUI offers the OAuth switch instead of a raw error.
self._mcp_auth_hints: set[str] = set()
@@ -2764,21 +2768,30 @@ class SessionManager:
for f in d.fields
if not f.secret and profile.get(f.key)
}
- out.append(
- {
- **d.to_dict(),
- "configured": configured,
- "values": values,
- "suggested_models": self._suggested_models(d.name),
- # Key hygiene for the Settings pane: when the key was saved (date, stamped
- # by set_provider) and when the provider last served a completion (epoch,
- # stamped by the router's on_use hook). Absent for env-only config.
- "key_set_at": profile.get("key_set_at"),
- "last_used_at": (self._prefs.get("provider_last_used") or {}).get(
- d.name
- ),
- }
- )
+ row = {
+ **d.to_dict(),
+ "configured": configured,
+ "values": values,
+ "suggested_models": self._suggested_models(d.name),
+ # Key hygiene for the Settings pane: when the key was saved (date, stamped
+ # by set_provider) and when the provider last served a completion (epoch,
+ # stamped by the router's on_use hook). Absent for env-only config.
+ "key_set_at": profile.get("key_set_at"),
+ "last_used_at": (self._prefs.get("provider_last_used") or {}).get(
+ d.name
+ ),
+ }
+ if d.auth == "oauth":
+ # Sign-in state instead of key state; the token values themselves
+ # never leave the SecretStore.
+ row["signed_in"] = configured
+ row["account"] = profile.get("account_email") or profile.get(
+ "account_id"
+ )
+ if d.name == "openai-codex":
+ row["authorizing"] = self._codex_authorizing
+ row["last_error"] = self._codex_error
+ out.append(row)
return out
def pick_native_folder(self) -> dict[str, Any]:
@@ -2919,6 +2932,58 @@ class SessionManager:
self._refresh_provider(name)
return {"ok": True, "provider": name}
+ # -- ChatGPT-subscription provider (OAuth, no key) ---------------------------
+ def begin_codex_signin(self) -> None:
+ """Flag `authorizing` BEFORE the background sign-in task starts, so the GUI's
+ first poll after the button press already shows it (same reasoning as
+ begin_mcp_connect)."""
+ self._codex_authorizing = True
+ self._codex_error = None
+
+ async def codex_signin(self) -> dict[str, Any]:
+ """Run the interactive browser sign-in and store the tokens. Long-running
+ (the user completes it in the browser) — routes run it as a background task
+ and the GUI polls codex_status for the flip."""
+ from ..providers import codex_auth
+
+ self._codex_authorizing = True
+ self._codex_error = None
+ try:
+ result = await codex_auth.sign_in(self.secrets)
+ except Exception as exc:
+ self._codex_error = str(exc)
+ return {"ok": False, "error": str(exc)}
+ finally:
+ self._codex_authorizing = False
+ self._refresh_provider("openai-codex")
+ # Same convenience as set_provider: surface the recommended model right away,
+ # and win the default when the current default's provider isn't usable.
+ added = "openai-codex:gpt-5.2-codex"
+ self.add_model(added)
+ if not self._provider_configured(self._model_provider(self.model)):
+ self.set_default_model(added)
+ return result
+
+ def codex_status(self) -> dict[str, Any]:
+ from ..providers import codex_auth
+
+ store = codex_auth.CodexTokenStore(self.secrets)
+ return {
+ "signed_in": store.signed_in(),
+ "account": store.account_label(),
+ "authorizing": self._codex_authorizing,
+ "last_error": self._codex_error,
+ "authorize_url": codex_auth.last_authorize_url,
+ }
+
+ def codex_signout(self) -> dict[str, Any]:
+ from ..providers import codex_auth
+
+ had_tokens = codex_auth.CodexTokenStore(self.secrets).clear()
+ self._codex_error = None
+ self._refresh_provider("openai-codex")
+ return {"ok": True, "had_tokens": had_tokens}
+
def verify_provider(
self, name: str, fields: Optional[dict[str, Any]]
) -> dict[str, Any]:
@@ -2930,6 +2995,11 @@ class SessionManager:
d = get_descriptor(name)
if d is None:
return {"ok": False, "error": f"unknown provider: {name}"}
+ if d.auth == "oauth":
+ # No key form — verify from the stored token set (signed-out / expired / OK).
+ from ..providers import codex_auth
+
+ return codex_auth.verify(self.secrets)
fields = fields or {}
profile = self.secrets.get(f"provider:{name}") or {}
merged = {}
diff --git a/tests/test_codex_provider.py b/tests/test_codex_provider.py
new file mode 100644
index 00000000..ba54782d
--- /dev/null
+++ b/tests/test_codex_provider.py
@@ -0,0 +1,562 @@
+"""ChatGPT-subscription provider (`openai-codex`): PKCE flow, token storage +
+refresh, backend request shape, failure modes, and the REST surface. No live
+network — the token endpoint, the SDK client, and the verify probe are all faked;
+the loopback callback server is exercised for real on its fixed port."""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+import json
+import socket
+import time
+from types import SimpleNamespace
+from urllib.parse import parse_qs, urlsplit
+
+import pytest
+from fastapi.testclient import TestClient
+
+from coworker.providers import codex_auth
+from coworker.providers.codex_auth import (
+ CODEX_BASE_URL,
+ CodexAuthError,
+ CodexSignInRequired,
+ CodexTokenStore,
+ account_id_from,
+ build_authorize_url,
+ create_pkce,
+)
+from coworker.providers.codex_provider import CodexProvider
+from coworker.secrets import SecretStore
+from coworker.server.app import create_app
+from coworker.server.manager import SessionManager
+
+ACCOUNT_CLAIM = "https://api.openai.com/auth"
+
+
+def _jwt(claims: dict) -> str:
+ def b64(obj) -> str:
+ raw = json.dumps(obj).encode()
+ return base64.urlsafe_b64encode(raw).rstrip(b"=").decode()
+
+ return f"{b64({'alg': 'none'})}.{b64(claims)}.sig"
+
+
+def _access(exp_offset: float = 3600, account: str = "acct_1") -> str:
+ return _jwt(
+ {"exp": time.time() + exp_offset, ACCOUNT_CLAIM: {"chatgpt_account_id": account}}
+ )
+
+
+def _id_token(email: str = "user@example.com", account: str = "acct_1") -> str:
+ return _jwt({"email": email, ACCOUNT_CLAIM: {"chatgpt_account_id": account}})
+
+
+def _token_response(status: int = 200, body: dict | None = None):
+ return SimpleNamespace(status_code=status, json=lambda: body or {})
+
+
+def _seed(secrets, exp_offset: float = 3600) -> None:
+ CodexTokenStore(secrets).save(
+ {
+ "access_token": _access(exp_offset),
+ "refresh_token": "rt-1",
+ "id_token": _id_token(),
+ }
+ )
+
+
+# -- PKCE / authorize URL -----------------------------------------------------------
+
+
+def test_pkce_challenge_is_s256_of_verifier():
+ import hashlib
+
+ verifier, challenge = create_pkce()
+ expected = (
+ base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest())
+ .rstrip(b"=")
+ .decode()
+ )
+ assert challenge == expected
+ assert create_pkce()[0] != verifier # fresh randomness per flow
+
+
+def test_authorize_url_shape():
+ url = build_authorize_url("st4te", "ch4llenge")
+ parts = urlsplit(url)
+ assert f"{parts.scheme}://{parts.netloc}{parts.path}" == codex_auth.AUTHORIZE_URL
+ q = {k: v[0] for k, v in parse_qs(parts.query).items()}
+ assert q["response_type"] == "code"
+ assert q["client_id"] == codex_auth.CLIENT_ID
+ assert q["redirect_uri"] == "http://localhost:1455/auth/callback"
+ assert q["state"] == "st4te"
+ assert q["code_challenge"] == "ch4llenge"
+ assert q["code_challenge_method"] == "S256"
+ assert q["codex_cli_simplified_flow"] == "true"
+ assert q["originator"] == codex_auth.ORIGINATOR
+
+
+def test_account_id_from_token_claim():
+ tokens = {"id_token": _id_token(account="acct_9")}
+ assert account_id_from(tokens) == "acct_9"
+ # Falls back to the access token, and to the plain account_id key.
+ tokens = {"access_token": _jwt({ACCOUNT_CLAIM: {"account_id": "acct_x"}})}
+ assert account_id_from(tokens) == "acct_x"
+ assert account_id_from({"access_token": "not-a-jwt"}) == ""
+
+
+# -- sign-in flow (loopback callback → exchange → storage) --------------------------
+
+
+async def _hit_callback(query: str) -> str:
+ reader, writer = await asyncio.open_connection("127.0.0.1", codex_auth.CALLBACK_PORT)
+ writer.write(
+ f"GET /auth/callback?{query} HTTP/1.1\r\nHost: localhost\r\n\r\n".encode()
+ )
+ await writer.drain()
+ data = await reader.read(-1)
+ writer.close()
+ return data.decode()
+
+
+async def test_sign_in_full_flow(tmp_path, monkeypatch):
+ secrets = SecretStore(tmp_path / "s.json")
+ opened: dict = {}
+ exchanged: dict = {}
+
+ monkeypatch.setattr("webbrowser.open", lambda url: opened.update(url=url))
+
+ def fake_token_post(data, timeout=30.0):
+ exchanged.update(data)
+ return _token_response(
+ body={
+ "access_token": _access(),
+ "refresh_token": "rt-1",
+ "id_token": _id_token("user@example.com"),
+ }
+ )
+
+ monkeypatch.setattr(codex_auth, "_token_post", fake_token_post)
+
+ task = asyncio.create_task(codex_auth.sign_in(secrets))
+ while not opened: # wait for the flow to bind the port and "open" the browser
+ await asyncio.sleep(0.01)
+ state = parse_qs(urlsplit(opened["url"]).query)["state"][0]
+
+ # A forged local hit with the wrong state is rejected and does NOT consume the flow.
+ resp = await _hit_callback("code=evil&state=wrong")
+ assert resp.startswith("HTTP/1.1 400")
+ assert not task.done()
+
+ resp = await _hit_callback(f"code=c0de&state={state}")
+ assert resp.startswith("HTTP/1.1 200") and "close this tab" in resp.lower()
+ result = await task
+
+ assert result == {"ok": True, "account": "user@example.com"}
+ assert exchanged["grant_type"] == "authorization_code"
+ assert exchanged["code"] == "c0de"
+ assert exchanged["redirect_uri"] == "http://localhost:1455/auth/callback"
+ assert exchanged["code_verifier"]
+ profile = secrets.get("provider:openai-codex")
+ assert profile["tokens"]["refresh_token"] == "rt-1"
+ assert profile["account_id"] == "acct_1"
+ assert profile["account_email"] == "user@example.com"
+ assert isinstance(profile["tokens_issued_at"], int)
+
+
+async def test_sign_in_provider_error_from_callback(tmp_path, monkeypatch):
+ secrets = SecretStore(tmp_path / "s.json")
+ opened: dict = {}
+ monkeypatch.setattr("webbrowser.open", lambda url: opened.update(url=url))
+
+ task = asyncio.create_task(codex_auth.sign_in(secrets))
+ while not opened:
+ await asyncio.sleep(0.01)
+ resp = await _hit_callback("error=access_denied")
+ assert resp.startswith("HTTP/1.1 400")
+ with pytest.raises(CodexAuthError, match="access_denied"):
+ await task
+ assert not CodexTokenStore(secrets).signed_in()
+
+
+async def test_sign_in_port_busy_names_the_usual_holder(tmp_path):
+ secrets = SecretStore(tmp_path / "s.json")
+ blocker = socket.socket()
+ blocker.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+ try:
+ blocker.bind(("127.0.0.1", codex_auth.CALLBACK_PORT))
+ blocker.listen(1)
+ with pytest.raises(CodexAuthError, match="1455"):
+ await codex_auth.sign_in(secrets, open_browser=False)
+ finally:
+ blocker.close()
+
+
+# -- token store: refresh ------------------------------------------------------------
+
+
+def test_access_token_fresh_needs_no_refresh(tmp_path, monkeypatch):
+ secrets = SecretStore(tmp_path / "s.json")
+ _seed(secrets)
+ monkeypatch.setattr(
+ codex_auth, "_token_post", lambda *a, **k: pytest.fail("refresh not needed")
+ )
+ token, account = CodexTokenStore(secrets).access_token()
+ assert token == secrets.get("provider:openai-codex")["tokens"]["access_token"]
+ assert account == "acct_1"
+
+
+def test_access_token_refreshes_near_expiry(tmp_path, monkeypatch):
+ secrets = SecretStore(tmp_path / "s.json")
+ _seed(secrets, exp_offset=30) # inside the refresh margin
+ sent: dict = {}
+ fresh = _access(7200)
+
+ def fake_token_post(data, timeout=30.0):
+ sent.update(data)
+ return _token_response(body={"access_token": fresh})
+
+ monkeypatch.setattr(codex_auth, "_token_post", fake_token_post)
+ token, account = CodexTokenStore(secrets).access_token()
+ assert token == fresh and account == "acct_1"
+ assert sent == {
+ "grant_type": "refresh_token",
+ "refresh_token": "rt-1",
+ "client_id": codex_auth.CLIENT_ID,
+ }
+ profile = secrets.get("provider:openai-codex")
+ # The refresh response omitted refresh/id tokens — prior values survive.
+ assert profile["tokens"]["access_token"] == fresh
+ assert profile["tokens"]["refresh_token"] == "rt-1"
+
+
+def test_rejected_refresh_blanks_tokens_to_signed_out(tmp_path, monkeypatch):
+ secrets = SecretStore(tmp_path / "s.json")
+ _seed(secrets, exp_offset=-10)
+ monkeypatch.setattr(
+ codex_auth, "_token_post", lambda *a, **k: _token_response(status=400)
+ )
+ store = CodexTokenStore(secrets)
+ with pytest.raises(CodexSignInRequired):
+ store.access_token()
+ assert not store.signed_in() # clean signed-out state, not a crash loop
+
+
+def test_access_token_signed_out_raises_typed_error(tmp_path):
+ store = CodexTokenStore(SecretStore(tmp_path / "s.json"))
+ with pytest.raises(CodexSignInRequired, match="Not signed in"):
+ store.access_token()
+
+
+# -- provider: request shape to the backend -----------------------------------------
+
+
+class _FakeSDKClient:
+ def __init__(self, events=None, errors=None):
+ self.kwargs: dict = {}
+ errors = list(errors or [])
+
+ def create(**kwargs):
+ self.kwargs = kwargs
+ if errors:
+ raise errors.pop(0)
+ return iter(events or [])
+
+ self.responses = SimpleNamespace(create=create)
+
+
+def _completed_event(text="hello"):
+ return SimpleNamespace(
+ type="response.completed",
+ response=SimpleNamespace(
+ output=[
+ {
+ "type": "message",
+ "role": "assistant",
+ "content": [{"type": "output_text", "text": text}],
+ }
+ ],
+ status="completed",
+ incomplete_details=None,
+ ),
+ )
+
+
+def _status_error(status: int, message: str = ""):
+ exc = Exception(message or f"HTTP {status}")
+ exc.status_code = status
+ return exc
+
+
+def _provider(tmp_path, monkeypatch, clients):
+ """A CodexProvider over seeded tokens whose SDK clients come from `clients`
+ (each OpenAI() construction pops the next one and records its init kwargs)."""
+ secrets = SecretStore(tmp_path / "s.json")
+ _seed(secrets)
+ built: list[dict] = []
+
+ def fake_openai(**kwargs):
+ built.append(kwargs)
+ return clients.pop(0)
+
+ monkeypatch.setattr("openai.OpenAI", fake_openai)
+ return CodexProvider(secrets=secrets), secrets, built
+
+
+def test_stream_request_headers_and_body(tmp_path, monkeypatch):
+ fake = _FakeSDKClient(events=[_completed_event()])
+ provider, secrets, built = _provider(tmp_path, monkeypatch, [fake])
+
+ out = list(
+ provider.stream(
+ model="gpt-5.2-codex",
+ messages=[
+ {"role": "system", "content": "sys"},
+ {"role": "user", "content": "hi"},
+ ],
+ tools=[{"type": "function", "function": {"name": "f"}}],
+ )
+ )
+
+ assert built[0]["api_key"] == secrets.get("provider:openai-codex")["tokens"][
+ "access_token"
+ ]
+ assert built[0]["base_url"] == CODEX_BASE_URL
+ headers = built[0]["default_headers"]
+ assert headers["chatgpt-account-id"] == "acct_1"
+ assert headers["originator"] == codex_auth.ORIGINATOR
+ assert headers["OpenAI-Beta"] == "responses=experimental"
+ assert headers["session-id"] # per-conversation uuid
+ assert fake.kwargs["model"] == "gpt-5.2-codex"
+ assert fake.kwargs["stream"] is True
+ assert fake.kwargs["store"] is False
+ assert fake.kwargs["include"] == ["reasoning.encrypted_content"]
+ assert fake.kwargs["instructions"] == "sys"
+ assert fake.kwargs["tools"] == [{"type": "function", "name": "f"}]
+ assert out[-1].turn.text == "hello"
+
+
+def test_complete_goes_through_stream_and_reasoning_effort_rides(tmp_path, monkeypatch):
+ fake = _FakeSDKClient(events=[_completed_event("done")])
+ provider, _, _ = _provider(tmp_path, monkeypatch, [fake])
+ turn = provider.complete(
+ model="gpt-5.2-codex",
+ messages=[{"role": "user", "content": "hi"}],
+ reasoning_effort="high",
+ )
+ assert turn.text == "done"
+ assert fake.kwargs["stream"] is True # the backend only serves streams
+ assert fake.kwargs["reasoning"] == {"summary": "auto", "effort": "high"}
+ assert fake.kwargs["instructions"] # bare-call fallback instructions present
+
+
+def test_401_refreshes_once_and_retries_with_new_bearer(tmp_path, monkeypatch):
+ first = _FakeSDKClient(errors=[_status_error(401, "Unauthorized")])
+ second = _FakeSDKClient(events=[_completed_event("after refresh")])
+ provider, secrets, built = _provider(tmp_path, monkeypatch, [first, second])
+ fresh = _access(7200)
+ monkeypatch.setattr(
+ codex_auth,
+ "_token_post",
+ lambda *a, **k: _token_response(body={"access_token": fresh}),
+ )
+ turn = provider.complete(
+ model="gpt-5.2-codex", messages=[{"role": "user", "content": "hi"}]
+ )
+ assert turn.text == "after refresh"
+ assert len(built) == 2 and built[1]["api_key"] == fresh
+
+
+def test_429_surfaces_plan_limit_message(tmp_path, monkeypatch):
+ fake = _FakeSDKClient(errors=[_status_error(429, "Too Many Requests")])
+ provider, _, _ = _provider(tmp_path, monkeypatch, [fake])
+ with pytest.raises(RuntimeError, match="plan limit"):
+ provider.complete(
+ model="gpt-5.2-codex", messages=[{"role": "user", "content": "hi"}]
+ )
+
+
+def test_signed_out_provider_raises_typed_error(tmp_path):
+ provider = CodexProvider(secrets=SecretStore(tmp_path / "s.json"))
+ with pytest.raises(CodexSignInRequired, match="Not signed in"):
+ provider.complete(
+ model="gpt-5.2-codex", messages=[{"role": "user", "content": "hi"}]
+ )
+
+
+# -- registry / matrix ---------------------------------------------------------------
+
+
+def test_registry_builds_codex_provider():
+ from coworker.providers.registry import build_provider_client, get_descriptor
+
+ assert isinstance(build_provider_client("openai-codex", {}, None), CodexProvider)
+ d = get_descriptor("openai-codex")
+ assert d.auth == "oauth" and d.fields == []
+ assert d.to_dict()["auth"] == "oauth"
+
+
+def test_descriptor_configured_means_tokens_present():
+ from coworker.providers.registry import descriptor_configured, get_descriptor
+
+ d = get_descriptor("openai-codex")
+ assert not descriptor_configured(d, {})
+ assert descriptor_configured(d, {"tokens": {"access_token": "a"}})
+
+
+def test_matrix_curates_subscription_models():
+ from coworker.providers.capabilities import capabilities_for
+ from coworker.providers.matrix import models_for_provider
+
+ assert models_for_provider("openai-codex") == [
+ "gpt-5.2-codex",
+ "gpt-5.2",
+ "gpt-5.1-codex",
+ "gpt-5.1-codex-mini",
+ ]
+ caps = capabilities_for("openai-codex:gpt-5.2-codex")
+ assert caps.tools and caps.vision and caps.streaming
+
+
+# -- verify probe --------------------------------------------------------------------
+
+
+def _verify_with_backend(tmp_path, monkeypatch, status: int):
+ secrets = SecretStore(tmp_path / "s.json")
+ _seed(secrets)
+ probes: dict = {}
+
+ def fake_post(url, headers=None, json=None, timeout=None):
+ probes.update(url=url, headers=headers, body=json)
+ return SimpleNamespace(status_code=status)
+
+ monkeypatch.setattr("httpx.post", fake_post)
+ return codex_auth.verify(secrets), probes
+
+
+def test_verify_signed_out(tmp_path):
+ result = codex_auth.verify(SecretStore(tmp_path / "s.json"))
+ assert result["ok"] is False and result["state"] == "signed_out"
+
+
+def test_verify_ok_probes_backend(tmp_path, monkeypatch):
+ result, probes = _verify_with_backend(tmp_path, monkeypatch, 200)
+ assert result == {"ok": True, "account": "user@example.com"}
+ assert probes["url"] == CODEX_BASE_URL + "/responses"
+ assert probes["headers"]["Authorization"].startswith("Bearer ")
+ assert probes["headers"]["chatgpt-account-id"] == "acct_1"
+ assert probes["body"]["store"] is False and probes["body"]["stream"] is True
+
+
+def test_verify_expired_and_plan_limited(tmp_path, monkeypatch):
+ result, _ = _verify_with_backend(tmp_path, monkeypatch, 401)
+ assert result["ok"] is False and result["state"] == "expired"
+ result, _ = _verify_with_backend(tmp_path, monkeypatch, 429)
+ assert result["ok"] is True and "note" in result # auth fine, window used up
+
+
+# -- REST surface --------------------------------------------------------------------
+
+
+def _rest(tmp_path):
+ manager = SessionManager(data_dir=tmp_path / "data")
+ return manager, TestClient(create_app(manager))
+
+
+def test_providers_list_shows_oauth_state(tmp_path):
+ manager, client = _rest(tmp_path)
+ rows = {p["name"]: p for p in client.get("/v1/providers").json()}
+ row = rows["openai-codex"]
+ assert row["auth"] == "oauth"
+ assert row["signed_in"] is False and row["configured"] is False
+ assert "gpt-5.2-codex" in row["suggested_models"]
+
+ manager.secrets.put(
+ "provider:openai-codex",
+ {"tokens": {"access_token": "a", "refresh_token": "r"}, "account_email": "u@x.com"},
+ )
+ row = {p["name"]: p for p in client.get("/v1/providers").json()}["openai-codex"]
+ assert row["signed_in"] is True and row["configured"] is True
+ assert row["account"] == "u@x.com"
+ assert "tokens" not in row.get("values", {}) # secrets never leave the store
+
+
+def test_signin_route_starts_background_flow(tmp_path, monkeypatch):
+ manager, _ = _rest(tmp_path)
+ seen = {}
+
+ async def fake_signin():
+ seen["called"] = True
+ return {"ok": True}
+
+ monkeypatch.setattr(manager, "codex_signin", fake_signin)
+ client = TestClient(create_app(manager))
+ assert client.post("/v1/providers/openai-codex/signin").json() == {
+ "ok": True,
+ "started": True,
+ }
+ assert seen["called"]
+ # begin_codex_signin flagged before the task ran, so the very first status
+ # poll after the button press already shows authorizing.
+ assert manager._codex_authorizing is True
+
+
+def test_status_and_signout_routes(tmp_path):
+ manager, client = _rest(tmp_path)
+ status = client.get("/v1/providers/openai-codex/status").json()
+ assert status["signed_in"] is False and status["authorizing"] is False
+
+ manager.secrets.put(
+ "provider:openai-codex",
+ {"tokens": {"access_token": "a"}, "account_email": "u@x.com"},
+ )
+ status = client.get("/v1/providers/openai-codex/status").json()
+ assert status["signed_in"] is True and status["account"] == "u@x.com"
+
+ assert client.post("/v1/providers/openai-codex/signout").json() == {
+ "ok": True,
+ "had_tokens": True,
+ }
+ assert client.get("/v1/providers/openai-codex/status").json()["signed_in"] is False
+
+
+def test_verify_route_reports_signed_out(tmp_path):
+ _, client = _rest(tmp_path)
+ result = client.post("/v1/providers/verify", json={"name": "openai-codex"}).json()
+ assert result["ok"] is False and result["state"] == "signed_out"
+
+
+async def test_manager_signin_stores_and_promotes_model(tmp_path, monkeypatch):
+ manager, _ = _rest(tmp_path)
+
+ async def fake_sign_in(secrets, **kwargs):
+ CodexTokenStore(secrets).save(
+ {
+ "access_token": _access(),
+ "refresh_token": "rt-1",
+ "id_token": _id_token(),
+ }
+ )
+ return {"ok": True, "account": "user@example.com"}
+
+ monkeypatch.setattr(codex_auth, "sign_in", fake_sign_in)
+ result = await manager.codex_signin()
+ assert result["ok"] is True
+ assert manager._codex_authorizing is False
+ settings = manager.get_settings()
+ assert "openai-codex:gpt-5.2-codex" in settings["models"]
+
+
+async def test_manager_signin_failure_lands_in_status(tmp_path, monkeypatch):
+ manager, client = _rest(tmp_path)
+
+ async def fake_sign_in(secrets, **kwargs):
+ raise CodexAuthError("Port 1455 is already in use — quit the other holder.")
+
+ monkeypatch.setattr(codex_auth, "sign_in", fake_sign_in)
+ result = await manager.codex_signin()
+ assert result["ok"] is False
+ status = client.get("/v1/providers/openai-codex/status").json()
+ assert "1455" in status["last_error"]
+ assert status["authorizing"] is False