feat: add runtime resilience for file deletion survival

Implement graceful degradation patterns that allow the proxy to continue
running even if core files are deleted during runtime. Changes only take
effect on restart, enabling safe development while the proxy is serving.

## Changes by Component

### Usage Manager (usage_manager.py)
- Wrap `_save_usage()` in try/except with directory auto-recreation
- Enhanced `_load_usage()` with explicit error handling
- In-memory state continues working if file operations fail

### Failure Logger (failure_logger.py)
- Add module-level `_file_handler` and `_fallback_mode` state
- Create `_create_file_handler()` with directory auto-recreation
- Create `_ensure_handler_valid()` for handler recovery
- Use NullHandler as fallback when file logging fails

### Detailed Logger (detailed_logger.py)
- Add class-level `_disk_available` and `_console_fallback_warned` flags
- Add instance-level `_in_memory_logs` list for fallback storage
- Skip disk writes gracefully when filesystem unavailable

### Google OAuth Base (google_oauth_base.py)
- Update memory cache FIRST before disk write (memory-first pattern)
- Use cached tokens as fallback when refresh/save fails
- Log warnings but don't crash on persistence failures

### Provider Cache (provider_cache.py)
- Add `_disk_available` health flag and `disk_errors` counter
- Track disk health status in get_stats()
- Gracefully degrade to memory-only caching on disk failures

### Documentation (DOCUMENTATION.md)
- Add Section 5: Runtime Resilience with resilience hierarchy
- Document "Develop While Running" workflow
- Explain graceful degradation and data loss scenarios

DOCUMENTATION.md

@@ -697,4 +697,35 @@ To facilitate robust debugging, the proxy includes a comprehensive transaction l
 
 This level of detail allows developers to trace exactly why a request failed or why a specific key was rotated.
 
+---
+
+## 5. Runtime Resilience
+
+The proxy is engineered to maintain high availability even in the face of runtime filesystem disruptions. This "Runtime Resilience" capability ensures that the service continues to process API requests even if core data directories (like `logs/`, `oauth_creds/`) or files are accidentally deleted or become unwritable while the application is running.
+
+### 5.1. Resilience Hierarchy
+
+The system follows a strict hierarchy of survival:
+
+1.  **Core API Handling (Level 1)**: The Python runtime keeps all necessary code in memory (`sys.modules`). Deleting source code files while the proxy is running will **not** crash active requests.
+2.  **Credential Management (Level 2)**: OAuth tokens are aggressively cached in memory. If credential files are deleted, the proxy continues using the cached tokens. If a token needs refresh and the file cannot be written, the new token is updated in memory only.
+3.  **Usage Tracking (Level 3)**: Usage statistics (`key_usage.json`) are maintained in memory. If the file is deleted, the system tracks usage internally. It attempts to recreate the file/directory on the next save interval. If save fails, data is effectively "memory-only" until the next successful write.
+4.  **Logging (Level 4)**: Logging is treated as non-critical. If the `logs/` directory is removed, the system attempts to recreate it. If creation fails (e.g., permission error), logging degrades gracefully (stops or falls back to console) without interrupting the request flow.
+
+### 5.2. "Develop While Running"
+
+This architecture supports a robust development workflow:
+
+*   **Log Cleanup**: You can safely run `rm -rf logs/` while the proxy is serving traffic. The system will simply recreate the directory structure on the next request.
+*   **Config Reset**: Deleting `key_usage.json` resets the persistence layer, but the running instance preserves its current in-memory counts to ensure load balancing consistency.
+*   **File Recovery**: If you delete a critical file, the system attempts **Directory Auto-Recreation** before every write operation.
+
+### 5.3. Graceful Degradation & Data Loss
+
+While functionality is preserved, persistence may be compromised during filesystem failures:
+
+*   **Logs**: If disk writes fail, detailed request logs may be lost (unless console fallback is active).
+*   **Usage Stats**: If `key_usage.json` cannot be written, usage data since the last successful save will be lost upon application restart.
+*   **Credentials**: Refreshed tokens held only in memory will require re-authentication after a restart if they cannot be persisted to disk.
+
 

