feat(io): ✨ add fault-tolerant file operations with automatic recovery

Enhances application reliability by introducing a comprehensive I/O abstraction layer that eliminates crashes from filesystem issues. The system distinguishes between critical state files (credentials, usage data) that require memory buffering with retry logic, and disposable logs that can be safely dropped on failure.

Key improvements:

- New `ResilientStateWriter` class maintains in-memory state for critical files with background retry mechanism on disk failure
- Introduced `safe_write_json`, `safe_log_write`, and `safe_mkdir` utility functions for one-shot operations with graceful degradation
- Logging subsystems (`DetailedLogger`, `failure_logger`) now drop data on disk failure to prevent memory exhaustion during streaming
- Authentication providers (`GoogleOAuthBase`, `IFlowAuthBase`, `QwenAuthBase`) preserve credentials in memory when filesystem becomes unavailable
- `UsageManager` delegates persistence to `ResilientStateWriter` for automatic recovery from transient failures
- `ProviderCache` disk operations now fail silently while maintaining in-memory functionality
- Replaced scattered tempfile/atomic write patterns with centralized implementation featuring consistent error handling
- All directory creation operations now proceed gracefully if parent paths are inaccessible
- Thread-safe writer implementation supports concurrent usage from async contexts

BREAKING CHANGE: `ProviderCache._save_to_disk()` no longer raises exceptions on filesystem errors. Consumers relying on exception handling for disk write failures must now check the `disk_available` field in `get_stats()` return value for monitoring disk health.

src/proxy_app/detailed_logger.py

@@ -3,20 +3,27 @@ import time
 import uuid
 from datetime import datetime
 from pathlib import Path
-from typing import Any, Dict, Optional, List
+from typing import Any, Dict, Optional
 import logging
 
+from rotator_library.utils.resilient_io import (
+    safe_write_json,
+    safe_log_write,
+    safe_mkdir,
+)
+
 LOGS_DIR = Path(__file__).resolve().parent.parent.parent / "logs"
 DETAILED_LOGS_DIR = LOGS_DIR / "detailed_logs"
 
+
 class DetailedLogger:
     """
     Logs comprehensive details of each API transaction to a unique, timestamped directory.
+
+    Uses fire-and-forget logging - if disk writes fail, logs are dropped (not buffered)
+    to prevent memory issues, especially with streaming responses.
     """
-    # 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.
@@ -26,33 +33,24 @@ class DetailedLogger:
         timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
         self.log_dir = DETAILED_LOGS_DIR / f"{timestamp}_{self.request_id}"
         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
+        self._dir_available = safe_mkdir(self.log_dir, logging)
 
     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 (OSError, PermissionError, IOError) as e:
-            DetailedLogger._disk_available = False
-            logging.error(f"[{self.request_id}] Failed to write to {filename}: {e}")
-            self._in_memory_logs.append({"file": filename, "data": data})
+        if not self._dir_available:
+            # Try to create directory again in case it was recreated
+            self._dir_available = safe_mkdir(self.log_dir, logging)
+            if not self._dir_available:
+                return
+
+        safe_write_json(
+            self.log_dir / filename,
+            data,
+            logging,
+            atomic=False,
+            indent=4,
+            ensure_ascii=False,
+        )
 
     def log_request(self, headers: Dict[str, Any], body: Dict[str, Any]):
         """Logs the initial request details."""
@@ -61,29 +59,22 @@ class DetailedLogger:
             "request_id": self.request_id,
             "timestamp_utc": datetime.utcnow().isoformat(),
             "headers": dict(headers),
-            "body": body
+            "body": body,
         }
         self._write_json("request.json", request_data)
 
     def log_stream_chunk(self, chunk: Dict[str, Any]):
         """Logs an individual chunk from a streaming response to a JSON Lines file."""
-        # Intentionally skip memory fallback for streams to prevent OOM - unlike _write_json, we don't buffer stream chunks in memory
-        if not DetailedLogger._disk_available:
+        if not self._dir_available:
             return
-        
-        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 (OSError, PermissionError, IOError) as e:
-            DetailedLogger._disk_available = False
-            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]):
+
+        log_entry = {"timestamp_utc": datetime.utcnow().isoformat(), "chunk": chunk}
+        content = json.dumps(log_entry, ensure_ascii=False) + "\n"
+        safe_log_write(self.log_dir / "streaming_chunks.jsonl", content, logging)
+
+    def log_final_response(
+        self, status_code: int, headers: Optional[Dict[str, Any]], body: Dict[str, Any]
+    ):
         """Logs the complete final response, either from a non-streaming call or after reassembling a stream."""
         end_time = time.time()
         duration_ms = (end_time - self.start_time) * 1000
@@ -94,7 +85,7 @@ class DetailedLogger:
             "status_code": status_code,
             "duration_ms": round(duration_ms),
             "headers": dict(headers) if headers else None,
-            "body": body
+            "body": body,
         }
         self._write_json("final_response.json", response_data)
         self._log_metadata(response_data)
@@ -103,10 +94,10 @@ class DetailedLogger:
         """Recursively searches for and extracts 'reasoning' fields from the response body."""
         if not isinstance(response_body, dict):
             return None
-        
+
         if "reasoning" in response_body:
             return response_body["reasoning"]
-            
+
         if "choices" in response_body and response_body["choices"]:
             message = response_body["choices"][0].get("message", {})
             if "reasoning" in message:
@@ -121,8 +112,13 @@ class DetailedLogger:
         usage = response_data.get("body", {}).get("usage") or {}
         model = response_data.get("body", {}).get("model", "N/A")
         finish_reason = "N/A"
