Provider: ChatGPT-subscription sign-in (OAuth PKCE, tokens local-only)

Browser sign-in on the registered loopback port; tokens in the SecretStore profile.
Rides the Responses provider for conversion/streaming; refresh-on-expiry and on-401.
REST: signin (background + status poll), status, signout; oauth rows in providers list.

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",

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 = """<!doctype html><meta charset="utf-8"><title>OpenWorker</title>
+<body style="font-family: system-ui; margin: 4rem auto; max-width: 28rem; text-align: center;">
+<h2>{title}</h2><p>{body}</p></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}.",
+    }

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()

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(

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:<name>` 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":

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]:

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(
+            row = {
-                {
+                **d.to_dict(),
-                    **d.to_dict(),
+                "configured": configured,
-                    "configured": configured,
+                "values": values,
-                    "values": values,
+                "suggested_models": self._suggested_models(d.name),
-                    "suggested_models": self._suggested_models(d.name),
+                # Key hygiene for the Settings pane: when the key was saved (date, stamped
-                    # 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,
-                    # 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.
-                    # stamped by the router's on_use hook). Absent for env-only config.
+                "key_set_at": profile.get("key_set_at"),
-                    "key_set_at": profile.get("key_set_at"),
+                "last_used_at": (self._prefs.get("provider_last_used") or {}).get(
-                    "last_used_at": (self._prefs.get("provider_last_used") or {}).get(
+                    d.name
-                        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 = {}

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