#!/usr/bin/env python3 """ TITAN REFRESH DAEMON ==================== Background service that auto-refreshes Titan caches before expiry. VERIFICATION_STAMP Story: TITAN-003 (Auto-Refresh Daemon) Verified By: Claude Verified At: 2026-01-23 """ import os import sys import time import logging import threading from pathlib import Path from enum import Enum from typing import Optional, Callable from datetime import datetime, timedelta # Add genesis root sys.path.insert(0, str(Path(__file__).parent.parent.parent)) logger = logging.getLogger(__name__) class DaemonState(Enum): """Daemon state machine states.""" IDLE = "idle" RUNNING = "running" REFRESHING = "refreshing" ERROR = "error" STOPPED = "stopped" class TitanRefreshDaemon: """ Background daemon that automatically refreshes Titan caches. Features: - Monitors cache TTL - Refreshes 10 minutes before expiry - Retry logic on failures - State machine for tracking status - Callback hooks for events """ # Default refresh 10 minutes before expiry REFRESH_BUFFER_MINUTES = 10 def __init__( self, cache_manager, check_interval: int = 60, # Check every 60 seconds refresh_buffer: int = REFRESH_BUFFER_MINUTES, max_retries: int = 3, on_refresh_success: Optional[Callable] = None, on_refresh_failure: Optional[Callable] = None, ): """ Initialize the refresh daemon. Args: cache_manager: TitanCacheManager instance check_interval: Seconds between TTL checks refresh_buffer: Minutes before expiry to trigger refresh max_retries: Max refresh retry attempts on_refresh_success: Callback on successful refresh on_refresh_failure: Callback on failed refresh """ self.cache_manager = cache_manager self.check_interval = check_interval self.refresh_buffer = refresh_buffer self.max_retries = max_retries self._on_refresh_success = on_refresh_success self._on_refresh_failure_callback = on_refresh_failure self._state = DaemonState.IDLE self._thread: Optional[threading.Thread] = None self._stop_event = threading.Event() self._failure_count = 0 self._last_refresh: Optional[datetime] = None self._current_cache_name: Optional[str] = None @property def state(self) -> DaemonState: """Get current daemon state.""" return self._state def is_running(self) -> bool: """Check if daemon is running.""" return self._state == DaemonState.RUNNING or self._state == DaemonState.REFRESHING def start(self) -> bool: """ Start the refresh daemon. Returns: True if started successfully """ if self._state in (DaemonState.RUNNING, DaemonState.REFRESHING): logger.warning("Daemon already running") return False self._stop_event.clear() self._state = DaemonState.RUNNING self._failure_count = 0 self._thread = threading.Thread( target=self._run_loop, name="TitanRefreshDaemon", daemon=True, ) self._thread.start() logger.info("Titan Refresh Daemon started") return True def stop(self, timeout: float = 5.0) -> bool: """ Stop the refresh daemon. Args: timeout: Max seconds to wait for stop Returns: True if stopped successfully """ if self._state == DaemonState.STOPPED: return True logger.info("Stopping Titan Refresh Daemon...") self._stop_event.set() if self._thread and self._thread.is_alive(): self._thread.join(timeout=timeout) self._state = DaemonState.STOPPED logger.info("Titan Refresh Daemon stopped") return True def force_refresh(self) -> bool: """ Force an immediate cache refresh. Returns: True if refresh succeeded """ return self._do_refresh() def get_status(self) -> dict: """ Get daemon status information. Returns: Status dictionary """ return { 'state': self._state.value, 'failure_count': self._failure_count, 'last_refresh': self._last_refresh.isoformat() if self._last_refresh else None, 'current_cache': self._current_cache_name, 'check_interval': self.check_interval, 'refresh_buffer': self.refresh_buffer, } def _run_loop(self): """Main daemon loop.""" logger.info("Daemon loop started") while not self._stop_event.is_set(): try: # Check if refresh is needed if self._should_refresh(): self._state = DaemonState.REFRESHING success = self._do_refresh() if success: self._failure_count = 0 self._state = DaemonState.RUNNING else: self._failure_count += 1 if self._failure_count >= self.max_retries: self._state = DaemonState.ERROR logger.error(f"Max retries ({self.max_retries}) exceeded") else: self._state = DaemonState.RUNNING except Exception as e: logger.error(f"Daemon loop error: {e}") self._failure_count += 1 # Wait for next check self._stop_event.wait(timeout=self.check_interval) logger.info("Daemon loop ended") def _should_refresh(self) -> bool: """ Check if cache needs refresh. Returns: True if refresh is needed """ caches = self.cache_manager.list_caches() if not caches: # No caches - need to create one logger.info("No caches found, triggering creation") return True # Check if any cache is close to expiry for cache in caches: time_remaining = cache.time_until_expiry() minutes_remaining = time_remaining.total_seconds() / 60 if minutes_remaining <= self.refresh_buffer: logger.info(f"Cache {cache.display_name} expires in {minutes_remaining:.1f} min, triggering refresh") return True return False def _do_refresh(self) -> bool: """ Perform cache refresh. Returns: True if refresh succeeded """ logger.info("Starting cache refresh...") try: # Create new cache new_cache = self.cache_manager.create_full_stack_cache() if not new_cache: self._on_refresh_failure("Failed to create new cache") return False self._current_cache_name = new_cache.name self._last_refresh = datetime.now() # Delete old caches (keep only the new one) old_caches = self.cache_manager.list_caches() for old_cache in old_caches: if old_cache.name != new_cache.name: try: self.cache_manager.delete_cache(old_cache.name) logger.info(f"Deleted old cache: {old_cache.display_name}") except: pass logger.info(f"Refresh complete: {new_cache.display_name} ({new_cache.token_count:,} tokens)") if self._on_refresh_success: self._on_refresh_success(new_cache) return True except Exception as e: self._on_refresh_failure(str(e)) return False def _on_refresh_failure(self, error: str): """Handle refresh failure.""" self._failure_count += 1 logger.error(f"Refresh failed: {error}") if self._on_refresh_failure_callback: self._on_refresh_failure_callback(error) def _calculate_refresh_time(self, ttl_minutes: int) -> int: """ Calculate when to refresh based on TTL. Args: ttl_minutes: Cache TTL in minutes Returns: Minutes after creation to trigger refresh """ return ttl_minutes - self.refresh_buffer class TitanDaemonManager: """ Singleton manager for the Titan Refresh Daemon. Usage: manager = TitanDaemonManager.get_instance() manager.start() """ _instance: Optional['TitanDaemonManager'] = None _lock = threading.Lock() def __init__(self): from .cache_manager import TitanCacheManager self.cache_manager = TitanCacheManager() self.daemon = TitanRefreshDaemon( self.cache_manager, on_refresh_success=self._log_success, on_refresh_failure=self._log_failure, ) @classmethod def get_instance(cls) -> 'TitanDaemonManager': """Get singleton instance.""" if cls._instance is None: with cls._lock: if cls._instance is None: cls._instance = cls() return cls._instance def start(self) -> bool: """Start the daemon.""" return self.daemon.start() def stop(self) -> bool: """Stop the daemon.""" return self.daemon.stop() def status(self) -> dict: """Get daemon status.""" return self.daemon.get_status() def _log_success(self, cache): """Log successful refresh.""" logger.info(f"[TitanDaemon] Refresh SUCCESS: {cache.display_name}") def _log_failure(self, error): """Log failed refresh.""" logger.error(f"[TitanDaemon] Refresh FAILED: {error}") if __name__ == '__main__': # Quick test logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') print("=== Titan Refresh Daemon Test ===\n") manager = TitanDaemonManager.get_instance() print("Starting daemon...") manager.start() print(f"Status: {manager.status()}") print("\nRunning for 10 seconds...") time.sleep(10) print(f"Status: {manager.status()}") print("\nStopping daemon...") manager.stop() print(f"Final status: {manager.status()}")