aios/runtime/agent-os/orchestrator/integration_test_v1r2c2.py

#!/usr/bin/env python3
"""V1R2C2 integration test: verify the recovery pipeline end-to-end.

Tests: Action Monitor → Error Classifier → Failure Router → Recovery Controller
"""

import sys
import json
import tempfile
import os
import time
import subprocess
from pathlib import Path

# Path to the gitops workspace
WORKSPACE = Path('/data/aios-gitops/workspace')
sys.path.insert(0, str(WORKSPACE / 'runtime' / 'agent-os'))

from orchestrator.action_monitor import ActionMonitor
from orchestrator.error_classifier import ErrorClassifier
from orchestrator.failure_router import FailureRouter
from orchestrator.recovery_controller import RecoveryController


def debug_imports():
    """Verify all modules load correctly."""
    print("=== Module Import Verification ===")
    print(f"ActionMonitor: {ActionMonitor}")
    print(f"ErrorClassifier: {ErrorClassifier}")
    print(f"FailureRouter: {FailureRouter}")
    print(f"RecoveryController: {RecoveryController}")
    print()


def test_action_monitor():
    """Test the ActionMonitor records failure events correctly."""
    print("=== ActionMonitor Test ===")
    monitor = ActionMonitor()
    assert monitor.latest() is None, "Monitor should start empty"
    assert monitor.list_events() == [], "Monitor events list should start empty"

    e = monitor.record_failure('branch:create', 'Permission denied', {'branch': 'test'})
    assert e is not None
    assert e.action == 'branch:create'
    assert e.error == 'Permission denied'
    assert e.context == {'branch': 'test'}
    assert e.event_id is not None
    assert e.created_at > 0

    e2 = monitor.record_failure('file:write', 'File not found', {'path': '/tmp/x'})
    assert len(monitor.list_events()) == 2
    assert monitor.latest() == e2
    print("PASS: ActionMonitor records and retrieves events correctly")
    return monitor


def test_error_classifier():
    """Test the ErrorClassifier categorizes errors correctly."""
    print("=== ErrorClassifier Test ===")
    clf = ErrorClassifier()

    tests = [
        # (error_string, expected_error_type, expected_recovery_priority)
        ('Permission denied', 'permission_error', 'openclaw_auto_repair'),
        ('git failed: merge conflict', 'git_error', 'openclaw_auto_repair'),
        ('git push rejected: permission', 'git_error', 'openclaw_auto_repair'),
        ('Connection refused', 'runtime_error', 'hermes_escalation'),
        ('timeout', 'runtime_error', 'hermes_escalation'),
        ('ClientResponseError', 'runtime_error', 'hermes_escalation'),
        ('secret_rejected: use secret_ref', 'secret_violation', 'windows_codex_escalation'),
        ('unauthorized', 'permission_error', 'openclaw_auto_repair'),
        ('forbidden', 'permission_error', 'openclaw_auto_repair'),
        ('API rate limit exceeded', 'runtime_error', 'hermes_escalation'),
        ('Internal Server Error', 'runtime_error', 'hermes_escalation'),
        ('repository is locked', 'git_error', 'openclaw_auto_repair'),
        ('unknown weird error', 'unknown', 'openclaw_auto_repair'),
        ('No such file or directory', 'unknown', 'openclaw_auto_repair'),
        ('File exists', 'unknown', 'openclaw_auto_repair'),
        ('authentication failed', 'permission_error', 'openclaw_auto_repair'),
    ]

    for error_str, expected_type, expected_priority in tests:
        result = clf.classify(error_str)
        assert result['error_type'] == expected_type, f"Expected {expected_type}, got {result['error_type']} for '{error_str}'"
        assert result['recovery_priority'] == expected_priority, f"Expected {expected_priority}, got {result['recovery_priority']} for '{error_str}'"

    print("PASS: ErrorClassifier correctly classifies all error types")
    return clf


def test_failure_router():
    """Test the FailureRouter routes failures to the correct agent."""
    print("=== FailureRouter Test ===")
    router = FailureRouter()

    tests = [
        # (error, expected_primary, expected_fallback, expected_escape)
        ('Permission denied', 'openclaw', 'hermes', 'windows_codex_app'),
        ('ClientResponseError', 'hermes', 'openclaw', 'windows_codex_app'),
        ('secret_rejected: use secret_ref', 'windows_codex_app', 'openclaw', 'openclaw'),
    ]

    for error, primary, fallback, escape in tests:
        result = router.route(error)
        assert result['primary_agent'] == primary, f"Expected primary={primary}, got {result['primary_agent']}"
        assert result['fallback_agent'] == fallback, f"Expected fallback={fallback}, got {result['fallback_agent']}"
        assert result['escape_agent'] == escape, f"Expected escape={escape}, got {result['escape_agent']}"

    print("PASS: FailureRouter routes errors to correct agents")
    return router


def test_recovery_controller():
    """Test the full recovery pipeline end-to-end."""
    print("=== RecoveryController Test ===")
    controller = RecoveryController()

    tests = [
        # (action, error, expected_error_type)
        ('branch:create', 'Permission denied', 'permission_error'),
        ('file:write', 'ClientResponseError', 'runtime_error'),
        ('model:write', 'secret_rejected: use secret_ref', 'secret_violation'),
    ]

    for action, error, expected_type in tests:
        result = controller.process_failure(action, error, {'source': 'chatgpt-api'})
        assert result['error_type'] == expected_type, f"Expected {expected_type}, got {result['error_type']}"
        assert result['action'] == action
        assert result['plan'] is not None
        assert result['execution'] is not None
        assert result['validation'] is not None
        assert 'timestamp' in result
        print(f"  {action}: {error} → {expected_type} ✓")

    assert len(controller.recovery_history) == 3
    assert controller.latest_recovery() == controller.recovery_history[-1]

    print("PASS: RecoveryController processes full pipeline correctly")
    return controller


def test_api_500_scenario():
    """Simulate the real-world scenario of API 500 errors."""
    print("=== API 500 Scenario Test ===")
    monitor = ActionMonitor()
    controller = RecoveryController()

    # Simulate the /api/file 500 errors seen in production
    failures = [
        ('file:write', 'Internal Server Error', {'path': 'test.md', 'method': 'POST'}),
        ('branch:create', 'ClientResponseError', {'branch': 'test', 'method': 'POST'}),
        ('file:write', 'timeout', {'path': 'test2.md', 'method': 'POST'}),
    ]

    for action, error, context in failures:
        event = monitor.record_failure(action, error, context)
        result = controller.process_failure(action, error, context)

        assert result['error_type'] is not None
        assert result['route_info'] is not None

    # Verify the most critical errors (network errors) route to hermes
    network_results = [
        r for r in controller.recovery_history
        if r['error_type'] == 'network_error'
    ]
    for r in network_results:
        assert r['route_info']['primary_agent'] == 'hermes'

    print(f"PASS: API 500 scenario processed {len(failures)} failures correctly")
    print(f"  Network errors routed to hermes: {len(network_results)}")


def main():
    """Run all tests."""
    print("=" * 60)
    print("V1R2C2 Integration Tests - Autonomous Recovery Pipeline")
    print("=" * 60)
    print()

    debug_imports()
    test_action_monitor()
    test_error_classifier()
    test_failure_router()
    test_recovery_controller()
    test_api_500_scenario()

    print()
    print("=" * 60)
    print("ALL TESTS PASSED")
    print("=" * 60)
    return 0


if __name__ == '__main__':
    sys.exit(main())