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
106 lines
4.0 KiB
Python
106 lines
4.0 KiB
Python
"""V1R2C2 Recovery Controller Module.
|
|
|
|
Orchestrates the full recovery pipeline:
|
|
Action Monitor → Error Classifier → Failure Router → Recovery Controller → Repair → Validate → GitOps Record
|
|
|
|
This module demonstrates the complete recovery chain.
|
|
"""
|
|
|
|
from typing import Any, Dict, Optional
|
|
from datetime import datetime, timezone
|
|
|
|
from orchestrator.error_classifier import ErrorClassifier
|
|
from orchestrator.failure_router import FailureRouter
|
|
|
|
|
|
class RecoveryController:
|
|
"""Capture and process action failures through the recovery pipeline."""
|
|
|
|
def __init__(self):
|
|
self.classifier = ErrorClassifier()
|
|
self.router = FailureRouter(self.classifier)
|
|
self.recovery_history: list[Dict[str, Any]] = []
|
|
|
|
def process_failure(self, action: str, error: str, context: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
|
"""Process a single failure through the complete recovery pipeline.
|
|
|
|
Pipeline: Action Monitor → Error Classifier → Failure Router → Recovery Plan → Execute → Validate → Record
|
|
|
|
Returns:
|
|
dict with full recovery chain output
|
|
"""
|
|
context = context or {}
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
|
|
# Step 1: Classify
|
|
classification = self.classifier.classify(error)
|
|
error_type = classification['error_type']
|
|
|
|
# Step 2: Route
|
|
route_info = self.router.route(error, context)
|
|
|
|
# Step 3: Generate recovery plan
|
|
plan = self._generate_plan(action, error_type, route_info, context)
|
|
|
|
# Step 4: Execute (simulated - actual execution requires agent infrastructure)
|
|
execution = self._execute_recovery(action, error_type, plan)
|
|
|
|
# Step 5: Validate
|
|
validation = self._validate_recovery(execution)
|
|
|
|
# Step 6: Record
|
|
record = {
|
|
'timestamp': now,
|
|
'action': action,
|
|
'error': error,
|
|
'error_type': error_type,
|
|
'classification': classification,
|
|
'route_info': route_info,
|
|
'plan': plan,
|
|
'execution': execution,
|
|
'validation': validation,
|
|
}
|
|
|
|
self.recovery_history.append(record)
|
|
return record
|
|
|
|
def _generate_plan(self, action: str, error_type: str, route_info: Dict[str, Any], context: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Generate a recovery plan based on classification and routing."""
|
|
primary = route_info.get('primary_agent', 'openclaw')
|
|
return {
|
|
'action': action,
|
|
'error_type': error_type,
|
|
'primary_agent': primary,
|
|
'fallback_agent': route_info.get('fallback_agent', 'hermes'),
|
|
'escape_agent': route_info.get('escape_agent', 'windows_codex_app'),
|
|
'steps': [
|
|
f'classify: {error_type}',
|
|
f'assign: {primary}',
|
|
f'fallback: {route_info.get("fallback_agent", "hermes")}',
|
|
f'escape: {route_info.get("escape_agent", "windows_codex_app")}',
|
|
],
|
|
}
|
|
|
|
def _execute_recovery(self, action: str, error_type: str, plan: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Execute the recovery plan (simulation)."""
|
|
# In production, this would invoke the selected agent with the plan
|
|
return {
|
|
'status': 'recovery_initiated',
|
|
'agent': plan.get('primary_agent'),
|
|
'action': action,
|
|
'error_type': error_type,
|
|
}
|
|
|
|
def _validate_recovery(self, execution: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Validate the recovery action was taken."""
|
|
# In production, this would check if the agent applied the fix
|
|
return {
|
|
'validated': execution.get('status') == 'recovery_initiated',
|
|
'execution_status': execution.get('status'),
|
|
'validation_time': datetime.now(timezone.utc).isoformat(),
|
|
}
|
|
|
|
def latest_recovery(self) -> Optional[Dict[str, Any]]:
|
|
"""Get the latest recovery record."""
|
|
return self.recovery_history[-1] if self.recovery_history else None
|