fix(oauth): 🐛 escape rich markup in oauth authorization urls

Prevent Rich markup interpretation issues when displaying OAuth authorization URLs in terminal output.

- Import `rich.markup.escape` to properly escape special characters (=, &, etc.) in URLs
- Add extensive inline documentation explaining the escaping rationale and known terminal compatibility issues
- Apply URL escaping to authorization URLs in Google OAuth, iFlow, and Qwen Code providers
- Refine headless environment detection to exclude macOS from DISPLAY checks (macOS uses Quartz, not X11)
- Improve code formatting consistency (string quotes, line wrapping) across OAuth providers

The escaped URLs display correctly in all terminal configurations while remaining clickable in supported terminals (iTerm2, Windows Terminal, etc.).

src/rotator_library/providers/google_oauth_base.py

@@ -16,35 +16,37 @@ import httpx
 from rich.console import Console
 from rich.panel import Panel
 from rich.text import Text
+from rich.markup import escape as rich_escape
 
 from ..utils.headless_detection import is_headless_environment
 
-lib_logger = logging.getLogger('rotator_library')
+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"
@@ -57,49 +59,65 @@ class GoogleOAuthBase:
         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")
+            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")
+            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
+        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)
-        
+        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._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
+        self._queue_processor_task: Optional[asyncio.Task] = (
+            None  # Background worker task
+        )
 
     def _parse_env_credential_path(self, path: str) -> Optional[str]:
         """
         Parse a virtual env:// path and return the credential index.
-        
+
         Supported formats:
         - "env://provider/0" - Legacy single credential (no index in env var names)
         - "env://provider/1" - First numbered credential (PROVIDER_1_ACCESS_TOKEN)
         - "env://provider/2" - Second numbered credential, etc.
-        
+
         Returns:
             The credential index as string ("0" for legacy, "1", "2", etc. for numbered)
             or None if path is not an env:// path
         """
         if not path.startswith("env://"):
             return None
-        
+
         # Parse: env://provider/index
         parts = path[6:].split("/")  # Remove "env://" prefix
         if len(parts) >= 2:
             return parts[1]  # Return the index
         return "0"  # Default to legacy format
 
