refactor(core): 🔨 centralize path management for PyInstaller compatibility

Introduce a new `paths.py` utility module that provides centralized path resolution for all data files (logs, cache, OAuth credentials, usage data) with support for both PyInstaller EXE and script/library runtime modes.

- Add `get_default_root()` to auto-detect EXE directory when frozen, otherwise uses CWD
- Add `get_logs_dir()`, `get_cache_dir()`, `get_oauth_dir()`, and `get_data_file()` helpers
- Update `RotatingClient` to accept optional `data_dir` parameter for override capability
- Refactor all hardcoded path calculations to use centralized utilities
- Update `CredentialManager` to accept `oauth_dir` parameter
- Convert `failure_logger` to lazy initialization with configurable logs directory
- Migrate `UsageManager` to accept optional `file_path` with automatic defaults
- Update all provider-specific logging directories to use centralized helpers
- Refactor credential tool and settings tool to use path utilities
- Update main.py to load .env files from correct root directory in both modes

This change eliminates fragile `Path(__file__).resolve().parent` patterns and ensures all file operations work correctly when the application is packaged as a standalone executable.

src/proxy_app/detailed_logger.py

@@ -11,9 +11,15 @@ from rotator_library.utils.resilient_io import (
     safe_log_write,
     safe_mkdir,
 )
+from rotator_library.utils.paths import get_logs_dir
 
-LOGS_DIR = Path(__file__).resolve().parent.parent.parent / "logs"
-DETAILED_LOGS_DIR = LOGS_DIR / "detailed_logs"
+
+def _get_detailed_logs_dir() -> Path:
+    """Get the detailed logs directory, creating it if needed."""
+    logs_dir = get_logs_dir()
+    detailed_dir = logs_dir / "detailed_logs"
+    detailed_dir.mkdir(parents=True, exist_ok=True)
+    return detailed_dir
 
 
 class DetailedLogger:
@@ -31,7 +37,7 @@ class DetailedLogger:
         self.start_time = time.time()
         self.request_id = str(uuid.uuid4())
         timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
-        self.log_dir = DETAILED_LOGS_DIR / f"{timestamp}_{self.request_id}"
+        self.log_dir = _get_detailed_logs_dir() / f"{timestamp}_{self.request_id}"
         self.streaming = False
         self._dir_available = safe_mkdir(self.log_dir, logging)
 

src/proxy_app/main.py

@@ -51,12 +51,15 @@ _start_time = time.time()
 # Load all .env files from root folder (main .env first, then any additional *.env files)
 from dotenv import load_dotenv
 from glob import glob
+from rotator_library.utils.paths import get_default_root, get_logs_dir, get_data_file
+
+# Get the application root directory (EXE dir if frozen, else CWD)
+_root_dir = get_default_root()
 
 # Load main .env first
-load_dotenv()
+load_dotenv(_root_dir / ".env")
 
 # Load any additional .env files (e.g., antigravity_all_combined.env, gemini_cli_all_combined.env)
-_root_dir = Path.cwd()
 _env_files_found = list(_root_dir.glob("*.env"))
 for _env_file in sorted(_root_dir.glob("*.env")):
     if _env_file.name != ".env":  # Skip main .env (already loaded)
@@ -234,8 +237,7 @@ print(
 # Note: Debug logging will be added after logging configuration below
 
 # --- Logging Configuration ---
-LOG_DIR = Path(__file__).resolve().parent.parent.parent / "logs"
-LOG_DIR.mkdir(exist_ok=True)
+LOG_DIR = get_logs_dir(_root_dir)
 
 # Configure a console handler with color (INFO and above only, no DEBUG)
 console_handler = colorlog.StreamHandler(sys.stdout)
@@ -570,11 +572,11 @@ async def lifespan(app: FastAPI):
     )
 
     # Log loaded credentials summary (compact, always visible for deployment verification)
-    #_api_summary = ', '.join([f"{p}:{len(c)}" for p, c in api_keys.items()]) if api_keys else "none"
-    #_oauth_summary = ', '.join([f"{p}:{len(c)}" for p, c in oauth_credentials.items()]) if oauth_credentials else "none"
-    #_total_summary = ', '.join([f"{p}:{len(c)}" for p, c in client.all_credentials.items()])
-    #print(f"🔑 Credentials loaded: {_total_summary} (API: {_api_summary} | OAuth: {_oauth_summary})")
-    client.background_refresher.start() # Start the background task
+    # _api_summary = ', '.join([f"{p}:{len(c)}" for p, c in api_keys.items()]) if api_keys else "none"
+    # _oauth_summary = ', '.join([f"{p}:{len(c)}" for p, c in oauth_credentials.items()]) if oauth_credentials else "none"
+    # _total_summary = ', '.join([f"{p}:{len(c)}" for p, c in client.all_credentials.items()])
+    # print(f"🔑 Credentials loaded: {_total_summary} (API: {_api_summary} | OAuth: {_oauth_summary})")
+    client.background_refresher.start()  # Start the background task
     app.state.rotating_client = client
 
     # Warn if no provider credentials are configured
@@ -1263,8 +1265,8 @@ async def cost_estimate(request: Request, _=Depends(verify_api_key)):
 
 
 if __name__ == "__main__":
-    # Define ENV_FILE for onboarding checks
-    ENV_FILE = Path.cwd() / ".env"
+    # Define ENV_FILE for onboarding checks using centralized path
+    ENV_FILE = get_data_file(".env")
 
     # Check if launcher TUI should be shown (no arguments provided)
     if len(sys.argv) == 1:

src/proxy_app/settings_tool.py

@@ -12,6 +12,8 @@ from rich.prompt import Prompt, IntPrompt, Confirm
 from rich.panel import Panel
 from dotenv import set_key, unset_key
 
+from rotator_library.utils.paths import get_data_file
+
 console = Console()
 
 # Import default OAuth port values from provider modules
@@ -54,7 +56,7 @@ class AdvancedSettings:
     """Manages pending changes to .env"""
 
     def __init__(self):
