feat(io): ✨ add shutdown flush mechanism for buffered writes

This commit introduces a global buffered write registry with automatic shutdown flush, ensuring critical data (auth tokens, usage stats) is saved even when disk writes fail temporarily.

- Add `BufferedWriteRegistry` singleton for centralized buffered write management
- Implement periodic retry (30s interval) and atexit shutdown flush for pending writes
- Enable `buffer_on_failure` parameter in `safe_write_json()` for credential files
- Integrate buffering with `ResilientStateWriter` for automatic registry registration
- Update OAuth providers (Google, Qwen, iFlow) to use buffered credential writes
- Change provider cache `_save_to_disk()` to return success status for better tracking
- Reduce log noise by changing missing thoughtSignature warnings to debug level
- Export `BufferedWriteRegistry` from utils module for monitoring access

The new architecture ensures data is never lost on graceful shutdown (Ctrl+C), with console output showing flush progress and results. All buffered writes are retried in a background thread and guaranteed a final save attempt on application exit.

DOCUMENTATION.md

@@ -939,31 +939,173 @@ This level of detail allows developers to trace exactly why a request failed or
 
 ## 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.
+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 data files or directories are deleted while the application is running.
 
-### 5.1. Resilience Hierarchy
+### 5.1. Centralized Resilient I/O (`resilient_io.py`)
+
+All file operations are centralized in a single utility module that provides consistent error handling, graceful degradation, and automatic retry with shutdown flush:
+
+#### `BufferedWriteRegistry` (Singleton)
+
+Global registry for buffered writes with periodic retry and shutdown flush. Ensures critical data is saved even if disk writes fail temporarily:
+
+- **Per-file buffering**: Each file path has its own pending write (latest data always wins)
+- **Periodic retries**: Background thread retries failed writes every 30 seconds
+- **Shutdown flush**: `atexit` hook ensures final write attempt on app exit (Ctrl+C)
+- **Thread-safe**: Safe for concurrent access from multiple threads
+
+```python
+# Get the singleton instance
+registry = BufferedWriteRegistry.get_instance()
+
+# Check pending writes (for monitoring)
+pending_count = registry.get_pending_count()
+pending_files = registry.get_pending_paths()
+
+# Manual flush (optional - atexit handles this automatically)
+results = registry.flush_all()  # Returns {path: success_bool}
+
+# Manual shutdown (if needed before atexit)
+results = registry.shutdown()
+```
+
+#### `ResilientStateWriter`
+
+For stateful files that must persist (usage stats):
+- **Memory-first**: Always updates in-memory state before attempting disk write
+- **Atomic writes**: Uses tempfile + move pattern to prevent corruption
+- **Automatic retry with backoff**: If disk fails, waits `retry_interval` seconds before trying again
+- **Shutdown integration**: Registers with `BufferedWriteRegistry` on failure for final flush
+- **Health monitoring**: Exposes `is_healthy` property for monitoring
+
+```python
+writer = ResilientStateWriter("data.json", logger, retry_interval=30.0)
+writer.write({"key": "value"})  # Always succeeds (memory update)
+if not writer.is_healthy:
+    logger.warning("Disk writes failing, data in memory only")
+# On next write() call after retry_interval, disk write is attempted again
+# On app exit (Ctrl+C), BufferedWriteRegistry attempts final save
+```
+
+#### `safe_write_json()`
+
+For JSON writes with configurable options (credentials, cache):
+
+| Parameter | Default | Description |
+|-----------|---------|-------------|
+| `path` | required | File path to write to |
+| `data` | required | JSON-serializable data |
+| `logger` | required | Logger for warnings |
+| `atomic` | `True` | Use atomic write pattern (tempfile + move) |
+| `indent` | `2` | JSON indentation level |
+| `ensure_ascii` | `True` | Escape non-ASCII characters |
+| `secure_permissions` | `False` | Set file permissions to 0o600 |
+| `buffer_on_failure` | `False` | Register with BufferedWriteRegistry on failure |
+
+When `buffer_on_failure=True`:
+- Failed writes are registered with `BufferedWriteRegistry`
+- Data is retried every 30 seconds in background
+- On app exit, final write attempt is made automatically
+- Success unregisters the pending write
+
+```python
+# For critical data (auth tokens) - use buffer_on_failure
+safe_write_json(path, creds, logger, secure_permissions=True, buffer_on_failure=True)
+
+# For non-critical data (logs) - no buffering needed
+safe_write_json(path, data, logger)
+```
+
+#### `safe_log_write()`
+
+For log files where occasional loss is acceptable:
+- Fire-and-forget pattern
+- Creates parent directories if needed
+- Returns `True`/`False`, never raises
+- **No buffering** - logs are dropped on failure
+
+#### `safe_mkdir()`
+
+For directory creation with error handling.
+
+### 5.2. 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.
+1. **Core API Handling (Level 1)**: The Python runtime keeps all necessary code in memory. Deleting source code files while the proxy is running will **not** crash active requests.
+
+2. **Credential Management (Level 2)**: OAuth tokens are cached in memory first. If credential files are deleted, the proxy continues using cached tokens. If a token refresh succeeds but the file cannot be written, the new token is buffered for retry and saved on shutdown.
+
+3. **Usage Tracking (Level 3)**: Usage statistics (`key_usage.json`) are maintained in memory via `ResilientStateWriter`. If the file is deleted, the system tracks usage internally and attempts to recreate the file on the next save interval. Pending writes are flushed on shutdown.
+
+4. **Provider Cache (Level 4)**: The provider cache tracks disk health and continues operating in memory-only mode if disk writes fail. Has its own shutdown mechanism.
+
+5. **Logging (Level 5)**: Logging is treated as non-critical. If the `logs/` directory is removed, the system attempts to recreate it. If creation fails, logging degrades gracefully without interrupting the request flow. **No buffering or retry**.
+
+### 5.3. Component Integration
 