-    def _load_from_env(self, credential_index: Optional[str] = None) -> Optional[Dict[str, Any]]:
+    def _load_from_env(
+        self, credential_index: Optional[str] = None
+    ) -> Optional[Dict[str, Any]]:
         """
         Load OAuth credentials from environment variables for stateless deployments.
 
@@ -133,7 +151,7 @@ class GoogleOAuthBase:
             # Legacy format: PROVIDER_ACCESS_TOKEN
             prefix = self.ENV_PREFIX
             default_email = "env-user"
-        
+
         access_token = os.getenv(f"{prefix}_ACCESS_TOKEN")
         refresh_token = os.getenv(f"{prefix}_REFRESH_TOKEN")
 
@@ -148,7 +166,9 @@ class GoogleOAuthBase:
         try:
             expiry_date = float(expiry_str)
         except ValueError:
-            lib_logger.warning(f"Invalid {prefix}_EXPIRY_DATE value: {expiry_str}, using 0")
+            lib_logger.warning(
+                f"Invalid {prefix}_EXPIRY_DATE value: {expiry_str}, using 0"
+            )
             expiry_date = 0
 
         creds = {
@@ -163,15 +183,16 @@ class GoogleOAuthBase:
                 "email": os.getenv(f"{prefix}_EMAIL", default_email),
                 "last_check_timestamp": time.time(),
                 "loaded_from_env": True,  # Flag to indicate env-based credentials
-                "env_credential_index": credential_index or "0"  # Track which env credential this is
-            }
+                "env_credential_index": credential_index
+                or "0",  # Track which env credential this is
+            },
         }
 
         # Add project_id if provided
         project_id = os.getenv(f"{prefix}_PROJECT_ID")
         if project_id:
             creds["_proxy_metadata"]["project_id"] = project_id
-        
+
         # Add tier if provided
         tier = os.getenv(f"{prefix}_TIER")
         if tier:
@@ -193,24 +214,32 @@ class GoogleOAuthBase:
                 # Load from environment variables with specific index
                 env_creds = self._load_from_env(credential_index)
                 if env_creds:
-                    lib_logger.info(f"Using {self.ENV_PREFIX} credentials from environment variables (index: {credential_index})")
+                    lib_logger.info(
+                        f"Using {self.ENV_PREFIX} credentials from environment variables (index: {credential_index})"
+                    )
                     self._credentials_cache[path] = env_creds
                     return env_creds
                 else:
-                    raise IOError(f"Environment variables for {self.ENV_PREFIX} credential index {credential_index} not found")
+                    raise IOError(
+                        f"Environment variables for {self.ENV_PREFIX} credential index {credential_index} not found"
+                    )
 
             # For file paths, first try loading from legacy env vars (for backwards compatibility)
             env_creds = self._load_from_env()
             if env_creds:
-                lib_logger.info(f"Using {self.ENV_PREFIX} credentials from environment variables")
+                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:
+                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:
@@ -218,11 +247,17 @@ class GoogleOAuthBase:
                 self._credentials_cache[path] = creds
                 return creds
             except FileNotFoundError:
-                raise IOError(f"{self.ENV_PREFIX} OAuth credential file not found at '{path}'")
+                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}")
+                raise IOError(
+                    f"Failed to load {self.ENV_PREFIX} OAuth credentials from '{path}': {e}"
+                )
             except Exception as e:
-                raise IOError(f"Failed to load {self.ENV_PREFIX} OAuth credentials from '{path}': {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
@@ -241,10 +276,12 @@ class GoogleOAuthBase:
         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)
+            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:
+            with os.fdopen(tmp_fd, "w") as f:
                 json.dump(creds, f, indent=2)
                 tmp_fd = None  # fdopen closes the fd
 
@@ -261,10 +298,14 @@ class GoogleOAuthBase:
 
             # 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).")
+            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}")
+            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:
@@ -279,20 +320,26 @@ class GoogleOAuthBase:
             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
+        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 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)):
+            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})...")
+            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.")
@@ -306,12 +353,18 @@ class GoogleOAuthBase:
             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 = 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
@@ -332,7 +385,9 @@ class GoogleOAuthBase:
                         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")
+                            lib_logger.warning(
+                                f"Rate limited (HTTP 429), retry after {retry_after}s"
+                            )
                             if attempt < max_retries - 1:
                                 await asyncio.sleep(retry_after)
                                 continue
@@ -341,8 +396,10 @@ class GoogleOAuthBase:
                         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")
+                                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
@@ -355,22 +412,30 @@ class GoogleOAuthBase:
                         # 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")
+                            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}'...")
+                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}")
+                    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:
@@ -379,7 +444,7 @@ class GoogleOAuthBase:
             # [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
+            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:
@@ -405,10 +470,20 @@ class GoogleOAuthBase:
             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)]
+            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}")
+                raise ValueError(
+                    f"Refreshed credentials missing required fields: {missing_fields}"
+                )
 
             # [VALIDATION] Optional: Test that the refreshed token is actually usable
             try:
@@ -416,17 +491,23 @@ class GoogleOAuthBase:
                     test_response = await client.get(
                         self.USER_INFO_URI,
                         headers={"Authorization": f"Bearer {creds['access_token']}"},
-                        timeout=5.0
+                        timeout=5.0,
                     )
                     test_response.raise_for_status()
-                    lib_logger.debug(f"Token validation successful for '{Path(path).name}'")
+                    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}")
+                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}'.") 
+            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):
@@ -451,11 +532,15 @@ class GoogleOAuthBase:
     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())
+            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):
+    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
@@ -470,9 +555,11 @@ class GoogleOAuthBase:
                 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)")
+                    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)
@@ -488,14 +575,13 @@ class GoogleOAuthBase:
                 # 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
+                        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):
@@ -506,16 +592,16 @@ class GoogleOAuthBase:
                             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:
@@ -530,18 +616,26 @@ class GoogleOAuthBase:
                     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]:
+    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")
+            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}'...")
+        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
+            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"
@@ -553,34 +647,51 @@ class GoogleOAuthBase:
                     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"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}."
+                )
 
-                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
+                        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:
+                        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>")
+                                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]
+                            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())
+                                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}")
@@ -588,15 +699,25 @@ class GoogleOAuthBase:
                         writer.close()
 
                 try:
-                    server = await asyncio.start_server(handle_callback, '127.0.0.1', self.CALLBACK_PORT)
+                    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"
-                    })
-                    
+
+                    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(
@@ -606,68 +727,118 @@ class GoogleOAuthBase:
                     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."
+                            "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(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")
-                    
+                    )
+                    # [URL DISPLAY] Print URL with proper escaping to prevent Rich markup issues.
+                    # IMPORTANT: OAuth URLs contain special characters (=, &, etc.) that Rich might
+                    # interpret as markup in some terminal configurations. We escape the URL to
+                    # ensure it displays correctly.
+                    #
+                    # KNOWN ISSUE: If Rich rendering fails entirely (e.g., terminal doesn't support
+                    # ANSI codes, or output is piped), the escaped URL should still be valid.
+                    # However, if the terminal strips or mangles the output, users should copy
+                    # the URL directly from logs or use --verbose to see the raw URL.
+                    #
+                    # The [link=...] markup creates a clickable hyperlink in supported terminals
+                    # (iTerm2, Windows Terminal, etc.), but the displayed text is the escaped URL
+                    # which can be safely copied even if the hyperlink doesn't work.
+                    escaped_url = rich_escape(auth_url)
+                    console.print(
+                        f"[bold]URL:[/bold] [link={auth_url}]{escaped_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")
+                            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)
+                            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...")
+
+                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 = 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
-                    
+                    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 = 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()
+                        "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}'.")
+                    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.")
+            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}")
+            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)
@@ -675,13 +846,15 @@ class GoogleOAuthBase:
             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]:
+    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:
@@ -695,11 +868,11 @@ class GoogleOAuthBase:
             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()
+                "last_check_timestamp": time.time(),
             }
             if path:
                 await self._save_credentials(path, creds)

src/rotator_library/providers/iflow_auth_base.py

@@ -21,9 +21,10 @@ from rich.console import Console
 from rich.panel import Panel
 from rich.prompt import Prompt
 from rich.text import Text
+from rich.markup import escape as rich_escape
 from ..utils.headless_detection import is_headless_environment
 
-lib_logger = logging.getLogger('rotator_library')
+lib_logger = logging.getLogger("rotator_library")
 
 IFLOW_OAUTH_AUTHORIZE_ENDPOINT = "https://iflow.cn/oauth"
 IFLOW_OAUTH_TOKEN_ENDPOINT = "https://iflow.cn/oauth/token"
@@ -61,7 +62,7 @@ class OAuthCallbackServer:
         """Checks if the callback port is available."""
         try:
             sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-            sock.bind(('', self.port))
+            sock.bind(("", self.port))
             sock.close()
             return True
         except OSError:
@@ -76,12 +77,12 @@ class OAuthCallbackServer:
         self.result_future = asyncio.Future()
 
         # Setup route
-        self.app.router.add_get('/oauth2callback', self._handle_callback)
+        self.app.router.add_get("/oauth2callback", self._handle_callback)
 
         # Start server
         self.runner = web.AppRunner(self.app)
         await self.runner.setup()
-        self.site = web.TCPSite(self.runner, 'localhost', self.port)
+        self.site = web.TCPSite(self.runner, "localhost", self.port)
         await self.site.start()
 
         lib_logger.debug(f"iFlow OAuth callback server started on port {self.port}")
@@ -99,34 +100,46 @@ class OAuthCallbackServer:
         query = request.query
 
         # Check for error parameter
-        if 'error' in query:
-            error = query.get('error', 'unknown_error')
+        if "error" in query:
+            error = query.get("error", "unknown_error")
             lib_logger.error(f"iFlow OAuth callback received error: {error}")
             if not self.result_future.done():
                 self.result_future.set_exception(ValueError(f"OAuth error: {error}"))
-            return web.Response(status=302, headers={'Location': IFLOW_ERROR_REDIRECT_URL})
+            return web.Response(
+                status=302, headers={"Location": IFLOW_ERROR_REDIRECT_URL}
+            )
 
         # Check for authorization code
-        code = query.get('code')
+        code = query.get("code")
         if not code:
             lib_logger.error("iFlow OAuth callback missing authorization code")
             if not self.result_future.done():
-                self.result_future.set_exception(ValueError("Missing authorization code"))
-            return web.Response(status=302, headers={'Location': IFLOW_ERROR_REDIRECT_URL})
+                self.result_future.set_exception(
+                    ValueError("Missing authorization code")
+                )
+            return web.Response(
+                status=302, headers={"Location": IFLOW_ERROR_REDIRECT_URL}
+            )
 
         # Validate state parameter
-        state = query.get('state', '')
+        state = query.get("state", "")
         if state != self.expected_state:
-            lib_logger.error(f"iFlow OAuth state mismatch. Expected: {self.expected_state}, Got: {state}")
+            lib_logger.error(
+                f"iFlow OAuth state mismatch. Expected: {self.expected_state}, Got: {state}"
+            )
             if not self.result_future.done():
                 self.result_future.set_exception(ValueError("State parameter mismatch"))
-            return web.Response(status=302, headers={'Location': IFLOW_ERROR_REDIRECT_URL})
+            return web.Response(
+                status=302, headers={"Location": IFLOW_ERROR_REDIRECT_URL}
+            )
 
         # Success - set result and redirect to success page
         if not self.result_future.done():
             self.result_future.set_result(code)
 
-        return web.Response(status=302, headers={'Location': IFLOW_SUCCESS_REDIRECT_URL})
+        return web.Response(
+            status=302, headers={"Location": IFLOW_SUCCESS_REDIRECT_URL}
+        )
 
     async def wait_for_callback(self, timeout: float = 300.0) -> str:
         """Waits for the OAuth callback and returns the authorization code."""
@@ -146,38 +159,50 @@ class IFlowAuthBase:
     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
+        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)
-        
+        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._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
+        self._queue_processor_task: Optional[asyncio.Task] = (
+            None  # Background worker task
+        )
 
     def _parse_env_credential_path(self, path: str) -> Optional[str]:
         """
         Parse a virtual env:// path and return the credential index.
-        
+
         Supported formats:
         - "env://provider/0" - Legacy single credential (no index in env var names)
         - "env://provider/1" - First numbered credential (IFLOW_1_ACCESS_TOKEN)
-        
+
         Returns:
             The credential index as string, or None if path is not an env:// path
         """
         if not path.startswith("env://"):
             return None
-        
+
         parts = path[6:].split("/")
         if len(parts) >= 2:
             return parts[1]
         return "0"
 
-    def _load_from_env(self, credential_index: Optional[str] = None) -> Optional[Dict[str, Any]]:
+    def _load_from_env(
+        self, credential_index: Optional[str] = None
+    ) -> Optional[Dict[str, Any]]:
         """
         Load OAuth credentials from environment variables for stateless deployments.
 
@@ -204,7 +229,7 @@ class IFlowAuthBase:
         else:
             prefix = "IFLOW"
             default_email = "env-user"
-        
+
         access_token = os.getenv(f"{prefix}_ACCESS_TOKEN")
         refresh_token = os.getenv(f"{prefix}_REFRESH_TOKEN")
         api_key = os.getenv(f"{prefix}_API_KEY")
@@ -213,7 +238,9 @@ class IFlowAuthBase:
         if not (access_token and refresh_token and api_key):
             return None
 
-        lib_logger.debug(f"Loading iFlow credentials from environment variables (prefix: {prefix})")
+        lib_logger.debug(
+            f"Loading iFlow credentials from environment variables (prefix: {prefix})"
+        )
 
         # Parse expiry_date as string (ISO 8601 format)
         expiry_str = os.getenv(f"{prefix}_EXPIRY_DATE", "")
@@ -230,8 +257,8 @@ class IFlowAuthBase:
                 "email": os.getenv(f"{prefix}_EMAIL", default_email),
                 "last_check_timestamp": time.time(),
                 "loaded_from_env": True,
