diff --git a/botspot/components/new/chatgpt_plan/README.md b/botspot/components/new/chatgpt_plan/README.md
new file mode 100644
index 0000000..39df33c
--- /dev/null
+++ b/botspot/components/new/chatgpt_plan/README.md
@@ -0,0 +1,95 @@
+# chatgpt_plan
+
+Lets a bot user link their own ChatGPT plan ("Sign in with ChatGPT", SIWC) so the
+bot's LLM calls run on that plan instead of the bot owner's API key. **Off by default.**
+With the flag off, every call raises `PlanDisabled` and does no I/O.
+
+## Eligibility blocker
+
+Self-serve SIWC client IDs are only for open-source, locally hosted apps (loopback
+redirect `http://127.0.0.1:{port}/callback`). Paid or remotely hosted bots must apply
+through the interest form, ,
+which is for a "select group of commercial partners". The client ID and the partner
+redirect rules come with approval. Until then nothing here can run against OpenAI, so
+every endpoint, scope and redirect is configurable.
+
+## What OpenAI allows on the plan path
+
+- OAuth 2 auth-code + PKCE (S256), OIDC. Access token 1 h; refresh token 30 days,
+ rolling on each refresh. No client secret.
+- Inference: Responses API only (`POST {api_base}/responses`) with `store: false` and
+ `stream: true`. Success only after `response.completed`.
+- Not allowed: `temperature`, `top_p`, `max_output_tokens`, `metadata`, `user`,
+ `previous_response_id`, background mode, hosted tools, function calling, audio/file
+ inputs, transcription. `respond()` strips these params before sending.
+- Models: per-user catalog, `GET {api_base}/models`, entries with `visibility == "list"`.
+
+## Config (env prefix `BOTSPOT_CHATGPT_PLAN_`)
+
+| Var | Default | Notes |
+|---|---|---|
+| `ENABLED` | `false` | master flag |
+| `CLIENT_ID` | | issued by OpenAI on partner approval |
+| `REDIRECT_URI` | | our public HTTPS callback; its path is the mounted route (default `/api/chatgpt/callback`) |
+| `TOKEN_ENCRYPTION_KEY` | | Fernet key: `python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"` |
+| `ISSUER` | `https://auth.openai.com` | |
+| `AUTHORIZE_PATH` | `/api/accounts/authorize` | |
+| `TOKEN_PATH` | `/api/accounts/oauth/token` | |
+| `RESOURCE` | `https://api.openai.com/v1` | sent on authorize, code exchange and refresh |
+| `API_BASE` | `https://api.openai.com/v1` | |
+| `SCOPES` | `openid profile email offline_access resource.invoke chatgpt.tokens.use.direct` | `chatgpt.tokens.use.direct` must be granted |
+| `AGENT_NAME_HINT` | | sent on authorize only when set |
+| `PENDING_TTL_SECONDS` | `600` | link state lifetime (Mongo TTL index) |
+| `REFRESH_MARGIN_SECONDS` | `300` | refresh when the access token expires within this |
+| `UNAVAILABLE_RETRIES` | `2` | bounded backoff on 503 / unavailable |
+| `UNAVAILABLE_BACKOFF_SECONDS` | `2.0` | doubles per retry |
+| `CAP_PAUSE_SECONDS` | `3600` | pause after a 429 that carries no reset hint |
+| `REQUEST_TIMEOUT_SECONDS` | `120` | |
+| `RETURN_URL` | | optional link on the callback page (e.g. `https://t.me/`) |
+
+Requires `BOTSPOT_MONGO_DATABASE_ENABLED=true` and the `cryptography` package.
+Collections: `chatgpt_oauth_pending` (TTL on `expires_at`), `chatgpt_links` (keyed by
+Telegram user id; refresh and access tokens are Fernet-encrypted). Tokens are never logged.
+
+## Usage
+
+```python
+from botspot.components.new import chatgpt_plan
+from botspot.components.new.chatgpt_plan import PlanNeedsRelink, PlanUnavailable, PlanUsageCapped
+
+url = await chatgpt_plan.start_link(user_id) # send as an inline URL button
+try:
+ answer = await chatgpt_plan.respond(user_id, "gpt-6-astra", text, instructions=system)
+except (PlanUsageCapped, PlanNeedsRelink, PlanUnavailable):
+ ... # fall back to the normal paid path
+
+# web server (only when enabled)
+app.add_routes(chatgpt_plan.aiohttp_routes(chatgpt_plan.get_chatgpt_plan(), on_linked=notify_user))
+```
+
+## Error mapping
+
+| OpenAI | Exception | Component action |
+|---|---|---|
+| 429 / `subscription_sharing_usage_limit_exceeded` | `PlanUsageCapped(reset_hint)` | link status `capped` until the hint (or `CAP_PAUSE_SECONDS`) |
+| 401 / `invalid_grant` / `invalid_refresh_token` / `token_expired` | `PlanNeedsRelink` | tokens cleared, status `needs_relink` |
+| 503 / `subscription_sharing_usage_unavailable` | `PlanUnavailable` | bounded backoff, then raise |
+| 403 (policy / region) | `PlanUnavailable(retryable=False)` | raise, no retry |
+| `subscription_sharing_unsupported_capability` | `PlanUnsupported(param)` | drop `error.param` and retry; raise if it can't be dropped |
+| no link | `PlanNotLinked` | |
+| flag off | `PlanDisabled` | no I/O |
+
+## Untested until OpenAI issues a client ID
+
+All HTTP in the tests is faked. These parts are written from the SIWC docs and have
+never talked to OpenAI:
+
+- the real authorize screen, partner redirect rules, and whether `resource` /
+ `agent_name_hint` are accepted on each request;
+- the token endpoint's exact response shape (`scope`, `id_token`, rolling refresh);
+- **id_token signature verification is NOT implemented** (TODO in
+ `oauth.verify_id_token_signature`, JWKS). We check `iss`, `aud`, `nonce`, `sub` and
+ rely on the token arriving straight from the token endpoint over TLS. Tests inject a
+ verifier via `ChatgptPlan(..., id_token_verifier=...)`;
+- real Responses stream events, error bodies and the 429 reset hint format;
+- `unlink` only deletes our stored tokens; it does not revoke them at OpenAI.
diff --git a/botspot/components/new/chatgpt_plan/__init__.py b/botspot/components/new/chatgpt_plan/__init__.py
new file mode 100644
index 0000000..1d6756a
--- /dev/null
+++ b/botspot/components/new/chatgpt_plan/__init__.py
@@ -0,0 +1,137 @@
+"""botspot chatgpt_plan component: users fund LLM calls with their own ChatGPT plan.
+
+"Sign in with ChatGPT" (OAuth 2 + PKCE) links a Telegram user to their ChatGPT plan;
+``respond`` then runs on the Responses API with the user's token. Off by default
+(``BOTSPOT_CHATGPT_PLAN_ENABLED``); with the flag off every call raises
+``PlanDisabled`` and does no I/O. Needs an OpenAI-issued client ID (see README.md).
+"""
+
+from __future__ import annotations
+
+from typing import Any, Optional
+
+from botspot.components.new.chatgpt_plan.client import PlanResponse, ResponseInput
+from botspot.components.new.chatgpt_plan.errors import (
+ PlanDisabled,
+ PlanError,
+ PlanLinkError,
+ PlanNeedsRelink,
+ PlanNotLinked,
+ PlanUnavailable,
+ PlanUnsupported,
+ PlanUsageCapped,
+)
+from botspot.components.new.chatgpt_plan.oauth import LinkResult
+from botspot.components.new.chatgpt_plan.plan import ChatgptPlan
+from botspot.components.new.chatgpt_plan.settings import ChatgptPlanSettings
+from botspot.components.new.chatgpt_plan.store import PlanLink
+from botspot.components.new.chatgpt_plan.web import aiohttp_routes, make_callback_handler
+from botspot.utils.internal import get_logger
+
+logger = get_logger()
+
+
+def initialize(settings: ChatgptPlanSettings) -> ChatgptPlan:
+ """Initialize the component. Requires mongo_database, client_id and an encryption key."""
+ from botspot.core.dependency_manager import get_dependency_manager
+ from botspot.core.errors import ConfigurationError
+
+ deps = get_dependency_manager()
+ if not deps.botspot_settings.mongo_database.enabled:
+ raise ConfigurationError(
+ "MongoDB is required for chatgpt_plan. Set BOTSPOT_MONGO_DATABASE_ENABLED=true."
+ )
+ if not settings.client_id or not settings.redirect_uri:
+ raise ConfigurationError(
+ "chatgpt_plan needs BOTSPOT_CHATGPT_PLAN_CLIENT_ID and _REDIRECT_URI "
+ "(client ID is issued by OpenAI to approved partners)."
+ )
+ from botspot.components.data.mongo_database import get_database
+
+ plan = ChatgptPlan(settings, db=get_database())
+ logger.info("chatgpt_plan initialized")
+ return plan
+
+
+def get_chatgpt_plan() -> ChatgptPlan:
+ """The initialized component. Raises ``PlanDisabled`` when the flag is off."""
+ from botspot.core.dependency_manager import DependencyManager
+
+ if not DependencyManager.is_initialized():
+ raise PlanDisabled()
+ plan = DependencyManager().chatgpt_plan
+ if plan is None or not plan.settings.enabled:
+ raise PlanDisabled()
+ return plan
+
+
+async def start_link(user_id: int) -> str:
+ """Return the "Sign in with ChatGPT" URL for this Telegram user."""
+ return await get_chatgpt_plan().start_link(user_id)
+
+
+async def complete_link(state: str, code: str) -> LinkResult:
+ """Finish the link from the OAuth callback (``state`` is single use)."""
+ return await get_chatgpt_plan().complete_link(state, code)
+
+
+async def get_access_token(user_id: int) -> str:
+ """A valid access token for the user, refreshed when near expiry."""
+ return await get_chatgpt_plan().get_access_token(user_id)
+
+
+async def get_link(user_id: int) -> Optional[PlanLink]:
+ """Link state (no tokens) for /settings, or None."""
+ return await get_chatgpt_plan().get_link(user_id)
+
+
+async def unlink(user_id: int) -> bool:
+ """Forget the user's tokens. True if a link existed."""
+ return await get_chatgpt_plan().unlink(user_id)
+
+
+async def respond(
+ user_id: int,
+ model: str,
+ input: ResponseInput,
+ *,
+ instructions: Optional[str] = None,
+ **params: Any,
+) -> PlanResponse:
+ """Run one Responses call on the user's plan (``store=False``, ``stream=True``)."""
+ return await get_chatgpt_plan().respond(
+ user_id, model, input, instructions=instructions, **params
+ )
+
+
+async def list_models(user_id: int) -> list[str]:
+ """Model ids available on the user's plan."""
+ return await get_chatgpt_plan().list_models(user_id)
+
+
+__all__ = [
+ "ChatgptPlanSettings",
+ "ChatgptPlan",
+ "LinkResult",
+ "PlanLink",
+ "PlanResponse",
+ "PlanError",
+ "PlanDisabled",
+ "PlanNotLinked",
+ "PlanNeedsRelink",
+ "PlanUsageCapped",
+ "PlanUnavailable",
+ "PlanUnsupported",
+ "PlanLinkError",
+ "initialize",
+ "get_chatgpt_plan",
+ "start_link",
+ "complete_link",
+ "get_access_token",
+ "get_link",
+ "unlink",
+ "respond",
+ "list_models",
+ "aiohttp_routes",
+ "make_callback_handler",
+]
diff --git a/botspot/components/new/chatgpt_plan/client.py b/botspot/components/new/chatgpt_plan/client.py
new file mode 100644
index 0000000..4c06861
--- /dev/null
+++ b/botspot/components/new/chatgpt_plan/client.py
@@ -0,0 +1,170 @@
+"""Plan-funded inference: Responses API only, ``store=False`` + ``stream=True``."""
+
+from __future__ import annotations
+
+import asyncio
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING, Any, Optional, Union
+
+from botspot.components.new.chatgpt_plan.errors import (
+ PlanNeedsRelink,
+ PlanUnavailable,
+ PlanUnsupported,
+ PlanUsageCapped,
+ error_from_response,
+)
+from botspot.components.new.chatgpt_plan.settings import (
+ DISALLOWED_PARAMS,
+ REQUIRED_PARAMS,
+ ChatgptPlanSettings,
+)
+from botspot.utils.internal import get_logger
+
+if TYPE_CHECKING:
+ from botspot.components.new.chatgpt_plan.http import HttpResult, Transport
+ from botspot.components.new.chatgpt_plan.oauth import OAuthClient
+ from botspot.components.new.chatgpt_plan.store import LinkStore
+
+logger = get_logger()
+
+ResponseInput = Union[str, list[dict[str, Any]]]
+MAX_PARAM_DROPS = 3
+
+
+@dataclass
+class PlanResponse:
+ text: str
+ model: str
+ usage: dict = field(default_factory=dict)
+
+
+def build_payload(
+ model: str, input: ResponseInput, instructions: Optional[str], params: dict
+) -> dict:
+ """Build a Responses payload, stripping params the plan path rejects."""
+ dropped = sorted(k for k in params if k in DISALLOWED_PARAMS)
+ if dropped:
+ logger.debug(f"chatgpt_plan: dropping unsupported params {dropped}")
+ payload = {k: v for k, v in params.items() if k not in DISALLOWED_PARAMS}
+ payload.update(model=model, input=input, store=False, stream=True)
+ if instructions:
+ payload["instructions"] = instructions
+ return payload
+
+
+def _output_text(response: dict) -> str:
+ """Fallback: collect output_text parts from a completed response object."""
+ parts = []
+ for item in response.get("output") or []:
+ for content in item.get("content") or []:
+ if content.get("type") == "output_text":
+ parts.append(content.get("text", ""))
+ return "".join(parts)
+
+
+def parse_stream(events: list[dict], requested_model: str) -> PlanResponse:
+ """Accumulate output text. Success only after ``response.completed``."""
+ deltas: list[str] = []
+ for event in events:
+ kind = event.get("type")
+ if kind == "response.output_text.delta":
+ deltas.append(event.get("delta", ""))
+ elif kind in ("error", "response.failed"):
+ body = event.get("response") or {"error": event.get("error") or event}
+ raise error_from_response(200, body)
+ elif kind == "response.completed":
+ response = event.get("response") or {}
+ return PlanResponse(
+ text="".join(deltas) or _output_text(response),
+ model=response.get("model") or requested_model,
+ usage=response.get("usage") or {},
+ )
+ raise PlanUnavailable("stream ended without response.completed")
+
+
+def drop_param(payload: dict, param: Optional[str]) -> bool:
+ """Drop the param named by ``unsupported_capability``. False if it can't be dropped."""
+ if not param or param in REQUIRED_PARAMS or param not in payload:
+ return False
+ del payload[param]
+ logger.info(f"chatgpt_plan: dropped param '{param}' after unsupported_capability")
+ return True
+
+
+class ResponsesClient:
+ def __init__(
+ self,
+ settings: ChatgptPlanSettings,
+ transport: Transport,
+ store: LinkStore,
+ oauth: OAuthClient,
+ ):
+ self.settings = settings
+ self.transport = transport
+ self.store = store
+ self.oauth = oauth
+
+ def _url(self, path: str) -> str:
+ return self.settings.api_base.rstrip("/") + path
+
+ async def _check(self, user_id: int, result: HttpResult) -> None:
+ if result.ok:
+ return
+ error = error_from_response(result.status, result.body, result.headers)
+ await self._record(user_id, error)
+ raise error
+
+ async def _record(self, user_id: int, error: Exception) -> None:
+ if isinstance(error, PlanNeedsRelink):
+ await self.store.mark_needs_relink(user_id)
+ elif isinstance(error, PlanUsageCapped):
+ await self.store.mark_capped(user_id, error)
+
+ async def _respond_once(self, user_id: int, payload: dict) -> PlanResponse:
+ token = await self.store.get_access_token(user_id, self.oauth)
+ result = await self.transport.post_sse(self._url("/responses"), token, payload)
+ await self._check(user_id, result)
+ try:
+ return parse_stream(result.events, payload["model"])
+ except (PlanNeedsRelink, PlanUsageCapped) as error:
+ await self._record(user_id, error)
+ raise
+
+ async def _backoff(self, attempt: int, error: PlanUnavailable) -> None:
+ if not error.retryable or attempt >= self.settings.unavailable_retries:
+ raise error
+ await asyncio.sleep(self.settings.unavailable_backoff_seconds * 2**attempt)
+
+ async def respond(
+ self,
+ user_id: int,
+ model: str,
+ input: ResponseInput,
+ *,
+ instructions: Optional[str] = None,
+ **params: Any,
+ ) -> PlanResponse:
+ payload = build_payload(model, input, instructions, params)
+ unavailable = drops = 0
+ while True:
+ try:
+ response = await self._respond_once(user_id, payload)
+ except PlanUnsupported as error:
+ drops += 1
+ if drops > MAX_PARAM_DROPS or not drop_param(payload, error.param):
+ raise
+ continue
+ except PlanUnavailable as error:
+ await self._backoff(unavailable, error)
+ unavailable += 1
+ continue
+ await self.store.touch(user_id)
+ return response
+
+ async def list_models(self, user_id: int) -> list[str]:
+ """Model ids from the user's catalog with ``visibility == "list"``."""
+ token = await self.store.get_access_token(user_id, self.oauth)
+ result = await self.transport.get_json(self._url("/models"), token)
+ await self._check(user_id, result)
+ models = (result.body or {}).get("data") or []
+ return [m["id"] for m in models if m.get("visibility") == "list" and m.get("id")]
diff --git a/botspot/components/new/chatgpt_plan/errors.py b/botspot/components/new/chatgpt_plan/errors.py
new file mode 100644
index 0000000..fabd034
--- /dev/null
+++ b/botspot/components/new/chatgpt_plan/errors.py
@@ -0,0 +1,153 @@
+"""Plan errors and the OpenAI error -> exception mapping table.
+
+Host bots fall back to their own paid path on ``PlanUsageCapped``,
+``PlanNeedsRelink`` and ``PlanUnavailable``.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Callable, Mapping, Optional
+
+from botspot.core.errors import BotspotError
+
+
+class PlanError(BotspotError):
+ """Base for chatgpt_plan errors. ``code`` is the OpenAI error code, if any."""
+
+ def __init__(self, message: str, *, code: str = "", report_to_dev: bool = False):
+ super().__init__(message, report_to_dev=report_to_dev)
+ self.args = (message,)
+ self.code = code
+
+
+class PlanDisabled(PlanError):
+ """The component flag is off. Raised before any I/O."""
+
+ def __init__(self, message: str = "chatgpt_plan is disabled"):
+ super().__init__(message)
+
+
+class PlanNotLinked(PlanError):
+ """The user has not linked a ChatGPT plan."""
+
+ def __init__(self, message: str = "no ChatGPT plan linked"):
+ super().__init__(message)
+
+
+class PlanNeedsRelink(PlanError):
+ """Tokens are revoked/expired; the user must link again."""
+
+
+class PlanUsageCapped(PlanError):
+ """The user hit their weekly per-app cap (ChatGPT settings -> Usage)."""
+
+ def __init__(self, message: str, *, reset_hint: Optional[str] = None, code: str = ""):
+ super().__init__(message, code=code)
+ self.reset_hint = reset_hint
+
+
+class PlanUnavailable(PlanError):
+ """Plan inference is unavailable. ``retryable`` is False for 403 policy/region."""
+
+ def __init__(self, message: str, *, code: str = "", retryable: bool = True):
+ super().__init__(message, code=code)
+ self.retryable = retryable
+
+
+class PlanUnsupported(PlanError):
+ """The plan path rejected a request param (named in ``param``)."""
+
+ def __init__(self, message: str, *, param: Optional[str] = None, code: str = ""):
+ super().__init__(message, code=code)
+ self.param = param
+
+
+class PlanLinkError(PlanError):
+ """The link flow failed: bad/expired state, missing scope, failed code exchange."""
+
+
+def _error_fields(body: Any) -> tuple[str, str, Optional[str]]:
+ """Return (code, message, param) from an API (`{"error": {...}}`) or OAuth body."""
+ if not isinstance(body, Mapping):
+ return "", "", None
+ err = body.get("error")
+ if isinstance(err, Mapping):
+ code = str(err.get("code") or err.get("type") or "")
+ return code, str(err.get("message") or ""), err.get("param")
+ # OAuth token endpoint shape: {"error": "invalid_grant", "error_description": "..."}
+ return str(err or ""), str(body.get("error_description") or ""), None
+
+
+def _reset_hint(body: Any, headers: Mapping[str, str]) -> Optional[str]:
+ err = body.get("error") if isinstance(body, Mapping) else None
+ if isinstance(err, Mapping) and err.get("resets_at"):
+ return str(err["resets_at"])
+ return headers.get("retry-after") or headers.get("Retry-After")
+
+
+def _capped(msg: str, code: str, param, hint) -> PlanError:
+ return PlanUsageCapped(msg or "ChatGPT plan usage cap reached", reset_hint=hint, code=code)
+
+
+def _relink(msg: str, code: str, param, hint) -> PlanError:
+ return PlanNeedsRelink(msg or "ChatGPT link expired, relink needed", code=code)
+
+
+def _unavailable(msg: str, code: str, param, hint) -> PlanError:
+ return PlanUnavailable(msg or "ChatGPT plan unavailable", code=code)
+
+
+def _forbidden(msg: str, code: str, param, hint) -> PlanError:
+ return PlanUnavailable(
+ msg or "ChatGPT plan refused (policy/region)", code=code, retryable=False
+ )
+
+
+def _unsupported(msg: str, code: str, param, hint) -> PlanError:
+ return PlanUnsupported(msg or f"unsupported param: {param}", param=param, code=code)
+
+
+_Factory = Callable[[str, str, Optional[str], Optional[str]], PlanError]
+
+# Error code -> exception. Codes win over HTTP status.
+CODE_ERRORS: dict[str, _Factory] = {
+ "subscription_sharing_usage_limit_exceeded": _capped,
+ "subscription_sharing_usage_unavailable": _unavailable,
+ "subscription_sharing_unsupported_capability": _unsupported,
+ "invalid_grant": _relink,
+ "invalid_refresh_token": _relink,
+ "token_expired": _relink,
+}
+
+# HTTP status -> exception, when the code is unknown (401 also covers revoked tokens).
+STATUS_ERRORS: dict[int, _Factory] = {
+ 401: _relink,
+ 403: _forbidden,
+ 429: _capped,
+ 503: _unavailable,
+}
+
+
+def error_from_response(
+ status: int, body: Any, headers: Optional[Mapping[str, str]] = None
+) -> PlanError:
+ """Map an OpenAI error response (or stream error event) to a ``PlanError``."""
+ code, message, param = _error_fields(body)
+ hint = _reset_hint(body, headers or {})
+ factory = CODE_ERRORS.get(code) or STATUS_ERRORS.get(status) or _unavailable
+ return factory(message, code, param, hint)
+
+
+__all__ = [
+ "PlanError",
+ "PlanDisabled",
+ "PlanNotLinked",
+ "PlanNeedsRelink",
+ "PlanUsageCapped",
+ "PlanUnavailable",
+ "PlanUnsupported",
+ "PlanLinkError",
+ "CODE_ERRORS",
+ "STATUS_ERRORS",
+ "error_from_response",
+]
diff --git a/botspot/components/new/chatgpt_plan/http.py b/botspot/components/new/chatgpt_plan/http.py
new file mode 100644
index 0000000..dd9ce4e
--- /dev/null
+++ b/botspot/components/new/chatgpt_plan/http.py
@@ -0,0 +1,85 @@
+"""Thin aiohttp transport. Tests swap it for a fake; nothing here logs headers or bodies."""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass, field
+from typing import Any, Iterable, Mapping, Protocol
+
+
+@dataclass
+class HttpResult:
+ status: int
+ body: Any = None # parsed JSON (or None)
+ events: list[dict] = field(default_factory=list) # parsed SSE events, 2xx streams only
+ headers: Mapping[str, str] = field(default_factory=dict)
+
+ @property
+ def ok(self) -> bool:
+ return 200 <= self.status < 300
+
+
+class Transport(Protocol):
+ async def post_form(self, url: str, form: Mapping[str, str]) -> HttpResult: ...
+
+ async def get_json(self, url: str, token: str) -> HttpResult: ...
+
+ async def post_sse(self, url: str, token: str, payload: Mapping[str, Any]) -> HttpResult: ...
+
+
+def parse_sse(lines: Iterable[str]) -> list[dict]:
+ """Parse ``data:`` lines of a server-sent event stream into JSON events."""
+ events: list[dict] = []
+ data: list[str] = []
+ for raw in list(lines) + [""]:
+ line = raw.rstrip("\r\n")
+ if line.startswith("data:"):
+ data.append(line[5:].lstrip())
+ elif not line and data:
+ chunk = "\n".join(data)
+ data = []
+ if chunk != "[DONE]":
+ events.append(json.loads(chunk))
+ return events
+
+
+def _json_or_none(text: str) -> Any:
+ try:
+ return json.loads(text)
+ except ValueError:
+ return None
+
+
+class AiohttpTransport:
+ def __init__(self, timeout_seconds: float = 120.0):
+ self.timeout_seconds = timeout_seconds
+
+ def _session(self):
+ import aiohttp
+
+ return aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=self.timeout_seconds))
+
+ async def post_form(self, url: str, form: Mapping[str, str]) -> HttpResult:
+ async with self._session() as session:
+ async with session.post(url, data=dict(form)) as resp:
+ text = await resp.text()
+ return HttpResult(resp.status, _json_or_none(text), headers=dict(resp.headers))
+
+ async def get_json(self, url: str, token: str) -> HttpResult:
+ headers = {"Authorization": f"Bearer {token}"}
+ async with self._session() as session:
+ async with session.get(url, headers=headers) as resp:
+ text = await resp.text()
+ return HttpResult(resp.status, _json_or_none(text), headers=dict(resp.headers))
+
+ async def post_sse(self, url: str, token: str, payload: Mapping[str, Any]) -> HttpResult:
+ headers = {"Authorization": f"Bearer {token}", "Accept": "text/event-stream"}
+ async with self._session() as session:
+ async with session.post(url, json=dict(payload), headers=headers) as resp:
+ text = await resp.text()
+ result = HttpResult(resp.status, headers=dict(resp.headers))
+ if result.ok:
+ result.events = parse_sse(text.splitlines())
+ else:
+ result.body = _json_or_none(text)
+ return result
diff --git a/botspot/components/new/chatgpt_plan/oauth.py b/botspot/components/new/chatgpt_plan/oauth.py
new file mode 100644
index 0000000..3a89571
--- /dev/null
+++ b/botspot/components/new/chatgpt_plan/oauth.py
@@ -0,0 +1,195 @@
+"""Sign in with ChatGPT: OAuth 2 auth-code + PKCE (S256), OIDC id_token claims."""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import json
+import secrets
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Awaitable, Callable, Optional
+from urllib.parse import urlencode
+
+from botspot.components.new.chatgpt_plan.errors import (
+ PlanLinkError,
+ PlanUnavailable,
+ error_from_response,
+)
+from botspot.components.new.chatgpt_plan.settings import REQUIRED_SCOPE, ChatgptPlanSettings
+from botspot.utils.internal import get_logger
+
+if TYPE_CHECKING:
+ from botspot.components.new.chatgpt_plan.http import Transport
+ from botspot.components.new.chatgpt_plan.store import LinkStore
+
+logger = get_logger()
+
+# (id_token, settings) -> None; raises PlanLinkError when the signature is invalid.
+IdTokenVerifier = Callable[[str, ChatgptPlanSettings], Awaitable[None]]
+
+
+@dataclass
+class TokenSet:
+ access_token: str
+ expires_in: int
+ refresh_token: str = ""
+ id_token: str = ""
+ scope: str = ""
+
+
+@dataclass
+class LinkResult:
+ user_id: int
+ sub: str
+ email: str
+ scopes: str
+
+
+def _b64url(data: bytes) -> str:
+ return base64.urlsafe_b64encode(data).rstrip(b"=").decode()
+
+
+def pkce_pair() -> tuple[str, str]:
+ """Return (code_verifier, S256 code_challenge)."""
+ verifier = _b64url(secrets.token_bytes(32))
+ challenge = _b64url(hashlib.sha256(verifier.encode()).digest())
+ return verifier, challenge
+
+
+def build_authorize_url(
+ settings: ChatgptPlanSettings, *, state: str, nonce: str, code_challenge: str
+) -> str:
+ params = {
+ "response_type": "code",
+ "client_id": settings.client_id,
+ "redirect_uri": settings.redirect_uri,
+ "scope": settings.scopes,
+ "state": state,
+ "nonce": nonce,
+ "code_challenge": code_challenge,
+ "code_challenge_method": "S256",
+ "resource": settings.resource,
+ }
+ if settings.agent_name_hint:
+ params["agent_name_hint"] = settings.agent_name_hint
+ return f"{settings.authorize_url}?{urlencode(params)}"
+
+
+def decode_id_token_claims(id_token: str) -> dict:
+ """Decode the JWT payload WITHOUT verifying its signature.
+
+ The signature is checked separately by ``verify_id_token_signature``.
+ """
+ parts = id_token.split(".")
+ if len(parts) != 3:
+ raise PlanLinkError("malformed id_token")
+ payload = parts[1] + "=" * (-len(parts[1]) % 4)
+ return json.loads(base64.urlsafe_b64decode(payload))
+
+
+async def verify_id_token_signature(id_token: str, settings: ChatgptPlanSettings) -> None:
+ """TODO(siwc-client-id): verify the id_token signature against the issuer JWKS.
+
+ NOT IMPLEMENTED. Until it is, we rely on the id_token arriving straight from the
+ token endpoint over TLS (OIDC Core 3.1.3.7 allows this for the code flow) plus the
+ iss / aud / nonce checks in ``_check_claims``. Implement once OpenAI issues a client
+ ID and the JWKS URL / signing algs can be tested. Tests inject a verifier through
+ ``OAuthClient(id_token_verifier=...)``.
+ """
+ logger.warning("chatgpt_plan: id_token signature NOT verified (JWKS TODO)")
+
+
+def _check_claims(claims: dict, settings: ChatgptPlanSettings, nonce: str) -> None:
+ aud = claims.get("aud")
+ audiences = aud if isinstance(aud, list) else [aud]
+ if claims.get("iss", "").rstrip("/") != settings.issuer.rstrip("/"):
+ raise PlanLinkError("id_token issuer mismatch")
+ if settings.client_id not in audiences:
+ raise PlanLinkError("id_token audience mismatch")
+ if claims.get("nonce") != nonce:
+ raise PlanLinkError("id_token nonce mismatch")
+ if not claims.get("sub"):
+ raise PlanLinkError("id_token has no sub")
+
+
+def _token_set(body: dict, requested_scope: str) -> TokenSet:
+ if not body.get("access_token"):
+ raise PlanUnavailable("token endpoint returned no access_token")
+ return TokenSet(
+ access_token=body["access_token"],
+ expires_in=int(body.get("expires_in") or 3600),
+ refresh_token=body.get("refresh_token") or "",
+ id_token=body.get("id_token") or "",
+ # RFC 6749 5.1: an omitted scope means "as requested".
+ scope=body.get("scope") or requested_scope,
+ )
+
+
+class OAuthClient:
+ def __init__(
+ self,
+ settings: ChatgptPlanSettings,
+ transport: Transport,
+ store: LinkStore,
+ id_token_verifier: Optional[IdTokenVerifier] = None,
+ ):
+ self.settings = settings
+ self.transport = transport
+ self.store = store
+ self.id_token_verifier = id_token_verifier or verify_id_token_signature
+
+ async def start_link(self, user_id: int) -> str:
+ """Create a single-use pending state and return the authorize URL."""
+ verifier, challenge = pkce_pair()
+ state, nonce = secrets.token_urlsafe(32), secrets.token_urlsafe(32)
+ await self.store.save_pending(state, user_id, verifier, nonce)
+ logger.info(f"chatgpt_plan: link started for user {user_id}")
+ return build_authorize_url(
+ self.settings, state=state, nonce=nonce, code_challenge=challenge
+ )
+
+ async def complete_link(self, state: str, code: str) -> LinkResult:
+ """Validate state, exchange the code, check scope + id_token, save the link."""
+ pending = await self.store.pop_pending(state)
+ if pending is None:
+ raise PlanLinkError("unknown, used or expired link state")
+ tokens = await self._exchange_code(code, pending["code_verifier"])
+ if REQUIRED_SCOPE not in tokens.scope.split():
+ raise PlanLinkError(f"granted scope lacks {REQUIRED_SCOPE}")
+ if not tokens.refresh_token or not tokens.id_token:
+ raise PlanLinkError("token response lacks refresh_token or id_token")
+ await self.id_token_verifier(tokens.id_token, self.settings)
+ claims = decode_id_token_claims(tokens.id_token)
+ _check_claims(claims, self.settings, pending["nonce"])
+ user_id = pending["user_id"]
+ email = claims.get("email", "")
+ await self.store.save_link(user_id, tokens, claims["sub"], email)
+ return LinkResult(user_id=user_id, sub=claims["sub"], email=email, scopes=tokens.scope)
+
+ async def _exchange_code(self, code: str, verifier: str) -> TokenSet:
+ form = {
+ "grant_type": "authorization_code",
+ "code": code,
+ "redirect_uri": self.settings.redirect_uri,
+ "client_id": self.settings.client_id,
+ "code_verifier": verifier,
+ "resource": self.settings.resource,
+ }
+ result = await self.transport.post_form(self.settings.token_url, form)
+ if not result.ok:
+ err = error_from_response(result.status, result.body, result.headers)
+ raise PlanLinkError(f"code exchange failed: {err.code or result.status}")
+ return _token_set(result.body or {}, self.settings.scopes)
+
+ async def refresh(self, refresh_token: str) -> TokenSet:
+ """Refresh grant. Raises ``PlanNeedsRelink`` on invalid_grant / 401."""
+ form = {
+ "grant_type": "refresh_token",
+ "refresh_token": refresh_token,
+ "client_id": self.settings.client_id,
+ "resource": self.settings.resource,
+ }
+ result = await self.transport.post_form(self.settings.token_url, form)
+ if not result.ok:
+ raise error_from_response(result.status, result.body, result.headers)
+ return _token_set(result.body or {}, self.settings.scopes)
diff --git a/botspot/components/new/chatgpt_plan/plan.py b/botspot/components/new/chatgpt_plan/plan.py
new file mode 100644
index 0000000..f1f9a37
--- /dev/null
+++ b/botspot/components/new/chatgpt_plan/plan.py
@@ -0,0 +1,77 @@
+"""ChatgptPlan: the component object. Every method raises ``PlanDisabled`` before I/O
+when the flag is off."""
+
+from __future__ import annotations
+
+from typing import Any, Optional
+
+from botspot.components.new.chatgpt_plan.client import (
+ PlanResponse,
+ ResponseInput,
+ ResponsesClient,
+)
+from botspot.components.new.chatgpt_plan.errors import PlanDisabled
+from botspot.components.new.chatgpt_plan.http import AiohttpTransport, Transport
+from botspot.components.new.chatgpt_plan.oauth import IdTokenVerifier, LinkResult, OAuthClient
+from botspot.components.new.chatgpt_plan.settings import ChatgptPlanSettings
+from botspot.components.new.chatgpt_plan.store import LinkStore, PlanLink, TokenCipher
+
+
+class ChatgptPlan:
+ def __init__(
+ self,
+ settings: ChatgptPlanSettings,
+ *,
+ db: Any,
+ transport: Optional[Transport] = None,
+ id_token_verifier: Optional[IdTokenVerifier] = None,
+ ):
+ self.settings = settings
+ if not settings.enabled:
+ return # nothing is built: no cipher, no collections, no sessions
+ transport = transport or AiohttpTransport(settings.request_timeout_seconds)
+ key = settings.token_encryption_key
+ cipher = TokenCipher(key.get_secret_value() if key else "")
+ self.store = LinkStore(settings, db, cipher)
+ self.oauth = OAuthClient(settings, transport, self.store, id_token_verifier)
+ self.client = ResponsesClient(settings, transport, self.store, self.oauth)
+
+ def _require_enabled(self) -> None:
+ if not self.settings.enabled:
+ raise PlanDisabled()
+
+ async def start_link(self, user_id: int) -> str:
+ self._require_enabled()
+ return await self.oauth.start_link(user_id)
+
+ async def complete_link(self, state: str, code: str) -> LinkResult:
+ self._require_enabled()
+ return await self.oauth.complete_link(state, code)
+
+ async def get_access_token(self, user_id: int) -> str:
+ self._require_enabled()
+ return await self.store.get_access_token(user_id, self.oauth)
+
+ async def get_link(self, user_id: int) -> Optional[PlanLink]:
+ self._require_enabled()
+ return await self.store.get_link(user_id)
+
+ async def unlink(self, user_id: int) -> bool:
+ self._require_enabled()
+ return await self.store.unlink(user_id)
+
+ async def respond(
+ self,
+ user_id: int,
+ model: str,
+ input: ResponseInput,
+ *,
+ instructions: Optional[str] = None,
+ **params: Any,
+ ) -> PlanResponse:
+ self._require_enabled()
+ return await self.client.respond(user_id, model, input, instructions=instructions, **params)
+
+ async def list_models(self, user_id: int) -> list[str]:
+ self._require_enabled()
+ return await self.client.list_models(user_id)
diff --git a/botspot/components/new/chatgpt_plan/settings.py b/botspot/components/new/chatgpt_plan/settings.py
new file mode 100644
index 0000000..4c98d45
--- /dev/null
+++ b/botspot/components/new/chatgpt_plan/settings.py
@@ -0,0 +1,85 @@
+"""Settings for the chatgpt_plan component ("Sign in with ChatGPT")."""
+
+from __future__ import annotations
+
+from urllib.parse import urlparse
+
+from pydantic import SecretStr
+from pydantic_settings import BaseSettings
+
+DEFAULT_SCOPES = "openid profile email offline_access resource.invoke chatgpt.tokens.use.direct"
+REQUIRED_SCOPE = "chatgpt.tokens.use.direct"
+DEFAULT_CALLBACK_PATH = "/api/chatgpt/callback"
+
+LINKS_COLLECTION = "chatgpt_links"
+PENDING_COLLECTION = "chatgpt_oauth_pending"
+
+STATUS_ACTIVE = "active"
+STATUS_NEEDS_RELINK = "needs_relink"
+STATUS_CAPPED = "capped"
+
+# Plan-funded inference rejects these (Responses API, SIWC preview limitations).
+DISALLOWED_PARAMS = frozenset(
+ {
+ "temperature",
+ "top_p",
+ "max_output_tokens",
+ "max_tokens",
+ "max_completion_tokens",
+ "metadata",
+ "user",
+ "previous_response_id",
+ "background",
+ "tools",
+ "tool_choice",
+ "parallel_tool_calls",
+ }
+)
+# Params the component sets itself; never dropped on `unsupported_capability`.
+REQUIRED_PARAMS = frozenset({"model", "input", "store", "stream"})
+
+
+class ChatgptPlanSettings(BaseSettings):
+ """Env prefix ``BOTSPOT_CHATGPT_PLAN_``.
+
+ Off by default. With ``ENABLED=false`` every call raises ``PlanDisabled`` and
+ does no I/O. A client ID comes only from OpenAI partner approval, so every
+ endpoint, scope and redirect stays configurable.
+ """
+
+ enabled: bool = False
+ client_id: str = ""
+ issuer: str = "https://auth.openai.com"
+ authorize_path: str = "/api/accounts/authorize"
+ token_path: str = "/api/accounts/oauth/token"
+ resource: str = "https://api.openai.com/v1"
+ api_base: str = "https://api.openai.com/v1"
+ scopes: str = DEFAULT_SCOPES
+ redirect_uri: str = "" # our public HTTPS callback; partner value TBD
+ agent_name_hint: str = "" # sent on authorize only when set
+ token_encryption_key: SecretStr | None = None # Fernet key
+ pending_ttl_seconds: int = 600
+ refresh_margin_seconds: int = 300
+ unavailable_retries: int = 2
+ unavailable_backoff_seconds: float = 2.0
+ cap_pause_seconds: int = 3600 # used when a 429 carries no reset hint
+ request_timeout_seconds: float = 120.0
+ return_url: str = "" # optional "back to Telegram" link on the callback page
+
+ @property
+ def authorize_url(self) -> str:
+ return self.issuer.rstrip("/") + self.authorize_path
+
+ @property
+ def token_url(self) -> str:
+ return self.issuer.rstrip("/") + self.token_path
+
+ @property
+ def callback_path(self) -> str:
+ return urlparse(self.redirect_uri).path or DEFAULT_CALLBACK_PATH
+
+ class Config:
+ env_prefix = "BOTSPOT_CHATGPT_PLAN_"
+ env_file = ".env"
+ env_file_encoding = "utf-8"
+ extra = "ignore"
diff --git a/botspot/components/new/chatgpt_plan/store.py b/botspot/components/new/chatgpt_plan/store.py
new file mode 100644
index 0000000..72bc24f
--- /dev/null
+++ b/botspot/components/new/chatgpt_plan/store.py
@@ -0,0 +1,238 @@
+"""Mongo storage for pending OAuth states and linked plans. Tokens are Fernet-encrypted.
+
+Never log token values: log user ids and statuses only.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from collections import defaultdict
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from typing import TYPE_CHECKING, Any, Optional
+
+from botspot.components.new.chatgpt_plan.errors import (
+ PlanNeedsRelink,
+ PlanNotLinked,
+ PlanUsageCapped,
+)
+from botspot.components.new.chatgpt_plan.settings import (
+ LINKS_COLLECTION,
+ PENDING_COLLECTION,
+ STATUS_ACTIVE,
+ STATUS_CAPPED,
+ STATUS_NEEDS_RELINK,
+ ChatgptPlanSettings,
+)
+from botspot.core.errors import ConfigurationError
+from botspot.utils.internal import get_logger
+
+if TYPE_CHECKING:
+ from botspot.components.new.chatgpt_plan.oauth import OAuthClient, TokenSet
+
+logger = get_logger()
+
+
+def utcnow() -> datetime:
+ return datetime.now(timezone.utc)
+
+
+def as_utc(value: Optional[datetime]) -> Optional[datetime]:
+ """Mongo returns naive datetimes unless tz_aware; treat them as UTC."""
+ if value is None or value.tzinfo is not None:
+ return value
+ return value.replace(tzinfo=timezone.utc)
+
+
+class TokenCipher:
+ """Fernet wrapper. ``cryptography`` is imported lazily."""
+
+ def __init__(self, key: str):
+ if not key:
+ raise ConfigurationError(
+ "BOTSPOT_CHATGPT_PLAN_TOKEN_ENCRYPTION_KEY is required (a Fernet key)."
+ )
+ from cryptography.fernet import Fernet
+
+ self._fernet = Fernet(key.encode())
+
+ def encrypt(self, value: str) -> str:
+ return self._fernet.encrypt(value.encode()).decode()
+
+ def decrypt(self, value: str) -> str:
+ return self._fernet.decrypt(value.encode()).decode()
+
+
+@dataclass
+class PlanLink:
+ """A user's link, without tokens. Safe to show in /settings."""
+
+ user_id: int
+ status: str
+ sub: str = ""
+ email: str = ""
+ scopes: str = ""
+ linked_at: Optional[datetime] = None
+ last_used_at: Optional[datetime] = None
+ capped_until: Optional[datetime] = None
+
+
+class LinkStore:
+ def __init__(self, settings: ChatgptPlanSettings, db: Any, cipher: TokenCipher):
+ self.settings = settings
+ self.pending = db[PENDING_COLLECTION]
+ self.links = db[LINKS_COLLECTION]
+ self.cipher = cipher
+ self._refresh_locks: dict[int, asyncio.Lock] = defaultdict(asyncio.Lock)
+ self._indexes_ready = False
+
+ async def ensure_indexes(self) -> None:
+ if self._indexes_ready:
+ return
+ await self.pending.create_index("expires_at", expireAfterSeconds=0)
+ self._indexes_ready = True
+
+ # -- pending OAuth states ------------------------------------------------
+
+ async def save_pending(self, state: str, user_id: int, verifier: str, nonce: str) -> None:
+ await self.ensure_indexes()
+ now = utcnow()
+ await self.pending.insert_one(
+ {
+ "_id": state,
+ "user_id": user_id,
+ "code_verifier": self.cipher.encrypt(verifier),
+ "nonce": nonce,
+ "created_at": now,
+ "expires_at": now + timedelta(seconds=self.settings.pending_ttl_seconds),
+ }
+ )
+
+ async def pop_pending(self, state: str) -> Optional[dict]:
+ """Single use: the record is deleted on read. Expired records return None."""
+ doc = await self.pending.find_one_and_delete({"_id": state})
+ expires_at = as_utc(doc.get("expires_at")) if doc else None
+ if doc is None or expires_at is None or expires_at <= utcnow():
+ return None
+ doc["code_verifier"] = self.cipher.decrypt(doc["code_verifier"])
+ return doc
+
+ # -- links ----------------------------------------------------------------
+
+ async def save_link(self, user_id: int, tokens: TokenSet, sub: str, email: str) -> None:
+ now = utcnow()
+ await self.links.update_one(
+ {"_id": user_id},
+ {
+ "$set": {
+ "sub": sub,
+ "email": email,
+ "client_id": self.settings.client_id,
+ "scopes": tokens.scope,
+ "status": STATUS_ACTIVE,
+ "capped_until": None,
+ "linked_at": now,
+ **self._token_fields(tokens),
+ }
+ },
+ upsert=True,
+ )
+ logger.info(f"chatgpt_plan: user {user_id} linked")
+
+ def _token_fields(self, tokens: TokenSet) -> dict:
+ fields: dict[str, Any] = {
+ "access_token": self.cipher.encrypt(tokens.access_token),
+ "expires_at": utcnow() + timedelta(seconds=tokens.expires_in),
+ }
+ if tokens.refresh_token: # rolling: keep the old one if none is returned
+ fields["refresh_token"] = self.cipher.encrypt(tokens.refresh_token)
+ return fields
+
+ async def get_link(self, user_id: int) -> Optional[PlanLink]:
+ doc = await self.links.find_one({"_id": user_id})
+ if doc is None:
+ return None
+ return PlanLink(
+ user_id=user_id,
+ status=doc.get("status", STATUS_ACTIVE),
+ sub=doc.get("sub", ""),
+ email=doc.get("email", ""),
+ scopes=doc.get("scopes", ""),
+ linked_at=as_utc(doc.get("linked_at")),
+ last_used_at=as_utc(doc.get("last_used_at")),
+ capped_until=as_utc(doc.get("capped_until")),
+ )
+
+ async def unlink(self, user_id: int) -> bool:
+ result = await self.links.delete_one({"_id": user_id})
+ logger.info(f"chatgpt_plan: user {user_id} unlinked")
+ return bool(result.deleted_count)
+
+ async def mark_needs_relink(self, user_id: int) -> None:
+ await self.links.update_one(
+ {"_id": user_id},
+ {
+ "$set": {"status": STATUS_NEEDS_RELINK},
+ "$unset": {"access_token": "", "refresh_token": "", "expires_at": ""},
+ },
+ )
+ logger.info(f"chatgpt_plan: user {user_id} needs relink")
+
+ async def mark_capped(self, user_id: int, error: PlanUsageCapped) -> None:
+ until = utcnow() + timedelta(seconds=self.settings.cap_pause_seconds)
+ hint = error.reset_hint or ""
+ if hint.isdigit(): # Retry-After seconds
+ until = utcnow() + timedelta(seconds=int(hint))
+ await self.links.update_one(
+ {"_id": user_id}, {"$set": {"status": STATUS_CAPPED, "capped_until": until}}
+ )
+ logger.info(f"chatgpt_plan: user {user_id} capped until {until.isoformat()}")
+
+ async def touch(self, user_id: int) -> None:
+ await self.links.update_one({"_id": user_id}, {"$set": {"last_used_at": utcnow()}})
+
+ # -- access tokens --------------------------------------------------------
+
+ async def _usable_doc(self, user_id: int) -> dict:
+ doc = await self.links.find_one({"_id": user_id})
+ if doc is None:
+ raise PlanNotLinked()
+ status = doc.get("status", STATUS_ACTIVE)
+ if status == STATUS_NEEDS_RELINK or not doc.get("refresh_token"):
+ raise PlanNeedsRelink("ChatGPT link expired, relink needed")
+ capped_until = as_utc(doc.get("capped_until"))
+ if status == STATUS_CAPPED and capped_until and capped_until > utcnow():
+ raise PlanUsageCapped("ChatGPT plan usage cap reached", reset_hint=str(capped_until))
+ return doc
+
+ def _fresh_access_token(self, doc: dict) -> Optional[str]:
+ expires_at = as_utc(doc.get("expires_at"))
+ margin = timedelta(seconds=self.settings.refresh_margin_seconds)
+ if doc.get("access_token") and expires_at and expires_at - margin > utcnow():
+ return self.cipher.decrypt(doc["access_token"])
+ return None
+
+ async def get_access_token(self, user_id: int, oauth: OAuthClient) -> str:
+ """Return a valid access token, refreshing when within the margin."""
+ async with self._refresh_locks[user_id]:
+ doc = await self._usable_doc(user_id)
+ token = self._fresh_access_token(doc)
+ if token is not None:
+ return token
+ try:
+ tokens = await oauth.refresh(self.cipher.decrypt(doc["refresh_token"]))
+ except PlanNeedsRelink:
+ await self.mark_needs_relink(user_id)
+ raise
+ await self.links.update_one(
+ {"_id": user_id},
+ {
+ "$set": {
+ "status": STATUS_ACTIVE,
+ "capped_until": None,
+ **self._token_fields(tokens),
+ }
+ },
+ )
+ logger.debug(f"chatgpt_plan: refreshed token for user {user_id}")
+ return tokens.access_token
diff --git a/botspot/components/new/chatgpt_plan/web.py b/botspot/components/new/chatgpt_plan/web.py
new file mode 100644
index 0000000..c1a592f
--- /dev/null
+++ b/botspot/components/new/chatgpt_plan/web.py
@@ -0,0 +1,65 @@
+"""aiohttp OAuth callback route the host app mounts (only when the flag is on)."""
+
+from __future__ import annotations
+
+from html import escape
+from typing import TYPE_CHECKING, Awaitable, Callable, Optional
+
+from botspot.components.new.chatgpt_plan.errors import PlanError
+from botspot.utils.internal import get_logger
+
+if TYPE_CHECKING:
+ from aiohttp import web
+
+ from botspot.components.new.chatgpt_plan.oauth import LinkResult
+ from botspot.components.new.chatgpt_plan.plan import ChatgptPlan
+
+logger = get_logger()
+
+OnLinked = Callable[["LinkResult"], Awaitable[None]]
+
+_PAGE = """
+{title}
+
+{title}
{body}
{link}"""
+
+
+def render_page(title: str, body: str, return_url: str = "", status: int = 200) -> web.Response:
+ from aiohttp import web
+
+ link = f'Telegram
' if return_url else ""
+ html = _PAGE.format(title=escape(title), body=escape(body), link=link)
+ return web.Response(text=html, content_type="text/html", status=status)
+
+
+def make_callback_handler(
+ plan: ChatgptPlan, on_linked: Optional[OnLinked] = None
+) -> Callable[[web.Request], Awaitable[web.Response]]:
+ """Handler for ``GET ?state=...&code=...``.
+
+ ``on_linked(result)`` lets the host bot message the user after a successful link.
+ """
+ return_url = plan.settings.return_url
+
+ async def callback(request: web.Request) -> web.Response:
+ state = request.query.get("state", "")
+ code = request.query.get("code", "")
+ if request.query.get("error") or not state or not code:
+ return render_page("Not linked", "Linking was cancelled.", return_url, 400)
+ try:
+ result = await plan.complete_link(state, code)
+ except PlanError as error:
+ logger.info(f"chatgpt_plan: callback failed: {error.message}")
+ return render_page("Not linked", "Link failed, try again.", return_url, 400)
+ if on_linked is not None:
+ await on_linked(result)
+ return render_page("Linked", "ChatGPT plan linked. Go back to Telegram.", return_url)
+
+ return callback
+
+
+def aiohttp_routes(plan: ChatgptPlan, on_linked: Optional[OnLinked] = None) -> list:
+ """Return the aiohttp route def for the OAuth callback (path from ``redirect_uri``)."""
+ from aiohttp import web
+
+ return [web.get(plan.settings.callback_path, make_callback_handler(plan, on_linked))]
diff --git a/botspot/core/bot_manager.py b/botspot/core/bot_manager.py
index a7f778b..eb5d0bd 100644
--- a/botspot/core/bot_manager.py
+++ b/botspot/core/bot_manager.py
@@ -25,6 +25,7 @@
auto_archive,
chat_binder,
chat_fetcher,
+ chatgpt_plan,
llm_provider,
message_aggregator,
queue_manager,
@@ -112,6 +113,9 @@ def __init__(
self.settings.subscription_manager
)
+ if self.settings.chatgpt_plan.enabled:
+ self.deps.chatgpt_plan = chatgpt_plan.initialize(self.settings.chatgpt_plan)
+
self.deps.simple_user_cache = simple_user_cache.initialize()
def setup_dispatcher(self, dp: Dispatcher):
diff --git a/botspot/core/botspot_settings.py b/botspot/core/botspot_settings.py
index f26412c..2a66995 100644
--- a/botspot/core/botspot_settings.py
+++ b/botspot/core/botspot_settings.py
@@ -15,6 +15,7 @@
from botspot.components.new.auto_archive import AutoArchiveSettings
from botspot.components.new.chat_binder import ChatBinderSettings
from botspot.components.new.chat_fetcher import ChatFetcherSettings
+from botspot.components.new.chatgpt_plan.settings import ChatgptPlanSettings
from botspot.components.new.llm_provider import LLMProviderSettings
from botspot.components.new.message_aggregator import MessageAggregatorSettings
from botspot.components.new.queue_manager import QueueManagerSettings
@@ -82,6 +83,7 @@ def friends(self) -> List[str]:
s3_storage: S3StorageSettings = S3StorageSettings()
access_control: AccessControlSettings = AccessControlSettings()
subscription_manager: SubscriptionManagerSettings = SubscriptionManagerSettings()
+ chatgpt_plan: ChatgptPlanSettings = ChatgptPlanSettings()
class Config:
env_prefix = "BOTSPOT_"
diff --git a/botspot/core/dependency_manager.py b/botspot/core/dependency_manager.py
index c46df54..73779cf 100644
--- a/botspot/core/dependency_manager.py
+++ b/botspot/core/dependency_manager.py
@@ -15,6 +15,7 @@
from botspot.components.new.auto_archive import AutoArchive
from botspot.components.new.chat_binder import ChatBinder
from botspot.components.new.chat_fetcher import ChatFetcher
+ from botspot.components.new.chatgpt_plan import ChatgptPlan
from botspot.components.new.llm_provider import LLMProvider
from botspot.components.new.message_aggregator import MessageAggregator
from botspot.components.new.queue_manager import QueueManager
@@ -59,6 +60,7 @@ def __init__(
self._simple_user_cache = None
self._access_control = None
self._subscription_manager = None
+ self._chatgpt_plan = None
self.__dict__.update(kwargs)
@property
@@ -253,6 +255,14 @@ def subscription_manager(self) -> "SubscriptionManager":
def subscription_manager(self, value: "SubscriptionManager"):
self._subscription_manager = value
+ @property
+ def chatgpt_plan(self) -> Optional["ChatgptPlan"]:
+ return self._chatgpt_plan
+
+ @chatgpt_plan.setter
+ def chatgpt_plan(self, value: Optional["ChatgptPlan"]):
+ self._chatgpt_plan = value
+
@classmethod
def is_initialized(cls) -> bool:
return cls in cls._instances
diff --git a/botspot/utils/__init__.py b/botspot/utils/__init__.py
index d6188ec..bf07281 100644
--- a/botspot/utils/__init__.py
+++ b/botspot/utils/__init__.py
@@ -32,6 +32,7 @@
get_dispatcher,
get_llm_provider,
get_subscription_manager,
+ get_chatgpt_plan,
get_scheduler,
get_telethon_client,
get_telethon_manager,
@@ -64,6 +65,7 @@
"get_message_aggregator",
"get_llm_provider",
"get_subscription_manager",
+ "get_chatgpt_plan",
"get_easter_egg",
"get_pong",
"strip_command",
diff --git a/botspot/utils/deps_getters.py b/botspot/utils/deps_getters.py
index bcf35e5..15bb4e3 100644
--- a/botspot/utils/deps_getters.py
+++ b/botspot/utils/deps_getters.py
@@ -108,6 +108,13 @@ def get_subscription_manager():
return _get_subscription_manager()
+def get_chatgpt_plan():
+ """Get the chatgpt_plan component. Raises PlanDisabled when the flag is off."""
+ from botspot.components.new.chatgpt_plan import get_chatgpt_plan as _get_chatgpt_plan
+
+ return _get_chatgpt_plan()
+
+
def get_s3_storage() -> Optional[S3StorageProvider]:
"""Get the S3 Storage provider from dependency manager."""
from botspot.core.dependency_manager import get_dependency_manager
@@ -144,4 +151,5 @@ def get_botspot_settings() -> "BotspotSettings":
"get_dependency_manager",
"get_botspot_settings",
"get_subscription_manager",
+ "get_chatgpt_plan",
]
diff --git a/docs/index.md b/docs/index.md
index eee7bf4..6870891 100644
--- a/docs/index.md
+++ b/docs/index.md
@@ -60,7 +60,7 @@ More demos: [examples/](https://github.com/calmmage/botspot/tree/main/examples).
| Features | `user_interactions` (`ask_user`), multi-forward |
| Main | Telethon, scheduler, trial / single-user mode |
| QoL | `@botspot_command` menu, bot info, print bot URL |
-| New | LLM provider, chat binder, queues, S3, message aggregator, `subscription_manager` (credits, Telegram Stars subscriptions, trial caps) |
+| New | LLM provider, chat binder, queues, S3, message aggregator, `subscription_manager` (credits, Telegram Stars subscriptions, trial caps), `chatgpt_plan` (users fund LLM calls with their own ChatGPT plan; off by default, needs an OpenAI partner client ID) |
Enable components in settings / `example.env`. `BotManager` wires them into the dispatcher.
diff --git a/examples/components_examples/chatgpt_plan_demo/bot.py b/examples/components_examples/chatgpt_plan_demo/bot.py
new file mode 100644
index 0000000..4e3b8a9
--- /dev/null
+++ b/examples/components_examples/chatgpt_plan_demo/bot.py
@@ -0,0 +1,34 @@
+"""Minimal chatgpt_plan demo. Needs an OpenAI-issued client ID; the host app must also
+serve ``chatgpt_plan.aiohttp_routes(...)`` at the redirect URI."""
+
+from aiogram import F
+from aiogram.filters import Command
+from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup, Message
+
+from botspot.commands_menu import botspot_command
+from botspot.components.new import chatgpt_plan
+from botspot.utils import send_safe
+from examples.base_bot import App, main, router
+
+
+class ChatgptPlanDemoApp(App):
+ name = "ChatGPT Plan Demo"
+
+
+@botspot_command("link_chatgpt", "Use your ChatGPT plan")
+@router.message(Command("link_chatgpt"))
+async def link_handler(message: Message):
+ url = await chatgpt_plan.start_link(message.from_user.id)
+ button = InlineKeyboardButton(text="Sign in with ChatGPT", url=url)
+ markup = InlineKeyboardMarkup(inline_keyboard=[[button]])
+ await message.answer("Link your ChatGPT plan:", reply_markup=markup)
+
+
+@router.message(F.text)
+async def chat_handler(message: Message):
+ answer = await chatgpt_plan.respond(message.from_user.id, "gpt-6-astra", message.text)
+ await send_safe(message.chat.id, answer.text)
+
+
+if __name__ == "__main__":
+ main(routers=[router], AppClass=ChatgptPlanDemoApp)
diff --git a/examples/components_examples/chatgpt_plan_demo/example.env b/examples/components_examples/chatgpt_plan_demo/example.env
new file mode 100644
index 0000000..6b50abb
--- /dev/null
+++ b/examples/components_examples/chatgpt_plan_demo/example.env
@@ -0,0 +1,7 @@
+TELEGRAM_BOT_TOKEN=your_bot_token_here
+BOTSPOT_MONGO_DATABASE_ENABLED=true
+BOTSPOT_CHATGPT_PLAN_ENABLED=true
+# Issued by OpenAI to approved partners only: https://openai.com/form/sign-in-with-chatgpt-interest/
+BOTSPOT_CHATGPT_PLAN_CLIENT_ID=
+BOTSPOT_CHATGPT_PLAN_REDIRECT_URI=https://bot.example.com/api/chatgpt/callback
+BOTSPOT_CHATGPT_PLAN_TOKEN_ENCRYPTION_KEY=
diff --git a/pyproject.toml b/pyproject.toml
index a8ca3dd..fea1680 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -31,6 +31,7 @@ extras = [
"sqlalchemy[asyncio]>=2.0.40",
"asyncpg>=0.31.0",
"alembic>=1.18.5",
+ "cryptography>=42", # chatgpt_plan: Fernet token encryption
]
test = [
"pytest>=8.3.5",
diff --git a/tests/components/new/test_chatgpt_plan.py b/tests/components/new/test_chatgpt_plan.py
new file mode 100644
index 0000000..dbafc37
--- /dev/null
+++ b/tests/components/new/test_chatgpt_plan.py
@@ -0,0 +1,762 @@
+"""Tests for chatgpt_plan. All HTTP is faked; no client ID, token or network needed."""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import json
+from datetime import timedelta
+from types import SimpleNamespace
+from urllib.parse import parse_qs, urlparse
+
+import pytest
+from aiohttp.test_utils import make_mocked_request
+from cryptography.fernet import Fernet
+
+import botspot.components.new.chatgpt_plan as chatgpt_plan
+from botspot.components.new.chatgpt_plan import (
+ ChatgptPlan,
+ ChatgptPlanSettings,
+ PlanDisabled,
+ PlanError,
+ PlanLinkError,
+ PlanNeedsRelink,
+ PlanNotLinked,
+ PlanUnavailable,
+ PlanUnsupported,
+ PlanUsageCapped,
+ aiohttp_routes,
+)
+from botspot.components.new.chatgpt_plan.client import build_payload, parse_stream
+from botspot.components.new.chatgpt_plan.errors import error_from_response
+from botspot.components.new.chatgpt_plan.http import AiohttpTransport, HttpResult, parse_sse
+from botspot.components.new.chatgpt_plan.settings import (
+ LINKS_COLLECTION,
+ PENDING_COLLECTION,
+ STATUS_CAPPED,
+ STATUS_NEEDS_RELINK,
+)
+from botspot.components.new.chatgpt_plan.store import TokenCipher, utcnow
+from botspot.core.errors import ConfigurationError
+
+CLIENT_ID = "app_test_client"
+REDIRECT = "https://bot.example.com/api/chatgpt/callback"
+USER = 42
+
+
+# -- fakes --------------------------------------------------------------------
+
+
+class FakeCollection:
+ def __init__(self):
+ self.docs: dict = {}
+ self.indexes: list = []
+
+ async def create_index(self, keys, **kwargs):
+ self.indexes.append((keys, kwargs))
+
+ async def insert_one(self, doc):
+ self.docs[doc["_id"]] = dict(doc)
+
+ async def find_one(self, query):
+ doc = self.docs.get(query["_id"])
+ return dict(doc) if doc else None
+
+ async def find_one_and_delete(self, query):
+ return self.docs.pop(query["_id"], None)
+
+ async def delete_one(self, query):
+ return SimpleNamespace(deleted_count=int(self.docs.pop(query["_id"], None) is not None))
+
+ async def update_one(self, query, update, upsert=False):
+ key = query["_id"]
+ if key not in self.docs:
+ if not upsert:
+ return
+ self.docs[key] = {"_id": key}
+ self.docs[key].update(update.get("$set", {}))
+ for field in update.get("$unset", {}):
+ self.docs[key].pop(field, None)
+
+
+class FakeDb(dict):
+ def __missing__(self, name):
+ self[name] = FakeCollection()
+ return self[name]
+
+
+class FakeTransport:
+ """Queues of HttpResult per method; records every call."""
+
+ def __init__(self):
+ self.queues: dict[str, list[HttpResult]] = {"form": [], "get": [], "sse": []}
+ self.calls: list[tuple] = []
+
+ def queue(self, method: str, *results: HttpResult):
+ self.queues[method].extend(results)
+
+ async def post_form(self, url, form):
+ self.calls.append(("form", url, dict(form)))
+ return self.queues["form"].pop(0)
+
+ async def get_json(self, url, token):
+ self.calls.append(("get", url, token))
+ return self.queues["get"].pop(0)
+
+ async def post_sse(self, url, token, payload):
+ self.calls.append(("sse", url, token, dict(payload)))
+ return self.queues["sse"].pop(0)
+
+
+class Explode:
+ """Any attribute access is a failure: proves no I/O happened."""
+
+ def __getattr__(self, name):
+ raise AssertionError(f"unexpected I/O: {name}")
+
+ def __getitem__(self, name):
+ raise AssertionError(f"unexpected I/O: {name}")
+
+
+# -- helpers ------------------------------------------------------------------
+
+
+def make_settings(**overrides) -> ChatgptPlanSettings:
+ values = dict(
+ enabled=True,
+ client_id=CLIENT_ID,
+ redirect_uri=REDIRECT,
+ token_encryption_key=Fernet.generate_key().decode(),
+ unavailable_backoff_seconds=0,
+ )
+ values.update(overrides)
+ return ChatgptPlanSettings(**values)
+
+
+def make_plan(verifier=None, **overrides):
+ db, transport = FakeDb(), FakeTransport()
+ plan = ChatgptPlan(
+ make_settings(**overrides), db=db, transport=transport, id_token_verifier=verifier
+ )
+ return plan, db, transport
+
+
+async def accept_any(id_token, settings):
+ return None
+
+
+def b64(data: dict) -> str:
+ return base64.urlsafe_b64encode(json.dumps(data).encode()).rstrip(b"=").decode()
+
+
+def id_token(**claims) -> str:
+ base = {"iss": "https://auth.openai.com", "aud": CLIENT_ID, "sub": "user-sub", "email": "a@b.c"}
+ base.update(claims)
+ return f"{b64({'alg': 'RS256'})}.{b64(base)}.sig"
+
+
+def token_body(nonce: str, **overrides) -> dict:
+ body = {
+ "access_token": "AT-1",
+ "refresh_token": "RT-1",
+ "expires_in": 3600,
+ "scope": "openid email offline_access chatgpt.tokens.use.direct",
+ "id_token": id_token(nonce=nonce),
+ }
+ body.update(overrides)
+ return body
+
+
+def pending_of(db, url: str) -> tuple[str, dict]:
+ state = parse_qs(urlparse(url).query)["state"][0]
+ return state, db[PENDING_COLLECTION].docs[state]
+
+
+async def link_user(plan, db, transport, user_id=USER) -> str:
+ url = await plan.start_link(user_id)
+ state, pending = pending_of(db, url)
+ transport.queue("form", HttpResult(200, token_body(pending["nonce"])))
+ await plan.complete_link(state, "code-1")
+ transport.calls.clear()
+ return state
+
+
+def sse_events(*texts: str, completed=True, model="gpt-6-astra"):
+ events = [{"type": "response.output_text.delta", "delta": t} for t in texts]
+ if completed:
+ usage = {"input_tokens": 3, "output_tokens": 2}
+ events.append({"type": "response.completed", "response": {"model": model, "usage": usage}})
+ return events
+
+
+# -- settings + wiring --------------------------------------------------------
+
+
+def test_settings_off_by_default_and_env_prefix(monkeypatch):
+ assert ChatgptPlanSettings().enabled is False
+ monkeypatch.setenv("BOTSPOT_CHATGPT_PLAN_ENABLED", "true")
+ monkeypatch.setenv("BOTSPOT_CHATGPT_PLAN_CLIENT_ID", "abc")
+ monkeypatch.setenv("BOTSPOT_CHATGPT_PLAN_REDIRECT_URI", REDIRECT)
+ settings = ChatgptPlanSettings()
+ assert settings.enabled and settings.client_id == "abc"
+ assert settings.token_url == "https://auth.openai.com/api/accounts/oauth/token"
+ assert settings.callback_path == "/api/chatgpt/callback"
+
+
+def test_botspot_settings_registers_component():
+ from botspot.core.botspot_settings import BotspotSettings
+
+ assert BotspotSettings().chatgpt_plan.enabled is False
+
+
+def test_initialize_requires_mongo_and_client_id():
+ from botspot.core.botspot_settings import BotspotSettings
+ from botspot.core.dependency_manager import DependencyManager
+
+ DependencyManager(botspot_settings=BotspotSettings())
+ with pytest.raises(ConfigurationError):
+ chatgpt_plan.initialize(make_settings())
+
+
+def test_missing_encryption_key_is_configuration_error():
+ with pytest.raises(ConfigurationError):
+ ChatgptPlan(make_settings(token_encryption_key=None), db=FakeDb())
+
+
+# -- flag off -----------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_flag_off_raises_disabled_without_io():
+ plan = ChatgptPlan(ChatgptPlanSettings(), db=Explode(), transport=Explode()) # type: ignore[arg-type]
+ calls = [
+ plan.start_link(USER),
+ plan.complete_link("s", "c"),
+ plan.get_access_token(USER),
+ plan.get_link(USER),
+ plan.unlink(USER),
+ plan.respond(USER, "m", "hi"),
+ plan.list_models(USER),
+ ]
+ for call in calls:
+ with pytest.raises(PlanDisabled):
+ await call
+
+
+@pytest.mark.asyncio
+async def test_module_functions_disabled_without_component():
+ with pytest.raises(PlanDisabled):
+ await chatgpt_plan.start_link(USER)
+ with pytest.raises(PlanDisabled):
+ await chatgpt_plan.respond(USER, "m", "hi")
+
+ from botspot.core.botspot_settings import BotspotSettings
+ from botspot.core.dependency_manager import DependencyManager
+
+ DependencyManager(botspot_settings=BotspotSettings()) # initialized, component off
+ with pytest.raises(PlanDisabled):
+ await chatgpt_plan.list_models(USER)
+
+
+@pytest.mark.asyncio
+async def test_module_functions_delegate_when_enabled():
+ from botspot.core.botspot_settings import BotspotSettings
+ from botspot.core.dependency_manager import DependencyManager
+
+ plan, db, transport = make_plan(accept_any)
+ DependencyManager(botspot_settings=BotspotSettings()).chatgpt_plan = plan
+ await link_user(plan, db, transport)
+ assert await chatgpt_plan.get_access_token(USER) == "AT-1"
+ assert (await chatgpt_plan.get_link(USER)).email == "a@b.c"
+ assert await chatgpt_plan.unlink(USER) is True
+
+
+# -- link flow ----------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_authorize_url_params_and_pkce():
+ plan, db, _ = make_plan()
+ url = await plan.start_link(USER)
+ parsed = urlparse(url)
+ assert f"{parsed.scheme}://{parsed.netloc}{parsed.path}" == (
+ "https://auth.openai.com/api/accounts/authorize"
+ )
+ q = {k: v[0] for k, v in parse_qs(parsed.query).items()}
+ assert q["response_type"] == "code"
+ assert q["client_id"] == CLIENT_ID
+ assert q["redirect_uri"] == REDIRECT
+ assert "chatgpt.tokens.use.direct" in q["scope"].split()
+ assert q["code_challenge_method"] == "S256"
+ assert q["resource"] == "https://api.openai.com/v1"
+ assert "agent_name_hint" not in q
+ state, pending = pending_of(db, url)
+ assert state == q["state"] and pending["nonce"] == q["nonce"]
+ assert pending["user_id"] == USER
+ verifier = plan.store.cipher.decrypt(pending["code_verifier"])
+ digest = hashlib.sha256(verifier.encode()).digest()
+ assert q["code_challenge"] == base64.urlsafe_b64encode(digest).rstrip(b"=").decode()
+ assert db[PENDING_COLLECTION].indexes == [("expires_at", {"expireAfterSeconds": 0})]
+
+
+@pytest.mark.asyncio
+async def test_complete_link_exchanges_code_and_stores_encrypted():
+ plan, db, transport = make_plan(accept_any)
+ url = await plan.start_link(USER)
+ state, pending = pending_of(db, url)
+ verifier = plan.store.cipher.decrypt(pending["code_verifier"])
+ transport.queue("form", HttpResult(200, token_body(pending["nonce"])))
+
+ result = await plan.complete_link(state, "code-1")
+
+ assert (result.user_id, result.sub, result.email) == (USER, "user-sub", "a@b.c")
+ _, url_called, form = transport.calls[0]
+ assert url_called == plan.settings.token_url
+ assert form == {
+ "grant_type": "authorization_code",
+ "code": "code-1",
+ "redirect_uri": REDIRECT,
+ "client_id": CLIENT_ID,
+ "code_verifier": verifier,
+ "resource": "https://api.openai.com/v1",
+ }
+ doc = db[LINKS_COLLECTION].docs[USER]
+ assert "AT-1" not in json.dumps(doc, default=str)
+ assert "RT-1" not in json.dumps(doc, default=str)
+ assert plan.store.cipher.decrypt(doc["refresh_token"]) == "RT-1"
+ assert doc["status"] == "active" and doc["client_id"] == CLIENT_ID
+
+
+@pytest.mark.asyncio
+async def test_state_is_single_use():
+ plan, db, transport = make_plan(accept_any)
+ state = await link_user(plan, db, transport)
+ with pytest.raises(PlanLinkError):
+ await plan.complete_link(state, "code-1")
+ assert transport.calls == []
+
+
+@pytest.mark.asyncio
+async def test_expired_state_is_rejected_without_http():
+ plan, db, transport = make_plan(accept_any)
+ state, pending = pending_of(db, await plan.start_link(USER))
+ pending["expires_at"] = utcnow() - timedelta(seconds=1)
+ with pytest.raises(PlanLinkError):
+ await plan.complete_link(state, "code-1")
+ assert transport.calls == []
+
+
+@pytest.mark.asyncio
+async def test_missing_scope_is_rejected():
+ plan, db, transport = make_plan(accept_any)
+ state, pending = pending_of(db, await plan.start_link(USER))
+ body = token_body(pending["nonce"], scope="openid email offline_access")
+ transport.queue("form", HttpResult(200, body))
+ with pytest.raises(PlanLinkError, match="chatgpt.tokens.use.direct"):
+ await plan.complete_link(state, "code-1")
+ assert USER not in db[LINKS_COLLECTION].docs
+
+
+@pytest.mark.parametrize(
+ "claims",
+ [{"nonce": "wrong"}, {"aud": "someone-else"}, {"iss": "https://evil.example.com"}],
+)
+@pytest.mark.asyncio
+async def test_id_token_claim_checks(claims):
+ plan, db, transport = make_plan(accept_any)
+ state, pending = pending_of(db, await plan.start_link(USER))
+ body = token_body(pending["nonce"], id_token=id_token(**{"nonce": pending["nonce"], **claims}))
+ transport.queue("form", HttpResult(200, body))
+ with pytest.raises(PlanLinkError):
+ await plan.complete_link(state, "code-1")
+
+
+@pytest.mark.asyncio
+async def test_id_token_signature_hook_is_called_and_can_reject():
+ seen = []
+
+ async def reject(token, settings):
+ seen.append(token)
+ raise PlanLinkError("bad signature")
+
+ plan, db, transport = make_plan(reject)
+ state, pending = pending_of(db, await plan.start_link(USER))
+ body = token_body(pending["nonce"])
+ transport.queue("form", HttpResult(200, body))
+ with pytest.raises(PlanLinkError, match="bad signature"):
+ await plan.complete_link(state, "code-1")
+ assert seen == [body["id_token"]]
+ assert USER not in db[LINKS_COLLECTION].docs
+
+
+@pytest.mark.asyncio
+async def test_failed_code_exchange_is_link_error():
+ plan, db, transport = make_plan(accept_any)
+ state, _ = pending_of(db, await plan.start_link(USER))
+ transport.queue("form", HttpResult(400, {"error": "invalid_grant"}))
+ with pytest.raises(PlanLinkError, match="invalid_grant"):
+ await plan.complete_link(state, "code-1")
+
+
+# -- tokens -------------------------------------------------------------------
+
+
+def test_encryption_round_trip():
+ cipher = TokenCipher(Fernet.generate_key().decode())
+ secret = cipher.encrypt("refresh-token-value")
+ assert secret != "refresh-token-value"
+ assert cipher.decrypt(secret) == "refresh-token-value"
+ with pytest.raises(ConfigurationError):
+ TokenCipher("")
+
+
+@pytest.mark.asyncio
+async def test_fresh_access_token_needs_no_http():
+ plan, db, transport = make_plan(accept_any)
+ await link_user(plan, db, transport)
+ assert await plan.get_access_token(USER) == "AT-1"
+ assert transport.calls == []
+
+
+@pytest.mark.asyncio
+async def test_refresh_is_rolling():
+ plan, db, transport = make_plan(accept_any)
+ await link_user(plan, db, transport)
+ links = db[LINKS_COLLECTION].docs
+
+ links[USER]["expires_at"] = utcnow() + timedelta(seconds=60) # inside 300 s margin
+ transport.queue("form", HttpResult(200, {"access_token": "AT-2", "refresh_token": "RT-2"}))
+ assert await plan.get_access_token(USER) == "AT-2"
+ assert transport.calls[-1][2]["refresh_token"] == "RT-1"
+ assert transport.calls[-1][2]["grant_type"] == "refresh_token"
+
+ links[USER]["expires_at"] = utcnow() - timedelta(seconds=1)
+ transport.queue("form", HttpResult(200, {"access_token": "AT-3", "expires_in": 3600}))
+ assert await plan.get_access_token(USER) == "AT-3"
+ assert transport.calls[-1][2]["refresh_token"] == "RT-2"
+ # no new refresh token returned -> the last one is kept
+ assert plan.store.cipher.decrypt(links[USER]["refresh_token"]) == "RT-2"
+
+
+@pytest.mark.parametrize(
+ "status,body",
+ [
+ (400, {"error": "invalid_grant"}),
+ (400, {"error": "invalid_refresh_token"}),
+ (401, {"error": {"code": "token_expired"}}),
+ (401, {}),
+ ],
+)
+@pytest.mark.asyncio
+async def test_refresh_failure_marks_needs_relink(status, body):
+ plan, db, transport = make_plan(accept_any)
+ await link_user(plan, db, transport)
+ db[LINKS_COLLECTION].docs[USER]["expires_at"] = utcnow() - timedelta(seconds=1)
+ transport.queue("form", HttpResult(status, body))
+ with pytest.raises(PlanNeedsRelink):
+ await plan.get_access_token(USER)
+ doc = db[LINKS_COLLECTION].docs[USER]
+ assert doc["status"] == STATUS_NEEDS_RELINK
+ assert "refresh_token" not in doc and "access_token" not in doc
+ transport.calls.clear()
+ with pytest.raises(PlanNeedsRelink):
+ await plan.get_access_token(USER)
+ assert transport.calls == []
+
+
+@pytest.mark.asyncio
+async def test_not_linked_and_unlink():
+ plan, db, transport = make_plan(accept_any)
+ with pytest.raises(PlanNotLinked):
+ await plan.get_access_token(USER)
+ assert await plan.get_link(USER) is None
+ await link_user(plan, db, transport)
+ assert (await plan.get_link(USER)).status == "active"
+ assert await plan.unlink(USER) is True
+ assert await plan.unlink(USER) is False
+
+
+# -- responses ----------------------------------------------------------------
+
+
+def test_parse_sse():
+ lines = [
+ 'data: {"type": "a"}',
+ "",
+ "event: x",
+ 'data: {"type":',
+ 'data: "b"}',
+ "",
+ "data: [DONE]",
+ "",
+ ]
+ assert parse_sse(lines) == [{"type": "a"}, {"type": "b"}]
+
+
+def test_build_payload_strips_disallowed_params():
+ payload = build_payload(
+ "m", "hi", "be brief", {"temperature": 0.2, "max_output_tokens": 5, "reasoning": {}}
+ )
+ assert payload == {
+ "model": "m",
+ "input": "hi",
+ "store": False,
+ "stream": True,
+ "instructions": "be brief",
+ "reasoning": {},
+ }
+
+
+def test_parse_stream_requires_completed():
+ assert parse_stream(sse_events("Hel", "lo"), "m").text == "Hello"
+ with pytest.raises(PlanUnavailable, match="response.completed"):
+ parse_stream(sse_events("Hel", completed=False), "m")
+
+
+def test_parse_stream_falls_back_to_output_items():
+ out = [{"content": [{"type": "output_text", "text": "Hi"}]}]
+ events = [{"type": "response.completed", "response": {"output": out}}]
+ assert parse_stream(events, "m").text == "Hi"
+
+
+@pytest.mark.asyncio
+async def test_respond_streams_on_user_plan():
+ plan, db, transport = make_plan(accept_any)
+ await link_user(plan, db, transport)
+ transport.queue("sse", HttpResult(200, events=sse_events("Hel", "lo")))
+
+ response = await plan.respond(USER, "gpt-6-astra", "hi", instructions="sys", temperature=1)
+
+ assert (response.text, response.model) == ("Hello", "gpt-6-astra")
+ assert response.usage == {"input_tokens": 3, "output_tokens": 2}
+ _, url, token, payload = transport.calls[0]
+ assert url == "https://api.openai.com/v1/responses" and token == "AT-1"
+ assert payload["store"] is False and payload["stream"] is True
+ assert "temperature" not in payload
+ assert db[LINKS_COLLECTION].docs[USER]["last_used_at"] is not None
+
+
+@pytest.mark.asyncio
+async def test_respond_without_completed_is_error():
+ plan, db, transport = make_plan(accept_any, unavailable_retries=0)
+ await link_user(plan, db, transport)
+ transport.queue("sse", HttpResult(200, events=sse_events("partial", completed=False)))
+ with pytest.raises(PlanUnavailable):
+ await plan.respond(USER, "m", "hi")
+
+
+@pytest.mark.parametrize(
+ "status,body,expected",
+ [
+ (429, {"error": {"code": "subscription_sharing_usage_limit_exceeded"}}, PlanUsageCapped),
+ (429, {}, PlanUsageCapped),
+ (401, {"error": {"code": "invalid_api_key"}}, PlanNeedsRelink),
+ (400, {"error": "invalid_grant"}, PlanNeedsRelink),
+ (400, {"error": "invalid_refresh_token"}, PlanNeedsRelink),
+ (401, {"error": {"code": "token_expired"}}, PlanNeedsRelink),
+ (403, {"error": {"code": "unsupported_country"}}, PlanUnavailable),
+ (503, {"error": {"code": "subscription_sharing_usage_unavailable"}}, PlanUnavailable),
+ (
+ 400,
+ {"error": {"code": "subscription_sharing_unsupported_capability", "param": "tools"}},
+ PlanUnsupported,
+ ),
+ (500, None, PlanUnavailable),
+ ],
+)
+def test_error_mapping_table(status, body, expected):
+ error = error_from_response(status, body)
+ assert type(error) is expected
+ assert isinstance(error, PlanError)
+
+
+def test_error_mapping_details():
+ capped = error_from_response(429, {}, {"Retry-After": "120"})
+ assert isinstance(capped, PlanUsageCapped) and capped.reset_hint == "120"
+ unsupported = error_from_response(
+ 400, {"error": {"code": "subscription_sharing_unsupported_capability", "param": "tools"}}
+ )
+ assert isinstance(unsupported, PlanUnsupported) and unsupported.param == "tools"
+ forbidden = error_from_response(403, {})
+ assert isinstance(forbidden, PlanUnavailable) and forbidden.retryable is False
+ assert str(forbidden)
+
+
+@pytest.mark.asyncio
+async def test_respond_429_marks_capped_then_short_circuits():
+ plan, db, transport = make_plan(accept_any)
+ await link_user(plan, db, transport)
+ body = {"error": {"code": "subscription_sharing_usage_limit_exceeded"}}
+ transport.queue("sse", HttpResult(429, body, headers={"Retry-After": "600"}))
+ with pytest.raises(PlanUsageCapped):
+ await plan.respond(USER, "m", "hi")
+ doc = db[LINKS_COLLECTION].docs[USER]
+ assert doc["status"] == STATUS_CAPPED
+ assert doc["capped_until"] > utcnow() + timedelta(seconds=500)
+ transport.calls.clear()
+ with pytest.raises(PlanUsageCapped):
+ await plan.respond(USER, "m", "hi")
+ assert transport.calls == []
+
+
+@pytest.mark.asyncio
+async def test_respond_401_marks_needs_relink():
+ plan, db, transport = make_plan(accept_any)
+ await link_user(plan, db, transport)
+ transport.queue("sse", HttpResult(401, {"error": {"code": "token_revoked"}}))
+ with pytest.raises(PlanNeedsRelink):
+ await plan.respond(USER, "m", "hi")
+ assert db[LINKS_COLLECTION].docs[USER]["status"] == STATUS_NEEDS_RELINK
+
+
+@pytest.mark.asyncio
+async def test_respond_stream_failed_event_is_mapped():
+ plan, db, transport = make_plan(accept_any)
+ await link_user(plan, db, transport)
+ failed = {
+ "type": "response.failed",
+ "response": {"error": {"code": "subscription_sharing_usage_limit_exceeded"}},
+ }
+ transport.queue("sse", HttpResult(200, events=[failed]))
+ with pytest.raises(PlanUsageCapped):
+ await plan.respond(USER, "m", "hi")
+ assert db[LINKS_COLLECTION].docs[USER]["status"] == STATUS_CAPPED
+
+
+@pytest.mark.asyncio
+async def test_respond_503_backs_off_then_succeeds():
+ plan, db, transport = make_plan(accept_any, unavailable_retries=2)
+ await link_user(plan, db, transport)
+ busy = HttpResult(503, {"error": {"code": "subscription_sharing_usage_unavailable"}})
+ transport.queue("sse", busy, busy, HttpResult(200, events=sse_events("ok")))
+ assert (await plan.respond(USER, "m", "hi")).text == "ok"
+ assert len(transport.calls) == 3
+
+
+@pytest.mark.asyncio
+async def test_respond_503_gives_up_after_bounded_retries():
+ plan, db, transport = make_plan(accept_any, unavailable_retries=1)
+ await link_user(plan, db, transport)
+ busy = HttpResult(503, {})
+ transport.queue("sse", busy, busy, busy)
+ with pytest.raises(PlanUnavailable):
+ await plan.respond(USER, "m", "hi")
+ assert len(transport.calls) == 2
+
+
+@pytest.mark.asyncio
+async def test_respond_403_is_not_retried():
+ plan, db, transport = make_plan(accept_any, unavailable_retries=3)
+ await link_user(plan, db, transport)
+ transport.queue("sse", HttpResult(403, {}))
+ with pytest.raises(PlanUnavailable):
+ await plan.respond(USER, "m", "hi")
+ assert len(transport.calls) == 1
+
+
+@pytest.mark.asyncio
+async def test_respond_drops_unsupported_param_and_retries():
+ plan, db, transport = make_plan(accept_any)
+ await link_user(plan, db, transport)
+ unsupported = {
+ "error": {"code": "subscription_sharing_unsupported_capability", "param": "reasoning"}
+ }
+ transport.queue("sse", HttpResult(400, unsupported), HttpResult(200, events=sse_events("ok")))
+ response = await plan.respond(USER, "m", "hi", reasoning={"effort": "low"})
+ assert response.text == "ok"
+ assert "reasoning" in transport.calls[0][3] and "reasoning" not in transport.calls[1][3]
+
+
+@pytest.mark.asyncio
+async def test_respond_unsupported_required_param_raises():
+ plan, db, transport = make_plan(accept_any)
+ await link_user(plan, db, transport)
+ unsupported = {
+ "error": {"code": "subscription_sharing_unsupported_capability", "param": "input"}
+ }
+ transport.queue("sse", HttpResult(400, unsupported))
+ with pytest.raises(PlanUnsupported) as info:
+ await plan.respond(USER, "m", "hi")
+ assert info.value.param == "input"
+
+
+@pytest.mark.asyncio
+async def test_list_models_filters_visibility():
+ plan, db, transport = make_plan(accept_any)
+ await link_user(plan, db, transport)
+ models = [
+ {"id": "gpt-6-astra", "visibility": "list"},
+ {"id": "hidden", "visibility": "hide"},
+ {"id": "gpt-6-mini", "visibility": "list"},
+ ]
+ transport.queue("get", HttpResult(200, {"data": models}))
+ assert await plan.list_models(USER) == ["gpt-6-astra", "gpt-6-mini"]
+ assert transport.calls[0][1:] == ("https://api.openai.com/v1/models", "AT-1")
+
+
+# -- web callback -------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_callback_route_links_and_notifies():
+ plan, db, transport = make_plan(accept_any)
+ state, pending = pending_of(db, await plan.start_link(USER))
+ transport.queue("form", HttpResult(200, token_body(pending["nonce"])))
+ linked = []
+
+ async def on_linked(result):
+ linked.append(result.user_id)
+
+ (route,) = aiohttp_routes(plan, on_linked)
+ assert route.method == "GET" and route.path == "/api/chatgpt/callback"
+ request = make_mocked_request("GET", f"/api/chatgpt/callback?state={state}&code=c")
+ response = await route.handler(request)
+ assert response.status == 200 and "Linked" in response.text
+ assert linked == [USER]
+
+
+@pytest.mark.parametrize("query", ["state=unknown&code=c", "error=access_denied", ""])
+@pytest.mark.asyncio
+async def test_callback_route_failures(query):
+ plan, _, _ = make_plan(accept_any)
+ (route,) = aiohttp_routes(plan)
+ response = await route.handler(make_mocked_request("GET", f"/api/chatgpt/callback?{query}"))
+ assert response.status == 400 and "Not linked" in response.text
+
+
+# -- real transport against a local test server -------------------------------
+
+
+@pytest.mark.asyncio
+async def test_aiohttp_transport_against_local_server():
+ from aiohttp import web
+ from aiohttp.test_utils import TestServer
+
+ async def token(request):
+ form = await request.post()
+ return web.json_response({"grant_type": form["grant_type"]})
+
+ async def responses(request):
+ assert request.headers["Authorization"] == "Bearer tok"
+ body = await request.json()
+ events = sse_events(body["input"])
+ text = "".join(f"data: {json.dumps(e)}\n\n" for e in events) + "data: [DONE]\n\n"
+ return web.Response(text=text, content_type="text/event-stream")
+
+ async def models(request):
+ return web.json_response({"error": {"code": "token_expired"}}, status=401)
+
+ app = web.Application()
+ app.router.add_post("/token", token)
+ app.router.add_post("/responses", responses)
+ app.router.add_get("/models", models)
+ async with TestServer(app) as server:
+ transport = AiohttpTransport(timeout_seconds=5)
+ form = await transport.post_form(str(server.make_url("/token")), {"grant_type": "x"})
+ assert form.ok and form.body == {"grant_type": "x"}
+ sse = await transport.post_sse(str(server.make_url("/responses")), "tok", {"input": "yo"})
+ assert parse_stream(sse.events, "m").text == "yo"
+ bad = await transport.get_json(str(server.make_url("/models")), "tok")
+ assert bad.status == 401 and isinstance(
+ error_from_response(bad.status, bad.body), PlanNeedsRelink
+ )
diff --git a/uv.lock b/uv.lock
index 3a214dc..9b53cc8 100644
--- a/uv.lock
+++ b/uv.lock
@@ -508,6 +508,7 @@ extras = [
{ name = "apscheduler" },
{ name = "asyncpg" },
{ name = "calmlib" },
+ { name = "cryptography" },
{ name = "litellm" },
{ name = "mistune" },
{ name = "pymongo" },
@@ -556,6 +557,7 @@ extras = [
{ name = "apscheduler", specifier = ">=3.10.4,<4.0.0" },
{ name = "asyncpg", specifier = ">=0.31.0" },
{ name = "calmlib", git = "https://github.com/calmmage/calmlib.git?rev=main" },
+ { name = "cryptography", specifier = ">=42" },
{ name = "litellm", specifier = ">=1.63,<2.0" },
{ name = "mistune", specifier = ">=3.1.3,<4.0.0" },
{ name = "pymongo", specifier = ">=4.9" },