feat(rotation): ✨ add sequential rotation mode with provider-specific quota parsing

Introduces a new credential rotation mode system that allows providers to choose between "balanced" (distribute load evenly) and "sequential" (use until exhausted) strategies. Sequential mode is particularly beneficial for providers with cache-preserving features like Antigravity's thinking signature caches.

Key changes:
- Added ROTATION_MODE_{PROVIDER} environment variable support with comprehensive documentation in .env.example
- Implemented provider-specific quota error parsing for Antigravity and Gemini CLI providers, extracting retry_after from Google RPC error format (handles compound durations like "143h4m52.73s")
- Extended ProviderInterface with rotation mode configuration and parse_quota_error() method
- Updated UsageManager to support sequential credential selection that preserves sticky credential usage until quota exhaustion
- Enhanced error_handler.py classify_error() to attempt provider-specific parsing before falling back to generic classification
- Added rotation mode management UI in settings_tool.py with visual indicators for configured vs default modes
- Preserved long-term cooldowns during daily reset to prevent premature quota retry
- Updated all classify_error() call sites to pass provider parameter for context-aware parsing

Provider defaults:
- Antigravity: sequential (preserves thinking caches, handles weekly quota reset)
- Gemini CLI: balanced (short cooldowns in seconds/minutes)
- All others: balanced (standard per-minute rate limits)

The sequential mode ensures the same credential is reused until it hits a cooldown (429 error), at which point the system switches to the next available credential. This maximizes cache hit rates for providers that maintain request context across API calls.

.env.example

@@ -159,6 +159,32 @@ MAX_CONCURRENT_REQUESTS_PER_KEY_GEMINI=1
 MAX_CONCURRENT_REQUESTS_PER_KEY_ANTHROPIC=1
 MAX_CONCURRENT_REQUESTS_PER_KEY_IFLOW=1
 
+# --- Credential Rotation Mode ---
+# Controls how credentials are rotated when multiple are available for a provider.
+# This affects how the proxy selects the next credential to use for requests.
+#
+# Available modes:
+#   balanced   - (Default) Rotate credentials evenly across requests to distribute load.
+#                Best for API keys with per-minute rate limits.
+#   sequential - Use one credential until it's exhausted (429 error), then switch to next.
+#                Best for credentials with daily/weekly quotas (e.g., free tier accounts).
+#                When a credential hits quota, it's put on cooldown based on the reset time
+#                parsed from the provider's error response.
+#
+# Format: ROTATION_MODE_<PROVIDER_NAME>=<mode>
+#
+# Provider Defaults:
+#   - antigravity: sequential (free tier accounts with daily quotas)
+#   - All others: balanced
+#
+# Example:
+# ROTATION_MODE_GEMINI=sequential    # Use Gemini keys until quota exhausted
+# ROTATION_MODE_OPENAI=balanced      # Distribute load across OpenAI keys (default)
+# ROTATION_MODE_ANTIGRAVITY=balanced # Override Antigravity's sequential default
+#
+# ROTATION_MODE_GEMINI=balanced
+# ROTATION_MODE_ANTIGRAVITY=sequential
+
 # ------------------------------------------------------------------------------
 # | [ADVANCED] Proxy Configuration                                             |
 # ------------------------------------------------------------------------------

src/proxy_app/settings_tool.py

@@ -17,37 +17,38 @@ console = Console()
 
 def clear_screen():
     """
-    Cross-platform terminal clear that works robustly on both 
+    Cross-platform terminal clear that works robustly on both
     classic Windows conhost and modern terminals (Windows Terminal, Linux, Mac).
-    
+
     Uses native OS commands instead of ANSI escape sequences:
     - Windows (conhost & Windows Terminal): cls
     - Unix-like systems (Linux, Mac): clear
     """
-    os.system('cls' if os.name == 'nt' else 'clear')
+    os.system("cls" if os.name == "nt" else "clear")
 
 
 class AdvancedSettings:
     """Manages pending changes to .env"""
-    
+
     def __init__(self):
         self.env_file = Path.cwd() / ".env"
         self.pending_changes = {}  # key -> value (None means delete)
         self.load_current_settings()
-    
+
     def load_current_settings(self):
         """Load current .env values into env vars"""
         from dotenv import load_dotenv
+
         load_dotenv(override=True)
-    
+
     def set(self, key: str, value: str):
         """Stage a change"""
         self.pending_changes[key] = value
-    
+
     def remove(self, key: str):
         """Stage a removal"""
         self.pending_changes[key] = None
-    
+
     def save(self):
         """Write pending changes to .env"""
         for key, value in self.pending_changes.items():
@@ -57,14 +58,14 @@ class AdvancedSettings:
             else:
                 # Set key
                 set_key(str(self.env_file), key, value)
-        
+
         self.pending_changes.clear()
         self.load_current_settings()
-    
+
     def discard(self):
         """Discard pending changes"""
         self.pending_changes.clear()
-    
+
     def has_pending(self) -> bool:
         """Check if there are pending changes"""
         return bool(self.pending_changes)
@@ -72,14 +73,14 @@ class AdvancedSettings:
 
 class CustomProviderManager:
     """Manages custom provider API bases"""
-    
+
     def __init__(self, settings: AdvancedSettings):
         self.settings = settings
-    
+
     def get_current_providers(self) -> Dict[str, str]:
         """Get currently configured custom providers"""
         from proxy_app.provider_urls import PROVIDER_URL_MAP
-        
+
         providers = {}
         for key, value in os.environ.items():
             if key.endswith("_API_BASE"):
@@ -88,16 +89,16 @@ class CustomProviderManager:
                 if provider not in PROVIDER_URL_MAP:
                     providers[provider] = value
         return providers
-    
+
     def add_provider(self, name: str, api_base: str):
         """Add PROVIDER_API_BASE"""
         key = f"{name.upper()}_API_BASE"
         self.settings.set(key, api_base)
-    
+
     def edit_provider(self, name: str, api_base: str):
         """Edit PROVIDER_API_BASE"""
         self.add_provider(name, api_base)
-    
+
     def remove_provider(self, name: str):
         """Remove PROVIDER_API_BASE"""
         key = f"{name.upper()}_API_BASE"
@@ -106,10 +107,10 @@ class CustomProviderManager:
 
 class ModelDefinitionManager:
     """Manages PROVIDER_MODELS"""
-    
+
     def __init__(self, settings: AdvancedSettings):
         self.settings = settings
-    
+
     def get_current_provider_models(self, provider: str) -> Optional[Dict]:
         """Get currently configured models for a provider"""
         key = f"{provider.upper()}_MODELS"
@@ -120,7 +121,7 @@ class ModelDefinitionManager:
             except (json.JSONDecodeError, ValueError):
                 return None
         return None
-    
+
     def get_all_providers_with_models(self) -> Dict[str, int]:
         """Get all providers with model definitions"""
         providers = {}
@@ -136,13 +137,13 @@ class ModelDefinitionManager:
                 except (json.JSONDecodeError, ValueError):
                     pass
         return providers
-    
+
     def set_models(self, provider: str, models: Dict[str, Dict[str, Any]]):
         """Set PROVIDER_MODELS"""
         key = f"{provider.upper()}_MODELS"
         value = json.dumps(models)
         self.settings.set(key, value)
-    
+
     def remove_models(self, provider: str):
         """Remove PROVIDER_MODELS"""
         key = f"{provider.upper()}_MODELS"
@@ -151,10 +152,10 @@ class ModelDefinitionManager:
 
 class ConcurrencyManager:
     """Manages MAX_CONCURRENT_REQUESTS_PER_KEY_PROVIDER"""
-    
+
     def __init__(self, settings: AdvancedSettings):
         self.settings = settings
-    
+
     def get_current_limits(self) -> Dict[str, int]:
         """Get currently configured concurrency limits"""
         limits = {}
@@ -166,18 +167,73 @@ class ConcurrencyManager:
                 except (json.JSONDecodeError, ValueError):
                     pass
         return limits
-    
+
     def set_limit(self, provider: str, limit: int):
         """Set concurrency limit"""
         key = f"MAX_CONCURRENT_REQUESTS_PER_KEY_{provider.upper()}"
         self.settings.set(key, str(limit))
-    
+
     def remove_limit(self, provider: str):
         """Remove concurrency limit (reset to default)"""
         key = f"MAX_CONCURRENT_REQUESTS_PER_KEY_{provider.upper()}"
         self.settings.remove(key)
 
 
+class RotationModeManager:
+    """Manages ROTATION_MODE_PROVIDER settings for sequential/balanced credential rotation"""
+
+    VALID_MODES = ["balanced", "sequential"]
+
+    def __init__(self, settings: AdvancedSettings):
+        self.settings = settings
+
+    def get_current_modes(self) -> Dict[str, str]:
+        """Get currently configured rotation modes"""
+        modes = {}
+        for key, value in os.environ.items():
+            if key.startswith("ROTATION_MODE_"):
+                provider = key.replace("ROTATION_MODE_", "").lower()
+                if value.lower() in self.VALID_MODES:
+                    modes[provider] = value.lower()
+        return modes
+
+    def get_default_mode(self, provider: str) -> str:
+        """Get the default rotation mode for a provider"""
+        # Import here to avoid circular imports
+        try:
+            from rotator_library.providers.provider_interface import (
+                LLMProviderInterface,
+            )
+
+            return LLMProviderInterface.get_rotation_mode(provider)
+        except ImportError:
+            # Fallback defaults if import fails
+            if provider.lower() == "antigravity":
+                return "sequential"
+            return "balanced"
+
+    def get_effective_mode(self, provider: str) -> str:
+        """Get the effective rotation mode (configured or default)"""
+        configured = self.get_current_modes().get(provider.lower())
+        if configured:
+            return configured
+        return self.get_default_mode(provider)
+
+    def set_mode(self, provider: str, mode: str):
+        """Set rotation mode for a provider"""
+        if mode.lower() not in self.VALID_MODES:
+            raise ValueError(
+                f"Invalid rotation mode: {mode}. Must be one of {self.VALID_MODES}"
+            )
+        key = f"ROTATION_MODE_{provider.upper()}"
+        self.settings.set(key, mode.lower())
+
+    def remove_mode(self, provider: str):
+        """Remove rotation mode (reset to provider default)"""
+        key = f"ROTATION_MODE_{provider.upper()}"
+        self.settings.remove(key)
+
+
 # =============================================================================
 # PROVIDER-SPECIFIC SETTINGS DEFINITIONS
 # =============================================================================