-                "env_credential_index": credential_index or "0"
-            }
+                "env_credential_index": credential_index or "0",
+            },
         }
 
         return creds
@@ -240,7 +267,7 @@ class IFlowAuthBase:
         """Reads credentials from file and populates the cache. No locking."""
         try:
             lib_logger.debug(f"Reading iFlow credentials from file: {path}")
-            with open(path, 'r') as f:
+            with open(path, "r") as f:
                 creds = json.load(f)
             self._credentials_cache[path] = creds
             return creds
@@ -264,11 +291,15 @@ class IFlowAuthBase:
             if credential_index is not None:
                 env_creds = self._load_from_env(credential_index)
                 if env_creds:
-                    lib_logger.info(f"Using iFlow credentials from environment variables (index: {credential_index})")
+                    lib_logger.info(
+                        f"Using iFlow credentials from environment variables (index: {credential_index})"
+                    )
                     self._credentials_cache[path] = env_creds
                     return env_creds
                 else:
-                    raise IOError(f"Environment variables for iFlow credential index {credential_index} not found")
+                    raise IOError(
+                        f"Environment variables for iFlow credential index {credential_index} not found"
+                    )
 
             # For file paths, try loading from legacy env vars first
             env_creds = self._load_from_env()
@@ -298,10 +329,12 @@ class IFlowAuthBase:
         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)
+            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:
+            with os.fdopen(tmp_fd, "w") as f:
                 json.dump(creds, f, indent=2)
                 tmp_fd = None  # fdopen closes the fd
 
@@ -318,10 +351,14 @@ class IFlowAuthBase:
 
             # Update cache AFTER successful file write
             self._credentials_cache[path] = creds
-            lib_logger.debug(f"Saved updated iFlow OAuth credentials to '{path}' (atomic write).")
+            lib_logger.debug(
+                f"Saved updated iFlow OAuth credentials to '{path}' (atomic write)."
+            )
 
         except Exception as e:
-            lib_logger.error(f"Failed to save updated iFlow OAuth credentials to '{path}': {e}")
+            lib_logger.error(
+                f"Failed to save updated iFlow OAuth credentials to '{path}': {e}"
+            )
             # Clean up temp file if it still exists
             if tmp_fd is not None:
                 try:
@@ -345,7 +382,8 @@ class IFlowAuthBase:
         try:
             # Parse ISO 8601 format (e.g., "2025-01-17T12:00:00Z")
             from datetime import datetime
-            expiry_dt = datetime.fromisoformat(expiry_str.replace('Z', '+00:00'))
+
+            expiry_dt = datetime.fromisoformat(expiry_str.replace("Z", "+00:00"))
             expiry_timestamp = expiry_dt.timestamp()
         except (ValueError, AttributeError):
             # Fallback: treat as numeric timestamp
@@ -389,7 +427,9 @@ class IFlowAuthBase:
 
         return {"api_key": api_key, "email": email}
 
-    async def _exchange_code_for_tokens(self, code: str, redirect_uri: str) -> Dict[str, Any]:
+    async def _exchange_code_for_tokens(
+        self, code: str, redirect_uri: str
+    ) -> Dict[str, Any]:
         """
         Exchanges authorization code for access and refresh tokens.
         Uses Basic Auth with client credentials.
@@ -401,7 +441,7 @@ class IFlowAuthBase:
         headers = {
             "Content-Type": "application/x-www-form-urlencoded",
             "Accept": "application/json",
-            "Authorization": f"Basic {basic_auth}"
+            "Authorization": f"Basic {basic_auth}",
         }
 
         data = {
@@ -409,16 +449,22 @@ class IFlowAuthBase:
             "code": code,
             "redirect_uri": redirect_uri,
             "client_id": IFLOW_CLIENT_ID,
-            "client_secret": IFLOW_CLIENT_SECRET
+            "client_secret": IFLOW_CLIENT_SECRET,
         }
 
         async with httpx.AsyncClient(timeout=30.0) as client:
-            response = await client.post(IFLOW_OAUTH_TOKEN_ENDPOINT, headers=headers, data=data)
+            response = await client.post(
+                IFLOW_OAUTH_TOKEN_ENDPOINT, headers=headers, data=data
+            )
 
             if response.status_code != 200:
                 error_text = response.text
-                lib_logger.error(f"iFlow token exchange failed: {response.status_code} {error_text}")
-                raise ValueError(f"Token exchange failed: {response.status_code} {error_text}")
+                lib_logger.error(
+                    f"iFlow token exchange failed: {response.status_code} {error_text}"
+                )
+                raise ValueError(
+                    f"Token exchange failed: {response.status_code} {error_text}"
+                )
 
             token_data = response.json()
 
@@ -436,7 +482,10 @@ class IFlowAuthBase:
 
         # Calculate expiry date
         from datetime import datetime, timedelta
-        expiry_date = (datetime.utcnow() + timedelta(seconds=expires_in)).isoformat() + 'Z'
+
+        expiry_date = (
+            datetime.utcnow() + timedelta(seconds=expires_in)
+        ).isoformat() + "Z"
 
         return {
             "access_token": access_token,
@@ -445,7 +494,7 @@ class IFlowAuthBase:
             "email": user_info["email"],
             "expiry_date": expiry_date,
             "token_type": token_type,
-            "scope": scope
+            "scope": scope,
         }
 
     async def _refresh_token(self, path: str, force: bool = False) -> Dict[str, Any]:
@@ -482,20 +531,22 @@ class IFlowAuthBase:
             headers = {
                 "Content-Type": "application/x-www-form-urlencoded",
                 "Accept": "application/json",
-                "Authorization": f"Basic {basic_auth}"
+                "Authorization": f"Basic {basic_auth}",
             }
 
             data = {
                 "grant_type": "refresh_token",
                 "refresh_token": refresh_token,
                 "client_id": IFLOW_CLIENT_ID,
-                "client_secret": IFLOW_CLIENT_SECRET
+                "client_secret": IFLOW_CLIENT_SECRET,
             }
 
             async with httpx.AsyncClient(timeout=30.0) as client:
                 for attempt in range(max_retries):
                     try:
-                        response = await client.post(IFLOW_OAUTH_TOKEN_ENDPOINT, headers=headers, data=data)
+                        response = await client.post(
+                            IFLOW_OAUTH_TOKEN_ENDPOINT, headers=headers, data=data
+                        )
                         response.raise_for_status()
                         new_token_data = response.json()
                         break  # Success
@@ -505,7 +556,9 @@ class IFlowAuthBase:
                         status_code = e.response.status_code
                         error_body = e.response.text
 
-                        lib_logger.error(f"[REFRESH HTTP ERROR] HTTP {status_code} for '{Path(path).name}': {error_body}")
+                        lib_logger.error(
+                            f"[REFRESH HTTP ERROR] HTTP {status_code} for '{Path(path).name}': {error_body}"
+                        )
 
                         # [STATUS CODE HANDLING]
                         # [INVALID GRANT HANDLING] Handle 401/403 by triggering re-authentication
@@ -519,7 +572,9 @@ class IFlowAuthBase:
 
                         elif status_code == 429:
                             retry_after = int(e.response.headers.get("Retry-After", 60))
-                            lib_logger.warning(f"Rate limited (HTTP 429), retry after {retry_after}s")
+                            lib_logger.warning(
+                                f"Rate limited (HTTP 429), retry after {retry_after}s"
+                            )
                             if attempt < max_retries - 1:
                                 await asyncio.sleep(retry_after)
                                 continue
@@ -527,8 +582,10 @@ class IFlowAuthBase:
 
                         elif 500 <= status_code < 600:
                             if attempt < max_retries - 1:
-                                wait_time = 2 ** attempt
-                                lib_logger.warning(f"Server error (HTTP {status_code}), retry {attempt + 1}/{max_retries} in {wait_time}s")
+                                wait_time = 2**attempt
+                                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
@@ -539,15 +596,19 @@ class IFlowAuthBase:
                     except (httpx.RequestError, httpx.TimeoutException) as e:
                         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")
+                            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}'...")
+                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)
@@ -556,20 +617,34 @@ class IFlowAuthBase:
                     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}")
+                    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._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}")
+                    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
-                backoff_seconds = min(300, 30 * (2 ** self._refresh_failures[path]))  # Max 5 min backoff
+                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")
+                lib_logger.debug(
+                    f"Setting backoff for '{Path(path).name}': {backoff_seconds}s"
+                )
                 raise last_error or Exception("Token refresh failed after all retries")
 
             # Update tokens
@@ -578,14 +653,23 @@ class IFlowAuthBase:
                 raise ValueError("Missing access_token in refresh response")
 
             creds_from_file["access_token"] = access_token
-            creds_from_file["refresh_token"] = new_token_data.get("refresh_token", creds_from_file["refresh_token"])
+            creds_from_file["refresh_token"] = new_token_data.get(
+                "refresh_token", creds_from_file["refresh_token"]
+            )
 
             expires_in = new_token_data.get("expires_in", 3600)
             from datetime import datetime, timedelta
-            creds_from_file["expiry_date"] = (datetime.utcnow() + timedelta(seconds=expires_in)).isoformat() + 'Z'
 
-            creds_from_file["token_type"] = new_token_data.get("token_type", creds_from_file.get("token_type", "Bearer"))
-            creds_from_file["scope"] = new_token_data.get("scope", creds_from_file.get("scope", ""))
+            creds_from_file["expiry_date"] = (
+                datetime.utcnow() + timedelta(seconds=expires_in)
+            ).isoformat() + "Z"
+
+            creds_from_file["token_type"] = new_token_data.get(
+                "token_type", creds_from_file.get("token_type", "Bearer")
+            )
+            creds_from_file["scope"] = new_token_data.get(
+                "scope", creds_from_file.get("scope", "")
+            )
 
             # CRITICAL: Re-fetch user info to get potentially updated API key
             try:
@@ -595,7 +679,9 @@ class IFlowAuthBase:
                 if user_info.get("email"):
                     creds_from_file["email"] = user_info["email"]
             except Exception as e:
-                lib_logger.warning(f"Failed to update API key during token refresh: {e}")
+                lib_logger.warning(
+                    f"Failed to update API key during token refresh: {e}"
+                )
 
             # Ensure _proxy_metadata exists and update timestamp
             if "_proxy_metadata" not in creds_from_file:
@@ -604,16 +690,22 @@ class IFlowAuthBase:
 
             # [VALIDATION] Verify required fields exist after refresh
             required_fields = ["access_token", "refresh_token", "api_key"]
-            missing_fields = [field for field in required_fields if not creds_from_file.get(field)]
+            missing_fields = [
+                field for field in required_fields if not creds_from_file.get(field)
+            ]
             if missing_fields:
-                raise ValueError(f"Refreshed credentials missing required fields: {missing_fields}")
+                raise ValueError(
+                    f"Refreshed credentials missing required fields: {missing_fields}"
+                )
 
             # [BACKOFF TRACKING] Clear failure count on successful refresh
             self._refresh_failures.pop(path, None)
             self._next_refresh_after.pop(path, None)
 
             await self._save_credentials(path, creds_from_file)
-            lib_logger.debug(f"Successfully refreshed iFlow OAuth token for '{Path(path).name}'.")
+            lib_logger.debug(
+                f"Successfully refreshed iFlow OAuth token for '{Path(path).name}'."
+            )
             return creds_from_file
 
     async def get_api_details(self, credential_identifier: str) -> Tuple[str, str]:
@@ -628,7 +720,9 @@ class IFlowAuthBase:
         # Detect credential type
         if os.path.isfile(credential_identifier):
             # OAuth credential: file path to JSON
-            lib_logger.debug(f"Using OAuth credentials from file: {credential_identifier}")
+            lib_logger.debug(
+                f"Using OAuth credentials from file: {credential_identifier}"
+            )
             creds = await self._load_credentials(credential_identifier)
 
             # Check if token needs refresh
@@ -653,7 +747,7 @@ class IFlowAuthBase:
         """
         # Check if it's an env:// virtual path (OAuth credentials from environment)
         is_env_path = credential_identifier.startswith("env://")
