feat(auth): extract GoogleOAuthBase and add antigravity provider

- Add providers/google_oauth_base.py to centralize Google OAuth logic (auth flow, token refresh, env loading, atomic saves, backoff/retry, queueing, headless support, and validation).
- Migrate GeminiAuthBase and AntigravityAuthBase to inherit from GoogleOAuthBase and expose provider-specific constants (CLIENT_ID, CLIENT_SECRET, OAUTH_SCOPES, ENV_PREFIX, CALLBACK_PORT, CALLBACK_PATH).
- Register "antigravity" in DEFAULT_OAUTH_DIRS and mark it as OAuth-only in credential_tool; include a user-friendly display name for interactive flows.
- Remove large duplicated OAuth implementations from provider-specific files and consolidate behavior to reduce maintenance surface and ensure consistent token handling.

src/rotator_library/credential_manager.py

@@ -14,6 +14,7 @@ DEFAULT_OAUTH_DIRS = {
     "gemini_cli": Path.home() / ".gemini",
     "qwen_code": Path.home() / ".qwen",
     "iflow": Path.home() / ".iflow",
+    "antigravity": Path.home() / ".antigravity",
     # Add other providers like 'claude' here if they have a standard CLI path
 }
 

src/rotator_library/credential_tool.py

@@ -98,7 +98,7 @@ async def setup_api_key():
     # Discover custom providers and add them to the list
     # Note: gemini_cli is OAuth-only, but qwen_code and iflow support both OAuth and API keys
     _, PROVIDER_PLUGINS = _ensure_providers_loaded()
