"""
Status Aggregation Utilities
Provides comprehensive status aggregation capabilities for session monitoring
with efficient collection, analysis, and reporting of agent status information.
Architecture Integration:
- Design Patterns: Aggregator pattern for status collection, Strategy for health evaluation
- Performance Profile: O(n) agent iteration with optimized metric collection
- Data Processing: Real-time aggregation with caching for frequently accessed metrics
Technical Decisions:
- Parallel Collection: Concurrent agent status retrieval for performance
- Health Scoring: Multi-factor health evaluation with configurable thresholds
- Graceful Degradation: Partial results when some agents unreachable
Dependencies & Integration:
- Internal: Agent management, resource monitoring, health checking
- Types: Agent status, resource metrics, health indicators
Quality Assurance:
- Test Coverage: Unit tests for all aggregation scenarios
- Performance Testing: Scalability testing with multiple agents
- Error Resilience: Comprehensive error handling for offline agents
Author: ADDER_4 | Created: 2025-06-26 | Last Modified: 2025-06-26
"""
import asyncio
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from typing import Any, Dict, List, Optional, Tuple
from src.models.agent import AgentState, AgentStatus
# Import type system
from src.models.ids import AgentId, SessionId
from src.models.monitoring import HealthStatus, ResourceMetrics
from src.models.security import SecurityContext
# Import utilities
from src.utils.errors import OperationError
from .contracts_shim import ensure, require
class SessionHealth(Enum):
"""Session health status levels."""
HEALTHY = "healthy"
DEGRADED = "degraded"
CRITICAL = "critical"
FAILED = "failed"
EMPTY = "empty"
class AggregationStrategy(Enum):
"""Status aggregation strategies."""
FAST = "fast" # Basic status only
DETAILED = "detailed" # Include metrics
COMPREHENSIVE = "comprehensive" # Full monitoring
@dataclass
class AgentStatusInfo:
"""Comprehensive agent status information."""
agent_id: AgentId
agent_name: str
status: AgentStatus
session_id: SessionId
specialization: Optional[str] = None
created_at: Optional[datetime] = None
last_activity: Optional[datetime] = None
# Process information
process_id: Optional[int] = None
iterm_tab_id: Optional[str] = None
# Resource metrics
cpu_percent: Optional[float] = None
memory_mb: Optional[int] = None
file_descriptors: Optional[int] = None
thread_count: Optional[int] = None
# Health information
health_status: Optional[HealthStatus] = None
last_health_check: Optional[datetime] = None
health_issues: List[str] = field(default_factory=list)
# Error information
error_message: Optional[str] = None
is_accessible: bool = True
@dataclass
class SessionMetrics:
"""Aggregated session-level metrics."""
total_agents: int = 0
active_agents: int = 0
degraded_agents: int = 0
failed_agents: int = 0
offline_agents: int = 0
# Resource totals
total_cpu_percent: float = 0.0
total_memory_mb: int = 0
total_file_descriptors: int = 0
total_threads: int = 0
# Performance indicators
avg_response_time_ms: float = 0.0
max_memory_usage_mb: int = 0
resource_utilization_percent: float = 0.0
# Health indicators
health_score: float = 0.0
critical_issues: int = 0
warnings: int = 0
@dataclass
class SessionStatusResult:
"""Complete session status aggregation result."""
session_id: SessionId
session_name: str
session_health: SessionHealth
metrics: SessionMetrics
agents: List[AgentStatusInfo]
collection_timestamp: datetime
collection_duration_ms: float
partial_results: bool = False
errors: List[str] = field(default_factory=list)
warnings: List[str] = field(default_factory=list)
class SessionStatusAggregator:
"""
Comprehensive session status aggregation with performance optimization.
Aggregates status information from multiple agents within a session,
providing real-time metrics, health evaluation, and error resilience.
Contracts:
Preconditions:
- Agent manager and monitoring systems must be available
- Session ID must be valid
Postconditions:
- Complete status aggregation or partial results with errors
- All accessible agents included in results
- Performance metrics captured
Invariants:
- Agent count matches aggregated metrics
- Health scores are consistent with agent states
- Timestamps are accurate and ordered
"""
def __init__(self):
"""Initialize status aggregator."""
self._collection_stats = {
"total_collections": 0,
"successful_collections": 0,
"partial_collections": 0,
"failed_collections": 0,
"avg_collection_time_ms": 0.0,
}
self._cache_timeout = timedelta(seconds=30)
self._cached_results: Dict[str, Tuple[SessionStatusResult, datetime]] = {}
@require(lambda self, session_id: session_id is not None)
@ensure(lambda result: isinstance(result, SessionStatusResult))
async def aggregate_session_status(
self,
session_id: SessionId,
agent_states: List[AgentState],
strategy: AggregationStrategy = AggregationStrategy.DETAILED,
include_health: bool = True,
include_metrics: bool = True,
use_cache: bool = True,
) -> SessionStatusResult:
"""
Aggregate comprehensive status for all agents in a session.
Args:
session_id: Session identifier
agent_states: List of agent states to aggregate
strategy: Aggregation strategy (fast/detailed/comprehensive)
include_health: Whether to include health information
include_metrics: Whether to include resource metrics
use_cache: Whether to use cached results
Returns:
SessionStatusResult with aggregated information
"""
collection_start = datetime.utcnow()
start_time = asyncio.get_event_loop().time()
# Check cache if enabled
if use_cache and strategy != AggregationStrategy.COMPREHENSIVE:
cached_result = self._get_cached_result(str(session_id))
if cached_result:
return cached_result
try:
self._collection_stats["total_collections"] += 1
# Phase 1: Parallel agent status collection
agent_status_infos = await self._collect_agent_statuses(
agent_states=agent_states,
strategy=strategy,
include_health=include_health,
include_metrics=include_metrics,
)
# Phase 2: Metrics aggregation
session_metrics = self._aggregate_metrics(agent_status_infos)
# Phase 3: Health evaluation
session_health = self._evaluate_session_health(
agent_status_infos, session_metrics
)
# Phase 4: Result compilation
collection_duration = (asyncio.get_event_loop().time() - start_time) * 1000
result = SessionStatusResult(
session_id=session_id,
session_name=f"Session_{session_id}", # Would get from session manager
session_health=session_health,
metrics=session_metrics,
agents=agent_status_infos,
collection_timestamp=collection_start,
collection_duration_ms=collection_duration,
partial_results=any(
not agent.is_accessible for agent in agent_status_infos
),
errors=[
agent.error_message
for agent in agent_status_infos
if agent.error_message
],
warnings=self._generate_warnings(agent_status_infos, session_metrics),
)
# Update statistics
self._collection_stats["successful_collections"] += 1
self._update_avg_collection_time(collection_duration)
# Cache result
if use_cache:
self._cache_result(str(session_id), result)
return result
except Exception as e:
self._collection_stats["failed_collections"] += 1
# Return minimal result with error
return SessionStatusResult(
session_id=session_id,
session_name=f"Session_{session_id}",
session_health=SessionHealth.FAILED,
metrics=SessionMetrics(),
agents=[],
collection_timestamp=collection_start,
collection_duration_ms=(asyncio.get_event_loop().time() - start_time)
* 1000,
partial_results=True,
errors=[f"Collection failed: {str(e)}"],
)
async def _collect_agent_statuses(
self,
agent_states: List[AgentState],
strategy: AggregationStrategy,
include_health: bool,
include_metrics: bool,
) -> List[AgentStatusInfo]:
"""Collect status information from all agents in parallel."""
if not agent_states:
return []
# Create collection tasks for parallel execution
tasks = [
self._collect_single_agent_status(
agent_state=agent,
strategy=strategy,
include_health=include_health,
include_metrics=include_metrics,
)
for agent in agent_states
]
# Execute all collections in parallel with timeout
try:
agent_statuses = await asyncio.wait_for(
asyncio.gather(*tasks, return_exceptions=True),
timeout=30.0, # 30-second timeout for all collections
)
# Process results and handle exceptions
results = []
for agent_state, status_result in zip(agent_states, agent_statuses):
if isinstance(status_result, Exception):
# Create error status for failed collection
error_status = AgentStatusInfo(
agent_id=agent_state.agent_id,
agent_name=agent_state.name,
status=AgentStatus.FAILED,
session_id=agent_state.session_id,
error_message=str(status_result),
is_accessible=False,
)
results.append(error_status)
else:
results.append(status_result)
return results
except asyncio.TimeoutError:
# Handle timeout by returning partial results
self._collection_stats["partial_collections"] += 1
# Create timeout status for all agents
return [
AgentStatusInfo(
agent_id=agent.agent_id,
agent_name=agent.name,
status=AgentStatus.DEGRADED,
session_id=agent.session_id,
error_message="Collection timeout",
is_accessible=False,
)
for agent in agent_states
]
async def _collect_single_agent_status(
self,
agent_state: AgentState,
strategy: AggregationStrategy,
include_health: bool,
include_metrics: bool,
) -> AgentStatusInfo:
"""Collect status information for a single agent."""
try:
# Base status information
status_info = AgentStatusInfo(
agent_id=agent_state.agent_id,
agent_name=agent_state.name,
status=agent_state.status,
session_id=agent_state.session_id,
specialization=(
agent_state.specialization.value
if agent_state.specialization
else None
),
created_at=agent_state.created_at,
last_activity=agent_state.last_activity,
process_id=agent_state.process_id,
iterm_tab_id=agent_state.iterm_tab_id,
)
# Add metrics if requested and strategy allows
if include_metrics and strategy != AggregationStrategy.FAST:
await self._add_resource_metrics(status_info, agent_state)
# Add health information if requested
if include_health and strategy == AggregationStrategy.COMPREHENSIVE:
await self._add_health_information(status_info, agent_state)
return status_info
except Exception as e:
# Return error status
return AgentStatusInfo(
agent_id=agent_state.agent_id,
agent_name=agent_state.name,
status=AgentStatus.FAILED,
session_id=agent_state.session_id,
error_message=str(e),
is_accessible=False,
)
async def _add_resource_metrics(
self, status_info: AgentStatusInfo, agent_state: AgentState
) -> None:
"""Add resource metrics to agent status."""
try:
# TODO: Integrate with actual resource monitoring
# For now, simulate metrics collection
status_info.cpu_percent = 15.5 # Placeholder
status_info.memory_mb = 256 # Placeholder
status_info.file_descriptors = 42 # Placeholder
status_info.thread_count = 8 # Placeholder
except Exception as e:
status_info.error_message = f"Metrics collection failed: {e}"
async def _add_health_information(
self, status_info: AgentStatusInfo, agent_state: AgentState
) -> None:
"""Add health information to agent status."""
try:
# TODO: Integrate with actual health monitoring
# For now, simulate health check
status_info.health_status = HealthStatus.HEALTHY # Placeholder
status_info.last_health_check = datetime.utcnow()
status_info.health_issues = [] # Placeholder
except Exception as e:
status_info.error_message = f"Health check failed: {e}"
def _aggregate_metrics(
self, agent_statuses: List[AgentStatusInfo]
) -> SessionMetrics:
"""Aggregate metrics across all agents."""
if not agent_statuses:
return SessionMetrics()
metrics = SessionMetrics()
# Count agents by status
metrics.total_agents = len(agent_statuses)
metrics.active_agents = sum(
1 for a in agent_statuses if a.status == AgentStatus.ACTIVE
)
metrics.degraded_agents = sum(
1 for a in agent_statuses if a.status == AgentStatus.DEGRADED
)
metrics.failed_agents = sum(
1 for a in agent_statuses if a.status == AgentStatus.FAILED
)
metrics.offline_agents = sum(1 for a in agent_statuses if not a.is_accessible)
# Aggregate resource metrics
accessible_agents = [a for a in agent_statuses if a.is_accessible]
if accessible_agents:
metrics.total_cpu_percent = sum(
a.cpu_percent or 0 for a in accessible_agents
)
metrics.total_memory_mb = sum(a.memory_mb or 0 for a in accessible_agents)
metrics.total_file_descriptors = sum(
a.file_descriptors or 0 for a in accessible_agents
)
metrics.total_threads = sum(a.thread_count or 0 for a in accessible_agents)
# Calculate derived metrics
metrics.max_memory_usage_mb = max(
(a.memory_mb or 0 for a in accessible_agents), default=0
)
# Resource utilization (percentage of system capacity)
# TODO: Get actual system limits from resource manager
system_memory_mb = 8192 # 8GB placeholder
metrics.resource_utilization_percent = (
metrics.total_memory_mb / system_memory_mb * 100
if system_memory_mb > 0
else 0
)
# Calculate health score
metrics.health_score = self._calculate_health_score(agent_statuses)
# Count issues
metrics.critical_issues = sum(
1
for a in agent_statuses
if a.status == AgentStatus.FAILED or not a.is_accessible
)
metrics.warnings = sum(
1 for a in agent_statuses if a.status == AgentStatus.DEGRADED
)
return metrics
def _evaluate_session_health(
self, agent_statuses: List[AgentStatusInfo], metrics: SessionMetrics
) -> SessionHealth:
"""Evaluate overall session health."""
if not agent_statuses:
return SessionHealth.EMPTY
total_agents = len(agent_statuses)
# Critical if >50% agents failed or all offline
if (
metrics.failed_agents > total_agents * 0.5
or metrics.offline_agents == total_agents
):
return SessionHealth.CRITICAL
# Failed if >75% agents have issues
problematic_agents = (
metrics.failed_agents + metrics.degraded_agents + metrics.offline_agents
)
if problematic_agents > total_agents * 0.75:
return SessionHealth.FAILED
# Degraded if >25% agents have issues
if problematic_agents > total_agents * 0.25:
return SessionHealth.DEGRADED
# Healthy if most agents are active
if metrics.active_agents >= total_agents * 0.8:
return SessionHealth.HEALTHY
return SessionHealth.DEGRADED
def _calculate_health_score(self, agent_statuses: List[AgentStatusInfo]) -> float:
"""Calculate numeric health score (0-100)."""
if not agent_statuses:
return 0.0
total_agents = len(agent_statuses)
active_agents = sum(1 for a in agent_statuses if a.status == AgentStatus.ACTIVE)
degraded_agents = sum(
1 for a in agent_statuses if a.status == AgentStatus.DEGRADED
)
# Weight: active=100%, degraded=50%, failed/offline=0%
score = (active_agents * 100 + degraded_agents * 50) / total_agents
return round(score, 1)
def _generate_warnings(
self, agent_statuses: List[AgentStatusInfo], metrics: SessionMetrics
) -> List[str]:
"""Generate warning messages based on status analysis."""
warnings = []
if metrics.offline_agents > 0:
warnings.append(
f"{metrics.offline_agents} agents are offline or unreachable"
)
if metrics.resource_utilization_percent > 80:
warnings.append(
f"High resource utilization: {metrics.resource_utilization_percent:.1f}%"
)
if metrics.health_score < 70:
warnings.append(f"Low session health score: {metrics.health_score}")
return warnings
def _get_cached_result(self, session_id: str) -> Optional[SessionStatusResult]:
"""Get cached result if available and not expired."""
if session_id in self._cached_results:
result, timestamp = self._cached_results[session_id]
if datetime.utcnow() - timestamp < self._cache_timeout:
return result
else:
del self._cached_results[session_id]
return None
def _cache_result(self, session_id: str, result: SessionStatusResult) -> None:
"""Cache result with timestamp."""
self._cached_results[session_id] = (result, datetime.utcnow())
# Clean up old cache entries
cutoff = datetime.utcnow() - self._cache_timeout
expired_keys = [
key
for key, (_, timestamp) in self._cached_results.items()
if timestamp < cutoff
]
for key in expired_keys:
del self._cached_results[key]
def _update_avg_collection_time(self, duration_ms: float) -> None:
"""Update running average of collection time."""
current_avg = self._collection_stats["avg_collection_time_ms"]
total_collections = self._collection_stats["total_collections"]
# Exponential moving average
self._collection_stats["avg_collection_time_ms"] = (
current_avg * 0.9 + duration_ms * 0.1
)
def get_aggregation_stats(self) -> Dict[str, Any]:
"""Get aggregation performance statistics."""
total = self._collection_stats["total_collections"]
return {
**self._collection_stats,
"success_rate": (
self._collection_stats["successful_collections"] / total * 100
if total > 0
else 0.0
),
"partial_rate": (
self._collection_stats["partial_collections"] / total * 100
if total > 0
else 0.0
),
"cache_entries": len(self._cached_results),
}
# Export utilities
__all__ = [
"SessionStatusAggregator",
"SessionStatusResult",
"AgentStatusInfo",
"SessionMetrics",
"SessionHealth",
"AggregationStrategy",
]