-        if "choices" in response_data.get("body", {}) and response_data["body"]["choices"]:
-            finish_reason = response_data["body"]["choices"][0].get("finish_reason", "N/A")
+        if (
+            "choices" in response_data.get("body", {})
+            and response_data["body"]["choices"]
+        ):
+            finish_reason = response_data["body"]["choices"][0].get(
+                "finish_reason", "N/A"
+            )
 
         metadata = {
             "request_id": self.request_id,
@@ -138,12 +134,12 @@ class DetailedLogger:
             },
             "finish_reason": finish_reason,
             "reasoning_found": False,
-            "reasoning_content": None
+            "reasoning_content": None,
         }
 
         reasoning = self._extract_reasoning(response_data.get("body", {}))
         if reasoning:
             metadata["reasoning_found"] = True
             metadata["reasoning_content"] = reasoning
-        
-        self._write_json("metadata.json", metadata)
+
+        self._write_json("metadata.json", metadata)

src/rotator_library/failure_logger.py

@@ -5,74 +5,42 @@ import os
 from datetime import datetime
 from .error_handler import mask_credential
 
-# Module-level state for resilience
-_file_handler = None
-_fallback_mode = False
 
-
-# Custom JSON formatter for structured logs (defined at module level for reuse)
 class JsonFormatter(logging.Formatter):
+    """Custom JSON formatter for structured logs."""
+
     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
+def setup_failure_logger():
+    """Sets up a dedicated JSON logger for writing detailed failure logs to a file."""
     log_dir = "logs"
-    
+    logger = logging.getLogger("failure_logger")
+    logger.setLevel(logging.INFO)
+    logger.propagate = False
+
+    # Clear existing handlers to prevent duplicates on re-setup
+    logger.handlers.clear()
+
     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
+        logger.addHandler(handler)
     except (OSError, PermissionError, IOError) as e:
         logging.warning(f"Cannot create failure log file handler: {e}")
-        _fallback_mode = True
-        return None
-
-
-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:
+        # Add NullHandler to prevent "no handlers" warning
         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)
+    return logger
 
 
 # Initialize the dedicated logger for detailed failure logs
@@ -180,25 +148,19 @@ def log_failure(
         "request_headers": request_headers,
         "error_chain": error_chain if len(error_chain) > 1 else None,
     }
-    
+
     # 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
+
+    # Log to failure logger with resilience - if it fails, just continue
     try:
         failure_logger.error(detailed_log_data)
     except (OSError, IOError) as e:
-        global _fallback_mode
-        _fallback_mode = True
-        # File logging failed - log to console instead
-        logging.error(f"Failed to write to failures.log: {e}")
-        logging.error(f"Failure summary: {summary_message}")
-    
+        # Log file write failed - log to console instead
+        logging.warning(f"Failed to write to failures.log: {e}")
+
     # Console log always succeeds
     main_lib_logger.error(summary_message)

src/rotator_library/providers/google_oauth_base.py

@@ -9,8 +9,6 @@ import asyncio
 import logging
 from pathlib import Path
 from typing import Dict, Any
-import tempfile
-import shutil
 
 import httpx
 from rich.console import Console
@@ -20,6 +18,7 @@ from rich.markup import escape as rich_escape
 
 from ..utils.headless_detection import is_headless_environment
 from ..utils.reauth_coordinator import get_reauth_coordinator
+from ..utils.resilient_io import safe_write_json
 
 lib_logger = logging.getLogger("rotator_library")
 
@@ -264,13 +263,8 @@ 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)
+        """Save credentials with in-memory fallback if disk unavailable."""
+        # Always update cache first (memory is reliable)
         self._credentials_cache[path] = creds
 
         # Don't save to file if credentials were loaded from environment
@@ -278,62 +272,15 @@ class GoogleOAuthBase:
             lib_logger.debug("Credentials loaded from env, skipping file save")
             return
 
-        try:
-            # [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
-
-                # Set secure permissions (0600 = owner read/write only)
-                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
-
-                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
+        # Attempt disk write - if it fails, we still have the cache
+        if safe_write_json(path, creds, lib_logger, secure_permissions=True):
+            lib_logger.debug(
+                f"Saved updated {self.ENV_PREFIX} OAuth credentials to '{path}'."
+            )
+        else:
             lib_logger.warning(
-                f"Failed to save credentials to {path}: {e}. "
-                "Credentials cached in memory only (will be lost on restart)."
+                f"Credentials for {self.ENV_PREFIX} 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
@@ -952,19 +899,14 @@ class GoogleOAuthBase:
             )
 
     async def get_auth_header(self, credential_path: str) -> Dict[str, str]:
-        """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.
-        """
+        """Get auth header with graceful degradation if refresh fails."""
         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
+                    # 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(
@@ -976,7 +918,7 @@ class GoogleOAuthBase:
                         raise
             return {"Authorization": f"Bearer {creds['access_token']}"}
         except Exception as e:
-            # [FINAL FALLBACK] Check if any cached credential exists as last resort
+            # 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(

src/rotator_library/providers/iflow_auth_base.py

@@ -12,8 +12,6 @@ import os
 from pathlib import Path
 from typing import Dict, Any, Tuple, Union, Optional
 from urllib.parse import urlencode, parse_qs, urlparse
-import tempfile
-import shutil
 
 import httpx
 from aiohttp import web
@@ -24,6 +22,7 @@ from rich.text import Text
 from rich.markup import escape as rich_escape
 from ..utils.headless_detection import is_headless_environment
 from ..utils.reauth_coordinator import get_reauth_coordinator
+from ..utils.resilient_io import safe_write_json
 
 lib_logger = logging.getLogger("rotator_library")
 
@@ -316,65 +315,22 @@ class IFlowAuthBase:
             return await self._read_creds_from_file(path)
 
     async def _save_credentials(self, path: str, creds: Dict[str, Any]):