-        
+
         # Only refresh if it's an OAuth credential (file path or env:// path)
         if not is_env_path and not os.path.isfile(credential_identifier):
             return  # Direct API key, no refresh needed
@@ -661,7 +755,9 @@ class IFlowAuthBase:
         creds = await self._load_credentials(credential_identifier)
         if self._is_token_expired(creds):
             # Queue for refresh with needs_reauth=False (automated refresh)
-            await self._queue_refresh(credential_identifier, force=False, needs_reauth=False)
+            await self._queue_refresh(
+                credential_identifier, force=False, needs_reauth=False
+            )
 
     async def _get_lock(self, path: str) -> asyncio.Lock:
         """Gets or creates a lock for the given credential path."""
@@ -678,11 +774,15 @@ class IFlowAuthBase:
     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())
+            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):
+    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
@@ -697,9 +797,11 @@ class IFlowAuthBase:
                 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)")
+                    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)
@@ -715,14 +817,13 @@ class IFlowAuthBase:
                 # 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
+                        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):
@@ -733,16 +834,16 @@ class IFlowAuthBase:
                             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, 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:
@@ -757,7 +858,9 @@ class IFlowAuthBase:
                     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]:
+    async def initialize_token(
+        self, creds_or_path: Union[Dict[str, Any], str]
+    ) -> Dict[str, Any]:
         """
         Initiates OAuth authorization code flow if tokens are missing or invalid.
         Uses local callback server to receive authorization code.
@@ -766,14 +869,18 @@ class IFlowAuthBase:
 
         # 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")
+            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 iFlow token for '{display_name}'...")
 
         try:
-            creds = await self._load_credentials(creds_or_path) if path else creds_or_path
+            creds = (
+                await self._load_credentials(creds_or_path) if path else creds_or_path
+            )
 
             reason = ""
             if not creds.get("refresh_token"):
@@ -787,11 +894,15 @@ class IFlowAuthBase:
                     try:
                         return await self._refresh_token(path)
                     except Exception as e:
-                        lib_logger.warning(f"Automatic token refresh for '{display_name}' failed: {e}. Proceeding to interactive login.")
+                        lib_logger.warning(
+                            f"Automatic token refresh for '{display_name}' failed: {e}. Proceeding to interactive login."
+                        )
 
                 # Interactive OAuth flow
-                lib_logger.warning(f"iFlow OAuth token for '{display_name}' needs setup: {reason}.")
-                
+                lib_logger.warning(
+                    f"iFlow OAuth token for '{display_name}' needs setup: {reason}."
+                )
+
                 # [HEADLESS DETECTION] Check if running in headless environment
                 is_headless = is_headless_environment()
 
@@ -805,7 +916,7 @@ class IFlowAuthBase:
                     "type": "phone",
                     "redirect": redirect_uri,
                     "state": state,
-                    "client_id": IFLOW_CLIENT_ID
+                    "client_id": IFLOW_CLIENT_ID,
                 }
                 auth_url = f"{IFLOW_OAUTH_AUTHORIZE_ENDPOINT}?{urlencode(auth_params)}"
 
@@ -829,49 +940,86 @@ class IFlowAuthBase:
                             "2. [bold]Authorize the application[/bold] to access your account.\n"
                             "3. You will be automatically redirected after authorization."
                         )
-                    
-                    console.print(Panel(auth_panel_text, title=f"iFlow 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")
+
+                    console.print(
+                        Panel(
+                            auth_panel_text,
+                            title=f"iFlow OAuth Setup for [bold yellow]{display_name}[/bold yellow]",
+                            style="bold blue",
+                        )
+                    )
+                    # [URL DISPLAY] Print URL with proper escaping to prevent Rich markup issues.
+                    # IMPORTANT: OAuth URLs contain special characters (=, &, etc.) that Rich might
+                    # interpret as markup in some terminal configurations. We escape the URL to
+                    # ensure it displays correctly.
+                    #
+                    # KNOWN ISSUE: If Rich rendering fails entirely (e.g., terminal doesn't support
+                    # ANSI codes, or output is piped), the escaped URL should still be valid.
+                    # However, if the terminal strips or mangles the output, users should copy
+                    # the URL directly from logs or use --verbose to see the raw URL.
+                    #
+                    # The [link=...] markup creates a clickable hyperlink in supported terminals
+                    # (iTerm2, Windows Terminal, etc.), but the displayed text is the escaped URL
+                    # which can be safely copied even if the hyperlink doesn't work.
+                    escaped_url = rich_escape(auth_url)
+                    console.print(
+                        f"[bold]URL:[/bold] [link={auth_url}]{escaped_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 iFlow OAuth flow")
+                            lib_logger.info(
+                                "Browser opened successfully for iFlow OAuth flow"
+                            )
                         except Exception as e:
-                            lib_logger.warning(f"Failed to open browser automatically: {e}. Please open the URL manually.")
+                            lib_logger.warning(
+                                f"Failed to open browser automatically: {e}. Please open the URL manually."
+                            )
 
                     # Wait for callback
-                    with console.status("[bold green]Waiting for authorization in the browser...[/bold green]", spinner="dots"):
+                    with console.status(
+                        "[bold green]Waiting for authorization in the browser...[/bold green]",
+                        spinner="dots",
+                    ):
                         code = await callback_server.wait_for_callback(timeout=300.0)
 
-                    lib_logger.info("Received authorization code, exchanging for tokens...")
+                    lib_logger.info(
+                        "Received authorization code, exchanging for tokens..."
+                    )
 
                     # Exchange code for tokens and API key
-                    token_data = await self._exchange_code_for_tokens(code, redirect_uri)
+                    token_data = await self._exchange_code_for_tokens(
+                        code, redirect_uri
+                    )
 
                     # Update credentials
-                    creds.update({
-                        "access_token": token_data["access_token"],
-                        "refresh_token": token_data["refresh_token"],
-                        "api_key": token_data["api_key"],
-                        "email": token_data["email"],
-                        "expiry_date": token_data["expiry_date"],
-                        "token_type": token_data["token_type"],
-                        "scope": token_data["scope"]
-                    })
+                    creds.update(
+                        {
+                            "access_token": token_data["access_token"],
+                            "refresh_token": token_data["refresh_token"],
+                            "api_key": token_data["api_key"],
+                            "email": token_data["email"],
+                            "expiry_date": token_data["expiry_date"],
+                            "token_type": token_data["token_type"],
+                            "scope": token_data["scope"],
+                        }
+                    )
 
                     # Create metadata object
                     if not creds.get("_proxy_metadata"):
                         creds["_proxy_metadata"] = {
                             "email": token_data["email"],
-                            "last_check_timestamp": time.time()
+                            "last_check_timestamp": time.time(),
                         }
 
                     if path:
                         await self._save_credentials(path, creds)
 
-                    lib_logger.info(f"iFlow OAuth initialized successfully for '{display_name}'.")
+                    lib_logger.info(
+                        f"iFlow OAuth initialized successfully for '{display_name}'."
+                    )
                     return creds
 
                 finally:
@@ -898,11 +1046,15 @@ class IFlowAuthBase:
 
         return {"Authorization": f"Bearer {api_key}"}
 
-    async def get_user_info(self, creds_or_path: Union[Dict[str, Any], str]) -> Dict[str, Any]:
+    async def get_user_info(
+        self, creds_or_path: Union[Dict[str, Any], str]
+    ) -> Dict[str, Any]:
         """Retrieves user info from the _proxy_metadata in the credential file."""
         try:
             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
+            creds = (
+                await self._load_credentials(creds_or_path) if path else creds_or_path
+            )
 
             # Ensure the token is valid
             if path:
@@ -912,7 +1064,9 @@ class IFlowAuthBase:
             email = creds.get("email") or creds.get("_proxy_metadata", {}).get("email")
 
             if not email:
-                lib_logger.warning(f"No email found in iFlow credentials for '{path or 'in-memory object'}'.")
+                lib_logger.warning(
+                    f"No email found in iFlow credentials for '{path or 'in-memory object'}'."
+                )
 
             # Update timestamp on check
             if path and "_proxy_metadata" in creds:

src/rotator_library/providers/qwen_auth_base.py

@@ -19,54 +19,70 @@ from rich.console import Console
 from rich.panel import Panel
 from rich.prompt import Prompt
 from rich.text import Text
+from rich.markup import escape as rich_escape
 
 from ..utils.headless_detection import is_headless_environment
 
-lib_logger = logging.getLogger('rotator_library')
+lib_logger = logging.getLogger("rotator_library")
 
-CLIENT_ID = "f0304373b74a44d2b584a3fb70ca9e56" #https://api.kilocode.ai/extension-config.json
+CLIENT_ID = (
+    "f0304373b74a44d2b584a3fb70ca9e56"  # https://api.kilocode.ai/extension-config.json
+)
 SCOPE = "openid profile email model.completion"
 TOKEN_ENDPOINT = "https://chat.qwen.ai/api/v1/oauth2/token"
 REFRESH_EXPIRY_BUFFER_SECONDS = 3 * 60 * 60  # 3 hours buffer before expiry
 
 console = Console()
 
+
 class QwenAuthBase:
     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
+        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)
-        
+        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._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
+        self._queue_processor_task: Optional[asyncio.Task] = (
+            None  # Background worker task
+        )
 
     def _parse_env_credential_path(self, path: str) -> Optional[str]:
         """
         Parse a virtual env:// path and return the credential index.
