Merge pull request #34 from MasuRii/fix/antigravity-credential-stuck-unavailable

src/proxy_app/main.py

@@ -570,23 +570,11 @@ async def lifespan(app: FastAPI):
     )
 
     # Log loaded credentials summary (compact, always visible for deployment verification)
-    _api_summary = (
-        ", ".join([f"{p}:{len(c)}" for p, c in api_keys.items()])
-        if api_keys
-        else "none"
-    )
-    _oauth_summary = (
-        ", ".join([f"{p}:{len(c)}" for p, c in oauth_credentials.items()])
-        if oauth_credentials
-        else "none"
-    )
-    _total_summary = ", ".join(
-        [f"{p}:{len(c)}" for p, c in client.all_credentials.items()]
-    )
-    print(
-        f"🔑 Credentials loaded: {_total_summary} (API: {_api_summary} | OAuth: {_oauth_summary})"
-    )
-    client.background_refresher.start()  # Start the background task
+    #_api_summary = ', '.join([f"{p}:{len(c)}" for p, c in api_keys.items()]) if api_keys else "none"
+    #_oauth_summary = ', '.join([f"{p}:{len(c)}" for p, c in oauth_credentials.items()]) if oauth_credentials else "none"
+    #_total_summary = ', '.join([f"{p}:{len(c)}" for p, c in client.all_credentials.items()])
+    #print(f"🔑 Credentials loaded: {_total_summary} (API: {_api_summary} | OAuth: {_oauth_summary})")
+    client.background_refresher.start() # Start the background task
     app.state.rotating_client = client
 
     # Warn if no provider credentials are configured

src/rotator_library/background_refresher.py

