# Performance Monitoring This document provides standardized performance monitoring utilities for tracking agent efficiency, resource usage, and system health. ## Core Metrics ```python from datetime import datetime, timedelta import time import json from typing import Dict, List, Any, Optional from collections import defaultdict class PerformanceMonitor: """Track and report performance metrics for agents.""" def __init__(self, agent_name: str, session_id: str): self.agent_name = agent_name self.session_id = session_id self.metrics = defaultdict(dict) self.operation_stack = [] def start_operation(self, operation_name: str, metadata: Optional[Dict] = None): """Start timing an operation.""" operation_id = f"{operation_name}_{int(time.time() * 1000)}" op_data = { 'id': operation_id, 'name': operation_name, 'start_time': time.time(), 'metadata': metadata or {} } self.operation_stack.append(op_data) self.metrics[operation_name][operation_id] = op_data return operation_id def end_operation(self, operation_id: Optional[str] = None, status: str = 'success', result_metadata: Optional[Dict] = None): """End timing and record results.""" # If no ID provided, use most recent operation if not operation_id and self.operation_stack: operation_data = self.operation_stack.pop() operation_id = operation_data['id'] else: # Find and remove from stack operation_data = None for i, op in enumerate(self.operation_stack): if op['id'] == operation_id: operation_data = self.operation_stack.pop(i) break if not operation_data: # Try to find in metrics for op_name, ops in self.metrics.items(): if operation_id in ops: operation_data = ops[operation_id] break if not operation_data: print(f"[WARNING] Unknown operation: {operation_id}") return # Calculate duration end_time = time.time() duration = end_time - operation_data['start_time'] # Update operation data operation_data.update({ 'end_time': end_time, 'duration_seconds': duration, 'status': status, 'result_metadata': result_metadata or {} }) return operation_data def get_operation_stats(self, operation_name: str) -> Dict[str, Any]: """Get statistics for a specific operation type.""" operations = self.metrics.get(operation_name, {}) if not operations: return {} completed = [op for op in operations.values() if 'duration_seconds' in op] if not completed: return {'count': len(operations), 'completed': 0} durations = [op['duration_seconds'] for op in completed] statuses = [op.get('status', 'unknown') for op in completed] return { 'count': len(operations), 'completed': len(completed), 'success_rate': statuses.count('success') / len(statuses), 'avg_duration': sum(durations) / len(durations), 'min_duration': min(durations), 'max_duration': max(durations), 'total_duration': sum(durations), 'status_breakdown': { status: statuses.count(status) for status in set(statuses) } } def get_summary(self) -> Dict[str, Any]: """Get overall performance summary.""" summary = { 'agent': self.agent_name, 'session': self.session_id, 'timestamp': datetime.utcnow().isoformat() + 'Z', 'operations': {} } for op_name in self.metrics: summary['operations'][op_name] = self.get_operation_stats(op_name) # Calculate totals total_completed = sum( stats.get('completed', 0) for stats in summary['operations'].values() ) total_duration = sum( stats.get('total_duration', 0) for stats in summary['operations'].values() ) summary['totals'] = { 'operations_completed': total_completed, 'total_duration_seconds': total_duration, 'operations_in_progress': len(self.operation_stack) } return summary ``` ## Resource Usage Tracking ```python import psutil import os class ResourceMonitor: """Monitor system resource usage.""" def __init__(self): self.process = psutil.Process(os.getpid()) self.start_time = time.time() self.snapshots = [] def take_snapshot(self, label: str = ""): """Capture current resource usage.""" try: cpu_percent = self.process.cpu_percent(interval=0.1) memory_info = self.process.memory_info() io_counters = self.process.io_counters() if hasattr(self.process, 'io_counters') else None snapshot = { 'timestamp': time.time(), 'label': label, 'cpu_percent': cpu_percent, 'memory_mb': memory_info.rss / 1024 / 1024, 'memory_percent': self.process.memory_percent(), } if io_counters: snapshot.update({ 'disk_read_mb': io_counters.read_bytes / 1024 / 1024, 'disk_write_mb': io_counters.write_bytes / 1024 / 1024 }) # System-wide metrics snapshot['system'] = { 'cpu_percent': psutil.cpu_percent(interval=0.1), 'memory_percent': psutil.virtual_memory().percent, 'disk_usage_percent': psutil.disk_usage('/').percent } self.snapshots.append(snapshot) return snapshot except Exception as e: print(f"[WARNING] Resource snapshot failed: {e}") return None def get_summary(self) -> Dict[str, Any]: """Get resource usage summary.""" if not self.snapshots: return {} # Calculate averages and peaks cpu_values = [s['cpu_percent'] for s in self.snapshots] memory_values = [s['memory_mb'] for s in self.snapshots] summary = { 'duration_seconds': time.time() - self.start_time, 'snapshots': len(self.snapshots), 'process': { 'cpu_avg_percent': sum(cpu_values) / len(cpu_values), 'cpu_peak_percent': max(cpu_values), 'memory_avg_mb': sum(memory_values) / len(memory_values), 'memory_peak_mb': max(memory_values), } } # Add I/O stats if available if 'disk_read_mb' in self.snapshots[-1]: summary['io'] = { 'total_read_mb': self.snapshots[-1]['disk_read_mb'], 'total_write_mb': self.snapshots[-1]['disk_write_mb'] } return summary ``` ## Integration with Agents ```python from shared.performance_monitoring import PerformanceMonitor, ResourceMonitor from shared.logging import AgentLogger class MonitoredAgent: """Base class for agents with integrated monitoring.""" def __init__(self, agent_name: str, session_id: str): self.logger = AgentLogger(agent_name, session_id) self.perf_monitor = PerformanceMonitor(agent_name, session_id) self.resource_monitor = ResourceMonitor() # Take initial resource snapshot self.resource_monitor.take_snapshot("startup") def monitored_operation(self, operation_name: str, operation_func, **kwargs): """Execute an operation with full monitoring.""" # Start performance tracking op_id = self.perf_monitor.start_operation(operation_name, kwargs) # Log operation start self.logger.info(f"Starting {operation_name}", action=operation_name) # Take resource snapshot before self.resource_monitor.take_snapshot(f"{operation_name}_start") try: # Execute the operation result = operation_func(**kwargs) # Success self.perf_monitor.end_operation(op_id, status='success') self.logger.info(f"Completed {operation_name}", action=operation_name) return result except Exception as e: # Failure self.perf_monitor.end_operation(op_id, status='failure', result_metadata={'error': str(e)}) self.logger.error(f"Failed {operation_name}: {e}", action=operation_name) raise finally: # Take resource snapshot after self.resource_monitor.take_snapshot(f"{operation_name}_end") def report_metrics(self, session_state_issue: int, owner: str, repo: str): """Report collected metrics to Forgejo.""" perf_summary = self.perf_monitor.get_summary() resource_summary = self.resource_monitor.get_summary() report = { 'performance': perf_summary, 'resources': resource_summary, 'timestamp': datetime.utcnow().isoformat() + 'Z' } comment = f"""[METRICS] {self.agent_name} Performance Summary: ```json {json.dumps(perf_summary, indent=2)} ``` Resource Usage: ```json {json.dumps(resource_summary, indent=2)} ``` --- **Automated by CleverAgents Bot** Agent: {self.agent_name}""" forgejo_create_issue_comment(owner, repo, session_state_issue, comment) ``` ## Usage Example ```python # Implementation worker with monitoring class MonitoredImplementationWorker(MonitoredAgent): def implement_issue(self, issue_number: int): """Implement issue with full monitoring.""" # Clone with monitoring def do_clone(): return git_clone(repo_url, target_dir) clone_dir = self.monitored_operation( "clone_repository", do_clone, repo_url=repo_url, target_dir=target_dir ) # Implement each subtask with monitoring for i, subtask in enumerate(subtasks): def do_subtask(): return implement_subtask(subtask, clone_dir) self.monitored_operation( f"implement_subtask_{i}", do_subtask, subtask_title=subtask['title'] ) # Report metrics at end self.report_metrics(session_state_issue, owner, repo) ``` ## Performance Optimization Patterns ### Batch Operations ```python def batch_monitored_operations(monitor: PerformanceMonitor, operation_name: str, items: List[Any], process_func, batch_size: int = 10): """Process items in batches with monitoring.""" total_op_id = monitor.start_operation(f"{operation_name}_batch", { 'total_items': len(items), 'batch_size': batch_size }) results = [] failures = [] for i in range(0, len(items), batch_size): batch = items[i:i + batch_size] batch_op_id = monitor.start_operation(f"{operation_name}_batch_{i//batch_size}", { 'batch_index': i // batch_size, 'batch_items': len(batch) }) try: batch_results = process_func(batch) results.extend(batch_results) monitor.end_operation(batch_op_id, 'success') except Exception as e: failures.extend(batch) monitor.end_operation(batch_op_id, 'failure', {'error': str(e)}) # End total operation monitor.end_operation(total_op_id, 'partial' if failures else 'success', { 'processed': len(results), 'failed': len(failures) }) return results, failures ``` ### Adaptive Timeouts ```python def adaptive_timeout(monitor: PerformanceMonitor, operation_name: str, default_timeout: int = 30): """Calculate adaptive timeout based on historical performance.""" stats = monitor.get_operation_stats(operation_name) if not stats or stats['completed'] < 3: # Not enough history return default_timeout # Use 99th percentile approximation (max of observed) # Add 50% buffer timeout = int(stats['max_duration'] * 1.5) # Bound between min and max return max(10, min(timeout, default_timeout * 3)) ``` ## Health Metrics ```python def calculate_health_score(perf_summary: Dict, resource_summary: Dict, failure_threshold: float = 0.2) -> float: """Calculate overall health score (0-1).""" scores = [] # Performance score based on success rates for op_name, stats in perf_summary.get('operations', {}).items(): if stats.get('completed', 0) > 0: success_rate = stats.get('success_rate', 0) scores.append(success_rate) # Resource score based on usage if resource_summary: cpu_score = 1 - min(resource_summary['process']['cpu_avg_percent'] / 100, 1) memory_score = 1 - min(resource_summary['process']['memory_avg_mb'] / 1000, 1) scores.extend([cpu_score, memory_score]) if not scores: return 1.0 # Weighted average health = sum(scores) / len(scores) # Penalize if too many operations in progress (might be stuck) in_progress = perf_summary.get('totals', {}).get('operations_in_progress', 0) if in_progress > 10: health *= 0.8 return health ``` ## Dashboard Metrics For system-wide monitoring, aggregate these metrics: ```python # Key Performance Indicators (KPIs) SYSTEM_KPIS = { # Throughput 'issues_per_hour': 'Count of issues completed per hour', 'prs_merged_per_hour': 'PRs successfully merged per hour', 'tests_written_per_hour': 'Test scenarios created per hour', # Quality 'first_attempt_success_rate': 'Work completed without retry', 'review_approval_rate': 'PRs approved on first review', 'test_coverage_trend': 'Coverage percentage over time', # Efficiency 'avg_issue_completion_time': 'Time from start to PR merge', 'avg_pr_fix_time': 'Time to fix failing PR', 'worker_utilization': 'Active workers / max workers', # Health 'system_health_score': 'Overall system health (0-1)', 'failure_rate_trend': 'Failure rate over time', 'resource_pressure': 'CPU/memory usage indicators' }