-        
+
         Supported formats:
         - "env://provider/0" - Legacy single credential (no index in env var names)
         - "env://provider/1" - First numbered credential (QWEN_CODE_1_ACCESS_TOKEN)
-        
+
         Returns:
             The credential index as string, or None if path is not an env:// path
         """
         if not path.startswith("env://"):
             return None
-        
+
         parts = path[6:].split("/")
         if len(parts) >= 2:
             return parts[1]
         return "0"
 
-    def _load_from_env(self, credential_index: Optional[str] = None) -> Optional[Dict[str, Any]]:
+    def _load_from_env(
+        self, credential_index: Optional[str] = None
+    ) -> Optional[Dict[str, Any]]:
         """
         Load OAuth credentials from environment variables for stateless deployments.
 
@@ -91,7 +107,7 @@ class QwenAuthBase:
         else:
             prefix = "QWEN_CODE"
             default_email = "env-user"
-        
+
         access_token = os.getenv(f"{prefix}_ACCESS_TOKEN")
         refresh_token = os.getenv(f"{prefix}_REFRESH_TOKEN")
 
@@ -99,27 +115,33 @@ class QwenAuthBase:
         if not (access_token and refresh_token):
             return None
 
-        lib_logger.debug(f"Loading Qwen Code credentials from environment variables (prefix: {prefix})")
+        lib_logger.debug(
+            f"Loading Qwen Code credentials from environment variables (prefix: {prefix})"
+        )
 
         # Parse expiry_date as float, default to 0 if not present
         expiry_str = os.getenv(f"{prefix}_EXPIRY_DATE", "0")
         try:
             expiry_date = float(expiry_str)
         except ValueError:
-            lib_logger.warning(f"Invalid {prefix}_EXPIRY_DATE value: {expiry_str}, using 0")
+            lib_logger.warning(
+                f"Invalid {prefix}_EXPIRY_DATE value: {expiry_str}, using 0"
+            )
             expiry_date = 0
 
         creds = {
             "access_token": access_token,
             "refresh_token": refresh_token,
             "expiry_date": expiry_date,
-            "resource_url": os.getenv(f"{prefix}_RESOURCE_URL", "https://portal.qwen.ai/v1"),
+            "resource_url": os.getenv(
+                f"{prefix}_RESOURCE_URL", "https://portal.qwen.ai/v1"
+            ),
             "_proxy_metadata": {
                 "email": os.getenv(f"{prefix}_EMAIL", default_email),
                 "last_check_timestamp": time.time(),
                 "loaded_from_env": True,
-                "env_credential_index": credential_index or "0"
-            }
+                "env_credential_index": credential_index or "0",
+            },
         }
 
         return creds
@@ -128,7 +150,7 @@ class QwenAuthBase:
         """Reads credentials from file and populates the cache. No locking."""
         try:
             lib_logger.debug(f"Reading Qwen credentials from file: {path}")
-            with open(path, 'r') as f:
+            with open(path, "r") as f:
                 creds = json.load(f)
             self._credentials_cache[path] = creds
             return creds