-### 5.2. "Develop While Running"
+| Component | Utility Used | Behavior on Disk Failure | Shutdown Flush |
+|-----------|--------------|--------------------------|----------------|
+| `UsageManager` | `ResilientStateWriter` | Continues in memory, retries after 30s | Yes (via registry) |
+| `GoogleOAuthBase` | `safe_write_json(buffer_on_failure=True)` | Memory cache preserved, buffered for retry | Yes (via registry) |
+| `QwenAuthBase` | `safe_write_json(buffer_on_failure=True)` | Memory cache preserved, buffered for retry | Yes (via registry) |
+| `IFlowAuthBase` | `safe_write_json(buffer_on_failure=True)` | Memory cache preserved, buffered for retry | Yes (via registry) |
+| `ProviderCache` | `safe_write_json` + own shutdown | Retries via own background loop | Yes (own mechanism) |
+| `DetailedLogger` | `safe_write_json` | Logs dropped, no crash | No |
+| `failure_logger` | Python `logging.RotatingFileHandler` | Falls back to NullHandler | No |
+
+### 5.4. Shutdown Behavior
+
+When the application exits (including Ctrl+C):
+
+1. **atexit handler fires**: `BufferedWriteRegistry._atexit_handler()` is called
+2. **Pending writes counted**: Registry checks how many files have pending writes
+3. **Flush attempted**: Each pending file gets a final write attempt
+4. **Results logged**:
+   - Success: `"Shutdown flush: all N write(s) succeeded"`
+   - Partial: `"Shutdown flush: X succeeded, Y failed"` with failed file names
+
+**Console output example:**
+```
+INFO:rotator_library.resilient_io:Flushing 2 pending write(s) on shutdown...
+INFO:rotator_library.resilient_io:Shutdown flush: all 2 write(s) succeeded
+```
+
+### 5.5. "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.
+- **Log Cleanup**: You can safely run `rm -rf logs/` while the proxy is serving traffic. The system will 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 for load balancing consistency.
+- **File Recovery**: If you delete a critical file, the system attempts directory auto-recreation before every write operation.
+- **Safe Exit**: Ctrl+C triggers graceful shutdown with final data flush attempt.
 
