""" Performance benchmark storage using Redis. Tracks operation timing, tool usage, and recommendation accuracy across sessions. Provides time-series data for performance analysis and optimization. """ import json from datetime import datetime, timezone from typing import Any, Literal, Optional import redis.asyncio as redis from pydantic import BaseModel, Field from .config import config from .logging_config import get_logger logger = get_logger(__name__) class PerformanceBenchmark(BaseModel): """ Performance benchmark record. Stores timing and metadata for operations like Steward analysis, tool calls, and agent execution. """ timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) operation: str # "steward_analysis", "tool_call", "tatlock_execution" duration_seconds: float success: bool # Steward-specific fields recommendation_count: Optional[int] = None confidence: Optional[float] = None # Tool-specific fields tool_name: Optional[str] = None was_recommended: Optional[bool] = None was_actually_used: Optional[bool] = None # Context conversation_id: Optional[str] = None metadata: dict[str, Any] = Field(default_factory=dict) def to_redis_dict(self) -> dict[str, Any]: """Convert to dict suitable for Redis storage.""" data = self.model_dump() data["timestamp"] = self.timestamp.isoformat() data["metadata"] = json.dumps(self.metadata) # Convert booleans to strings (Redis doesn't accept bool type) for key, value in data.items(): if isinstance(value, bool): data[key] = str(value) return data @classmethod def from_redis_dict(cls, data: dict[str, Any]) -> "PerformanceBenchmark": """Reconstruct from Redis dict.""" data["timestamp"] = datetime.fromisoformat(data["timestamp"]) data["metadata"] = json.loads(data.get("metadata", "{}")) # Convert string booleans back to bool for key in ["success", "was_recommended", "was_actually_used"]: if key in data and isinstance(data[key], str): data[key] = data[key] == "True" return cls(**data) class BenchmarkStore: """ Redis-backed benchmark storage with automatic expiry. Stores performance metrics in time-series format with 30-day retention. Provides querying capabilities for analysis and reporting. """ def __init__(self, redis_client: Optional[redis.Redis] = None): """ Initialize benchmark store. Args: redis_client: Optional Redis client. If None, creates from config. """ self._client = redis_client self._ttl_days = 30 # 30-day retention async def _get_client(self) -> redis.Redis: """Get or create Redis client.""" if self._client is None: self._client = redis.from_url( config.redis_url, encoding="utf-8", decode_responses=True, socket_timeout=config.REDIS_TIMEOUT, socket_connect_timeout=config.REDIS_TIMEOUT, ) return self._client async def record(self, benchmark: PerformanceBenchmark) -> None: """ Record a performance benchmark. Args: benchmark: Performance benchmark to record Example: >>> await store.record(PerformanceBenchmark( ... operation="steward_analysis", ... duration_seconds=1.23, ... success=True, ... recommendation_count=3, ... )) """ if not config.ENABLE_BENCHMARKS: return try: client = await self._get_client() # Generate key: benchmark:{operation}:{timestamp_ms} timestamp_ms = int(benchmark.timestamp.timestamp() * 1000) key = f"benchmark:{benchmark.operation}:{timestamp_ms}" # Store as hash await client.hset(key, mapping=benchmark.to_redis_dict()) # Set expiry await client.expire(key, self._ttl_days * 24 * 60 * 60) # Add to sorted set for time-based queries index_key = f"benchmark_index:{benchmark.operation}" await client.zadd(index_key, {key: timestamp_ms}) await client.expire(index_key, self._ttl_days * 24 * 60 * 60) logger.debug( "benchmark_recorded", operation=benchmark.operation, duration=benchmark.duration_seconds, success=benchmark.success, ) except Exception as e: logger.warning( "benchmark_recording_failed", error=str(e), operation=benchmark.operation, ) # Don't fail the request if benchmarking fails async def query( self, operation: str, start_time: Optional[datetime] = None, end_time: Optional[datetime] = None, limit: int = 100, ) -> list[PerformanceBenchmark]: """ Query benchmarks by operation and time range. Args: operation: Operation name to filter by start_time: Start of time range (inclusive) end_time: End of time range (inclusive) limit: Maximum number of results Returns: List of benchmarks matching the query Example: >>> from datetime import timedelta >>> now = datetime.now(timezone.utc) >>> yesterday = now - timedelta(days=1) >>> benchmarks = await store.query( ... "steward_analysis", ... start_time=yesterday, ... limit=50 ... ) """ if not config.ENABLE_BENCHMARKS: return [] try: client = await self._get_client() index_key = f"benchmark_index:{operation}" # Convert time range to timestamps min_score = ( int(start_time.timestamp() * 1000) if start_time else "-inf" ) max_score = ( int(end_time.timestamp() * 1000) if end_time else "+inf" ) # Query sorted set keys = await client.zrevrangebyscore( index_key, max_score, min_score, start=0, num=limit, ) # Fetch benchmark data benchmarks = [] for key in keys: data = await client.hgetall(key) if data: benchmarks.append(PerformanceBenchmark.from_redis_dict(data)) return benchmarks except Exception as e: logger.error( "benchmark_query_failed", error=str(e), operation=operation, ) return [] async def get_statistics( self, operation: str, start_time: Optional[datetime] = None, end_time: Optional[datetime] = None, ) -> dict[str, Any]: """ Get aggregate statistics for an operation. Args: operation: Operation name start_time: Start of time range end_time: End of time range Returns: Dictionary with statistics (count, avg_duration, success_rate, etc.) Example: >>> stats = await store.get_statistics("steward_analysis") >>> print(f"Average duration: {stats['avg_duration']}s") >>> print(f"Success rate: {stats['success_rate']}%") """ benchmarks = await self.query(operation, start_time, end_time, limit=1000) if not benchmarks: return { "count": 0, "avg_duration": 0.0, "min_duration": 0.0, "max_duration": 0.0, "success_rate": 0.0, } durations = [b.duration_seconds for b in benchmarks] successes = sum(1 for b in benchmarks if b.success) return { "count": len(benchmarks), "avg_duration": sum(durations) / len(durations), "min_duration": min(durations), "max_duration": max(durations), "success_rate": (successes / len(benchmarks)) * 100, "total_successes": successes, "total_failures": len(benchmarks) - successes, } async def get_tool_accuracy( self, start_time: Optional[datetime] = None, end_time: Optional[datetime] = None, ) -> dict[str, Any]: """ Analyze tool recommendation accuracy. Compares recommended tools vs actually used tools to measure Steward's recommendation precision. Args: start_time: Start of time range end_time: End of time range Returns: Dictionary with accuracy metrics Example: >>> accuracy = await store.get_tool_accuracy() >>> print(f"Precision: {accuracy['precision']}%") """ tool_calls = await self.query("tool_call", start_time, end_time, limit=1000) if not tool_calls: return { "total_calls": 0, "recommended_and_used": 0, "recommended_not_used": 0, "not_recommended_but_used": 0, "precision": 0.0, } recommended_and_used = sum( 1 for b in tool_calls if b.was_recommended and b.was_actually_used ) not_recommended_but_used = sum( 1 for b in tool_calls if not b.was_recommended and b.was_actually_used ) total_used = sum(1 for b in tool_calls if b.was_actually_used) precision = ( (recommended_and_used / total_used * 100) if total_used > 0 else 0.0 ) return { "total_calls": len(tool_calls), "total_used": total_used, "recommended_and_used": recommended_and_used, "not_recommended_but_used": not_recommended_but_used, "precision": precision, } async def close(self) -> None: """Close Redis connection.""" if self._client: await self._client.aclose() self._client = None # Global benchmark store instance _benchmark_store: Optional[BenchmarkStore] = None def get_benchmark_store() -> BenchmarkStore: """ Get global benchmark store instance. Returns: BenchmarkStore instance """ global _benchmark_store if _benchmark_store is None: _benchmark_store = BenchmarkStore() return _benchmark_store