-    oauth_only_providers = {'gemini_cli'}
+    oauth_only_providers = {'gemini_cli', 'antigravity'}
     discovered_providers = {
         p.replace('_', ' ').title(): p.upper() + "_API_KEY"
         for p in PROVIDER_PLUGINS.keys()
@@ -195,7 +195,8 @@ async def setup_new_credential(provider_name: str):
         oauth_friendly_names = {
             "gemini_cli": "Gemini CLI (OAuth)",
             "qwen_code": "Qwen Code (OAuth - also supports API keys)",
-            "iflow": "iFlow (OAuth - also supports API keys)"
+            "iflow": "iFlow (OAuth - also supports API keys)",
+            "antigravity": "Antigravity (OAuth)"
         }
         display_name = oauth_friendly_names.get(provider_name, provider_name.replace('_', ' ').title())
 
@@ -578,7 +579,8 @@ async def main(clear_on_start=True):
             oauth_friendly_names = {
                 "gemini_cli": "Gemini CLI (OAuth)",
                 "qwen_code": "Qwen Code (OAuth - also supports API keys)",
-                "iflow": "iFlow (OAuth - also supports API keys)"
+                "iflow": "iFlow (OAuth - also supports API keys)",
+                "antigravity": "Antigravity (OAuth)",
             }
             
             provider_text = Text()

src/rotator_library/providers/antigravity_auth_base.py

@@ -1,466 +1,24 @@
 # src/rotator_library/providers/antigravity_auth_base.py
 
-import os
-import webbrowser
-from typing import Union, Optional
-import json
-import time
-import asyncio
-import logging
-from pathlib import Path
-from typing import Dict, Any
-import tempfile
-import shutil
+from .google_oauth_base import GoogleOAuthBase
 
-import httpx
-from rich.console import Console
-from rich.panel import Panel
-from rich.text import Text
-
-from ..utils.headless_detection import is_headless_environment
-
-lib_logger = logging.getLogger('rotator_library')
-
-# Antigravity OAuth credentials from CLIProxyAPI
-CLIENT_ID = "1071006060591-tmhssin2h21lcre235vtolojh4g403ep.apps.googleusercontent.com"
-CLIENT_SECRET = "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"
-TOKEN_URI = "https://oauth2.googleapis.com/token"
-USER_INFO_URI = "https://www.googleapis.com/oauth2/v1/userinfo"
-REFRESH_EXPIRY_BUFFER_SECONDS = 30 * 60  # 30 minutes buffer before expiry
-
-# Antigravity requires additional scopes
-OAUTH_SCOPES = [
-    "https://www.googleapis.com/auth/cloud-platform",
-    "https://www.googleapis.com/auth/userinfo.email",
-    "https://www.googleapis.com/auth/userinfo.profile",
-    "https://www.googleapis.com/auth/cclog",  # Antigravity-specific
-    "https://www.googleapis.com/auth/experimentsandconfigs"  # Antigravity-specific
-]
-
-console = Console()
-
-class AntigravityAuthBase:
+class AntigravityAuthBase(GoogleOAuthBase):
     """
-    Base authentication class for Antigravity provider.
-    Handles OAuth2 flow, token management, and refresh logic.
+    Antigravity OAuth2 authentication implementation.
     
-    Based on GeminiAuthBase but uses Antigravity-specific OAuth credentials and scopes.
+    Inherits all OAuth functionality from GoogleOAuthBase with Antigravity-specific configuration.
+    Uses Antigravity's OAuth credentials and includes additional scopes for cclog and experimentsandconfigs.
     """
     
-    def __init__(self):
-        self._credentials_cache: Dict[str, Dict[str, Any]] = {}
-        self._refresh_locks: Dict[str, asyncio.Lock] = {}
-        self._locks_lock = asyncio.Lock()  # Protects the locks dict from race conditions
-        # [BACKOFF TRACKING] Track consecutive failures per credential
-        self._refresh_failures: Dict[str, int] = {}  # Track consecutive failures per credential
-        self._next_refresh_after: Dict[str, float] = {}  # Track backoff timers (Unix timestamp)
-        
-        # [QUEUE SYSTEM] Sequential refresh processing
-        self._refresh_queue: asyncio.Queue = asyncio.Queue()
-        self._queued_credentials: set = set()  # Track credentials already in queue
-        self._unavailable_credentials: set = set()  # Mark credentials unavailable during re-auth
-        self._queue_tracking_lock = asyncio.Lock()  # Protects queue sets
-        self._queue_processor_task: Optional[asyncio.Task] = None  # Background worker task
-
-    def _load_from_env(self) -> Optional[Dict[str, Any]]:
-        """
-        Load OAuth credentials from environment variables for stateless deployments.
-
-        Expected environment variables:
-        - ANTIGRAVITY_ACCESS_TOKEN (required)
-        - ANTIGRAVITY_REFRESH_TOKEN (required)
-        - ANTIGRAVITY_EXPIRY_DATE (optional, defaults to 0)
-        - ANTIGRAVITY_CLIENT_ID (optional, uses default)
-        - ANTIGRAVITY_CLIENT_SECRET (optional, uses default)
-        - ANTIGRAVITY_TOKEN_URI (optional, uses default)
-        - ANTIGRAVITY_UNIVERSE_DOMAIN (optional, defaults to googleapis.com)
-        - ANTIGRAVITY_EMAIL (optional, defaults to "env-user")
-
-        Returns:
-            Dict with credential structure if env vars present, None otherwise
-        """
-        access_token = os.getenv("ANTIGRAVITY_ACCESS_TOKEN")
-        refresh_token = os.getenv("ANTIGRAVITY_REFRESH_TOKEN")
-
-        # Both access and refresh tokens are required
-        if not (access_token and refresh_token):
-            return None
-
-        lib_logger.debug("Loading Antigravity credentials from environment variables")
-
-        # Parse expiry_date as float, default to 0 if not present
-        expiry_str = os.getenv("ANTIGRAVITY_EXPIRY_DATE", "0")
-        try:
-            expiry_date = float(expiry_str)
-        except ValueError:
-            lib_logger.warning(f"Invalid ANTIGRAVITY_EXPIRY_DATE value: {expiry_str}, using 0")
-            expiry_date = 0
-
-        creds = {
-            "access_token": access_token,
-            "refresh_token": refresh_token,
-            "expiry_date": expiry_date,
-            "client_id": os.getenv("ANTIGRAVITY_CLIENT_ID", CLIENT_ID),
-            "client_secret": os.getenv("ANTIGRAVITY_CLIENT_SECRET", CLIENT_SECRET),
-            "token_uri": os.getenv("ANTIGRAVITY_TOKEN_URI", TOKEN_URI),
-            "universe_domain": os.getenv("ANTIGRAVITY_UNIVERSE_DOMAIN", "googleapis.com"),
-            "_proxy_metadata": {
-                "email": os.getenv("ANTIGRAVITY_EMAIL", "env-user"),
-                "last_check_timestamp": time.time(),
-                "loaded_from_env": True  # Flag to indicate env-based credentials
-            }
-        }
-
-        return creds
-
-    async def _load_credentials(self, path: str) -> Dict[str, Any]:
-        """
-        Load credentials from a file. First attempts file-based load,
-        then falls back to environment variables if file not found.
-
-        Args:
-            path: File path to load credentials from
-
-        Returns:
-            Dict containing the credentials
-
-        Raises:
-            ValueError: If credentials cannot be loaded from either source
-        """
-        # If path is special marker "env", load from environment
-        if path == "env":
-            env_creds = self._load_from_env()
-            if env_creds:
-                lib_logger.debug("Using Antigravity credentials from environment variables")
-                return env_creds
-            raise ValueError("ANTIGRAVITY_ACCESS_TOKEN and ANTIGRAVITY_REFRESH_TOKEN environment variables not set")
-
-        # Try loading from cache first
-        if path in self._credentials_cache:
-            cached_creds = self._credentials_cache[path]
-            lib_logger.debug(f"Using cached Antigravity credentials for: {Path(path).name}")
-            return cached_creds
-
-        # Try loading from file
-        try:
-            with open(path, 'r') as f:
-                creds = json.load(f)
-            self._credentials_cache[path] = creds
-            lib_logger.debug(f"Loaded Antigravity credentials from file: {Path(path).name}")
-            return creds
-        except FileNotFoundError:
-            # Fall back to environment variables
-            lib_logger.debug(f"Credential file not found: {path}, attempting environment variables")
-            env_creds = self._load_from_env()
-            if env_creds:
-                lib_logger.debug("Using Antigravity credentials from environment variables as fallback")
-                # Cache with special path marker
-                self._credentials_cache[path] = env_creds
-                return env_creds
-            raise ValueError(f"Credential file not found: {path} and environment variables not set")
-        except json.JSONDecodeError as e:
-            raise ValueError(f"Invalid JSON in credential file {path}: {e}")
-
-    async def _save_credentials(self, path: str, creds: Dict[str, Any]) -> None:
-        """
-        Save credentials to a file. Skip if credentials were loaded from environment.
-
-        Args:
-            path: File path to save credentials to
-            creds: Credentials dictionary to save
-        """
-        # Don't save environment-based credentials to file
-        if creds.get("_proxy_metadata", {}).get("loaded_from_env"):
-            lib_logger.debug("Skipping credential save (loaded from environment)")
-            return
-
-        # Don't save if path is special marker
-        if path == "env":
-            return
-
-        try:
-            # Ensure directory exists
-            Path(path).parent.mkdir(parents=True, exist_ok=True)
-            
-            # Write atomically using temp file + rename
-            temp_fd, temp_path = tempfile.mkstemp(
-                dir=Path(path).parent,
-                prefix='.tmp_',
-                suffix='.json'
-            )
-            try:
-                with os.fdopen(temp_fd, 'w') as f:
-                    json.dump(creds, f, indent=2)
-                shutil.move(temp_path, path)
-                lib_logger.debug(f"Saved Antigravity credentials to: {Path(path).name}")
-            except Exception:
-                # Clean up temp file on error
-                try:
-                    os.unlink(temp_path)
-                except Exception:
-                    pass
-                raise
-        except Exception as e:
-            lib_logger.warning(f"Failed to save Antigravity credentials to {path}: {e}")
-
-    def _is_token_expired(self, creds: Dict[str, Any]) -> bool:
-        """
-        Check if the access token is expired or close to expiry.
-
-        Args:
-            creds: Credentials dict with expiry_date field (in milliseconds)
-
-        Returns:
-            True if token is expired or within buffer time of expiry
-        """
-        if 'expiry_date' not in creds:
-            return True
-
-        # expiry_date is in milliseconds
-        expiry_timestamp = creds['expiry_date'] / 1000.0
-        current_time = time.time()
-        
-        # Consider expired if within buffer time
-        return (expiry_timestamp - current_time) <= REFRESH_EXPIRY_BUFFER_SECONDS
-
-    async def _refresh_token(self, path: str, creds: Dict[str, Any]) -> Dict[str, Any]:
-        """
-        Refresh an expired OAuth token using the refresh token.
-
-        Args:
-            path: Credential file path (for saving updated credentials)
-            creds: Current credentials dict with refresh_token
-
-        Returns:
-            Updated credentials dict with fresh access token
-
-        Raises:
-            ValueError: If refresh fails
-        """
-        if 'refresh_token' not in creds:
-            raise ValueError("No refresh token available")
-
-        lib_logger.debug(f"Refreshing Antigravity OAuth token for: {Path(path).name if path != 'env' else 'env'}")
-
-        client_id = creds.get('client_id', CLIENT_ID)
-        client_secret = creds.get('client_secret', CLIENT_SECRET)
-        token_uri = creds.get('token_uri', TOKEN_URI)
-
-        async with httpx.AsyncClient() as client:
-            try:
-                response = await client.post(
-                    token_uri,
-                    data={
-                        'client_id': client_id,
-                        'client_secret': client_secret,
-                        'refresh_token': creds['refresh_token'],
-                        'grant_type': 'refresh_token'
-                    },
-                    timeout=30.0
-                )
-                response.raise_for_status()
-                token_data = response.json()
-
-                # Update credentials with new token
-                creds['access_token'] = token_data['access_token']
-                creds['expiry_date'] = (time.time() + token_data['expires_in']) * 1000
-
-                # Update metadata
-                if '_proxy_metadata' not in creds:
-                    creds['_proxy_metadata'] = {}
-                creds['_proxy_metadata']['last_check_timestamp'] = time.time()
-
-                # Save updated credentials
-                await self._save_credentials(path, creds)
-
-                # Update cache
-                self._credentials_cache[path] = creds
-
-                # Reset failure count on success
-                self._refresh_failures[path] = 0
-
-                lib_logger.info(f"Successfully refreshed Antigravity OAuth token for: {Path(path).name if path != 'env' else 'env'}")
-                return creds
-
-            except httpx.HTTPStatusError as e:
-                # Track failures for backoff
-                self._refresh_failures[path] = self._refresh_failures.get(path, 0) + 1
-                raise ValueError(f"Failed to refresh Antigravity token (HTTP {e.response.status_code}): {e.response.text}")
-            except Exception as e:
-                self._refresh_failures[path] = self._refresh_failures.get(path, 0) + 1
-                raise ValueError(f"Failed to refresh Antigravity token: {e}")
-
-    async def initialize_token(self, creds_or_path: Union[Dict[str, Any], str]) -> Dict[str, Any]:
-        """
-        Initialize or refresh an OAuth token. Handles the complete OAuth flow if needed.
-
-        Args:
-            creds_or_path: Either a credentials dict or a file path string
-
-        Returns:
-            Valid credentials dict with fresh access token
-        """
-        path = creds_or_path if isinstance(creds_or_path, str) else None
-
-        if isinstance(creds_or_path, dict):
-            display_name = creds_or_path.get("_proxy_metadata", {}).get("display_name", "in-memory object")
-        else:
-            display_name = Path(path).name if path and path != "env" else "env"
-
-        lib_logger.debug(f"Initializing Antigravity token for '{display_name}'...")
-        
-        try:
-            creds = await self._load_credentials(creds_or_path) if path else creds_or_path
-            reason = ""
-            if not creds.get("refresh_token"):
-                reason = "refresh token is missing"
-            elif self._is_token_expired(creds):
-                reason = "token is expired"
-
-            if reason:
-                if reason == "token is expired" and creds.get("refresh_token"):
-                    try:
-                        return await self._refresh_token(path, creds)
-                    except Exception as e:
-                        lib_logger.warning(f"Automatic token refresh for '{display_name}' failed: {e}. Proceeding to interactive login.")
-
-                lib_logger.warning(f"Antigravity OAuth token for '{display_name}' needs setup: {reason}.")
-                
-                is_headless = is_headless_environment()
-                
-                auth_code_future = asyncio.get_event_loop().create_future()
-                server = None
-
-                async def handle_callback(reader, writer):
-                    try:
-                        request_line_bytes = await reader.readline()
-                        if not request_line_bytes:
-                            return
-                        path_str = request_line_bytes.decode('utf-8').strip().split(' ')[1]
-                        # Consume headers
-                        while await reader.readline() != b'\r\n':
-                            pass
-                        
-                        from urllib.parse import urlparse, parse_qs
-                        query_params = parse_qs(urlparse(path_str).query)
-                        
-                        writer.write(b"HTTP/1.1 200 OK\r\nContent-Type: text/html\r\n\r\n")
-                        if 'code' in query_params:
-                            if not auth_code_future.done():
-                                auth_code_future.set_result(query_params['code'][0])
-                            writer.write(b"<html><body><h1>Authentication successful!</h1><p>You can close this window.</p></body></html>")
-                        else:
-                            error = query_params.get('error', ['Unknown error'])[0]
-                            if not auth_code_future.done():
-                                auth_code_future.set_exception(Exception(f"OAuth failed: {error}"))
-                            writer.write(f"<html><body><h1>Authentication Failed</h1><p>Error: {error}. Please try again.</p></body></html>".encode())
-                        await writer.drain()
-                    except Exception as e:
-                        lib_logger.error(f"Error in OAuth callback handler: {e}")
-                    finally:
-                        writer.close()
-
-                try:
-                    server = await asyncio.start_server(handle_callback, '127.0.0.1', 8085)
-                    
-                    from urllib.parse import urlencode
-                    auth_url = "https://accounts.google.com/o/oauth2/v2/auth?" + urlencode({
-                        "client_id": CLIENT_ID,
-                        "redirect_uri": "http://localhost:8085/oauth2callback",
-                        "scope": " ".join(OAUTH_SCOPES),
-                        "access_type": "offline",
-                        "response_type": "code",
-                        "prompt": "consent"
-                    })
-                    
-                    if is_headless:
-                        auth_panel_text = Text.from_markup(
-                            "Running in headless environment (no GUI detected).\n"
-                            "Please open the URL below in a browser on another machine to authorize:\n"
-                        )
-                    else:
-                        auth_panel_text = Text.from_markup(
-                            "1. Your browser will now open to log in and authorize the application.\n"
-                            "2. If it doesn't open automatically, please open the URL below manually."
-                        )
-                    
-                    console.print(Panel(auth_panel_text, title=f"Antigravity OAuth Setup for [bold yellow]{display_name}[/bold yellow]", style="bold blue"))
-                    console.print(f"[bold]URL:[/bold] [link={auth_url}]{auth_url}[/link]\n")
-                    
-                    if not is_headless:
-                        try:
-                            webbrowser.open(auth_url)
-                            lib_logger.info("Browser opened successfully for OAuth flow")
-                        except Exception as e:
-                            lib_logger.warning(f"Failed to open browser automatically: {e}. Please open the URL manually.")
-                    
-                    with console.status("[bold green]Waiting for you to complete authentication in the browser...[/bold green]", spinner="dots"):
-                        auth_code = await asyncio.wait_for(auth_code_future, timeout=300)
-                except asyncio.TimeoutError:
-                    raise Exception("OAuth flow timed out. Please try again.")
-                finally:
-                    if server:
-                        server.close()
-                        await server.wait_closed()
-                
-                lib_logger.info(f"Attempting to exchange authorization code for tokens...")
-                async with httpx.AsyncClient() as client:
-                    response = await client.post(TOKEN_URI, data={
-                        "code": auth_code.strip(),
-                        "client_id": CLIENT_ID,
-                        "client_secret": CLIENT_SECRET,
-                        "redirect_uri": "http://localhost:8085/oauth2callback",
-                        "grant_type": "authorization_code"
-                    })
-                    response.raise_for_status()
-                    token_data = response.json()
-                    
-                    creds = token_data.copy()
-                    creds["expiry_date"] = (time.time() + creds.pop("expires_in")) * 1000
-                    creds["client_id"] = CLIENT_ID
-                    creds["client_secret"] = CLIENT_SECRET
-                    creds["token_uri"] = TOKEN_URI
-                    creds["universe_domain"] = "googleapis.com"
-                    
-                    # Fetch user info
-                    user_info_response = await client.get(
-                        USER_INFO_URI,
-                        headers={"Authorization": f"Bearer {creds['access_token']}"}
-                    )
-                    user_info_response.raise_for_status()
-                    user_info = user_info_response.json()
-                    
-                    creds["_proxy_metadata"] = {
-                        "email": user_info.get("email"),
-                        "last_check_timestamp": time.time()
-                    }
-
-                    if path:
-                        await self._save_credentials(path, creds)
-                    
-                    lib_logger.info(f"Antigravity OAuth initialized successfully for '{display_name}'.")
-                    return creds
-
-            lib_logger.info(f"Antigravity OAuth token at '{display_name}' is valid.")
-            return creds
-        except Exception as e:
-            raise ValueError(f"Failed to initialize Antigravity OAuth for '{display_name}': {e}")
-
-    async def get_valid_token(self, credential_path: str) -> str:
-        """
-        Get a valid access token, refreshing if necessary.
-
-        Args:
-            credential_path: Path to credential file or "env" for environment variables
-
-        Returns:
-            Valid access token string
-
-        Raises:
-            ValueError: If token cannot be obtained
-        """
-        try:
-            creds = await self.initialize_token(credential_path)
-            return creds['access_token']
-        except Exception as e:
-            raise ValueError(f"Failed to get valid Antigravity token: {e}")
+    CLIENT_ID = "1071006060591-tmhssin2h21lcre235vtolojh4g403ep.apps.googleusercontent.com"
+    CLIENT_SECRET = "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"
+    OAUTH_SCOPES = [
+        "https://www.googleapis.com/auth/cloud-platform",
+        "https://www.googleapis.com/auth/userinfo.email",
+        "https://www.googleapis.com/auth/userinfo.profile",
+        "https://www.googleapis.com/auth/cclog",  # Antigravity-specific
+        "https://www.googleapis.com/auth/experimentsandconfigs",  # Antigravity-specific
+    ]
+    ENV_PREFIX = "ANTIGRAVITY"
+    CALLBACK_PORT = 51121
+    CALLBACK_PATH = "/oauthcallback"

src/rotator_library/providers/gemini_auth_base.py

@@ -1,625 +1,21 @@
 # src/rotator_library/providers/gemini_auth_base.py
 
-import os
-import webbrowser
-from typing import Union, Optional
-import json
-import time
-import asyncio
-import logging
-from pathlib import Path
-from typing import Dict, Any
-import tempfile
-import shutil
-
-import httpx
-from rich.console import Console
-from rich.panel import Panel
-from rich.text import Text
-
-from ..utils.headless_detection import is_headless_environment
-
-lib_logger = logging.getLogger('rotator_library')
-
-CLIENT_ID = "681255809395-oo8ft2oprdrnp9e3aqf6av3hmdib135j.apps.googleusercontent.com" #https://api.kilocode.ai/extension-config.json
-CLIENT_SECRET = "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" #https://api.kilocode.ai/extension-config.json
-TOKEN_URI = "https://oauth2.googleapis.com/token"
-USER_INFO_URI = "https://www.googleapis.com/oauth2/v1/userinfo"
-REFRESH_EXPIRY_BUFFER_SECONDS = 30 * 60  # 30 minutes buffer before expiry
-
-console = Console()
-
-class GeminiAuthBase:
-    def __init__(self):
-        self._credentials_cache: Dict[str, Dict[str, Any]] = {}
-        self._refresh_locks: Dict[str, asyncio.Lock] = {}
-        self._locks_lock = asyncio.Lock()  # Protects the locks dict from race conditions
-        # [BACKOFF TRACKING] Track consecutive failures per credential
-        self._refresh_failures: Dict[str, int] = {}  # Track consecutive failures per credential
-        self._next_refresh_after: Dict[str, float] = {}  # Track backoff timers (Unix timestamp)
-        
-        # [QUEUE SYSTEM] Sequential refresh processing
-        self._refresh_queue: asyncio.Queue = asyncio.Queue()
-        self._queued_credentials: set = set()  # Track credentials already in queue
-        self._unavailable_credentials: set = set()  # Mark credentials unavailable during re-auth
-        self._queue_tracking_lock = asyncio.Lock()  # Protects queue sets
-        self._queue_processor_task: Optional[asyncio.Task] = None  # Background worker task
-
-    def _load_from_env(self) -> Optional[Dict[str, Any]]:
-        """
-        Load OAuth credentials from environment variables for stateless deployments.
-
-        Expected environment variables:
-        - GEMINI_CLI_ACCESS_TOKEN (required)
-        - GEMINI_CLI_REFRESH_TOKEN (required)
-        - GEMINI_CLI_EXPIRY_DATE (optional, defaults to 0)
-        - GEMINI_CLI_CLIENT_ID (optional, uses default)
-        - GEMINI_CLI_CLIENT_SECRET (optional, uses default)
-        - GEMINI_CLI_TOKEN_URI (optional, uses default)
-        - GEMINI_CLI_UNIVERSE_DOMAIN (optional, defaults to googleapis.com)
-        - GEMINI_CLI_EMAIL (optional, defaults to "env-user")
-        - GEMINI_CLI_PROJECT_ID (optional)
-        - GEMINI_CLI_TIER (optional)
-
-        Returns:
-            Dict with credential structure if env vars present, None otherwise
-        """
-        access_token = os.getenv("GEMINI_CLI_ACCESS_TOKEN")
-        refresh_token = os.getenv("GEMINI_CLI_REFRESH_TOKEN")
-
-        # Both access and refresh tokens are required
-        if not (access_token and refresh_token):
-            return None
-
-        lib_logger.debug("Loading Gemini CLI credentials from environment variables")
-
-        # Parse expiry_date as float, default to 0 if not present
-        expiry_str = os.getenv("GEMINI_CLI_EXPIRY_DATE", "0")
-        try:
-            expiry_date = float(expiry_str)
-        except ValueError:
-            lib_logger.warning(f"Invalid GEMINI_CLI_EXPIRY_DATE value: {expiry_str}, using 0")
-            expiry_date = 0
-
-        creds = {
-            "access_token": access_token,
-            "refresh_token": refresh_token,
-            "expiry_date": expiry_date,
-            "client_id": os.getenv("GEMINI_CLI_CLIENT_ID", CLIENT_ID),
-            "client_secret": os.getenv("GEMINI_CLI_CLIENT_SECRET", CLIENT_SECRET),
-            "token_uri": os.getenv("GEMINI_CLI_TOKEN_URI", TOKEN_URI),
-            "universe_domain": os.getenv("GEMINI_CLI_UNIVERSE_DOMAIN", "googleapis.com"),
-            "_proxy_metadata": {
-                "email": os.getenv("GEMINI_CLI_EMAIL", "env-user"),
-                "last_check_timestamp": time.time(),
-                "loaded_from_env": True  # Flag to indicate env-based credentials
-            }
-        }
-
-        # Add project_id if provided
-        project_id = os.getenv("GEMINI_CLI_PROJECT_ID")
-        if project_id:
-            creds["_proxy_metadata"]["project_id"] = project_id
-        
-        # Add tier if provided
-        tier = os.getenv("GEMINI_CLI_TIER")
-        if tier:
-            creds["_proxy_metadata"]["tier"] = tier
-
-        return creds
-
-    async def _load_credentials(self, path: str) -> Dict[str, Any]:
-        if path in self._credentials_cache:
-            return self._credentials_cache[path]
-
-        async with await self._get_lock(path):
-            if path in self._credentials_cache:
-                return self._credentials_cache[path]
-
-            # First, try loading from environment variables
-            env_creds = self._load_from_env()
-            if env_creds:
-                lib_logger.info("Using Gemini CLI credentials from environment variables")
-                # Cache env-based credentials using the path as key
-                self._credentials_cache[path] = env_creds
-                return env_creds
-
-            # Fall back to file-based loading
-            try:
-                lib_logger.debug(f"Loading Gemini credentials from file: {path}")
-                with open(path, 'r') as f:
-                    creds = json.load(f)
-                # Handle gcloud-style creds file which nest tokens under "credential"
-                if "credential" in creds:
-                    creds = creds["credential"]
-                self._credentials_cache[path] = creds
-                return creds
-            except FileNotFoundError:
-                raise IOError(f"Gemini OAuth credential file not found at '{path}'")
-            except Exception as e:
-                raise IOError(f"Failed to load Gemini OAuth credentials from '{path}': {e}")
-
-    async def _save_credentials(self, path: str, creds: Dict[str, Any]):
-        # Don't save to file if credentials were loaded from environment
-        if creds.get("_proxy_metadata", {}).get("loaded_from_env"):
-            lib_logger.debug("Credentials loaded from env, skipping file save")
-            # Still update cache for in-memory consistency
-            self._credentials_cache[path] = creds
-            return
-
-        # [ATOMIC WRITE] Use tempfile + move pattern to ensure atomic writes
-        # This prevents credential corruption if the process is interrupted during write
-        parent_dir = os.path.dirname(os.path.abspath(path))
-        os.makedirs(parent_dir, exist_ok=True)
-
-        tmp_fd = None
-        tmp_path = None
-        try:
-            # Create temp file in same directory as target (ensures same filesystem)
-            tmp_fd, tmp_path = tempfile.mkstemp(dir=parent_dir, prefix='.tmp_', suffix='.json', text=True)
-
-            # Write JSON to temp file
-            with os.fdopen(tmp_fd, 'w') as f:
-                json.dump(creds, f, indent=2)
-                tmp_fd = None  # fdopen closes the fd
-
-            # Set secure permissions (0600 = owner read/write only)
-            try:
-                os.chmod(tmp_path, 0o600)
-            except (OSError, AttributeError):
-                # Windows may not support chmod, ignore
-                pass
-
-            # Atomic move (overwrites target if it exists)
-            shutil.move(tmp_path, path)
-            tmp_path = None  # Successfully moved
-
-            # Update cache AFTER successful file write (prevents cache/file inconsistency)
-            self._credentials_cache[path] = creds
-            lib_logger.debug(f"Saved updated Gemini OAuth credentials to '{path}' (atomic write).")
-
-        except Exception as e:
-            lib_logger.error(f"Failed to save updated Gemini OAuth credentials to '{path}': {e}")
-            # Clean up temp file if it still exists
-            if tmp_fd is not None:
-                try:
-                    os.close(tmp_fd)
-                except:
-                    pass
-            if tmp_path and os.path.exists(tmp_path):
-                try:
-                    os.unlink(tmp_path)
-                except:
-                    pass
-            raise
-
-    def _is_token_expired(self, creds: Dict[str, Any]) -> bool:
-        expiry = creds.get("token_expiry") # gcloud format
-        if not expiry: # gemini-cli format
-             expiry_timestamp = creds.get("expiry_date", 0) / 1000
-        else:
-            expiry_timestamp = time.mktime(time.strptime(expiry, "%Y-%m-%dT%H:%M:%SZ"))
-        return expiry_timestamp < time.time() + REFRESH_EXPIRY_BUFFER_SECONDS
-
-    async def _refresh_token(self, path: str, creds: Dict[str, Any], force: bool = False) -> Dict[str, Any]:
-        async with await self._get_lock(path):
-            # Skip the expiry check if a refresh is being forced
-            if not force and not self._is_token_expired(self._credentials_cache.get(path, creds)):
-                return self._credentials_cache.get(path, creds)
-
-            lib_logger.debug(f"Refreshing Gemini OAuth token for '{Path(path).name}' (forced: {force})...")
-            refresh_token = creds.get("refresh_token")
-            if not refresh_token:
-                raise ValueError("No refresh_token found in credentials file.")
-
-            # [RETRY LOGIC] Implement exponential backoff for transient errors
-            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):
-                    try:
-                        response = await client.post(TOKEN_URI, data={
-                            "client_id": creds.get("client_id", CLIENT_ID),
-                            "client_secret": creds.get("client_secret", CLIENT_SECRET),
-                            "refresh_token": refresh_token,
-                            "grant_type": "refresh_token",
-                        }, timeout=30.0)
-                        response.raise_for_status()
-                        new_token_data = response.json()
-                        break  # Success, exit retry loop
-
-                    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..."
-                            )
-                            needs_reauth = True
-                            break  # Exit retry loop to trigger re-auth
-
-                        elif status_code == 429:
-                            # Rate limit - honor Retry-After header if present
-                            retry_after = int(e.response.headers.get("Retry-After", 60))
-                            lib_logger.warning(f"Rate limited (HTTP 429), retry after {retry_after}s")
-                            if attempt < max_retries - 1:
-                                await asyncio.sleep(retry_after)
-                                continue
-                            raise
-
-                        elif status_code >= 500 and status_code < 600:
-                            # Server error - retry with exponential backoff
-                            if attempt < max_retries - 1:
-                                wait_time = 2 ** attempt  # 1s, 2s, 4s
-                                lib_logger.warning(f"Server error (HTTP {status_code}), retry {attempt + 1}/{max_retries} in {wait_time}s")
-                                await asyncio.sleep(wait_time)
-                                continue
-                            raise  # Final attempt failed
-
-                        else:
-                            # Other errors - don't retry
-                            raise
-
-                    except (httpx.RequestError, httpx.TimeoutException) as e:
-                        # Network errors - retry with backoff
-                        last_error = e
-                        if attempt < max_retries - 1:
-                            wait_time = 2 ** attempt
-                            lib_logger.warning(f"Network error during refresh: {e}, retry {attempt + 1}/{max_retries} in {wait_time}s")
-                            await asyncio.sleep(wait_time)
-                            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")
-
-            # [FIX 1] Update OAuth token fields from response
-            creds["access_token"] = new_token_data["access_token"]
-            expiry_timestamp = time.time() + new_token_data["expires_in"]
-            creds["expiry_date"] = expiry_timestamp * 1000 # gemini-cli format
-
-            # [FIX 2] Update refresh_token if server provided a new one (rare but possible with Google OAuth)
-            if "refresh_token" in new_token_data:
-                creds["refresh_token"] = new_token_data["refresh_token"]
-
-            # [FIX 3] Ensure all required OAuth client fields are present (restore if missing)
-            if "client_id" not in creds or not creds["client_id"]:
-                creds["client_id"] = CLIENT_ID
-            if "client_secret" not in creds or not creds["client_secret"]:
-                creds["client_secret"] = CLIENT_SECRET
-            if "token_uri" not in creds or not creds["token_uri"]:
-                creds["token_uri"] = TOKEN_URI
-            if "universe_domain" not in creds or not creds["universe_domain"]:
-                creds["universe_domain"] = "googleapis.com"
-
-            # [FIX 4] Add scopes array if missing
-            if "scopes" not in creds:
-                creds["scopes"] = [
-                    "https://www.googleapis.com/auth/cloud-platform",
-                    "https://www.googleapis.com/auth/userinfo.email",
-                    "https://www.googleapis.com/auth/userinfo.profile",
-                ]
-
-            # [FIX 5] Ensure _proxy_metadata exists and update timestamp
-            if "_proxy_metadata" not in creds:
-                creds["_proxy_metadata"] = {}
-            creds["_proxy_metadata"]["last_check_timestamp"] = time.time()
-
-            # [VALIDATION] Verify refreshed credentials have all required fields
-            required_fields = ["access_token", "refresh_token", "client_id", "client_secret", "token_uri"]
-            missing_fields = [field for field in required_fields if not creds.get(field)]
-            if missing_fields:
-                raise ValueError(f"Refreshed credentials missing required fields: {missing_fields}")
-
-            # [VALIDATION] Optional: Test that the refreshed token is actually usable
-            try:
-                async with httpx.AsyncClient() as client:
-                    test_response = await client.get(
-                        USER_INFO_URI,
-                        headers={"Authorization": f"Bearer {creds['access_token']}"},
-                        timeout=5.0
-                    )
-                    test_response.raise_for_status()
-                    lib_logger.debug(f"Token validation successful for '{Path(path).name}'")
-            except Exception as e:
-                lib_logger.warning(f"Refreshed token validation failed for '{Path(path).name}': {e}")
-                # Don't fail the refresh - the token might still work for other endpoints
-                # But log it for debugging purposes
-
-            await self._save_credentials(path, creds)
-            lib_logger.debug(f"Successfully refreshed Gemini OAuth token for '{Path(path).name}'.") 
-            return creds
-
-    async def proactively_refresh(self, credential_path: str):
-        """Proactively refresh a credential by queueing it for refresh."""
-        creds = await self._load_credentials(credential_path)
-        if self._is_token_expired(creds):
-            # Queue for refresh with needs_reauth=False (automated refresh)
-            await self._queue_refresh(credential_path, force=False, needs_reauth=False)
-
-    async def _get_lock(self, path: str) -> asyncio.Lock:
-        # [FIX RACE CONDITION] Protect lock creation with a master lock
-        # This prevents TOCTOU bug where multiple coroutines check and create simultaneously
-        async with self._locks_lock:
-            if path not in self._refresh_locks:
-                self._refresh_locks[path] = asyncio.Lock()
-            return self._refresh_locks[path]
-
-    def is_credential_available(self, path: str) -> bool:
-        """Check if a credential is available for rotation (not queued/refreshing)."""
-        return path not in self._unavailable_credentials
-
-    async def _ensure_queue_processor_running(self):
-        """Lazily starts the queue processor if not already running."""
-        if self._queue_processor_task is None or self._queue_processor_task.done():
-            self._queue_processor_task = asyncio.create_task(self._process_refresh_queue())
-
-    async def _queue_refresh(self, path: str, force: bool = False, needs_reauth: bool = False):
-        """Add a credential to the refresh queue if not already queued.
-        
-        Args:
-            path: Credential file path
-            force: Force refresh even if not expired
-            needs_reauth: True if full re-authentication needed (bypasses backoff)
-        """
-        # IMPORTANT: Only check backoff for simple automated refreshes
-        # Re-authentication (interactive OAuth) should BYPASS backoff since it needs user input
-        if not needs_reauth:
-            now = time.time()
-            if path in self._next_refresh_after:
-                backoff_until = self._next_refresh_after[path]
-                if now < backoff_until:
-                    # Credential is in backoff for automated refresh, do not queue
-                    remaining = int(backoff_until - now)
-                    lib_logger.debug(f"Skipping automated refresh for '{Path(path).name}' (in backoff for {remaining}s)")
-                    return
-        
-        async with self._queue_tracking_lock:
-            if path not in self._queued_credentials:
-                self._queued_credentials.add(path)
-                self._unavailable_credentials.add(path)  # Mark as unavailable
-                await self._refresh_queue.put((path, force, needs_reauth))
-                await self._ensure_queue_processor_running()
-
-    async def _process_refresh_queue(self):
-        """Background worker that processes refresh requests sequentially."""
-        while True:
-            path = None
-            try:
-                # Wait for an item with timeout to allow graceful shutdown
-                try:
-                    path, force, needs_reauth = await asyncio.wait_for(
-                        self._refresh_queue.get(), 
-                        timeout=60.0
-                    )
-                except asyncio.TimeoutError:
-                    # No items for 60s, exit to save resources
-                    self._queue_processor_task = None
-                    return
-                
-                try:
-                    # Perform the actual refresh (still using per-credential lock)
-                    async with await self._get_lock(path):
-                        # Re-check if still expired (may have changed since queueing)
-                        creds = self._credentials_cache.get(path)
-                        if creds and not self._is_token_expired(creds):
-                            # No longer expired, mark as available
-                            async with self._queue_tracking_lock:
-                                self._unavailable_credentials.discard(path)
-                            continue
-                        
-                        # Perform refresh
-                        if not creds:
-                            creds = await self._load_credentials(path)
-                        await self._refresh_token(path, creds, force=force)
-                        
-                        # SUCCESS: Mark as available again
-                        async with self._queue_tracking_lock:
-                            self._unavailable_credentials.discard(path)
-                        
-                finally:
-                    # Remove from queued set
-                    async with self._queue_tracking_lock:
-                        self._queued_credentials.discard(path)
-                    self._refresh_queue.task_done()
-            except asyncio.CancelledError:
-                break
-            except Exception as e:
-                lib_logger.error(f"Error in queue processor: {e}")
-                # Even on error, mark as available (backoff will prevent immediate retry)
-                if path:
-                    async with self._queue_tracking_lock:
-                        self._unavailable_credentials.discard(path)
-
-    async def initialize_token(self, creds_or_path: Union[Dict[str, Any], str]) -> Dict[str, Any]:
-        path = creds_or_path if isinstance(creds_or_path, str) else None
-
-        # Get display name from metadata if available, otherwise derive from path
-        if isinstance(creds_or_path, dict):
-            display_name = creds_or_path.get("_proxy_metadata", {}).get("display_name", "in-memory object")
-        else:
-            display_name = Path(path).name if path else "in-memory object"
-
-        lib_logger.debug(f"Initializing Gemini token for '{display_name}'...")
-        try:
-            creds = await self._load_credentials(creds_or_path) if path else creds_or_path
-            reason = ""
-            if not creds.get("refresh_token"):
-                reason = "refresh token is missing"
-            elif self._is_token_expired(creds):
-                reason = "token is expired"
-
-            if reason:
-                if reason == "token is expired" and creds.get("refresh_token"):
-                    try:
-                        return await self._refresh_token(path, creds)
-                    except Exception as e:
-                        lib_logger.warning(f"Automatic token refresh for '{display_name}' failed: {e}. Proceeding to interactive login.")
-
-                lib_logger.warning(f"Gemini OAuth token for '{display_name}' needs setup: {reason}.")
-                
-                # [HEADLESS DETECTION] Check if running in headless environment
-                is_headless = is_headless_environment()
-                
-                auth_code_future = asyncio.get_event_loop().create_future()
-                server = None
-
-                async def handle_callback(reader, writer):
-                    try:
-                        request_line_bytes = await reader.readline()
-                        if not request_line_bytes: return
-                        path = request_line_bytes.decode('utf-8').strip().split(' ')[1]
-                        while await reader.readline() != b'\r\n': pass
-                        from urllib.parse import urlparse, parse_qs
-                        query_params = parse_qs(urlparse(path).query)
-                        writer.write(b"HTTP/1.1 200 OK\r\nContent-Type: text/html\r\n\r\n")
-                        if 'code' in query_params:
-                            if not auth_code_future.done():
-                                auth_code_future.set_result(query_params['code'][0])
-                            writer.write(b"<html><body><h1>Authentication successful!</h1><p>You can close this window.</p></body></html>")
-                        else:
-                            error = query_params.get('error', ['Unknown error'])[0]
-                            if not auth_code_future.done():
-                                auth_code_future.set_exception(Exception(f"OAuth failed: {error}"))
-                            writer.write(f"<html><body><h1>Authentication Failed</h1><p>Error: {error}. Please try again.</p></body></html>".encode())
-                        await writer.drain()
-                    except Exception as e:
-                        lib_logger.error(f"Error in OAuth callback handler: {e}")
-                    finally:
-                        writer.close()
-
-                try:
-                    server = await asyncio.start_server(handle_callback, '127.0.0.1', 8085)
-                    from urllib.parse import urlencode
-                    auth_url = "https://accounts.google.com/o/oauth2/v2/auth?" + urlencode({
-                        "client_id": CLIENT_ID,
-                        "redirect_uri": "http://localhost:8085/oauth2callback",
-                        "scope": " ".join(["https://www.googleapis.com/auth/cloud-platform", "https://www.googleapis.com/auth/userinfo.email", "https://www.googleapis.com/auth/userinfo.profile"]),
-                        "access_type": "offline", "response_type": "code", "prompt": "consent"
-                    })
-                    
-                    # [HEADLESS SUPPORT] Display appropriate instructions
-                    if is_headless:
-                        auth_panel_text = Text.from_markup(
-                            "Running in headless environment (no GUI detected).\n"
-                            "Please open the URL below in a browser on another machine to authorize:\n"
-                        )
-                    else:
-                        auth_panel_text = Text.from_markup(
-                            "1. Your browser will now open to log in and authorize the application.\n"
-                           "2. If it doesn't open automatically, please open the URL below manually."
-                        )
-                    
-                    console.print(Panel(auth_panel_text, title=f"Gemini OAuth Setup for [bold yellow]{display_name}[/bold yellow]", style="bold blue"))
-                    console.print(f"[bold]URL:[/bold] [link={auth_url}]{auth_url}[/link]\n")
-                    
-                    # [HEADLESS SUPPORT] Only attempt browser open if NOT headless
-                    if not is_headless:
-                        try:
-                            webbrowser.open(auth_url)
-                            lib_logger.info("Browser opened successfully for OAuth flow")
-                        except Exception as e:
-                            lib_logger.warning(f"Failed to open browser automatically: {e}. Please open the URL manually.")
-                    
-                    with console.status("[bold green]Waiting for you to complete authentication in the browser...[/bold green]", spinner="dots"):
-                        auth_code = await asyncio.wait_for(auth_code_future, timeout=300)
-                except asyncio.TimeoutError:
-                    raise Exception("OAuth flow timed out. Please try again.")
-                finally:
-                    if server:
-                        server.close()
-                        await server.wait_closed()
-                
-                lib_logger.info(f"Attempting to exchange authorization code for tokens...")
-                async with httpx.AsyncClient() as client:
-                    response = await client.post(TOKEN_URI, data={
-                        "code": auth_code.strip(), "client_id": CLIENT_ID, "client_secret": CLIENT_SECRET,
-                        "redirect_uri": "http://localhost:8085/oauth2callback", "grant_type": "authorization_code"
-                    })
-                    response.raise_for_status()
-                    token_data = response.json()
-                    # Start with the full token data from the exchange
-                    creds = token_data.copy()
-                    
-                    # Convert 'expires_in' to 'expiry_date' in milliseconds
-                    creds["expiry_date"] = (time.time() + creds.pop("expires_in")) * 1000
-                    
-                    # Ensure client_id and client_secret are present
-                    creds["client_id"] = CLIENT_ID
-                    creds["client_secret"] = CLIENT_SECRET
-
-                    creds["token_uri"] = TOKEN_URI
-                    creds["universe_domain"] = "googleapis.com"
-                    
-                    # Fetch user info and add metadata
-                    user_info_response = await client.get(USER_INFO_URI, headers={"Authorization": f"Bearer {creds['access_token']}"})
-                    user_info_response.raise_for_status()
-                    user_info = user_info_response.json()
-                    creds["_proxy_metadata"] = {
-                        "email": user_info.get("email"),
-                        "last_check_timestamp": time.time()
-                    }
-
-                    if path:
-                        await self._save_credentials(path, creds)
-                    lib_logger.info(f"Gemini OAuth initialized successfully for '{display_name}'.")
-                return creds
-
-            lib_logger.info(f"Gemini OAuth token at '{display_name}' is valid.")
-            return creds
-        except Exception as e:
-            raise ValueError(f"Failed to initialize Gemini OAuth for '{path}': {e}")
-
-    async def get_auth_header(self, credential_path: str) -> Dict[str, str]:
-        creds = await self._load_credentials(credential_path)
-        if self._is_token_expired(creds):
-            creds = await self._refresh_token(credential_path, creds)
-        return {"Authorization": f"Bearer {creds['access_token']}"}
-
-    async def get_user_info(self, creds_or_path: Union[Dict[str, Any], str]) -> Dict[str, Any]:
-        path = creds_or_path if isinstance(creds_or_path, str) else None
-        creds = await self._load_credentials(creds_or_path) if path else creds_or_path
-
-        if path and self._is_token_expired(creds):
-            creds = await self._refresh_token(path, creds)
-        
-        # Prefer locally stored metadata
-        if creds.get("_proxy_metadata", {}).get("email"):
-            if path:
-                creds["_proxy_metadata"]["last_check_timestamp"] = time.time()
-                await self._save_credentials(path, creds)
-            return {"email": creds["_proxy_metadata"]["email"]}
-
-        # Fallback to API call if metadata is missing
-        headers = {"Authorization": f"Bearer {creds['access_token']}"}
-        async with httpx.AsyncClient() as client:
-            response = await client.get(USER_INFO_URI, headers=headers)
-            response.raise_for_status()
-            user_info = response.json()
-            
-            # Save the retrieved info for future use
-            creds["_proxy_metadata"] = {
-                "email": user_info.get("email"),
-                "last_check_timestamp": time.time()
-            }
-            if path:
-                await self._save_credentials(path, creds)
-            return {"email": user_info.get("email")}
+from .google_oauth_base import GoogleOAuthBase
+
+class GeminiAuthBase(GoogleOAuthBase):
+    """
+    Gemini CLI OAuth2 authentication implementation.
+    
+    Inherits all OAuth functionality from GoogleOAuthBase with Gemini-specific configuration.
+    """
+    
+    CLIENT_ID = "681255809395-oo8ft2oprdrnp9e3aqf6av3hmdib135j.apps.googleusercontent.com"
+    CLIENT_SECRET = "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"
+    OAUTH_SCOPES = [
+        "https://www.googleapis.com/auth/cloud-platform",
+        "https://www.googleapis.com/auth/userinfo.email",
+        "https://www.googleapis.com/auth/userinfo.profile",
+    ]
+    ENV_PREFIX = "GEMINI_CLI"
+    CALLBACK_PORT = 8085
+    CALLBACK_PATH = "/oauth2callback"