-        """Saves credentials to cache and file using atomic writes."""
+        """Save credentials with in-memory fallback if disk unavailable."""
+        # Always update cache first (memory is 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
-
-            # Set secure permissions (0600 = owner read/write only)
-            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
-
-            # Update cache AFTER successful file write
-            self._credentials_cache[path] = creds
-            lib_logger.debug(
-                f"Saved updated iFlow OAuth credentials to '{path}' (atomic write)."
-            )
-
-        except Exception as e:
-            lib_logger.error(
-                f"Failed to save updated iFlow OAuth credentials to '{path}': {e}"
+        # Attempt disk write - if it fails, we still have the cache
+        if safe_write_json(path, creds, lib_logger, secure_permissions=True):
+            lib_logger.debug(f"Saved updated iFlow OAuth credentials to '{path}'.")
+        else:
+            lib_logger.warning(
+                "iFlow credentials cached in memory only (will be lost on restart)."
             )
-            # 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
 
     def _is_token_expired(self, creds: Dict[str, Any]) -> bool:
         """Checks if the token is expired (with buffer for proactive refresh)."""

src/rotator_library/providers/provider_cache.py

@@ -20,19 +20,20 @@ import asyncio
 import json
 import logging
 import os
-import shutil
-import tempfile
 import time
 from pathlib import Path
 from typing import Any, Dict, Optional, Tuple
 
-lib_logger = logging.getLogger('rotator_library')
+from ..utils.resilient_io import safe_write_json
+
+lib_logger = logging.getLogger("rotator_library")
 
 
 # =============================================================================
 # UTILITY FUNCTIONS
 # =============================================================================
 
+
 def _env_bool(key: str, default: bool = False) -> bool:
     """Get boolean from environment variable."""
     return os.getenv(key, str(default).lower()).lower() in ("true", "1", "yes")
@@ -47,18 +48,19 @@ def _env_int(key: str, default: int) -> int:
 # PROVIDER CACHE CLASS
 # =============================================================================
 
+
 class ProviderCache:
     """
     Server-side cache for provider conversation state preservation.
-    
+
     A generic, modular cache supporting any key-value data that providers need
     to persist across requests. Features:
-    
+
     - Dual-TTL system: configurable memory TTL, longer disk TTL
     - Async disk persistence with batched writes
     - Background cleanup task for expired entries
     - Statistics tracking (hits, misses, writes)
-    
+
     Args:
         cache_file: Path to disk cache file
         memory_ttl_seconds: In-memory entry lifetime (default: 1 hour)
@@ -67,13 +69,13 @@ class ProviderCache:
         write_interval: Seconds between background disk writes (default: 60)
         cleanup_interval: Seconds between expired entry cleanup (default: 30 min)
         env_prefix: Environment variable prefix for configuration overrides
-    
+
     Environment Variables (with default prefix "PROVIDER_CACHE"):
         {PREFIX}_ENABLE: Enable/disable disk persistence
         {PREFIX}_WRITE_INTERVAL: Background write interval in seconds
         {PREFIX}_CLEANUP_INTERVAL: Cleanup interval in seconds
     """
-    
+
     def __init__(
         self,
         cache_file: Path,
@@ -82,7 +84,7 @@ class ProviderCache:
         enable_disk: Optional[bool] = None,
         write_interval: Optional[int] = None,
         cleanup_interval: Optional[int] = None,
-        env_prefix: str = "PROVIDER_CACHE"
+        env_prefix: str = "PROVIDER_CACHE",
     ):
         # In-memory cache: {cache_key: (data, timestamp)}
         self._cache: Dict[str, Tuple[str, float]] = {}
@@ -90,28 +92,42 @@ class ProviderCache:
         self._disk_ttl = disk_ttl_seconds
         self._lock = asyncio.Lock()
         self._disk_lock = asyncio.Lock()
-        
+
         # Disk persistence configuration
         self._cache_file = cache_file
-        self._enable_disk = enable_disk if enable_disk is not None else _env_bool(f"{env_prefix}_ENABLE", True)
+        self._enable_disk = (
+            enable_disk
+            if enable_disk is not None
+            else _env_bool(f"{env_prefix}_ENABLE", True)
+        )
         self._dirty = False
-        self._write_interval = write_interval or _env_int(f"{env_prefix}_WRITE_INTERVAL", 60)
-        self._cleanup_interval = cleanup_interval or _env_int(f"{env_prefix}_CLEANUP_INTERVAL", 1800)
-        
+        self._write_interval = write_interval or _env_int(
+            f"{env_prefix}_WRITE_INTERVAL", 60
+        )
+        self._cleanup_interval = cleanup_interval or _env_int(
+            f"{env_prefix}_CLEANUP_INTERVAL", 1800
+        )
+
         # Background tasks
         self._writer_task: Optional[asyncio.Task] = None
         self._cleanup_task: Optional[asyncio.Task] = None
         self._running = False
-        
+
         # Statistics
-        self._stats = {"memory_hits": 0, "disk_hits": 0, "misses": 0, "writes": 0, "disk_errors": 0}
-        
-        # [RUNTIME RESILIENCE] Track disk health for monitoring
+        self._stats = {
+            "memory_hits": 0,
+            "disk_hits": 0,
+            "misses": 0,
+            "writes": 0,
+            "disk_errors": 0,
+        }
+
+        # 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"
-        
+
         if self._enable_disk:
             lib_logger.debug(
                 f"ProviderCache[{self._cache_name}]: Disk enabled "
@@ -120,142 +136,114 @@ class ProviderCache:
             asyncio.create_task(self._async_init())
         else:
             lib_logger.debug(f"ProviderCache[{self._cache_name}]: Memory-only mode")
-    
+
     # =========================================================================
     # INITIALIZATION
     # =========================================================================
-    
+
     async def _async_init(self) -> None:
         """Async initialization: load from disk and start background tasks."""
         try:
             await self._load_from_disk()
             await self._start_background_tasks()
         except Exception as e:
-            lib_logger.error(f"ProviderCache[{self._cache_name}] async init failed: {e}")
-    
+            lib_logger.error(
+                f"ProviderCache[{self._cache_name}] async init failed: {e}"
+            )
+
     async def _load_from_disk(self) -> None:
         """Load cache from disk file with TTL validation."""
         if not self._enable_disk or not self._cache_file.exists():
             return
-        
+
         try:
             async with self._disk_lock:
-                with open(self._cache_file, 'r', encoding='utf-8') as f:
+                with open(self._cache_file, "r", encoding="utf-8") as f:
                     data = json.load(f)
-                
+
                 if data.get("version") != "1.0":
-                    lib_logger.warning(f"ProviderCache[{self._cache_name}]: Version mismatch, starting fresh")
+                    lib_logger.warning(
+                        f"ProviderCache[{self._cache_name}]: Version mismatch, starting fresh"
+                    )
                     return
-                
+
                 now = time.time()
                 entries = data.get("entries", {})
                 loaded = expired = 0
-                
+
                 for cache_key, entry in entries.items():
                     age = now - entry.get("timestamp", 0)
                     if age <= self._disk_ttl:
-                        value = entry.get("value", entry.get("signature", ""))  # Support both formats
+                        value = entry.get(
+                            "value", entry.get("signature", "")
+                        )  # Support both formats
                         if value:
                             self._cache[cache_key] = (value, entry["timestamp"])
                             loaded += 1
                     else:
                         expired += 1
-                
+
                 lib_logger.debug(
                     f"ProviderCache[{self._cache_name}]: Loaded {loaded} entries ({expired} expired)"
                 )
         except json.JSONDecodeError as e:
-            lib_logger.warning(f"ProviderCache[{self._cache_name}]: File corrupted: {e}")
+            lib_logger.warning(
+                f"ProviderCache[{self._cache_name}]: File corrupted: {e}"
+            )
         except Exception as e:
             lib_logger.error(f"ProviderCache[{self._cache_name}]: Load failed: {e}")
-    
+
     # =========================================================================
     # DISK PERSISTENCE
     # =========================================================================
-    
+
     async def _save_to_disk(self) -> None:
-        """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.
-        """
+        """Persist cache to disk using atomic write with health tracking."""
         if not self._enable_disk:
             return
-        
-        try:
-            async with self._disk_lock:
-                # [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",
-                    "memory_ttl_seconds": self._memory_ttl,
-                    "disk_ttl_seconds": self._disk_ttl,
-                    "entries": {
-                        key: {"value": val, "timestamp": ts}
-                        for key, (val, ts) in self._cache.items()
-                    },
-                    "statistics": {
-                        "total_entries": len(self._cache),
-                        "last_write": time.time(),
-                        **self._stats
-                    }
-                }
-                
-                # Atomic write using temp file
-                parent_dir = self._cache_file.parent
-                tmp_fd, tmp_path = tempfile.mkstemp(dir=parent_dir, prefix='.tmp_', suffix='.json')
-                
-                try:
-                    with os.fdopen(tmp_fd, 'w', encoding='utf-8') as f:
-                        json.dump(cache_data, f, indent=2)
-                    
-                    # Set restrictive permissions (if supported)
-                    try:
-                        os.chmod(tmp_path, 0o600)
-                    except (OSError, AttributeError):
-                        pass
-                    
-                    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"
-                    )
-                except Exception:
-                    if tmp_path and os.path.exists(tmp_path):
-                        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}")
-    
+
+        async with self._disk_lock:
+            cache_data = {
+                "version": "1.0",
+                "memory_ttl_seconds": self._memory_ttl,
+                "disk_ttl_seconds": self._disk_ttl,
+                "entries": {
+                    key: {"value": val, "timestamp": ts}
+                    for key, (val, ts) in self._cache.items()
+                },
+                "statistics": {
+                    "total_entries": len(self._cache),
+                    "last_write": time.time(),
+                    **self._stats,
+                },
+            }
+
+            if safe_write_json(
+                self._cache_file, cache_data, lib_logger, secure_permissions=True
+            ):
+                self._stats["writes"] += 1
+                self._disk_available = True
+                lib_logger.debug(
+                    f"ProviderCache[{self._cache_name}]: Saved {len(self._cache)} entries"
+                )
+            else:
+                self._stats["disk_errors"] += 1
+                self._disk_available = False
+
     # =========================================================================
     # BACKGROUND TASKS
     # =========================================================================
