fix(oauth): 🐛 prevent deadlock and token desync for rotating refresh tokens

This commit fixes critical issues in OAuth providers (Google, iFlow, Qwen) that use rotating refresh tokens, where each token refresh invalidates the previous token.

**Problems Fixed:**

1. **Deadlock Prevention**: Removed inline re-authentication from `refresh_token()` method that was called while holding a lock. When refresh failed with HTTP 400/401/403, the method would call `initialize_token()` directly, which would try to acquire the same lock, causing a deadlock. Now, invalid token errors are caught and queued for background re-authentication via `asyncio.create_task()`.

2. **Token Desync**: Changed `_save_credentials()` to write to disk FIRST, then update cache. Previously, cache was updated first with `buffer_on_failure=True`, which could leave stale tokens on disk if the write failed. For rotating tokens, this caused the old refresh_token on disk to become invalid after a successful API call, requiring re-auth on restart.

3. **Stale Cache Usage**: Modified `refresh_token()` to always read fresh credentials from disk before refreshing, preventing use of stale cached tokens that may have been invalidated by another process.

4. **New Error Type**: Introduced `CredentialNeedsReauthError` exception to signal rotatable authentication failures. This allows the client to rotate to the next credential without logging scary tracebacks, while background re-auth fixes the broken credential.

**Changes:**

- Add `CredentialNeedsReauthError` exception class and classification in error_handler.py
- Catch and wrap `CredentialNeedsReauthError` in client.py retry loop
- Replace inline re-auth with background task queuing in all OAuth providers
- Change `_save_credentials()` to disk-first writes with no buffering for rotating tokens
- Add `force_interactive` parameter to `initialize_token()` for explicit re-auth requests
- Always reload credentials from disk before refresh to prevent stale token usage
- Return boolean from `_save_credentials()` and raise IOError on critical failures
- Update re-auth queue processing to call `initialize_token(force_interactive=True)`

src/rotator_library/client.py

