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
59 lines
1.9 KiB
Python
59 lines
1.9 KiB
Python
"""V1R2C2 Failure Router Module.
|
|
|
|
Routes failures to appropriate recovery agent based on error type and recovery priority.
|
|
"""
|
|
|
|
from typing import Any, Dict, Optional
|
|
|
|
from orchestrator.error_classifier import ErrorClassifier
|
|
|
|
# Agent routing table
|
|
AGENT_ROUTING = {
|
|
'openclaw_auto_repair': {
|
|
'primary': 'openclaw',
|
|
'fallback': 'hermes', # if openclaw fails
|
|
'escape': 'windows_codex_app',
|
|
'description': 'OpenClaw auto-repair route',
|
|
},
|
|
'hermes_escalation': {
|
|
'primary': 'hermes',
|
|
'fallback': 'openclaw',
|
|
'escape': 'windows_codex_app',
|
|
'description': 'Hermes escalation route',
|
|
},
|
|
'windows_codex_escalation': {
|
|
'primary': 'windows_codex_app',
|
|
'fallback': 'openclaw',
|
|
'escape': 'openclaw',
|
|
'description': 'Windows Codex App escalation route',
|
|
},
|
|
}
|
|
|
|
|
|
class FailureRouter:
|
|
"""Route action failures to the appropriate recovery agent."""
|
|
|
|
def __init__(self, classifier: Optional[ErrorClassifier] = None):
|
|
self.classifier = classifier or ErrorClassifier()
|
|
|
|
def route(self, error: str, context: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
|
"""Route a failure event to the appropriate recovery agent.
|
|
|
|
Returns:
|
|
dict with 'primary_agent', 'fallback_agent', 'escape_agent', 'selected_error_type'
|
|
"""
|
|
classification = self.classifier.classify(error)
|
|
error_type = classification['error_type']
|
|
recovery_priority = classification['recovery_priority']
|
|
|
|
route = AGENT_ROUTING.get(recovery_priority, AGENT_ROUTING['openclaw_auto_repair'])
|
|
|
|
return {
|
|
'error_type': error_type,
|
|
'recovery_priority': recovery_priority,
|
|
'primary_agent': route['primary'],
|
|
'fallback_agent': route['fallback'],
|
|
'escape_agent': route['escape'],
|
|
'selected_error_type': error_type,
|
|
}
|