-    
+
     async def _start_background_tasks(self) -> None:
         """Start background writer and cleanup tasks."""
         if not self._enable_disk or self._running:
             return
-        
+
         self._running = True
         self._writer_task = asyncio.create_task(self._writer_loop())
         self._cleanup_task = asyncio.create_task(self._cleanup_loop())
         lib_logger.debug(f"ProviderCache[{self._cache_name}]: Started background tasks")
-    
+
     async def _writer_loop(self) -> None:
         """Background task: periodically flush dirty cache to disk."""
         try:
@@ -266,10 +254,12 @@ class ProviderCache:
                         await self._save_to_disk()
                         self._dirty = False
                     except Exception as e:
-                        lib_logger.error(f"ProviderCache[{self._cache_name}]: Writer error: {e}")
+                        lib_logger.error(
+                            f"ProviderCache[{self._cache_name}]: Writer error: {e}"
+                        )
         except asyncio.CancelledError:
             pass
-    
+
     async def _cleanup_loop(self) -> None:
         """Background task: periodically clean up expired entries."""
         try:
@@ -278,12 +268,14 @@ class ProviderCache:
                 await self._cleanup_expired()
         except asyncio.CancelledError:
             pass
-    
+
     async def _cleanup_expired(self) -> None:
         """Remove expired entries from memory cache."""
         async with self._lock:
             now = time.time()