@@ -18,6 +18,9 @@ class BackgroundRefresher:
     """
 
     def __init__(self, client: "RotatingClient"):
+        self._client = client
+        self._task: Optional[asyncio.Task] = None
+        self._initialized = False
         try:
             interval_str = os.getenv("OAUTH_REFRESH_INTERVAL", "600")
             self._interval = int(interval_str)
@@ -26,9 +29,6 @@ class BackgroundRefresher:
                 f"Invalid OAUTH_REFRESH_INTERVAL '{interval_str}'. Falling back to 600s."
             )
             self._interval = 600
-        self._client = client
-        self._task: Optional[asyncio.Task] = None
-        self._initialized = False
 
     def start(self):
         """Starts the background refresh task."""

src/rotator_library/providers/google_oauth_base.py

@@ -19,6 +19,7 @@ from rich.text import Text
 from rich.markup import escape as rich_escape
 
 from ..utils.headless_detection import is_headless_environment
+from ..utils.reauth_coordinator import get_reauth_coordinator
 
 lib_logger = logging.getLogger("rotator_library")
 
@@ -85,9 +86,12 @@ class GoogleOAuthBase:
         # [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
+        # [FIX PR#34] Changed from set to dict mapping credential path to timestamp
+        # This enables TTL-based stale entry cleanup as defense in depth
+        self._unavailable_credentials: Dict[
+            str, float
+        ] = {}  # Maps credential path -> timestamp when marked unavailable
+        self._unavailable_ttl_seconds: int = 300  # 5 minutes TTL for stale entries
         self._queue_tracking_lock = asyncio.Lock()  # Protects queue sets
         self._queue_processor_task: Optional[asyncio.Task] = (
             None  # Background worker task
@@ -526,8 +530,33 @@ class GoogleOAuthBase:
             return self._refresh_locks[path]
 
     def is_credential_available(self, path: str) -> bool:
-        """Check if a credential is available for rotation (not queued/refreshing)."""
-        return path not in self._unavailable_credentials
+        """Check if a credential is available for rotation (not queued/refreshing).
+
+        [FIX PR#34] Now includes TTL-based stale entry cleanup as defense in depth.
+        If a credential has been unavailable for longer than _unavailable_ttl_seconds,
+        it is automatically cleaned up and considered available.
+        """
+        if path not in self._unavailable_credentials:
+            return True
+
+        # [FIX PR#34] Check if the entry is stale (TTL expired)
+        marked_time = self._unavailable_credentials.get(path)
+        if marked_time is not None:
+            now = time.time()
+            if now - marked_time > self._unavailable_ttl_seconds:
+                # Entry is stale - clean it up and return available
+                lib_logger.warning(
+                    f"Credential '{Path(path).name}' was stuck in unavailable state for "
+                    f"{int(now - marked_time)}s (TTL: {self._unavailable_ttl_seconds}s). "
+                    f"Auto-cleaning stale entry."
+                )
+                # Note: This is a sync method, so we can't use async lock here.
+                # However, pop from dict is thread-safe for single operations.
+                # The _queue_tracking_lock protects concurrent modifications in async context.
+                self._unavailable_credentials.pop(path, None)
+                return True
+
+        return False
 
     async def _ensure_queue_processor_running(self):
         """Lazily starts the queue processor if not already running."""
@@ -563,7 +592,12 @@ class GoogleOAuthBase:
         async with self._queue_tracking_lock:
             if path not in self._queued_credentials:
                 self._queued_credentials.add(path)
-                self._unavailable_credentials.add(path)  # Mark as unavailable
+                # [FIX PR#34] Store timestamp when marking unavailable (for TTL cleanup)
+                self._unavailable_credentials[path] = time.time()
+                lib_logger.debug(
+                    f"Marked '{Path(path).name}' as unavailable. "
+                    f"Total unavailable: {len(self._unavailable_credentials)}"
+                )
                 await self._refresh_queue.put((path, force, needs_reauth))
                 await self._ensure_queue_processor_running()
 
@@ -578,7 +612,16 @@ class GoogleOAuthBase:
                         self._refresh_queue.get(), timeout=60.0
                     )
                 except asyncio.TimeoutError:
-                    # No items for 60s, exit to save resources
+                    # [FIX PR#34] Clean up any stale unavailable entries before exiting
+                    # If we're idle for 60s, no refreshes are in progress
+                    async with self._queue_tracking_lock:
+                        if self._unavailable_credentials:
+                            stale_count = len(self._unavailable_credentials)
+                            lib_logger.warning(
+                                f"Queue processor idle timeout. Cleaning {stale_count} "
+                                f"stale unavailable credentials: {list(self._unavailable_credentials.keys())}"
+                            )
+                            self._unavailable_credentials.clear()
                     self._queue_processor_task = None
                     return
 
@@ -590,7 +633,11 @@ class GoogleOAuthBase:
                         if creds and not self._is_token_expired(creds):
                             # No longer expired, mark as available
                             async with self._queue_tracking_lock:
-                                self._unavailable_credentials.discard(path)
+                                self._unavailable_credentials.pop(path, None)
+                                lib_logger.debug(
+                                    f"Credential '{Path(path).name}' no longer expired, marked available. "
+                                    f"Remaining unavailable: {len(self._unavailable_credentials)}"
+                                )
                             continue
 
                         # Perform refresh
@@ -600,25 +647,235 @@ class GoogleOAuthBase:
 
                         # SUCCESS: Mark as available again
                         async with self._queue_tracking_lock:
-                            self._unavailable_credentials.discard(path)
+                            self._unavailable_credentials.pop(path, None)
+                            lib_logger.debug(
+                                f"Refresh SUCCESS for '{Path(path).name}', marked available. "
+                                f"Remaining unavailable: {len(self._unavailable_credentials)}"
+                            )
 
                 finally:
-                    # Remove from queued set
+                    # [FIX PR#34] Remove from BOTH queued set AND unavailable credentials
+                    # This ensures cleanup happens in ALL exit paths (success, exception, etc.)
                     async with self._queue_tracking_lock:
                         self._queued_credentials.discard(path)
+                        # [FIX PR#34] Always clean up unavailable credentials in finally block
+                        self._unavailable_credentials.pop(path, None)
+                        lib_logger.debug(
+                            f"Finally cleanup for '{Path(path).name}'. "
+                            f"Remaining unavailable: {len(self._unavailable_credentials)}"
+                        )
                     self._refresh_queue.task_done()
             except asyncio.CancelledError:
+                # [FIX PR#34] Clean up the current credential before breaking
+                if path:
+                    async with self._queue_tracking_lock:
+                        self._unavailable_credentials.pop(path, None)
+                        lib_logger.debug(
+                            f"CancelledError cleanup for '{Path(path).name}'. "
+                            f"Remaining unavailable: {len(self._unavailable_credentials)}"
+                        )
                 break
             except Exception as e:
                 lib_logger.error(f"Error in queue processor: {e}")
                 # Even on error, mark as available (backoff will prevent immediate retry)
                 if path:
                     async with self._queue_tracking_lock:
-                        self._unavailable_credentials.discard(path)
+                        self._unavailable_credentials.pop(path, None)
+                        lib_logger.debug(
+                            f"Error cleanup for '{Path(path).name}': {e}. "
+                            f"Remaining unavailable: {len(self._unavailable_credentials)}"
+                        )
+
+    async def _perform_interactive_oauth(
+        self, path: str, creds: Dict[str, Any], display_name: str
+    ) -> Dict[str, Any]:
+        """
+        Perform interactive OAuth flow (browser-based authentication).
+
+        This method is called via the global ReauthCoordinator to ensure
+        only one interactive OAuth flow runs at a time across all providers.
+
+        Args:
+            path: Credential file path
+            creds: Current credentials dict (will be updated)
+            display_name: Display name for logging/UI
+
+        Returns:
+            Updated credentials dict with new tokens
+        """
+        # [HEADLESS DETECTION] Check if running in headless environment
+        is_headless = is_headless_environment()
+
+        auth_code_future = asyncio.get_event_loop().create_future()
+        server = None
+
+        async def handle_callback(reader, writer):
+            try:
+                request_line_bytes = await reader.readline()
+                if not request_line_bytes:
+                    return
+                path_str = request_line_bytes.decode("utf-8").strip().split(" ")[1]
+                while await reader.readline() != b"\r\n":
+                    pass
+                from urllib.parse import urlparse, parse_qs
+
+                query_params = parse_qs(urlparse(path_str).query)
+                writer.write(b"HTTP/1.1 200 OK\r\nContent-Type: text/html\r\n\r\n")
+                if "code" in query_params:
+                    if not auth_code_future.done():
+                        auth_code_future.set_result(query_params["code"][0])
+                    writer.write(
+                        b"<html><body><h1>Authentication successful!</h1><p>You can close this window.</p></body></html>"
+                    )
+                else:
+                    error = query_params.get("error", ["Unknown error"])[0]
+                    if not auth_code_future.done():
+                        auth_code_future.set_exception(
+                            Exception(f"OAuth failed: {error}")
+                        )
+                    writer.write(
+                        f"<html><body><h1>Authentication Failed</h1><p>Error: {error}. Please try again.</p></body></html>".encode()
+                    )
+                await writer.drain()
+            except Exception as e:
+                lib_logger.error(f"Error in OAuth callback handler: {e}")
+            finally:
+                writer.close()
+
+        try:
+            server = await asyncio.start_server(
+                handle_callback, "127.0.0.1", self.CALLBACK_PORT
+            )
+            from urllib.parse import urlencode
+
+            auth_url = "https://accounts.google.com/o/oauth2/v2/auth?" + urlencode(
+                {
+                    "client_id": self.CLIENT_ID,
+                    "redirect_uri": f"http://localhost:{self.CALLBACK_PORT}{self.CALLBACK_PATH}",
+                    "scope": " ".join(self.OAUTH_SCOPES),
+                    "access_type": "offline",
+                    "response_type": "code",
+                    "prompt": "consent",
+                }
+            )
+
+            # [HEADLESS SUPPORT] Display appropriate instructions
+            if is_headless:
+                auth_panel_text = Text.from_markup(
+                    "Running in headless environment (no GUI detected).\n"
+                    "Please open the URL below in a browser on another machine to authorize:\n"
+                )
+            else:
+                auth_panel_text = Text.from_markup(
+                    "1. Your browser will now open to log in and authorize the application.\n"
+                    "2. If it doesn't open automatically, please open the URL below manually."
+                )
+
+            console.print(
+                Panel(
+                    auth_panel_text,
+                    title=f"{self.ENV_PREFIX} OAuth Setup for [bold yellow]{display_name}[/bold yellow]",
+                    style="bold blue",
+                )
+            )
+            # [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")
+                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",
+            ):
+                # Note: The 300s timeout here is handled by the ReauthCoordinator
+                # We use a slightly longer internal timeout to let the coordinator handle it
+                auth_code = await asyncio.wait_for(auth_code_future, timeout=310)
+        except asyncio.TimeoutError:
+            raise Exception("OAuth flow timed out. Please try again.")
+        finally:
+            if server:
+                server.close()
+                await server.wait_closed()
+
+        lib_logger.info(f"Attempting to exchange authorization code for tokens...")
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                self.TOKEN_URI,
+                data={
+                    "code": auth_code.strip(),
+                    "client_id": self.CLIENT_ID,
+                    "client_secret": self.CLIENT_SECRET,
+                    "redirect_uri": f"http://localhost:{self.CALLBACK_PORT}{self.CALLBACK_PATH}",
+                    "grant_type": "authorization_code",
+                },
+            )
+            response.raise_for_status()
+            token_data = response.json()
+            # Start with the full token data from the exchange
+            new_creds = token_data.copy()
+
+            # Convert 'expires_in' to 'expiry_date' in milliseconds
+            new_creds["expiry_date"] = (
+                time.time() + new_creds.pop("expires_in")
+            ) * 1000
+
+            # Ensure client_id and client_secret are present
+            new_creds["client_id"] = self.CLIENT_ID
+            new_creds["client_secret"] = self.CLIENT_SECRET
+
+            new_creds["token_uri"] = self.TOKEN_URI
+            new_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 {new_creds['access_token']}"},
+            )
+            user_info_response.raise_for_status()
+            user_info = user_info_response.json()
+            new_creds["_proxy_metadata"] = {
+                "email": user_info.get("email"),
+                "last_check_timestamp": time.time(),
+            }
+
+            if path:
+                await self._save_credentials(path, new_creds)
+            lib_logger.info(
+                f"{self.ENV_PREFIX} OAuth initialized successfully for '{display_name}'."
+            )
+        return new_creds
 
     async def initialize_token(
         self, creds_or_path: Union[Dict[str, Any], str]
     ) -> Dict[str, Any]:
+        """
+        Initialize OAuth token, triggering interactive OAuth flow if needed.
+
+        If interactive OAuth is required (expired refresh token, missing credentials, etc.),
+        the flow is coordinated globally via ReauthCoordinator to ensure only one
+        interactive OAuth flow runs at a time across all providers.
+        """
         path = creds_or_path if isinstance(creds_or_path, str) else None
 
         # Get display name from metadata if available, otherwise derive from path
@@ -655,181 +912,23 @@ class GoogleOAuthBase:
                     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
+                # [GLOBAL REAUTH COORDINATION] Use the global coordinator to ensure
+                # only one interactive OAuth flow runs at a time across all providers
+                coordinator = get_reauth_coordinator()
 
-                async def handle_callback(reader, writer):
-                    try:
-                        request_line_bytes = await reader.readline()
-                        if not request_line_bytes:
-                            return
-                        path_str = (
-                            request_line_bytes.decode("utf-8").strip().split(" ")[1]
-                        )
-                        while await reader.readline() != b"\r\n":
-                            pass
-                        from urllib.parse import urlparse, parse_qs
-
-                        query_params = parse_qs(urlparse(path_str).query)
-                        writer.write(
-                            b"HTTP/1.1 200 OK\r\nContent-Type: text/html\r\n\r\n"
-                        )
-                        if "code" in query_params:
-                            if not auth_code_future.done():
-                                auth_code_future.set_result(query_params["code"][0])
-                            writer.write(
-                                b"<html><body><h1>Authentication successful!</h1><p>You can close this window.</p></body></html>"
-                            )
-                        else:
-                            error = query_params.get("error", ["Unknown error"])[0]
-                            if not auth_code_future.done():
-                                auth_code_future.set_exception(
-                                    Exception(f"OAuth failed: {error}")
-                                )
-                            writer.write(
-                                f"<html><body><h1>Authentication Failed</h1><p>Error: {error}. Please try again.</p></body></html>".encode()
-                            )
-                        await writer.drain()
-                    except Exception as e:
-                        lib_logger.error(f"Error in OAuth callback handler: {e}")
-                    finally:
-                        writer.close()
-
-                try:
-                    server = await asyncio.start_server(
-                        handle_callback, "127.0.0.1", self.CALLBACK_PORT
-                    )
-                    from urllib.parse import urlencode
-
-                    auth_url = (
-                        "https://accounts.google.com/o/oauth2/v2/auth?"
-                        + urlencode(
-                            {
-                                "client_id": self.CLIENT_ID,
-                                "redirect_uri": f"http://localhost:{self.CALLBACK_PORT}{self.CALLBACK_PATH}",
-                                "scope": " ".join(self.OAUTH_SCOPES),
-                                "access_type": "offline",
-                                "response_type": "code",
-                                "prompt": "consent",
-                            }
-                        )
+                # Define the interactive OAuth function to be executed by coordinator
+                async def _do_interactive_oauth():
+                    return await self._perform_interactive_oauth(
+                        path, creds, display_name
                     )
 
-                    # [HEADLESS SUPPORT] Display appropriate instructions
-                    if is_headless:
-                        auth_panel_text = Text.from_markup(
-                            "Running in headless environment (no GUI detected).\n"
-                            "Please open the URL below in a browser on another machine to authorize:\n"
-                        )
-                    else:
-                        auth_panel_text = Text.from_markup(
-                            "1. Your browser will now open to log in and authorize the application.\n"
-                            "2. If it doesn't open automatically, please open the URL below manually."
-                        )
-
-                    console.print(
-                        Panel(
-                            auth_panel_text,
-                            title=f"{self.ENV_PREFIX} OAuth Setup for [bold yellow]{display_name}[/bold yellow]",
-                            style="bold blue",
-                        )
-                    )
-                    # [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"
-                            )
-                        except Exception as e:
-                            lib_logger.warning(
-                                f"Failed to open browser automatically: {e}. Please open the URL manually."
-                            )
-
-                    with console.status(
-                        f"[bold green]Waiting for you to complete authentication in the browser...[/bold green]",
-                        spinner="dots",
-                    ):
-                        auth_code = await asyncio.wait_for(
-                            auth_code_future, timeout=300
-                        )
-                except asyncio.TimeoutError:
-                    raise Exception("OAuth flow timed out. Please try again.")
-                finally:
-                    if server:
-                        server.close()
-                        await server.wait_closed()
-
-                lib_logger.info(
-                    f"Attempting to exchange authorization code for tokens..."
+                # Execute via global coordinator (ensures only one at a time)
+                return await coordinator.execute_reauth(
+                    credential_path=path or display_name,
+                    provider_name=self.ENV_PREFIX,
+                    reauth_func=_do_interactive_oauth,
+                    timeout=300.0,  # 5 minute timeout for user to complete OAuth
                 )
-                async with httpx.AsyncClient() as client:
-                    response = await client.post(
-                        self.TOKEN_URI,
-                        data={
-                            "code": auth_code.strip(),
-                            "client_id": self.CLIENT_ID,
-                            "client_secret": self.CLIENT_SECRET,
-                            "redirect_uri": f"http://localhost:{self.CALLBACK_PORT}{self.CALLBACK_PATH}",
-                            "grant_type": "authorization_code",
-                        },
-                    )
-                    response.raise_for_status()
-                    token_data = response.json()
-                    # Start with the full token data from the exchange
-                    creds = token_data.copy()
-
-                    # Convert 'expires_in' to 'expiry_date' in milliseconds
-                    creds["expiry_date"] = (
-                        time.time() + creds.pop("expires_in")
-                    ) * 1000
-
-                    # Ensure client_id and client_secret are present
-                    creds["client_id"] = self.CLIENT_ID
-                    creds["client_secret"] = self.CLIENT_SECRET
-
-                    creds["token_uri"] = self.TOKEN_URI
-                    creds["universe_domain"] = "googleapis.com"
-
-                    # Fetch user info and add metadata
-                    user_info_response = await client.get(
-                        self.USER_INFO_URI,
-                        headers={"Authorization": f"Bearer {creds['access_token']}"},
-                    )
-                    user_info_response.raise_for_status()
-                    user_info = user_info_response.json()
-                    creds["_proxy_metadata"] = {
-                        "email": user_info.get("email"),
-                        "last_check_timestamp": time.time(),
-                    }
-
-                    if path:
-                        await self._save_credentials(path, creds)
-                    lib_logger.info(
-                        f"{self.ENV_PREFIX} OAuth initialized successfully for '{display_name}'."
-                    )
-                return creds
 
             lib_logger.info(
                 f"{self.ENV_PREFIX} OAuth token at '{display_name}' is valid."

src/rotator_library/providers/iflow_auth_base.py

@@ -23,6 +23,7 @@ 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
+from ..utils.reauth_coordinator import get_reauth_coordinator
 
 lib_logger = logging.getLogger("rotator_library")
 
@@ -173,9 +174,12 @@ class IFlowAuthBase:
         # [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
+        # [FIX PR#34] Changed from set to dict mapping credential path to timestamp
+        # This enables TTL-based stale entry cleanup as defense in depth
+        self._unavailable_credentials: Dict[
+            str, float
+        ] = {}  # Maps credential path -> timestamp when marked unavailable
+        self._unavailable_ttl_seconds: int = 300  # 5 minutes TTL for stale entries
         self._queue_tracking_lock = asyncio.Lock()  # Protects queue sets
         self._queue_processor_task: Optional[asyncio.Task] = (
             None  # Background worker task
@@ -745,15 +749,28 @@ class IFlowAuthBase:
         Proactively refreshes tokens if they're close to expiry.
         Only applies to OAuth credentials (file paths or env:// paths). Direct API keys are skipped.
         """
-        # Check if it's an env:// virtual path (OAuth credentials from environment)
-        is_env_path = credential_identifier.startswith("env://")
+        lib_logger.debug(f"proactively_refresh called for: {credential_identifier}")
 
-        # 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
+        # Try to load credentials - this will fail for direct API keys
+        # and succeed for OAuth credentials (file paths or env:// paths)
+        try:
+            creds = await self._load_credentials(credential_identifier)
+        except IOError as e:
+            # Not a valid credential path (likely a direct API key string)
+            lib_logger.debug(
+                f"Skipping refresh for '{credential_identifier}' - not an OAuth credential: {e}"
+            )
+            return
 
-        creds = await self._load_credentials(credential_identifier)
-        if self._is_token_expired(creds):
+        is_expired = self._is_token_expired(creds)
+        lib_logger.debug(
+            f"Token expired check for '{Path(credential_identifier).name}': {is_expired}"
+        )
+
+        if is_expired:
+            lib_logger.debug(
+                f"Queueing refresh for '{Path(credential_identifier).name}'"
+            )
             # Queue for refresh with needs_reauth=False (automated refresh)
             await self._queue_refresh(
                 credential_identifier, force=False, needs_reauth=False
@@ -768,8 +785,30 @@ class IFlowAuthBase:
             return self._refresh_locks[path]
 
     def is_credential_available(self, path: str) -> bool:
-        """Check if a credential is available for rotation (not queued/refreshing)."""
-        return path not in self._unavailable_credentials
+        """Check if a credential is available for rotation (not queued/refreshing).
+
+        [FIX PR#34] Now includes TTL-based stale entry cleanup as defense in depth.
+        If a credential has been unavailable for longer than _unavailable_ttl_seconds,
+        it is automatically cleaned up and considered available.
+        """
+        if path not in self._unavailable_credentials:
+            return True
+
+        # [FIX PR#34] Check if the entry is stale (TTL expired)
+        marked_time = self._unavailable_credentials.get(path)
+        if marked_time is not None:
+            now = time.time()
+            if now - marked_time > self._unavailable_ttl_seconds:
+                # Entry is stale - clean it up and return available
+                lib_logger.warning(
+                    f"Credential '{Path(path).name}' was stuck in unavailable state for "
+                    f"{int(now - marked_time)}s (TTL: {self._unavailable_ttl_seconds}s). "
+                    f"Auto-cleaning stale entry."
+                )
+                self._unavailable_credentials.pop(path, None)
+                return True
+
+        return False
 
     async def _ensure_queue_processor_running(self):
         """Lazily starts the queue processor if not already running."""
@@ -805,7 +844,12 @@ class IFlowAuthBase:
         async with self._queue_tracking_lock:
             if path not in self._queued_credentials:
                 self._queued_credentials.add(path)
-                self._unavailable_credentials.add(path)  # Mark as unavailable
+                # [FIX PR#34] Store timestamp when marking unavailable (for TTL cleanup)
+                self._unavailable_credentials[path] = time.time()
+                lib_logger.debug(
+                    f"Marked '{Path(path).name}' as unavailable. "
+                    f"Total unavailable: {len(self._unavailable_credentials)}"
+                )
                 await self._refresh_queue.put((path, force, needs_reauth))
                 await self._ensure_queue_processor_running()
 
@@ -820,7 +864,22 @@ class IFlowAuthBase:
                         self._refresh_queue.get(), timeout=60.0
                     )
                 except asyncio.TimeoutError:
-                    # No items for 60s, exit to save resources
+                    # [FIX PR#34] Clean up any stale unavailable entries before exiting
+                    # If we're idle for 60s, no refreshes are in progress
+                    async with self._queue_tracking_lock:
+                        if self._unavailable_credentials:
+                            stale_count = len(self._unavailable_credentials)
+                            lib_logger.warning(
+                                f"Queue processor idle timeout. Cleaning {stale_count} "
+                                f"stale unavailable credentials: {list(self._unavailable_credentials.keys())}"
+                            )
+                            self._unavailable_credentials.clear()
+                        # [FIX BUG#6] Also clear queued credentials to prevent stuck state
+                        if self._queued_credentials:
+                            lib_logger.debug(
+                                f"Clearing {len(self._queued_credentials)} queued credentials on timeout"
+                            )
+                            self._queued_credentials.clear()
                     self._queue_processor_task = None
                     return
 
@@ -832,7 +891,11 @@ class IFlowAuthBase:
                         if creds and not self._is_token_expired(creds):
                             # No longer expired, mark as available
                             async with self._queue_tracking_lock:
-                                self._unavailable_credentials.discard(path)
+                                self._unavailable_credentials.pop(path, None)
+                                lib_logger.debug(
+                                    f"Credential '{Path(path).name}' no longer expired, marked available. "
+                                    f"Remaining unavailable: {len(self._unavailable_credentials)}"
+                                )
                             continue
 
                         # Perform refresh
@@ -842,28 +905,174 @@ class IFlowAuthBase:
 
                         # SUCCESS: Mark as available again
                         async with self._queue_tracking_lock:
-                            self._unavailable_credentials.discard(path)
+                            self._unavailable_credentials.pop(path, None)
+                            lib_logger.debug(
+                                f"Refresh SUCCESS for '{Path(path).name}', marked available. "
+                                f"Remaining unavailable: {len(self._unavailable_credentials)}"
+                            )
 
                 finally:
-                    # Remove from queued set
+                    # [FIX PR#34] Remove from BOTH queued set AND unavailable credentials
+                    # This ensures cleanup happens in ALL exit paths (success, exception, etc.)
                     async with self._queue_tracking_lock:
                         self._queued_credentials.discard(path)
+                        # [FIX PR#34] Always clean up unavailable credentials in finally block
+                        self._unavailable_credentials.pop(path, None)
+                        lib_logger.debug(
+                            f"Finally cleanup for '{Path(path).name}'. "
+                            f"Remaining unavailable: {len(self._unavailable_credentials)}"
+                        )
                     self._refresh_queue.task_done()
             except asyncio.CancelledError:
+                # [FIX PR#34] Clean up the current credential before breaking
+                if path:
+                    async with self._queue_tracking_lock:
+                        self._unavailable_credentials.pop(path, None)
+                        lib_logger.debug(
+                            f"CancelledError cleanup for '{Path(path).name}'. "
+                            f"Remaining unavailable: {len(self._unavailable_credentials)}"
+                        )
                 break
             except Exception as e:
                 lib_logger.error(f"Error in queue processor: {e}")
                 # Even on error, mark as available (backoff will prevent immediate retry)
                 if path:
                     async with self._queue_tracking_lock:
-                        self._unavailable_credentials.discard(path)
+                        self._unavailable_credentials.pop(path, None)
+                        lib_logger.debug(
+                            f"Error cleanup for '{Path(path).name}': {e}. "
+                            f"Remaining unavailable: {len(self._unavailable_credentials)}"
+                        )
+
+    async def _perform_interactive_oauth(
+        self, path: str, creds: Dict[str, Any], display_name: str
+    ) -> Dict[str, Any]:
+        """
+        Perform interactive OAuth authorization code flow (browser-based authentication).
+
+        This method is called via the global ReauthCoordinator to ensure
+        only one interactive OAuth flow runs at a time across all providers.
+
+        Args:
+            path: Credential file path
+            creds: Current credentials dict (will be updated)
+            display_name: Display name for logging/UI
+
+        Returns:
+            Updated credentials dict with new tokens
+        """
+        # [HEADLESS DETECTION] Check if running in headless environment
+        is_headless = is_headless_environment()
+
+        # Generate random state for CSRF protection
+        state = secrets.token_urlsafe(32)
+
+        # Build authorization URL
+        redirect_uri = f"http://localhost:{CALLBACK_PORT}/oauth2callback"
+        auth_params = {
+            "loginMethod": "phone",
+            "type": "phone",
+            "redirect": redirect_uri,
+            "state": state,
+            "client_id": IFLOW_CLIENT_ID,
+        }
+        auth_url = f"{IFLOW_OAUTH_AUTHORIZE_ENDPOINT}?{urlencode(auth_params)}"
+
+        # Start OAuth callback server
+        callback_server = OAuthCallbackServer(port=CALLBACK_PORT)
+        try:
+            await callback_server.start(expected_state=state)
+
+            # [HEADLESS SUPPORT] Display appropriate instructions
+            if is_headless:
+                auth_panel_text = Text.from_markup(
+                    "Running in headless environment (no GUI detected).\n"
+                    "Please open the URL below in a browser on another machine to authorize:\n"
+                    "1. Visit the URL below to sign in with your phone number.\n"
+                    "2. [bold]Authorize the application[/bold] to access your account.\n"
+                    "3. You will be automatically redirected after authorization."
+                )
+            else:
+                auth_panel_text = Text.from_markup(
+                    "1. Visit the URL below to sign in with your phone number.\n"
+                    "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",
+                )
+            )
+            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")
+                except Exception as e:
+                    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",
+            ):
+                # Note: The 300s timeout here is handled by the ReauthCoordinator
+                # We use a slightly longer internal timeout to let the coordinator handle it
+                code = await callback_server.wait_for_callback(timeout=310.0)
+
+            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)
+
+            # 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"],
+                }
+            )
+
+            # Create metadata object
+            if not creds.get("_proxy_metadata"):
+                creds["_proxy_metadata"] = {
+                    "email": token_data["email"],
+                    "last_check_timestamp": time.time(),
+                }
+
+            if path:
+                await self._save_credentials(path, creds)
+
+            lib_logger.info(
+                f"iFlow OAuth initialized successfully for '{display_name}'."
+            )
+            return creds
+
+        finally:
+            await callback_server.stop()
 
     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.
