"""Agent OS health score module for V1R1C5. Calculates health scores based on quality, acceptance rate, and historical reliability. Used for scheduler selection and consensus weighting. """ from typing import Dict, Any class AgentHealthScore: """Health score calculator for agents. Combines multiple factors into a single health score (0.0 - 1.0) used by: - Scheduler for task assignment priority - Consensus engine for weighted voting - Escalation and recovery decisions """ def __init__(self): self.default_quality = 0.5 self.default_acceptance_rate = 0.8 self.default_reliability = 0.7 def calculate_health_score( self, quality_score: float = None, acceptance_rate: float = None, historical_reliability: float = None, ) -> float: """Calculate composite health score from individual metrics. Formula: health_score = (quality_score * 0.35) + (acceptance_rate * 0.35) + (historical_reliability * 0.30) Args: quality_score: Quality score from agent runtime metrics (0.0-1.0) acceptance_rate: Historical acceptance rate from agent store (0.0-1.0) historical_reliability: Reliability score from agent store (0.0-1.0) Returns: Composite health score between 0.0 and 1.0 """ quality = quality_score if quality_score is not None else self.default_quality acceptance = acceptance_rate if acceptance_rate is not None else self.default_acceptance_rate reliability = historical_reliability if historical_reliability is not None else self.default_reliability score = ( quality * 0.35 + acceptance * 0.35 + reliability * 0.30 ) return min(max(score, 0.0), 1.0) def get_health_status(self, score: float) -> str: """Get health status string from score. Args: score: Health score (0.0-1.0) Returns: One of: "healthy", "degraded", "unhealthy" """ if score >= 0.75: return "healthy" elif score >= 0.4: return "degraded" else: return "unhealthy" def select_agent(self, agents: list) -> str: """Select best agent from a list based on health scores. Args: agents: List of dicts with agent_id and health_score keys Returns: Agent ID of the healthiest agent """ if not agents: return "" return max(agents, key=lambda a: a.get("health_score", 0.0))["agent_id"] def get_weight_multiplier(self, score: float) -> float: """Get weight multiplier based on health score. Higher health scores yield higher multipliers for consensus weighting. Args: score: Health score (0.0-1.0) Returns: Weight multiplier (0.0-1.0) """ return score ** 1.5 # Amplify differences