aios
19a2ce8439
- Integrate ActionMonitor into ChatGPT GitOps API for failure tracking - Add ErrorClassifier with permission, git, runtime, system, secret violation error types - Add FailureRouter to route errors to correct recovery agents (openclaw/hermes/windows_codex_app) - Add RecoveryController orchestrating full pipeline: capture→classify→route→plan→execute→validate→record - Update error_classifier to include ClientResponseError, rate limit, and secret_violation patterns - All 8 integration tests passing (ActionMonitor, ErrorClassifier, FailureRouter, RecoveryController, API 500 scenario) - Push baseline to V1R2C2 with tag v1r2c2-baseline
96 lines
2.9 KiB
Python
96 lines
2.9 KiB
Python
"""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
|