@@ -152,16 +174,22 @@ class QwenAuthBase:
             if credential_index is not None:
                 env_creds = self._load_from_env(credential_index)
                 if env_creds:
-                    lib_logger.info(f"Using Qwen Code credentials from environment variables (index: {credential_index})")
+                    lib_logger.info(
+                        f"Using Qwen Code credentials from environment variables (index: {credential_index})"
+                    )
                     self._credentials_cache[path] = env_creds
                     return env_creds
                 else:
-                    raise IOError(f"Environment variables for Qwen Code credential index {credential_index} not found")
+                    raise IOError(
+                        f"Environment variables for Qwen Code credential index {credential_index} not found"
+                    )
 
             # For file paths, try loading from legacy env vars first
             env_creds = self._load_from_env()
             if env_creds:
-                lib_logger.info("Using Qwen Code credentials from environment variables")
+                lib_logger.info(
+                    "Using Qwen Code credentials from environment variables"
+                )
                 self._credentials_cache[path] = env_creds
                 return env_creds
 
@@ -184,10 +212,12 @@ class QwenAuthBase:
         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)
+            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:
+            with os.fdopen(tmp_fd, "w") as f:
                 json.dump(creds, f, indent=2)
                 tmp_fd = None  # fdopen closes the fd
 
@@ -204,10 +234,14 @@ class QwenAuthBase:
 
             # Update cache AFTER successful file write
             self._credentials_cache[path] = creds
-            lib_logger.debug(f"Saved updated Qwen OAuth credentials to '{path}' (atomic write).")
+            lib_logger.debug(
+                f"Saved updated Qwen OAuth credentials to '{path}' (atomic write)."
+            )
 
         except Exception as e:
-            lib_logger.error(f"Failed to save updated Qwen OAuth credentials to '{path}': {e}")
+            lib_logger.error(
+                f"Failed to save updated Qwen OAuth credentials to '{path}': {e}"
+            )
             # Clean up temp file if it still exists
             if tmp_fd is not None:
                 try:
@@ -252,17 +286,22 @@ class QwenAuthBase:
             headers = {
                 "Content-Type": "application/x-www-form-urlencoded",
                 "Accept": "application/json",
-                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
+                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
             }
 
             async with httpx.AsyncClient() as client:
                 for attempt in range(max_retries):
                     try:
-                        response = await client.post(TOKEN_ENDPOINT, headers=headers, data={
-                            "grant_type": "refresh_token",
-                            "refresh_token": refresh_token,
-                            "client_id": CLIENT_ID,
-                        }, timeout=30.0)
+                        response = await client.post(
+                            TOKEN_ENDPOINT,
+                            headers=headers,
+                            data={
+                                "grant_type": "refresh_token",
+                                "refresh_token": refresh_token,
+                                "client_id": CLIENT_ID,
+                            },
+                            timeout=30.0,
+                        )
                         response.raise_for_status()
                         new_token_data = response.json()
                         break  # Success
@@ -271,7 +310,9 @@ class QwenAuthBase:
                         last_error = e
                         status_code = e.response.status_code
                         error_body = e.response.text
-                        lib_logger.error(f"HTTP {status_code} for '{Path(path).name}': {error_body}")
+                        lib_logger.error(
+                            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):
@@ -284,7 +325,9 @@ class QwenAuthBase:
 
                         elif status_code == 429:
                             retry_after = int(e.response.headers.get("Retry-After", 60))
-                            lib_logger.warning(f"Rate limited (HTTP 429), retry after {retry_after}s")
+                            lib_logger.warning(
+                                f"Rate limited (HTTP 429), retry after {retry_after}s"
+                            )
                             if attempt < max_retries - 1:
                                 await asyncio.sleep(retry_after)
                                 continue
@@ -292,8 +335,10 @@ class QwenAuthBase:
 
                         elif 500 <= status_code < 600:
                             if attempt < max_retries - 1:
-                                wait_time = 2 ** attempt
-                                lib_logger.warning(f"Server error (HTTP {status_code}), retry {attempt + 1}/{max_retries} in {wait_time}s")
+                                wait_time = 2**attempt
+                                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
@@ -304,15 +349,19 @@ class QwenAuthBase:
                     except (httpx.RequestError, httpx.TimeoutException) as e:
                         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")
+                            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}'...")
+                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)
@@ -321,26 +370,46 @@ class QwenAuthBase:
                     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}")
+                    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._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}")
+                    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
-                backoff_seconds = min(300, 30 * (2 ** self._refresh_failures[path]))  # Max 5 min backoff
+                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")
+                lib_logger.debug(
+                    f"Setting backoff for '{Path(path).name}': {backoff_seconds}s"
+                )
                 raise last_error or Exception("Token refresh failed after all retries")
 
             creds_from_file["access_token"] = new_token_data["access_token"]
-            creds_from_file["refresh_token"] = new_token_data.get("refresh_token", creds_from_file["refresh_token"])
-            creds_from_file["expiry_date"] = (time.time() + new_token_data["expires_in"]) * 1000
-            creds_from_file["resource_url"] = new_token_data.get("resource_url", creds_from_file.get("resource_url"))
+            creds_from_file["refresh_token"] = new_token_data.get(
+                "refresh_token", creds_from_file["refresh_token"]
+            )
+            creds_from_file["expiry_date"] = (
+                time.time() + new_token_data["expires_in"]
+            ) * 1000
+            creds_from_file["resource_url"] = new_token_data.get(
+                "resource_url", creds_from_file.get("resource_url")
+            )
 
             # Ensure _proxy_metadata exists and update timestamp
             if "_proxy_metadata" not in creds_from_file:
@@ -349,16 +418,22 @@ class QwenAuthBase:
 
             # [VALIDATION] Verify required fields exist after refresh
             required_fields = ["access_token", "refresh_token"]
-            missing_fields = [field for field in required_fields if not creds_from_file.get(field)]
+            missing_fields = [
+                field for field in required_fields if not creds_from_file.get(field)
+            ]
             if missing_fields:
-                raise ValueError(f"Refreshed credentials missing required fields: {missing_fields}")
+                raise ValueError(
+                    f"Refreshed credentials missing required fields: {missing_fields}"
+                )
 
             # [BACKOFF TRACKING] Clear failure count on successful refresh
             self._refresh_failures.pop(path, None)
             self._next_refresh_after.pop(path, None)
 
             await self._save_credentials(path, creds_from_file)
-            lib_logger.debug(f"Successfully refreshed Qwen OAuth token for '{Path(path).name}'.")
+            lib_logger.debug(
+                f"Successfully refreshed Qwen OAuth token for '{Path(path).name}'."
+            )
             return creds_from_file
 
     async def get_api_details(self, credential_identifier: str) -> Tuple[str, str]:
@@ -372,12 +447,14 @@ class QwenAuthBase:
         # Detect credential type
         if os.path.isfile(credential_identifier):
             # OAuth credential: file path to JSON
-            lib_logger.debug(f"Using OAuth credentials from file: {credential_identifier}")
+            lib_logger.debug(
+                f"Using OAuth credentials from file: {credential_identifier}"
+            )
             creds = await self._load_credentials(credential_identifier)
 
             if self._is_token_expired(creds):
                 creds = await self._refresh_token(credential_identifier)
-                
+
             base_url = creds.get("resource_url", "https://portal.qwen.ai/v1")
             if not base_url.startswith("http"):
                 base_url = f"https://{base_url}"