@@ -23,6 +23,7 @@ from .usage_manager import UsageManager
 from .failure_logger import log_failure, configure_failure_logger
 from .error_handler import (
     PreRequestCallbackError,
+    CredentialNeedsReauthError,
     classify_error,
     AllProviders,
     NoAvailableKeysError,
@@ -755,6 +756,12 @@ class RotatingClient:
                         await self.usage_manager.record_success(key, model)
                     break
 
+                except CredentialNeedsReauthError as e:
+                    # This credential needs re-authentication but re-auth is already queued.
+                    # Wrap it so the outer retry loop can rotate to the next credential.
+                    # No scary traceback needed - this is an expected recovery scenario.
+                    raise StreamedAPIError("Credential needs re-authentication", data=e)
+
                 except (
                     litellm.RateLimitError,
                     litellm.ServiceUnavailableError,

src/rotator_library/error_handler.py

@@ -117,6 +117,31 @@ class PreRequestCallbackError(Exception):
     pass
 
 
+class CredentialNeedsReauthError(Exception):
+    """
+    Raised when a credential's refresh token is invalid and re-authentication is required.
+
+    This is a rotatable error - the request should try the next credential while
+    the broken credential is queued for re-authentication in the background.
+
+    Unlike generic HTTPStatusError, this exception signals:
+    - The credential is temporarily unavailable (needs user action)
+    - Re-auth has already been queued
+    - The request should rotate to the next credential without logging scary tracebacks
+
+    Attributes:
+        credential_path: Path to the credential file that needs re-auth
+        message: Human-readable message about the error
+    """
+
+    def __init__(self, credential_path: str, message: str = ""):
+        self.credential_path = credential_path
+        self.message = (
+            message or f"Credential '{credential_path}' requires re-authentication"
+        )
+        super().__init__(self.message)
+
+
 # =============================================================================
 # ERROR TRACKING FOR CLIENT REPORTING
 # =============================================================================
@@ -698,6 +723,14 @@ def classify_error(e: Exception, provider: Optional[str] = None) -> ClassifiedEr
             status_code=400,  # Treat as a bad request
         )
 
+    if isinstance(e, CredentialNeedsReauthError):
+        # This is a rotatable error - credential is broken but re-auth is queued
+        return ClassifiedError(
+            error_type="credential_reauth_needed",
+            original_exception=e,
+            status_code=401,  # Treat as auth error for reporting purposes
+        )
+
     if isinstance(e, RateLimitError):
         retry_after = get_retry_after(e)
         # Check if this is a quota error vs rate limit
@@ -789,6 +822,7 @@ def should_rotate_on_error(classified_error: ClassifiedError) -> bool:
     - quota_exceeded: Current key/account exhausted
     - forbidden: Current credential denied access
     - authentication: Current credential invalid
+    - credential_reauth_needed: Credential needs interactive re-auth (queued)
     - server_error: Provider having issues (might work with different endpoint/key)
     - api_connection: Network issues (might be transient)
     - unknown: Safer to try another key

src/rotator_library/providers/google_oauth_base.py

@@ -22,6 +22,7 @@ from rich.markup import escape as rich_escape
 from ..utils.headless_detection import is_headless_environment
 from ..utils.reauth_coordinator import get_reauth_coordinator
 from ..utils.resilient_io import safe_write_json
+from ..error_handler import CredentialNeedsReauthError
 
 lib_logger = logging.getLogger("rotator_library")
 
@@ -366,7 +367,6 @@ class GoogleOAuthBase:
             max_retries = 3
             new_token_data = None
             last_error = None
-            needs_reauth = False
 
             async with httpx.AsyncClient() as client:
                 for attempt in range(max_retries):
@@ -390,15 +390,42 @@ class GoogleOAuthBase:
                     except httpx.HTTPStatusError as e:
                         last_error = e
                         status_code = e.response.status_code
-
-                        # [INVALID GRANT HANDLING] Handle 401/403 by triggering re-authentication
-                        if status_code == 401 or status_code == 403:
-                            lib_logger.warning(
-                                f"Refresh token invalid for '{Path(path).name}' (HTTP {status_code}). "
-                                f"Token may have been revoked or expired. Starting re-authentication..."
+                        error_body = e.response.text
+
+                        # [INVALID GRANT HANDLING] Handle 400/401/403 by queuing for re-auth
+                        # We must NOT call initialize_token from here as we hold a lock (would deadlock)
+                        if status_code == 400:
+                            # Check if this is an invalid_grant error
+                            if "invalid_grant" in error_body.lower():
+                                lib_logger.info(
+                                    f"Credential '{Path(path).name}' needs re-auth (HTTP 400: invalid_grant). "
+                                    f"Queued for re-authentication, rotating to next credential."
+                                )
+                                asyncio.create_task(
+                                    self._queue_refresh(
+                                        path, force=True, needs_reauth=True
+                                    )
+                                )
+                                raise CredentialNeedsReauthError(
+                                    credential_path=path,
+                                    message=f"Refresh token invalid for '{Path(path).name}'. Re-auth queued.",
+                                )
+                            else:
+                                # Other 400 error - raise it
+                                raise
+
+                        elif status_code in (401, 403):
+                            lib_logger.info(
+                                f"Credential '{Path(path).name}' needs re-auth (HTTP {status_code}). "
+                                f"Queued for re-authentication, rotating to next credential."
+                            )
+                            asyncio.create_task(
+                                self._queue_refresh(path, force=True, needs_reauth=True)
+                            )
+                            raise CredentialNeedsReauthError(
+                                credential_path=path,
+                                message=f"Token invalid for '{Path(path).name}' (HTTP {status_code}). Re-auth queued.",
                             )
-                            needs_reauth = True
-                            break  # Exit retry loop to trigger re-auth
 
                         elif status_code == 429:
                             # Rate limit - honor Retry-After header if present
@@ -438,23 +465,6 @@ class GoogleOAuthBase:
                             continue
                         raise
 
-            # [INVALID GRANT RE-AUTH] Trigger OAuth flow if refresh token is invalid
-            if needs_reauth:
-                lib_logger.info(
-                    f"Starting re-authentication for '{Path(path).name}'..."
-                )
-                try:
-                    # Call initialize_token to trigger OAuth flow
-                    new_creds = await self.initialize_token(path)
-                    return new_creds
-                except Exception as reauth_error:
-                    lib_logger.error(
-                        f"Re-authentication failed for '{Path(path).name}': {reauth_error}"
-                    )
-                    raise ValueError(
-                        f"Refresh token invalid and re-authentication failed: {reauth_error}"
-                    )
-
             # If we exhausted retries without success
             if new_token_data is None:
                 raise last_error or Exception("Token refresh failed after all retries")
@@ -832,7 +842,7 @@ class GoogleOAuthBase:
 
                 try:
                     lib_logger.info(f"Starting re-auth for '{Path(path).name}'...")
-                    await self.initialize_token(path)
+                    await self.initialize_token(path, force_interactive=True)
                     lib_logger.info(f"Re-auth SUCCESS for '{Path(path).name}'")
 
                 except Exception as e:
@@ -1058,7 +1068,9 @@ class GoogleOAuthBase:
         return new_creds
 
     async def initialize_token(
-        self, creds_or_path: Union[Dict[str, Any], str]
+        self,
+        creds_or_path: Union[Dict[str, Any], str],
+        force_interactive: bool = False,
     ) -> Dict[str, Any]:
         """
         Initialize OAuth token, triggering interactive OAuth flow if needed.
@@ -1066,6 +1078,12 @@ class GoogleOAuthBase:
         If interactive OAuth is required (expired refresh token, missing credentials, etc.),
         the flow is coordinated globally via ReauthCoordinator to ensure only one
         interactive OAuth flow runs at a time across all providers.
+
+        Args:
+            creds_or_path: Either a credentials dict or path to credentials file.
+            force_interactive: If True, skip expiry checks and force interactive OAuth.
+                               Use this when the refresh token is known to be invalid
+                               (e.g., after HTTP 400 from token endpoint).
         """
         path = creds_or_path if isinstance(creds_or_path, str) else None
 
@@ -1085,7 +1103,11 @@ class GoogleOAuthBase:
                 await self._load_credentials(creds_or_path) if path else creds_or_path
             )
             reason = ""
-            if not creds.get("refresh_token"):
+            if force_interactive:
+                reason = (
+                    "re-authentication was explicitly requested (refresh token invalid)"
+                )
+            elif not creds.get("refresh_token"):
                 reason = "refresh token is missing"
             elif self._is_token_expired(creds):
                 reason = "token is expired"

src/rotator_library/providers/iflow_auth_base.py

@@ -26,6 +26,7 @@ from rich.markup import escape as rich_escape
 from ..utils.headless_detection import is_headless_environment
 from ..utils.reauth_coordinator import get_reauth_coordinator
 from ..utils.resilient_io import safe_write_json
+from ..error_handler import CredentialNeedsReauthError
 
 lib_logger = logging.getLogger("rotator_library")
 
@@ -371,26 +372,42 @@ class IFlowAuthBase:
                     return env_creds
                 raise  # Re-raise the original file not found error
 
-    async def _save_credentials(self, path: str, creds: Dict[str, Any]):
-        """Save credentials with in-memory fallback if disk unavailable."""
-        # Always update cache first (memory is reliable)
-        self._credentials_cache[path] = creds
+    async def _save_credentials(self, path: str, creds: Dict[str, Any]) -> bool:
+        """Save credentials to disk, then update cache. Returns True only if disk write succeeded.
+
+        For providers with rotating refresh tokens, disk persistence is CRITICAL.
+        If we update the cache but fail to write to disk:
+        - The old refresh_token on disk may become invalid (consumed by API)
+        - On restart, we'd load the invalid token and require re-auth
 
+        By writing to disk FIRST, we ensure:
+        - Cache only updated after disk succeeds (guaranteed parity)
+        - If disk fails, cache keeps old tokens, refresh is retried
+        - No desync between cache and disk is possible
+        """
         # Don't save to file if credentials were loaded from environment
         if creds.get("_proxy_metadata", {}).get("loaded_from_env"):
+            self._credentials_cache[path] = creds
             lib_logger.debug("Credentials loaded from env, skipping file save")
-            return
+            return True
 
-        # Attempt disk write - if it fails, we still have the cache
-        # buffer_on_failure ensures data is retried periodically and saved on shutdown
-        if safe_write_json(
-            path, creds, lib_logger, secure_permissions=True, buffer_on_failure=True
+        # Write to disk FIRST - do NOT buffer on failure for rotating tokens
+        # Buffering is dangerous because the refresh_token may be stale by retry time
+        if not safe_write_json(
+            path, creds, lib_logger, secure_permissions=True, buffer_on_failure=False
         ):
-            lib_logger.debug(f"Saved updated iFlow OAuth credentials to '{path}'.")
-        else:
-            lib_logger.warning(
-                "iFlow credentials cached in memory only (buffered for retry)."
+            lib_logger.error(
+                f"Failed to write iFlow credentials to disk for '{Path(path).name}'. "
+                f"Cache NOT updated to maintain parity with disk."
             )
+            return False
+
+        # Disk write succeeded - now update cache (guaranteed parity)
+        self._credentials_cache[path] = creds
+        lib_logger.debug(
+            f"Saved updated iFlow OAuth credentials to '{Path(path).name}'."
+        )
+        return True
 
     def _is_token_expired(self, creds: Dict[str, Any]) -> bool:
         """Checks if the token is expired (with buffer for proactive refresh)."""
@@ -550,10 +567,11 @@ class IFlowAuthBase:
             if not force and cached_creds and not self._is_token_expired(cached_creds):
                 return cached_creds
 
-            # If cache is empty, read from file
-            if path not in self._credentials_cache:
-                await self._read_creds_from_file(path)
-
+            # [ROTATING TOKEN FIX] Always read fresh from disk before refresh.
+            # iFlow may use rotating refresh tokens - each refresh could invalidate the previous token.
+            # If we use a stale cached token, refresh will fail.
+            # Reading fresh from disk ensures we have the latest token.
+            await self._read_creds_from_file(path)
             creds_from_file = self._credentials_cache[path]
 
             lib_logger.debug(f"Refreshing iFlow OAuth token for '{Path(path).name}'...")
@@ -565,7 +583,6 @@ class IFlowAuthBase:
             max_retries = 3
             new_token_data = None
             last_error = None
-            needs_reauth = False
 
             # Create Basic Auth header
             auth_string = f"{IFLOW_CLIENT_ID}:{IFLOW_CLIENT_SECRET}"
@@ -624,14 +641,58 @@ class IFlowAuthBase:
                         )
 
                         # [STATUS CODE HANDLING]
-                        # [INVALID GRANT HANDLING] Handle 401/403 by triggering re-authentication
-                        if status_code in (401, 403):
-                            lib_logger.warning(
-                                f"Refresh token invalid for '{Path(path).name}' (HTTP {status_code}). "
-                                f"Token may have been revoked or expired. Starting re-authentication..."
+                        # [INVALID GRANT HANDLING] Handle 400/401/403 by raising
+                        # Queue for re-auth in background so credential gets fixed automatically
+                        if status_code == 400:
+                            # Check if this is an invalid refresh token error
+                            try:
+                                error_data = e.response.json()
+                                error_type = error_data.get("error", "")
+                                error_desc = error_data.get("error_description", "")
+                                if not error_desc:
+                                    error_desc = error_data.get("message", error_body)
+                            except Exception:
+                                error_type = ""
+                                error_desc = error_body
+
+                            if (
+                                "invalid" in error_desc.lower()
+                                or error_type == "invalid_request"
+                            ):
+                                lib_logger.info(
+                                    f"Credential '{Path(path).name}' needs re-auth (HTTP 400: {error_desc}). "
+                                    f"Queued for re-authentication, rotating to next credential."
+                                )
+                                # Queue for re-auth in background (non-blocking, fire-and-forget)
+                                # This ensures credential gets fixed even if caller doesn't handle it
+                                asyncio.create_task(
+                                    self._queue_refresh(
+                                        path, force=True, needs_reauth=True
+                                    )
+                                )
+                                # Raise rotatable error instead of raw HTTPStatusError
+                                raise CredentialNeedsReauthError(
+                                    credential_path=path,
+                                    message=f"Refresh token invalid for '{Path(path).name}'. Re-auth queued.",
+                                )
+                            else:
+                                # Other 400 error - raise it
+                                raise
+
+                        elif status_code in (401, 403):
+                            lib_logger.info(
+                                f"Credential '{Path(path).name}' needs re-auth (HTTP {status_code}). "
+                                f"Queued for re-authentication, rotating to next credential."
+                            )
+                            # Queue for re-auth in background (non-blocking, fire-and-forget)
+                            asyncio.create_task(
+                                self._queue_refresh(path, force=True, needs_reauth=True)
+                            )
+                            # Raise rotatable error instead of raw HTTPStatusError
+                            raise CredentialNeedsReauthError(
+                                credential_path=path,
+                                message=f"Token invalid for '{Path(path).name}' (HTTP {status_code}). Re-auth queued.",
                             )
-                            needs_reauth = True
-                            break  # Exit retry loop to trigger re-auth
 
                         elif status_code == 429:
                             retry_after = int(e.response.headers.get("Retry-After", 60))
@@ -667,37 +728,6 @@ class IFlowAuthBase:
                             continue
                         raise
 
-            # [INVALID GRANT RE-AUTH] Trigger OAuth flow if refresh token is invalid
-            if needs_reauth:
-                lib_logger.info(
-                    f"Starting re-authentication for '{Path(path).name}'..."
-                )
-                try:
-                    # Call initialize_token to trigger OAuth flow
-                    new_creds = await self.initialize_token(path)
-                    # Clear backoff on successful re-auth
-                    self._refresh_failures.pop(path, None)
-                    self._next_refresh_after.pop(path, None)
-                    return new_creds
-                except Exception as reauth_error:
-                    lib_logger.error(
-                        f"Re-authentication failed for '{Path(path).name}': {reauth_error}"
-                    )
-                    # [BACKOFF TRACKING] Increment failure count and set backoff timer
-                    self._refresh_failures[path] = (
-                        self._refresh_failures.get(path, 0) + 1
-                    )
-                    backoff_seconds = min(
-                        300, 30 * (2 ** self._refresh_failures[path])
-                    )  # Max 5 min backoff
-                    self._next_refresh_after[path] = time.time() + backoff_seconds
-                    lib_logger.debug(
-                        f"Setting backoff for '{Path(path).name}': {backoff_seconds}s"
-                    )
-                    raise ValueError(
-                        f"Refresh token invalid and re-authentication failed: {reauth_error}"
-                    )
-
             if new_token_data is None:
                 # [BACKOFF TRACKING] Increment failure count and set backoff timer
                 self._refresh_failures[path] = self._refresh_failures.get(path, 0) + 1
@@ -775,11 +805,19 @@ class IFlowAuthBase:
             self._refresh_failures.pop(path, None)
             self._next_refresh_after.pop(path, None)
 
-            await self._save_credentials(path, creds_from_file)
+            # Save credentials - MUST succeed for rotating token providers
+            if not await self._save_credentials(path, creds_from_file):
+                # CRITICAL: If we can't persist the new token, the old token may be
+                # invalidated. This is a critical failure - raise so retry logic kicks in.
+                raise IOError(
+                    f"Failed to persist refreshed credentials for '{Path(path).name}'. "
+                    f"Disk write failed - refresh will be retried."
+                )
+
             lib_logger.debug(
                 f"Successfully refreshed iFlow OAuth token for '{Path(path).name}'."
             )
-            return creds_from_file
+            return self._credentials_cache[path]  # Return from cache (synced with disk)
 
     async def get_api_details(self, credential_identifier: str) -> Tuple[str, str]:
         """
@@ -1026,12 +1064,39 @@ class IFlowAuthBase:
 
                     except httpx.HTTPStatusError as e:
                         status_code = e.response.status_code
-                        if status_code in (401, 403):
-                            # Invalid refresh token - route to re-auth queue
-                            lib_logger.warning(
-                                f"Refresh token invalid for '{Path(path).name}' (HTTP {status_code}). "
+                        # Check for invalid refresh token errors (400/401/403)
+                        # These need to be routed to re-auth queue for interactive OAuth
+                        needs_reauth = False
+
+                        if status_code == 400:
+                            # Check if this is an invalid refresh token error
+                            try:
+                                error_data = e.response.json()
+                                error_type = error_data.get("error", "")
+                                error_desc = error_data.get("error_description", "")
+                                if not error_desc:
+                                    error_desc = error_data.get("message", str(e))
+                            except Exception:
+                                error_type = ""
+                                error_desc = str(e)
+
+                            if (
+                                "invalid" in error_desc.lower()
+                                or error_type == "invalid_request"
+                            ):
+                                needs_reauth = True
+                                lib_logger.info(
+                                    f"Credential '{Path(path).name}' needs re-auth (HTTP 400: {error_desc}). "
+                                    f"Routing to re-auth queue."
+                                )
+                        elif status_code in (401, 403):
+                            needs_reauth = True
+                            lib_logger.info(
+                                f"Credential '{Path(path).name}' needs re-auth (HTTP {status_code}). "
                                 f"Routing to re-auth queue."
                             )
+
+                        if needs_reauth:
                             self._queue_retry_count.pop(path, None)  # Clear retry count
                             async with self._queue_tracking_lock:
                                 self._queued_credentials.discard(
@@ -1126,7 +1191,7 @@ class IFlowAuthBase:
 
                 try:
                     lib_logger.info(f"Starting re-auth for '{Path(path).name}'...")
-                    await self.initialize_token(path)
+                    await self.initialize_token(path, force_interactive=True)
                     lib_logger.info(f"Re-auth SUCCESS for '{Path(path).name}'")
 
                 except Exception as e:
@@ -1270,7 +1335,11 @@ class IFlowAuthBase:
                 }
 
             if path:
-                await self._save_credentials(path, creds)
+                if not await self._save_credentials(path, creds):
+                    raise IOError(
+                        f"Failed to save OAuth credentials to disk for '{display_name}'. "
+                        f"Please retry authentication."
+                    )
 
             lib_logger.info(
                 f"iFlow OAuth initialized successfully for '{display_name}'."
@@ -1281,7 +1350,9 @@ class IFlowAuthBase:
             await callback_server.stop()
 
     async def initialize_token(
-        self, creds_or_path: Union[Dict[str, Any], str]
+        self,
+        creds_or_path: Union[Dict[str, Any], str],
+        force_interactive: bool = False,
     ) -> Dict[str, Any]:
         """
         Initialize OAuth token, triggering interactive authorization flow if needed.
@@ -1289,6 +1360,12 @@ class IFlowAuthBase:
         If interactive OAuth is required (expired refresh token, missing credentials, etc.),
         the flow is coordinated globally via ReauthCoordinator to ensure only one
         interactive OAuth flow runs at a time across all providers.
+
+        Args:
+            creds_or_path: Either a credentials dict or path to credentials file.
+            force_interactive: If True, skip expiry checks and force interactive OAuth.
+                               Use this when the refresh token is known to be invalid
+                               (e.g., after HTTP 400 from token endpoint).
         """
         path = creds_or_path if isinstance(creds_or_path, str) else None
 
@@ -1308,7 +1385,11 @@ class IFlowAuthBase:
             )
 
             reason = ""
-            if not creds.get("refresh_token"):
+            if force_interactive:
+                reason = (
+                    "re-authentication was explicitly requested (refresh token invalid)"
+                )
+            elif not creds.get("refresh_token"):
                 reason = "refresh token is missing"
             elif self._is_token_expired(creds):
                 reason = "token is expired"
@@ -1389,10 +1470,15 @@ class IFlowAuthBase:
                     f"No email found in iFlow credentials for '{path or 'in-memory object'}'."
                 )
 
-            # Update timestamp on check
+            # Update timestamp in cache only (not disk) to avoid overwriting
+            # potentially newer tokens that were saved by another process/refresh.
+            # The timestamp is non-critical metadata - losing it on restart is fine.
             if path and "_proxy_metadata" in creds:
                 creds["_proxy_metadata"]["last_check_timestamp"] = time.time()
-                await self._save_credentials(path, creds)
+                # Note: We intentionally don't save to disk here because:
+                # 1. The cache may have older tokens than disk (if external refresh occurred)
+                # 2. Saving would overwrite the newer disk tokens with stale cached ones
+                # 3. The timestamp is non-critical and will be updated on next refresh
 
             return {"email": email}
         except Exception as e:
@@ -1513,7 +1599,11 @@ class IFlowAuthBase:
                 )
 
             # Step 4: Save credentials to file
-            await self._save_credentials(str(file_path), new_creds)
+            if not await self._save_credentials(str(file_path), new_creds):
+                return IFlowCredentialSetupResult(
+                    success=False,
+                    error=f"Failed to save credentials to disk at {file_path.name}",
+                )
 
             return IFlowCredentialSetupResult(
                 success=True,

src/rotator_library/providers/qwen_auth_base.py

@@ -25,6 +25,7 @@ from rich.markup import escape as rich_escape
 from ..utils.headless_detection import is_headless_environment
 from ..utils.reauth_coordinator import get_reauth_coordinator
 from ..utils.resilient_io import safe_write_json
+from ..error_handler import CredentialNeedsReauthError
 
 lib_logger = logging.getLogger("rotator_library")
 
@@ -235,26 +236,42 @@ class QwenAuthBase:
                     return env_creds
                 raise  # Re-raise the original file not found error
 
-    async def _save_credentials(self, path: str, creds: Dict[str, Any]):
-        """Save credentials with in-memory fallback if disk unavailable."""
-        # Always update cache first (memory is reliable)
-        self._credentials_cache[path] = creds
+    async def _save_credentials(self, path: str, creds: Dict[str, Any]) -> bool:
+        """Save credentials to disk, then update cache. Returns True only if disk write succeeded.
+
+        For providers with rotating refresh tokens (like Qwen), disk persistence is CRITICAL.
+        If we update the cache but fail to write to disk:
+        - The old refresh_token on disk is now invalid (consumed by API)
+        - On restart, we'd load the invalid token and require re-auth
 
+        By writing to disk FIRST, we ensure:
+        - Cache only updated after disk succeeds (guaranteed parity)
+        - If disk fails, cache keeps old tokens, refresh is retried
+        - No desync between cache and disk is possible
+        """
         # Don't save to file if credentials were loaded from environment
         if creds.get("_proxy_metadata", {}).get("loaded_from_env"):
+            self._credentials_cache[path] = creds
             lib_logger.debug("Credentials loaded from env, skipping file save")
-            return
+            return True
 
-        # Attempt disk write - if it fails, we still have the cache
-        # buffer_on_failure ensures data is retried periodically and saved on shutdown
-        if safe_write_json(
-            path, creds, lib_logger, secure_permissions=True, buffer_on_failure=True
+        # Write to disk FIRST - do NOT buffer on failure for rotating tokens
+        # Buffering is dangerous because the refresh_token may be stale by retry time
+        if not safe_write_json(
+            path, creds, lib_logger, secure_permissions=True, buffer_on_failure=False
         ):
-            lib_logger.debug(f"Saved updated Qwen OAuth credentials to '{path}'.")
-        else:
-            lib_logger.warning(
-                "Qwen credentials cached in memory only (buffered for retry)."
+            lib_logger.error(
+                f"Failed to write Qwen credentials to disk for '{Path(path).name}'. "
+                f"Cache NOT updated to maintain parity with disk."
             )
+            return False
+
+        # Disk write succeeded - now update cache (guaranteed parity)
+        self._credentials_cache[path] = creds
+        lib_logger.debug(
+            f"Saved updated Qwen OAuth credentials to '{Path(path).name}'."
+        )
+        return True
 
     def _is_token_expired(self, creds: Dict[str, Any]) -> bool:
         expiry_timestamp = creds.get("expiry_date", 0) / 1000
@@ -275,10 +292,11 @@ class QwenAuthBase:
             if not force and cached_creds and not self._is_token_expired(cached_creds):
                 return cached_creds
 
-            # If cache is empty, read from file. This is safe because we hold the lock.
-            if path not in self._credentials_cache:
-                await self._read_creds_from_file(path)
-
+            # [ROTATING TOKEN FIX] Always read fresh from disk before refresh.
+            # Qwen uses rotating refresh tokens - each refresh invalidates the previous token.
+            # If we use a stale cached token, refresh will fail with HTTP 400.
+            # Reading fresh from disk ensures we have the latest token.
+            await self._read_creds_from_file(path)
             creds_from_file = self._credentials_cache[path]
 
             lib_logger.debug(f"Refreshing Qwen OAuth token for '{Path(path).name}'...")
@@ -291,7 +309,6 @@ class QwenAuthBase:
             max_retries = 3
             new_token_data = None
             last_error = None
-            needs_reauth = False
 
             headers = {
                 "Content-Type": "application/x-www-form-urlencoded",
@@ -324,14 +341,57 @@ class QwenAuthBase:
                             f"HTTP {status_code} for '{Path(path).name}': {error_body}"
                         )
 
-                        # [INVALID GRANT HANDLING] Handle 401/403 by triggering re-authentication
-                        if status_code in (401, 403):
-                            lib_logger.warning(
-                                f"Refresh token invalid for '{Path(path).name}' (HTTP {status_code}). "
-                                f"Token may have been revoked or expired. Starting re-authentication..."
+                        # [INVALID GRANT HANDLING] Handle 400/401/403 by raising
+                        # The caller (_process_refresh_queue or initialize_token) will handle re-auth
+                        # We must NOT call initialize_token from here as we hold a lock (would deadlock)
+                        if status_code == 400:
+                            # Check if this is an invalid refresh token error
+                            try:
+                                error_data = e.response.json()
+                                error_type = error_data.get("error", "")
+                                error_desc = error_data.get("error_description", "")
+                            except Exception:
+                                error_type = ""
+                                error_desc = error_body
+
+                            if (
+                                "invalid" in error_desc.lower()
+                                or error_type == "invalid_request"
+                            ):
+                                lib_logger.info(
+                                    f"Credential '{Path(path).name}' needs re-auth (HTTP 400: {error_desc}). "
+                                    f"Queued for re-authentication, rotating to next credential."
+                                )
+                                # Queue for re-auth in background (non-blocking, fire-and-forget)
+                                # This ensures credential gets fixed even if caller doesn't handle it
+                                asyncio.create_task(
+                                    self._queue_refresh(
+                                        path, force=True, needs_reauth=True
+                                    )
+                                )
+                                # Raise rotatable error instead of raw HTTPStatusError
+                                raise CredentialNeedsReauthError(
+                                    credential_path=path,
+                                    message=f"Refresh token invalid for '{Path(path).name}'. Re-auth queued.",
+                                )
+                            else:
+                                # Other 400 error - raise it
+                                raise
+
+                        elif status_code in (401, 403):
+                            lib_logger.info(
+                                f"Credential '{Path(path).name}' needs re-auth (HTTP {status_code}). "
+                                f"Queued for re-authentication, rotating to next credential."
+                            )
+                            # Queue for re-auth in background (non-blocking, fire-and-forget)
+                            asyncio.create_task(
+                                self._queue_refresh(path, force=True, needs_reauth=True)
+                            )
+                            # Raise rotatable error instead of raw HTTPStatusError
+                            raise CredentialNeedsReauthError(
+                                credential_path=path,
+                                message=f"Token invalid for '{Path(path).name}' (HTTP {status_code}). Re-auth queued.",
                             )
-                            needs_reauth = True
-                            break  # Exit retry loop to trigger re-auth
 
                         elif status_code == 429:
                             retry_after = int(e.response.headers.get("Retry-After", 60))
@@ -367,37 +427,6 @@ class QwenAuthBase:
                             continue
                         raise
 
-            # [INVALID GRANT RE-AUTH] Trigger OAuth flow if refresh token is invalid
-            if needs_reauth:
-                lib_logger.info(
-                    f"Starting re-authentication for '{Path(path).name}'..."
-                )
-                try:
-                    # Call initialize_token to trigger OAuth flow
-                    new_creds = await self.initialize_token(path)
-                    # Clear backoff on successful re-auth
-                    self._refresh_failures.pop(path, None)
-                    self._next_refresh_after.pop(path, None)
-                    return new_creds
-                except Exception as reauth_error:
-                    lib_logger.error(
-                        f"Re-authentication failed for '{Path(path).name}': {reauth_error}"
-                    )
-                    # [BACKOFF TRACKING] Increment failure count and set backoff timer
-                    self._refresh_failures[path] = (
-                        self._refresh_failures.get(path, 0) + 1
-                    )
-                    backoff_seconds = min(
-                        300, 30 * (2 ** self._refresh_failures[path])
-                    )  # Max 5 min backoff
-                    self._next_refresh_after[path] = time.time() + backoff_seconds
-                    lib_logger.debug(
-                        f"Setting backoff for '{Path(path).name}': {backoff_seconds}s"
-                    )
-                    raise ValueError(
-                        f"Refresh token invalid and re-authentication failed: {reauth_error}"
-                    )
-
             if new_token_data is None:
                 # [BACKOFF TRACKING] Increment failure count and set backoff timer
                 self._refresh_failures[path] = self._refresh_failures.get(path, 0) + 1
@@ -440,11 +469,20 @@ class QwenAuthBase:
             self._refresh_failures.pop(path, None)
             self._next_refresh_after.pop(path, None)
 
-            await self._save_credentials(path, creds_from_file)
+            # Save credentials - MUST succeed for rotating token providers
+            if not await self._save_credentials(path, creds_from_file):
+                # CRITICAL: For rotating tokens, if we can't persist the new token,
+                # the old token is already invalidated by Qwen. This is a critical failure.
+                # Raise an error so retry logic kicks in.
+                raise IOError(
+                    f"Failed to persist refreshed credentials for '{Path(path).name}'. "
+                    f"Disk write failed - refresh will be retried."
+                )
+
             lib_logger.debug(
                 f"Successfully refreshed Qwen OAuth token for '{Path(path).name}'."
             )
-            return creds_from_file
+            return self._credentials_cache[path]  # Return from cache (synced with disk)
 
     async def get_api_details(self, credential_identifier: str) -> Tuple[str, str]:
         """
@@ -689,12 +727,37 @@ class QwenAuthBase:
 
                     except httpx.HTTPStatusError as e:
                         status_code = e.response.status_code
-                        if status_code in (401, 403):
-                            # Invalid refresh token - route to re-auth queue
-                            lib_logger.warning(
-                                f"Refresh token invalid for '{Path(path).name}' (HTTP {status_code}). "
+                        # Check for invalid refresh token errors (400/401/403)
+                        # These need to be routed to re-auth queue for interactive OAuth
+                        needs_reauth = False
+
+                        if status_code == 400:
+                            # Check if this is an invalid refresh token error
+                            try:
+                                error_data = e.response.json()
+                                error_type = error_data.get("error", "")
+                                error_desc = error_data.get("error_description", "")
+                            except Exception:
+                                error_type = ""
+                                error_desc = str(e)
+
+                            if (
+                                "invalid" in error_desc.lower()
+                                or error_type == "invalid_request"
+                            ):
+                                needs_reauth = True
+                                lib_logger.info(
+                                    f"Credential '{Path(path).name}' needs re-auth (HTTP 400: {error_desc}). "
+                                    f"Routing to re-auth queue."
+                                )
+                        elif status_code in (401, 403):
+                            needs_reauth = True
+                            lib_logger.info(
+                                f"Credential '{Path(path).name}' needs re-auth (HTTP {status_code}). "
                                 f"Routing to re-auth queue."
                             )
+
+                        if needs_reauth:
                             self._queue_retry_count.pop(path, None)  # Clear retry count
                             async with self._queue_tracking_lock:
                                 self._queued_credentials.discard(
@@ -789,7 +852,7 @@ class QwenAuthBase:
 
                 try:
                     lib_logger.info(f"Starting re-auth for '{Path(path).name}'...")
-                    await self.initialize_token(path)
+                    await self.initialize_token(path, force_interactive=True)
                     lib_logger.info(f"Re-auth SUCCESS for '{Path(path).name}'")
 
                 except Exception as e:
@@ -1001,14 +1064,20 @@ class QwenAuthBase:
                     }
 
             if path:
-                await self._save_credentials(path, creds)
+                if not await self._save_credentials(path, creds):
+                    raise IOError(
+                        f"Failed to save OAuth credentials to disk for '{display_name}'. "
+                        f"Please retry authentication."
+                    )
             lib_logger.info(
                 f"Qwen OAuth initialized successfully for '{display_name}'."
             )
         return creds
 
     async def initialize_token(
-        self, creds_or_path: Union[Dict[str, Any], str]
+        self,
+        creds_or_path: Union[Dict[str, Any], str],
+        force_interactive: bool = False,
     ) -> Dict[str, Any]:
         """
         Initialize OAuth token, triggering interactive device flow if needed.
@@ -1016,6 +1085,12 @@ class QwenAuthBase:
         If interactive OAuth is required (expired refresh token, missing credentials, etc.),
         the flow is coordinated globally via ReauthCoordinator to ensure only one
         interactive OAuth flow runs at a time across all providers.
+
+        Args:
+            creds_or_path: Either a credentials dict or path to credentials file.
+            force_interactive: If True, skip expiry checks and force interactive OAuth.
+                               Use this when the refresh token is known to be invalid
+                               (e.g., after HTTP 400 from token endpoint).
         """
         path = creds_or_path if isinstance(creds_or_path, str) else None
 
@@ -1034,7 +1109,11 @@ class QwenAuthBase:
             )
 
             reason = ""
-            if not creds.get("refresh_token"):
+            if force_interactive:
+                reason = (
+                    "re-authentication was explicitly requested (refresh token invalid)"
+                )
+            elif not creds.get("refresh_token"):
                 reason = "refresh token is missing"
             elif self._is_token_expired(creds):
                 reason = "token is expired"
@@ -1108,10 +1187,15 @@ class QwenAuthBase:
                     f"No email found in _proxy_metadata for '{path or 'in-memory object'}'."
                 )
 
-            # Update timestamp on check and save if it's a file-based credential
+            # Update timestamp in cache only (not disk) to avoid overwriting
+            # potentially newer tokens that were saved by another process/refresh.
+            # The timestamp is non-critical metadata - losing it on restart is fine.
             if path and "_proxy_metadata" in creds:
                 creds["_proxy_metadata"]["last_check_timestamp"] = time.time()
-                await self._save_credentials(path, creds)
+                # Note: We intentionally don't save to disk here because:
+                # 1. The cache may have older tokens than disk (if external refresh occurred)
+                # 2. Saving would overwrite the newer disk tokens with stale cached ones
+                # 3. The timestamp is non-critical and will be updated on next refresh
 
             return {"email": email}
         except Exception as e:
@@ -1230,7 +1314,11 @@ class QwenAuthBase:
                 )
 
             # Step 4: Save credentials to file
-            await self._save_credentials(str(file_path), new_creds)
+            if not await self._save_credentials(str(file_path), new_creds):
+                return QwenCredentialSetupResult(
+                    success=False,
+                    error=f"Failed to save credentials to disk at {file_path.name}",
+                )
 
             return QwenCredentialSetupResult(
                 success=True,