+        Initialize OAuth token, triggering interactive authorization flow if needed.
+
+        If interactive OAuth is required (expired refresh token, missing credentials, etc.),
+        the flow is coordinated globally via ReauthCoordinator to ensure only one
+        interactive OAuth flow runs at a time across all providers.
         """
         path = creds_or_path if isinstance(creds_or_path, str) else None
 
@@ -903,127 +1112,23 @@ class IFlowAuthBase:
                     f"iFlow OAuth token for '{display_name}' needs setup: {reason}."
                 )
 
-                # [HEADLESS DETECTION] Check if running in headless environment
-                is_headless = is_headless_environment()
-
-                # Generate random state for CSRF protection
-                state = secrets.token_urlsafe(32)
-
-                # Build authorization URL
-                redirect_uri = f"http://localhost:{CALLBACK_PORT}/oauth2callback"
-                auth_params = {
-                    "loginMethod": "phone",
-                    "type": "phone",
-                    "redirect": redirect_uri,
-                    "state": state,
-                    "client_id": IFLOW_CLIENT_ID,
-                }
-                auth_url = f"{IFLOW_OAUTH_AUTHORIZE_ENDPOINT}?{urlencode(auth_params)}"
-
-                # Start OAuth callback server
-                callback_server = OAuthCallbackServer(port=CALLBACK_PORT)
-                try:
-                    await callback_server.start(expected_state=state)
-
-                    # [HEADLESS SUPPORT] Display appropriate instructions
-                    if is_headless:
-                        auth_panel_text = Text.from_markup(
-                            "Running in headless environment (no GUI detected).\n"
-                            "Please open the URL below in a browser on another machine to authorize:\n"
-                            "1. Visit the URL below to sign in with your phone number.\n"
-                            "2. [bold]Authorize the application[/bold] to access your account.\n"
-                            "3. You will be automatically redirected after authorization."
-                        )
-                    else:
-                        auth_panel_text = Text.from_markup(
-                            "1. Visit the URL below to sign in with your phone number.\n"
-                            "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",
-                        )
-                    )
-                    # [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"
-                            )
-                        except Exception as e:
-                            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",
-                    ):
-                        code = await callback_server.wait_for_callback(timeout=300.0)
+                # [GLOBAL REAUTH COORDINATION] Use the global coordinator to ensure
+                # only one interactive OAuth flow runs at a time across all providers
+                coordinator = get_reauth_coordinator()
 
-                    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
-                    )
-
-                    # 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"],
-                        }
-                    )
-
-                    # Create metadata object
-                    if not creds.get("_proxy_metadata"):
-                        creds["_proxy_metadata"] = {
-                            "email": token_data["email"],
-                            "last_check_timestamp": time.time(),
-                        }
-
-                    if path:
-                        await self._save_credentials(path, creds)
-
-                    lib_logger.info(
-                        f"iFlow OAuth initialized successfully for '{display_name}'."
+                # Define the interactive OAuth function to be executed by coordinator
+                async def _do_interactive_oauth():
+                    return await self._perform_interactive_oauth(
+                        path, creds, display_name
                     )
-                    return creds
 
-                finally:
-                    await callback_server.stop()
+                # Execute via global coordinator (ensures only one at a time)
+                return await coordinator.execute_reauth(
+                    credential_path=path or display_name,
+                    provider_name="IFLOW",
+                    reauth_func=_do_interactive_oauth,
+                    timeout=300.0,  # 5 minute timeout for user to complete OAuth
+                )
 
             lib_logger.info(f"iFlow OAuth token at '{display_name}' is valid.")
             return creds

src/rotator_library/providers/qwen_auth_base.py

@@ -22,6 +22,7 @@ from rich.text import Text
 from rich.markup import escape as rich_escape
 
 from ..utils.headless_detection import is_headless_environment
+from ..utils.reauth_coordinator import get_reauth_coordinator
 
 lib_logger = logging.getLogger("rotator_library")
 
@@ -53,9 +54,12 @@ class QwenAuthBase:
         # [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
+        # [FIX PR#34] Changed from set to dict mapping credential path to timestamp
+        # This enables TTL-based stale entry cleanup as defense in depth
+        self._unavailable_credentials: Dict[
+            str, float
+        ] = {}  # Maps credential path -> timestamp when marked unavailable
+        self._unavailable_ttl_seconds: int = 300  # 5 minutes TTL for stale entries
         self._queue_tracking_lock = asyncio.Lock()  # Protects queue sets
         self._queue_processor_task: Optional[asyncio.Task] = (
             None  # Background worker task
@@ -472,15 +476,28 @@ class QwenAuthBase:
         Proactively refreshes tokens if they're close to expiry.
         Only applies to OAuth credentials (file paths or env:// paths). Direct API keys are skipped.
         """