@@ -397,7 +474,7 @@ class QwenAuthBase:
         """
         # Check if it's an env:// virtual path (OAuth credentials from environment)
         is_env_path = credential_identifier.startswith("env://")
-        
+
         # Only refresh if it's an OAuth credential (file path or env:// path)
         if not is_env_path and not os.path.isfile(credential_identifier):
             return  # Direct API key, no refresh needed
@@ -405,7 +482,9 @@ class QwenAuthBase:
         creds = await self._load_credentials(credential_identifier)
         if self._is_token_expired(creds):
             # Queue for refresh with needs_reauth=False (automated refresh)
-            await self._queue_refresh(credential_identifier, force=False, needs_reauth=False)
+            await self._queue_refresh(
+                credential_identifier, force=False, needs_reauth=False
+            )
 
     async def _get_lock(self, path: str) -> asyncio.Lock:
         # [FIX RACE CONDITION] Protect lock creation with a master lock
@@ -421,11 +500,15 @@ class QwenAuthBase:
     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())
+            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):
+    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
@@ -440,9 +523,11 @@ class QwenAuthBase:
                 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)")
+                    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)
@@ -458,14 +543,13 @@ class QwenAuthBase:
                 # 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
+                        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):
@@ -476,16 +560,16 @@ class QwenAuthBase:
                             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, 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:
@@ -500,19 +584,25 @@ class QwenAuthBase:
                     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]:
+    async def initialize_token(
+        self, creds_or_path: Union[Dict[str, Any], str]
+    ) -> Dict[str, Any]:
         """Initiates device flow if tokens are missing or invalid."""
         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")
+            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 Qwen token for '{display_name}'...")
         try:
-            creds = await self._load_credentials(creds_or_path) if path else creds_or_path
+            creds = (
+                await self._load_credentials(creds_or_path) if path else creds_or_path
+            )
 
             reason = ""
             if not creds.get("refresh_token"):
@@ -525,44 +615,58 @@ class QwenAuthBase:
                     try:
                         return await self._refresh_token(path)
                     except Exception as e:
-                        lib_logger.warning(f"Automatic token refresh for '{display_name}' failed: {e}. Proceeding to interactive login.")
+                        lib_logger.warning(
+                            f"Automatic token refresh for '{display_name}' failed: {e}. Proceeding to interactive login."
+                        )
+
+                lib_logger.warning(
+                    f"Qwen OAuth token for '{display_name}' needs setup: {reason}."
+                )
 
-                lib_logger.warning(f"Qwen OAuth token for '{display_name}' needs setup: {reason}.")
-                
                 # [HEADLESS DETECTION] Check if running in headless environment
                 is_headless = is_headless_environment()
-                
-                code_verifier = base64.urlsafe_b64encode(secrets.token_bytes(32)).decode('utf-8').rstrip('=')
-                code_challenge = base64.urlsafe_b64encode(
-                    hashlib.sha256(code_verifier.encode('utf-8')).digest()
-                ).decode('utf-8').rstrip('=')
-                
+
+                code_verifier = (
+                    base64.urlsafe_b64encode(secrets.token_bytes(32))
+                    .decode("utf-8")
+                    .rstrip("=")
+                )
+                code_challenge = (
+                    base64.urlsafe_b64encode(
+                        hashlib.sha256(code_verifier.encode("utf-8")).digest()
+                    )
+                    .decode("utf-8")
+                    .rstrip("=")
+                )
+
                 headers = {
                     "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
                     "Content-Type": "application/x-www-form-urlencoded",
-                    "Accept": "application/json"
+                    "Accept": "application/json",
                 }
                 async with httpx.AsyncClient() as client:
                     request_data = {
                         "client_id": CLIENT_ID,
                         "scope": SCOPE,
                         "code_challenge": code_challenge,
-                        "code_challenge_method": "S256"
+                        "code_challenge_method": "S256",
                     }
                     lib_logger.debug(f"Qwen device code request data: {request_data}")
                     try:
                         dev_response = await client.post(
                             "https://chat.qwen.ai/api/v1/oauth2/device/code",
                             headers=headers,
-                            data=request_data
+                            data=request_data,
                         )
                         dev_response.raise_for_status()
                         dev_data = dev_response.json()
                         lib_logger.debug(f"Qwen device auth response: {dev_data}")
                     except httpx.HTTPStatusError as e:
-                        lib_logger.error(f"Qwen device code request failed with status {e.response.status_code}: {e.response.text}")
+                        lib_logger.error(
+                            f"Qwen device code request failed with status {e.response.status_code}: {e.response.text}"
+                        )
                         raise e
-                    
+
                     # [HEADLESS SUPPORT] Display appropriate instructions
                     if is_headless:
                         auth_panel_text = Text.from_markup(
@@ -578,33 +682,63 @@ class QwenAuthBase:
                             "2. [bold]Copy your email[/bold] or another unique identifier and authorize the application.\n"
                             "3. You will be prompted to enter your identifier after authorization."
                         )
-                    
-                    console.print(Panel(auth_panel_text, title=f"Qwen OAuth Setup for [bold yellow]{display_name}[/bold yellow]", style="bold blue"))
-                    console.print(f"[bold]URL:[/bold] [link={dev_data['verification_uri_complete']}]{dev_data['verification_uri_complete']}[/link]\n")
-                    
+
+                    console.print(
+                        Panel(
+                            auth_panel_text,
+                            title=f"Qwen OAuth Setup for [bold yellow]{display_name}[/bold yellow]",
+                            style="bold blue",
+                        )
+                    )
+                    # [URL DISPLAY] Print URL with proper escaping to prevent Rich markup issues.
+                    # IMPORTANT: OAuth URLs contain special characters (=, &, etc.) that Rich might
+                    # interpret as markup in some terminal configurations. We escape the URL to
+                    # ensure it displays correctly.
+                    #
+                    # KNOWN ISSUE: If Rich rendering fails entirely (e.g., terminal doesn't support
+                    # ANSI codes, or output is piped), the escaped URL should still be valid.
+                    # However, if the terminal strips or mangles the output, users should copy
+                    # the URL directly from logs or use --verbose to see the raw URL.
+                    #
+                    # The [link=...] markup creates a clickable hyperlink in supported terminals
+                    # (iTerm2, Windows Terminal, etc.), but the displayed text is the escaped URL
+                    # which can be safely copied even if the hyperlink doesn't work.
+                    verification_url = dev_data["verification_uri_complete"]
+                    escaped_url = rich_escape(verification_url)
+                    console.print(
+                        f"[bold]URL:[/bold] [link={verification_url}]{escaped_url}[/link]\n"
+                    )
+
                     # [HEADLESS SUPPORT] Only attempt browser open if NOT headless
                     if not is_headless:
                         try:
-                            webbrowser.open(dev_data['verification_uri_complete'])
-                            lib_logger.info("Browser opened successfully for Qwen OAuth flow")
+                            webbrowser.open(dev_data["verification_uri_complete"])
+                            lib_logger.info(
+                                "Browser opened successfully for Qwen OAuth flow"
+                            )
                         except Exception as e:
-                            lib_logger.warning(f"Failed to open browser automatically: {e}. Please open the URL manually.")
-                    
+                            lib_logger.warning(
+                                f"Failed to open browser automatically: {e}. Please open the URL manually."
+                            )
+
                     token_data = None
                     start_time = time.time()
-                    interval = dev_data.get('interval', 5)
+                    interval = dev_data.get("interval", 5)
 
-                    with console.status("[bold green]Polling for token, please complete authentication in the browser...[/bold green]", spinner="dots") as status:
-                        while time.time() - start_time < dev_data['expires_in']:
+                    with console.status(
+                        "[bold green]Polling for token, please complete authentication in the browser...[/bold green]",
+                        spinner="dots",
+                    ) as status:
+                        while time.time() - start_time < dev_data["expires_in"]:
                             poll_response = await client.post(
                                 TOKEN_ENDPOINT,
                                 headers=headers,
                                 data={
                                     "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
-                                    "device_code": dev_data['device_code'],
+                                    "device_code": dev_data["device_code"],
                                     "client_id": CLIENT_ID,
-                                    "code_verifier": code_verifier
-                                }
+                                    "code_verifier": code_verifier,
+                                },
                             )
                             if poll_response.status_code == 200:
                                 token_data = poll_response.json()
@@ -614,45 +748,63 @@ class QwenAuthBase:
                                 poll_data = poll_response.json()
                                 error_type = poll_data.get("error")
                                 if error_type == "authorization_pending":
-                                    lib_logger.debug(f"Polling status: {error_type}, waiting {interval}s")
+                                    lib_logger.debug(
+                                        f"Polling status: {error_type}, waiting {interval}s"
+                                    )
                                 elif error_type == "slow_down":
                                     interval = int(interval * 1.5)
                                     if interval > 10:
                                         interval = 10
-                                    lib_logger.debug(f"Polling status: {error_type}, waiting {interval}s")
+                                    lib_logger.debug(
+                                        f"Polling status: {error_type}, waiting {interval}s"
+                                    )
                                 else:
-                                    raise ValueError(f"Token polling failed: {poll_data.get('error_description', error_type)}")
+                                    raise ValueError(
+                                        f"Token polling failed: {poll_data.get('error_description', error_type)}"
+                                    )
                             else:
                                 poll_response.raise_for_status()
-                            
+
                             await asyncio.sleep(interval)
-                    
+
                     if not token_data:
                         raise TimeoutError("Qwen device flow timed out.")
-                    
-                    creds.update({
-                        "access_token": token_data["access_token"],
-                        "refresh_token": token_data.get("refresh_token"),
-                        "expiry_date": (time.time() + token_data["expires_in"]) * 1000,
-                        "resource_url": token_data.get("resource_url")
-                    })
+
+                    creds.update(
+                        {
+                            "access_token": token_data["access_token"],
+                            "refresh_token": token_data.get("refresh_token"),
+                            "expiry_date": (time.time() + token_data["expires_in"])
+                            * 1000,
+                            "resource_url": token_data.get("resource_url"),
+                        }
+                    )
 
                     # Prompt for user identifier and create metadata object if needed
                     if not creds.get("_proxy_metadata", {}).get("email"):
                         try:
-                            prompt_text = Text.from_markup(f"\\n[bold]Please enter your email or a unique identifier for [yellow]'{display_name}'[/yellow][/bold]")
+                            prompt_text = Text.from_markup(
+                                f"\\n[bold]Please enter your email or a unique identifier for [yellow]'{display_name}'[/yellow][/bold]"
+                            )
                             email = Prompt.ask(prompt_text)
                             creds["_proxy_metadata"] = {
                                 "email": email.strip(),
-                                "last_check_timestamp": time.time()
+                                "last_check_timestamp": time.time(),
                             }
                         except (EOFError, KeyboardInterrupt):
-                            console.print("\\n[bold yellow]No identifier provided. Deduplication will not be possible.[/bold yellow]")
-                            creds["_proxy_metadata"] = {"email": None, "last_check_timestamp": time.time()}
+                            console.print(
+                                "\\n[bold yellow]No identifier provided. Deduplication will not be possible.[/bold yellow]"
+                            )
+                            creds["_proxy_metadata"] = {
+                                "email": None,
+                                "last_check_timestamp": time.time(),
+                            }
 
                     if path:
                         await self._save_credentials(path, creds)
-                    lib_logger.info(f"Qwen OAuth initialized successfully for '{display_name}'.")
+                    lib_logger.info(
+                        f"Qwen OAuth initialized successfully for '{display_name}'."
+                    )
                 return creds
 
             lib_logger.info(f"Qwen OAuth token at '{display_name}' is valid.")
@@ -666,24 +818,32 @@ class QwenAuthBase:
             creds = await self._refresh_token(credential_path)
         return {"Authorization": f"Bearer {creds['access_token']}"}
 
-    async def get_user_info(self, creds_or_path: Union[Dict[str, Any], str]) -> Dict[str, Any]:
+    async def get_user_info(
+        self, creds_or_path: Union[Dict[str, Any], str]
+    ) -> Dict[str, Any]:
         """
         Retrieves user info from the _proxy_metadata in the credential file.
         """
         try:
             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
-            
+            creds = (
+                await self._load_credentials(creds_or_path) if path else creds_or_path
+            )
+
             # This will ensure the token is valid and metadata exists if the flow was just run
             if path:
                 await self.initialize_token(path)
-                creds = await self._load_credentials(path) # Re-load after potential init
+                creds = await self._load_credentials(
+                    path
+                )  # Re-load after potential init
 
             metadata = creds.get("_proxy_metadata", {"email": None})
             email = metadata.get("email")
 
             if not email:
-                lib_logger.warning(f"No email found in _proxy_metadata for '{path or 'in-memory object'}'.")
+                lib_logger.warning(
+                    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
             if path and "_proxy_metadata" in creds:
@@ -693,4 +853,4 @@ class QwenAuthBase:
             return {"email": email}
         except Exception as e:
             lib_logger.error(f"Failed to get Qwen user info from credentials: {e}")
-            return {"email": None}
+            return {"email": None}

src/rotator_library/utils/headless_detection.py

@@ -1,24 +1,27 @@
 # src/rotator_library/utils/headless_detection.py
 
 import os
+import sys
 import logging
 
-lib_logger = logging.getLogger('rotator_library')
+lib_logger = logging.getLogger("rotator_library")
 
 # Import console for user-visible output
 try:
     from rich.console import Console
+
     console = Console()
 except ImportError:
     console = None
 
+
 def is_headless_environment() -> bool:
     """
     Detects if the current environment is headless (no GUI available).