src/rotator_library/providers/google_oauth_base.py

@@ -0,0 +1,653 @@
+# src/rotator_library/providers/google_oauth_base.py
+
+import os
+import webbrowser
+from typing import Union, Optional
+import json
+import time
+import asyncio
+import logging
+from pathlib import Path
+from typing import Dict, Any
+import tempfile
+import shutil
+
+import httpx
+from rich.console import Console
+from rich.panel import Panel
+from rich.text import Text
+
+from ..utils.headless_detection import is_headless_environment
+
+lib_logger = logging.getLogger('rotator_library')
+
+console = Console()
+
+class GoogleOAuthBase:
+    """
+    Base class for Google OAuth2 authentication providers.
+    
+    Subclasses must override:
+        - CLIENT_ID: OAuth client ID
+        - CLIENT_SECRET: OAuth client secret
+        - OAUTH_SCOPES: List of OAuth scopes
+        - ENV_PREFIX: Prefix for environment variables (e.g., "GEMINI_CLI", "ANTIGRAVITY")
+    
+    Subclasses may optionally override:
+        - CALLBACK_PORT: Local OAuth callback server port (default: 8085)
+        - CALLBACK_PATH: OAuth callback path (default: "/oauth2callback")
+        - REFRESH_EXPIRY_BUFFER_SECONDS: Time buffer before token expiry (default: 30 minutes)
+    """
+    
+    # Subclasses MUST override these
+    CLIENT_ID: str = None
+    CLIENT_SECRET: str = None
+    OAUTH_SCOPES: list = None
+    ENV_PREFIX: str = None
+    
+    # Subclasses MAY override these
+    TOKEN_URI: str = "https://oauth2.googleapis.com/token"
+    USER_INFO_URI: str = "https://www.googleapis.com/oauth2/v1/userinfo"
+    CALLBACK_PORT: int = 8085
+    CALLBACK_PATH: str = "/oauth2callback"
+    REFRESH_EXPIRY_BUFFER_SECONDS: int = 30 * 60  # 30 minutes
+
+    def __init__(self):
+        # Validate that subclass has set required attributes
+        if self.CLIENT_ID is None:
+            raise NotImplementedError(f"{self.__class__.__name__} must set CLIENT_ID")
+        if self.CLIENT_SECRET is None:
+            raise NotImplementedError(f"{self.__class__.__name__} must set CLIENT_SECRET")
+        if self.OAUTH_SCOPES is None:
+            raise NotImplementedError(f"{self.__class__.__name__} must set OAUTH_SCOPES")
+        if self.ENV_PREFIX is None:
+            raise NotImplementedError(f"{self.__class__.__name__} must set ENV_PREFIX")
+        
+        self._credentials_cache: Dict[str, Dict[str, Any]] = {}
+        self._refresh_locks: Dict[str, asyncio.Lock] = {}
+        self._locks_lock = asyncio.Lock()  # Protects the locks dict from race conditions
+        # [BACKOFF TRACKING] Track consecutive failures per credential
+        self._refresh_failures: Dict[str, int] = {}  # Track consecutive failures per credential
+        self._next_refresh_after: Dict[str, float] = {}  # Track backoff timers (Unix timestamp)
+        
+        # [QUEUE SYSTEM] Sequential refresh processing
+        self._refresh_queue: asyncio.Queue = asyncio.Queue()
+        self._queued_credentials: set = set()  # Track credentials already in queue
+        self._unavailable_credentials: set = set()  # Mark credentials unavailable during re-auth
+        self._queue_tracking_lock = asyncio.Lock()  # Protects queue sets
+        self._queue_processor_task: Optional[asyncio.Task] = None  # Background worker task
+
+    def _load_from_env(self) -> Optional[Dict[str, Any]]:
+        """
+        Load OAuth credentials from environment variables for stateless deployments.
+
+        Expected environment variables:
+        - {ENV_PREFIX}_ACCESS_TOKEN (required)
+        - {ENV_PREFIX}_REFRESH_TOKEN (required)
+        - {ENV_PREFIX}_EXPIRY_DATE (optional, defaults to 0)
+        - {ENV_PREFIX}_CLIENT_ID (optional, uses default)
+        - {ENV_PREFIX}_CLIENT_SECRET (optional, uses default)
+        - {ENV_PREFIX}_TOKEN_URI (optional, uses default)
+        - {ENV_PREFIX}_UNIVERSE_DOMAIN (optional, defaults to googleapis.com)
+        - {ENV_PREFIX}_EMAIL (optional, defaults to "env-user")
+        - {ENV_PREFIX}_PROJECT_ID (optional)
+        - {ENV_PREFIX}_TIER (optional)
+
+        Returns:
+            Dict with credential structure if env vars present, None otherwise
+        """
+        access_token = os.getenv(f"{self.ENV_PREFIX}_ACCESS_TOKEN")
+        refresh_token = os.getenv(f"{self.ENV_PREFIX}_REFRESH_TOKEN")
+
+        # Both access and refresh tokens are required
+        if not (access_token and refresh_token):
+            return None
+
+        lib_logger.debug(f"Loading {self.ENV_PREFIX} credentials from environment variables")
+
+        # Parse expiry_date as float, default to 0 if not present
+        expiry_str = os.getenv(f"{self.ENV_PREFIX}_EXPIRY_DATE", "0")
+        try:
+            expiry_date = float(expiry_str)
+        except ValueError:
+            lib_logger.warning(f"Invalid {self.ENV_PREFIX}_EXPIRY_DATE value: {expiry_str}, using 0")
+            expiry_date = 0
+
+        creds = {
+            "access_token": access_token,
+            "refresh_token": refresh_token,
+            "expiry_date": expiry_date,
+            "client_id": os.getenv(f"{self.ENV_PREFIX}_CLIENT_ID", self.CLIENT_ID),
+            "client_secret": os.getenv(f"{self.ENV_PREFIX}_CLIENT_SECRET", self.CLIENT_SECRET),
+            "token_uri": os.getenv(f"{self.ENV_PREFIX}_TOKEN_URI", self.TOKEN_URI),
+            "universe_domain": os.getenv(f"{self.ENV_PREFIX}_UNIVERSE_DOMAIN", "googleapis.com"),
+            "_proxy_metadata": {
+                "email": os.getenv(f"{self.ENV_PREFIX}_EMAIL", "env-user"),
+                "last_check_timestamp": time.time(),
+                "loaded_from_env": True  # Flag to indicate env-based credentials
+            }
+        }
+
+        # Add project_id if provided
+        project_id = os.getenv(f"{self.ENV_PREFIX}_PROJECT_ID")
+        if project_id:
+            creds["_proxy_metadata"]["project_id"] = project_id
+        
+        # Add tier if provided
+        tier = os.getenv(f"{self.ENV_PREFIX}_TIER")
+        if tier:
+            creds["_proxy_metadata"]["tier"] = tier
+
+        return creds
+
+    async def _load_credentials(self, path: str) -> Dict[str, Any]:
+        if path in self._credentials_cache:
+            return self._credentials_cache[path]
+
+        async with await self._get_lock(path):
+            if path in self._credentials_cache:
+                return self._credentials_cache[path]
+
+            # First, try loading from environment variables
+            env_creds = self._load_from_env()
+            if env_creds:
+                lib_logger.info(f"Using {self.ENV_PREFIX} credentials from environment variables")
+                # Cache env-based credentials using the path as key
+                self._credentials_cache[path] = env_creds
+                return env_creds
+
+            # Fall back to file-based loading
+            try:
+                lib_logger.debug(f"Loading {self.ENV_PREFIX} credentials from file: {path}")
+                with open(path, 'r') as f:
+                    creds = json.load(f)
+                # Handle gcloud-style creds file which nest tokens under "credential"
+                if "credential" in creds:
+                    creds = creds["credential"]
+                self._credentials_cache[path] = creds
+                return creds
+            except FileNotFoundError:
+                raise IOError(f"{self.ENV_PREFIX} OAuth credential file not found at '{path}'")
+            except Exception as e:
+                raise IOError(f"Failed to load {self.ENV_PREFIX} OAuth credentials from '{path}': {e}")
+
+    async def _save_credentials(self, path: str, creds: Dict[str, Any]):
+        # Don't save to file if credentials were loaded from environment
+        if creds.get("_proxy_metadata", {}).get("loaded_from_env"):
+            lib_logger.debug("Credentials loaded from env, skipping file save")
+            # Still update cache for in-memory consistency
+            self._credentials_cache[path] = creds
+            return
+
+        # [ATOMIC WRITE] Use tempfile + move pattern to ensure atomic writes
+        # This prevents credential corruption if the process is interrupted during write
+        parent_dir = os.path.dirname(os.path.abspath(path))
+        os.makedirs(parent_dir, exist_ok=True)
+
+        tmp_fd = None
+        tmp_path = None
+        try:
+            # Create temp file in same directory as target (ensures same filesystem)
+            tmp_fd, tmp_path = tempfile.mkstemp(dir=parent_dir, prefix='.tmp_', suffix='.json', text=True)
+
+            # Write JSON to temp file
+            with os.fdopen(tmp_fd, 'w') as f:
+                json.dump(creds, f, indent=2)
+                tmp_fd = None  # fdopen closes the fd
+
+            # Set secure permissions (0600 = owner read/write only)
+            try:
+                os.chmod(tmp_path, 0o600)
+            except (OSError, AttributeError):
+                # Windows may not support chmod, ignore
+                pass
+
+            # Atomic move (overwrites target if it exists)
+            shutil.move(tmp_path, path)
+            tmp_path = None  # Successfully moved
+
+            # Update cache AFTER successful file write (prevents cache/file inconsistency)
+            self._credentials_cache[path] = creds
+            lib_logger.debug(f"Saved updated {self.ENV_PREFIX} OAuth credentials to '{path}' (atomic write).")
+
+        except Exception as e:
+            lib_logger.error(f"Failed to save updated {self.ENV_PREFIX} OAuth credentials to '{path}': {e}")
+            # Clean up temp file if it still exists
+            if tmp_fd is not None:
+                try:
+                    os.close(tmp_fd)
+                except:
+                    pass
+            if tmp_path and os.path.exists(tmp_path):
+                try:
+                    os.unlink(tmp_path)
+                except:
+                    pass
+            raise
+
+    def _is_token_expired(self, creds: Dict[str, Any]) -> bool:
+        expiry = creds.get("token_expiry") # gcloud format
+        if not expiry: # gemini-cli format
+             expiry_timestamp = creds.get("expiry_date", 0) / 1000
+        else:
+            expiry_timestamp = time.mktime(time.strptime(expiry, "%Y-%m-%dT%H:%M:%SZ"))
+        return expiry_timestamp < time.time() + self.REFRESH_EXPIRY_BUFFER_SECONDS
+
+    async def _refresh_token(self, path: str, creds: Dict[str, Any], force: bool = False) -> Dict[str, Any]:
+        async with await self._get_lock(path):
+            # Skip the expiry check if a refresh is being forced
+            if not force and not self._is_token_expired(self._credentials_cache.get(path, creds)):
+                return self._credentials_cache.get(path, creds)
+
+            lib_logger.debug(f"Refreshing {self.ENV_PREFIX} OAuth token for '{Path(path).name}' (forced: {force})...")
+            refresh_token = creds.get("refresh_token")
+            if not refresh_token:
+                raise ValueError("No refresh_token found in credentials file.")
+
+            # [RETRY LOGIC] Implement exponential backoff for transient errors
+            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):
+                    try:
+                        response = await client.post(self.TOKEN_URI, data={
+                            "client_id": creds.get("client_id", self.CLIENT_ID),
+                            "client_secret": creds.get("client_secret", self.CLIENT_SECRET),
+                            "refresh_token": refresh_token,
+                            "grant_type": "refresh_token",
+                        }, timeout=30.0)
+                        response.raise_for_status()
+                        new_token_data = response.json()
+                        break  # Success, exit retry loop
+
+                    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..."
+                            )
+                            needs_reauth = True
+                            break  # Exit retry loop to trigger re-auth
+
+                        elif status_code == 429:
+                            # Rate limit - honor Retry-After header if present
+                            retry_after = int(e.response.headers.get("Retry-After", 60))
+                            lib_logger.warning(f"Rate limited (HTTP 429), retry after {retry_after}s")
+                            if attempt < max_retries - 1:
+                                await asyncio.sleep(retry_after)
+                                continue
+                            raise
+
+                        elif status_code >= 500 and status_code < 600:
+                            # Server error - retry with exponential backoff
+                            if attempt < max_retries - 1:
+                                wait_time = 2 ** attempt  # 1s, 2s, 4s
+                                lib_logger.warning(f"Server error (HTTP {status_code}), retry {attempt + 1}/{max_retries} in {wait_time}s")
+                                await asyncio.sleep(wait_time)
+                                continue
+                            raise  # Final attempt failed
+
+                        else:
+                            # Other errors - don't retry
+                            raise
+
+                    except (httpx.RequestError, httpx.TimeoutException) as e:
+                        # Network errors - retry with backoff
+                        last_error = e
+                        if attempt < max_retries - 1:
+                            wait_time = 2 ** attempt
+                            lib_logger.warning(f"Network error during refresh: {e}, retry {attempt + 1}/{max_retries} in {wait_time}s")
+                            await asyncio.sleep(wait_time)
+                            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")
+
+            # [FIX 1] Update OAuth token fields from response
+            creds["access_token"] = new_token_data["access_token"]
+            expiry_timestamp = time.time() + new_token_data["expires_in"]
+            creds["expiry_date"] = expiry_timestamp * 1000 # gemini-cli format
+
+            # [FIX 2] Update refresh_token if server provided a new one (rare but possible with Google OAuth)
+            if "refresh_token" in new_token_data:
+                creds["refresh_token"] = new_token_data["refresh_token"]
+
+            # [FIX 3] Ensure all required OAuth client fields are present (restore if missing)
+            if "client_id" not in creds or not creds["client_id"]:
+                creds["client_id"] = self.CLIENT_ID
+            if "client_secret" not in creds or not creds["client_secret"]:
+                creds["client_secret"] = self.CLIENT_SECRET
+            if "token_uri" not in creds or not creds["token_uri"]:
+                creds["token_uri"] = self.TOKEN_URI
+            if "universe_domain" not in creds or not creds["universe_domain"]:
+                creds["universe_domain"] = "googleapis.com"
+
+            # [FIX 4] Add scopes array if missing
+            if "scopes" not in creds:
+                creds["scopes"] = self.OAUTH_SCOPES
+
+            # [FIX 5] Ensure _proxy_metadata exists and update timestamp
+            if "_proxy_metadata" not in creds:
+                creds["_proxy_metadata"] = {}
+            creds["_proxy_metadata"]["last_check_timestamp"] = time.time()
+
+            # [VALIDATION] Verify refreshed credentials have all required fields
+            required_fields = ["access_token", "refresh_token", "client_id", "client_secret", "token_uri"]
+            missing_fields = [field for field in required_fields if not creds.get(field)]
+            if missing_fields:
+                raise ValueError(f"Refreshed credentials missing required fields: {missing_fields}")
+
+            # [VALIDATION] Optional: Test that the refreshed token is actually usable
+            try:
+                async with httpx.AsyncClient() as client:
+                    test_response = await client.get(
+                        self.USER_INFO_URI,
+                        headers={"Authorization": f"Bearer {creds['access_token']}"},
+                        timeout=5.0
+                    )
+                    test_response.raise_for_status()
+                    lib_logger.debug(f"Token validation successful for '{Path(path).name}'")
+            except Exception as e:
+                lib_logger.warning(f"Refreshed token validation failed for '{Path(path).name}': {e}")
+                # Don't fail the refresh - the token might still work for other endpoints
+                # But log it for debugging purposes
+
+            await self._save_credentials(path, creds)
+            lib_logger.debug(f"Successfully refreshed {self.ENV_PREFIX} OAuth token for '{Path(path).name}'.") 
+            return creds
+
+    async def proactively_refresh(self, credential_path: str):
+        """Proactively refresh a credential by queueing it for refresh."""
+        creds = await self._load_credentials(credential_path)
+        if self._is_token_expired(creds):
+            # Queue for refresh with needs_reauth=False (automated refresh)
+            await self._queue_refresh(credential_path, force=False, needs_reauth=False)
+
+    async def _get_lock(self, path: str) -> asyncio.Lock:
+        # [FIX RACE CONDITION] Protect lock creation with a master lock
+        # This prevents TOCTOU bug where multiple coroutines check and create simultaneously
+        async with self._locks_lock:
+            if path not in self._refresh_locks:
+                self._refresh_locks[path] = asyncio.Lock()
+            return self._refresh_locks[path]
+
+    def is_credential_available(self, path: str) -> bool:
+        """Check if a credential is available for rotation (not queued/refreshing)."""
+        return path not in self._unavailable_credentials
+
+    async def _ensure_queue_processor_running(self):
+        """Lazily starts the queue processor if not already running."""
+        if self._queue_processor_task is None or self._queue_processor_task.done():
+            self._queue_processor_task = asyncio.create_task(self._process_refresh_queue())
+
+    async def _queue_refresh(self, path: str, force: bool = False, needs_reauth: bool = False):
+        """Add a credential to the refresh queue if not already queued.
+        
+        Args:
+            path: Credential file path
+            force: Force refresh even if not expired
+            needs_reauth: True if full re-authentication needed (bypasses backoff)
+        """
+        # IMPORTANT: Only check backoff for simple automated refreshes
+        # Re-authentication (interactive OAuth) should BYPASS backoff since it needs user input
+        if not needs_reauth:
+            now = time.time()
+            if path in self._next_refresh_after:
+                backoff_until = self._next_refresh_after[path]
+                if now < backoff_until:
+                    # Credential is in backoff for automated refresh, do not queue
+                    remaining = int(backoff_until - now)
+                    lib_logger.debug(f"Skipping automated refresh for '{Path(path).name}' (in backoff for {remaining}s)")
+                    return
+        
+        async with self._queue_tracking_lock:
+            if path not in self._queued_credentials:
+                self._queued_credentials.add(path)
+                self._unavailable_credentials.add(path)  # Mark as unavailable
+                await self._refresh_queue.put((path, force, needs_reauth))
+                await self._ensure_queue_processor_running()
+
+    async def _process_refresh_queue(self):
+        """Background worker that processes refresh requests sequentially."""
+        while True:
+            path = None
+            try:
+                # Wait for an item with timeout to allow graceful shutdown
+                try:
+                    path, force, needs_reauth = await asyncio.wait_for(
+                        self._refresh_queue.get(), 
+                        timeout=60.0
+                    )
+                except asyncio.TimeoutError:
+                    # No items for 60s, exit to save resources
+                    self._queue_processor_task = None
+                    return
+                
+                try:
+                    # Perform the actual refresh (still using per-credential lock)
+                    async with await self._get_lock(path):
+                        # Re-check if still expired (may have changed since queueing)
+                        creds = self._credentials_cache.get(path)
+                        if creds and not self._is_token_expired(creds):
+                            # No longer expired, mark as available
+                            async with self._queue_tracking_lock:
+                                self._unavailable_credentials.discard(path)
+                            continue
+                        
+                        # Perform refresh
+                        if not creds:
+                            creds = await self._load_credentials(path)
+                        await self._refresh_token(path, creds, force=force)
+                        
+                        # SUCCESS: Mark as available again
+                        async with self._queue_tracking_lock:
+                            self._unavailable_credentials.discard(path)
+                        
+                finally:
+                    # Remove from queued set
+                    async with self._queue_tracking_lock:
+                        self._queued_credentials.discard(path)
+                    self._refresh_queue.task_done()
+            except asyncio.CancelledError:
+                break
+            except Exception as e:
+                lib_logger.error(f"Error in queue processor: {e}")
+                # Even on error, mark as available (backoff will prevent immediate retry)
+                if path:
+                    async with self._queue_tracking_lock:
+                        self._unavailable_credentials.discard(path)
+
+    async def initialize_token(self, creds_or_path: Union[Dict[str, Any], str]) -> Dict[str, Any]:
+        path = creds_or_path if isinstance(creds_or_path, str) else None
+
+        # Get display name from metadata if available, otherwise derive from path
+        if isinstance(creds_or_path, dict):
+            display_name = creds_or_path.get("_proxy_metadata", {}).get("display_name", "in-memory object")
+        else:
+            display_name = Path(path).name if path else "in-memory object"
+
+        lib_logger.debug(f"Initializing {self.ENV_PREFIX} token for '{display_name}'...")
+        try:
+            creds = await self._load_credentials(creds_or_path) if path else creds_or_path
+            reason = ""
+            if not creds.get("refresh_token"):
+                reason = "refresh token is missing"
+            elif self._is_token_expired(creds):
+                reason = "token is expired"
+
+            if reason:
+                if reason == "token is expired" and creds.get("refresh_token"):
+                    try:
+                        return await self._refresh_token(path, creds)
+                    except Exception as e:
+                        lib_logger.warning(f"Automatic token refresh for '{display_name}' failed: {e}. Proceeding to interactive login.")
+
+                lib_logger.warning(f"{self.ENV_PREFIX} OAuth token for '{display_name}' needs setup: {reason}.")
+                
+                # [HEADLESS DETECTION] Check if running in headless environment
+                is_headless = is_headless_environment()
+                
+                auth_code_future = asyncio.get_event_loop().create_future()
+                server = None
+
+                async def handle_callback(reader, writer):
+                    try:
+                        request_line_bytes = await reader.readline()
+                        if not request_line_bytes: return
+                        path_str = request_line_bytes.decode('utf-8').strip().split(' ')[1]
+                        while await reader.readline() != b'\r\n': pass
+                        from urllib.parse import urlparse, parse_qs
+                        query_params = parse_qs(urlparse(path_str).query)
+                        writer.write(b"HTTP/1.1 200 OK\r\nContent-Type: text/html\r\n\r\n")
+                        if 'code' in query_params:
+                            if not auth_code_future.done():
+                                auth_code_future.set_result(query_params['code'][0])
+                            writer.write(b"<html><body><h1>Authentication successful!</h1><p>You can close this window.</p></body></html>")
+                        else:
+                            error = query_params.get('error', ['Unknown error'])[0]
+                            if not auth_code_future.done():
+                                auth_code_future.set_exception(Exception(f"OAuth failed: {error}"))
+                            writer.write(f"<html><body><h1>Authentication Failed</h1><p>Error: {error}. Please try again.</p></body></html>".encode())
+                        await writer.drain()
+                    except Exception as e:
+                        lib_logger.error(f"Error in OAuth callback handler: {e}")
+                    finally:
+                        writer.close()
+
+                try:
+                    server = await asyncio.start_server(handle_callback, '127.0.0.1', self.CALLBACK_PORT)
+                    from urllib.parse import urlencode
+                    auth_url = "https://accounts.google.com/o/oauth2/v2/auth?" + urlencode({
+                        "client_id": self.CLIENT_ID,
+                        "redirect_uri": f"http://localhost:{self.CALLBACK_PORT}{self.CALLBACK_PATH}",
+                        "scope": " ".join(self.OAUTH_SCOPES),
+                        "access_type": "offline", "response_type": "code", "prompt": "consent"
+                    })
+                    
+                    # [HEADLESS SUPPORT] Display appropriate instructions
+                    if is_headless:
+                        auth_panel_text = Text.from_markup(
+                            "Running in headless environment (no GUI detected).\n"
+                            "Please open the URL below in a browser on another machine to authorize:\n"
+                        )
+                    else:
+                        auth_panel_text = Text.from_markup(
+                            "1. Your browser will now open to log in and authorize the application.\n"
+                           "2. If it doesn't open automatically, please open the URL below manually."
+                        )
+                    
+                    console.print(Panel(auth_panel_text, title=f"{self.ENV_PREFIX} OAuth Setup for [bold yellow]{display_name}[/bold yellow]", style="bold blue"))
+                    console.print(f"[bold]URL:[/bold] [link={auth_url}]{auth_url}[/link]\n")
+                    
+                    # [HEADLESS SUPPORT] Only attempt browser open if NOT headless
+                    if not is_headless:
+                        try:
+                            webbrowser.open(auth_url)
+                            lib_logger.info("Browser opened successfully for OAuth flow")
+                        except Exception as e:
+                            lib_logger.warning(f"Failed to open browser automatically: {e}. Please open the URL manually.")
+                    
+                    with console.status(f"[bold green]Waiting for you to complete authentication in the browser...[/bold green]", spinner="dots"):
+                        auth_code = await asyncio.wait_for(auth_code_future, timeout=300)
+                except asyncio.TimeoutError:
+                    raise Exception("OAuth flow timed out. Please try again.")
+                finally:
+                    if server:
+                        server.close()
+                        await server.wait_closed()
+                
+                lib_logger.info(f"Attempting to exchange authorization code for tokens...")
+                async with httpx.AsyncClient() as client:
+                    response = await client.post(self.TOKEN_URI, data={
+                        "code": auth_code.strip(), "client_id": self.CLIENT_ID, "client_secret": self.CLIENT_SECRET,
+                        "redirect_uri": f"http://localhost:{self.CALLBACK_PORT}{self.CALLBACK_PATH}", "grant_type": "authorization_code"
+                    })
+                    response.raise_for_status()
+                    token_data = response.json()
+                    # Start with the full token data from the exchange
+                    creds = token_data.copy()
+                    
+                    # Convert 'expires_in' to 'expiry_date' in milliseconds
+                    creds["expiry_date"] = (time.time() + creds.pop("expires_in")) * 1000
+                    
+                    # Ensure client_id and client_secret are present
+                    creds["client_id"] = self.CLIENT_ID
+                    creds["client_secret"] = self.CLIENT_SECRET
+
+                    creds["token_uri"] = self.TOKEN_URI
+                    creds["universe_domain"] = "googleapis.com"
+                    
+                    # Fetch user info and add metadata
+                    user_info_response = await client.get(self.USER_INFO_URI, headers={"Authorization": f"Bearer {creds['access_token']}"})
+                    user_info_response.raise_for_status()
+                    user_info = user_info_response.json()
+                    creds["_proxy_metadata"] = {
+                        "email": user_info.get("email"),
+                        "last_check_timestamp": time.time()
+                    }
+
+                    if path:
+                        await self._save_credentials(path, creds)
+                    lib_logger.info(f"{self.ENV_PREFIX} OAuth initialized successfully for '{display_name}'.")
+                return creds
+
+            lib_logger.info(f"{self.ENV_PREFIX} OAuth token at '{display_name}' is valid.")
+            return creds
+        except Exception as e:
+            raise ValueError(f"Failed to initialize {self.ENV_PREFIX} OAuth for '{path}': {e}")
+
+    async def get_auth_header(self, credential_path: str) -> Dict[str, str]:
+        creds = await self._load_credentials(credential_path)
+        if self._is_token_expired(creds):
+            creds = await self._refresh_token(credential_path, creds)
+        return {"Authorization": f"Bearer {creds['access_token']}"}
+
+    async def get_user_info(self, creds_or_path: Union[Dict[str, Any], str]) -> Dict[str, Any]:
+        path = creds_or_path if isinstance(creds_or_path, str) else None
+        creds = await self._load_credentials(creds_or_path) if path else creds_or_path
+
+        if path and self._is_token_expired(creds):
+            creds = await self._refresh_token(path, creds)
+        
+        # Prefer locally stored metadata
+        if creds.get("_proxy_metadata", {}).get("email"):
+            if path:
+                creds["_proxy_metadata"]["last_check_timestamp"] = time.time()
+                await self._save_credentials(path, creds)
+            return {"email": creds["_proxy_metadata"]["email"]}
+
+        # Fallback to API call if metadata is missing
+        headers = {"Authorization": f"Bearer {creds['access_token']}"}
+        async with httpx.AsyncClient() as client:
+            response = await client.get(self.USER_INFO_URI, headers=headers)
+            response.raise_for_status()
+            user_info = response.json()
+            
+            # Save the retrieved info for future use
+            creds["_proxy_metadata"] = {
+                "email": user_info.get("email"),
+                "last_check_timestamp": time.time()
+            }
+            if path:
+                await self._save_credentials(path, creds)
+            return {"email": user_info.get("email")}