-### 5.3. Graceful Degradation & Data Loss
+### 5.6. 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.
+- **Logs**: If disk writes fail, detailed request logs may be lost (no buffering).
+- **Usage Stats**: Buffered in memory and flushed on shutdown. Data loss only if shutdown flush also fails.
+- **Credentials**: Buffered in memory and flushed on shutdown. Re-authentication only needed if shutdown flush fails.
+- **Cache**: Provider cache entries may need to be regenerated after restart if its own shutdown mechanism fails.
+
+### 5.7. Monitoring Disk Health
+
+Components expose health information for monitoring:
 
+```python
+# BufferedWriteRegistry
+registry = BufferedWriteRegistry.get_instance()
+pending = registry.get_pending_count()  # Number of files with pending writes
+files = registry.get_pending_paths()    # List of pending file names
+
+# UsageManager
+writer = usage_manager._state_writer
+health = writer.get_health_info()
+# Returns: {"healthy": True, "failure_count": 0, "last_success": 1234567890.0, ...}
+
+# ProviderCache
+stats = cache.get_stats()
+# Includes: {"disk_available": True, "disk_errors": 0, ...}
+```
 

src/rotator_library/providers/antigravity_provider.py

@@ -2424,7 +2424,7 @@ class AntigravityProvider(AntigravityAuthBase, ProviderInterface):
                 elif first_func_in_msg:
                     # Only add bypass to the first function call if no sig available
                     func_part["thoughtSignature"] = "skip_thought_signature_validator"