-    
+
     Returns:
         True if headless environment is detected, False otherwise
-    
+
     Detection logic:
     - Linux/Unix: Check DISPLAY environment variable
     - SSH detection: Check SSH_CONNECTION or SSH_CLIENT
@@ -26,17 +29,20 @@ def is_headless_environment() -> bool:
     - Windows: Check SESSIONNAME for service/headless indicators
     """
     headless_indicators = []
-    
-    # Check DISPLAY for Linux/Unix GUI availability (skip on Windows)
-    if os.name != 'nt':  # Only check DISPLAY on non-Windows systems
+
+    # Check DISPLAY for Linux GUI availability (skip on Windows and macOS)
+    # NOTE: DISPLAY is an X11 (X Window System) variable used on Linux.
+    # macOS uses its native Quartz windowing system, NOT X11, so DISPLAY is
+    # typically unset on macOS even with a full GUI. Only check DISPLAY on Linux.
+    if os.name != "nt" and sys.platform != "darwin":  # Linux only
         display = os.getenv("DISPLAY")
         if display is None or display.strip() == "":
-            headless_indicators.append("No DISPLAY variable (Linux/Unix headless)")
-    
+            headless_indicators.append("No DISPLAY variable (Linux headless)")
+
     # Check for SSH connection
     if os.getenv("SSH_CONNECTION") or os.getenv("SSH_CLIENT") or os.getenv("SSH_TTY"):
         headless_indicators.append("SSH connection detected")
-    
+
     # Check for CI environments
     ci_vars = [
         "CI",  # Generic CI indicator
@@ -55,30 +61,38 @@ def is_headless_environment() -> bool:
         if os.getenv(var):
             headless_indicators.append(f"CI environment detected ({var})")
             break
-    
+
     # Check Windows session type
-    if os.name == 'nt':  # Windows
+    if os.name == "nt":  # Windows
         session_name = os.getenv("SESSIONNAME", "").lower()
         if session_name in ["services", "rdp-tcp"]:
             headless_indicators.append(f"Windows headless session ({session_name})")
-    
+
     # Detect Docker/container environment
     if os.path.exists("/.dockerenv") or os.path.exists("/run/.containerenv"):
         headless_indicators.append("Container environment detected")
-    
+
     # Determine if headless
     is_headless = len(headless_indicators) > 0
-    
+
     if is_headless:
         # Log to logger
-        lib_logger.info(f"Headless environment detected: {'; '.join(headless_indicators)}")
-        
+        lib_logger.info(
+            f"Headless environment detected: {'; '.join(headless_indicators)}"
+        )
+
         # Print to console for user visibility
         if console:
-            console.print(f"[yellow]ℹ Headless environment detected:[/yellow] {'; '.join(headless_indicators)}")
-            console.print("[yellow]→ Browser will NOT open automatically. Please use the URL below.[/yellow]\n")
+            console.print(
+                f"[yellow]ℹ Headless environment detected:[/yellow] {'; '.join(headless_indicators)}"
+            )
+            console.print(
+                "[yellow]→ Browser will NOT open automatically. Please use the URL below.[/yellow]\n"
+            )
     else:
         # Only log to debug, no console output
-        lib_logger.debug("GUI environment detected, browser auto-open will be attempted")
-    
+        lib_logger.debug(
+            "GUI environment detected, browser auto-open will be attempted"
+        )
+
     return is_headless