-        # Check if it's an env:// virtual path (OAuth credentials from environment)
-        is_env_path = credential_identifier.startswith("env://")
+        lib_logger.debug(f"proactively_refresh called for: {credential_identifier}")
 
-        # 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
+        # Try to load credentials - this will fail for direct API keys
+        # and succeed for OAuth credentials (file paths or env:// paths)
+        try:
+            creds = await self._load_credentials(credential_identifier)
+        except IOError as e:
+            # Not a valid credential path (likely a direct API key string)
+            lib_logger.debug(
+                f"Skipping refresh for '{credential_identifier}' - not an OAuth credential: {e}"
+            )
+            return
 
-        creds = await self._load_credentials(credential_identifier)
-        if self._is_token_expired(creds):
+        is_expired = self._is_token_expired(creds)
+        lib_logger.debug(
+            f"Token expired check for '{Path(credential_identifier).name}': {is_expired}"
+        )
+
+        if is_expired:
+            lib_logger.debug(
+                f"Queueing refresh for '{Path(credential_identifier).name}'"
+            )
             # Queue for refresh with needs_reauth=False (automated refresh)
             await self._queue_refresh(
                 credential_identifier, force=False, needs_reauth=False
@@ -494,8 +511,30 @@ class QwenAuthBase:
             return self._refresh_locks[path]
 
     def is_credential_available(self, path: str) -> bool:
-        """Check if a credential is available for rotation (not queued/refreshing)."""
-        return path not in self._unavailable_credentials
+        """Check if a credential is available for rotation (not queued/refreshing).
+
+        [FIX PR#34] Now includes TTL-based stale entry cleanup as defense in depth.
+        If a credential has been unavailable for longer than _unavailable_ttl_seconds,
+        it is automatically cleaned up and considered available.
+        """
+        if path not in self._unavailable_credentials:
+            return True
+
+        # [FIX PR#34] Check if the entry is stale (TTL expired)
+        marked_time = self._unavailable_credentials.get(path)
+        if marked_time is not None:
+            now = time.time()
+            if now - marked_time > self._unavailable_ttl_seconds:
+                # Entry is stale - clean it up and return available
+                lib_logger.warning(
+                    f"Credential '{Path(path).name}' was stuck in unavailable state for "
+                    f"{int(now - marked_time)}s (TTL: {self._unavailable_ttl_seconds}s). "
+                    f"Auto-cleaning stale entry."
+                )
+                self._unavailable_credentials.pop(path, None)
+                return True
+
+        return False
 
     async def _ensure_queue_processor_running(self):
         """Lazily starts the queue processor if not already running."""
@@ -531,7 +570,12 @@ class QwenAuthBase:
         async with self._queue_tracking_lock:
             if path not in self._queued_credentials:
                 self._queued_credentials.add(path)
-                self._unavailable_credentials.add(path)  # Mark as unavailable
+                # [FIX PR#34] Store timestamp when marking unavailable (for TTL cleanup)
+                self._unavailable_credentials[path] = time.time()
+                lib_logger.debug(
+                    f"Marked '{Path(path).name}' as unavailable. "
+                    f"Total unavailable: {len(self._unavailable_credentials)}"
+                )
                 await self._refresh_queue.put((path, force, needs_reauth))
                 await self._ensure_queue_processor_running()
 
@@ -546,7 +590,22 @@ class QwenAuthBase:
                         self._refresh_queue.get(), timeout=60.0
                     )
                 except asyncio.TimeoutError:
-                    # No items for 60s, exit to save resources
+                    # [FIX PR#34] Clean up any stale unavailable entries before exiting
+                    # If we're idle for 60s, no refreshes are in progress
+                    async with self._queue_tracking_lock:
+                        if self._unavailable_credentials:
+                            stale_count = len(self._unavailable_credentials)
+                            lib_logger.warning(
+                                f"Queue processor idle timeout. Cleaning {stale_count} "
+                                f"stale unavailable credentials: {list(self._unavailable_credentials.keys())}"
+                            )
+                            self._unavailable_credentials.clear()
+                        # [FIX BUG#6] Also clear queued credentials to prevent stuck state
+                        if self._queued_credentials:
+                            lib_logger.debug(
+                                f"Clearing {len(self._queued_credentials)} queued credentials on timeout"
+                            )
+                            self._queued_credentials.clear()
                     self._queue_processor_task = None
                     return
 
@@ -558,7 +617,11 @@ class QwenAuthBase:
                         if creds and not self._is_token_expired(creds):
                             # No longer expired, mark as available
                             async with self._queue_tracking_lock:
-                                self._unavailable_credentials.discard(path)
+                                self._unavailable_credentials.pop(path, None)
+                                lib_logger.debug(
+                                    f"Credential '{Path(path).name}' no longer expired, marked available. "
+                                    f"Remaining unavailable: {len(self._unavailable_credentials)}"
+                                )
                             continue
 
                         # Perform refresh