@@ -294,24 +350,26 @@ PROVIDER_SETTINGS_MAP = {
 
 class ProviderSettingsManager:
     """Manages provider-specific configuration settings"""
-    
+
     def __init__(self, settings: AdvancedSettings):
         self.settings = settings
-    
+
     def get_available_providers(self) -> List[str]:
         """Get list of providers with specific settings available"""
         return list(PROVIDER_SETTINGS_MAP.keys())
-    
-    def get_provider_settings_definitions(self, provider: str) -> Dict[str, Dict[str, Any]]:
+
+    def get_provider_settings_definitions(
+        self, provider: str
+    ) -> Dict[str, Dict[str, Any]]:
         """Get settings definitions for a provider"""
         return PROVIDER_SETTINGS_MAP.get(provider, {})
-    
+
     def get_current_value(self, key: str, definition: Dict[str, Any]) -> Any:
         """Get current value of a setting from environment"""
         env_value = os.getenv(key)
         if env_value is None:
             return definition.get("default")
-        
+
         setting_type = definition.get("type", "str")
         try:
             if setting_type == "bool":
@@ -322,7 +380,7 @@ class ProviderSettingsManager:
                 return env_value
         except (ValueError, AttributeError):
             return definition.get("default")
-    
+
     def get_all_current_values(self, provider: str) -> Dict[str, Any]:
         """Get all current values for a provider"""
         definitions = self.get_provider_settings_definitions(provider)
@@ -330,7 +388,7 @@ class ProviderSettingsManager:
         for key, definition in definitions.items():
             values[key] = self.get_current_value(key, definition)
         return values
-    
+
     def set_value(self, key: str, value: Any, definition: Dict[str, Any]):
         """Set a setting value, converting to string for .env storage"""
         setting_type = definition.get("type", "str")
@@ -339,11 +397,11 @@ class ProviderSettingsManager:
         else:
             str_value = str(value)
         self.settings.set(key, str_value)
-    
+
     def reset_to_default(self, key: str):
         """Remove a setting to reset it to default"""
         self.settings.remove(key)
-    
+
     def get_modified_settings(self, provider: str) -> Dict[str, Any]:
         """Get settings that differ from defaults"""
         definitions = self.get_provider_settings_definitions(provider)
@@ -358,80 +416,96 @@ class ProviderSettingsManager:
 
 class SettingsTool:
     """Main settings tool TUI"""
-    
+
     def __init__(self):
         self.console = Console()
         self.settings = AdvancedSettings()
         self.provider_mgr = CustomProviderManager(self.settings)
         self.model_mgr = ModelDefinitionManager(self.settings)
         self.concurrency_mgr = ConcurrencyManager(self.settings)
+        self.rotation_mgr = RotationModeManager(self.settings)
         self.provider_settings_mgr = ProviderSettingsManager(self.settings)
         self.running = True
-    
+
     def get_available_providers(self) -> List[str]:
         """Get list of providers that have credentials configured"""
         env_file = Path.cwd() / ".env"
         providers = set()
-        
+
         # Scan for providers with API keys from local .env
         if env_file.exists():
             try:
-                with open(env_file, 'r', encoding='utf-8') as f:
+                with open(env_file, "r", encoding="utf-8") as f:
                     for line in f:
                         line = line.strip()
-                        if "_API_KEY" in line and "PROXY_API_KEY" not in line and "=" in line:
+                        if (
+                            "_API_KEY" in line
+                            and "PROXY_API_KEY" not in line
+                            and "=" in line
+                        ):
                             provider = line.split("_API_KEY")[0].strip().lower()
                             providers.add(provider)
             except (IOError, OSError):
                 pass
-        
+
         # Also check for OAuth providers from files
         oauth_dir = Path("oauth_credentials")
         if oauth_dir.exists():
             for file in oauth_dir.glob("*_oauth_*.json"):
                 provider = file.name.split("_oauth_")[0]
                 providers.add(provider)
-        
+
         return sorted(list(providers))
 
     def run(self):
         """Main loop"""
         while self.running:
             self.show_main_menu()
-    
+
     def show_main_menu(self):
         """Display settings categories"""
         clear_screen()
-        
-        self.console.print(Panel.fit(
-            "[bold cyan]🔧 Advanced Settings Configuration[/bold cyan]",
-            border_style="cyan"
-        ))
-        
+
+        self.console.print(
+            Panel.fit(
+                "[bold cyan]🔧 Advanced Settings Configuration[/bold cyan]",
+                border_style="cyan",
+            )
+        )
+
         self.console.print()
         self.console.print("[bold]⚙️  Configuration Categories[/bold]")
         self.console.print()
         self.console.print("   1. 🌐 Custom Provider API Bases")
         self.console.print("   2. 📦 Provider Model Definitions")
         self.console.print("   3. ⚡ Concurrency Limits")
-        self.console.print("   4. 🔬 Provider-Specific Settings")
-        self.console.print("   5. 💾 Save & Exit")
-        self.console.print("   6. 🚫 Exit Without Saving")
-        
+        self.console.print("   4. 🔄 Rotation Modes")
+        self.console.print("   5. 🔬 Provider-Specific Settings")
+        self.console.print("   6. 💾 Save & Exit")
+        self.console.print("   7. 🚫 Exit Without Saving")
+
         self.console.print()
         self.console.print("━" * 70)
-        
+
         if self.settings.has_pending():
-            self.console.print("[yellow]ℹ️  Changes are pending until you select \"Save & Exit\"[/yellow]")
+            self.console.print(
+                '[yellow]ℹ️  Changes are pending until you select "Save & Exit"[/yellow]'
+            )
         else:
             self.console.print("[dim]ℹ️  No pending changes[/dim]")
-        
+
         self.console.print()
-        self.console.print("[dim]⚠️  Model filters not supported - edit .env for IGNORE_MODELS_* / WHITELIST_MODELS_*[/dim]")
+        self.console.print(
+            "[dim]⚠️  Model filters not supported - edit .env for IGNORE_MODELS_* / WHITELIST_MODELS_*[/dim]"
+        )
         self.console.print()
-        
-        choice = Prompt.ask("Select option", choices=["1", "2", "3", "4", "5", "6"], show_choices=False)
-        
+
+        choice = Prompt.ask(
+            "Select option",
+            choices=["1", "2", "3", "4", "5", "6", "7"],
+            show_choices=False,
+        )
+
         if choice == "1":
             self.manage_custom_providers()
         elif choice == "2":
@@ -439,34 +513,38 @@ class SettingsTool:
         elif choice == "3":
             self.manage_concurrency_limits()
         elif choice == "4":
-            self.manage_provider_settings()
+            self.manage_rotation_modes()
         elif choice == "5":
-            self.save_and_exit()
+            self.manage_provider_settings()
         elif choice == "6":
+            self.save_and_exit()
+        elif choice == "7":
             self.exit_without_saving()
-    
+
     def manage_custom_providers(self):
         """Manage custom provider API bases"""
         while True:
             clear_screen()
-            
+
             providers = self.provider_mgr.get_current_providers()
-            
-            self.console.print(Panel.fit(
-                "[bold cyan]🌐 Custom Provider API Bases[/bold cyan]",
-                border_style="cyan"
-            ))
-            
+
+            self.console.print(
+                Panel.fit(
+                    "[bold cyan]🌐 Custom Provider API Bases[/bold cyan]",
+                    border_style="cyan",
+                )
+            )
+
             self.console.print()
             self.console.print("[bold]📋 Configured Custom Providers[/bold]")
             self.console.print("━" * 70)
-            
+
             if providers:
                 for name, base in providers.items():
                     self.console.print(f"   • {name:15} {base}")
             else:
                 self.console.print("   [dim]No custom providers configured[/dim]")
-            
+
             self.console.print()
             self.console.print("━" * 70)
             self.console.print()
@@ -476,94 +554,116 @@ class SettingsTool:
             self.console.print("   2. ✏️  Edit Existing Provider")
             self.console.print("   3. 🗑️  Remove Provider")
             self.console.print("   4. ↩️  Back to Settings Menu")
-            
+
             self.console.print()
             self.console.print("━" * 70)
             self.console.print()
-            
-            choice = Prompt.ask("Select option", choices=["1", "2", "3", "4"], show_choices=False)
-            
+
+            choice = Prompt.ask(
+                "Select option", choices=["1", "2", "3", "4"], show_choices=False
+            )
+
             if choice == "1":
                 name = Prompt.ask("Provider name (e.g., 'opencode')").strip().lower()
                 if name:
                     api_base = Prompt.ask("API Base URL").strip()
                     if api_base:
                         self.provider_mgr.add_provider(name, api_base)
-                        self.console.print(f"\n[green]✅ Custom provider '{name}' configured![/green]")
-                        self.console.print(f"   To use: set {name.upper()}_API_KEY in credentials")
+                        self.console.print(
+                            f"\n[green]✅ Custom provider '{name}' configured![/green]"
+                        )
+                        self.console.print(
+                            f"   To use: set {name.upper()}_API_KEY in credentials"
+                        )
                         input("\nPress Enter to continue...")
-            
+
             elif choice == "2":
                 if not providers:
                     self.console.print("\n[yellow]No providers to edit[/yellow]")
                     input("\nPress Enter to continue...")
                     continue
-                
+
                 # Show numbered list
                 self.console.print("\n[bold]Select provider to edit:[/bold]")
                 providers_list = list(providers.keys())
                 for idx, prov in enumerate(providers_list, 1):
                     self.console.print(f"   {idx}. {prov}")
-                
-                choice_idx = IntPrompt.ask("Select option", choices=[str(i) for i in range(1, len(providers_list) + 1)])
+
+                choice_idx = IntPrompt.ask(
+                    "Select option",
+                    choices=[str(i) for i in range(1, len(providers_list) + 1)],
+                )
                 name = providers_list[choice_idx - 1]
                 current_base = providers.get(name, "")
-                
+
                 self.console.print(f"\nCurrent API Base: {current_base}")
-                new_base = Prompt.ask("New API Base [press Enter to keep current]", default=current_base).strip()
-                
+                new_base = Prompt.ask(
+                    "New API Base [press Enter to keep current]", default=current_base
+                ).strip()
+
                 if new_base and new_base != current_base:
                     self.provider_mgr.edit_provider(name, new_base)
-                    self.console.print(f"\n[green]✅ Custom provider '{name}' updated![/green]")
+                    self.console.print(
+                        f"\n[green]✅ Custom provider '{name}' updated![/green]"
+                    )
                 else:
                     self.console.print("\n[yellow]No changes made[/yellow]")
                 input("\nPress Enter to continue...")
-            
+
             elif choice == "3":
                 if not providers:
                     self.console.print("\n[yellow]No providers to remove[/yellow]")
                     input("\nPress Enter to continue...")
                     continue
-                
+
                 # Show numbered list
                 self.console.print("\n[bold]Select provider to remove:[/bold]")
                 providers_list = list(providers.keys())
                 for idx, prov in enumerate(providers_list, 1):
                     self.console.print(f"   {idx}. {prov}")
-                
-                choice_idx = IntPrompt.ask("Select option", choices=[str(i) for i in range(1, len(providers_list) + 1)])
+
+                choice_idx = IntPrompt.ask(
+                    "Select option",
+                    choices=[str(i) for i in range(1, len(providers_list) + 1)],
+                )
                 name = providers_list[choice_idx - 1]
-                
+
                 if Confirm.ask(f"Remove '{name}'?"):
                     self.provider_mgr.remove_provider(name)
-                    self.console.print(f"\n[green]✅ Provider '{name}' removed![/green]")
+                    self.console.print(
+                        f"\n[green]✅ Provider '{name}' removed![/green]"
+                    )
                     input("\nPress Enter to continue...")
-            
+
             elif choice == "4":
                 break
-    
+
     def manage_model_definitions(self):
         """Manage provider model definitions"""
         while True:
             clear_screen()
-            
+
             all_providers = self.model_mgr.get_all_providers_with_models()
-            
-            self.console.print(Panel.fit(
-                "[bold cyan]📦 Provider Model Definitions[/bold cyan]",
-                border_style="cyan"
-            ))
-            
+
+            self.console.print(
+                Panel.fit(
+                    "[bold cyan]📦 Provider Model Definitions[/bold cyan]",
+                    border_style="cyan",
+                )
+            )
+
             self.console.print()
             self.console.print("[bold]📋 Configured Provider Models[/bold]")
             self.console.print("━" * 70)
-            
+
             if all_providers:
                 for provider, count in all_providers.items():
-                    self.console.print(f"   • {provider:15} {count} model{'s' if count > 1 else ''}")
+                    self.console.print(
+                        f"   • {provider:15} {count} model{'s' if count > 1 else ''}"
+                    )
             else:
                 self.console.print("   [dim]No model definitions configured[/dim]")
-            
+
             self.console.print()
             self.console.print("━" * 70)
             self.console.print()
@@ -574,13 +674,15 @@ class SettingsTool:
             self.console.print("   3. 👁️  View Provider Models")
             self.console.print("   4. 🗑️  Remove Provider Models")
             self.console.print("   5. ↩️  Back to Settings Menu")
-            
+
             self.console.print()
             self.console.print("━" * 70)
             self.console.print()
-            
-            choice = Prompt.ask("Select option", choices=["1", "2", "3", "4", "5"], show_choices=False)
-            
+
+            choice = Prompt.ask(
+                "Select option", choices=["1", "2", "3", "4", "5"], show_choices=False
+            )
+
             if choice == "1":
                 self.add_model_definitions()
             elif choice == "2":
@@ -600,57 +702,71 @@ class SettingsTool:
                     self.console.print("\n[yellow]No providers to remove[/yellow]")
                     input("\nPress Enter to continue...")
                     continue
-                
+
                 # Show numbered list
-                self.console.print("\n[bold]Select provider to remove models from:[/bold]")
+                self.console.print(
+                    "\n[bold]Select provider to remove models from:[/bold]"
+                )
                 providers_list = list(all_providers.keys())
                 for idx, prov in enumerate(providers_list, 1):
                     self.console.print(f"   {idx}. {prov}")
-                
-                choice_idx = IntPrompt.ask("Select option", choices=[str(i) for i in range(1, len(providers_list) + 1)])
+
+                choice_idx = IntPrompt.ask(
+                    "Select option",
+                    choices=[str(i) for i in range(1, len(providers_list) + 1)],
+                )
                 provider = providers_list[choice_idx - 1]
-                
+
                 if Confirm.ask(f"Remove all model definitions for '{provider}'?"):
                     self.model_mgr.remove_models(provider)
-                    self.console.print(f"\n[green]✅ Model definitions removed for '{provider}'![/green]")
+                    self.console.print(
+                        f"\n[green]✅ Model definitions removed for '{provider}'![/green]"
+                    )
                     input("\nPress Enter to continue...")
             elif choice == "5":
                 break
-    
+
     def add_model_definitions(self):
         """Add model definitions for a provider"""
         # Get available providers from credentials
         available_providers = self.get_available_providers()
-        
+
         if not available_providers:
-            self.console.print("\n[yellow]No providers with credentials found. Please add credentials first.[/yellow]")
+            self.console.print(
+                "\n[yellow]No providers with credentials found. Please add credentials first.[/yellow]"
+            )
             input("\nPress Enter to continue...")
             return
-        
+
         # Show provider selection menu
         self.console.print("\n[bold]Select provider:[/bold]")
         for idx, prov in enumerate(available_providers, 1):
             self.console.print(f"   {idx}. {prov}")
-        self.console.print(f"   {len(available_providers) + 1}. Enter custom provider name")
-        
-        choice = IntPrompt.ask("Select option", choices=[str(i) for i in range(1, len(available_providers) + 2)])
-        
+        self.console.print(
+            f"   {len(available_providers) + 1}. Enter custom provider name"
+        )
+
+        choice = IntPrompt.ask(
+            "Select option",
+            choices=[str(i) for i in range(1, len(available_providers) + 2)],
+        )
+
         if choice == len(available_providers) + 1:
             provider = Prompt.ask("Provider name").strip().lower()
         else:
             provider = available_providers[choice - 1]
-        
+
         if not provider:
             return
-        
+
         self.console.print("\nHow would you like to define models?")
         self.console.print("   1. Simple list (names only)")
         self.console.print("   2. Advanced (names with IDs and options)")
-        
+
         mode = Prompt.ask("Select mode", choices=["1", "2"], show_choices=False)
-        
+
         models = {}
-        
+
         if mode == "1":
             # Simple mode
             while True:
@@ -667,13 +783,19 @@ class SettingsTool:
                     break
                 if name:
                     model_def = {}
-                    model_id = Prompt.ask(f"Model ID [press Enter to use '{name}']", default=name).strip()
+                    model_id = Prompt.ask(
+                        f"Model ID [press Enter to use '{name}']", default=name
+                    ).strip()
                     if model_id and model_id != name:
                         model_def["id"] = model_id
-                    
+
                     # Optional: model options
-                    if Confirm.ask("Add model options (e.g., temperature limits)?", default=False):
-                        self.console.print("\nEnter options as key=value pairs (one per line, 'done' to finish):")
+                    if Confirm.ask(
+                        "Add model options (e.g., temperature limits)?", default=False
+                    ):
+                        self.console.print(
+                            "\nEnter options as key=value pairs (one per line, 'done' to finish):"
+                        )
                         options = {}
                         while True:
                             opt = Prompt.ask("Option").strip()
@@ -690,121 +812,143 @@ class SettingsTool:
                                 options[key.strip()] = value
                         if options:
                             model_def["options"] = options
-                    
+
                     models[name] = model_def
-        
+
         if models:
             self.model_mgr.set_models(provider, models)
-            self.console.print(f"\n[green]✅ Model definitions saved for '{provider}'![/green]")
+            self.console.print(
+                f"\n[green]✅ Model definitions saved for '{provider}'![/green]"
+            )
         else:
             self.console.print("\n[yellow]No models added[/yellow]")
-        
+
         input("\nPress Enter to continue...")
-    
+
     def edit_model_definitions(self, providers: List[str]):
         """Edit existing model definitions"""
         # Show numbered list
         self.console.print("\n[bold]Select provider to edit:[/bold]")
         for idx, prov in enumerate(providers, 1):
             self.console.print(f"   {idx}. {prov}")
-        
-        choice_idx = IntPrompt.ask("Select option", choices=[str(i) for i in range(1, len(providers) + 1)])
+
+        choice_idx = IntPrompt.ask(
+            "Select option", choices=[str(i) for i in range(1, len(providers) + 1)]
+        )
         provider = providers[choice_idx - 1]
-        
+
         current_models = self.model_mgr.get_current_provider_models(provider)
         if not current_models:
             self.console.print(f"\n[yellow]No models found for '{provider}'[/yellow]")
             input("\nPress Enter to continue...")
             return
-        
+
         # Convert to dict if list
         if isinstance(current_models, list):
             current_models = {m: {} for m in current_models}
-        
+
         while True:
             clear_screen()
             self.console.print(f"[bold]Editing models for: {provider}[/bold]\n")
             self.console.print("Current models:")
             for i, (name, definition) in enumerate(current_models.items(), 1):
-                model_id = definition.get("id", name) if isinstance(definition, dict) else name
+                model_id = (
+                    definition.get("id", name) if isinstance(definition, dict) else name
+                )
                 self.console.print(f"   {i}. {name} (ID: {model_id})")
-            
+
             self.console.print("\nOptions:")
             self.console.print("   1. Add new model")
             self.console.print("   2. Edit existing model")
             self.console.print("   3. Remove model")
             self.console.print("   4. Done")
-            
-            choice = Prompt.ask("\nSelect option", choices=["1", "2", "3", "4"], show_choices=False)
-            
+
+            choice = Prompt.ask(
+                "\nSelect option", choices=["1", "2", "3", "4"], show_choices=False
+            )
+
             if choice == "1":
                 name = Prompt.ask("New model name").strip()
                 if name and name not in current_models:
                     model_id = Prompt.ask("Model ID", default=name).strip()
                     current_models[name] = {"id": model_id} if model_id != name else {}
-            
+
             elif choice == "2":
                 # Show numbered list
                 models_list = list(current_models.keys())
                 self.console.print("\n[bold]Select model to edit:[/bold]")
                 for idx, model_name in enumerate(models_list, 1):
                     self.console.print(f"   {idx}. {model_name}")
-                
-                model_idx = IntPrompt.ask("Select option", choices=[str(i) for i in range(1, len(models_list) + 1)])
+
+                model_idx = IntPrompt.ask(
+                    "Select option",
+                    choices=[str(i) for i in range(1, len(models_list) + 1)],
+                )
                 name = models_list[model_idx - 1]
-                
+
                 current_def = current_models[name]
-                current_id = current_def.get("id", name) if isinstance(current_def, dict) else name
-                
+                current_id = (
+                    current_def.get("id", name)
+                    if isinstance(current_def, dict)
+                    else name
+                )
+
                 new_id = Prompt.ask("Model ID", default=current_id).strip()
                 current_models[name] = {"id": new_id} if new_id != name else {}
-            
+
             elif choice == "3":
                 # Show numbered list
                 models_list = list(current_models.keys())
                 self.console.print("\n[bold]Select model to remove:[/bold]")
                 for idx, model_name in enumerate(models_list, 1):
                     self.console.print(f"   {idx}. {model_name}")
-                
-                model_idx = IntPrompt.ask("Select option", choices=[str(i) for i in range(1, len(models_list) + 1)])
+
+                model_idx = IntPrompt.ask(
+                    "Select option",
+                    choices=[str(i) for i in range(1, len(models_list) + 1)],
+                )
                 name = models_list[model_idx - 1]
-                
+
                 if Confirm.ask(f"Remove '{name}'?"):
                     del current_models[name]
-            
+
             elif choice == "4":
                 break
-        
+
         if current_models:
             self.model_mgr.set_models(provider, current_models)
             self.console.print(f"\n[green]✅ Models updated for '{provider}'![/green]")
         else:
-            self.console.print("\n[yellow]No models left - removing definition[/yellow]")
+            self.console.print(
+                "\n[yellow]No models left - removing definition[/yellow]"
+            )
             self.model_mgr.remove_models(provider)
-        
+
         input("\nPress Enter to continue...")
-    
+
     def view_model_definitions(self, providers: List[str]):
         """View model definitions for a provider"""
         # Show numbered list
         self.console.print("\n[bold]Select provider to view:[/bold]")
         for idx, prov in enumerate(providers, 1):
             self.console.print(f"   {idx}. {prov}")
-        
-        choice_idx = IntPrompt.ask("Select option", choices=[str(i) for i in range(1, len(providers) + 1)])
+
+        choice_idx = IntPrompt.ask(
+            "Select option", choices=[str(i) for i in range(1, len(providers) + 1)]
+        )
         provider = providers[choice_idx - 1]
-        
+
         models = self.model_mgr.get_current_provider_models(provider)
         if not models:
             self.console.print(f"\n[yellow]No models found for '{provider}'[/yellow]")
             input("\nPress Enter to continue...")
             return
-        
+
         clear_screen()
         self.console.print(f"[bold]Provider: {provider}[/bold]\n")
         self.console.print("[bold]📦 Configured Models:[/bold]")
         self.console.print("━" * 50)
-        
+
         # Handle both dict and list formats
         if isinstance(models, dict):
             for name, definition in models.items():
@@ -822,74 +966,88 @@ class SettingsTool:
             for name in models:
                 self.console.print(f"   Name: {name}")
                 self.console.print()
-        
+
         input("Press Enter to return...")
-    
+
     def manage_provider_settings(self):
         """Manage provider-specific settings (Antigravity, Gemini CLI)"""
         while True:
             clear_screen()
-            
+
             available_providers = self.provider_settings_mgr.get_available_providers()
-            
-            self.console.print(Panel.fit(
-                "[bold cyan]🔬 Provider-Specific Settings[/bold cyan]",
-                border_style="cyan"
-            ))
-            
+
+            self.console.print(
+                Panel.fit(
+                    "[bold cyan]🔬 Provider-Specific Settings[/bold cyan]",
+                    border_style="cyan",
+                )
+            )
+
             self.console.print()
-            self.console.print("[bold]📋 Available Providers with Custom Settings[/bold]")
+            self.console.print(
+                "[bold]📋 Available Providers with Custom Settings[/bold]"
+            )
             self.console.print("━" * 70)
-            
+
             for provider in available_providers:
                 modified = self.provider_settings_mgr.get_modified_settings(provider)
-                status = f"[yellow]{len(modified)} modified[/yellow]" if modified else "[dim]defaults[/dim]"
+                status = (
+                    f"[yellow]{len(modified)} modified[/yellow]"
+                    if modified
+                    else "[dim]defaults[/dim]"
+                )
                 display_name = provider.replace("_", " ").title()
                 self.console.print(f"   • {display_name:20} {status}")
-            
+
             self.console.print()
             self.console.print("━" * 70)
             self.console.print()
             self.console.print("[bold]⚙️  Select Provider to Configure[/bold]")
             self.console.print()
-            
+
             for idx, provider in enumerate(available_providers, 1):
                 display_name = provider.replace("_", " ").title()
                 self.console.print(f"   {idx}. {display_name}")
-            self.console.print(f"   {len(available_providers) + 1}. ↩️  Back to Settings Menu")
-            
+            self.console.print(
+                f"   {len(available_providers) + 1}. ↩️  Back to Settings Menu"
+            )
+
             self.console.print()
             self.console.print("━" * 70)
             self.console.print()
-            
+
             choices = [str(i) for i in range(1, len(available_providers) + 2)]
             choice = Prompt.ask("Select option", choices=choices, show_choices=False)
             choice_idx = int(choice)
-            
+
             if choice_idx == len(available_providers) + 1:
                 break
-            
+
             provider = available_providers[choice_idx - 1]
             self._manage_single_provider_settings(provider)
-    
+
     def _manage_single_provider_settings(self, provider: str):
         """Manage settings for a single provider"""
         while True:
             clear_screen()
-            
+
             display_name = provider.replace("_", " ").title()
-            definitions = self.provider_settings_mgr.get_provider_settings_definitions(provider)
+            definitions = self.provider_settings_mgr.get_provider_settings_definitions(
+                provider
+            )
             current_values = self.provider_settings_mgr.get_all_current_values(provider)
-            
-            self.console.print(Panel.fit(
-                f"[bold cyan]🔬 {display_name} Settings[/bold cyan]",
-                border_style="cyan"
-            ))
-            
+
+            self.console.print(
+                Panel.fit(
+                    f"[bold cyan]🔬 {display_name} Settings[/bold cyan]",
+                    border_style="cyan",
+                )
+            )
+
             self.console.print()
             self.console.print("[bold]📋 Current Settings[/bold]")
             self.console.print("━" * 70)
-            
+
             # Display all settings with current values
             settings_list = list(definitions.keys())
             for idx, key in enumerate(settings_list, 1):
@@ -898,25 +1056,35 @@ class SettingsTool:
                 default = definition.get("default")
                 setting_type = definition.get("type", "str")
                 description = definition.get("description", "")
-                
+
                 # Format value display
                 if setting_type == "bool":
-                    value_display = "[green]✓ Enabled[/green]" if current else "[red]✗ Disabled[/red]"
+                    value_display = (
+                        "[green]✓ Enabled[/green]"
+                        if current
+                        else "[red]✗ Disabled[/red]"
+                    )
                 elif setting_type == "int":
                     value_display = f"[cyan]{current}[/cyan]"
                 else:
-                    value_display = f"[cyan]{current or '(not set)'}[/cyan]" if current else "[dim](not set)[/dim]"
-                
+                    value_display = (
+                        f"[cyan]{current or '(not set)'}[/cyan]"
+                        if current
+                        else "[dim](not set)[/dim]"
+                    )
+
                 # Check if modified from default
                 modified = current != default
                 mod_marker = "[yellow]*[/yellow]" if modified else " "
-                
+
                 # Short key name for display (strip provider prefix)
                 short_key = key.replace(f"{provider.upper()}_", "")
-                
-                self.console.print(f"  {mod_marker}{idx:2}. {short_key:35} {value_display}")
+
+                self.console.print(
+                    f"  {mod_marker}{idx:2}. {short_key:35} {value_display}"
+                )
                 self.console.print(f"       [dim]{description}[/dim]")
-            
+
             self.console.print()
             self.console.print("━" * 70)
             self.console.print("[dim]* = modified from default[/dim]")
@@ -927,13 +1095,17 @@ class SettingsTool:
             self.console.print("   R. 🔄 Reset Setting to Default")
             self.console.print("   A. 🔄 Reset All to Defaults")
             self.console.print("   B. ↩️  Back to Provider Selection")
-            
+
             self.console.print()
             self.console.print("━" * 70)
             self.console.print()
-            
-            choice = Prompt.ask("Select action", choices=["e", "r", "a", "b", "E", "R", "A", "B"], show_choices=False).lower()
-            
+
+            choice = Prompt.ask(
+                "Select action",
+                choices=["e", "r", "a", "b", "E", "R", "A", "B"],
+                show_choices=False,
+            ).lower()
+
             if choice == "b":
                 break
             elif choice == "e":
@@ -942,26 +1114,31 @@ class SettingsTool:
                 self._reset_provider_setting(provider, settings_list, definitions)
             elif choice == "a":
                 self._reset_all_provider_settings(provider, settings_list)
-    
-    def _edit_provider_setting(self, provider: str, settings_list: List[str], definitions: Dict[str, Dict[str, Any]]):
+
+    def _edit_provider_setting(
+        self,
+        provider: str,
+        settings_list: List[str],
+        definitions: Dict[str, Dict[str, Any]],
+    ):
         """Edit a single provider setting"""
         self.console.print("\n[bold]Select setting number to edit:[/bold]")
-        
+
         choices = [str(i) for i in range(1, len(settings_list) + 1)]
         choice = IntPrompt.ask("Setting number", choices=choices)
         key = settings_list[choice - 1]
         definition = definitions[key]
-        
+
         current = self.provider_settings_mgr.get_current_value(key, definition)
         default = definition.get("default")
         setting_type = definition.get("type", "str")
         short_key = key.replace(f"{provider.upper()}_", "")
-        
+
         self.console.print(f"\n[bold]Editing: {short_key}[/bold]")
         self.console.print(f"Current value: [cyan]{current}[/cyan]")
         self.console.print(f"Default value: [dim]{default}[/dim]")
         self.console.print(f"Type: {setting_type}")
-        
+
         if setting_type == "bool":
             new_value = Confirm.ask("\nEnable this setting?", default=current)
             self.provider_settings_mgr.set_value(key, new_value, definition)
@@ -972,71 +1149,252 @@ class SettingsTool:
             self.provider_settings_mgr.set_value(key, new_value, definition)
             self.console.print(f"\n[green]✅ {short_key} set to {new_value}![/green]")
         else:
-            new_value = Prompt.ask("\nNew value", default=str(current) if current else "").strip()
+            new_value = Prompt.ask(
+                "\nNew value", default=str(current) if current else ""
+            ).strip()
             if new_value:
                 self.provider_settings_mgr.set_value(key, new_value, definition)
                 self.console.print(f"\n[green]✅ {short_key} updated![/green]")
             else:
                 self.console.print("\n[yellow]No changes made[/yellow]")
-        
+
         input("\nPress Enter to continue...")
-    
-    def _reset_provider_setting(self, provider: str, settings_list: List[str], definitions: Dict[str, Dict[str, Any]]):
+
+    def _reset_provider_setting(
+        self,
+        provider: str,
+        settings_list: List[str],
+        definitions: Dict[str, Dict[str, Any]],
+    ):
         """Reset a single provider setting to default"""
         self.console.print("\n[bold]Select setting number to reset:[/bold]")
-        
+
         choices = [str(i) for i in range(1, len(settings_list) + 1)]
         choice = IntPrompt.ask("Setting number", choices=choices)
         key = settings_list[choice - 1]
         definition = definitions[key]
-        
+
         default = definition.get("default")
         short_key = key.replace(f"{provider.upper()}_", "")
-        
+
         if Confirm.ask(f"\nReset {short_key} to default ({default})?"):
             self.provider_settings_mgr.reset_to_default(key)
             self.console.print(f"\n[green]✅ {short_key} reset to default![/green]")
         else:
             self.console.print("\n[yellow]No changes made[/yellow]")
-        
+
         input("\nPress Enter to continue...")
-    
+
     def _reset_all_provider_settings(self, provider: str, settings_list: List[str]):
         """Reset all provider settings to defaults"""
         display_name = provider.replace("_", " ").title()
-        
-        if Confirm.ask(f"\n[bold red]Reset ALL {display_name} settings to defaults?[/bold red]"):
+
+        if Confirm.ask(
+            f"\n[bold red]Reset ALL {display_name} settings to defaults?[/bold red]"
+        ):
             for key in settings_list:
                 self.provider_settings_mgr.reset_to_default(key)
-            self.console.print(f"\n[green]✅ All {display_name} settings reset to defaults![/green]")
+            self.console.print(
+                f"\n[green]✅ All {display_name} settings reset to defaults![/green]"
+            )
         else:
             self.console.print("\n[yellow]No changes made[/yellow]")
-        
+
         input("\nPress Enter to continue...")
-    
+
+    def manage_rotation_modes(self):
+        """Manage credential rotation modes (sequential vs balanced)"""
+        while True:
+            clear_screen()
+
+            modes = self.rotation_mgr.get_current_modes()
+            available_providers = self.get_available_providers()
+
+            self.console.print(
+                Panel.fit(
+                    "[bold cyan]🔄 Credential Rotation Mode Configuration[/bold cyan]",
+                    border_style="cyan",
+                )
+            )
+
+            self.console.print()
+            self.console.print("[bold]📋 Rotation Modes Explained[/bold]")
+            self.console.print("━" * 70)
+            self.console.print(
+                "   [cyan]balanced[/cyan]   - Rotate credentials evenly across requests (default)"
+            )
+            self.console.print(
+                "   [cyan]sequential[/cyan] - Use one credential until exhausted (429), then switch"
+            )
+            self.console.print()
+            self.console.print("[bold]📋 Current Rotation Mode Settings[/bold]")
+            self.console.print("━" * 70)
+
+            if modes:
+                for provider, mode in modes.items():
+                    default_mode = self.rotation_mgr.get_default_mode(provider)
+                    is_custom = mode != default_mode
+                    marker = "[yellow]*[/yellow]" if is_custom else " "
+                    mode_display = (
+                        f"[green]{mode}[/green]"
+                        if mode == "sequential"
+                        else f"[blue]{mode}[/blue]"
+                    )
+                    self.console.print(f"  {marker}• {provider:20} {mode_display}")
+
+            # Show providers with default modes
+            providers_with_defaults = [p for p in available_providers if p not in modes]
+            if providers_with_defaults:
+                self.console.print()
+                self.console.print("[dim]Providers using default modes:[/dim]")
+                for provider in providers_with_defaults:
+                    default_mode = self.rotation_mgr.get_default_mode(provider)
+                    mode_display = (
+                        f"[green]{default_mode}[/green]"
+                        if default_mode == "sequential"
+                        else f"[blue]{default_mode}[/blue]"
+                    )
+                    self.console.print(
+                        f"   • {provider:20} {mode_display} [dim](default)[/dim]"
+                    )
+
+            self.console.print()
+            self.console.print("━" * 70)
+            self.console.print(
+                "[dim]* = custom setting (differs from provider default)[/dim]"
+            )
+            self.console.print()
+            self.console.print("[bold]⚙️  Actions[/bold]")
+            self.console.print()
+            self.console.print("   1. ➕ Set Rotation Mode for Provider")
+            self.console.print("   2. 🗑️  Reset to Provider Default")
+            self.console.print("   3. ↩️  Back to Settings Menu")
+
+            self.console.print()
+            self.console.print("━" * 70)
+            self.console.print()
+
+            choice = Prompt.ask(
+                "Select option", choices=["1", "2", "3"], show_choices=False
+            )
+
+            if choice == "1":
+                if not available_providers:
+                    self.console.print(
+                        "\n[yellow]No providers with credentials found. Please add credentials first.[/yellow]"
+                    )
+                    input("\nPress Enter to continue...")
+                    continue
+
+                # Show provider selection menu
+                self.console.print("\n[bold]Select provider:[/bold]")
+                for idx, prov in enumerate(available_providers, 1):
+                    current_mode = self.rotation_mgr.get_effective_mode(prov)
+                    mode_display = (
+                        f"[green]{current_mode}[/green]"
+                        if current_mode == "sequential"
+                        else f"[blue]{current_mode}[/blue]"
+                    )
+                    self.console.print(f"   {idx}. {prov} ({mode_display})")
+                self.console.print(
+                    f"   {len(available_providers) + 1}. Enter custom provider name"
+                )
+
+                choice_idx = IntPrompt.ask(
+                    "Select option",
+                    choices=[str(i) for i in range(1, len(available_providers) + 2)],
+                )
+
+                if choice_idx == len(available_providers) + 1:
+                    provider = Prompt.ask("Provider name").strip().lower()
+                else:
+                    provider = available_providers[choice_idx - 1]
+
+                if provider:
+                    current_mode = self.rotation_mgr.get_effective_mode(provider)
+                    self.console.print(
+                        f"\nCurrent mode for {provider}: [cyan]{current_mode}[/cyan]"
+                    )
+                    self.console.print("\nSelect new rotation mode:")
+                    self.console.print(
+                        "   1. [blue]balanced[/blue] - Rotate credentials evenly"
+                    )
+                    self.console.print(
+                        "   2. [green]sequential[/green] - Use until exhausted"
+                    )
+
+                    mode_choice = Prompt.ask(
+                        "Select mode", choices=["1", "2"], show_choices=False
+                    )
+                    new_mode = "balanced" if mode_choice == "1" else "sequential"
+
+                    self.rotation_mgr.set_mode(provider, new_mode)
+                    self.console.print(
+                        f"\n[green]✅ Rotation mode for '{provider}' set to {new_mode}![/green]"
+                    )
+                    input("\nPress Enter to continue...")
+
+            elif choice == "2":
+                if not modes:
+                    self.console.print(
+                        "\n[yellow]No custom rotation modes to reset[/yellow]"
+                    )
+                    input("\nPress Enter to continue...")
+                    continue
+
+                # Show numbered list
+                self.console.print(
+                    "\n[bold]Select provider to reset to default:[/bold]"
+                )
+                modes_list = list(modes.keys())
+                for idx, prov in enumerate(modes_list, 1):
+                    default_mode = self.rotation_mgr.get_default_mode(prov)
+                    self.console.print(
+                        f"   {idx}. {prov} (will reset to: {default_mode})"
+                    )
+
+                choice_idx = IntPrompt.ask(
+                    "Select option",
+                    choices=[str(i) for i in range(1, len(modes_list) + 1)],
+                )
+                provider = modes_list[choice_idx - 1]
+                default_mode = self.rotation_mgr.get_default_mode(provider)
+
+                if Confirm.ask(f"Reset '{provider}' to default mode ({default_mode})?"):
+                    self.rotation_mgr.remove_mode(provider)
+                    self.console.print(
+                        f"\n[green]✅ Rotation mode for '{provider}' reset to default ({default_mode})![/green]"
+                    )
+                    input("\nPress Enter to continue...")
+
+            elif choice == "3":
+                break
+
     def manage_concurrency_limits(self):
         """Manage concurrency limits"""
         while True:
             clear_screen()
-            
+
             limits = self.concurrency_mgr.get_current_limits()
-            
-            self.console.print(Panel.fit(
-                "[bold cyan]⚡ Concurrency Limits Configuration[/bold cyan]",
-                border_style="cyan"
-            ))
-            
+
+            self.console.print(
+                Panel.fit(
+                    "[bold cyan]⚡ Concurrency Limits Configuration[/bold cyan]",
+                    border_style="cyan",
+                )
+            )
+
             self.console.print()
             self.console.print("[bold]📋 Current Concurrency Settings[/bold]")
             self.console.print("━" * 70)
-            
+
             if limits:
                 for provider, limit in limits.items():
                     self.console.print(f"   • {provider:15} {limit} requests/key")
                 self.console.print(f"   • Default:        1 request/key (all others)")
             else:
                 self.console.print("   • Default:        1 request/key (all providers)")
-            
+
             self.console.print()
             self.console.print("━" * 70)
             self.console.print()
@@ -1046,96 +1404,128 @@ class SettingsTool:
             self.console.print("   2. ✏️  Edit Existing Limit")
             self.console.print("   3. 🗑️  Remove Limit (reset to default)")
             self.console.print("   4. ↩️  Back to Settings Menu")
-            
+
             self.console.print()
             self.console.print("━" * 70)
             self.console.print()
-            
-            choice = Prompt.ask("Select option", choices=["1", "2", "3", "4"], show_choices=False)
-            
+
+            choice = Prompt.ask(
+                "Select option", choices=["1", "2", "3", "4"], show_choices=False
+            )
+
             if choice == "1":
                 # Get available providers
                 available_providers = self.get_available_providers()
-                
+
                 if not available_providers:
-                    self.console.print("\n[yellow]No providers with credentials found. Please add credentials first.[/yellow]")
+                    self.console.print(
+                        "\n[yellow]No providers with credentials found. Please add credentials first.[/yellow]"
+                    )
                     input("\nPress Enter to continue...")
                     continue
-                
+
                 # Show provider selection menu
                 self.console.print("\n[bold]Select provider:[/bold]")
                 for idx, prov in enumerate(available_providers, 1):
                     self.console.print(f"   {idx}. {prov}")
-                self.console.print(f"   {len(available_providers) + 1}. Enter custom provider name")
-                
-                choice_idx = IntPrompt.ask("Select option", choices=[str(i) for i in range(1, len(available_providers) + 2)])
-                
+                self.console.print(
+                    f"   {len(available_providers) + 1}. Enter custom provider name"
+                )
+
+                choice_idx = IntPrompt.ask(
+                    "Select option",
+                    choices=[str(i) for i in range(1, len(available_providers) + 2)],
+                )
+
                 if choice_idx == len(available_providers) + 1:
                     provider = Prompt.ask("Provider name").strip().lower()
                 else:
                     provider = available_providers[choice_idx - 1]
-                
+
                 if provider:
-                    limit = IntPrompt.ask("Max concurrent requests per key (1-100)", default=1)
+                    limit = IntPrompt.ask(
+                        "Max concurrent requests per key (1-100)", default=1
+                    )
                     if 1 <= limit <= 100:
                         self.concurrency_mgr.set_limit(provider, limit)
-                        self.console.print(f"\n[green]✅ Concurrency limit set for '{provider}': {limit} requests/key[/green]")
+                        self.console.print(
+                            f"\n[green]✅ Concurrency limit set for '{provider}': {limit} requests/key[/green]"
+                        )
                     else:
-                        self.console.print("\n[red]❌ Limit must be between 1-100[/red]")
+                        self.console.print(
+                            "\n[red]❌ Limit must be between 1-100[/red]"
+                        )
                     input("\nPress Enter to continue...")
-            
+
             elif choice == "2":
                 if not limits:
                     self.console.print("\n[yellow]No limits to edit[/yellow]")
                     input("\nPress Enter to continue...")
                     continue
-                
+
                 # Show numbered list
                 self.console.print("\n[bold]Select provider to edit:[/bold]")
                 limits_list = list(limits.keys())
                 for idx, prov in enumerate(limits_list, 1):
                     self.console.print(f"   {idx}. {prov}")
-                
-                choice_idx = IntPrompt.ask("Select option", choices=[str(i) for i in range(1, len(limits_list) + 1)])
+
+                choice_idx = IntPrompt.ask(
+                    "Select option",
+                    choices=[str(i) for i in range(1, len(limits_list) + 1)],
+                )
                 provider = limits_list[choice_idx - 1]
                 current_limit = limits.get(provider, 1)
-                
+
                 self.console.print(f"\nCurrent limit: {current_limit} requests/key")
-                new_limit = IntPrompt.ask("New limit (1-100) [press Enter to keep current]", default=current_limit)
-                
+                new_limit = IntPrompt.ask(
+                    "New limit (1-100) [press Enter to keep current]",
+                    default=current_limit,
+                )
+
                 if 1 <= new_limit <= 100:
                     if new_limit != current_limit:
                         self.concurrency_mgr.set_limit(provider, new_limit)
-                        self.console.print(f"\n[green]✅ Concurrency limit updated for '{provider}': {new_limit} requests/key[/green]")
+                        self.console.print(
+                            f"\n[green]✅ Concurrency limit updated for '{provider}': {new_limit} requests/key[/green]"
+                        )
                     else:
                         self.console.print("\n[yellow]No changes made[/yellow]")
                 else:
                     self.console.print("\n[red]Limit must be between 1-100[/red]")
                 input("\nPress Enter to continue...")
-            
+
             elif choice == "3":
                 if not limits:
                     self.console.print("\n[yellow]No limits to remove[/yellow]")
                     input("\nPress Enter to continue...")
                     continue
-                
+
                 # Show numbered list
-                self.console.print("\n[bold]Select provider to remove limit from:[/bold]")
+                self.console.print(
+                    "\n[bold]Select provider to remove limit from:[/bold]"
+                )
                 limits_list = list(limits.keys())
                 for idx, prov in enumerate(limits_list, 1):
                     self.console.print(f"   {idx}. {prov}")
-                
-                choice_idx = IntPrompt.ask("Select option", choices=[str(i) for i in range(1, len(limits_list) + 1)])
+
+                choice_idx = IntPrompt.ask(
+                    "Select option",
+                    choices=[str(i) for i in range(1, len(limits_list) + 1)],
+                )
                 provider = limits_list[choice_idx - 1]
-                
-                if Confirm.ask(f"Remove concurrency limit for '{provider}' (reset to default 1)?"):
+
+                if Confirm.ask(
+                    f"Remove concurrency limit for '{provider}' (reset to default 1)?"
+                ):
                     self.concurrency_mgr.remove_limit(provider)
-                    self.console.print(f"\n[green]✅ Limit removed for '{provider}' - using default (1 request/key)[/green]")
+                    self.console.print(
+                        f"\n[green]✅ Limit removed for '{provider}' - using default (1 request/key)[/green]"
+                    )
                     input("\nPress Enter to continue...")
-            
+
             elif choice == "4":
                 break
-    
+
     def save_and_exit(self):
         """Save pending changes and exit"""
         if self.settings.has_pending():
@@ -1150,9 +1540,9 @@ class SettingsTool:
         else:
             self.console.print("\n[dim]No changes to save[/dim]")
             input("\nPress Enter to return to launcher...")
-        
+
         self.running = False
-    
+
     def exit_without_saving(self):
         """Exit without saving"""
         if self.settings.has_pending():

src/rotator_library/client.py

@@ -139,8 +139,28 @@ class RotatingClient:
         self.max_retries = max_retries
         self.global_timeout = global_timeout
         self.abort_on_callback_error = abort_on_callback_error
+
+        # Build provider rotation modes map
+        # Each provider can specify its preferred rotation mode ("balanced" or "sequential")
+        provider_rotation_modes = {}
+        for provider in self.all_credentials.keys():
+            provider_class = self._provider_plugins.get(provider)
+            if provider_class and hasattr(provider_class, "get_rotation_mode"):
+                # Use class method to get rotation mode (checks env var + class default)
+                mode = provider_class.get_rotation_mode(provider)
+            else:
+                # Fallback: check environment variable directly
+                env_key = f"ROTATION_MODE_{provider.upper()}"
+                mode = os.getenv(env_key, "balanced")
+
+            provider_rotation_modes[provider] = mode
+            if mode != "balanced":
+                lib_logger.info(f"Provider '{provider}' using rotation mode: {mode}")
+
         self.usage_manager = UsageManager(
-            file_path=usage_file_path, rotation_tolerance=rotation_tolerance
+            file_path=usage_file_path,
+            rotation_tolerance=rotation_tolerance,
+            provider_rotation_modes=provider_rotation_modes,
         )
         self._model_list_cache = {}
         self._provider_plugins = PROVIDER_PLUGINS
@@ -1070,7 +1090,7 @@ class RotatingClient:
                                 if request
                                 else {},
                             )