-            expired = [k for k, (_, ts) in self._cache.items() if now - ts > self._memory_ttl]
+            expired = [
+                k for k, (_, ts) in self._cache.items() if now - ts > self._memory_ttl
+            ]
             for k in expired:
                 del self._cache[k]
             if expired:
@@ -291,42 +283,42 @@ class ProviderCache:
                 lib_logger.debug(
                     f"ProviderCache[{self._cache_name}]: Cleaned {len(expired)} expired entries"
                 )
-    
+
     # =========================================================================
     # CORE OPERATIONS
     # =========================================================================
-    
+
     def store(self, key: str, value: str) -> None:
         """
         Store a value synchronously (schedules async storage).
-        
+
         Args:
             key: Cache key
             value: Value to store (typically JSON-serialized data)
         """
         asyncio.create_task(self._async_store(key, value))
-    
+
     async def _async_store(self, key: str, value: str) -> None:
         """Async implementation of store."""
         async with self._lock:
             self._cache[key] = (value, time.time())
             self._dirty = True
-    
+
     async def store_async(self, key: str, value: str) -> None:
         """
         Store a value asynchronously (awaitable).
-        
+
         Use this when you need to ensure the value is stored before continuing.
         """
         await self._async_store(key, value)
-    
+
     def retrieve(self, key: str) -> Optional[str]:
         """
         Retrieve a value by key (synchronous, with optional async disk fallback).
-        
+
         Args:
             key: Cache key
-            
+
         Returns:
             Cached value if found and not expired, None otherwise
         """
@@ -338,17 +330,17 @@ class ProviderCache:
             else:
                 del self._cache[key]
                 self._dirty = True
-        
+
         self._stats["misses"] += 1
         if self._enable_disk:
             # Schedule async disk lookup for next time
             asyncio.create_task(self._check_disk_fallback(key))
         return None
-    
+
     async def retrieve_async(self, key: str) -> Optional[str]:
         """
         Retrieve a value asynchronously (checks disk if not in memory).
-        
+
         Use this when you can await and need guaranteed disk fallback.
         """
         # Check memory first
@@ -362,24 +354,24 @@ class ProviderCache:
                     if key in self._cache:
                         del self._cache[key]
                         self._dirty = True
-        
+
         # Check disk
         if self._enable_disk:
             return await self._disk_retrieve(key)
-        
+
         self._stats["misses"] += 1
         return None
-    
+
     async def _check_disk_fallback(self, key: str) -> None:
         """Check disk for key and load into memory if found (background)."""
         try:
             if not self._cache_file.exists():
                 return
-            
+
             async with self._disk_lock:
-                with open(self._cache_file, 'r', encoding='utf-8') as f:
+                with open(self._cache_file, "r", encoding="utf-8") as f:
                     data = json.load(f)
-                
+
                 entries = data.get("entries", {})
                 if key in entries:
                     entry = entries[key]
@@ -394,19 +386,21 @@ class ProviderCache:
                                 f"ProviderCache[{self._cache_name}]: Loaded {key} from disk"
                             )
         except Exception as e:
-            lib_logger.debug(f"ProviderCache[{self._cache_name}]: Disk fallback failed: {e}")
-    
+            lib_logger.debug(
+                f"ProviderCache[{self._cache_name}]: Disk fallback failed: {e}"
+            )
+
     async def _disk_retrieve(self, key: str) -> Optional[str]:
         """Direct disk retrieval with loading into memory."""
         try:
             if not self._cache_file.exists():
                 self._stats["misses"] += 1
                 return None
-            
+
             async with self._disk_lock:
-                with open(self._cache_file, 'r', encoding='utf-8') as f:
+                with open(self._cache_file, "r", encoding="utf-8") as f:
                     data = json.load(f)
-                
+
                 entries = data.get("entries", {})
                 if key in entries:
                     entry = entries[key]
@@ -418,39 +412,37 @@ class ProviderCache:
                                 self._cache[key] = (value, ts)
                             self._stats["disk_hits"] += 1
                             return value
-            
+
             self._stats["misses"] += 1
             return None
         except Exception as e:
-            lib_logger.debug(f"ProviderCache[{self._cache_name}]: Disk retrieve failed: {e}")
+            lib_logger.debug(
+                f"ProviderCache[{self._cache_name}]: Disk retrieve failed: {e}"
+            )
             self._stats["misses"] += 1
             return None
-    
+
     # =========================================================================
     # UTILITY METHODS
     # =========================================================================
-    
+
     def contains(self, key: str) -> bool:
         """Check if key exists in memory cache (without updating stats)."""
         if key in self._cache:
             _, timestamp = self._cache[key]
             return time.time() - timestamp <= self._memory_ttl
         return False
-    
+
     def get_stats(self) -> Dict[str, Any]:
-        """Get cache statistics including disk health.
-        
-        [RUNTIME RESILIENCE] Includes disk_available flag for monitoring
-        the health of disk persistence.
-        """
+        """Get cache statistics including disk health."""
         return {
             **self._stats,
             "memory_entries": len(self._cache),
             "dirty": self._dirty,
             "disk_enabled": self._enable_disk,
-            "disk_available": self._disk_available  # [RUNTIME RESILIENCE] Health indicator
+            "disk_available": self._disk_available,
         }
-    
+
     async def clear(self) -> None:
         """Clear all cached data."""
         async with self._lock:
@@ -458,12 +450,12 @@ class ProviderCache:
             self._dirty = True
         if self._enable_disk:
             await self._save_to_disk()