src/proxy_app/detailed_logger.py

@@ -13,6 +13,10 @@ class DetailedLogger:
     """
     Logs comprehensive details of each API transaction to a unique, timestamped directory.
     """
+    # Class-level fallback flags for resilience
+    _disk_available = True
+    _console_fallback_warned = False
+    
     def __init__(self):
         """
         Initializes the logger for a single request, creating a unique directory to store all related log files.
@@ -21,16 +25,33 @@ class DetailedLogger:
         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.mkdir(parents=True, exist_ok=True)
         self.streaming = False
+        self._in_memory_logs = []  # Fallback storage
+        
+        # Attempt directory creation with resilience
+        try:
+            self.log_dir.mkdir(parents=True, exist_ok=True)
+            DetailedLogger._disk_available = True
+        except (OSError, PermissionError) as e:
+            DetailedLogger._disk_available = False
+            if not DetailedLogger._console_fallback_warned:
+                logging.warning(f"Detailed logging disabled - cannot create log directory: {e}")
+                DetailedLogger._console_fallback_warned = True
 
     def _write_json(self, filename: str, data: Dict[str, Any]):
         """Helper to write data to a JSON file in the log directory."""
+        if not DetailedLogger._disk_available:
+            self._in_memory_logs.append({"file": filename, "data": data})
+            return
+        
         try:
+            # Attempt directory recreation if needed
+            self.log_dir.mkdir(parents=True, exist_ok=True)
             with open(self.log_dir / filename, "w", encoding="utf-8") as f:
                 json.dump(data, f, indent=4, ensure_ascii=False)
-        except Exception as e:
+        except (OSError, PermissionError, IOError) as e:
             logging.error(f"[{self.request_id}] Failed to write to {filename}: {e}")
+            self._in_memory_logs.append({"file": filename, "data": data})
 
     def log_request(self, headers: Dict[str, Any], body: Dict[str, Any]):
         """Logs the initial request details."""
@@ -45,14 +66,18 @@ class DetailedLogger:
 
     def log_stream_chunk(self, chunk: Dict[str, Any]):
         """Logs an individual chunk from a streaming response to a JSON Lines file."""
+        if not DetailedLogger._disk_available:
+            return  # Skip chunk logging when disk unavailable
+        
         try:
+            self.log_dir.mkdir(parents=True, exist_ok=True)
             log_entry = {
                 "timestamp_utc": datetime.utcnow().isoformat(),
                 "chunk": chunk
             }
             with open(self.log_dir / "streaming_chunks.jsonl", "a", encoding="utf-8") as f:
                 f.write(json.dumps(log_entry, ensure_ascii=False) + "\n")
-        except Exception as e:
+        except (OSError, PermissionError, IOError) as e:
             logging.error(f"[{self.request_id}] Failed to write stream chunk: {e}")
 
     def log_final_response(self, status_code: int, headers: Optional[Dict[str, Any]], body: Dict[str, Any]):

src/rotator_library/failure_logger.py

@@ -5,41 +5,76 @@ import os
 from datetime import datetime
 from .error_handler import mask_credential
 
+# Module-level state for resilience
+_file_handler = None
+_fallback_mode = False
 
-def setup_failure_logger():
-    """Sets up a dedicated JSON logger for writing detailed failure logs to a file."""
-    log_dir = "logs"
-    if not os.path.exists(log_dir):
-        os.makedirs(log_dir)
 
-    # Create a logger specifically for failures.
-    # This logger will NOT propagate to the root logger.
-    logger = logging.getLogger("failure_logger")
-    logger.setLevel(logging.INFO)
-    logger.propagate = False
+# Custom JSON formatter for structured logs (defined at module level for reuse)
+class JsonFormatter(logging.Formatter):
+    def format(self, record):
+        # The message is already a dict, so we just format it as a JSON string
+        return json.dumps(record.msg)
 
-    # Use a rotating file handler
-    handler = RotatingFileHandler(
-        os.path.join(log_dir, "failures.log"),
-        maxBytes=5 * 1024 * 1024,  # 5 MB
-        backupCount=2,
-    )
 
-    # Custom JSON formatter for structured logs
-    class JsonFormatter(logging.Formatter):
-        def format(self, record):
-            # The message is already a dict, so we just format it as a JSON string
-            return json.dumps(record.msg)
+def _create_file_handler():
+    """Create file handler with directory auto-recreation."""
+    global _file_handler, _fallback_mode
+    log_dir = "logs"
+    
+    try:
+        if not os.path.exists(log_dir):
+            os.makedirs(log_dir, exist_ok=True)
+        
+        handler = RotatingFileHandler(
+            os.path.join(log_dir, "failures.log"),
+            maxBytes=5 * 1024 * 1024,  # 5 MB
+            backupCount=2,
+        )
+        
+        handler.setFormatter(JsonFormatter())
+        _file_handler = handler
+        _fallback_mode = False
+        return handler
+    except (OSError, PermissionError, IOError) as e:
+        logging.warning(f"Cannot create failure log file handler: {e}")
+        _fallback_mode = True
+        return None
 
-    handler.setFormatter(JsonFormatter())
 
-    # Add handler only if it hasn't been added before
-    if not logger.handlers:
+def setup_failure_logger():
+    """Sets up a dedicated JSON logger for writing detailed failure logs."""
+    logger = logging.getLogger("failure_logger")
+    logger.setLevel(logging.INFO)
+    logger.propagate = False
+    
+    # Remove existing handlers to prevent duplicates
+    logger.handlers.clear()
+    
+    # Try to add file handler
+    handler = _create_file_handler()
+    if handler:
         logger.addHandler(handler)
-
+    
+    # Always add a NullHandler as fallback to prevent "no handlers" warning
+    if not logger.handlers:
+        logger.addHandler(logging.NullHandler())
+    
     return logger
 
 
+def _ensure_handler_valid():
+    """Check if file handler is still valid, recreate if needed."""
+    global _file_handler, _fallback_mode
+    
+    if _file_handler is None or _fallback_mode:
+        handler = _create_file_handler()
+        if handler:
+            failure_logger = logging.getLogger("failure_logger")
+            failure_logger.handlers.clear()
+            failure_logger.addHandler(handler)
+
+
 # Initialize the dedicated logger for detailed failure logs
 failure_logger = setup_failure_logger()
 
@@ -145,11 +180,23 @@ def log_failure(
         "request_headers": request_headers,
         "error_chain": error_chain if len(error_chain) > 1 else None,
     }
-    failure_logger.error(detailed_log_data)
-
+    
     # 2. Log a concise summary to the main library logger, which will propagate
     summary_message = (
         f"API call failed for model {model} with key {mask_credential(api_key)}. "
         f"Error: {type(error).__name__}. See failures.log for details."
     )
+    
+    # Attempt to ensure handler is valid before logging
+    _ensure_handler_valid()
+    
+    # Wrap the actual log call with resilience
+    try:
+        failure_logger.error(detailed_log_data)
+    except (OSError, IOError) as e:
+        # File logging failed - log to console instead
+        logging.error(f"Failed to write to failures.log: {e}")
+        logging.error(f"Failure summary: {summary_message}")
+    
+    # Console log always succeeds
     main_lib_logger.error(summary_message)

src/rotator_library/providers/google_oauth_base.py

@@ -260,64 +260,76 @@ class GoogleOAuthBase:
                 )
 
     async def _save_credentials(self, path: str, creds: Dict[str, Any]):
+        """Save credentials with in-memory fallback if disk unavailable.
+        
+        [RUNTIME RESILIENCE] Always updates the in-memory cache first (memory is reliable),
+        then attempts disk persistence. If disk write fails, logs a warning but does NOT
+        raise an exception - the in-memory state continues to work.
+        """
+        # [IN-MEMORY FIRST] Always update cache first (reliable)
+        self._credentials_cache[path] = creds
+
         # Don't save to file if credentials were loaded from environment
         if creds.get("_proxy_metadata", {}).get("loaded_from_env"):
             lib_logger.debug("Credentials loaded from env, skipping file save")
-            # Still update cache for in-memory consistency
-            self._credentials_cache[path] = creds
             return
 
-        # [ATOMIC WRITE] Use tempfile + move pattern to ensure atomic writes
-        # This prevents credential corruption if the process is interrupted during write
-        parent_dir = os.path.dirname(os.path.abspath(path))
-        os.makedirs(parent_dir, exist_ok=True)
-
-        tmp_fd = None
-        tmp_path = None
         try:
-            # Create temp file in same directory as target (ensures same filesystem)
-            tmp_fd, tmp_path = tempfile.mkstemp(
-                dir=parent_dir, prefix=".tmp_", suffix=".json", text=True
-            )
-
-            # Write JSON to temp file
-            with os.fdopen(tmp_fd, "w") as f:
-                json.dump(creds, f, indent=2)
-                tmp_fd = None  # fdopen closes the fd
+            # [ATOMIC WRITE] Use tempfile + move pattern to ensure atomic writes
+            # This prevents credential corruption if the process is interrupted during write
+            parent_dir = os.path.dirname(os.path.abspath(path))
+            os.makedirs(parent_dir, exist_ok=True)
 
-            # Set secure permissions (0600 = owner read/write only)
+            tmp_fd = None
+            tmp_path = None
             try:
-                os.chmod(tmp_path, 0o600)
-            except (OSError, AttributeError):
-                # Windows may not support chmod, ignore
-                pass
-
-            # Atomic move (overwrites target if it exists)
-            shutil.move(tmp_path, path)
-            tmp_path = None  # Successfully moved
+                # Create temp file in same directory as target (ensures same filesystem)
+                tmp_fd, tmp_path = tempfile.mkstemp(
+                    dir=parent_dir, prefix=".tmp_", suffix=".json", text=True
+                )
 
-            # Update cache AFTER successful file write (prevents cache/file inconsistency)
-            self._credentials_cache[path] = creds
-            lib_logger.debug(
-                f"Saved updated {self.ENV_PREFIX} OAuth credentials to '{path}' (atomic write)."
-            )
+                # Write JSON to temp file
+                with os.fdopen(tmp_fd, "w") as f:
+                    json.dump(creds, f, indent=2)
+                    tmp_fd = None  # fdopen closes the fd
 
-        except Exception as e:
-            lib_logger.error(
-                f"Failed to save updated {self.ENV_PREFIX} OAuth credentials to '{path}': {e}"
-            )
-            # Clean up temp file if it still exists
-            if tmp_fd is not None:
+                # Set secure permissions (0600 = owner read/write only)
                 try:
-                    os.close(tmp_fd)
-                except:
+                    os.chmod(tmp_path, 0o600)
+                except (OSError, AttributeError):
+                    # Windows may not support chmod, ignore
                     pass
-            if tmp_path and os.path.exists(tmp_path):
-                try:
-                    os.unlink(tmp_path)
-                except:
-                    pass
-            raise
+
+                # Atomic move (overwrites target if it exists)
+                shutil.move(tmp_path, path)
+                tmp_path = None  # Successfully moved
+
+                lib_logger.debug(
+                    f"Saved updated {self.ENV_PREFIX} OAuth credentials to '{path}' (atomic write)."
+                )
+
+            except Exception as e:
+                # Clean up temp file if it still exists
+                if tmp_fd is not None:
+                    try:
+                        os.close(tmp_fd)
+                    except:
+                        pass
+                if tmp_path and os.path.exists(tmp_path):
+                    try:
+                        os.unlink(tmp_path)
+                    except:
+                        pass
+                raise
+
+        except (OSError, PermissionError, IOError) as e:
+            # [FAIL SILENTLY, LOG LOUDLY] Log the error but don't crash
+            # The in-memory cache was already updated, so we can continue operating
+            lib_logger.warning(
+                f"Failed to save credentials to {path}: {e}. "
+                "Credentials cached in memory only (will be lost on restart)."
+            )
+            # Don't raise - we already updated the memory cache
 
     def _is_token_expired(self, creds: Dict[str, Any]) -> bool:
         expiry = creds.get("token_expiry")  # gcloud format
@@ -841,10 +853,39 @@ class GoogleOAuthBase:
             )
 
     async def get_auth_header(self, credential_path: str) -> Dict[str, str]:
-        creds = await self._load_credentials(credential_path)
-        if self._is_token_expired(creds):
-            creds = await self._refresh_token(credential_path, creds)
-        return {"Authorization": f"Bearer {creds['access_token']}"}
+        """Get auth header with graceful degradation if refresh fails.
+        
+        [RUNTIME RESILIENCE] If credential file is deleted or refresh fails,
+        attempts to use cached credentials. This allows the proxy to continue
+        operating with potentially stale tokens rather than crashing.
+        """
+        try:
+            creds = await self._load_credentials(credential_path)
+            if self._is_token_expired(creds):
+                try:
+                    creds = await self._refresh_token(credential_path, creds)
+                except Exception as e:
+                    # [CACHED TOKEN FALLBACK] Check if we have a cached token that might still work
+                    cached = self._credentials_cache.get(credential_path)
+                    if cached and cached.get("access_token"):
+                        lib_logger.warning(
+                            f"Token refresh failed for {Path(credential_path).name}: {e}. "
+                            "Using cached token (may be expired)."
+                        )
+                        creds = cached
+                    else:
+                        raise
+            return {"Authorization": f"Bearer {creds['access_token']}"}
+        except Exception as e:
+            # [FINAL FALLBACK] Check if any cached credential exists as last resort
+            cached = self._credentials_cache.get(credential_path)
+            if cached and cached.get("access_token"):
+                lib_logger.error(
+                    f"Credential load failed for {credential_path}: {e}. "
+                    "Using stale cached token as last resort."
+                )
+                return {"Authorization": f"Bearer {cached['access_token']}"}
+            raise
 
     async def get_user_info(
         self, creds_or_path: Union[Dict[str, Any], str]

src/rotator_library/providers/provider_cache.py

@@ -104,7 +104,10 @@ class ProviderCache:
         self._running = False
         
         # Statistics
-        self._stats = {"memory_hits": 0, "disk_hits": 0, "misses": 0, "writes": 0}
+        self._stats = {"memory_hits": 0, "disk_hits": 0, "misses": 0, "writes": 0, "disk_errors": 0}
+        
+        # [RUNTIME RESILIENCE] Track disk health for monitoring
+        self._disk_available = True
         
         # Metadata about this cache instance
         self._cache_name = cache_file.stem if cache_file else "unnamed"
@@ -171,13 +174,27 @@ class ProviderCache:
     # =========================================================================
     
     async def _save_to_disk(self) -> None:
-        """Persist cache to disk using atomic write."""
+        """Persist cache to disk using atomic write with health tracking.
+        
+        [RUNTIME RESILIENCE] Tracks disk health and records errors. If disk
+        operations fail, the memory cache continues to work. Health status
+        is available via get_stats() for monitoring.
+        """
         if not self._enable_disk:
             return
         
         try:
             async with self._disk_lock:
-                self._cache_file.parent.mkdir(parents=True, exist_ok=True)
+                # [DIRECTORY AUTO-RECREATION] Attempt to create directory
+                try:
+                    self._cache_file.parent.mkdir(parents=True, exist_ok=True)
+                except (OSError, PermissionError) as e:
+                    self._stats["disk_errors"] += 1
+                    self._disk_available = False
+                    lib_logger.warning(
+                        f"ProviderCache[{self._cache_name}]: Cannot create cache directory: {e}"
+                    )
+                    return
                 
                 cache_data = {
                     "version": "1.0",
@@ -210,6 +227,8 @@ class ProviderCache:
                     
                     shutil.move(tmp_path, self._cache_file)
                     self._stats["writes"] += 1
+                    # [RUNTIME RESILIENCE] Mark disk as healthy on success
+                    self._disk_available = True
                     lib_logger.debug(
                         f"ProviderCache[{self._cache_name}]: Saved {len(self._cache)} entries"
                     )
@@ -218,6 +237,9 @@ class ProviderCache:
                         os.unlink(tmp_path)
                     raise
         except Exception as e:
+            # [RUNTIME RESILIENCE] Track disk errors for monitoring
+            self._stats["disk_errors"] += 1
+            self._disk_available = False
             lib_logger.error(f"ProviderCache[{self._cache_name}]: Disk save failed: {e}")
     
     # =========================================================================
@@ -416,12 +438,17 @@ class ProviderCache:
         return False
     
     def get_stats(self) -> Dict[str, Any]:
-        """Get cache statistics."""
+        """Get cache statistics including disk health.
+        
+        [RUNTIME RESILIENCE] Includes disk_available flag for monitoring
+        the health of disk persistence.
+        """
         return {
             **self._stats,
             "memory_entries": len(self._cache),
             "dirty": self._dirty,
-            "disk_enabled": self._enable_disk
+            "disk_enabled": self._enable_disk,
+            "disk_available": self._disk_available  # [RUNTIME RESILIENCE] Health indicator
         }
     
     async def clear(self) -> None:

src/rotator_library/usage_manager.py

@@ -90,25 +90,56 @@ class UsageManager:
                 self._initialized.set()
 
     async def _load_usage(self):
-        """Loads usage data from the JSON file asynchronously."""
+        """Loads usage data from the JSON file asynchronously with enhanced resilience.
+        
+        [RUNTIME RESILIENCE] Handles various file system errors gracefully,
+        including race conditions where file is deleted between exists check and open.
+        """
         async with self._data_lock:
             if not os.path.exists(self.file_path):
                 self._usage_data = {}
                 return
+
             try:
                 async with aiofiles.open(self.file_path, "r") as f:
                     content = await f.read()
-                    self._usage_data = json.loads(content)
-            except (json.JSONDecodeError, IOError, FileNotFoundError):
+                    self._usage_data = json.loads(content) if content.strip() else {}
+            except FileNotFoundError:
+                # [RACE CONDITION HANDLING] File deleted between exists check and open
+                self._usage_data = {}
+            except json.JSONDecodeError as e:
+                lib_logger.warning(f"Corrupted usage file {self.file_path}: {e}. Starting fresh.")
+                self._usage_data = {}
+            except (OSError, PermissionError, IOError) as e:
+                lib_logger.warning(f"Cannot read usage file {self.file_path}: {e}. Using empty state.")
                 self._usage_data = {}
 
     async def _save_usage(self):
-        """Saves the current usage data to the JSON file asynchronously."""
+        """Saves the current usage data to the JSON file asynchronously with resilience.
+        
+        [RUNTIME RESILIENCE] Wraps file operations in try/except to prevent crashes
+        if the file or directory is deleted during runtime. The in-memory state
+        continues to work even if disk persistence fails.
+        """
         if self._usage_data is None:
             return
-        async with self._data_lock:
-            async with aiofiles.open(self.file_path, "w") as f:
-                await f.write(json.dumps(self._usage_data, indent=2))
+
+        try:
+            async with self._data_lock:
+                # [DIRECTORY AUTO-RECREATION] Ensure directory exists before write
+                file_dir = os.path.dirname(os.path.abspath(self.file_path))
+                if file_dir and not os.path.exists(file_dir):
+                    os.makedirs(file_dir, exist_ok=True)
+
+                async with aiofiles.open(self.file_path, "w") as f:
+                    await f.write(json.dumps(self._usage_data, indent=2))
+        except (OSError, PermissionError, IOError) as e:
+            # [FAIL SILENTLY, LOG LOUDLY] Log the error but don't crash
+            # In-memory state is preserved and will continue to work
+            lib_logger.warning(
+                f"Failed to save usage data to {self.file_path}: {e}. "
+                "Data will be retained in memory but may be lost on restart."
+            )
 
     async def _reset_daily_stats_if_needed(self):
         """Checks if daily stats need to be reset for any key."""