temp/.opencode/agents/shared/performance_monitoring.md

Performance Monitoring

This document provides standardized performance monitoring utilities for tracking agent efficiency, resource usage, and system health.

Core Metrics

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

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

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.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

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

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

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:

# 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'
}