-        self.env_file = Path.cwd() / ".env"
+        self.env_file = get_data_file(".env")
         self.pending_changes = {}  # key -> value (None means delete)
         self.load_current_settings()
 
@@ -561,7 +563,7 @@ class SettingsTool:
 
     def get_available_providers(self) -> List[str]:
         """Get list of providers that have credentials configured"""
-        env_file = Path.cwd() / ".env"
+        env_file = get_data_file(".env")
         providers = set()
 
         # Scan for providers with API keys from local .env
@@ -584,7 +586,9 @@ class SettingsTool:
                 pass
 
         # Also check for OAuth providers from files
-        oauth_dir = Path("oauth_creds")
+        from rotator_library.utils.paths import get_oauth_dir
+
+        oauth_dir = get_oauth_dir()
         if oauth_dir.exists():
             for file in oauth_dir.glob("*_oauth_*.json"):
                 provider = file.name.split("_oauth_")[0]

src/rotator_library/client.py

@@ -10,6 +10,7 @@ import litellm
 from litellm.exceptions import APIConnectionError
 from litellm.litellm_core_utils.token_counter import token_counter
 import logging
+from pathlib import Path
 from typing import List, Dict, Any, AsyncGenerator, Optional, Union
 
 lib_logger = logging.getLogger("rotator_library")
@@ -19,7 +20,7 @@ lib_logger = logging.getLogger("rotator_library")
 lib_logger.propagate = False
 
 from .usage_manager import UsageManager
-from .failure_logger import log_failure
+from .failure_logger import log_failure, configure_failure_logger
 from .error_handler import (
     PreRequestCallbackError,
     classify_error,
@@ -37,6 +38,7 @@ from .cooldown_manager import CooldownManager
 from .credential_manager import CredentialManager
 from .background_refresher import BackgroundRefresher
 from .model_definitions import ModelDefinitions
+from .utils.paths import get_default_root, get_logs_dir, get_oauth_dir, get_data_file
 
 
 class StreamedAPIError(Exception):
@@ -58,7 +60,7 @@ class RotatingClient:
         api_keys: Optional[Dict[str, List[str]]] = None,
         oauth_credentials: Optional[Dict[str, List[str]]] = None,
         max_retries: int = 2,
-        usage_file_path: str = "key_usage.json",
+        usage_file_path: Optional[Union[str, Path]] = None,
         configure_logging: bool = True,
         global_timeout: int = 30,
         abort_on_callback_error: bool = True,
@@ -68,6 +70,7 @@ class RotatingClient:
         enable_request_logging: bool = False,
         max_concurrent_requests_per_key: Optional[Dict[str, int]] = None,
         rotation_tolerance: float = 3.0,
+        data_dir: Optional[Union[str, Path]] = None,
     ):
         """
         Initialize the RotatingClient with intelligent credential rotation.
@@ -76,7 +79,7 @@ class RotatingClient:
             api_keys: Dictionary mapping provider names to lists of API keys
             oauth_credentials: Dictionary mapping provider names to OAuth credential paths
             max_retries: Maximum number of retry attempts per credential
-            usage_file_path: Path to store usage statistics
+            usage_file_path: Path to store usage statistics. If None, uses data_dir/key_usage.json
             configure_logging: Whether to configure library logging
             global_timeout: Global timeout for requests in seconds
             abort_on_callback_error: Whether to abort on pre-request callback errors
@@ -89,7 +92,18 @@ class RotatingClient:
                 - 0.0: Deterministic, least-used credential always selected
                 - 2.0 - 4.0 (default, recommended): Balanced randomness, can pick credentials within 2 uses of max
                 - 5.0+: High randomness, more unpredictable selection patterns
+            data_dir: Root directory for all data files (logs, cache, oauth_creds, key_usage.json).
+                      If None, auto-detects: EXE directory if frozen, else current working directory.
         """
+        # Resolve data_dir early - this becomes the root for all file operations
+        if data_dir is not None:
+            self.data_dir = Path(data_dir).resolve()
+        else:
+            self.data_dir = get_default_root()
+
+        # Configure failure logger to use correct logs directory
+        configure_failure_logger(get_logs_dir(self.data_dir))
+
         os.environ["LITELLM_LOG"] = "ERROR"
         litellm.set_verbose = False
         litellm.drop_params = True
@@ -124,7 +138,9 @@ class RotatingClient:
         if oauth_credentials:
             self.oauth_credentials = oauth_credentials
         else:
-            self.credential_manager = CredentialManager(os.environ)
+            self.credential_manager = CredentialManager(
+                os.environ, oauth_dir=get_oauth_dir(self.data_dir)
+            )
             self.oauth_credentials = self.credential_manager.discover_and_prepare()
         self.background_refresher = BackgroundRefresher(self)
         self.oauth_providers = set(self.oauth_credentials.keys())
@@ -242,8 +258,14 @@ class RotatingClient:
                 f"Provider '{provider}' sequential fallback multiplier: {fallback}x"
             )
 