@@ -568,26 +631,240 @@ class QwenAuthBase:
 
                         # SUCCESS: Mark as available again
                         async with self._queue_tracking_lock:
-                            self._unavailable_credentials.discard(path)
+                            self._unavailable_credentials.pop(path, None)
+                            lib_logger.debug(
+                                f"Refresh SUCCESS for '{Path(path).name}', marked available. "
+                                f"Remaining unavailable: {len(self._unavailable_credentials)}"
+                            )
 
                 finally:
-                    # Remove from queued set
+                    # [FIX PR#34] Remove from BOTH queued set AND unavailable credentials
+                    # This ensures cleanup happens in ALL exit paths (success, exception, etc.)
                     async with self._queue_tracking_lock:
                         self._queued_credentials.discard(path)
+                        # [FIX PR#34] Always clean up unavailable credentials in finally block
+                        self._unavailable_credentials.pop(path, None)
+                        lib_logger.debug(
+                            f"Finally cleanup for '{Path(path).name}'. "
+                            f"Remaining unavailable: {len(self._unavailable_credentials)}"
+                        )
                     self._refresh_queue.task_done()
             except asyncio.CancelledError:
+                # [FIX PR#34] Clean up the current credential before breaking
+                if path:
+                    async with self._queue_tracking_lock:
+                        self._unavailable_credentials.pop(path, None)
+                        lib_logger.debug(
+                            f"CancelledError cleanup for '{Path(path).name}'. "
+                            f"Remaining unavailable: {len(self._unavailable_credentials)}"
+                        )
                 break
             except Exception as e:
                 lib_logger.error(f"Error in queue processor: {e}")
                 # Even on error, mark as available (backoff will prevent immediate retry)
                 if path:
                     async with self._queue_tracking_lock:
-                        self._unavailable_credentials.discard(path)
+                        self._unavailable_credentials.pop(path, None)
+                        lib_logger.debug(
+                            f"Error cleanup for '{Path(path).name}': {e}. "
+                            f"Remaining unavailable: {len(self._unavailable_credentials)}"
+                        )
+
+    async def _perform_interactive_oauth(
+        self, path: str, creds: Dict[str, Any], display_name: str
+    ) -> Dict[str, Any]:
+        """
+        Perform interactive OAuth device flow (browser-based authentication).
+
+        This method is called via the global ReauthCoordinator to ensure
+        only one interactive OAuth flow runs at a time across all providers.
+
+        Args:
+            path: Credential file path
+            creds: Current credentials dict (will be updated)
+            display_name: Display name for logging/UI
+
+        Returns:
+            Updated credentials dict with new tokens
+        """
+        # [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("=")
+        )
+
+        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",
+        }
+        async with httpx.AsyncClient() as client:
+            request_data = {
+                "client_id": CLIENT_ID,
+                "scope": SCOPE,
+                "code_challenge": code_challenge,
+                "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,
+                )
+                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}"
+                )
+                raise e
+
+            # [HEADLESS SUPPORT] Display appropriate instructions
+            if is_headless:
+                auth_panel_text = Text.from_markup(
+                    "Running in headless environment (no GUI detected).\n"
+                    "Please open the URL below in a browser on another machine to authorize:\n"
+                    "1. Visit the URL below to sign in.\n"
+                    "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."
+                )
+            else:
+                auth_panel_text = Text.from_markup(
+                    "1. Visit the URL below to sign in.\n"
+                    "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",
+                )
+            )
+            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")
+                except Exception as e:
+                    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)
+
+            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"],
+                            "client_id": CLIENT_ID,
+                            "code_verifier": code_verifier,
+                        },
+                    )
+                    if poll_response.status_code == 200:
+                        token_data = poll_response.json()
+                        lib_logger.info("Successfully received token.")
+                        break
+                    elif poll_response.status_code == 400:
+                        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"
+                            )
+                        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"
+                            )
+                        else:
+                            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"),
+                }
+            )
+
+            # 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]"
+                    )
+                    email = Prompt.ask(prompt_text)
+                    creds["_proxy_metadata"] = {
+                        "email": email.strip(),
+                        "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(),
+                    }
+
+            if path:
+                await self._save_credentials(path, creds)
+            lib_logger.info(
+                f"Qwen OAuth initialized successfully for '{display_name}'."
+            )
+        return creds
 
     async def initialize_token(
         self, creds_or_path: Union[Dict[str, Any], str]
     ) -> Dict[str, Any]:
-        """Initiates device flow if tokens are missing or invalid."""
+        """
+        Initialize OAuth token, triggering interactive device flow if needed.
+
+        If interactive OAuth is required (expired refresh token, missing credentials, etc.),
+        the flow is coordinated globally via ReauthCoordinator to ensure only one
+        interactive OAuth flow runs at a time across all providers.
+        """
         path = creds_or_path if isinstance(creds_or_path, str) else None
 
         # Get display name from metadata if available, otherwise derive from path
@@ -623,189 +900,23 @@ class QwenAuthBase:
                     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("=")
-                )
-
-                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",
-                }
-                async with httpx.AsyncClient() as client:
-                    request_data = {
-                        "client_id": CLIENT_ID,
-                        "scope": SCOPE,
-                        "code_challenge": code_challenge,
-                        "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,
-                        )
-                        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}"
-                        )
-                        raise e
-
-                    # [HEADLESS SUPPORT] Display appropriate instructions
-                    if is_headless:
-                        auth_panel_text = Text.from_markup(
-                            "Running in headless environment (no GUI detected).\n"
-                            "Please open the URL below in a browser on another machine to authorize:\n"
-                            "1. Visit the URL below to sign in.\n"
-                            "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."
-                        )
-                    else:
-                        auth_panel_text = Text.from_markup(
-                            "1. Visit the URL below to sign in.\n"
-                            "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",
-                        )
-                    )
-                    # [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"
-                            )
-                        except Exception as e:
-                            lib_logger.warning(
-                                f"Failed to open browser automatically: {e}. Please open the URL manually."
-                            )
+                # [GLOBAL REAUTH COORDINATION] Use the global coordinator to ensure
+                # only one interactive OAuth flow runs at a time across all providers
+                coordinator = get_reauth_coordinator()
 
-                    token_data = None
-                    start_time = time.time()
-                    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"]:
-                            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"],
-                                    "client_id": CLIENT_ID,
-                                    "code_verifier": code_verifier,
-                                },
-                            )
-                            if poll_response.status_code == 200:
-                                token_data = poll_response.json()
-                                lib_logger.info("Successfully received token.")
-                                break
-                            elif poll_response.status_code == 400:
-                                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"
-                                    )
-                                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"
-                                    )
-                                else:
-                                    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"),
-                        }
+                # Define the interactive OAuth function to be executed by coordinator
+                async def _do_interactive_oauth():
+                    return await self._perform_interactive_oauth(
+                        path, creds, display_name
                     )
 
-                    # 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]"
-                            )
-                            email = Prompt.ask(prompt_text)
-                            creds["_proxy_metadata"] = {
-                                "email": email.strip(),
-                                "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(),
-                            }
-
-                    if path:
-                        await self._save_credentials(path, creds)
-                    lib_logger.info(
-                        f"Qwen OAuth initialized successfully for '{display_name}'."
-                    )
-                return creds
+                # Execute via global coordinator (ensures only one at a time)
+                return await coordinator.execute_reauth(
+                    credential_path=path or display_name,
+                    provider_name="QWEN_CODE",
+                    reauth_func=_do_interactive_oauth,
+                    timeout=300.0,  # 5 minute timeout for user to complete OAuth
+                )
 
             lib_logger.info(f"Qwen OAuth token at '{display_name}' is valid.")
             return creds

src/rotator_library/utils/__init__.py

@@ -1,5 +1,6 @@
 # src/rotator_library/utils/__init__.py
 
 from .headless_detection import is_headless_environment
+from .reauth_coordinator import get_reauth_coordinator, ReauthCoordinator
 
-__all__ = ['is_headless_environment']
+__all__ = ["is_headless_environment", "get_reauth_coordinator", "ReauthCoordinator"]

src/rotator_library/utils/reauth_coordinator.py

