feat(intelligent-eval): implement fault tolerance and recovery (ticket 06)
- Add fault_tolerance.py with stuck cron detection and handling - Implement state reconciliation (platform DB vs OpenClaw state) - Implement platform restart recovery (requeue inactive tasks) - Implement OpenClaw restart recovery (sync cron states) - Add 6 unit tests and 5 integration tests All 824 tests passing.
This commit is contained in:
sinohqb
parent
e6f98aaa6d
commit
1d9228fd86
204
backend/agenteval/intelligent_eval/fault_tolerance.py
Normal file
204
backend/agenteval/intelligent_eval/fault_tolerance.py
Normal file
@ -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
|
||||
227
tests/integration/test_fault_tolerance_e2e.py
Normal file
227
tests/integration/test_fault_tolerance_e2e.py
Normal file
@ -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
|
||||
206
tests/unit/test_fault_tolerance.py
Normal file
206
tests/unit/test_fault_tolerance.py
Normal file
@ -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
|
||||
Loading…
Reference in New Issue
Block a user