-                            classified_error = classify_error(e)
+                            classified_error = classify_error(e, provider=provider)
 
                             # Extract a clean error message for the user-facing log
                             error_message = str(e).split("\n")[0]
@@ -1114,7 +1134,7 @@ class RotatingClient:
                                 if request
                                 else {},
                             )
-                            classified_error = classify_error(e)
+                            classified_error = classify_error(e, provider=provider)
                             error_message = str(e).split("\n")[0]
 
                             # Provider-level error: don't increment consecutive failures
@@ -1170,7 +1190,7 @@ class RotatingClient:
                                 else {},
                             )
 
-                            classified_error = classify_error(e)
+                            classified_error = classify_error(e, provider=provider)
                             error_message = str(e).split("\n")[0]
 
                             lib_logger.warning(
@@ -1239,7 +1259,7 @@ class RotatingClient:
                                 )
                                 raise last_exception
 
-                            classified_error = classify_error(e)
+                            classified_error = classify_error(e, provider=provider)
                             error_message = str(e).split("\n")[0]
 
                             lib_logger.warning(
@@ -1566,7 +1586,9 @@ class RotatingClient:
                                 last_exception = e
                                 # If the exception is our custom wrapper, unwrap the original error
                                 original_exc = getattr(e, "data", e)
-                                classified_error = classify_error(original_exc)
+                                classified_error = classify_error(
+                                    original_exc, provider=provider
+                                )
                                 error_message = str(original_exc).split("\n")[0]
 
                                 log_failure(
@@ -1623,7 +1645,7 @@ class RotatingClient:
                                     if request
                                     else {},
                                 )
-                                classified_error = classify_error(e)
+                                classified_error = classify_error(e, provider=provider)
                                 error_message = str(e).split("\n")[0]
 
                                 # Provider-level error: don't increment consecutive failures
@@ -1673,7 +1695,7 @@ class RotatingClient:
                                     if request
                                     else {},
                                 )