@@ -0,0 +1,236 @@
+# src/rotator_library/utils/reauth_coordinator.py
+
+"""
+Global Re-authentication Coordinator
+
+Ensures only ONE interactive OAuth flow runs at a time across ALL providers.
+This prevents port conflicts and user confusion when multiple credentials
+need re-authentication simultaneously.
+
+When a credential needs interactive re-auth (expired refresh token, revoked, etc.),
+it queues a request here. The coordinator ensures only one re-auth happens at a time,
+regardless of which provider the credential belongs to.
+"""
+
+import asyncio
+import logging
+import time
+from typing import Callable, Optional, Dict, Any, Awaitable
+from pathlib import Path
+
+lib_logger = logging.getLogger("rotator_library")
+
+
+class ReauthCoordinator:
+    """
+    Singleton coordinator for global re-authentication serialization.
+
+    When a credential needs interactive re-auth (expired refresh token, revoked, etc.),
+    it queues a request here. The coordinator ensures only one re-auth happens at a time.
+
+    This is critical because:
+    1. Different providers may use the same callback ports
+    2. User can only complete one OAuth flow at a time
+    3. Prevents race conditions in credential state management
+    """
+
+    _instance: Optional["ReauthCoordinator"] = None
+    _initialized: bool = False  # Class-level declaration for Pylint
+
+    def __new__(cls):
+        # Singleton pattern - only one coordinator exists
+        if cls._instance is None:
+            cls._instance = super().__new__(cls)
+            cls._instance._initialized = False
+        return cls._instance
+
+    def __init__(self):
+        if self._initialized:
+            return
+
+        # Global semaphore - only 1 re-auth at a time
+        self._reauth_semaphore: asyncio.Semaphore = asyncio.Semaphore(1)
+
+        # Tracking for observability
+        self._pending_reauths: Dict[str, float] = {}  # credential -> queue_time
+        self._current_reauth: Optional[str] = None
+        self._current_provider: Optional[str] = None
+        self._reauth_start_time: Optional[float] = None
+
+        # Lock for tracking dict modifications
+        self._tracking_lock: asyncio.Lock = asyncio.Lock()
+
+        # Statistics
+        self._total_reauths: int = 0
+        self._successful_reauths: int = 0
+        self._failed_reauths: int = 0
+        self._timeout_reauths: int = 0
+
+        self._initialized = True
+        lib_logger.info("Global ReauthCoordinator initialized")
+
+    def _get_display_name(self, credential_path: str) -> str:
+        """Get a display-friendly name for a credential path."""
+        if credential_path.startswith("env://"):
+            return credential_path
+        return Path(credential_path).name
+
+    async def execute_reauth(
+        self,
+        credential_path: str,
+        provider_name: str,
+        reauth_func: Callable[[], Awaitable[Dict[str, Any]]],
+        timeout: float = 300.0,  # 5 minutes default timeout
+    ) -> Dict[str, Any]:
+        """
+        Execute a re-authentication function with global serialization.
+
+        Only one re-auth can run at a time across all providers.
+        Other requests wait in queue.
+
+        Args:
+            credential_path: Path/identifier of the credential needing re-auth
+            provider_name: Name of the provider (for logging)
+            reauth_func: Async function that performs the actual re-auth
+            timeout: Maximum time to wait for re-auth to complete
+
+        Returns:
+            The result from reauth_func (new credentials dict)
+
+        Raises:
+            TimeoutError: If re-auth doesn't complete within timeout
+            Exception: Any exception from reauth_func is re-raised
+        """
+        display_name = self._get_display_name(credential_path)
+
+        # Track that this credential is waiting
+        async with self._tracking_lock:
+            self._pending_reauths[credential_path] = time.time()
+            pending_count = len(self._pending_reauths)
+
+            # Log queue status
+            if self._current_reauth:
+                current_display = self._get_display_name(self._current_reauth)
+                lib_logger.info(
+                    f"[ReauthCoordinator] Credential '{display_name}' ({provider_name}) queued for re-auth. "
+                    f"Position in queue: {pending_count}. "
+                    f"Currently processing: '{current_display}' ({self._current_provider})"
+                )
+            else:
+                lib_logger.info(
+                    f"[ReauthCoordinator] Credential '{display_name}' ({provider_name}) requesting re-auth."
+                )
+
+        try:
+            # Acquire global semaphore - blocks until our turn
+            async with self._reauth_semaphore:
+                # Calculate how long we waited in queue
+                async with self._tracking_lock:
+                    queue_time = self._pending_reauths.pop(credential_path, time.time())
+                    wait_duration = time.time() - queue_time
+                    self._current_reauth = credential_path
+                    self._current_provider = provider_name
+                    self._reauth_start_time = time.time()
+                    self._total_reauths += 1
+
+                if wait_duration > 1.0:
+                    lib_logger.info(
+                        f"[ReauthCoordinator] Starting re-auth for '{display_name}' ({provider_name}) "
+                        f"after waiting {wait_duration:.1f}s in queue"
+                    )
+                else:
+                    lib_logger.info(
+                        f"[ReauthCoordinator] Starting re-auth for '{display_name}' ({provider_name})"
+                    )
+
+                try:
+                    # Execute the actual re-auth with timeout
+                    result = await asyncio.wait_for(reauth_func(), timeout=timeout)
+
+                    async with self._tracking_lock:
+                        self._successful_reauths += 1
+                        duration = time.time() - self._reauth_start_time
+
+                    lib_logger.info(
+                        f"[ReauthCoordinator] Re-auth SUCCESS for '{display_name}' ({provider_name}) "
+                        f"in {duration:.1f}s"
+                    )
+                    return result
+
+                except asyncio.TimeoutError:
+                    async with self._tracking_lock:
+                        self._failed_reauths += 1
+                        self._timeout_reauths += 1
+                    lib_logger.error(
+                        f"[ReauthCoordinator] Re-auth TIMEOUT for '{display_name}' ({provider_name}) "
+                        f"after {timeout}s. User did not complete OAuth flow in time."
+                    )
+                    raise TimeoutError(
+                        f"Re-authentication timed out after {timeout}s. "
+                        f"Please try again and complete the OAuth flow within the time limit."
+                    )
+
+                except Exception as e:
+                    async with self._tracking_lock:
+                        self._failed_reauths += 1
+                    lib_logger.error(
+                        f"[ReauthCoordinator] Re-auth FAILED for '{display_name}' ({provider_name}): {e}"
+                    )
+                    raise
+
+                finally:
+                    async with self._tracking_lock:
+                        self._current_reauth = None
+                        self._current_provider = None
+                        self._reauth_start_time = None
+
+                        # Log if there are still pending reauths
+                        if self._pending_reauths:
+                            lib_logger.info(
+                                f"[ReauthCoordinator] {len(self._pending_reauths)} credential(s) "
+                                f"still waiting for re-auth"
+                            )
+
+        finally:
+            # Ensure we're removed from pending even if something goes wrong
+            async with self._tracking_lock:
+                self._pending_reauths.pop(credential_path, None)
+
+    def is_reauth_in_progress(self) -> bool:
+        """Check if a re-auth is currently in progress."""
+        return self._current_reauth is not None
+
+    def get_pending_count(self) -> int:
+        """Get number of credentials waiting for re-auth."""
+        return len(self._pending_reauths)
+
+    def get_status(self) -> Dict[str, Any]:
+        """Get current coordinator status for debugging/monitoring."""
+        return {
+            "current_reauth": self._current_reauth,
+            "current_provider": self._current_provider,
+            "reauth_in_progress": self._current_reauth is not None,
+            "reauth_duration": (time.time() - self._reauth_start_time)
+            if self._reauth_start_time
+            else None,
+            "pending_count": len(self._pending_reauths),
+            "pending_credentials": list(self._pending_reauths.keys()),
+            "stats": {
+                "total": self._total_reauths,
+                "successful": self._successful_reauths,
+                "failed": self._failed_reauths,
+                "timeouts": self._timeout_reauths,
+            },
+        }
+
+
+# Global singleton instance
+_coordinator: Optional[ReauthCoordinator] = None
+
+
+def get_reauth_coordinator() -> ReauthCoordinator:
+    """Get the global ReauthCoordinator instance."""
+    global _coordinator
+    if _coordinator is None:
+        _coordinator = ReauthCoordinator()
+    return _coordinator