-    
+
     async def shutdown(self) -> None:
         """Graceful shutdown: flush pending writes and stop background tasks."""
         lib_logger.info(f"ProviderCache[{self._cache_name}]: Shutting down...")
         self._running = False
-        
+
         # Cancel background tasks
         for task in (self._writer_task, self._cleanup_task):
             if task:
@@ -472,11 +464,11 @@ class ProviderCache:
                     await task
                 except asyncio.CancelledError:
                     pass
-        
+
         # Final save
         if self._dirty and self._enable_disk:
             await self._save_to_disk()
-        
+
         lib_logger.info(
             f"ProviderCache[{self._cache_name}]: Shutdown complete "
             f"(stats: mem_hits={self._stats['memory_hits']}, "
@@ -488,38 +480,39 @@ class ProviderCache:
 # CONVENIENCE FACTORY
 # =============================================================================
 
+
 def create_provider_cache(
     name: str,
     cache_dir: Optional[Path] = None,
     memory_ttl_seconds: int = 3600,
     disk_ttl_seconds: int = 86400,
-    env_prefix: Optional[str] = None
+    env_prefix: Optional[str] = None,
 ) -> ProviderCache:
     """
     Factory function to create a provider cache with sensible defaults.
-    
+
     Args:
         name: Cache name (used as filename and for logging)
         cache_dir: Directory for cache file (default: project_root/cache/provider_name)
         memory_ttl_seconds: In-memory TTL
         disk_ttl_seconds: Disk TTL
         env_prefix: Environment variable prefix (default: derived from name)
-    
+
     Returns:
         Configured ProviderCache instance
     """
     if cache_dir is None:
         cache_dir = Path(__file__).resolve().parent.parent.parent.parent / "cache"
-    
+
     cache_file = cache_dir / f"{name}.json"
-    
+
     if env_prefix is None:
         # Convert name to env prefix: "gemini3_signatures" -> "GEMINI3_SIGNATURES_CACHE"
         env_prefix = f"{name.upper().replace('-', '_')}_CACHE"
-    
+
     return ProviderCache(
         cache_file=cache_file,
         memory_ttl_seconds=memory_ttl_seconds,
         disk_ttl_seconds=disk_ttl_seconds,
-        env_prefix=env_prefix
+        env_prefix=env_prefix,
     )

src/rotator_library/providers/qwen_auth_base.py

@@ -11,8 +11,6 @@ import webbrowser
 import os
 from pathlib import Path
 from typing import Dict, Any, Tuple, Union, Optional
-import tempfile
-import shutil
 
 import httpx
 from rich.console import Console
@@ -23,6 +21,7 @@ from rich.markup import escape as rich_escape
 
 from ..utils.headless_detection import is_headless_environment
 from ..utils.reauth_coordinator import get_reauth_coordinator
+from ..utils.resilient_io import safe_write_json
 
 lib_logger = logging.getLogger("rotator_library")
 
@@ -201,63 +200,22 @@ class QwenAuthBase:
             return await self._read_creds_from_file(path)
 
     async def _save_credentials(self, path: str, creds: Dict[str, Any]):
+        """Save credentials with in-memory fallback if disk unavailable."""
+        # Always update cache first (memory is 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
-        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
-
-            # Set secure permissions (0600 = owner read/write only)
-            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
-
-            # Update cache AFTER successful file write
-            self._credentials_cache[path] = creds
-            lib_logger.debug(
-                f"Saved updated Qwen OAuth credentials to '{path}' (atomic write)."
-            )
-
-        except Exception as e:
-            lib_logger.error(
-                f"Failed to save updated Qwen OAuth credentials to '{path}': {e}"
+        # Attempt disk write - if it fails, we still have the cache
+        if safe_write_json(path, creds, lib_logger, secure_permissions=True):
+            lib_logger.debug(f"Saved updated Qwen OAuth credentials to '{path}'.")
+        else:
+            lib_logger.warning(
+                "Qwen credentials cached in memory only (will be lost on restart)."
             )
-            # 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
 
     def _is_token_expired(self, creds: Dict[str, Any]) -> bool:
         expiry_timestamp = creds.get("expiry_date", 0) / 1000

src/rotator_library/usage_manager.py

@@ -11,6 +11,7 @@ import litellm
 
 from .error_handler import ClassifiedError, NoAvailableKeysError, mask_credential
 from .providers import PROVIDER_PLUGINS
+from .utils.resilient_io import ResilientStateWriter
 
 lib_logger = logging.getLogger("rotator_library")
 lib_logger.propagate = False
@@ -103,8 +104,8 @@ class UsageManager:
         self._timeout_lock = asyncio.Lock()
         self._claimed_on_timeout: Set[str] = set()
 
-        # Circuit breaker for disk write failures
-        self._disk_available = True
+        # Resilient writer for usage data persistence
+        self._state_writer = ResilientStateWriter(file_path, lib_logger)
 
         if daily_reset_time_utc:
             hour, minute = map(int, daily_reset_time_utc.split(":"))
@@ -543,11 +544,7 @@ class UsageManager:
                 self._initialized.set()
 
     async def _load_usage(self):
-        """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.
-        """
+        """Loads usage data from the JSON file asynchronously with resilience."""
         async with self._data_lock:
             if not os.path.exists(self.file_path):
                 self._usage_data = {}
@@ -558,7 +555,7 @@ class UsageManager:
                     content = await f.read()
                     self._usage_data = json.loads(content) if content.strip() else {}
             except FileNotFoundError:
-                # [RACE CONDITION HANDLING] File deleted between exists check and open
+                # File deleted between exists check and open
                 self._usage_data = {}
             except json.JSONDecodeError as e:
                 lib_logger.warning(
@@ -570,43 +567,17 @@ class UsageManager:
                     f"Cannot read usage file {self.file_path}: {e}. Using empty state."
                 )
                 self._usage_data = {}
-            else:
-                # [CIRCUIT BREAKER RESET] Successfully loaded, re-enable disk writes
-                self._disk_available = True
 
     async def _save_usage(self):
-        """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.
-        """
+        """Saves the current usage data using the resilient state writer."""
         if self._usage_data is None:
             return
 
-        if not self._disk_available:
-            return  # Skip disk write when unavailable
-
-        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)
-
-                # Add human-readable timestamp fields before saving
-                self._add_readable_timestamps(self._usage_data)
-                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:
-            # [CIRCUIT BREAKER] Disable disk writes to prevent repeated failures
-            self._disk_available = False
-            # [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 with self._data_lock:
+            # Add human-readable timestamp fields before saving
+            self._add_readable_timestamps(self._usage_data)
+            # Hand off to resilient writer - handles retries and disk failures
+            self._state_writer.write(self._usage_data)
 
     async def _reset_daily_stats_if_needed(self):
         """

src/rotator_library/utils/__init__.py

@@ -2,5 +2,19 @@
 
 from .headless_detection import is_headless_environment
 from .reauth_coordinator import get_reauth_coordinator, ReauthCoordinator
+from .resilient_io import (
+    ResilientStateWriter,
+    safe_write_json,
+    safe_log_write,
+    safe_mkdir,
+)
 
-__all__ = ["is_headless_environment", "get_reauth_coordinator", "ReauthCoordinator"]
+__all__ = [
+    "is_headless_environment",
+    "get_reauth_coordinator",
+    "ReauthCoordinator",
+    "ResilientStateWriter",
+    "safe_write_json",
+    "safe_log_write",
+    "safe_mkdir",
+]

src/rotator_library/utils/resilient_io.py

@@ -0,0 +1,339 @@
+# src/rotator_library/utils/resilient_io.py
+"""
+Resilient I/O utilities for handling file operations gracefully.
+
+Provides two main patterns:
+1. ResilientStateWriter - For stateful files (usage.json, credentials, cache)
+   that should be buffered in memory and retried on disk failure.
+2. safe_log_write / safe_write_json - For logs that can be dropped on failure.
+"""
+
+import json
+import os
+import shutil
+import tempfile
+import threading
+import time
+import logging
+from pathlib import Path
+from typing import Any, Callable, Dict, Optional, Union
+
+
+class ResilientStateWriter:
+    """
+    Manages resilient writes for stateful files (usage stats, credentials, cache).
+
+    Design:
+    - Caller hands off data via write() - always succeeds (memory update)
+    - Attempts disk write immediately
+    - If disk fails, retries periodically in background
+    - On recovery, writes full current state (not just new data)
+
+    Thread-safe for use in async contexts with sync file I/O.
+
+    Usage:
+        writer = ResilientStateWriter("data.json", logger)
+        writer.write({"key": "value"})  # Always succeeds
+        # ... later ...
+        if not writer.is_healthy:
+            logger.warning("Disk writes failing, data in memory only")
+    """
+
+    def __init__(
+        self,
+        path: Union[str, Path],
+        logger: logging.Logger,
+        retry_interval: float = 30.0,
+        serializer: Optional[Callable[[Any], str]] = None,
+    ):
+        """
+        Initialize the resilient writer.
+
+        Args:
+            path: File path to write to
+            logger: Logger for warnings/errors
+            retry_interval: Seconds between retry attempts when disk is unhealthy
+            serializer: Custom serializer function (defaults to JSON with indent=2)
+        """
+        self.path = Path(path)
+        self.logger = logger
+        self.retry_interval = retry_interval
+        self._serializer = serializer or (lambda d: json.dumps(d, indent=2))
+
+        self._current_state: Optional[Any] = None
+        self._disk_healthy = True
+        self._last_attempt: float = 0
+        self._last_success: Optional[float] = None
+        self._failure_count = 0
+        self._lock = threading.Lock()
+
+    def write(self, data: Any) -> bool:
+        """
+        Update state and attempt disk write.
+
+        Always updates in-memory state (guaranteed to succeed).
+        Attempts disk write - if it fails, schedules for retry.
+
+        Args:
+            data: Data to persist (must be serializable)
+
+        Returns:
+            True if disk write succeeded, False if failed (data still in memory)
+        """
+        with self._lock:
+            self._current_state = data
+            return self._try_disk_write()
+
+    def retry_if_needed(self) -> bool:
+        """
+        Retry disk write if unhealthy and retry interval has passed.
+
+        Call this periodically (e.g., on each save attempt) to recover
+        from transient disk failures.
+
+        Returns:
+            True if healthy (no retry needed or retry succeeded)
+        """
+        with self._lock:
+            if self._disk_healthy:
+                return True
+
+            if self._current_state is None:
+                return True
+
+            now = time.time()
+            if now - self._last_attempt < self.retry_interval:
+                return False
+
+            return self._try_disk_write()
+
+    def _try_disk_write(self) -> bool:
+        """
+        Attempt atomic write to disk. Updates health status.
+
+        Uses tempfile + move pattern for atomic writes on POSIX systems.
+        On Windows, uses direct write (still safe for our use case).
+        """
+        if self._current_state is None:
+            return True
+
+        self._last_attempt = time.time()
+
+        try:
+            # Ensure directory exists
+            self.path.parent.mkdir(parents=True, exist_ok=True)
+
+            # Serialize data
+            content = self._serializer(self._current_state)
+
+            # Atomic write: write to temp file, then move
+            tmp_fd = None
+            tmp_path = None
+            try:
+                tmp_fd, tmp_path = tempfile.mkstemp(
+                    dir=self.path.parent, prefix=".tmp_", suffix=".json", text=True
+                )
+
+                with os.fdopen(tmp_fd, "w", encoding="utf-8") as f:
+                    f.write(content)
+                    tmp_fd = None  # fdopen closes the fd
+
+                # Atomic move
+                shutil.move(tmp_path, self.path)
+                tmp_path = None
+
+            finally:
+                # Cleanup on failure
+                if tmp_fd is not None:
+                    try:
+                        os.close(tmp_fd)
+                    except OSError:
+                        pass
+                if tmp_path and os.path.exists(tmp_path):
+                    try:
+                        os.unlink(tmp_path)
+                    except OSError:
+                        pass
+
+            # Success - update health
+            self._disk_healthy = True
+            self._last_success = time.time()
+            self._failure_count = 0
+            return True
+
+        except (OSError, PermissionError, IOError) as e:
+            self._disk_healthy = False
+            self._failure_count += 1
+
+            # Log warning (rate-limited to avoid flooding)
+            if self._failure_count == 1 or self._failure_count % 10 == 0:
+                self.logger.warning(
+                    f"Failed to write {self.path.name}: {e}. "
+                    f"Data retained in memory (failure #{self._failure_count})."
+                )
+            return False
+
+    @property
+    def is_healthy(self) -> bool:
+        """Check if disk writes are currently working."""
+        return self._disk_healthy
+
+    @property
+    def current_state(self) -> Optional[Any]:
+        """Get the current in-memory state (for inspection/debugging)."""
+        return self._current_state
+
+    def get_health_info(self) -> Dict[str, Any]:
+        """
+        Get detailed health information for monitoring.
+
+        Returns dict with:
+            - healthy: bool
+            - failure_count: int
+            - last_success: Optional[float] (timestamp)
+            - last_attempt: float (timestamp)
+            - path: str
+        """
+        return {
+            "healthy": self._disk_healthy,
+            "failure_count": self._failure_count,
+            "last_success": self._last_success,
+            "last_attempt": self._last_attempt,
+            "path": str(self.path),
+        }
+
+
+def safe_write_json(
+    path: Union[str, Path],
+    data: Dict[str, Any],
+    logger: logging.Logger,
+    atomic: bool = True,
+    indent: int = 2,
+    ensure_ascii: bool = True,
+    secure_permissions: bool = False,
+) -> bool:
+    """
+    Write JSON data to file with error handling. No buffering or retry.
+
+    Suitable for one-off writes where failure is acceptable (e.g., logs).
+    Creates parent directories if needed.
+
+    Args:
+        path: File path to write to
+        data: JSON-serializable data
+        logger: Logger for warnings
+        atomic: Use atomic write pattern (tempfile + move)
+        indent: JSON indentation level (default: 2)
+        ensure_ascii: Escape non-ASCII characters (default: True)
+        secure_permissions: Set file permissions to 0o600 (default: False)
+
+    Returns:
+        True on success, False on failure (never raises)
+    """
+    path = Path(path)
+
+    try:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        content = json.dumps(data, indent=indent, ensure_ascii=ensure_ascii)
+
+        if atomic:
+            tmp_fd = None
+            tmp_path = None
+            try:
+                tmp_fd, tmp_path = tempfile.mkstemp(
+                    dir=path.parent, prefix=".tmp_", suffix=".json", text=True
+                )
+                with os.fdopen(tmp_fd, "w", encoding="utf-8") as f:
+                    f.write(content)
+                    tmp_fd = None
+
+                # Set secure permissions if requested (before move for security)
+                if secure_permissions:
+                    try:
+                        os.chmod(tmp_path, 0o600)
+                    except (OSError, AttributeError):
+                        # Windows may not support chmod, ignore
+                        pass
+
+                shutil.move(tmp_path, path)
+                tmp_path = None
+            finally:
+                if tmp_fd is not None:
+                    try:
+                        os.close(tmp_fd)
+                    except OSError:
+                        pass
+                if tmp_path and os.path.exists(tmp_path):
+                    try:
+                        os.unlink(tmp_path)
+                    except OSError:
+                        pass
+        else:
+            with open(path, "w", encoding="utf-8") as f:
+                f.write(content)
+
+            # Set secure permissions if requested
+            if secure_permissions:
+                try:
+                    os.chmod(path, 0o600)
+                except (OSError, AttributeError):
+                    pass
+
+        return True
+
+    except (OSError, PermissionError, IOError, TypeError, ValueError) as e:
+        logger.warning(f"Failed to write JSON to {path}: {e}")
+        return False
+
+
+def safe_log_write(
+    path: Union[str, Path],
+    content: str,
+    logger: logging.Logger,
+    mode: str = "a",
+) -> bool:
+    """
+    Write content to log file with error handling. No buffering or retry.
+
+    Suitable for log files where occasional loss is acceptable.
+    Creates parent directories if needed.
+
+    Args:
+        path: File path to write to
+        content: String content to write
+        logger: Logger for warnings
+        mode: File mode ('a' for append, 'w' for overwrite)
+
+    Returns:
+        True on success, False on failure (never raises)
+    """
+    path = Path(path)
+
+    try:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        with open(path, mode, encoding="utf-8") as f:
+            f.write(content)
+        return True
+
+    except (OSError, PermissionError, IOError) as e:
+        logger.warning(f"Failed to write log to {path}: {e}")
+        return False
+
+
+def safe_mkdir(path: Union[str, Path], logger: logging.Logger) -> bool:
+    """
+    Create directory with error handling.
+
+    Args:
+        path: Directory path to create
+        logger: Logger for warnings
+
+    Returns:
+        True on success (or already exists), False on failure
+    """
+    try:
+        Path(path).mkdir(parents=True, exist_ok=True)
+        return True
+    except (OSError, PermissionError) as e:
+        logger.warning(f"Failed to create directory {path}: {e}")
+        return False