fix(chatgpt): fail fast instead of running device code login inside a running event loop

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Devin AI 2026-09-03 10:55:12 +00:00
parent 658f50663d
commit 17f7b338f0
2 changed files with 43 additions and 3 deletions

View file

@ -1,3 +1,4 @@
import asyncio
import base64
import json
import os
@ -40,6 +41,14 @@ def _optional_str(value: JsonValue | None) -> str | None:
return value if isinstance(value, str) else None
def _event_loop_is_running() -> bool:
try:
asyncio.get_running_loop()
except RuntimeError:
return False
return True
class Authenticator:
def __init__(self) -> None:
self.token_dir = os.getenv(
@ -66,6 +75,16 @@ class Authenticator:
except RefreshAccessTokenError as exc:
verbose_logger.warning("ChatGPT refresh token failed, re-login required: %s", exc)
if _event_loop_is_running():
raise GetAccessTokenError(
message=(
"No usable cached ChatGPT credentials, and the interactive device code login cannot run inside "
"a running event loop. Sign in once from a synchronous session (for example litellm.completion "
"in a terminal) so the tokens are cached in CHATGPT_TOKEN_DIR, then retry."
),
status_code=401,
)
cooldown_remaining: Final = self._get_device_code_cooldown_remaining(auth_data)
if cooldown_remaining > 0:
token: Final = self._wait_for_access_token(cooldown_remaining)

View file

@ -1,3 +1,4 @@
import asyncio
import base64
import json
import time
@ -6,6 +7,7 @@ from unittest.mock import mock_open, patch
import pytest
from litellm.llms.chatgpt.authenticator import Authenticator
from litellm.llms.chatgpt.common_utils import GetAccessTokenError
def _make_jwt(payload: dict) -> str:
@ -54,10 +56,29 @@ class TestChatGPTAuthenticator:
token = authenticator.get_access_token()
assert token == "token-new"
def test_get_access_token_inside_event_loop_fails_fast_without_device_login(self, authenticator, tmp_path):
authenticator.auth_file = str(tmp_path / "missing.json")
async def _get_token_on_loop() -> str:
return authenticator.get_access_token()
with patch.object(authenticator, "_request_device_code") as mock_device_code:
with pytest.raises(GetAccessTokenError) as exc_info:
asyncio.run(_get_token_on_loop())
mock_device_code.assert_not_called()
assert exc_info.value.status_code == 401
assert "event loop" in str(exc_info.value)
def test_get_access_token_outside_event_loop_runs_device_login(self, authenticator, tmp_path):
authenticator.auth_file = str(tmp_path / "missing.json")
with patch.object(authenticator, "_login_device_code", return_value={"access_token": "token-dev"}) as login:
assert authenticator.get_access_token() == "token-dev"
login.assert_called_once()
def test_get_account_id_from_id_token(self, authenticator):
id_token = _make_jwt(
{"https://api.openai.com/auth": {"chatgpt_account_id": "acct-123"}}
)
id_token = _make_jwt({"https://api.openai.com/auth": {"chatgpt_account_id": "acct-123"}})
auth_data = json.dumps({"id_token": id_token})
with (