-                                classified_error = classify_error(e)
+                                classified_error = classify_error(e, provider=provider)
                                 error_message = str(e).split("\n")[0]
 
                                 # Record in accumulator
@@ -1812,7 +1834,9 @@ class RotatingClient:
                             cleaned_str = None
                             # The actual exception might be wrapped in our StreamedAPIError.
                             original_exc = getattr(e, "data", e)
-                            classified_error = classify_error(original_exc)
+                            classified_error = classify_error(
+                                original_exc, provider=provider
+                            )
 
                             # Check if this error should trigger rotation
                             if not should_rotate_on_error(classified_error):
@@ -1939,7 +1963,7 @@ class RotatingClient:
                                 if request
                                 else {},
                             )
-                            classified_error = classify_error(e)
+                            classified_error = classify_error(e, provider=provider)
                             error_message_text = str(e).split("\n")[0]
 
                             # Record error in accumulator (server errors are transient, not abnormal)
@@ -1990,7 +2014,7 @@ class RotatingClient:
                                 if request
                                 else {},
                             )
-                            classified_error = classify_error(e)
+                            classified_error = classify_error(e, provider=provider)
                             error_message_text = str(e).split("\n")[0]
 
                             # Record error in accumulator
@@ -2232,7 +2256,7 @@ class RotatingClient:
                     self._model_list_cache[provider] = final_models
                     return final_models
                 except Exception as e:
