diff --git a/core/database.py b/core/database.py
index d98360046..6addc95c4 100644
--- a/core/database.py
+++ b/core/database.py
@@ -203,6 +203,12 @@ class Session(TimestampMixin, Base):
# Organization
folder = Column(String, nullable=True, default=None)
cwd = Column(String, nullable=True, default=None)
+ # Registered ModelEndpoint this session is bound to. endpoint_url alone
+ # cannot distinguish two endpoints that share a provider URL but use
+ # different credentials (e.g. two ChatGPT Subscription accounts), so the
+ # exact endpoint id is remembered here. NULL = legacy session; the first
+ # deterministic, owner-scoped resolution persists a binding.
+ endpoint_id = Column(String, nullable=True, index=True)
# Headers stored as JSON
headers = Column(JSON, default=dict)
@@ -1472,6 +1478,19 @@ def _migrate_add_session_cwd_column():
except Exception:
pass
+def _migrate_add_session_endpoint_id_column():
+ """Add the nullable binding and index without rewriting existing sessions."""
+ with engine.begin() as connection:
+ schema = inspect(connection)
+ if not schema.has_table("sessions"):
+ return
+ columns = {column["name"] for column in schema.get_columns("sessions")}
+ if "endpoint_id" not in columns:
+ connection.execute(text("ALTER TABLE sessions ADD COLUMN endpoint_id VARCHAR"))
+ index = next(index for index in Session.__table__.indexes if index.name == "ix_sessions_endpoint_id")
+ index.create(bind=connection, checkfirst=True)
+
+
def _migrate_add_token_columns():
"""Add cumulative token tracking columns to sessions table."""
import sqlite3
@@ -2378,6 +2397,7 @@ def init_db():
_migrate_add_session_generation_settings_columns()
_migrate_add_folder_column()
_migrate_add_session_cwd_column()
+ _migrate_add_session_endpoint_id_column()
_migrate_add_token_columns()
_migrate_add_total_cost_usd()
_migrate_add_mode_column()
diff --git a/core/models.py b/core/models.py
index 851950713..a03e391fc 100644
--- a/core/models.py
+++ b/core/models.py
@@ -115,6 +115,10 @@ class Session:
temperature_override: Optional[float] = None
max_tokens_override: Optional[int] = None
cwd: Optional[str] = None
+ # Registered ModelEndpoint id this session is bound to (None = legacy /
+ # URL-matched). Lets two endpoints that share a provider URL but not
+ # credentials stay distinguishable.
+ endpoint_id: Optional[str] = None
def __post_init__(self):
if self.headers is None:
diff --git a/core/session_manager.py b/core/session_manager.py
index 6708fb69d..f7467eb04 100644
--- a/core/session_manager.py
+++ b/core/session_manager.py
@@ -157,6 +157,7 @@ class SessionManager:
temperature_override=getattr(db_session, "temperature_override", None),
max_tokens_override=getattr(db_session, "max_tokens_override", None),
cwd=getattr(db_session, "cwd", None) or None,
+ endpoint_id=getattr(db_session, "endpoint_id", None) or None,
)
session.message_count = getattr(db_session, "message_count", 0) or 0
return session
@@ -222,6 +223,7 @@ class SessionManager:
temperature_override=getattr(db_session, "temperature_override", None),
max_tokens_override=getattr(db_session, "max_tokens_override", None),
cwd=getattr(db_session, "cwd", None) or None,
+ endpoint_id=getattr(db_session, "endpoint_id", None) or None,
)
# The rows just loaded are the whole transcript, so they — not the
@@ -493,6 +495,7 @@ class SessionManager:
headers = {}
session.name = db_session.name
session.endpoint_url = db_session.endpoint_url or ""
+ session.endpoint_id = getattr(db_session, "endpoint_id", None) or None
session.model = db_session.model or ""
session.headers = headers or {}
session.rag = db_session.rag
@@ -563,9 +566,12 @@ class SessionManager:
owner: str = None,
cwd: str = None,
headers: Optional[Dict[str, str]] = None,
+ endpoint_id: Optional[str] = None,
) -> Session:
"""Create a new session and save to database."""
- session_headers = dict(headers or {})
+ from src.chatgpt_subscription import is_chatgpt_subscription_base
+ session_headers = {} if is_chatgpt_subscription_base(endpoint_url) else dict(headers or {})
+ endpoint_id = (endpoint_id or "").strip() or None
db = SessionLocal()
try:
db_session = DbSession(
@@ -577,6 +583,7 @@ class SessionManager:
headers=session_headers,
owner=owner,
cwd=cwd or None,
+ endpoint_id=endpoint_id,
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc)
)
@@ -592,6 +599,7 @@ class SessionManager:
headers=session_headers,
owner=owner,
cwd=cwd or None,
+ endpoint_id=endpoint_id,
)
self.sessions[session_id] = session
diff --git a/routes/chat_helpers.py b/routes/chat_helpers.py
index 35962dba5..96ac103d8 100644
--- a/routes/chat_helpers.py
+++ b/routes/chat_helpers.py
@@ -794,6 +794,52 @@ def _session_url_matches_endpoint(session_url: str, endpoint_base: str) -> bool:
return False
+def _endpoint_created_sort_key(ep) -> tuple:
+ created = getattr(ep, "created_at", None)
+ try:
+ ts = float(created.timestamp()) if created else 0.0
+ except Exception:
+ ts = 0.0
+ return (ts, str(getattr(ep, "id", "") or ""))
+
+
+def _select_session_endpoint(sess, target_url: str, endpoints) -> tuple:
+ """Pick the endpoint a session should use for credential resolution.
+
+ Two endpoints may share one provider URL but not credentials (e.g. two
+ ChatGPT Subscription accounts), so an explicit ``sess.endpoint_id`` binding
+ wins whenever it still matches the session URL. Without a binding the
+ oldest URL-matching endpoint is chosen deterministically and persisted.
+
+ Returns ``(endpoint, bound_by_fallback)``; ``bound_by_fallback`` is True
+ when the choice came from URL matching and may be persisted as a binding.
+ """
+ matching = [ep for ep in endpoints if _session_url_matches_endpoint(target_url, getattr(ep, "base_url", "") or "")]
+ if not matching:
+ return None, False
+ bound_id = getattr(sess, "endpoint_id", None) or None
+ if bound_id:
+ for ep in matching:
+ if str(ep.id) == str(bound_id):
+ sess.endpoint_id = ep.id
+ return ep, False
+ # The bound endpoint is gone or disabled. Never silently borrow another
+ # endpoint's credentials when several routes share this URL.
+ return None, False
+ matching.sort(key=_endpoint_created_sort_key)
+ chosen = matching[0]
+ if len(matching) > 1:
+ logger.warning(
+ "Session %s has no endpoint binding and %d endpoints share its URL; using oldest endpoint %s",
+ getattr(sess, "id", "?"), len(matching), chosen.id,
+ )
+ try:
+ sess.endpoint_id = chosen.id
+ except Exception:
+ pass
+ return chosen, True
+
+
def _has_auth_keys(headers) -> bool:
"""True if a headers dict carries an Authorization/x-api-key entry."""
return isinstance(headers, dict) and any(
@@ -803,6 +849,7 @@ def _has_auth_keys(headers) -> bool:
def resolve_session_auth(sess, session_id: str, owner: Optional[str] = None):
"""Ensure session has auth headers — resolve from endpoint DB if missing."""
+ owner = owner or getattr(sess, "owner", None)
try:
from src.chatgpt_subscription import is_chatgpt_subscription_base
is_chatgpt_subscription = is_chatgpt_subscription_base(getattr(sess, "endpoint_url", "") or "")
@@ -811,11 +858,22 @@ def resolve_session_auth(sess, session_id: str, owner: Optional[str] = None):
has_auth = _has_auth_keys(sess.headers)
if has_auth and not is_chatgpt_subscription:
return
+ if is_chatgpt_subscription:
+ # Never reuse a stale bearer after deletion, disablement or failed refresh.
+ sess.headers = {}
try:
from src.endpoint_resolver import build_headers, resolve_endpoint_runtime
db = SessionLocal()
try:
+ stored_q = db.query(DBSession).filter(DBSession.id == session_id)
+ if owner:
+ stored_q = stored_q.filter(DBSession.owner == owner)
+ if is_chatgpt_subscription:
+ stored = stored_q.first()
+ if stored is not None and _has_auth_keys(stored.headers):
+ stored_q.update({"headers": {}})
+ db.commit()
target_url = getattr(sess, "endpoint_url", "") or ""
if not target_url:
return
@@ -826,44 +884,36 @@ def resolve_session_auth(sess, session_id: str, owner: Optional[str] = None):
# with similar endpoint URLs can borrow each other's API key.
from src.auth_helpers import owner_filter
q = owner_filter(q, ModelEndpoint, owner)
- for ep in q.all():
- if not _session_url_matches_endpoint(target_url, ep.base_url or ""):
- continue
- try:
- base, api_key = resolve_endpoint_runtime(ep, owner=owner)
- except Exception as e:
- logger.warning("Failed to resolve provider auth for session %s: %s", session_id, e)
- return
- if not api_key:
- # No usable key (e.g. ChatGPT Subscription needs re-auth).
- return
- sess.headers = build_headers(api_key, base)
- if is_chatgpt_subscription:
- # The bearer is short-lived and re-resolved per request, so it
- # stays request-local and is never written to the plaintext
- # sessions.headers column. Proactively strip any bearer an
- # older code path may have persisted so it does not linger.
- stale_q = db.query(DBSession).filter(DBSession.id == session_id)
- if owner:
- stale_q = stale_q.filter(DBSession.owner == owner)
- stored = stale_q.first()
- if stored is not None and _has_auth_keys(stored.headers):
- stale_q.update({"headers": {}})
- db.commit()
- logger.info(f"Cleared persisted ChatGPT Subscription bearer from session {session_id}")
- logger.debug(f"Resolved request-local ChatGPT Subscription auth for session {session_id}")
- return
- update_q = db.query(DBSession).filter(DBSession.id == session_id)
- if owner:
- update_q = update_q.filter(DBSession.owner == owner)
- update_q.update({"headers": sess.headers})
- db.commit()
- logger.info(f"Resolved and persisted auth headers for session {session_id} from endpoint {ep.name}")
+ ep, bound_here = _select_session_endpoint(sess, target_url, q.all())
+ if ep is None:
return
+ if bound_here:
+ # Bind before authentication, including failed/expired credentials.
+ stored_q.filter(DBSession.endpoint_id == None).update({"endpoint_id": ep.id})
+ db.commit()
+ try:
+ base, api_key = resolve_endpoint_runtime(ep, owner=owner)
+ except Exception as e:
+ logger.warning("Failed to resolve provider auth for session %s: %s", session_id, type(e).__name__)
+ return
+ if not api_key:
+ # No usable key (e.g. ChatGPT Subscription needs re-auth).
+ return
+ sess.headers = build_headers(api_key, base)
+ if is_chatgpt_subscription:
+ # Request-local only; persistence was cleaned before resolution.
+ return
+ update_q = db.query(DBSession).filter(DBSession.id == session_id)
+ if owner:
+ update_q = update_q.filter(DBSession.owner == owner)
+ update_q.update({"headers": sess.headers})
+ db.commit()
+ logger.info(f"Resolved and persisted auth headers for session {session_id} from endpoint {ep.name}")
+ return
finally:
db.close()
except Exception as e:
- logger.warning(f"Failed to resolve session headers: {e}")
+ logger.warning("Failed to resolve session headers: %s", type(e).__name__)
def _match_cached_model_id(requested: str, models) -> Optional[str]:
diff --git a/routes/chat_routes.py b/routes/chat_routes.py
index 5d58ef7fb..33e8d6bd3 100644
--- a/routes/chat_routes.py
+++ b/routes/chat_routes.py
@@ -1701,9 +1701,21 @@ def _clear_orphaned_session_endpoint(sess, owner: str | None = None) -> bool:
from src.auth_helpers import owner_filter
q = owner_filter(q, ModelEndpoint, owner)
endpoints = q.all()
+ bound_id = getattr(sess, "endpoint_id", None)
for ep in endpoints:
+ if bound_id and ep.id != bound_id:
+ continue
if _session_url_matches_endpoint(sess.endpoint_url or "", ep.base_url or ""):
return False
+ if bound_id:
+ # Keep the identity so re-enabling/reconnecting A can recover A.
+ # Returning True stops chat; B must never replace a missing A.
+ sess.headers = {}
+ stored = db.query(DBSession).filter(DBSession.id == sess.id, DBSession.owner == owner).first()
+ if stored is not None:
+ stored.headers = {}
+ db.commit()
+ return True
db_session = db.query(DBSession).filter(DBSession.id == sess.id).first()
if db_session:
db_session.endpoint_url = ""
@@ -1803,6 +1815,13 @@ def _first_image_attachment(chat_handler, att_ids: List[str], owner: str | None
return None
+def _ts_or_zero(value) -> float:
+ try:
+ return float(value.timestamp()) if value else 0.0
+ except Exception:
+ return 0.0
+
+
def _recover_empty_session_model(sess, session_id: str, owner: str | None = None) -> bool:
"""Re-populate sess.model from the matching endpoint's cached models.
@@ -1835,10 +1854,19 @@ def _recover_empty_session_model(sess, session_id: str, owner: str | None = None
from src.auth_helpers import owner_filter
q = owner_filter(q, ModelEndpoint, owner)
endpoints = q.all()
- for cand in endpoints:
- if _session_url_matches_endpoint(sess.endpoint_url or "", cand.base_url or ""):
- ep = cand
- break
+ # Honour the session's exact endpoint binding first: two endpoints
+ # can share a provider URL (e.g. two ChatGPT Subscription accounts).
+ bound_id = getattr(sess, "endpoint_id", None) or None
+ if bound_id:
+ for cand in endpoints:
+ if cand.id == bound_id and _session_url_matches_endpoint(sess.endpoint_url or "", cand.base_url or ""):
+ ep = cand
+ break
+ if ep is None and not bound_id:
+ for cand in sorted(endpoints, key=lambda row: (_ts_or_zero(getattr(row, "created_at", None)), str(row.id))):
+ if _session_url_matches_endpoint(sess.endpoint_url or "", cand.base_url or ""):
+ ep = cand
+ break
if not ep:
return False
if not is_chatgpt_subscription:
@@ -1938,6 +1966,7 @@ def _reconcile_selected_route_from_request(
endpoint_url = ""
headers = None
+ resolved_endpoint_id = None
if selected_endpoint_id or selected_endpoint_url:
try:
from src.auth_helpers import owner_filter
@@ -1949,7 +1978,15 @@ def _reconcile_selected_route_from_request(
q = q.filter(ModelEndpoint.id == selected_endpoint_id)
if owner:
q = owner_filter(q, ModelEndpoint, owner)
- candidates = q.all() if selected_endpoint_url and not selected_endpoint_id else [q.first()]
+ if selected_endpoint_url and not selected_endpoint_id:
+ candidates = [row for row in q.all() if _session_url_matches_endpoint(selected_endpoint_url, row.base_url or "")]
+ bound_id = getattr(sess, "endpoint_id", None)
+ if bound_id:
+ candidates = [row for row in candidates if row.id == bound_id]
+ if len(candidates) != 1:
+ return False
+ else:
+ candidates = [q.first()]
ep = None
for cand in candidates:
if not cand:
@@ -1961,6 +1998,7 @@ def _reconcile_selected_route_from_request(
return False
endpoint_url = build_chat_url(normalize_base(ep.base_url or ""))
headers = build_headers(ep.api_key or "", ep.base_url or "") if ep.api_key else {}
+ resolved_endpoint_id = ep.id
finally:
db.close()
except Exception as e:
@@ -1975,18 +2013,29 @@ def _reconcile_selected_route_from_request(
and endpoint_url == (getattr(sess, "endpoint_url", "") or "")
)
headers_changed = dict(getattr(sess, "headers", None) or {}) != dict(headers or {})
- if not route_changed and not headers_changed:
+ binding_changed = bool(
+ resolved_endpoint_id
+ and resolved_endpoint_id != (getattr(sess, "endpoint_id", None) or None)
+ )
+ if not route_changed and not headers_changed and not binding_changed:
return False
sess.model = selected_model
sess.endpoint_url = endpoint_url
sess.headers = headers or {}
+ if resolved_endpoint_id:
+ sess.endpoint_id = resolved_endpoint_id
+ elif route_changed:
+ # The route moved without an explicit endpoint id: drop a stale binding
+ # rather than keep pointing at an endpoint the session no longer uses.
+ sess.endpoint_id = None
db = SessionLocal()
try:
db_session = db.query(DBSession).filter(DBSession.id == session_id).first()
if db_session:
db_session.model = selected_model
db_session.endpoint_url = endpoint_url
+ db_session.endpoint_id = getattr(sess, "endpoint_id", None) or None
db_session.headers = sess.headers or {}
db_session.updated_at = datetime.utcnow()
db.commit()
@@ -2087,6 +2136,10 @@ def setup_chat_routes(
if not supports_user_thinking_toggle(sess.model):
thinking_mode = "off"
owner = effective_user(request)
+ _reconcile_selected_route_from_request(request, sess, session, {
+ "selected_model": sess.model,
+ "selected_endpoint_id": chat_request.selected_endpoint_id,
+ }, owner=owner)
if _clear_orphaned_session_endpoint(sess, owner=owner):
raise HTTPException(400, "Selected model endpoint was removed. Pick another model in Settings.")
@@ -2102,6 +2155,8 @@ def setup_chat_routes(
if not (getattr(sess, "endpoint_url", "") or "").strip():
raise HTTPException(400, "Selected model endpoint is not configured")
+ resolve_session_auth(sess, session, owner=owner)
+
# Same allowed_models + daily-cap gate as chat_stream (mirror so the
# non-streaming path can't be used to bypass).
_enforce_chat_privileges(request, sess)
@@ -2169,7 +2224,7 @@ def setup_chat_routes(
sess.headers,
owner=owner,
policy=foreground_policy,
- selected_endpoint_id=chat_request.selected_endpoint_id,
+ selected_endpoint_id=chat_request.selected_endpoint_id or getattr(sess, "endpoint_id", None),
)
candidate_request_factory = None
selected_context_length = getattr(ctx, "context_length", 0)
@@ -3704,7 +3759,7 @@ def setup_chat_routes(
sess.headers,
owner=_user,
policy=_foreground_policy,
- selected_endpoint_id=selected_endpoint_id,
+ selected_endpoint_id=selected_endpoint_id or getattr(sess, "endpoint_id", None),
)
_chat_request_factory = None
_selected_context_length = getattr(ctx, "context_length", 0)
diff --git a/routes/chatgpt_subscription_routes.py b/routes/chatgpt_subscription_routes.py
index 9c695b371..a9a19b83a 100644
--- a/routes/chatgpt_subscription_routes.py
+++ b/routes/chatgpt_subscription_routes.py
@@ -1,28 +1,123 @@
-"""ChatGPT Subscription device-flow setup routes."""
+"""ChatGPT Subscription device-flow setup, multi-account and usage routes.
+
+One Odysseus owner may connect several independent ChatGPT subscriptions. Each
+connection is its own ``ProviderAuthSession`` + ``ModelEndpoint`` pair; the
+endpoint id decides which account a request is billed to. Labels are cosmetic
+only — stable ids drive lookup, reconnect, deletion and usage reads.
+"""
import json
import logging
import uuid
-from typing import Dict, Optional
+from typing import Any, Dict, Optional
-from fastapi import HTTPException, Request
+from fastapi import HTTPException, Request, Response
from core.database import ModelEndpoint, ProviderAuthSession, SessionLocal, utcnow_naive
+from core.middleware import require_admin
from routes.device_flow import (
DeviceFlowPoll,
DeviceFlowStart,
PendingDeviceFlowStore,
create_device_flow_router,
)
-from src.auth_helpers import get_current_user
+from src.auth_helpers import effective_user, get_current_user
from src import chatgpt_subscription
logger = logging.getLogger(__name__)
_DEVICE_FLOW_STORE = PendingDeviceFlowStore()
+_PROVIDER = chatgpt_subscription.CHATGPT_SUBSCRIPTION_PROVIDER
-def _provision_endpoint(tokens: Dict, owner: Optional[str]) -> Dict:
+
+def _owner_scope(query, model_cls, owner: Optional[str]):
+ return query.filter(model_cls.owner == owner)
+
+
+def _owner_chatgpt_auths(db, owner: Optional[str]):
+ q = db.query(ProviderAuthSession).filter(ProviderAuthSession.provider == _PROVIDER)
+ return _owner_scope(q, ProviderAuthSession, owner).order_by(ProviderAuthSession.created_at).all()
+
+
+def _endpoints_for_auth(db, auth_id: str):
+ return db.query(ModelEndpoint).filter(ModelEndpoint.provider_auth_id == auth_id).all()
+
+
+def _display_label(auth, ep) -> str:
+ """Display label from persisted metadata (never from credentials)."""
+ label = chatgpt_subscription.account_label_from_name(getattr(auth, "label", None))
+ if not label and ep is not None:
+ label = chatgpt_subscription.account_label_from_name(getattr(ep, "name", None))
+ return label
+
+
+def _assert_label_available(db, owner: Optional[str], label: str, *, exclude_auth_id: Optional[str] = None) -> None:
+ """Reject a label already used by another ChatGPT account of this owner."""
+ if not label:
+ return
+ for auth in _owner_chatgpt_auths(db, owner):
+ if exclude_auth_id and auth.id == exclude_auth_id:
+ continue
+ existing = _display_label(auth, None)
+ if not existing:
+ for ep in _endpoints_for_auth(db, auth.id):
+ existing = _display_label(auth, ep)
+ if existing:
+ break
+ if chatgpt_subscription.labels_conflict(existing, label):
+ raise ValueError(f"A ChatGPT subscription labelled '{label}' is already connected.")
+
+
+def _default_new_label(db, owner: Optional[str]) -> str:
+ """Label for a new connection when the user did not supply one.
+
+ The first account keeps the legacy unlabelled name so existing single
+ account setups look unchanged; later accounts get a distinguishable
+ ``account N`` label that is unique for this owner.
+ """
+ existing = _owner_chatgpt_auths(db, owner)
+ if not existing:
+ return ""
+ taken = set()
+ for auth in existing:
+ label = _display_label(auth, None)
+ if not label:
+ for ep in _endpoints_for_auth(db, auth.id):
+ label = _display_label(auth, ep)
+ if label:
+ break
+ if label:
+ taken.add(label.casefold())
+ n = len(existing) + 1
+ while f"account {n}".casefold() in taken:
+ n += 1
+ return f"account {n}"
+
+
+def _new_id(db, model_cls) -> str:
+ for _ in range(8):
+ candidate = str(uuid.uuid4())[:8]
+ if db.query(model_cls).filter(model_cls.id == candidate).first() is None:
+ return candidate
+ return uuid.uuid4().hex[:12]
+
+
+def _provision_endpoint(
+ tokens: Dict,
+ owner: Optional[str],
+ *,
+ label: str = "",
+ reconnect_auth_id: Optional[str] = None,
+ reconnect_endpoint_id: Optional[str] = None,
+) -> Dict:
+ """Create a new ChatGPT account (auth + endpoint) or refresh exactly one.
+
+ Without ``reconnect_auth_id`` a brand-new ``ProviderAuthSession`` and
+ ``ModelEndpoint`` are created even when the owner already has other ChatGPT
+ subscriptions. With it, only that owner-scoped auth row (and its endpoint)
+ is updated; every other account is left untouched.
+ """
access_token = tokens.get("access_token")
refresh_token = tokens.get("refresh_token")
if not access_token or not refresh_token:
@@ -32,22 +127,28 @@ def _provision_endpoint(tokens: Dict, owner: Optional[str]) -> Dict:
models = chatgpt_subscription.fetch_available_models(access_token)
if not models:
raise ValueError("ChatGPT Subscription connected, but no usable Codex models were discovered for this account.")
+ label = chatgpt_subscription.normalize_account_label(label)
db = SessionLocal()
try:
- auth = (
- db.query(ProviderAuthSession)
- .filter(
- ProviderAuthSession.provider == chatgpt_subscription.CHATGPT_SUBSCRIPTION_PROVIDER,
- ProviderAuthSession.owner == owner,
- )
- .first()
- )
- if auth is None:
+ auth = None
+ if reconnect_auth_id:
+ auth = chatgpt_subscription.find_owned_auth_session(db, reconnect_auth_id, owner)
+ if auth is None:
+ raise chatgpt_subscription.ChatGPTSubscriptionAuthNotFound(
+ "The ChatGPT subscription being reconnected no longer exists for this user."
+ )
+ # A reconnect keeps the existing label unless a new one was given.
+ if label:
+ _assert_label_available(db, owner, label, exclude_auth_id=auth.id)
+ else:
+ if not label:
+ label = _default_new_label(db, owner)
+ _assert_label_available(db, owner, label)
auth = ProviderAuthSession(
- id=str(uuid.uuid4())[:8],
- provider=chatgpt_subscription.CHATGPT_SUBSCRIPTION_PROVIDER,
+ id=_new_id(db, ProviderAuthSession),
+ provider=_PROVIDER,
owner=owner,
- label="ChatGPT Subscription",
+ label=chatgpt_subscription.endpoint_name_for_label(label),
base_url=base,
auth_mode="chatgpt",
)
@@ -57,31 +158,39 @@ def _provision_endpoint(tokens: Dict, owner: Optional[str]) -> Dict:
auth.refresh_token = refresh_token
auth.last_refresh = utcnow_naive()
auth.auth_mode = "chatgpt"
+ if label:
+ auth.label = chatgpt_subscription.endpoint_name_for_label(label)
- ep = (
- db.query(ModelEndpoint)
- .filter(
- ModelEndpoint.base_url == base,
- ModelEndpoint.provider_auth_id == auth.id,
- ModelEndpoint.owner == owner,
- )
- .first()
- )
+ ep = None
+ if reconnect_auth_id:
+ ep_q = db.query(ModelEndpoint).filter(ModelEndpoint.provider_auth_id == auth.id)
+ ep_q = _owner_scope(ep_q, ModelEndpoint, owner)
+ if reconnect_endpoint_id:
+ ep = ep_q.filter(ModelEndpoint.id == reconnect_endpoint_id).first()
+ if ep is None:
+ raise chatgpt_subscription.ChatGPTSubscriptionAuthNotFound(
+ "The ChatGPT subscription endpoint no longer exists for this user."
+ )
+ else:
+ ep = ep_q.order_by(ModelEndpoint.created_at).first()
if ep is None:
ep = ModelEndpoint(
- id=str(uuid.uuid4())[:8],
- name="ChatGPT Subscription",
+ id=_new_id(db, ModelEndpoint),
+ name=chatgpt_subscription.endpoint_name_for_label(label),
base_url=base,
model_type="llm",
endpoint_kind="api",
owner=owner,
)
db.add(ep)
- ep.name = "ChatGPT Subscription"
+ if label or not (ep.name or "").strip():
+ ep.name = chatgpt_subscription.endpoint_name_for_label(label)
ep.base_url = base
ep.api_key = None
ep.provider_auth_id = auth.id
ep.is_enabled = True
+ # ChatGPT provides inference only. Odysseus is the only agent: no
+ # provider-native tool schemas are ever sent on this route.
ep.supports_tools = False
ep.model_type = "llm"
ep.endpoint_kind = "api"
@@ -93,10 +202,14 @@ def _provision_endpoint(tokens: Dict, owner: Optional[str]) -> Dict:
"name": ep.name,
"base_url": ep.base_url,
"models": models,
+ "provider_auth_id": auth.id,
+ "account_label": _display_label(auth, ep),
+ "reconnected": bool(reconnect_auth_id),
}
finally:
db.close()
+ chatgpt_subscription.USAGE_CACHE.invalidate(result["provider_auth_id"])
try:
from routes.model_routes import _invalidate_models_cache
@@ -106,7 +219,53 @@ def _provision_endpoint(tokens: Dict, owner: Optional[str]) -> Dict:
return result
-def _start_device_flow(request: Request, _form) -> DeviceFlowStart:
+def _form_value(form, key: str) -> str:
+ try:
+ value = form.get(key) if form is not None else None
+ except Exception:
+ value = None
+ return str(value).strip() if value is not None else ""
+
+
+def _start_device_flow(request: Request, form) -> DeviceFlowStart:
+ owner = effective_user(request) or None
+ try:
+ label = chatgpt_subscription.normalize_account_label(_form_value(form, "label"))
+ except ValueError as exc:
+ raise HTTPException(400, str(exc))
+ reconnect_auth_id = _form_value(form, "reconnect_auth_id") or None
+ reconnect_endpoint_id = _form_value(form, "reconnect_endpoint_id") or None
+ if reconnect_endpoint_id and not reconnect_auth_id:
+ raise HTTPException(400, "Reconnect requires an account id")
+
+ # Validate the intended operation up front so the user is not sent through
+ # OAuth for a request that can never be provisioned.
+ db = SessionLocal()
+ try:
+ if reconnect_auth_id:
+ auth = chatgpt_subscription.find_owned_auth_session(db, reconnect_auth_id, owner)
+ if auth is None:
+ raise HTTPException(404, "ChatGPT subscription account not found")
+ if reconnect_endpoint_id:
+ ep_q = db.query(ModelEndpoint).filter(
+ ModelEndpoint.id == reconnect_endpoint_id,
+ ModelEndpoint.provider_auth_id == auth.id,
+ )
+ if _owner_scope(ep_q, ModelEndpoint, owner).first() is None:
+ raise HTTPException(404, "ChatGPT subscription endpoint not found")
+ if label:
+ try:
+ _assert_label_available(db, owner, label, exclude_auth_id=auth.id)
+ except ValueError as exc:
+ raise HTTPException(409, str(exc))
+ else:
+ try:
+ _assert_label_available(db, owner, label)
+ except ValueError as exc:
+ raise HTTPException(409, str(exc))
+ finally:
+ db.close()
+
try:
data = chatgpt_subscription.request_device_code()
except Exception as exc:
@@ -116,37 +275,82 @@ def _start_device_flow(request: Request, _form) -> DeviceFlowStart:
user_code = data.get("user_code")
if not device_auth_id or not user_code:
raise HTTPException(502, "ChatGPT did not return a complete device code")
- verification_uri = data.get("verification_uri") or f"{chatgpt_subscription.CHATGPT_OAUTH_ISSUER}/codex/device"
+ # Never pass an arbitrary upstream URL into an authorization link.
+ from urllib.parse import urlsplit
+ fallback_uri = f"{chatgpt_subscription.CHATGPT_OAUTH_ISSUER}/codex/device"
+ verification_uri = data.get("verification_uri") or fallback_uri
+ try:
+ parsed = urlsplit(verification_uri)
+ if parsed.scheme != "https" or parsed.netloc != "auth.openai.com":
+ verification_uri = fallback_uri
+ except (TypeError, ValueError):
+ verification_uri = fallback_uri
+ # The pending payload carries only what provisioning needs: the device
+ # handle, the owner and the intended operation. No access/refresh tokens.
+ pending: Dict[str, Any] = {
+ "device_auth_id": device_auth_id,
+ "user_code": user_code,
+ "owner": owner,
+ "label": label,
+ "reconnect_auth_id": reconnect_auth_id,
+ "reconnect_endpoint_id": reconnect_endpoint_id,
+ }
+ response: Dict[str, Any] = {
+ "user_code": user_code,
+ "verification_uri": verification_uri,
+ "mode": "reconnect" if reconnect_auth_id else "connect",
+ }
+ if label:
+ response["account_label"] = label
return DeviceFlowStart(
- pending={
- "device_auth_id": device_auth_id,
- "user_code": user_code,
- "owner": get_current_user(request) or None,
- },
- response={
- "user_code": user_code,
- "verification_uri": verification_uri,
- },
+ pending=pending,
+ response=response,
interval=int(data.get("interval") or 5),
expires_in=int(data.get("expires_in") or 900),
)
-def _poll_device_flow(_request: Request, pending: Dict) -> DeviceFlowPoll:
+def _poll_device_flow(request: Request, pending: Dict) -> DeviceFlowPoll:
+ # The poller must be the same user who started the flow: a poll id is not
+ # a bearer for provisioning into someone else's account list.
+ current_owner = effective_user(request) or None
+ if (pending.get("owner") or None) != current_owner:
+ raise HTTPException(403, "This sign-in belongs to another user")
+ if pending.get("reconnect_auth_id"):
+ db = SessionLocal()
+ try:
+ auth = chatgpt_subscription.find_owned_auth_session(db, pending["reconnect_auth_id"], current_owner)
+ if auth is None:
+ raise HTTPException(404, "ChatGPT subscription account not found")
+ if pending.get("reconnect_endpoint_id"):
+ ep = _owner_scope(db.query(ModelEndpoint).filter(
+ ModelEndpoint.id == pending["reconnect_endpoint_id"],
+ ModelEndpoint.provider_auth_id == auth.id,
+ ), ModelEndpoint, current_owner).first()
+ if ep is None:
+ raise HTTPException(404, "ChatGPT subscription endpoint not found")
+ finally:
+ db.close()
try:
data = chatgpt_subscription.poll_device_auth(pending["device_auth_id"], pending["user_code"])
except Exception as exc:
- logger.debug("ChatGPT device poll failed: %s", exc)
- return DeviceFlowPoll.pending(str(exc))
+ logger.debug("ChatGPT device poll failed: %s", type(exc).__name__)
+ return DeviceFlowPoll.pending("Sign-in status temporarily unavailable")
authorization_code = data.get("authorization_code")
code_verifier = data.get("code_verifier")
if authorization_code and code_verifier:
try:
tokens = chatgpt_subscription.exchange_authorization_code(authorization_code, code_verifier)
- result = _provision_endpoint(tokens, pending["owner"])
+ result = _provision_endpoint(
+ tokens,
+ pending.get("owner"),
+ label=pending.get("label") or "",
+ reconnect_auth_id=pending.get("reconnect_auth_id") or None,
+ reconnect_endpoint_id=pending.get("reconnect_endpoint_id") or None,
+ )
except Exception as exc:
- logger.exception("ChatGPT Subscription endpoint provisioning failed")
+ logger.warning("ChatGPT Subscription endpoint provisioning failed: %s", type(exc).__name__)
raise chatgpt_subscription.to_http_exception(exc)
return DeviceFlowPoll.authorized(result)
@@ -157,14 +361,97 @@ def _poll_device_flow(_request: Request, pending: Dict) -> DeviceFlowPoll:
return DeviceFlowPoll.slow_down(int(data.get("interval") or 0) or None)
if err in ("expired_token", "access_denied", "denied"):
return DeviceFlowPoll.failed(err)
- return DeviceFlowPoll.pending(err or "unknown")
+ return DeviceFlowPoll.pending("unknown")
+
+
+# ── Account listing / usage ─────────────────────────────────────────────────
+
+def _account_summary(auth, endpoints) -> Dict[str, Any]:
+ ep = endpoints[0] if endpoints else None
+ return {
+ "auth_id": auth.id,
+ "label": _display_label(auth, ep),
+ "name": (ep.name if ep is not None else None) or auth.label or chatgpt_subscription.CHATGPT_SUBSCRIPTION_LEGACY_NAME,
+ "endpoint_ids": [row.id for row in endpoints],
+ "connected_at": auth.created_at.isoformat() if getattr(auth, "created_at", None) else None,
+ "last_refresh": auth.last_refresh.isoformat() if getattr(auth, "last_refresh", None) else None,
+ "connected": bool(auth.refresh_token),
+ }
+
+
+def usage_error_payload(exc: chatgpt_subscription.ChatGPTUsageUnavailable) -> Dict[str, Any]:
+ """Safe, credential-free payload for a failed usage read."""
+ return {
+ "available": False,
+ "reason": exc.reason,
+ "message": str(exc),
+ "status_code": exc.status_code,
+ "reconnect_suggested": exc.reason == "reauth",
+ }
+
+
+def _load_owned_account(request: Request, auth_id: str):
+ """Admin gate + owner scope for account-level operations."""
+ require_admin(request)
+ owner = effective_user(request) or get_current_user(request) or None
+ db = SessionLocal()
+ try:
+ auth = chatgpt_subscription.find_owned_auth_session(db, auth_id, owner)
+ if auth is None:
+ raise HTTPException(404, "ChatGPT subscription account not found")
+ endpoints = _endpoints_for_auth(db, auth.id)
+ return owner, _account_summary(auth, endpoints)
+ finally:
+ db.close()
def setup_chatgpt_subscription_routes():
- return create_device_flow_router(
+ router = create_device_flow_router(
prefix="/api/chatgpt-subscription",
tags=["chatgpt-subscription"],
store=_DEVICE_FLOW_STORE,
start_flow=_start_device_flow,
poll_flow=_poll_device_flow,
)
+
+ @router.get("/accounts")
+ def list_accounts(request: Request):
+ require_admin(request)
+ owner = effective_user(request) or get_current_user(request) or None
+ db = SessionLocal()
+ try:
+ return [
+ _account_summary(auth, _endpoints_for_auth(db, auth.id))
+ for auth in _owner_chatgpt_auths(db, owner)
+ ]
+ finally:
+ db.close()
+
+ @router.get("/accounts/{auth_id}/usage")
+ def account_usage(auth_id: str, request: Request, refresh: bool = False, response: Response = None):
+ """Read-only, owner-scoped usage for exactly one ChatGPT account.
+
+ Failures here are telemetry failures only: the model endpoint is never
+ disabled, credentials are never destroyed and no reconnect is started.
+ """
+ if response is not None:
+ response.headers["Cache-Control"] = "no-store"
+ owner, account = _load_owned_account(request, auth_id)
+ try:
+ usage = chatgpt_subscription.get_account_usage(account["auth_id"], owner=owner, force_refresh=refresh)
+ except chatgpt_subscription.ChatGPTSubscriptionAuthNotFound:
+ raise HTTPException(404, "ChatGPT subscription account not found")
+ except chatgpt_subscription.ChatGPTUsageUnavailable as exc:
+ payload = usage_error_payload(exc)
+ payload["account"] = account
+ return payload
+ except Exception as exc: # pragma: no cover - defensive
+ logger.warning("ChatGPT usage read failed for auth %s: %s", auth_id, type(exc).__name__)
+ payload = usage_error_payload(
+ chatgpt_subscription.ChatGPTUsageUnavailable("upstream", "ChatGPT usage is unavailable.")
+ )
+ payload["account"] = account
+ return payload
+ return {"available": True, "account": account, "usage": usage}
+
+ return router
diff --git a/routes/history/history_routes.py b/routes/history/history_routes.py
index 411a54cbd..26d3dca48 100644
--- a/routes/history/history_routes.py
+++ b/routes/history/history_routes.py
@@ -315,6 +315,7 @@ def setup_history_routes(session_manager, upload_handler=None) -> APIRouter:
"history": history_dict,
"model": db_session.model,
"endpoint_url": db_session.endpoint_url,
+ "endpoint_id": getattr(db_session, "endpoint_id", None),
"name": db_session.name,
"offset": page_offset,
"limit": page_limit,
@@ -378,6 +379,7 @@ def setup_history_routes(session_manager, upload_handler=None) -> APIRouter:
"history": history_dict,
"model": session.model,
"endpoint_url": session.endpoint_url,
+ "endpoint_id": getattr(session, "endpoint_id", None),
"name": session.name,
}
@@ -802,6 +804,7 @@ def setup_history_routes(session_manager, upload_handler=None) -> APIRouter:
model=source.model,
rag=False,
owner=getattr(source, 'owner', None),
+ endpoint_id=getattr(source, 'endpoint_id', None),
)
# Copy messages up to keep_count
@@ -820,8 +823,6 @@ def setup_history_routes(session_manager, upload_handler=None) -> APIRouter:
except Exception:
logger.debug("session_created event dispatch failed", exc_info=True)
- from src.model_profiles import supports_user_thinking_toggle
- thinking_supported = supports_user_thinking_toggle(session.model)
return {
"status": "ok",
"id": new_id,
@@ -882,6 +883,7 @@ def setup_history_routes(session_manager, upload_handler=None) -> APIRouter:
"session_id": session_id,
"model": session.model,
"endpoint_url": session.endpoint_url,
+ "endpoint_id": getattr(session, "endpoint_id", None),
"used_tokens": used,
"context_length": ctx_len,
"context_percent": pct,
diff --git a/routes/model_routes.py b/routes/model_routes.py
index 5238b1746..069aeb322 100644
--- a/routes/model_routes.py
+++ b/routes/model_routes.py
@@ -695,6 +695,11 @@ def _delete_orphaned_provider_auth(db, auth_id: Optional[str], exclude_ep_id: Op
if auth_row is None:
return False
db.delete(auth_row)
+ try:
+ from src.chatgpt_subscription import USAGE_CACHE
+ USAGE_CACHE.invalidate(auth_id)
+ except Exception:
+ pass
return True
@@ -738,7 +743,7 @@ def _resolve_probe_key(ep) -> Optional[str]:
_base, key = resolve_endpoint_runtime(ep, owner=getattr(ep, "owner", None))
return key
except Exception as exc:
- logger.warning("Probe key resolution failed for %s: %s", getattr(ep, "id", "?"), exc)
+ logger.warning("Probe key resolution failed for %s: %s", getattr(ep, "id", "?"), type(exc).__name__)
return None
@@ -1410,6 +1415,35 @@ def _picker_models_for_endpoint(ep, base_url: str, kind: str):
), pinned
+def _chatgpt_endpoint_visible(ep: Any, request: Request) -> bool:
+ from src.chatgpt_subscription import is_chatgpt_subscription_base
+ if not is_chatgpt_subscription_base(getattr(ep, "base_url", "") or ""):
+ return True
+ try:
+ user = effective_user(request)
+ except AttributeError:
+ user = getattr(getattr(request, "state", None), "current_user", None)
+ return (getattr(ep, "owner", None) or None) == (user or None)
+
+
+def _provider_account_metadata(ep: Any) -> Dict[str, Any]:
+ """Non-secret account metadata for session-backed provider endpoints."""
+ auth_id = getattr(ep, "provider_auth_id", None)
+ if not auth_id:
+ return {"provider_auth_id": None, "provider": None, "account_label": None}
+ base = getattr(ep, "base_url", "") or ""
+ provider = None
+ account_label = None
+ try:
+ from src.chatgpt_subscription import account_label_from_name, is_chatgpt_subscription_base
+ if is_chatgpt_subscription_base(base):
+ provider = "chatgpt-subscription"
+ account_label = account_label_from_name(getattr(ep, "name", None)) or None
+ except Exception:
+ provider = None
+ return {"provider_auth_id": auth_id, "provider": provider, "account_label": account_label}
+
+
def _api_key_fingerprint(api_key: Optional[str]) -> str:
"""Stable, non-secret label for distinguishing same-URL credentials."""
key = (api_key or "").strip()
@@ -1444,8 +1478,10 @@ def setup_model_routes(model_discovery):
_REFRESH_FAILURE_BASE = 300.0
_REFRESH_FAILURE_MAX = 3600.0
- def _refresh_key(base: str, api_key: Optional[str]) -> str:
- return f"{base.rstrip('/')}\x00{api_key or ''}"
+ def _refresh_key(base: str, api_key: Optional[str], provider_auth_id: Optional[str] = None) -> str:
+ # Session-backed endpoints carry no static key; include their auth id
+ # so two accounts on one provider URL never share refresh state.
+ return f"{base.rstrip('/')}\x00{api_key or ''}\x00{provider_auth_id or ''}"
def _ts(value: Any) -> float:
try:
@@ -1466,7 +1502,7 @@ def setup_model_routes(model_discovery):
category = _classify_endpoint(base, kind)
mode = _endpoint_refresh_mode(ep, kind)
cached = _cached_model_ids(ep)
- key = _refresh_key(base, getattr(ep, "api_key", None))
+ key = _refresh_key(base, getattr(ep, "api_key", None), getattr(ep, "provider_auth_id", None))
state = _refresh_state.get(key, {})
info = {
@@ -1528,9 +1564,10 @@ def setup_model_routes(model_discovery):
ok, info = _should_refresh_endpoint(ep, now, force=force)
if not ok:
continue
+ credential = _resolve_probe_key(ep)
groups.setdefault(info["key"], {
"base": info["base"],
- "api_key": info["api_key"],
+ "api_key": credential,
"timeout": info["timeout"],
"endpoint_ids": [],
})["endpoint_ids"].append(info["id"])
@@ -1606,6 +1643,9 @@ def setup_model_routes(model_discovery):
db.close()
for ep in endpoints:
+ from src.chatgpt_subscription import is_chatgpt_subscription_base
+ if is_chatgpt_subscription_base(ep.base_url or "") and (ep.owner or None) != (owner or None):
+ continue
base = _normalize_base(ep.base_url)
provider = _safe_detect_provider(base)
ep_model_type = getattr(ep, "model_type", None) or "llm"
@@ -1858,8 +1898,8 @@ def setup_model_routes(model_discovery):
# Cache endpoint lookups
if ep_id and ep_id not in endpoints_cache:
ep = db.query(ModelEndpoint).filter(ModelEndpoint.id == ep_id).first()
- if ep:
- endpoints_cache[ep_id] = {"base_url": ep.base_url, "api_key": ep.api_key}
+ if ep and _chatgpt_endpoint_visible(ep, request):
+ endpoints_cache[ep_id] = {"base_url": ep.base_url, "api_key": _resolve_probe_key(ep)}
ep_data = endpoints_cache.get(ep_id)
if not ep_data:
# Try to find by base_url from the model's endpoint field
@@ -1894,11 +1934,13 @@ def setup_model_routes(model_discovery):
# Detach from session
ep_data = []
for ep in endpoints:
+ if not _chatgpt_endpoint_visible(ep, request):
+ continue
ep_data.append({
"id": ep.id,
"name": ep.name,
"base_url": ep.base_url,
- "api_key": ep.api_key,
+ "api_key": _resolve_probe_key(ep),
})
finally:
db.close()
@@ -1983,6 +2025,8 @@ def setup_model_routes(model_discovery):
results = []
upgraded_legacy_pins = False
for r in rows:
+ if not _chatgpt_endpoint_visible(r, request):
+ continue
all_models = _cached_model_ids(r)
hidden = _hidden_model_ids(r)
pinned = _normalize_model_ids(getattr(r, "pinned_models", None))
@@ -2022,6 +2066,7 @@ def setup_model_routes(model_discovery):
"model_refresh_mode": _endpoint_refresh_mode(r, kind),
"model_refresh_interval": getattr(r, "model_refresh_interval", None),
"model_refresh_timeout": getattr(r, "model_refresh_timeout", None),
+ **_provider_account_metadata(r),
})
if upgraded_legacy_pins:
db.commit()
@@ -2096,6 +2141,8 @@ def setup_model_routes(model_discovery):
existing = None
_empty_key_existing = None
for _candidate in _same_url_rows:
+ if getattr(_candidate, "provider_auth_id", None):
+ continue # OAuth account mutations require its explicit identity.
_candidate_key = (getattr(_candidate, "api_key", None) or "").strip()
if _candidate_key == _incoming_api_key:
existing = _candidate
@@ -2192,6 +2239,9 @@ def setup_model_routes(model_discovery):
try:
_st_raw = (supports_tools or "").strip().lower()
_st = True if _st_raw in ("true", "1", "yes") else (False if _st_raw in ("false", "0", "no") else None)
+ from src.chatgpt_subscription import is_chatgpt_subscription_base
+ if is_chatgpt_subscription_base(base_url):
+ _st = False
_pinned = _normalize_model_ids(pinned_models)
# Stamp owner so the picker only shows this endpoint to the admin
# who added it. Pass `shared=true` to mark it null-owner (visible
@@ -2305,9 +2355,9 @@ def setup_model_routes(model_discovery):
db = SessionLocal()
try:
ep = db.query(ModelEndpoint).filter(ModelEndpoint.id == ep_id).first()
- if not ep:
+ if not ep or not _chatgpt_endpoint_visible(ep, request):
raise HTTPException(404, "Endpoint not found")
- ep_data = {"id": ep.id, "name": ep.name, "base_url": ep.base_url, "api_key": ep.api_key}
+ ep_data = {"id": ep.id, "name": ep.name, "base_url": ep.base_url, "api_key": _resolve_probe_key(ep)}
finally:
db.close()
@@ -2361,7 +2411,7 @@ def setup_model_routes(model_discovery):
db = SessionLocal()
try:
ep = db.query(ModelEndpoint).filter(ModelEndpoint.id == ep_id).first()
- if not ep:
+ if not ep or not _chatgpt_endpoint_visible(ep, request):
raise HTTPException(404, "Endpoint not found")
hidden = _hidden_model_ids(ep)
all_models = _cached_model_ids(ep)
@@ -2372,7 +2422,7 @@ def setup_model_routes(model_discovery):
category = _classify_endpoint(base, kind)
timeout = _manual_refresh_timeout(ep, category, refresh_timeout)
try:
- probed = _probe_endpoint(base, ep.api_key, timeout=timeout)
+ probed = _probe_endpoint(base, _resolve_probe_key(ep), timeout=timeout)
except Exception as exc:
logger.warning("Manual model refresh failed for endpoint %s at %s: %s", ep_id, base, exc)
probed = []
@@ -2416,7 +2466,7 @@ def setup_model_routes(model_discovery):
db = SessionLocal()
try:
ep = db.query(ModelEndpoint).filter(ModelEndpoint.id == ep_id).first()
- if not ep:
+ if not ep or not _chatgpt_endpoint_visible(ep, request):
raise HTTPException(404, "Endpoint not found")
body = await request.json()
if not isinstance(body, dict):
@@ -2573,7 +2623,7 @@ def setup_model_routes(model_discovery):
db = SessionLocal()
try:
ep = db.query(ModelEndpoint).filter(ModelEndpoint.id == ep_id).first()
- if not ep:
+ if not ep or not _chatgpt_endpoint_visible(ep, request):
raise HTTPException(404, "Endpoint not found")
if body:
if "supports_tools" in body:
@@ -2627,6 +2677,9 @@ def setup_model_routes(model_discovery):
ep.base_url = _new_base
else:
ep.is_enabled = not ep.is_enabled
+ from src.chatgpt_subscription import is_chatgpt_subscription_base
+ if is_chatgpt_subscription_base(ep.base_url or ""):
+ ep.supports_tools = False
db.commit()
_invalidate_models_cache()
_local_probe_cache["data"] = None
@@ -2684,7 +2737,7 @@ def setup_model_routes(model_discovery):
}
return sess in variants or sess.startswith(base + "/")
- def _clear_sessions_for_endpoint(db, base_url: str) -> int:
+ def _clear_sessions_for_endpoint(db, base_url: str, endpoint_id: str, owner) -> int:
"""Drop stored auth for sessions using an endpoint being deleted.
Keep the session's endpoint URL and model intact. If the admin is
@@ -2694,15 +2747,17 @@ def setup_model_routes(model_discovery):
matching enabled endpoint exists.
"""
cleared = 0
- rows = db.query(DbSession).filter(DbSession.endpoint_url.isnot(None)).all()
+ rows = db.query(DbSession).filter(DbSession.endpoint_url.isnot(None), DbSession.owner == owner).all()
for row in rows:
+ if getattr(row, "endpoint_id", None) not in (None, endpoint_id):
+ continue
if _session_uses_endpoint_url(row.endpoint_url or "", base_url):
row.headers = {}
row.updated_at = datetime.utcnow()
cleared += 1
return cleared
- def _clear_loaded_sessions_for_endpoint(base_url: str) -> int:
+ def _clear_loaded_sessions_for_endpoint(base_url: str, endpoint_id: str, owner) -> int:
try:
from src.ai_interaction import get_session_manager
manager = get_session_manager()
@@ -2713,6 +2768,8 @@ def setup_model_routes(model_discovery):
cleared = 0
try:
for sess in list(getattr(manager, "sessions", {}).values()):
+ if getattr(sess, "owner", None) != owner or getattr(sess, "endpoint_id", None) not in (None, endpoint_id):
+ continue
if _session_uses_endpoint_url(getattr(sess, "endpoint_url", "") or "", base_url):
sess.headers = {}
cleared += 1
@@ -2732,13 +2789,13 @@ def setup_model_routes(model_discovery):
db = SessionLocal()
try:
ep = db.query(ModelEndpoint).filter(ModelEndpoint.id == ep_id).first()
- if not ep:
+ if not ep or not _chatgpt_endpoint_visible(ep, request):
raise HTTPException(404, "Endpoint not found")
# Clean up any settings that reference this endpoint
cleared = _clear_settings_for_endpoint(ep_id)
cleared_user_preferences = _clear_user_prefs_for_endpoint(ep_id)
- cleared_sessions = _clear_sessions_for_endpoint(db, ep.base_url)
- cleared_loaded_sessions = _clear_loaded_sessions_for_endpoint(ep.base_url)
+ cleared_sessions = _clear_sessions_for_endpoint(db, ep.base_url, ep.id, ep.owner)
+ cleared_loaded_sessions = _clear_loaded_sessions_for_endpoint(ep.base_url, ep.id, ep.owner)
auth_id = getattr(ep, "provider_auth_id", None)
db.delete(ep)
cleared_provider_auth = _delete_orphaned_provider_auth(db, auth_id, exclude_ep_id=ep_id)
diff --git a/routes/session_routes.py b/routes/session_routes.py
index 58e262696..3714a8347 100644
--- a/routes/session_routes.py
+++ b/routes/session_routes.py
@@ -455,13 +455,17 @@ def setup_session_routes(
else (s.created_at.isoformat() if s.created_at else None))
)
matches = endpoint_routes.get((s.endpoint_url or '').rstrip('/'), [])
- selected_endpoint = matches[0] if len(matches) == 1 else None
+ bound_id = getattr(s, "endpoint_id", None)
+ selected_endpoint = (
+ next((ep for ep in matches if ep.id == bound_id), None)
+ if bound_id else (matches[0] if len(matches) == 1 else None)
+ )
sessions.append({
"id": s.id,
"name": s.name,
"model": _public_model(s.name, s.model),
"endpoint_url": s.endpoint_url,
- "endpoint_id": selected_endpoint.id if selected_endpoint else None,
+ "endpoint_id": bound_id or (selected_endpoint.id if selected_endpoint else None),
"endpoint_name": selected_endpoint.name if selected_endpoint else None,
"rag": s.rag,
"archived": s.archived,
@@ -590,6 +594,7 @@ def setup_session_routes(
rag=str(rag).lower() == "true" if rag else False,
owner=user,
cwd=cwd or None,
+ endpoint_id=endpoint_id.strip() if endpoint_id else None,
)
# Set auth headers for custom API-key endpoints
resolved_key = request_api_key
@@ -597,7 +602,8 @@ def setup_session_routes(
if not resolved_key and endpoint_api_key:
resolved_key = endpoint_api_key
resolved_base = endpoint_base_url
- if resolved_key:
+ from src.chatgpt_subscription import is_chatgpt_subscription_base
+ if resolved_key and not is_chatgpt_subscription_base(endpoint_url):
from src.endpoint_resolver import build_headers
session.headers = build_headers(resolved_key, resolved_base)
_persist_session_headers(sid, session.headers)
@@ -685,8 +691,14 @@ def setup_session_routes(
endpoint_url = build_chat_url(normalize_base(endpoint_base_url))
finally:
_db.close()
+ previous_url = session.endpoint_url
session.model = model
session.endpoint_url = endpoint_url
+ # A registered endpoint id pins the exact route; a raw URL switch
+ # (admin only) clears any previous binding.
+ session.endpoint_id = (endpoint_id or "").strip() or (
+ getattr(session, "endpoint_id", None) if endpoint_url == previous_url else None
+ )
# Update auth headers from the endpoint's stored API key
if endpoint_api_key:
from src.endpoint_resolver import build_headers
@@ -700,6 +712,7 @@ def setup_session_routes(
if db_session:
db_session.model = model
db_session.endpoint_url = endpoint_url
+ db_session.endpoint_id = session.endpoint_id
db_session.headers = session.headers or {}
db_session.updated_at = utcnow_naive()
db.commit()
diff --git a/src/chatgpt_subscription.py b/src/chatgpt_subscription.py
index e65ccbc8d..0ee313b3e 100644
--- a/src/chatgpt_subscription.py
+++ b/src/chatgpt_subscription.py
@@ -9,10 +9,13 @@ from __future__ import annotations
import base64
import json
+import math
import os
+import re
import threading
import time
-from typing import Any, Dict, Optional
+import unicodedata
+from typing import Any, Dict, List, Optional
import httpx
from fastapi import HTTPException
@@ -22,6 +25,16 @@ DEFAULT_CHATGPT_SUBSCRIPTION_BASE_URL = (
or "https://chatgpt.com/backend-api/codex"
)
CHATGPT_SUBSCRIPTION_PROVIDER = "chatgpt-subscription"
+# Legacy single-account endpoint/auth name. Rows provisioned before
+# multi-account support keep this name and stay functional.
+CHATGPT_SUBSCRIPTION_LEGACY_NAME = "ChatGPT Subscription"
+CHATGPT_ACCOUNT_LABEL_MAX_LENGTH = 40
+# Read-only account usage (rate-limit windows) on the authenticated ChatGPT
+# backend. Mirrors openai/codex ``backend-client`` ``PathStyle::ChatGptApi``:
+# ``{base}/wham/usage`` where base is ``https://chatgpt.com/backend-api``.
+CHATGPT_USAGE_URL = "https://chatgpt.com/backend-api/wham/usage"
+CHATGPT_USAGE_TIMEOUT_SECONDS = 8.0
+CHATGPT_USAGE_CACHE_TTL_SECONDS = 45.0
CHATGPT_OAUTH_CLIENT_ID = "app_EMoamEEZ73f0CkXaXp7hrann"
CHATGPT_OAUTH_TOKEN_URL = "https://auth.openai.com/oauth/token"
CHATGPT_OAUTH_ISSUER = "https://auth.openai.com"
@@ -87,6 +100,53 @@ def chatgpt_headers(access_token: Optional[str]) -> Dict[str, str]:
return headers
+# ── Account labels ──────────────────────────────────────────────────────────
+
+_LABEL_WHITESPACE_RE = re.compile(r"\s+")
+
+
+def normalize_account_label(raw: Any) -> str:
+ """Return a trimmed, display-safe account label ("" when absent).
+
+ Labels are purely cosmetic: they never participate in authentication or
+ authorization (stable auth/endpoint ids do). Control characters are
+ stripped, whitespace collapsed and the length bounded so the label is safe
+ to echo in Settings, the model picker and provenance metadata.
+ """
+ if raw is None:
+ return ""
+ text = str(raw)
+ text = "".join(ch for ch in text if unicodedata.category(ch)[0] != "C")
+ text = _LABEL_WHITESPACE_RE.sub(" ", text).strip()
+ if len(text) > CHATGPT_ACCOUNT_LABEL_MAX_LENGTH:
+ raise ValueError(
+ f"Account label must be at most {CHATGPT_ACCOUNT_LABEL_MAX_LENGTH} characters."
+ )
+ return text
+
+
+def endpoint_name_for_label(label: str) -> str:
+ """User-visible endpoint name for a ChatGPT account label."""
+ label = (label or "").strip()
+ if not label:
+ return CHATGPT_SUBSCRIPTION_LEGACY_NAME
+ return f"ChatGPT · {label}"
+
+
+def account_label_from_name(name: Optional[str], fallback: str = "") -> str:
+ """Inverse of :func:`endpoint_name_for_label` for display purposes."""
+ text = (name or "").strip()
+ if text.startswith("ChatGPT · "):
+ return text[len("ChatGPT · "):].strip() or fallback
+ if text == CHATGPT_SUBSCRIPTION_LEGACY_NAME:
+ return fallback
+ return text or fallback
+
+
+def labels_conflict(a: str, b: str) -> bool:
+ return bool(a) and bool(b) and a.casefold() == b.casefold()
+
+
def fetch_available_models(access_token: str, timeout: float = 10.0) -> list[str]:
if not access_token:
return []
@@ -135,14 +195,8 @@ def _raise_for_oauth_response(response: httpx.Response, action: str) -> None:
err = payload.get("error") if isinstance(payload, dict) else None
if isinstance(err, dict):
code = str(err.get("code") or err.get("type") or "").strip()
- msg = err.get("message")
- if msg:
- message = f"ChatGPT Subscription {action} failed: {msg}"
elif isinstance(err, str):
code = err.strip()
- desc = payload.get("error_description") or payload.get("message")
- if desc:
- message = f"ChatGPT Subscription {action} failed: {desc}"
except Exception:
pass
if response.status_code == 429:
@@ -259,8 +313,7 @@ def resolve_runtime_credentials(auth_id: str, owner: Optional[str] = None, *, fo
ProviderAuthSession.id == auth_id,
ProviderAuthSession.provider == CHATGPT_SUBSCRIPTION_PROVIDER,
)
- if owner:
- q = q.filter(ProviderAuthSession.owner == owner)
+ q = q.filter(ProviderAuthSession.owner == owner)
row = q.first()
if row is None:
raise ChatGPTSubscriptionAuthNotFound("ChatGPT Subscription credentials were not found for this user.")
@@ -291,12 +344,359 @@ def resolve_runtime_credentials(auth_id: str, owner: Optional[str] = None, *, fo
db.close()
+def find_owned_auth_session(db, auth_id: str, owner: Optional[str]):
+ """Return the owner-scoped ChatGPT ProviderAuthSession row or None.
+
+ OAuth credentials belong to exactly one owner, including the legacy
+ anonymous owner. Labels are never used for lookup.
+ """
+ ProviderAuthSession, _SessionLocal, _now = _database_handles()
+ auth_id = (auth_id or "").strip()
+ if not auth_id:
+ return None
+ q = db.query(ProviderAuthSession).filter(
+ ProviderAuthSession.id == auth_id,
+ ProviderAuthSession.provider == CHATGPT_SUBSCRIPTION_PROVIDER,
+ )
+ return q.filter(ProviderAuthSession.owner == owner).first()
+
+
+def chatgpt_account_id_from_token(access_token: str) -> Optional[str]:
+ """Extract the ChatGPT account id claim from an access token, if any."""
+ try:
+ payload = _decode_jwt_payload(access_token)
+ except Exception:
+ return None
+ auth_claims = payload.get("https://api.openai.com/auth")
+ if isinstance(auth_claims, dict):
+ account_id = auth_claims.get("chatgpt_account_id")
+ if isinstance(account_id, str) and account_id.strip():
+ return account_id.strip()
+ return None
+
+
+def usage_request_headers(access_token: str) -> Dict[str, str]:
+ headers = {
+ "Accept": "application/json",
+ "User-Agent": "Odysseus ChatGPT Subscription",
+ "Authorization": f"Bearer {access_token}",
+ }
+ account_id = chatgpt_account_id_from_token(access_token)
+ if account_id:
+ headers["ChatGPT-Account-Id"] = account_id
+ return headers
+
+
+class ChatGPTUsageUnavailable(ChatGPTSubscriptionError):
+ """Usage telemetry could not be read; the model endpoint is unaffected."""
+
+ def __init__(self, reason: str, message: str, *, status_code: Optional[int] = None):
+ super().__init__(message)
+ self.reason = reason
+ self.status_code = status_code
+
+
+def _coerce_number(value: Any) -> Optional[float]:
+ if isinstance(value, bool):
+ return None
+ if isinstance(value, (int, float, str)):
+ try:
+ number = float(value)
+ return number if math.isfinite(number) else None
+ except (ValueError, OverflowError):
+ return None
+ return None
+
+
+def _coerce_int(value: Any) -> Optional[int]:
+ number = _coerce_number(value)
+ if number is None:
+ return None
+ try:
+ return int(number)
+ except (OverflowError, ValueError):
+ return None
+
+
+def _optional_str(value: Any) -> Optional[str]:
+ if isinstance(value, str):
+ text = value.strip()
+ return text or None
+ return None
+
+
+def window_minutes_from_seconds(seconds: Any) -> Optional[int]:
+ """Codex-compatible ceil(seconds / 60); None for missing/non-positive."""
+ value = _coerce_int(seconds)
+ if value is None or value <= 0:
+ return None
+ return (value + 59) // 60
+
+
+def friendly_window_name(window_minutes: Optional[int]) -> str:
+ """Compact label derived from the actual window length (never assumed)."""
+ if not window_minutes or window_minutes <= 0:
+ return "LIMIT"
+ if window_minutes % (7 * 24 * 60) == 0:
+ weeks = window_minutes // (7 * 24 * 60)
+ return "WEEK" if weeks == 1 else f"{weeks}W"
+ if window_minutes % (24 * 60) == 0:
+ return f"{window_minutes // (24 * 60)}D"
+ if window_minutes % 60 == 0:
+ return f"{window_minutes // 60}H"
+ return f"{window_minutes}M"
+
+
+def normalize_usage_window(raw: Any, kind: str) -> Optional[Dict[str, Any]]:
+ """Normalize one ``primary_window``/``secondary_window`` snapshot.
+
+ Upstream fields (openai/codex ``RateLimitWindowSnapshot``): ``used_percent``,
+ ``limit_window_seconds``, ``reset_after_seconds``, ``reset_at``. Absent
+ values stay ``None``; nothing is invented.
+ """
+ if not isinstance(raw, dict):
+ return None
+ used_percent = _coerce_number(raw.get("used_percent"))
+ if used_percent is not None:
+ used_percent = max(0.0, min(100.0, used_percent))
+ window_minutes = window_minutes_from_seconds(raw.get("limit_window_seconds"))
+ if window_minutes is None:
+ window_minutes = _coerce_int(raw.get("window_minutes"))
+ if window_minutes is not None and window_minutes <= 0:
+ window_minutes = None
+ resets_at = _coerce_int(raw.get("reset_at"))
+ if resets_at is None:
+ resets_at = _coerce_int(raw.get("resets_at"))
+ if resets_at is not None and resets_at <= 0:
+ resets_at = None
+ reset_after_seconds = _coerce_int(raw.get("reset_after_seconds"))
+ if reset_after_seconds is not None and reset_after_seconds < 0:
+ reset_after_seconds = None
+ return {
+ "kind": kind,
+ "name": friendly_window_name(window_minutes),
+ "used_percent": used_percent,
+ "remaining_percent": (None if used_percent is None else round(100.0 - used_percent, 2)),
+ "window_minutes": window_minutes,
+ "resets_at": resets_at,
+ "reset_after_seconds": reset_after_seconds,
+ }
+
+
+def _normalize_rate_limit_details(raw: Any) -> Dict[str, Any]:
+ details = raw if isinstance(raw, dict) else {}
+ windows: List[Dict[str, Any]] = []
+ # Preserve new window kinds without assigning a duration to their names.
+ keys = ["primary_window", "secondary_window"]
+ keys.extend(key for key in details if key.endswith("_window") and key not in keys)
+ for key in keys:
+ window = normalize_usage_window(details.get(key), key[:-7])
+ if window is not None:
+ windows.append(window)
+ allowed = details.get("allowed")
+ limit_reached = details.get("limit_reached")
+ return {
+ "allowed": allowed if isinstance(allowed, bool) else None,
+ "limit_reached": limit_reached if isinstance(limit_reached, bool) else None,
+ "windows": windows,
+ }
+
+
+def normalize_usage_payload(payload: Any) -> Dict[str, Any]:
+ """Normalize a ``GET /wham/usage`` JSON body into Odysseus' safe contract.
+
+ Returns only non-credential fields. The main Codex limit is reported as
+ ``limit_id == "codex"`` (as openai/codex does); each entry of
+ ``additional_rate_limits`` becomes its own bucket keyed by
+ ``metered_feature``. Unknown fields are ignored, unknown buckets kept.
+ """
+ if not isinstance(payload, dict):
+ raise ChatGPTUsageUnavailable("malformed", "ChatGPT usage response was not a JSON object.")
+
+ limits: List[Dict[str, Any]] = []
+ main = _normalize_rate_limit_details(payload.get("rate_limit"))
+ limits.append({
+ "limit_id": "codex",
+ "limit_name": None,
+ "normal_model_slug": None,
+ "allowed": main["allowed"],
+ "limit_reached": main["limit_reached"],
+ "windows": main["windows"],
+ })
+
+ additional = payload.get("additional_rate_limits")
+ if isinstance(additional, list):
+ for entry in additional:
+ if not isinstance(entry, dict):
+ continue
+ details = _normalize_rate_limit_details(entry.get("rate_limit"))
+ limit_name = _optional_str(entry.get("limit_name"))
+ limit_id = _optional_str(entry.get("metered_feature")) or limit_name
+ if not limit_id and not details["windows"]:
+ continue
+ limits.append({
+ "limit_id": limit_id or "additional",
+ "limit_name": limit_name,
+ "normal_model_slug": _optional_str(entry.get("normal_model_slug")),
+ "allowed": details["allowed"],
+ "limit_reached": details["limit_reached"],
+ "windows": details["windows"],
+ })
+
+ reached = payload.get("rate_limit_reached_type")
+ if isinstance(reached, dict):
+ reached = _optional_str(reached.get("type") or reached.get("kind"))
+ else:
+ reached = _optional_str(reached)
+
+ ordinary_usage_allowed = main["allowed"]
+ return {
+ "account_id": _optional_str(payload.get("account_id")),
+ "plan_type": _optional_str(payload.get("plan_type")),
+ "ordinary_usage_allowed": ordinary_usage_allowed,
+ "rate_limit_reached_type": reached,
+ "limits": limits,
+ }
+
+
+def fetch_usage_payload(access_token: str, timeout: float = CHATGPT_USAGE_TIMEOUT_SECONDS) -> Dict[str, Any]:
+ """Read the raw usage JSON for one access token; classify failures."""
+ if not access_token:
+ raise ChatGPTUsageUnavailable("reauth", "ChatGPT Subscription has no access token.")
+ try:
+ response = httpx.get(CHATGPT_USAGE_URL, headers=usage_request_headers(access_token), timeout=timeout)
+ except httpx.TimeoutException as exc:
+ raise ChatGPTUsageUnavailable("timeout", "ChatGPT usage request timed out.") from exc
+ except httpx.HTTPError as exc:
+ raise ChatGPTUsageUnavailable("network", "ChatGPT usage request failed.") from exc
+ status = response.status_code
+ if status in (401, 403):
+ raise ChatGPTUsageUnavailable(
+ "reauth",
+ "ChatGPT rejected the usage request; the account may need reconnecting.",
+ status_code=status,
+ )
+ if status == 429:
+ raise ChatGPTUsageUnavailable("rate_limited", "ChatGPT usage is temporarily rate limited.", status_code=status)
+ if status >= 500:
+ raise ChatGPTUsageUnavailable("upstream", f"ChatGPT usage service returned HTTP {status}.", status_code=status)
+ if status != 200:
+ raise ChatGPTUsageUnavailable("upstream", f"ChatGPT usage request returned HTTP {status}.", status_code=status)
+ try:
+ data = response.json()
+ except Exception as exc:
+ raise ChatGPTUsageUnavailable("malformed", "ChatGPT usage response was not valid JSON.") from exc
+ if not isinstance(data, dict):
+ raise ChatGPTUsageUnavailable("malformed", "ChatGPT usage response was not a JSON object.")
+ return data
+
+
+class UsageCache:
+ """Short per-auth-session cache for normalized usage snapshots."""
+
+ def __init__(self, ttl_seconds: float = CHATGPT_USAGE_CACHE_TTL_SECONDS, time_func=time.monotonic, max_entries: int = 256):
+ self._ttl = float(ttl_seconds)
+ self._max_entries = max(1, int(max_entries))
+ self._time = time_func
+ self._entries: Dict[str, tuple] = {}
+ self._lock = threading.Lock()
+
+ def get(self, auth_id: str) -> Optional[Dict[str, Any]]:
+ now = float(self._time())
+ with self._lock:
+ entry = self._entries.get(auth_id)
+ if entry is None:
+ return None
+ stored_at, value = entry
+ if now - stored_at >= self._ttl:
+ self._entries.pop(auth_id, None)
+ return None
+ return json.loads(json.dumps(value))
+
+ def put(self, auth_id: str, value: Dict[str, Any]) -> None:
+ with self._lock:
+ now = float(self._time())
+ for key, (stored_at, _) in list(self._entries.items()):
+ if now - stored_at >= self._ttl:
+ self._entries.pop(key, None)
+ self._entries.pop(auth_id, None)
+ while len(self._entries) >= self._max_entries:
+ self._entries.pop(next(iter(self._entries)))
+ self._entries[auth_id] = (now, json.loads(json.dumps(value)))
+
+ def invalidate(self, auth_id: str) -> None:
+ with self._lock:
+ self._entries.pop(auth_id, None)
+
+ def clear(self) -> None:
+ with self._lock:
+ self._entries.clear()
+
+
+USAGE_CACHE = UsageCache()
+
+
+def get_account_usage(
+ auth_id: str,
+ owner: Optional[str] = None,
+ *,
+ force_refresh: bool = False,
+ cache: Optional[UsageCache] = None,
+) -> Dict[str, Any]:
+ """Return normalized usage for exactly one owner-scoped auth session.
+
+ The access token is resolved (and refreshed if needed) for that auth
+ session only. Results are cached per auth id; ``force_refresh`` bypasses
+ and replaces the cached entry. Raises :class:`ChatGPTUsageUnavailable` on
+ any read failure; callers must not treat that as an endpoint failure.
+ """
+ cache = USAGE_CACHE if cache is None else cache
+ auth_id = (auth_id or "").strip()
+ if not auth_id:
+ raise ChatGPTSubscriptionAuthNotFound("ChatGPT Subscription account was not found.")
+ # Authorize even cache hits: cached telemetry must not outlive ownership.
+ _Auth, SessionLocal, _now = _database_handles()
+ db = SessionLocal()
+ try:
+ if find_owned_auth_session(db, auth_id, owner) is None:
+ raise ChatGPTSubscriptionAuthNotFound("ChatGPT Subscription account was not found.")
+ finally:
+ db.close()
+ if not force_refresh:
+ cached = cache.get(auth_id)
+ if cached is not None:
+ cached["cached"] = True
+ return cached
+ else:
+ cache.invalidate(auth_id)
+ try:
+ creds = resolve_runtime_credentials(auth_id, owner=owner)
+ except ChatGPTSubscriptionAuthNotFound:
+ raise
+ except ChatGPTSubscriptionRateLimited as exc:
+ raise ChatGPTUsageUnavailable("rate_limited", str(exc), status_code=429) from exc
+ except ChatGPTSubscriptionReauthRequired as exc:
+ raise ChatGPTUsageUnavailable("reauth", str(exc), status_code=401) from exc
+ except ChatGPTSubscriptionError as exc:
+ raise ChatGPTUsageUnavailable("upstream", str(exc)) from exc
+ raw = fetch_usage_payload(creds.get("api_key") or "")
+ normalized = normalize_usage_payload(raw)
+ normalized["auth_id"] = auth_id
+ normalized["fetched_at"] = int(time.time())
+ cache.put(auth_id, normalized)
+ result = json.loads(json.dumps(normalized))
+ result["cached"] = False
+ return result
+
+
def to_http_exception(exc: Exception) -> HTTPException:
if isinstance(exc, ChatGPTSubscriptionRateLimited):
return HTTPException(429, str(exc))
if isinstance(exc, (ChatGPTSubscriptionReauthRequired, ChatGPTSubscriptionAuthNotFound)):
return HTTPException(401, f"{exc} Reconnect the provider.")
- return HTTPException(502, str(exc))
+ if isinstance(exc, (ChatGPTSubscriptionError, ValueError)):
+ return HTTPException(502, str(exc))
+ return HTTPException(502, "ChatGPT Subscription request failed.")
def build_responses_input(messages: list[dict]) -> list[dict]:
diff --git a/src/foreground_model_routing.py b/src/foreground_model_routing.py
index 76e254614..898aa2ca8 100644
--- a/src/foreground_model_routing.py
+++ b/src/foreground_model_routing.py
@@ -191,7 +191,16 @@ def build_foreground_route_descriptors(
owner=owner,
)
if selected is None:
- selected = resolve_route_descriptor(endpoint_url, model, headers or {}, owner=owner)
+ if selected_endpoint_id:
+ # An explicit identity must never be relabelled as another account
+ # with the same URL/model (or even the same current bearer).
+ selected = {
+ "endpoint_id": None,
+ "endpoint_label": "Selected route",
+ "endpoint_cost_tracked": endpoint_cost_tracked(endpoint_url),
+ }
+ else:
+ selected = resolve_route_descriptor(endpoint_url, model, headers or {}, owner=owner)
primary = (endpoint_url, model, headers or {})
candidates = [primary]
descriptors = [selected]
diff --git a/src/llm_core.py b/src/llm_core.py
index 7b7298ad7..fa9cf045b 100644
--- a/src/llm_core.py
+++ b/src/llm_core.py
@@ -1496,6 +1496,33 @@ def _chatgpt_subscription_instructions(messages: List[Dict]) -> str:
return "You are a helpful AI assistant."
+# Provider-native agentic surfaces that must never be sent on the ChatGPT
+# Subscription route. ChatGPT provides model inference only; Odysseus is the
+# only agent (planning, tool selection/execution, filesystem, shell, browser,
+# MCP). Odysseus' own text tool protocol travels inside ``instructions``/``input``.
+CHATGPT_FORBIDDEN_PAYLOAD_KEYS = frozenset({
+ "tools",
+ "tool_choice",
+ "parallel_tool_calls",
+ "web_search",
+ "web_search_preview",
+ "file_search",
+ "computer",
+ "computer_use",
+ "computer_use_preview",
+ "shell",
+ "local_shell",
+ "code_interpreter",
+ "image_generation",
+ "mcp",
+ "function",
+ "functions",
+ "include",
+ "previous_response_id",
+ "background",
+})
+
+
def _build_chatgpt_responses_payload(
model: str,
messages: List[Dict],
@@ -1503,9 +1530,18 @@ def _build_chatgpt_responses_payload(
max_tokens: int,
*,
stream: bool = False,
+ tools: Optional[List[Dict]] = None,
+ **_ignored,
) -> Dict:
+ """Build the ChatGPT/Codex Responses request: model inference only.
+
+ ``tools`` (and any other provider-native tool declaration) is accepted for
+ signature compatibility with the other transports and deliberately
+ discarded. See :data:`CHATGPT_FORBIDDEN_PAYLOAD_KEYS`.
+ """
from src.chatgpt_subscription import build_responses_input
+ del tools, _ignored
conversation = [msg for msg in (messages or []) if (msg.get("role") or "") != "system"]
payload: Dict = {
"model": model,
@@ -1519,7 +1555,27 @@ def _build_chatgpt_responses_payload(
# ChatGPT Subscription Codex API does not support max_output_tokens —
# passing it returns HTTP 400 "Unsupported parameter: max_output_tokens".
# Do not include it in the payload.
- return payload
+ return _strip_chatgpt_native_tool_surfaces(payload)
+
+
+CHATGPT_ALLOWED_PAYLOAD_KEYS = frozenset({
+ "model", "instructions", "input", "stream", "store", "temperature",
+})
+
+
+def _strip_chatgpt_native_tool_surfaces(payload: Dict) -> Dict:
+ """Only explicitly approved inference fields may cross this boundary."""
+ return {key: value for key, value in payload.items() if key in CHATGPT_ALLOWED_PAYLOAD_KEYS}
+
+
+def _chatgpt_safe_error(message: str, headers: Dict) -> str:
+ """Upstream diagnostics must never echo the request bearer to the UI."""
+ result = str(message)
+ for key, value in (headers or {}).items():
+ if key.lower() in {"authorization", "x-api-key"} and value:
+ secret = str(value).removeprefix("Bearer ")
+ result = result.replace(str(value), "[redacted]").replace(secret, "[redacted]")
+ return result
def _format_chatgpt_subscription_error(status_code: int, text: str) -> str:
@@ -2977,8 +3033,8 @@ async def _stream_llm_inner(url: str, model: str, messages: List[Dict], temperat
_clear_host_dead(target_url)
if r.status_code != 200:
raw = (await r.aread()).decode(errors="replace")
- friendly = _format_chatgpt_subscription_error(r.status_code, raw)
- yield f'event: error\ndata: {json.dumps({"status": r.status_code, "text": friendly, "raw": raw[:500]})}\n\n'
+ friendly = _format_chatgpt_subscription_error(r.status_code, _chatgpt_safe_error(raw, h))
+ yield f'event: error\ndata: {json.dumps({"status": r.status_code, "text": friendly})}\n\n'
return
async for line in r.aiter_lines():
if not line:
@@ -3063,13 +3119,13 @@ async def _stream_llm_inner(url: str, model: str, messages: List[Dict], temperat
}
text = err.get("message") if isinstance(err, dict) else str(err or "ChatGPT Subscription request failed")
status = _provider_stream_error_status(err, default=400)
- yield f'event: error\ndata: {json.dumps({"status": status, "text": text})}\n\n'
+ yield f'event: error\ndata: {json.dumps({"status": status, "text": _chatgpt_safe_error(text, h)})}\n\n'
return
yield "data: [DONE]\n\n"
except (httpx.ConnectError, httpx.ConnectTimeout) as e:
_cooled = _mark_host_dead(target_url)
_tail = f" — host cooled for {DEAD_HOST_COOLDOWN:.0f}s" if _cooled else " — transient, will retry"
- logger.warning(f"ChatGPT Subscription stream connect to {target_url} failed: {e}{_tail}")
+ logger.warning("ChatGPT Subscription stream connect failed: %s%s", type(e).__name__, _tail)
yield f'event: error\ndata: {json.dumps({"error": f"Cannot reach {_host_key(target_url)}", "status": 503})}\n\n'
except httpx.ReadTimeout:
yield f'event: error\ndata: {json.dumps({"error": "Read timeout", "status": 504})}\n\n'
@@ -3082,8 +3138,8 @@ async def _stream_llm_inner(url: str, model: str, messages: List[Dict], temperat
except httpx.NetworkError:
yield f'event: error\ndata: {json.dumps({"error": "Network error", "status": 502, "fallback_eligible": False})}\n\n'
except Exception as e:
- logger.error(f"ChatGPT Subscription stream error: {e}")
- yield f'event: error\ndata: {json.dumps({"error": str(e), "status": 502, "fallback_eligible": False})}\n\n'
+ logger.error("ChatGPT Subscription stream error: %s", type(e).__name__)
+ yield f'event: error\ndata: {json.dumps({"error": "ChatGPT Subscription stream failed", "status": 502, "fallback_eligible": False})}\n\n'
return
# ── Native Ollama streaming ──
@@ -3099,7 +3155,7 @@ async def _stream_llm_inner(url: str, model: str, messages: List[Dict], temperat
if r.status_code != 200:
raw = (await r.aread()).decode(errors="replace")
friendly = _format_upstream_error(r.status_code, raw, target_url)
- yield f'event: error\ndata: {json.dumps({"status": r.status_code, "text": friendly, "raw": raw[:500]})}\n\n'
+ yield f'event: error\ndata: {json.dumps({"status": r.status_code, "text": friendly})}\n\n'
return
async for line in r.aiter_lines():
if not line:
@@ -3198,7 +3254,7 @@ async def _stream_llm_inner(url: str, model: str, messages: List[Dict], temperat
if r.status_code != 200:
raw = (await r.aread()).decode(errors="replace")
friendly = _format_upstream_error(r.status_code, raw, target_url)
- yield f'event: error\ndata: {json.dumps({"status": r.status_code, "text": friendly, "raw": raw[:500]})}\n\n'
+ yield f'event: error\ndata: {json.dumps({"status": r.status_code, "text": friendly})}\n\n'
return
async for line in r.aiter_lines():
# SSE allows "data:value" with no space after the colon
@@ -3382,7 +3438,7 @@ async def _stream_llm_inner(url: str, model: str, messages: List[Dict], temperat
if r.status_code != 200:
raw = (await r.aread()).decode(errors="replace")
friendly = _format_upstream_error(r.status_code, raw, target_url)
- yield f'event: error\ndata: {json.dumps({"status": r.status_code, "text": friendly, "raw": raw[:500]})}\n\n'
+ yield f'event: error\ndata: {json.dumps({"status": r.status_code, "text": friendly})}\n\n'
return
first_token_budget = _first_token_timeout(target_url, timeout)
diff --git a/static/app.js b/static/app.js
index 0a562ac73..f5014f29e 100644
--- a/static/app.js
+++ b/static/app.js
@@ -37,7 +37,7 @@ import { UI_VIS_DEFAULT_OFF, resolveVisibility } from './js/ui_visibility.js?v=2
import tasksModule from './js/tasks.js?v=20260914taskmodel1';
import calendarModule from './js/calendar.js?v=20260914emailsource11';
import notesModule from './js/notes.js?v=20260911notesselectioncancel1';
-import adminModule from './js/admin.js?v=20260914toolschemaprofiles1';
+import adminModule from './js/admin.js?v=20260921chatgptusage1';
import settingsModule from './js/settings.js?v=20260912writingstyle3';
// Eagerly bind unified minimize/restore behavior across all tool modals.
import './js/modalManager.js';
diff --git a/static/index.html b/static/index.html
index 808f78340..1b87eff7f 100644
--- a/static/index.html
+++ b/static/index.html
@@ -246,7 +246,7 @@
real request is discarded and the font fetched a second time. -->
-
+
diff --git a/static/js/admin.js b/static/js/admin.js
index 0dccd8e68..e6098908f 100644
--- a/static/js/admin.js
+++ b/static/js/admin.js
@@ -6,6 +6,12 @@ import settingsModule from './settings.js?v=20260912writingstyle3';
import { providerLogo, providerLogoFromUrl } from './providers.js';
import { sortModelObjects } from './modelSort.js';
import { PROVIDER_DEVICE_FLOWS, formatDeviceFlowError, runProviderDeviceFlow } from './providerDeviceFlow.js';
+import {
+ accountTitle as chatgptAccountTitle,
+ buildUsageViewModel as buildChatgptUsageViewModel,
+ isChatgptSubscriptionEndpoint,
+ renderUsageCardHtml as renderChatgptUsageCardHtml,
+} from './chatgptSubscriptionUsage.js';
import { getSettings, getTools, invalidateSettings, invalidateTools } from './appConfig.js';
let initialized = false;
@@ -652,6 +658,7 @@ async function loadEndpoints() {
const keyLabel = ep.has_key
? (ep.api_key_fingerprint ? ` (key ${esc(ep.api_key_fingerprint)})` : ' (key set)')
: '';
+ const isChatgptAccount = isChatgptSubscriptionEndpoint(ep);
return `
@@ -661,7 +668,7 @@ async function loadEndpoints() {
` : ''}
${providerLogoFromUrl(ep.base_url) || ''}
-
${esc(ep.name)}
+
${esc(isChatgptAccount ? chatgptAccountTitle(ep) : ep.name)}
${ep.model_type === 'image' ? '
Image' : ''}
${kindLabel ? `
${esc(kindLabel)}` : ''}
${statusBadge}
@@ -675,6 +682,7 @@ async function loadEndpoints() {
${hasModels ? '
' : ''}`}
+ ${isChatgptAccount ? `` : ''}
${esc(ep.base_url)}${category === 'local' ? `
` : ''}${keyLabel}
' +
+ '
' +
+ '' + esc(waitLabel) + '
' +
+ '
' +
+ 'Code' +
+ '' + esc(start.user_code) + '' +
+ '' +
+ '
' +
+ '
' + esc(authLabel) + ' ↗' +
+ '
' +
- '
' +
- '' + esc(waitLabel) + '
' +
- '
' +
- 'Code' +
- '' + esc(start.user_code) + '' +
- '' +
- '
' +
- '
' + esc(authLabel) + ' ↗' +
- '
` +
+ `
` +
+ `${esc(win.name)}` +
+ `${esc(win.remainingLabel || win.usedLabel)}` +
+ `
` +
+ `
` +
+ `` +
+ `
` +
+ `
` +
+ `${esc(win.usedLabel)}` +
+ (win.resetLabel ? `${esc(win.resetLabel)}` : '') +
+ `
` +
+ `
` +
+ `` +
+ `` +
+ `
`;
+ if (!vm.available) {
+ return (
+ `` +
+ `
${esc(vm.message || 'Usage unavailable')}
` +
+ buttons +
+ `
${esc(limit.title)}${limit.modelSlug ? ` ${esc(limit.modelSlug)}` : ''}
`
+ : '';
+ return `${title}${limit.windows.length ? limit.windows.map((w) => _barHtml(w, esc)).join('') : '
No rate-limit windows reported
'}
Usage currently blocked${vm.rateLimitReachedType ? ` (${esc(vm.rateLimitReachedType.replace(/_/g, ' '))})` : ''}
`
+ : '';
+ return (
+ `` +
+ (plan ? `
${plan}
` : '') +
+ blocked +
+ limitsHtml +
+ buttons +
+ `
"},
+ "usage": {"plan_type": "plus", "limits": [{"limit_id": "codex", "limit_name": "