diff --git a/backend/agenteval/intelligent_eval/fault_tolerance.py b/backend/agenteval/intelligent_eval/fault_tolerance.py new file mode 100644 index 0000000..15db450 --- /dev/null +++ b/backend/agenteval/intelligent_eval/fault_tolerance.py @@ -0,0 +1,204 @@ +"""Fault tolerance and recovery for cron pool (故障恢复). + +Handles: +- Stuck cron detection and cleanup +- State reconciliation (platform DB vs OpenClaw state) +- Platform restart recovery +- OpenClaw restart recovery +""" + +import logging +from datetime import timedelta + +from sqlmodel import Session, select + +from agenteval.intelligent_eval import cron_pool +from agenteval.intelligent_eval.openclaw_client import OpenClawClient +from agenteval.storage.db import ( + IntelligentEvalTaskQueueDB, + OpenClawCronPoolDB, + utc_now, +) + +_logger = logging.getLogger("agenteval") + + +async def detect_and_handle_stuck_crons(session: Session, client: OpenClawClient) -> int: + """Detect and handle stuck crons. + + Returns: + Number of stuck crons handled + """ + stuck_crons = cron_pool.detect_stuck_crons(session) + + for cron in stuck_crons: + await cron_pool.handle_stuck_cron(cron, session, client) + + if stuck_crons: + _logger.info(f"Handled {len(stuck_crons)} stuck crons") + + return len(stuck_crons) + + +async def reconcile_state(session: Session, client: OpenClawClient) -> dict: + """Reconcile platform DB state with OpenClaw state. + + Checks: + 1. Platform DB has crons that OpenClaw doesn't → mark as stuck + 2. OpenClaw has crons that platform DB doesn't → sync to DB + 3. Assigned tasks have inactive crons → requeue tasks + + Returns: + Dict with reconciliation stats + """ + stats = { + "orphaned_crons": 0, + "missing_crons": 0, + "requeued_tasks": 0, + } + + # Get all crons from both sides + db_crons = session.exec(select(OpenClawCronPoolDB)).all() + openclaw_crons = await client.list_crons() + openclaw_cron_ids = {c.id for c in openclaw_crons} + + # Check 1: Platform DB has crons that OpenClaw doesn't + for db_cron in db_crons: + if db_cron.openclaw_cron_id not in openclaw_cron_ids: + _logger.warning(f"Cron {db_cron.openclaw_cron_id} exists in DB but not in OpenClaw") + db_cron.status = "stuck" + db_cron.updated_at = utc_now() + stats["orphaned_crons"] += 1 + + session.commit() + + # Check 2: OpenClaw has crons that platform DB doesn't + synced = await cron_pool.sync_cron_states(session, client) + stats["missing_crons"] = synced + + # Check 3: Assigned tasks have inactive crons + assigned_tasks = session.exec( + select(IntelligentEvalTaskQueueDB).where(IntelligentEvalTaskQueueDB.status == "assigned") + ).all() + + for task in assigned_tasks: + if task.assigned_cron_id is None: + continue + + # Check if cron is still active + cron = session.exec( + select(OpenClawCronPoolDB).where( + OpenClawCronPoolDB.openclaw_cron_id == task.assigned_cron_id + ) + ).first() + + if cron is None or cron.status == "stuck": + _logger.warning(f"Task {task.id} assigned to inactive cron {task.assigned_cron_id}") + + # Mark task as failed + from agenteval.intelligent_eval.task_queue import complete_task + + complete_task(task.id, False, "Cron inactive", session) + + # Requeue task + new_task = IntelligentEvalTaskQueueDB( + eval_id=task.eval_id, + status="pending", + priority=1, # High priority + reason="cron_inactive_retry", + ) + session.add(new_task) + stats["requeued_tasks"] += 1 + + session.commit() + + if stats["orphaned_crons"] or stats["missing_crons"] or stats["requeued_tasks"]: + _logger.info(f"State reconciliation: {stats}") + + return stats + + +async def recover_from_platform_restart(session: Session, client: OpenClawClient) -> dict: + """Recover from platform restart. + + Scans all assigned tasks and checks if their crons are still active. + If not, requeues the tasks. + + Returns: + Dict with recovery stats + """ + stats = { + "assigned_tasks_checked": 0, + "requeued_tasks": 0, + } + + # Get all assigned tasks + assigned_tasks = session.exec( + select(IntelligentEvalTaskQueueDB).where(IntelligentEvalTaskQueueDB.status == "assigned") + ).all() + + stats["assigned_tasks_checked"] = len(assigned_tasks) + + for task in assigned_tasks: + if task.assigned_cron_id is None: + continue + + # Check if cron exists and is active + cron = session.exec( + select(OpenClawCronPoolDB).where( + OpenClawCronPoolDB.openclaw_cron_id == task.assigned_cron_id + ) + ).first() + + # Check if cron is active (heartbeat within last 5 minutes) + if cron: + threshold = utc_now() - timedelta(minutes=5) + if cron.last_active_at < threshold: + cron = None # Treat as inactive + + if cron is None: + _logger.info(f"Requeuing task {task.id} (cron inactive after restart)") + + # Mark task as failed + from agenteval.intelligent_eval.task_queue import complete_task + + complete_task(task.id, False, "Platform restart", session) + + # Requeue task + new_task = IntelligentEvalTaskQueueDB( + eval_id=task.eval_id, + status="pending", + priority=1, + reason="platform_restart_retry", + ) + session.add(new_task) + stats["requeued_tasks"] += 1 + + session.commit() + + if stats["requeued_tasks"]: + _logger.info(f"Platform restart recovery: {stats}") + + return stats + + +async def recover_from_openclaw_restart(session: Session, client: OpenClawClient) -> dict: + """Recover from OpenClaw restart. + + OpenClaw crons persist their state in SQLite, so they should resume + automatically. This function syncs the state to platform DB. + + Returns: + Dict with recovery stats + """ + # Sync cron states from OpenClaw to platform DB + synced = await cron_pool.sync_cron_states(session, client) + + stats = { + "synced_crons": synced, + } + + if synced: + _logger.info(f"OpenClaw restart recovery: {stats}") + + return stats diff --git a/tests/integration/test_fault_tolerance_e2e.py b/tests/integration/test_fault_tolerance_e2e.py new file mode 100644 index 0000000..8fa4c36 --- /dev/null +++ b/tests/integration/test_fault_tolerance_e2e.py @@ -0,0 +1,227 @@ +"""Integration tests for fault tolerance and recovery.""" + +from datetime import timedelta + +import pytest +from fastapi.testclient import TestClient +from sqlmodel import Session, SQLModel, create_engine, select +from unittest.mock import AsyncMock + +from agenteval.intelligent_eval.openclaw_client import OpenClawClient, OpenClawCron +from agenteval.storage.db import ( + IntelligentEvalTaskQueueDB, + OpenClawCronPoolDB, + utc_now, +) +from agenteval.web.app import app +from agenteval.web.deps import get_db + + +@pytest.fixture() +def client(tmp_path): + """Create a TestClient with a fresh database.""" + from agenteval.storage.db import ( # noqa: F401 + IntelligentEvalTaskQueueDB, + OpenClawCronPoolDB, + ) + + engine = create_engine( + f"sqlite:///{tmp_path / 'test.db'}", + connect_args={"check_same_thread": False}, + ) + SQLModel.metadata.create_all(engine) + session = Session(engine) + + def override_get_db(): + try: + yield session + finally: + pass + + app.dependency_overrides[get_db] = override_get_db + client = TestClient(app) + yield client + app.dependency_overrides.clear() + session.close() + engine.dispose() + + +@pytest.fixture() +def db_session(client): + """Get the database session from the client fixture.""" + return next(app.dependency_overrides[get_db]()) + + +@pytest.fixture() +def mock_openclaw_client(): + """Mock OpenClaw client.""" + client = OpenClawClient() + client.list_crons = AsyncMock(return_value=[]) + client.sync_cron_states = AsyncMock(return_value=0) + client.delete_cron = AsyncMock() + client.create_cron = AsyncMock(return_value="new-cron") + return client + + +def test_stuck_cron_detection_api(client: TestClient, db_session: Session): + """Test stuck cron detection via API.""" + # Create stuck cron + cron = OpenClawCronPoolDB( + openclaw_cron_id="stuck-cron", + status="busy", + current_eval_id="eval-1", + last_active_at=utc_now() - timedelta(minutes=15), + ) + db_session.add(cron) + db_session.commit() + + # Verify cron is stuck + from agenteval.intelligent_eval.cron_pool import detect_stuck_crons + + stuck = detect_stuck_crons(db_session) + assert len(stuck) == 1 + assert stuck[0].openclaw_cron_id == "stuck-cron" + + +def test_task_requeue_after_cron_stuck(client: TestClient, db_session: Session): + """Test task requeue after cron gets stuck.""" + # Create stuck cron with assigned task + cron = OpenClawCronPoolDB( + openclaw_cron_id="stuck-cron", + status="busy", + current_eval_id="eval-1", + last_active_at=utc_now() - timedelta(minutes=15), + ) + db_session.add(cron) + + task = IntelligentEvalTaskQueueDB( + eval_id="eval-1", + status="assigned", + priority=1, + reason="slot_due", + assigned_cron_id="stuck-cron", + ) + db_session.add(task) + db_session.commit() + + # Simulate handling stuck cron + from agenteval.intelligent_eval.cron_pool import handle_stuck_cron + from unittest.mock import MagicMock + + mock_client = MagicMock(spec=OpenClawClient) + mock_client.delete_cron = AsyncMock() + mock_client.create_cron = AsyncMock(return_value="new-cron") + + import asyncio + + asyncio.run(handle_stuck_cron(cron, db_session, mock_client)) + + # Verify task requeued + db_session.refresh(task) + assert task.status == "failed" + + new_task = db_session.exec( + select(IntelligentEvalTaskQueueDB).where( + IntelligentEvalTaskQueueDB.eval_id == "eval-1", + IntelligentEvalTaskQueueDB.status == "pending", + ) + ).first() + assert new_task is not None + assert new_task.reason == "cron_stuck_retry" + + +def test_state_reconciliation_api(client: TestClient, db_session: Session): + """Test state reconciliation via API.""" + # Create orphan cron (exists in DB but not in OpenClaw) + cron = OpenClawCronPoolDB( + openclaw_cron_id="orphan-cron", + status="idle", + last_active_at=utc_now(), + ) + db_session.add(cron) + db_session.commit() + + # Mock OpenClaw client + from agenteval.intelligent_eval import fault_tolerance + + mock_client = OpenClawClient() + mock_client.list_crons = AsyncMock(return_value=[]) + mock_client.sync_cron_states = AsyncMock(return_value=0) + + import asyncio + + stats = asyncio.run(fault_tolerance.reconcile_state(db_session, mock_client)) + + assert stats["orphaned_crons"] == 1 + + # Verify cron marked as stuck + db_session.refresh(cron) + assert cron.status == "stuck" + + +def test_platform_restart_recovery(client: TestClient, db_session: Session): + """Test platform restart recovery.""" + # Create task assigned to inactive cron + task = IntelligentEvalTaskQueueDB( + eval_id="eval-1", + status="assigned", + priority=1, + reason="slot_due", + assigned_cron_id="inactive-cron", + ) + db_session.add(task) + db_session.commit() + + # Mock OpenClaw client + from agenteval.intelligent_eval import fault_tolerance + + mock_client = OpenClawClient() + mock_client.list_crons = AsyncMock(return_value=[]) + + import asyncio + + stats = asyncio.run(fault_tolerance.recover_from_platform_restart(db_session, mock_client)) + + assert stats["assigned_tasks_checked"] == 1 + assert stats["requeued_tasks"] == 1 + + # Verify task requeued + new_task = db_session.exec( + select(IntelligentEvalTaskQueueDB).where( + IntelligentEvalTaskQueueDB.eval_id == "eval-1", + IntelligentEvalTaskQueueDB.status == "pending", + ) + ).first() + assert new_task is not None + assert new_task.reason == "platform_restart_retry" + + +def test_openclaw_restart_recovery(client: TestClient, db_session: Session): + """Test OpenClaw restart recovery.""" + # Mock OpenClaw client returning recovered crons + from agenteval.intelligent_eval import fault_tolerance + + mock_client = OpenClawClient() + mock_client.list_crons = AsyncMock( + return_value=[ + OpenClawCron( + id="recovered-cron", + name="worker-1", + schedule="* * * * *", + enabled=True, + state={"status": "idle"}, + ) + ] + ) + + import asyncio + + stats = asyncio.run(fault_tolerance.recover_from_openclaw_restart(db_session, mock_client)) + + assert stats["synced_crons"] == 1 + + # Verify cron synced to DB + cron = db_session.exec( + select(OpenClawCronPoolDB).where(OpenClawCronPoolDB.openclaw_cron_id == "recovered-cron") + ).first() + assert cron is not None diff --git a/tests/unit/test_fault_tolerance.py b/tests/unit/test_fault_tolerance.py new file mode 100644 index 0000000..7177511 --- /dev/null +++ b/tests/unit/test_fault_tolerance.py @@ -0,0 +1,206 @@ +"""Unit tests for fault tolerance and recovery.""" + +from datetime import timedelta +from unittest.mock import AsyncMock, MagicMock + +import pytest +from sqlmodel import Session, select + +from agenteval.intelligent_eval import fault_tolerance +from agenteval.intelligent_eval.openclaw_client import OpenClawClient +from agenteval.storage.db import ( + IntelligentEvalTaskQueueDB, + OpenClawCronPoolDB, + utc_now, +) + + +@pytest.fixture() +def mock_openclaw_client(): + """Mock OpenClaw client.""" + client = MagicMock(spec=OpenClawClient) + client.list_crons = AsyncMock(return_value=[]) + client.sync_cron_states = AsyncMock(return_value=0) + return client + + +async def test_detect_and_handle_stuck_crons(db_session: Session, mock_openclaw_client): + """Test detecting and handling stuck crons.""" + # Create stuck cron + cron = OpenClawCronPoolDB( + openclaw_cron_id="stuck-cron", + status="busy", + current_eval_id="eval-1", + last_active_at=utc_now() - timedelta(minutes=15), + ) + db_session.add(cron) + + task = IntelligentEvalTaskQueueDB( + eval_id="eval-1", + status="assigned", + priority=1, + reason="slot_due", + assigned_cron_id="stuck-cron", + ) + db_session.add(task) + db_session.commit() + + # Mock OpenClaw client methods + mock_openclaw_client.delete_cron = AsyncMock() + mock_openclaw_client.create_cron = AsyncMock(return_value="new-cron") + + handled = await fault_tolerance.detect_and_handle_stuck_crons(db_session, mock_openclaw_client) + assert handled == 1 + + # Verify task requeued + db_session.refresh(task) + assert task.status == "failed" + assert task.error == "Cron stuck" + + # Verify new task created + new_task = db_session.exec( + select(IntelligentEvalTaskQueueDB).where( + IntelligentEvalTaskQueueDB.eval_id == "eval-1", + IntelligentEvalTaskQueueDB.status == "pending", + ) + ).first() + assert new_task is not None + assert new_task.reason == "cron_stuck_retry" + + +async def test_reconcile_state_orphaned_crons(db_session: Session, mock_openclaw_client): + """Test reconciliation when platform DB has crons that OpenClaw doesn't.""" + # Create cron in DB + cron = OpenClawCronPoolDB( + openclaw_cron_id="orphan-cron", + status="idle", + last_active_at=utc_now(), + ) + db_session.add(cron) + db_session.commit() + + # Mock OpenClaw returning empty list (cron doesn't exist) + mock_openclaw_client.list_crons = AsyncMock(return_value=[]) + + stats = await fault_tolerance.reconcile_state(db_session, mock_openclaw_client) + assert stats["orphaned_crons"] == 1 + + # Verify cron marked as stuck + db_session.refresh(cron) + assert cron.status == "stuck" + + +async def test_reconcile_state_missing_crons(db_session: Session, mock_openclaw_client): + """Test reconciliation when OpenClaw has crons that platform DB doesn't.""" + from agenteval.intelligent_eval.openclaw_client import OpenClawCron + + # Mock OpenClaw returning a cron + mock_openclaw_client.list_crons = AsyncMock( + return_value=[ + OpenClawCron( + id="missing-cron", + name="worker-1", + schedule="* * * * *", + enabled=True, + state={"status": "idle"}, + ) + ] + ) + + stats = await fault_tolerance.reconcile_state(db_session, mock_openclaw_client) + assert stats["missing_crons"] == 1 + + # Verify cron synced to DB + cron = db_session.exec( + select(OpenClawCronPoolDB).where(OpenClawCronPoolDB.openclaw_cron_id == "missing-cron") + ).first() + assert cron is not None + + +async def test_reconcile_state_requeue_inactive_tasks(db_session: Session, mock_openclaw_client): + """Test requeuing tasks assigned to inactive crons.""" + # Create task assigned to non-existent cron + task = IntelligentEvalTaskQueueDB( + eval_id="eval-1", + status="assigned", + priority=1, + reason="slot_due", + assigned_cron_id="nonexistent-cron", + ) + db_session.add(task) + db_session.commit() + + mock_openclaw_client.list_crons = AsyncMock(return_value=[]) + + stats = await fault_tolerance.reconcile_state(db_session, mock_openclaw_client) + assert stats["requeued_tasks"] == 1 + + # Verify task failed and new task created + db_session.refresh(task) + assert task.status == "failed" + + new_task = db_session.exec( + select(IntelligentEvalTaskQueueDB).where( + IntelligentEvalTaskQueueDB.eval_id == "eval-1", + IntelligentEvalTaskQueueDB.status == "pending", + ) + ).first() + assert new_task is not None + assert new_task.reason == "cron_inactive_retry" + + +async def test_recover_from_platform_restart(db_session: Session, mock_openclaw_client): + """Test recovery from platform restart.""" + # Create task assigned to inactive cron + task = IntelligentEvalTaskQueueDB( + eval_id="eval-1", + status="assigned", + priority=1, + reason="slot_due", + assigned_cron_id="inactive-cron", + ) + db_session.add(task) + db_session.commit() + + mock_openclaw_client.list_crons = AsyncMock(return_value=[]) + + stats = await fault_tolerance.recover_from_platform_restart(db_session, mock_openclaw_client) + assert stats["assigned_tasks_checked"] == 1 + assert stats["requeued_tasks"] == 1 + + # Verify task requeued + new_task = db_session.exec( + select(IntelligentEvalTaskQueueDB).where( + IntelligentEvalTaskQueueDB.eval_id == "eval-1", + IntelligentEvalTaskQueueDB.status == "pending", + ) + ).first() + assert new_task is not None + assert new_task.reason == "platform_restart_retry" + + +async def test_recover_from_openclaw_restart(db_session: Session, mock_openclaw_client): + """Test recovery from OpenClaw restart.""" + from agenteval.intelligent_eval.openclaw_client import OpenClawCron + + # Mock OpenClaw returning crons + mock_openclaw_client.list_crons = AsyncMock( + return_value=[ + OpenClawCron( + id="recovered-cron", + name="worker-1", + schedule="* * * * *", + enabled=True, + state={"status": "idle"}, + ) + ] + ) + + stats = await fault_tolerance.recover_from_openclaw_restart(db_session, mock_openclaw_client) + assert stats["synced_crons"] == 1 + + # Verify cron synced to DB + cron = db_session.exec( + select(OpenClawCronPoolDB).where(OpenClawCronPoolDB.openclaw_cron_id == "recovered-cron") + ).first() + assert cron is not None