-                    classified_error = classify_error(e)
+                    classified_error = classify_error(e, provider=provider)
                     cred_display = mask_credential(credential)
                     lib_logger.debug(
                         f"Failed to get models for provider {provider} with credential {cred_display}: {classified_error.error_type}. Trying next credential."

src/rotator_library/error_handler.py

@@ -1,6 +1,7 @@
 import re
 import json
 import os
+import logging
 from typing import Optional, Dict, Any
 import httpx
 
@@ -17,6 +18,8 @@ from litellm.exceptions import (
     ContextWindowExceededError,
 )
 
+lib_logger = logging.getLogger("rotator_library")
+
 
 def _parse_duration_string(duration_str: str) -> Optional[int]:
     """
@@ -513,11 +516,15 @@ def get_retry_after(error: Exception) -> Optional[int]:
     return None
 
 
-def classify_error(e: Exception) -> ClassifiedError:
+def classify_error(e: Exception, provider: Optional[str] = None) -> ClassifiedError:
     """
     Classifies an exception into a structured ClassifiedError object.
     Now handles both litellm and httpx exceptions.
 
+    If provider is specified and has a parse_quota_error() method,
+    attempts provider-specific error parsing first before falling back
+    to generic classification.
+
     Error types and their typical handling:
     - rate_limit (429): Rotate key, may retry with backoff
     - server_error (5xx): Retry with backoff, then rotate
@@ -528,7 +535,60 @@ def classify_error(e: Exception) -> ClassifiedError:
     - context_window_exceeded: Don't retry - request too large
     - api_connection: Retry with backoff, then rotate
     - unknown: Rotate key (safer to try another)
+
+    Args:
+        e: The exception to classify
+        provider: Optional provider name for provider-specific error parsing
+
+    Returns:
+        ClassifiedError with error_type, status_code, retry_after, etc.
     """
+    # Try provider-specific parsing first for 429/rate limit errors
+    if provider:
+        try:
+            from .providers import PROVIDER_PLUGINS
+
+            provider_class = PROVIDER_PLUGINS.get(provider)
+
+            if provider_class and hasattr(provider_class, "parse_quota_error"):
+                # Get error body if available
+                error_body = None
+                if hasattr(e, "response") and hasattr(e.response, "text"):
+                    try:
+                        error_body = e.response.text
+                    except Exception:
+                        pass
+                elif hasattr(e, "body"):
+                    error_body = str(e.body)
+
+                quota_info = provider_class.parse_quota_error(e, error_body)
+
+                if quota_info and quota_info.get("retry_after"):
+                    retry_after = quota_info["retry_after"]
+                    reason = quota_info.get("reason", "QUOTA_EXHAUSTED")
+                    reset_ts = quota_info.get("reset_timestamp")
+
+                    # Log the parsed result with human-readable duration
+                    hours = retry_after / 3600
+                    lib_logger.info(
+                        f"Provider '{provider}' parsed quota error: "
+                        f"retry_after={retry_after}s ({hours:.1f}h), reason={reason}"
+                        + (f", resets at {reset_ts}" if reset_ts else "")
+                    )
+
+                    return ClassifiedError(
+                        error_type="quota_exceeded",
+                        original_exception=e,
+                        status_code=429,
+                        retry_after=retry_after,
+                    )
+        except Exception as parse_error:
+            lib_logger.debug(
+                f"Provider-specific error parsing failed for '{provider}': {parse_error}"
+            )
+            # Fall through to generic classification
+
+    # Generic classification logic
     status_code = getattr(e, "status_code", None)
 
     if isinstance(e, httpx.HTTPStatusError):  # [NEW] Handle httpx errors first

src/rotator_library/providers/antigravity_provider.py

@@ -494,6 +494,147 @@ class AntigravityProvider(AntigravityAuthBase, ProviderInterface):
 
     skip_cost_calculation = True
 
+    # Sequential mode by default - preserves thinking signature caches between requests
+    default_rotation_mode: str = "sequential"
+
+    @staticmethod
+    def parse_quota_error(
+        error: Exception, error_body: Optional[str] = None
+    ) -> Optional[Dict[str, Any]]:
+        """
+        Parse Antigravity/Google RPC quota errors.
+
+        Handles the Google Cloud API error format with ErrorInfo and RetryInfo details.
+
+        Example error format:
+        {
+          "error": {
+            "code": 429,
+            "details": [
+              {
+                "@type": "type.googleapis.com/google.rpc.ErrorInfo",
+                "reason": "QUOTA_EXHAUSTED",
+                "metadata": {
+                  "quotaResetDelay": "143h4m52.730699158s",
+                  "quotaResetTimeStamp": "2025-12-11T22:53:16Z"
+                }
+              },
+              {
+                "@type": "type.googleapis.com/google.rpc.RetryInfo",
+                "retryDelay": "515092.730699158s"
+              }
+            ]
+          }
+        }
+
+        Args:
+            error: The caught exception
+            error_body: Optional raw response body string
+
+        Returns:
+            None if not a parseable quota error, otherwise:
+            {
+                "retry_after": int,
+                "reason": str,
+                "reset_timestamp": str | None,
+            }
+        """
+        import re as regex_module
+
+        def parse_duration(duration_str: str) -> Optional[int]:
+            """Parse duration strings like '143h4m52.73s' or '515092.73s' to seconds."""
+            if not duration_str:
+                return None
+
+            # Handle pure seconds format: "515092.730699158s"
+            pure_seconds_match = regex_module.match(r"^([\d.]+)s$", duration_str)
+            if pure_seconds_match:
+                return int(float(pure_seconds_match.group(1)))
+
+            # Handle compound format: "143h4m52.730699158s"
+            total_seconds = 0
+            patterns = [
+                (r"(\d+)h", 3600),  # hours
+                (r"(\d+)m", 60),  # minutes
+                (r"([\d.]+)s", 1),  # seconds
+            ]
+            for pattern, multiplier in patterns:
+                match = regex_module.search(pattern, duration_str)
+                if match:
+                    total_seconds += float(match.group(1)) * multiplier
+
+            return int(total_seconds) if total_seconds > 0 else None
+
+        # Get error body from exception if not provided
+        body = error_body
+        if not body:
+            # Try to extract from various exception attributes
+            if hasattr(error, "response") and hasattr(error.response, "text"):
+                body = error.response.text
+            elif hasattr(error, "body"):
+                body = str(error.body)
+            elif hasattr(error, "message"):
+                body = str(error.message)
+            else:
+                body = str(error)
+
+        # Try to find JSON in the body
+        try:
+            # Handle cases where JSON is embedded in a larger string
+            json_match = regex_module.search(r"\{[\s\S]*\}", body)
+            if not json_match:
+                return None
+
+            data = json.loads(json_match.group(0))
+        except (json.JSONDecodeError, AttributeError, TypeError):
+            return None
+
+        # Navigate to error.details
+        error_obj = data.get("error", data)
+        details = error_obj.get("details", [])
+
+        if not details:
+            return None
+
+        result = {
+            "retry_after": None,
+            "reason": None,
+            "reset_timestamp": None,
+        }
+
+        for detail in details:
+            detail_type = detail.get("@type", "")
+
+            # Parse RetryInfo - most authoritative source for retry delay
+            if "RetryInfo" in detail_type:
+                retry_delay = detail.get("retryDelay")
+                if retry_delay:
+                    parsed = parse_duration(retry_delay)
+                    if parsed:
+                        result["retry_after"] = parsed
+
+            # Parse ErrorInfo - contains reason and quota reset metadata
+            elif "ErrorInfo" in detail_type:
+                result["reason"] = detail.get("reason")
+                metadata = detail.get("metadata", {})
+
+                # Get quotaResetDelay as fallback if RetryInfo not present
+                if not result["retry_after"]:
+                    quota_delay = metadata.get("quotaResetDelay")
+                    if quota_delay:
+                        parsed = parse_duration(quota_delay)
+                        if parsed:
+                            result["retry_after"] = parsed
+
+                # Capture reset timestamp for logging
+                result["reset_timestamp"] = metadata.get("quotaResetTimeStamp")
+
+        # Return None if we couldn't extract retry_after
+        if not result["retry_after"]:
+            return None
+
+        return result
+
     def __init__(self):
         super().__init__()
         self.model_definitions = ModelDefinitions()

src/rotator_library/providers/gemini_cli_provider.py

@@ -186,6 +186,31 @@ def _env_int(key: str, default: int) -> int:
 class GeminiCliProvider(GeminiAuthBase, ProviderInterface):
     skip_cost_calculation = True
 
+    # Balanced by default - Gemini CLI has short cooldowns (seconds, not hours)
+    default_rotation_mode: str = "balanced"
+
+    @staticmethod
+    def parse_quota_error(
+        error: Exception, error_body: Optional[str] = None
+    ) -> Optional[Dict[str, Any]]:
+        """
+        Parse Gemini CLI quota errors.
+
+        Uses the same Google RPC format as Antigravity but typically has
+        much shorter cooldown durations (seconds to minutes, not hours).
+
+        Args:
+            error: The caught exception
+            error_body: Optional raw response body string
+
+        Returns:
+            Same format as AntigravityProvider.parse_quota_error()
+        """
+        # Reuse the same parsing logic as Antigravity since both use Google RPC format
+        from .antigravity_provider import AntigravityProvider
+
+        return AntigravityProvider.parse_quota_error(error, error_body)
+
     def __init__(self):
         super().__init__()
         self.model_definitions = ModelDefinitions()

src/rotator_library/providers/provider_interface.py

@@ -1,5 +1,6 @@
 from abc import ABC, abstractmethod
 from typing import List, Dict, Any, Optional, AsyncGenerator, Union
+import os
 import httpx
 import litellm
 
@@ -12,6 +13,11 @@ class ProviderInterface(ABC):
 
     skip_cost_calculation: bool = False
 
+    # Default rotation mode for this provider ("balanced" or "sequential")
+    # - "balanced": Rotate credentials to distribute load evenly
+    # - "sequential": Use one credential until exhausted, then switch to next
+    default_rotation_mode: str = "balanced"
+
     @abstractmethod
     async def get_models(self, api_key: str, client: httpx.AsyncClient) -> List[str]:
         """
@@ -153,3 +159,69 @@ class ProviderInterface(ABC):
             Tier name string (e.g., "free-tier", "paid-tier") or None if unknown
         """
         return None
+
+    # =========================================================================
+    # Sequential Rotation Support
+    # =========================================================================
+
+    @classmethod
+    def get_rotation_mode(cls, provider_name: str) -> str:
+        """
+        Get the rotation mode for this provider.
+
+        Checks ROTATION_MODE_{PROVIDER} environment variable first,
+        then falls back to the class's default_rotation_mode.
+
+        Args:
+            provider_name: The provider name (e.g., "antigravity", "gemini_cli")
+
+        Returns:
+            "balanced" or "sequential"
+        """
+        env_key = f"ROTATION_MODE_{provider_name.upper()}"
+        return os.getenv(env_key, cls.default_rotation_mode)
+
+    @staticmethod
+    def parse_quota_error(
+        error: Exception, error_body: Optional[str] = None
+    ) -> Optional[Dict[str, Any]]:
+        """
+        Parse a quota/rate-limit error and extract structured information.
+
+        Providers should override this method to handle their specific error formats.
+        This allows the error_handler to use provider-specific parsing when available,
+        falling back to generic parsing otherwise.
+
+        Args:
+            error: The caught exception
+            error_body: Optional raw response body string
+
+        Returns:
+            None if not a parseable quota error, otherwise:
+            {
+                "retry_after": int,  # seconds until quota resets
+                "reason": str,       # e.g., "QUOTA_EXHAUSTED", "RATE_LIMITED"
+                "reset_timestamp": str | None,  # ISO timestamp if available
+            }
+        """
+        return None  # Default: no provider-specific parsing
+
+    # TODO: Implement provider-specific quota reset schedules
+    # Different providers have different quota reset periods:
+    # - Most providers: Daily reset at a specific time
+    # - Antigravity free tier: Weekly reset
+    # - Antigravity paid tier: 5-hour rolling window
+    #
+    # Future implementation should add:
+    # @classmethod
+    # def get_quota_reset_behavior(cls) -> Dict[str, Any]:
+    #     """
+    #     Get provider-specific quota reset behavior.
+    #     Returns:
+    #         {
+    #             "type": "daily" | "weekly" | "rolling",
+    #             "reset_time_utc": "03:00",  # For daily/weekly
+    #             "rolling_hours": 5,          # For rolling
+    #         }
+    #     """
+    #     return {"type": "daily", "reset_time_utc": "03:00"}

src/rotator_library/usage_manager.py

@@ -5,7 +5,7 @@ import logging
 import asyncio
 import random
 from datetime import date, datetime, timezone, time as dt_time
-from typing import Any, Dict, List, Optional, Set
+from typing import Any, Dict, List, Optional, Set, Tuple
 import aiofiles
 import litellm
 
@@ -42,6 +42,10 @@ class UsageManager:
 
     This ensures lower-usage credentials are preferred while tolerance controls how much
     randomness is introduced into the selection process.
+
+    Additionally, providers can specify a rotation mode:
+    - "balanced" (default): Rotate credentials to distribute load evenly
+    - "sequential": Use one credential until exhausted (preserves caching)
     """
 
     def __init__(
@@ -49,6 +53,7 @@ class UsageManager:
         file_path: str = "key_usage.json",
         daily_reset_time_utc: Optional[str] = "03:00",
         rotation_tolerance: float = 0.0,
+        provider_rotation_modes: Optional[Dict[str, str]] = None,
     ):
         """
         Initialize the UsageManager.
@@ -60,9 +65,13 @@ class UsageManager:
                 - 0.0: Deterministic, least-used credential always selected
                 - tolerance = 2.0 - 4.0 (default, recommended): Balanced randomness, can pick credentials within 2 uses of max
                 - 5.0+: High randomness, more unpredictable selection patterns
+            provider_rotation_modes: Dict mapping provider names to rotation modes.
+                - "balanced": Rotate credentials to distribute load evenly (default)
+                - "sequential": Use one credential until exhausted (preserves caching)
         """
         self.file_path = file_path
         self.rotation_tolerance = rotation_tolerance
+        self.provider_rotation_modes = provider_rotation_modes or {}
         self.key_states: Dict[str, Dict[str, Any]] = {}
 
         self._data_lock = asyncio.Lock()
@@ -81,6 +90,72 @@ class UsageManager:
         else:
             self.daily_reset_time_utc = None
 
+    def _get_rotation_mode(self, provider: str) -> str:
+        """
+        Get the rotation mode for a provider.
+
+        Args:
+            provider: Provider name (e.g., "antigravity", "gemini_cli")
+
+        Returns:
+            "balanced" or "sequential"
+        """
+        return self.provider_rotation_modes.get(provider, "balanced")
+
+    def _select_sequential(
+        self,
+        candidates: List[Tuple[str, int]],
+        credential_priorities: Optional[Dict[str, int]] = None,
+    ) -> str:
+        """
+        Select credential in strict sequential order for cache-preserving rotation.
+
+        This method ensures the same credential is reused until it hits a cooldown,
+        which preserves provider-side caching (e.g., thinking signature caches).
+
+        Selection logic:
+        1. Sort by priority (lowest number = highest priority)
+        2. Within same priority, sort by last_used_ts (most recent first = sticky)
+        3. Return the first candidate
+
+        Args:
+            candidates: List of (credential_id, usage_count) tuples
+            credential_priorities: Optional dict mapping credentials to priority levels
+
+        Returns:
+            Selected credential ID
+        """
+        if not candidates:
+            raise ValueError("Cannot select from empty candidate list")
+
+        if len(candidates) == 1:
+            return candidates[0][0]
+
+        def sort_key(item: Tuple[str, int]) -> Tuple[int, float]:
+            cred, _ = item
+            # Priority: lower is better (1 = highest priority)
+            priority = (
+                credential_priorities.get(cred, 999) if credential_priorities else 999
+            )
+            # Last used: higher (more recent) is better for stickiness
+            last_used = (
+                self._usage_data.get(cred, {}).get("last_used_ts", 0)
+                if self._usage_data
+                else 0
+            )
+            # Negative last_used so most recent sorts first
+            return (priority, -last_used)
+
+        sorted_candidates = sorted(candidates, key=sort_key)
+        selected = sorted_candidates[0][0]
+
+        lib_logger.debug(
+            f"Sequential selection: chose {mask_credential(selected)} "
+            f"(priority={credential_priorities.get(selected, 999) if credential_priorities else 'N/A'})"
+        )
+
+        return selected
+
     async def _lazy_init(self):
         """Initializes the usage data by loading it from the file asynchronously."""
         async with self._init_lock:
@@ -144,14 +219,63 @@ class UsageManager:
                     )
                     needs_saving = True
 
-                    # Reset cooldowns
-                    data["model_cooldowns"] = {}
-                    data["key_cooldown_until"] = None
+                    # Reset cooldowns - BUT preserve unexpired long-term cooldowns
+                    # This is important for quota errors with long cooldowns (e.g., 143 hours)
+                    now_ts = time.time()
+                    if "model_cooldowns" in data:
+                        active_cooldowns = {
+                            model: end_time
+                            for model, end_time in data["model_cooldowns"].items()
+                            if end_time > now_ts
+                        }
+                        if active_cooldowns:
+                            # Calculate how long the longest cooldown has remaining
+                            max_remaining = max(
+                                end_time - now_ts
+                                for end_time in active_cooldowns.values()
+                            )
+                            hours_remaining = max_remaining / 3600
+                            lib_logger.info(
+                                f"Preserving {len(active_cooldowns)} active cooldown(s) "
+                                f"for key {mask_credential(key)} during daily reset "
+                                f"(longest: {hours_remaining:.1f}h remaining)"
+                            )
+                        data["model_cooldowns"] = active_cooldowns
+                    else:
+                        data["model_cooldowns"] = {}
+
+                    # Clear key-level cooldown only if expired
+                    if data.get("key_cooldown_until"):
+                        if data["key_cooldown_until"] <= now_ts:
+                            data["key_cooldown_until"] = None
+                        else:
+                            hours_remaining = (
+                                data["key_cooldown_until"] - now_ts
+                            ) / 3600
+                            lib_logger.info(
+                                f"Preserving key-level cooldown for {mask_credential(key)} "
+                                f"during daily reset ({hours_remaining:.1f}h remaining)"
+                            )
+                    else:
+                        data["key_cooldown_until"] = None
 
                     # Reset consecutive failures
                     if "failures" in data:
                         data["failures"] = {}
 
+                    # TODO: Implement provider-specific reset schedules
+                    # Different providers have different quota reset periods:
+                    # - Most providers: Daily reset at daily_reset_time_utc
+                    # - Antigravity free tier: Weekly reset
+                    # - Antigravity paid tier: 5-hour rolling window
+                    #
+                    # Future implementation should:
+                    # 1. Group credentials by provider (extracted from key path or metadata)
+                    # 2. Check each provider's get_quota_reset_behavior()
+                    # 3. Apply provider-specific reset logic instead of universal daily reset
+                    #
+                    # For now, we preserve unexpired cooldowns which handles long cooldowns correctly.
+
                     # Archive global stats from the previous day's 'daily'
                     daily_data = data.get("daily", {})
                     if daily_data:
@@ -336,15 +460,30 @@ class UsageManager:
                         elif key_state["models_in_use"].get(model, 0) < max_concurrent:
                             tier2_keys.append((key, usage_count))
 
-                    # Apply weighted random selection or deterministic sorting
-                    selection_method = (
-                        "weighted-random"
-                        if self.rotation_tolerance > 0
-                        else "least-used"
-                    )
+                    # Determine selection method based on provider's rotation mode
+                    provider = model.split("/")[0] if "/" in model else ""
+                    rotation_mode = self._get_rotation_mode(provider)
 
-                    if self.rotation_tolerance > 0:
-                        # Weighted random selection within each tier
+                    if rotation_mode == "sequential":
+                        # Sequential mode: stick with same credential until exhausted
+                        selection_method = "sequential"
+                        if tier1_keys:
+                            selected_key = self._select_sequential(
+                                tier1_keys, credential_priorities
+                            )
+                            tier1_keys = [
+                                (k, u) for k, u in tier1_keys if k == selected_key
+                            ]
+                        if tier2_keys:
+                            selected_key = self._select_sequential(
+                                tier2_keys, credential_priorities
+                            )
+                            tier2_keys = [
+                                (k, u) for k, u in tier2_keys if k == selected_key
+                            ]
+                    elif self.rotation_tolerance > 0:
+                        # Balanced mode with weighted randomness
+                        selection_method = "weighted-random"
                         if tier1_keys:
                             selected_key = self._select_weighted_random(
                                 tier1_keys, self.rotation_tolerance
@@ -361,6 +500,7 @@ class UsageManager:
                             ]
                     else:
                         # Deterministic: sort by usage within each tier
+                        selection_method = "least-used"
                         tier1_keys.sort(key=lambda x: x[1])
                         tier2_keys.sort(key=lambda x: x[1])
 
@@ -452,13 +592,30 @@ class UsageManager:
                         elif key_state["models_in_use"].get(model, 0) < max_concurrent:
                             tier2_keys.append((key, usage_count))
 
-                # Apply weighted random selection or deterministic sorting
-                selection_method = (
-                    "weighted-random" if self.rotation_tolerance > 0 else "least-used"
-                )
+                # Determine selection method based on provider's rotation mode
+                provider = model.split("/")[0] if "/" in model else ""
+                rotation_mode = self._get_rotation_mode(provider)
 
-                if self.rotation_tolerance > 0:
-                    # Weighted random selection within each tier
+                if rotation_mode == "sequential":
+                    # Sequential mode: stick with same credential until exhausted
+                    selection_method = "sequential"
+                    if tier1_keys:
+                        selected_key = self._select_sequential(
+                            tier1_keys, credential_priorities
+                        )
+                        tier1_keys = [
+                            (k, u) for k, u in tier1_keys if k == selected_key
+                        ]
+                    if tier2_keys:
+                        selected_key = self._select_sequential(
+                            tier2_keys, credential_priorities
+                        )
+                        tier2_keys = [
+                            (k, u) for k, u in tier2_keys if k == selected_key
+                        ]
+                elif self.rotation_tolerance > 0:
+                    # Balanced mode with weighted randomness
+                    selection_method = "weighted-random"
                     if tier1_keys:
                         selected_key = self._select_weighted_random(
                             tier1_keys, self.rotation_tolerance
@@ -475,6 +632,7 @@ class UsageManager:
                         ]
                 else:
                     # Deterministic: sort by usage within each tier
+                    selection_method = "least-used"
                     tier1_keys.sort(key=lambda x: x[1])
                     tier2_keys.sort(key=lambda x: x[1])
 
@@ -726,10 +884,24 @@ class UsageManager:
             if classified_error.error_type in ["rate_limit", "quota_exceeded"]:
                 # Rate limit / Quota errors: use retry_after if available, otherwise default to 60s
                 cooldown_seconds = classified_error.retry_after or 60
-                lib_logger.info(
-                    f"Rate limit error on key {mask_credential(key)} for model {model}. "
-                    f"Using {'provided' if classified_error.retry_after else 'default'} retry_after: {cooldown_seconds}s"
-                )
+                if classified_error.retry_after:
+                    # Log with human-readable duration for provider-parsed cooldowns
+                    hours = cooldown_seconds / 3600
+                    if hours >= 1:
+                        lib_logger.info(
+                            f"Quota/rate limit on key {mask_credential(key)} for model {model}. "
+                            f"Applying provider-specified cooldown: {cooldown_seconds}s ({hours:.1f}h)"
+                        )
+                    else:
+                        lib_logger.info(
+                            f"Rate limit on key {mask_credential(key)} for model {model}. "
+                            f"Applying provider-specified cooldown: {cooldown_seconds}s"
+                        )
+                else:
+                    lib_logger.info(
+                        f"Rate limit on key {mask_credential(key)} for model {model}. "
+                        f"Using default cooldown: {cooldown_seconds}s"
+                    )
             elif classified_error.error_type == "authentication":
                 # Apply a 5-minute key-level lockout for auth errors
                 key_data["key_cooldown_until"] = time.time() + 300