+        # Resolve usage file path - use provided path or default to data_dir
+        if usage_file_path is not None:
+            resolved_usage_path = Path(usage_file_path)
+        else:
+            resolved_usage_path = self.data_dir / "key_usage.json"
+
         self.usage_manager = UsageManager(
-            file_path=usage_file_path,
+            file_path=resolved_usage_path,
             rotation_tolerance=rotation_tolerance,
             provider_rotation_modes=provider_rotation_modes,
             provider_plugins=PROVIDER_PLUGINS,

src/rotator_library/credential_manager.py

@@ -3,12 +3,11 @@ import re
 import shutil
 import logging
 from pathlib import Path
-from typing import Dict, List, Optional, Set
+from typing import Dict, List, Optional, Set, Union
 
-lib_logger = logging.getLogger('rotator_library')
+from .utils.paths import get_oauth_dir
 
-OAUTH_BASE_DIR = Path.cwd() / "oauth_creds"
-OAUTH_BASE_DIR.mkdir(exist_ok=True)
+lib_logger = logging.getLogger("rotator_library")
 
 # Standard directories where tools like `gemini login` store credentials.
 DEFAULT_OAUTH_DIRS = {
@@ -33,38 +32,53 @@ class CredentialManager:
     """
     Discovers OAuth credential files from standard locations, copies them locally,
     and updates the configuration to use the local paths.
-    
+
     Also discovers environment variable-based OAuth credentials for stateless deployments.
     Supports two env var formats:
-    
+
     1. Single credential (legacy): PROVIDER_ACCESS_TOKEN, PROVIDER_REFRESH_TOKEN
     2. Multiple credentials (numbered): PROVIDER_1_ACCESS_TOKEN, PROVIDER_2_ACCESS_TOKEN, etc.
-    
+
     When env-based credentials are detected, virtual paths like "env://provider/1" are created.
     """
-    def __init__(self, env_vars: Dict[str, str]):
+
+    def __init__(
+        self,
+        env_vars: Dict[str, str],
+        oauth_dir: Optional[Union[Path, str]] = None,
+    ):
+        """
+        Initialize the CredentialManager.
+
+        Args:
+            env_vars: Dictionary of environment variables (typically os.environ).
+            oauth_dir: Directory for storing OAuth credentials.
+                       If None, uses get_oauth_dir() which respects EXE vs script mode.
+        """
         self.env_vars = env_vars
+        self.oauth_base_dir = Path(oauth_dir) if oauth_dir else get_oauth_dir()
+        self.oauth_base_dir.mkdir(parents=True, exist_ok=True)
 
     def _discover_env_oauth_credentials(self) -> Dict[str, List[str]]:
         """
         Discover OAuth credentials defined via environment variables.
-        
+
         Supports two formats:
         1. Single credential: ANTIGRAVITY_ACCESS_TOKEN + ANTIGRAVITY_REFRESH_TOKEN
         2. Multiple credentials: ANTIGRAVITY_1_ACCESS_TOKEN + ANTIGRAVITY_1_REFRESH_TOKEN, etc.
-        
+
         Returns:
             Dict mapping provider name to list of virtual paths (e.g., "env://antigravity/1")
         """
         env_credentials: Dict[str, Set[str]] = {}
-        
+
         for provider, env_prefix in ENV_OAUTH_PROVIDERS.items():
             found_indices: Set[str] = set()
-            
+
             # Check for numbered credentials (PROVIDER_N_ACCESS_TOKEN pattern)
             # Pattern: ANTIGRAVITY_1_ACCESS_TOKEN, ANTIGRAVITY_2_ACCESS_TOKEN, etc.
             numbered_pattern = re.compile(rf"^{env_prefix}_(\d+)_ACCESS_TOKEN$")
-            
+
             for key in self.env_vars.keys():
                 match = numbered_pattern.match(key)
                 if match:
@@ -73,28 +87,34 @@ class CredentialManager:
                     refresh_key = f"{env_prefix}_{index}_REFRESH_TOKEN"
                     if refresh_key in self.env_vars and self.env_vars[refresh_key]:
                         found_indices.add(index)
-            
+
             # Check for legacy single credential (PROVIDER_ACCESS_TOKEN pattern)
             # Only use this if no numbered credentials exist
             if not found_indices:
                 access_key = f"{env_prefix}_ACCESS_TOKEN"
                 refresh_key = f"{env_prefix}_REFRESH_TOKEN"
-                if (access_key in self.env_vars and self.env_vars[access_key] and
-                    refresh_key in self.env_vars and self.env_vars[refresh_key]):
+                if (
+                    access_key in self.env_vars
+                    and self.env_vars[access_key]
+                    and refresh_key in self.env_vars
+                    and self.env_vars[refresh_key]
+                ):
                     # Use "0" as the index for legacy single credential
                     found_indices.add("0")
-            
+
             if found_indices:
                 env_credentials[provider] = found_indices
-                lib_logger.info(f"Found {len(found_indices)} env-based credential(s) for {provider}")
-        
+                lib_logger.info(
+                    f"Found {len(found_indices)} env-based credential(s) for {provider}"
+                )
+
         # Convert to virtual paths
         result: Dict[str, List[str]] = {}
         for provider, indices in env_credentials.items():
             # Sort indices numerically for consistent ordering
             sorted_indices = sorted(indices, key=lambda x: int(x))
             result[provider] = [f"env://{provider}/{idx}" for idx in sorted_indices]
-        
+
         return result
 
     def discover_and_prepare(self) -> Dict[str, List[str]]:
@@ -105,7 +125,9 @@ class CredentialManager:
         # These take priority for stateless deployments
         env_oauth_creds = self._discover_env_oauth_credentials()
         for provider, virtual_paths in env_oauth_creds.items():
-            lib_logger.info(f"Using {len(virtual_paths)} env-based credential(s) for {provider}")
+            lib_logger.info(
+                f"Using {len(virtual_paths)} env-based credential(s) for {provider}"
+            )
             final_config[provider] = virtual_paths
 
         # Extract OAuth file paths from environment variables
@@ -115,21 +137,29 @@ class CredentialManager:
                 provider = key.split("_OAUTH_")[0].lower()
                 if provider not in env_oauth_paths:
                     env_oauth_paths[provider] = []
-                if value: # Only consider non-empty values
+                if value:  # Only consider non-empty values
                     env_oauth_paths[provider].append(value)
 
         # PHASE 2: Discover file-based OAuth credentials
         for provider, default_dir in DEFAULT_OAUTH_DIRS.items():
             # Skip if already discovered from environment variables
             if provider in final_config:
-                lib_logger.debug(f"Skipping file discovery for {provider} - using env-based credentials")
+                lib_logger.debug(
+                    f"Skipping file discovery for {provider} - using env-based credentials"
+                )
                 continue
-            
+
             # Check for existing local credentials first. If found, use them and skip discovery.
-            local_provider_creds = sorted(list(OAUTH_BASE_DIR.glob(f"{provider}_oauth_*.json")))
+            local_provider_creds = sorted(
+                list(self.oauth_base_dir.glob(f"{provider}_oauth_*.json"))
+            )
             if local_provider_creds:
-                lib_logger.info(f"Found {len(local_provider_creds)} existing local credential(s) for {provider}. Skipping discovery.")
-                final_config[provider] = [str(p.resolve()) for p in local_provider_creds]
+                lib_logger.info(
+                    f"Found {len(local_provider_creds)} existing local credential(s) for {provider}. Skipping discovery."
+                )
+                final_config[provider] = [
+                    str(p.resolve()) for p in local_provider_creds
+                ]
                 continue
 
             # If no local credentials exist, proceed with a one-time discovery and copy.
@@ -140,13 +170,13 @@ class CredentialManager:
                 path = Path(path_str).expanduser()
                 if path.exists():
                     discovered_paths.add(path)
-            
+
             # 2. If no overrides are provided via .env, scan the default directory
             # [MODIFIED] This logic is now disabled to prefer local-first credential management.
             # if not discovered_paths and default_dir.exists():
             #     for json_file in default_dir.glob('*.json'):
             #         discovered_paths.add(json_file)
-            
+
             if not discovered_paths:
                 lib_logger.debug(f"No credential files found for provider: {provider}")
                 continue
@@ -156,18 +186,24 @@ class CredentialManager:
             for i, source_path in enumerate(sorted(list(discovered_paths))):
                 account_id = i + 1
                 local_filename = f"{provider}_oauth_{account_id}.json"
-                local_path = OAUTH_BASE_DIR / local_filename
+                local_path = self.oauth_base_dir / local_filename
 
                 try:
                     # Since we've established no local files exist, we can copy directly.
                     shutil.copy(source_path, local_path)
-                    lib_logger.info(f"Copied '{source_path.name}' to local pool at '{local_path}'.")
+                    lib_logger.info(
+                        f"Copied '{source_path.name}' to local pool at '{local_path}'."
+                    )
                     prepared_paths.append(str(local_path.resolve()))
                 except Exception as e:
-                    lib_logger.error(f"Failed to process OAuth file from '{source_path}': {e}")
-            
+                    lib_logger.error(
+                        f"Failed to process OAuth file from '{source_path}': {e}"
+                    )
+
             if prepared_paths:
-                lib_logger.info(f"Discovered and prepared {len(prepared_paths)} credential(s) for provider: {provider}")
+                lib_logger.info(
+                    f"Discovered and prepared {len(prepared_paths)} credential(s) for provider: {provider}"
+                )
                 final_config[provider] = prepared_paths
 
         lib_logger.info("OAuth credential discovery complete.")

src/rotator_library/credential_tool.py

@@ -14,10 +14,20 @@ from rich.panel import Panel
 from rich.prompt import Prompt
 from rich.text import Text
 
-OAUTH_BASE_DIR = Path.cwd() / "oauth_creds"
-OAUTH_BASE_DIR.mkdir(exist_ok=True)
-# Use a direct path to the .env file in the project root
-ENV_FILE = Path.cwd() / ".env"
+from .utils.paths import get_oauth_dir, get_data_file
+
+
+def _get_oauth_base_dir() -> Path:
+    """Get the OAuth base directory (lazy, respects EXE vs script mode)."""
+    oauth_dir = get_oauth_dir()
+    oauth_dir.mkdir(parents=True, exist_ok=True)
+    return oauth_dir
+
+
+def _get_env_file() -> Path:
+    """Get the .env file path (lazy, respects EXE vs script mode)."""
+    return get_data_file(".env")
+
 
 console = Console()
 
@@ -54,19 +64,19 @@ def ensure_env_defaults():
     """
     Ensures the .env file exists and contains essential default values like PROXY_API_KEY.
     """
-    if not ENV_FILE.is_file():
-        ENV_FILE.touch()
+    if not _get_env_file().is_file():
+        _get_env_file().touch()
         console.print(
-            f"Creating a new [bold yellow]{ENV_FILE.name}[/bold yellow] file..."
+            f"Creating a new [bold yellow]{_get_env_file().name}[/bold yellow] file..."
         )
 
     # Check for PROXY_API_KEY, similar to setup_env.bat
-    if get_key(str(ENV_FILE), "PROXY_API_KEY") is None:
+    if get_key(str(_get_env_file()), "PROXY_API_KEY") is None:
         default_key = "VerysecretKey"
         console.print(
-            f"Adding default [bold cyan]PROXY_API_KEY[/bold cyan] to [bold yellow]{ENV_FILE.name}[/bold yellow]..."
+            f"Adding default [bold cyan]PROXY_API_KEY[/bold cyan] to [bold yellow]{_get_env_file().name}[/bold yellow]..."
         )
-        set_key(str(ENV_FILE), "PROXY_API_KEY", default_key)
+        set_key(str(_get_env_file()), "PROXY_API_KEY", default_key)
 
 
 async def setup_api_key():
@@ -224,8 +234,8 @@ async def setup_api_key():
             api_key = Prompt.ask(f"Enter the API key for {display_name}")
 
             # Check for duplicate API key value
-            if ENV_FILE.is_file():
-                with open(ENV_FILE, "r") as f:
+            if _get_env_file().is_file():
+                with open(_get_env_file(), "r") as f:
                     for line in f:
                         line = line.strip()
                         if line.startswith(api_var_base) and "=" in line:
@@ -244,7 +254,9 @@ async def setup_api_key():
                                     )
                                 )
 