-                    lib_logger.warning(
+                    lib_logger.debug(
                         f"Missing thoughtSignature for first func call {tool_id}, using bypass"
                     )
                 # Subsequent parallel calls: no signature field at all

src/rotator_library/providers/gemini_cli_provider.py

@@ -1166,7 +1166,7 @@ class GeminiCliProvider(GeminiAuthBase, ProviderInterface):
                                     func_part["thoughtSignature"] = (
                                         "skip_thought_signature_validator"
                                     )
-                                    lib_logger.warning(
+                                    lib_logger.debug(
                                         f"Missing thoughtSignature for first func call {tool_id}, using bypass"
                                     )
                                 # Subsequent parallel calls: no signature field at all

src/rotator_library/providers/google_oauth_base.py

@@ -273,13 +273,16 @@ class GoogleOAuthBase:
             return
 
         # Attempt disk write - if it fails, we still have the cache
-        if safe_write_json(path, creds, lib_logger, secure_permissions=True):
+        # buffer_on_failure ensures data is retried periodically and saved on shutdown
+        if safe_write_json(
+            path, creds, lib_logger, secure_permissions=True, buffer_on_failure=True
+        ):
             lib_logger.debug(
                 f"Saved updated {self.ENV_PREFIX} OAuth credentials to '{path}'."
             )
         else:
             lib_logger.warning(
-                f"Credentials for {self.ENV_PREFIX} cached in memory only (will be lost on restart)."
+                f"Credentials for {self.ENV_PREFIX} cached in memory only (buffered for retry)."
             )
 
     def _is_token_expired(self, creds: Dict[str, Any]) -> bool:

src/rotator_library/providers/iflow_auth_base.py

@@ -325,11 +325,14 @@ class IFlowAuthBase:
             return
 
         # Attempt disk write - if it fails, we still have the cache
-        if safe_write_json(path, creds, lib_logger, secure_permissions=True):
+        # buffer_on_failure ensures data is retried periodically and saved on shutdown
+        if safe_write_json(
+            path, creds, lib_logger, secure_permissions=True, buffer_on_failure=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)."
+                "iFlow credentials cached in memory only (buffered for retry)."
             )
 
     def _is_token_expired(self, creds: Dict[str, Any]) -> bool:

src/rotator_library/providers/provider_cache.py

@@ -197,10 +197,14 @@ class ProviderCache:
     # DISK PERSISTENCE
     # =========================================================================
 
-    async def _save_to_disk(self) -> None:
-        """Persist cache to disk using atomic write with health tracking."""
+    async def _save_to_disk(self) -> bool:
+        """Persist cache to disk using atomic write with health tracking.
+
+        Returns:
+            True if write succeeded, False otherwise.
+        """
         if not self._enable_disk:
-            return
+            return True  # Not an error if disk is disabled
 
         async with self._disk_lock:
             cache_data = {
@@ -226,9 +230,11 @@ class ProviderCache:
                 lib_logger.debug(
                     f"ProviderCache[{self._cache_name}]: Saved {len(self._cache)} entries"
                 )
+                return True
             else:
                 self._stats["disk_errors"] += 1
                 self._disk_available = False
+                return False
 
     # =========================================================================
     # BACKGROUND TASKS
@@ -251,8 +257,10 @@ class ProviderCache:
                 await asyncio.sleep(self._write_interval)
                 if self._dirty:
                     try:
-                        await self._save_to_disk()
-                        self._dirty = False
+                        success = await self._save_to_disk()
+                        if success:
+                            self._dirty = False
+                        # If save failed, _dirty remains True so we retry next interval
                     except Exception as e:
                         lib_logger.error(
                             f"ProviderCache[{self._cache_name}]: Writer error: {e}"

src/rotator_library/providers/qwen_auth_base.py

@@ -210,11 +210,14 @@ class QwenAuthBase:
             return
 
         # Attempt disk write - if it fails, we still have the cache
-        if safe_write_json(path, creds, lib_logger, secure_permissions=True):
+        # buffer_on_failure ensures data is retried periodically and saved on shutdown
+        if safe_write_json(
+            path, creds, lib_logger, secure_permissions=True, buffer_on_failure=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)."
+                "Qwen credentials cached in memory only (buffered for retry)."
             )
 
     def _is_token_expired(self, creds: Dict[str, Any]) -> bool:

src/rotator_library/utils/__init__.py

@@ -3,6 +3,7 @@
 from .headless_detection import is_headless_environment
 from .reauth_coordinator import get_reauth_coordinator, ReauthCoordinator
 from .resilient_io import (
+    BufferedWriteRegistry,
     ResilientStateWriter,
     safe_write_json,
     safe_log_write,
@@ -13,6 +14,7 @@ __all__ = [
     "is_headless_environment",
     "get_reauth_coordinator",
     "ReauthCoordinator",
+    "BufferedWriteRegistry",
     "ResilientStateWriter",
     "safe_write_json",
     "safe_log_write",

src/rotator_library/utils/resilient_io.py

@@ -2,12 +2,17 @@
 """
 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.
+Provides three main patterns:
+1. BufferedWriteRegistry - Global singleton for buffered writes with periodic
+   retry and shutdown flush. Ensures data is saved on app exit (Ctrl+C).
+2. ResilientStateWriter - For stateful files (usage.json) that should be
+   buffered in memory and retried on disk failure.
+3. safe_write_json (with buffer_on_failure) - For critical files (auth tokens)
+   that should be buffered and retried if write fails.
+4. safe_log_write - For logs that can be dropped on failure.
 """
 
+import atexit
 import json
 import os
 import shutil
@@ -16,7 +21,284 @@ import threading
 import time
 import logging
 from pathlib import Path
-from typing import Any, Callable, Dict, Optional, Union
+from typing import Any, Callable, Dict, Optional, Tuple, Union
+
+
+# =============================================================================
+# BUFFERED WRITE REGISTRY (SINGLETON)
+# =============================================================================
+
+
+class BufferedWriteRegistry:
+    """
+    Global singleton registry for buffered writes with periodic retry and shutdown flush.
+
+    This ensures that critical data (auth tokens, usage stats) is saved even if
+    disk writes fail temporarily. On app exit (including Ctrl+C), all pending
+    writes are flushed.
+
+    Features:
+    - Per-file buffering: each file path has its own pending write
+    - Periodic retries: background thread retries failed writes every N seconds
+    - Shutdown flush: atexit hook ensures final write attempt on app exit
+    - Thread-safe: safe for concurrent access from multiple threads
+
+    Usage:
+        # Get the singleton instance
+        registry = BufferedWriteRegistry.get_instance()
+
+        # Register a pending write (usually called by safe_write_json on failure)
+        registry.register_pending(path, data, serializer_fn, options)
+
+        # Manual flush (optional - atexit handles this automatically)
+        results = registry.flush_all()
+    """
+
+    _instance: Optional["BufferedWriteRegistry"] = None
+    _instance_lock = threading.Lock()
+
+    def __init__(self, retry_interval: float = 30.0):
+        """
+        Initialize the registry. Use get_instance() instead of direct construction.
+
+        Args:
+            retry_interval: Seconds between retry attempts (default: 30)
+        """
+        self._pending: Dict[str, Tuple[Any, Callable[[Any], str], Dict[str, Any]]] = {}
+        self._retry_interval = retry_interval
+        self._lock = threading.Lock()
+        self._running = False
+        self._retry_thread: Optional[threading.Thread] = None
+        self._logger = logging.getLogger("rotator_library.resilient_io")
+
+        # Start background retry thread
+        self._start_retry_thread()
+
+        # Register atexit handler for shutdown flush
+        atexit.register(self._atexit_handler)
+
+    @classmethod
+    def get_instance(cls, retry_interval: float = 30.0) -> "BufferedWriteRegistry":
+        """
+        Get or create the singleton instance.
+
+        Args:
+            retry_interval: Seconds between retry attempts (only used on first call)
+
+        Returns:
+            The singleton BufferedWriteRegistry instance
+        """
+        if cls._instance is None:
+            with cls._instance_lock:
+                if cls._instance is None:
+                    cls._instance = cls(retry_interval)
+        return cls._instance
+
+    def _start_retry_thread(self) -> None:
+        """Start the background retry thread."""
+        if self._running:
+            return
+
+        self._running = True
+        self._retry_thread = threading.Thread(
+            target=self._retry_loop,
+            name="BufferedWriteRegistry-Retry",
+            daemon=True,  # Daemon so it doesn't block app exit
+        )
+        self._retry_thread.start()
+
+    def _retry_loop(self) -> None:
+        """Background thread: periodically retry pending writes."""
+        while self._running:
+            time.sleep(self._retry_interval)
+            if not self._running:
+                break
+            self._retry_pending()
+
+    def _retry_pending(self) -> None:
+        """Attempt to write all pending files."""
+        with self._lock:
+            if not self._pending:
+                return
+
+            # Copy paths to avoid modifying dict during iteration
+            paths = list(self._pending.keys())
+
+        for path_str in paths:
+            self._try_write(path_str, remove_on_success=True)
+
+    def register_pending(
+        self,
+        path: Union[str, Path],
+        data: Any,
+        serializer: Callable[[Any], str],
+        options: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        """
+        Register a pending write for later retry.
+
+        If a write is already pending for this path, it is replaced with the new data
+        (we always want to write the latest state).
+
+        Args:
+            path: File path to write to
+            data: Data to serialize and write
+            serializer: Function to serialize data to string
+            options: Additional options (e.g., secure_permissions)
+        """
+        path_str = str(Path(path).resolve())
+        with self._lock:
+            self._pending[path_str] = (data, serializer, options or {})
+            self._logger.debug(f"Registered pending write for {Path(path).name}")
+
+    def unregister(self, path: Union[str, Path]) -> None:
+        """
+        Remove a pending write (called when write succeeds elsewhere).
+
+        Args:
+            path: File path to remove from pending
+        """
+        path_str = str(Path(path).resolve())
+        with self._lock:
+            self._pending.pop(path_str, None)
+
+    def _try_write(self, path_str: str, remove_on_success: bool = True) -> bool:
+        """
+        Attempt to write a pending file.
+
+        Args:
+            path_str: Resolved path string
+            remove_on_success: Remove from pending if successful
+
+        Returns:
+            True if write succeeded, False otherwise
+        """
+        with self._lock:
+            if path_str not in self._pending:
+                return True
+            data, serializer, options = self._pending[path_str]
+
+        path = Path(path_str)
+        try:
+            # Ensure directory exists
+            path.parent.mkdir(parents=True, exist_ok=True)
+
+            # Serialize data
+            content = serializer(data)
+
+            # Atomic write
+            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
+                if options.get("secure_permissions"):
+                    try:
+                        os.chmod(tmp_path, 0o600)
+                    except (OSError, AttributeError):
+                        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
+
+            # Success - remove from pending
+            if remove_on_success:
+                with self._lock:
+                    self._pending.pop(path_str, None)
+
+            self._logger.debug(f"Retry succeeded for {path.name}")
+            return True
+
+        except (OSError, PermissionError, IOError) as e:
+            self._logger.debug(f"Retry failed for {path.name}: {e}")
+            return False
+
+    def flush_all(self) -> Dict[str, bool]:
+        """
+        Attempt to write all pending files immediately.
+
+        Returns:
+            Dict mapping file paths to success status
+        """
+        with self._lock:
+            paths = list(self._pending.keys())
+
+        results = {}
+        for path_str in paths:
+            results[path_str] = self._try_write(path_str, remove_on_success=True)
+
+        return results
+
+    def _atexit_handler(self) -> None:
+        """Called on app exit to flush pending writes."""
+        self._running = False
+
+        with self._lock:
+            pending_count = len(self._pending)
+
+        if pending_count == 0:
+            return
+
+        self._logger.info(f"Flushing {pending_count} pending write(s) on shutdown...")
+        results = self.flush_all()
+
+        succeeded = sum(1 for v in results.values() if v)
+        failed = pending_count - succeeded
+
+        if failed > 0:
+            self._logger.warning(
+                f"Shutdown flush: {succeeded} succeeded, {failed} failed"
+            )
+            for path_str, success in results.items():
+                if not success:
+                    self._logger.warning(f"  Failed to save: {Path(path_str).name}")
+        else:
+            self._logger.info(f"Shutdown flush: all {succeeded} write(s) succeeded")
+
+    def get_pending_count(self) -> int:
+        """Get the number of pending writes."""
+        with self._lock:
+            return len(self._pending)
+
+    def get_pending_paths(self) -> list:
+        """Get list of paths with pending writes (for monitoring)."""
+        with self._lock:
+            return [Path(p).name for p in self._pending.keys()]
+
+    def shutdown(self) -> Dict[str, bool]:
+        """
+        Manually trigger shutdown: stop retry thread and flush all pending writes.
+
+        Returns:
+            Dict mapping file paths to success status
+        """
+        self._running = False
+        if self._retry_thread and self._retry_thread.is_alive():
+            self._retry_thread.join(timeout=1.0)
+        return self.flush_all()
+
+
+# =============================================================================
+# RESILIENT STATE WRITER
+# =============================================================================
 
 
 class ResilientStateWriter:
@@ -72,7 +354,8 @@ class ResilientStateWriter:
         Update state and attempt disk write.
 
         Always updates in-memory state (guaranteed to succeed).
-        Attempts disk write - if it fails, schedules for retry.
+        Attempts disk write - if disk is unhealthy, respects retry_interval
+        before attempting again to avoid flooding with failed writes.
 
         Args:
             data: Data to persist (must be serializable)
@@ -82,6 +365,14 @@ class ResilientStateWriter:
         """
         with self._lock:
             self._current_state = data
+
+            # If disk is unhealthy, only retry after retry_interval has passed
+            if not self._disk_healthy:
+                now = time.time()
+                if now - self._last_attempt < self.retry_interval:
+                    # Too soon to retry, data is safe in memory
+                    return False
+
             return self._try_disk_write()
 
     def retry_if_needed(self) -> bool:
@@ -113,6 +404,8 @@ class ResilientStateWriter:
 
         Uses tempfile + move pattern for atomic writes on POSIX systems.
         On Windows, uses direct write (still safe for our use case).
+
+        Also registers/unregisters with BufferedWriteRegistry for shutdown flush.
         """
         if self._current_state is None:
             return True
@@ -155,16 +448,26 @@ class ResilientStateWriter:
                     except OSError:
                         pass
 
-            # Success - update health
+            # Success - update health and unregister from shutdown flush
             self._disk_healthy = True
             self._last_success = time.time()
             self._failure_count = 0
+            BufferedWriteRegistry.get_instance().unregister(self.path)
             return True
 
         except (OSError, PermissionError, IOError) as e:
             self._disk_healthy = False
             self._failure_count += 1
 
+            # Register with BufferedWriteRegistry for shutdown flush
+            registry = BufferedWriteRegistry.get_instance()
+            registry.register_pending(
+                self.path,
+                self._current_state,
+                self._serializer,
+                {},  # No special options for ResilientStateWriter
+            )
+
             # Log warning (rate-limited to avoid flooding)
             if self._failure_count == 1 or self._failure_count % 10 == 0:
                 self.logger.warning(
@@ -211,12 +514,14 @@ def safe_write_json(
     indent: int = 2,
     ensure_ascii: bool = True,
     secure_permissions: bool = False,
+    buffer_on_failure: bool = False,
 ) -> bool:
     """
-    Write JSON data to file with error handling. No buffering or retry.
+    Write JSON data to file with error handling and optional buffering.
 
-    Suitable for one-off writes where failure is acceptable (e.g., logs).
-    Creates parent directories if needed.
+    When buffer_on_failure is True, failed writes are registered with the
+    BufferedWriteRegistry for periodic retry and shutdown flush. This ensures
+    critical data (like auth tokens) is eventually saved.
 
     Args:
         path: File path to write to
@@ -226,15 +531,20 @@ def safe_write_json(
         indent: JSON indentation level (default: 2)
         ensure_ascii: Escape non-ASCII characters (default: True)
         secure_permissions: Set file permissions to 0o600 (default: False)
+        buffer_on_failure: Register with BufferedWriteRegistry on failure (default: False)
 
     Returns:
         True on success, False on failure (never raises)
     """
     path = Path(path)
 
+    # Create serializer function that matches the requested formatting
+    def serializer(d: Any) -> str:
+        return json.dumps(d, indent=indent, ensure_ascii=ensure_ascii)
+
     try:
         path.parent.mkdir(parents=True, exist_ok=True)
-        content = json.dumps(data, indent=indent, ensure_ascii=ensure_ascii)
+        content = serializer(data)
 
         if atomic:
             tmp_fd = None
@@ -279,10 +589,26 @@ def safe_write_json(
                 except (OSError, AttributeError):
                     pass
 
+        # Success - remove from pending if it was there
+        if buffer_on_failure:
+            BufferedWriteRegistry.get_instance().unregister(path)
+
         return True
 
     except (OSError, PermissionError, IOError, TypeError, ValueError) as e:
         logger.warning(f"Failed to write JSON to {path}: {e}")
+
+        # Register for retry if buffering is enabled
+        if buffer_on_failure:
+            registry = BufferedWriteRegistry.get_instance()
+            registry.register_pending(
+                path,
+                data,
+                serializer,
+                {"secure_permissions": secure_permissions},
+            )
+            logger.debug(f"Buffered {path.name} for retry on next interval or shutdown")
+
         return False