-                                set_key(str(ENV_FILE), existing_key_name, api_key)
+                                set_key(
+                                    str(_get_env_file()), existing_key_name, api_key
+                                )
 
                                 success_text = Text.from_markup(
                                     f"Successfully updated existing key [bold yellow]'{existing_key_name}'[/bold yellow]."
@@ -275,8 +287,8 @@ async def setup_api_key():
             key_index = 1
             while True:
                 key_name = f"{api_var_base}_{key_index}"
-                if ENV_FILE.is_file():
-                    with open(ENV_FILE, "r") as f:
+                if _get_env_file().is_file():
+                    with open(_get_env_file(), "r") as f:
                         if not any(line.startswith(f"{key_name}=") for line in f):
                             break
                 else:
@@ -284,7 +296,7 @@ async def setup_api_key():
                 key_index += 1
 
             key_name = f"{api_var_base}_{key_index}"
-            set_key(str(ENV_FILE), key_name, api_key)
+            set_key(str(_get_env_file()), key_name, api_key)
 
             success_text = Text.from_markup(
                 f"Successfully added {display_name} API key as [bold yellow]'{key_name}'[/bold yellow]."
@@ -327,7 +339,7 @@ async def setup_new_credential(provider_name: str):
         # - File path determination (new or existing)
         # - Credential file saving
         # - Post-auth discovery (tier/project for Google OAuth providers)
-        result = await auth_instance.setup_credential(OAUTH_BASE_DIR)
+        result = await auth_instance.setup_credential(_get_oauth_base_dir())
 
         if not result.success:
             console.print(
@@ -386,7 +398,7 @@ async def export_gemini_cli_to_env():
     auth_instance = auth_class()
 
     # List available credentials using auth class
-    credentials = auth_instance.list_credentials(OAUTH_BASE_DIR)
+    credentials = auth_instance.list_credentials(_get_oauth_base_dir())
 
     if not credentials:
         console.print(
@@ -427,7 +439,7 @@ async def export_gemini_cli_to_env():
 
             # Use auth class to export
             env_path = auth_instance.export_credential_to_env(
-                cred_info["file_path"], OAUTH_BASE_DIR
+                cred_info["file_path"], _get_oauth_base_dir()
             )
 
             if env_path:
@@ -481,7 +493,7 @@ async def export_qwen_code_to_env():
     auth_instance = auth_class()
 
     # List available credentials using auth class
-    credentials = auth_instance.list_credentials(OAUTH_BASE_DIR)
+    credentials = auth_instance.list_credentials(_get_oauth_base_dir())
 
     if not credentials:
         console.print(
@@ -522,7 +534,7 @@ async def export_qwen_code_to_env():
 
             # Use auth class to export
             env_path = auth_instance.export_credential_to_env(
-                cred_info["file_path"], OAUTH_BASE_DIR
+                cred_info["file_path"], _get_oauth_base_dir()
             )
 
             if env_path:
@@ -573,7 +585,7 @@ async def export_iflow_to_env():
     auth_instance = auth_class()
 
     # List available credentials using auth class
-    credentials = auth_instance.list_credentials(OAUTH_BASE_DIR)
+    credentials = auth_instance.list_credentials(_get_oauth_base_dir())
 
     if not credentials:
         console.print(
@@ -614,7 +626,7 @@ async def export_iflow_to_env():
 
             # Use auth class to export
             env_path = auth_instance.export_credential_to_env(
-                cred_info["file_path"], OAUTH_BASE_DIR
+                cred_info["file_path"], _get_oauth_base_dir()
             )
 
             if env_path:
@@ -667,7 +679,7 @@ async def export_antigravity_to_env():
     auth_instance = auth_class()
 
     # List available credentials using auth class
-    credentials = auth_instance.list_credentials(OAUTH_BASE_DIR)
+    credentials = auth_instance.list_credentials(_get_oauth_base_dir())
 
     if not credentials:
         console.print(
@@ -708,7 +720,7 @@ async def export_antigravity_to_env():
 
             # Use auth class to export
             env_path = auth_instance.export_credential_to_env(
-                cred_info["file_path"], OAUTH_BASE_DIR
+                cred_info["file_path"], _get_oauth_base_dir()
             )
 
             if env_path:
@@ -769,7 +781,7 @@ async def export_all_provider_credentials(provider_name: str):
     )
 
     # List all credentials using auth class
-    credentials = auth_instance.list_credentials(OAUTH_BASE_DIR)
+    credentials = auth_instance.list_credentials(_get_oauth_base_dir())
 
     if not credentials:
         console.print(
@@ -786,7 +798,7 @@ async def export_all_provider_credentials(provider_name: str):
         try:
             # Use auth class to export
             env_path = auth_instance.export_credential_to_env(
-                cred_info["file_path"], OAUTH_BASE_DIR
+                cred_info["file_path"], _get_oauth_base_dir()
             )
 
             if env_path:
@@ -837,7 +849,7 @@ async def combine_provider_credentials(provider_name: str):
     )
 
     # List all credentials using auth class
-    credentials = auth_instance.list_credentials(OAUTH_BASE_DIR)
+    credentials = auth_instance.list_credentials(_get_oauth_base_dir())
 
     if not credentials:
         console.print(
@@ -879,7 +891,7 @@ async def combine_provider_credentials(provider_name: str):
 
     # Write combined file
     combined_filename = f"{provider_name}_all_combined.env"
-    combined_filepath = OAUTH_BASE_DIR / combined_filename
+    combined_filepath = _get_oauth_base_dir() / combined_filename
 
     with open(combined_filepath, "w") as f:
         f.write("\n".join(combined_lines))
@@ -929,7 +941,7 @@ async def combine_all_credentials():
         except Exception:
             continue  # Skip providers that don't have auth classes
 
-        credentials = auth_instance.list_credentials(OAUTH_BASE_DIR)
+        credentials = auth_instance.list_credentials(_get_oauth_base_dir())
 
         if not credentials:
             continue
@@ -972,7 +984,7 @@ async def combine_all_credentials():
 
     # Write combined file
     combined_filename = "all_providers_combined.env"
-    combined_filepath = OAUTH_BASE_DIR / combined_filename
+    combined_filepath = _get_oauth_base_dir() / combined_filename
 
     with open(combined_filepath, "w") as f:
         f.write("\n".join(combined_lines))

src/rotator_library/failure_logger.py

@@ -1,9 +1,12 @@
 import logging
 import json
 from logging.handlers import RotatingFileHandler
-import os
+from pathlib import Path
 from datetime import datetime
+from typing import Optional, Union
+
 from .error_handler import mask_credential
+from .utils.paths import get_logs_dir
 
 
 class JsonFormatter(logging.Formatter):
@@ -14,9 +17,37 @@ class JsonFormatter(logging.Formatter):
         return json.dumps(record.msg)
 
 
-def setup_failure_logger():
-    """Sets up a dedicated JSON logger for writing detailed failure logs to a file."""
-    log_dir = "logs"
+# Module-level state for lazy initialization
+_failure_logger: Optional[logging.Logger] = None
+_configured_logs_dir: Optional[Path] = None
+
+
+def configure_failure_logger(logs_dir: Optional[Union[Path, str]] = None) -> None:
+    """
+    Configure the failure logger to use a specific logs directory.
+
+    Call this before first use if you want to override the default location.
+    If not called, the logger will use get_logs_dir() on first use.
+
+    Args:
+        logs_dir: Path to the logs directory. If None, uses get_logs_dir().
+    """
+    global _configured_logs_dir, _failure_logger
+    _configured_logs_dir = Path(logs_dir) if logs_dir else None
+    # Reset logger so it gets reconfigured on next use
+    _failure_logger = None
+
+
+def _setup_failure_logger(logs_dir: Path) -> logging.Logger:
+    """
+    Sets up a dedicated JSON logger for writing detailed failure logs to a file.
+
+    Args:
+        logs_dir: Path to the logs directory.
+
+    Returns:
+        Configured logger instance.
+    """
     logger = logging.getLogger("failure_logger")
     logger.setLevel(logging.INFO)
     logger.propagate = False
@@ -25,11 +56,10 @@ def setup_failure_logger():
     logger.handlers.clear()
 
     try:
-        if not os.path.exists(log_dir):
-            os.makedirs(log_dir, exist_ok=True)
+        logs_dir.mkdir(parents=True, exist_ok=True)
 
         handler = RotatingFileHandler(
-            os.path.join(log_dir, "failures.log"),
+            logs_dir / "failures.log",
             maxBytes=5 * 1024 * 1024,  # 5 MB
             backupCount=2,
         )
@@ -43,8 +73,21 @@ def setup_failure_logger():
     return logger
 
 
-# Initialize the dedicated logger for detailed failure logs
-failure_logger = setup_failure_logger()
+def get_failure_logger() -> logging.Logger:
+    """
+    Get the failure logger, initializing it lazily if needed.
+
+    Returns:
+        The configured failure logger.
+    """
+    global _failure_logger, _configured_logs_dir
+
+    if _failure_logger is None:
+        logs_dir = _configured_logs_dir if _configured_logs_dir else get_logs_dir()
+        _failure_logger = _setup_failure_logger(logs_dir)
+
+    return _failure_logger
+
 
 # Get the main library logger for concise, propagated messages
 main_lib_logger = logging.getLogger("rotator_library")
@@ -174,7 +217,7 @@ def log_failure(
 
     # Log to failure logger with resilience - if it fails, just continue
     try:
-        failure_logger.error(detailed_log_data)
+        get_failure_logger().error(detailed_log_data)
     except (OSError, IOError) as e:
         # Log file write failed - log to console instead
         logging.warning(f"Failed to write to failures.log: {e}")

src/rotator_library/providers/antigravity_provider.py

@@ -39,6 +39,7 @@ from .antigravity_auth_base import AntigravityAuthBase
 from .provider_cache import ProviderCache
 from ..model_definitions import ModelDefinitions
 from ..timeout_config import TimeoutConfig
+from ..utils.paths import get_logs_dir, get_cache_dir
 
 
 # =============================================================================
@@ -106,12 +107,23 @@ DEFAULT_SAFETY_SETTINGS = [
     {"category": "HARM_CATEGORY_CIVIC_INTEGRITY", "threshold": "BLOCK_NONE"},
 ]
 
-# Directory paths
-_BASE_DIR = Path(__file__).resolve().parent.parent.parent.parent
-LOGS_DIR = _BASE_DIR / "logs" / "antigravity_logs"
-CACHE_DIR = _BASE_DIR / "cache" / "antigravity"
-GEMINI3_SIGNATURE_CACHE_FILE = CACHE_DIR / "gemini3_signatures.json"
-CLAUDE_THINKING_CACHE_FILE = CACHE_DIR / "claude_thinking.json"
+
+# Directory paths - use centralized path management
+def _get_antigravity_logs_dir():
+    return get_logs_dir() / "antigravity_logs"
+
+
+def _get_antigravity_cache_dir():
+    return get_cache_dir(subdir="antigravity")
+
+
+def _get_gemini3_signature_cache_file():
+    return _get_antigravity_cache_dir() / "gemini3_signatures.json"
+
+
+def _get_claude_thinking_cache_file():
+    return _get_antigravity_cache_dir() / "claude_thinking.json"
+
 
 # Gemini 3 tool fix system instruction (prevents hallucination)
 DEFAULT_GEMINI3_SYSTEM_INSTRUCTION = """<CRITICAL_TOOL_USAGE_INSTRUCTIONS>
@@ -426,7 +438,9 @@ class AntigravityFileLogger:
 
         timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
         safe_model = model_name.replace("/", "_").replace(":", "_")
-        self.log_dir = LOGS_DIR / f"{timestamp}_{safe_model}_{uuid.uuid4()}"
+        self.log_dir = (
+            _get_antigravity_logs_dir() / f"{timestamp}_{safe_model}_{uuid.uuid4()}"
+        )
 
         try:
             self.log_dir.mkdir(parents=True, exist_ok=True)
@@ -731,13 +745,13 @@ class AntigravityProvider(AntigravityAuthBase, ProviderInterface):
 
         # Initialize caches using shared ProviderCache
         self._signature_cache = ProviderCache(
-            GEMINI3_SIGNATURE_CACHE_FILE,
+            _get_gemini3_signature_cache_file(),
             memory_ttl,
             disk_ttl,
             env_prefix="ANTIGRAVITY_SIGNATURE",
         )
         self._thinking_cache = ProviderCache(
-            CLAUDE_THINKING_CACHE_FILE,
+            _get_claude_thinking_cache_file(),
             memory_ttl,
             disk_ttl,
             env_prefix="ANTIGRAVITY_THINKING",

src/rotator_library/providers/gemini_cli_provider.py

@@ -12,6 +12,7 @@ from .gemini_auth_base import GeminiAuthBase
 from .provider_cache import ProviderCache
 from ..model_definitions import ModelDefinitions
 from ..timeout_config import TimeoutConfig
+from ..utils.paths import get_logs_dir, get_cache_dir
 import litellm
 from litellm.exceptions import RateLimitError
 from ..error_handler import extract_retry_after_from_body
@@ -22,8 +23,22 @@ from datetime import datetime
 
 lib_logger = logging.getLogger("rotator_library")
 
-LOGS_DIR = Path(__file__).resolve().parent.parent.parent.parent / "logs"
-GEMINI_CLI_LOGS_DIR = LOGS_DIR / "gemini_cli_logs"
+
+def _get_gemini_cli_logs_dir() -> Path:
+    """Get the Gemini CLI logs directory."""
+    logs_dir = get_logs_dir() / "gemini_cli_logs"
+    logs_dir.mkdir(parents=True, exist_ok=True)
+    return logs_dir
+
+
+def _get_gemini_cli_cache_dir() -> Path:
+    """Get the Gemini CLI cache directory."""
+    return get_cache_dir(subdir="gemini_cli")
+
+
+def _get_gemini3_signature_cache_file() -> Path:
+    """Get the Gemini 3 signature cache file path."""
+    return _get_gemini_cli_cache_dir() / "gemini3_signatures.json"
 
 
 class _GeminiCliFileLogger:
@@ -39,7 +54,7 @@ class _GeminiCliFileLogger:
         # Sanitize model name for directory
         safe_model_name = model_name.replace("/", "_").replace(":", "_")
         self.log_dir = (
-            GEMINI_CLI_LOGS_DIR / f"{timestamp}_{safe_model_name}_{request_id}"
+            _get_gemini_cli_logs_dir() / f"{timestamp}_{safe_model_name}_{request_id}"
         )
         try:
             self.log_dir.mkdir(parents=True, exist_ok=True)
@@ -103,12 +118,6 @@ HARDCODED_MODELS = [
     "gemini-3-pro-preview",
 ]
 
-# Cache directory for Gemini CLI
-CACHE_DIR = (
-    Path(__file__).resolve().parent.parent.parent.parent / "cache" / "gemini_cli"
-)
-GEMINI3_SIGNATURE_CACHE_FILE = CACHE_DIR / "gemini3_signatures.json"
-
 # Gemini 3 tool fix system instruction (prevents hallucination)
 DEFAULT_GEMINI3_SYSTEM_INSTRUCTION = """<CRITICAL_TOOL_USAGE_INSTRUCTIONS>
 You are operating in a CUSTOM ENVIRONMENT where tool definitions COMPLETELY DIFFER from your training data.
@@ -392,7 +401,7 @@ class GeminiCliProvider(GeminiAuthBase, ProviderInterface):
 
         # Initialize signature cache for Gemini 3 thoughtSignatures
         self._signature_cache = ProviderCache(
-            GEMINI3_SIGNATURE_CACHE_FILE,
+            _get_gemini3_signature_cache_file(),
             memory_ttl,
             disk_ttl,
             env_prefix="GEMINI_CLI_SIGNATURE",

src/rotator_library/providers/iflow_provider.py

@@ -11,6 +11,7 @@ from .provider_interface import ProviderInterface
 from .iflow_auth_base import IFlowAuthBase
 from ..model_definitions import ModelDefinitions
 from ..timeout_config import TimeoutConfig
+from ..utils.paths import get_logs_dir
 import litellm
 from litellm.exceptions import RateLimitError, AuthenticationError
 from pathlib import Path
@@ -19,8 +20,12 @@ from datetime import datetime
 
 lib_logger = logging.getLogger("rotator_library")
 
-LOGS_DIR = Path(__file__).resolve().parent.parent.parent.parent / "logs"
-IFLOW_LOGS_DIR = LOGS_DIR / "iflow_logs"
+
+def _get_iflow_logs_dir() -> Path:
+    """Get the iFlow logs directory."""
+    logs_dir = get_logs_dir() / "iflow_logs"
+    logs_dir.mkdir(parents=True, exist_ok=True)
+    return logs_dir
 
 
 class _IFlowFileLogger:
@@ -35,7 +40,9 @@ class _IFlowFileLogger:
         request_id = str(uuid.uuid4())
         # Sanitize model name for directory
         safe_model_name = model_name.replace("/", "_").replace(":", "_")
-        self.log_dir = IFLOW_LOGS_DIR / f"{timestamp}_{safe_model_name}_{request_id}"
+        self.log_dir = (
+            _get_iflow_logs_dir() / f"{timestamp}_{safe_model_name}_{request_id}"
+        )
         try:
             self.log_dir.mkdir(parents=True, exist_ok=True)
         except Exception as e:

src/rotator_library/providers/qwen_code_provider.py

@@ -11,6 +11,7 @@ from .provider_interface import ProviderInterface
 from .qwen_auth_base import QwenAuthBase
 from ..model_definitions import ModelDefinitions
 from ..timeout_config import TimeoutConfig
+from ..utils.paths import get_logs_dir
 import litellm
 from litellm.exceptions import RateLimitError, AuthenticationError
 from pathlib import Path
@@ -19,8 +20,12 @@ from datetime import datetime
 
 lib_logger = logging.getLogger("rotator_library")
 
-LOGS_DIR = Path(__file__).resolve().parent.parent.parent.parent / "logs"
-QWEN_CODE_LOGS_DIR = LOGS_DIR / "qwen_code_logs"
+
+def _get_qwen_code_logs_dir() -> Path:
+    """Get the Qwen Code logs directory."""
+    logs_dir = get_logs_dir() / "qwen_code_logs"
+    logs_dir.mkdir(parents=True, exist_ok=True)
+    return logs_dir
 
 
 class _QwenCodeFileLogger:
@@ -36,7 +41,7 @@ class _QwenCodeFileLogger:
         # Sanitize model name for directory
         safe_model_name = model_name.replace("/", "_").replace(":", "_")
         self.log_dir = (
-            QWEN_CODE_LOGS_DIR / f"{timestamp}_{safe_model_name}_{request_id}"
+            _get_qwen_code_logs_dir() / f"{timestamp}_{safe_model_name}_{request_id}"
         )
         try:
             self.log_dir.mkdir(parents=True, exist_ok=True)

src/rotator_library/usage_manager.py

@@ -5,13 +5,15 @@ import logging
 import asyncio
 import random
 from datetime import date, datetime, timezone, time as dt_time
-from typing import Any, Dict, List, Optional, Set, Tuple
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Set, Tuple, Union
 import aiofiles
 import litellm
 
 from .error_handler import ClassifiedError, NoAvailableKeysError, mask_credential
 from .providers import PROVIDER_PLUGINS
 from .utils.resilient_io import ResilientStateWriter
+from .utils.paths import get_data_file
 
 lib_logger = logging.getLogger("rotator_library")
 lib_logger.propagate = False
@@ -51,7 +53,7 @@ class UsageManager:
 
     def __init__(
         self,
-        file_path: str = "key_usage.json",
+        file_path: Optional[Union[str, Path]] = None,
         daily_reset_time_utc: Optional[str] = "03:00",
         rotation_tolerance: float = 0.0,
         provider_rotation_modes: Optional[Dict[str, str]] = None,
@@ -66,7 +68,8 @@ class UsageManager:
         Initialize the UsageManager.
 
         Args:
-            file_path: Path to the usage data JSON file
+            file_path: Path to the usage data JSON file. If None, uses get_data_file("key_usage.json").
+                       Can be absolute Path, relative Path, or string.
             daily_reset_time_utc: Time in UTC when daily stats should reset (HH:MM format)
             rotation_tolerance: Tolerance for weighted random credential rotation.
                 - 0.0: Deterministic, least-used credential always selected
@@ -86,7 +89,14 @@ class UsageManager:
                 Used in sequential mode when priority not in priority_multipliers.
                 Example: {"antigravity": 2}
         """
-        self.file_path = file_path
+        # Resolve file_path - use default if not provided
+        if file_path is None:
+            self.file_path = str(get_data_file("key_usage.json"))
+        elif isinstance(file_path, Path):
+            self.file_path = str(file_path)
+        else:
+            # String path - could be relative or absolute
+            self.file_path = file_path
         self.rotation_tolerance = rotation_tolerance
         self.provider_rotation_modes = provider_rotation_modes or {}
         self.provider_plugins = provider_plugins or PROVIDER_PLUGINS

src/rotator_library/utils/__init__.py

@@ -1,6 +1,13 @@
 # src/rotator_library/utils/__init__.py
 
 from .headless_detection import is_headless_environment
+from .paths import (
+    get_default_root,
+    get_logs_dir,
+    get_cache_dir,
+    get_oauth_dir,
+    get_data_file,
+)
 from .reauth_coordinator import get_reauth_coordinator, ReauthCoordinator
 from .resilient_io import (
     BufferedWriteRegistry,
@@ -12,6 +19,11 @@ from .resilient_io import (
 
 __all__ = [
     "is_headless_environment",
+    "get_default_root",
+    "get_logs_dir",
+    "get_cache_dir",
+    "get_oauth_dir",
+    "get_data_file",
     "get_reauth_coordinator",
     "ReauthCoordinator",
     "BufferedWriteRegistry",

src/rotator_library/utils/paths.py

@@ -0,0 +1,99 @@
+# src/rotator_library/utils/paths.py
+"""
+Centralized path management for the rotator library.
+
+Supports two runtime modes:
+1. PyInstaller EXE -> files in the directory containing the executable
+2. Script/Library  -> files in the current working directory (overridable)
+
+Library users can override by passing `data_dir` to RotatingClient.
+"""
+
+import sys
+from pathlib import Path
+from typing import Optional, Union
+
+
+def get_default_root() -> Path:
+    """
+    Get the default root directory for data files.
+
+    - EXE mode (PyInstaller): directory containing the executable
+    - Otherwise: current working directory
+
+    Returns:
+        Path to the root directory
+    """
+    if getattr(sys, "frozen", False):
+        # Running as PyInstaller bundle - use executable's directory
+        return Path(sys.executable).parent
+    # Running as script or library - use current working directory
+    return Path.cwd()
+
+
+def get_logs_dir(root: Optional[Union[Path, str]] = None) -> Path:
+    """
+    Get the logs directory, creating it if needed.
+
+    Args:
+        root: Optional root directory. If None, uses get_default_root().
+
+    Returns:
+        Path to the logs directory
+    """
+    base = Path(root) if root else get_default_root()
+    logs_dir = base / "logs"
+    logs_dir.mkdir(exist_ok=True)
+    return logs_dir
+
+
+def get_cache_dir(
+    root: Optional[Union[Path, str]] = None, subdir: Optional[str] = None
+) -> Path:
+    """
+    Get the cache directory, optionally with a subdirectory.
+
+    Args:
+        root: Optional root directory. If None, uses get_default_root().
+        subdir: Optional subdirectory name (e.g., "gemini_cli", "antigravity")
+
+    Returns:
+        Path to the cache directory (or subdirectory)
+    """
+    base = Path(root) if root else get_default_root()
+    cache_dir = base / "cache"
+    if subdir:
+        cache_dir = cache_dir / subdir
+    cache_dir.mkdir(parents=True, exist_ok=True)
+    return cache_dir
+
+
+def get_oauth_dir(root: Optional[Union[Path, str]] = None) -> Path:
+    """
+    Get the OAuth credentials directory, creating it if needed.
+
+    Args:
+        root: Optional root directory. If None, uses get_default_root().
+
+    Returns:
+        Path to the oauth_creds directory
+    """
+    base = Path(root) if root else get_default_root()
+    oauth_dir = base / "oauth_creds"
+    oauth_dir.mkdir(exist_ok=True)
+    return oauth_dir
+
+
+def get_data_file(filename: str, root: Optional[Union[Path, str]] = None) -> Path:
+    """
+    Get the path to a data file in the root directory.
+
+    Args:
+        filename: Name of the file (e.g., "key_usage.json", ".env")
+        root: Optional root directory. If None, uses get_default_root().
+
+    Returns:
+        Path to the file (does not create the file)
+    """
+    base = Path(root) if root else get_default_root()
+    return base / filename