diff --git a/backend/agenteval/intelligent_eval/cron_pool.py b/backend/agenteval/intelligent_eval/cron_pool.py new file mode 100644 index 0000000..c58948f --- /dev/null +++ b/backend/agenteval/intelligent_eval/cron_pool.py @@ -0,0 +1,311 @@ +"""Cron pool management for intelligent evaluations (Cron 池管理). + +Platform manages a pool of OpenClaw crons (5-20) that can process any +intelligent evaluation. Pool automatically scales up/down based on load. +""" + +import logging +from datetime import datetime, timedelta +from typing import Optional + +from sqlmodel import Session, select + +from agenteval.intelligent_eval.openclaw_client import OpenClawClient +from agenteval.storage.db import OpenClawCronPoolDB, utc_now + +_logger = logging.getLogger("agenteval") + +# Pool configuration +MIN_POOL_SIZE = 5 +MAX_POOL_SIZE = 20 +SCALE_UP_THRESHOLD = 0.8 # busy/total > 0.8 triggers scale up +SCALE_DOWN_THRESHOLD = 2 # idle > min_size * 2 triggers scale down +STUCK_THRESHOLD_MINUTES = 10 + + +async def initialize_pool(session: Session, client: OpenClawClient) -> int: + """Initialize cron pool on startup. + + Creates MIN_POOL_SIZE crons if pool is empty. + + Returns: + Number of crons created + """ + # Check if pool already initialized + existing = session.exec(select(OpenClawCronPoolDB)).all() + if existing: + _logger.info(f"Cron pool already initialized with {len(existing)} crons") + return 0 + + # Create MIN_POOL_SIZE crons + created = 0 + for i in range(MIN_POOL_SIZE): + try: + cron_id = await client.create_cron( + name=f"intelligent-eval-worker-{i}", + schedule="* * * * *", # Every minute + skill="agenteval-intelligent-worker", + state={"status": "idle"}, + ) + + # Record in DB + cron_db = OpenClawCronPoolDB( + openclaw_cron_id=cron_id, + status="idle", + last_active_at=utc_now(), + ) + session.add(cron_db) + created += 1 + except Exception as exc: + _logger.error(f"Failed to create cron {i}: {exc}") + + session.commit() + _logger.info(f"Initialized cron pool with {created} crons") + return created + + +async def scale_up(count: int, session: Session, client: OpenClawClient) -> int: + """Scale up the pool by creating new crons. + + Args: + count: Number of crons to create + session: Database session + client: OpenClaw client + + Returns: + Number of crons created + """ + # Check current pool size + current_size = len(session.exec(select(OpenClawCronPoolDB)).all()) + if current_size >= MAX_POOL_SIZE: + _logger.warning(f"Pool already at max size ({MAX_POOL_SIZE})") + return 0 + + # Limit count to not exceed max size + count = min(count, MAX_POOL_SIZE - current_size) + + created = 0 + for i in range(count): + try: + cron_id = await client.create_cron( + name=f"intelligent-eval-worker-{current_size + i}", + schedule="* * * * *", + skill="agenteval-intelligent-worker", + state={"status": "idle"}, + ) + + cron_db = OpenClawCronPoolDB( + openclaw_cron_id=cron_id, + status="idle", + last_active_at=utc_now(), + ) + session.add(cron_db) + created += 1 + except Exception as exc: + _logger.error(f"Failed to create cron during scale up: {exc}") + + session.commit() + _logger.info(f"Scaled up pool by {created} crons (total: {current_size + created})") + return created + + +async def scale_down(count: int, session: Session, client: OpenClawClient) -> int: + """Scale down the pool by deleting idle crons. + + Args: + count: Number of crons to delete + session: Database session + client: OpenClaw client + + Returns: + Number of crons deleted + """ + # Check current pool size + current_size = len(session.exec(select(OpenClawCronPoolDB)).all()) + if current_size <= MIN_POOL_SIZE: + _logger.warning(f"Pool already at min size ({MIN_POOL_SIZE})") + return 0 + + # Limit count to not go below min size + count = min(count, current_size - MIN_POOL_SIZE) + + # Find idle crons to delete + idle_crons = session.exec( + select(OpenClawCronPoolDB) + .where(OpenClawCronPoolDB.status == "idle") + .order_by(OpenClawCronPoolDB.last_active_at) + .limit(count) + ).all() + + deleted = 0 + for cron in idle_crons: + try: + await client.delete_cron(cron.openclaw_cron_id) + session.delete(cron) + deleted += 1 + except Exception as exc: + _logger.error(f"Failed to delete cron {cron.openclaw_cron_id}: {exc}") + + session.commit() + _logger.info(f"Scaled down pool by {deleted} crons (total: {current_size - deleted})") + return deleted + + +async def auto_scale(session: Session, client: OpenClawClient) -> tuple[int, int]: + """Automatically scale pool based on load. + + Returns: + (scaled_up, scaled_down) counts + """ + crons = session.exec(select(OpenClawCronPoolDB)).all() + total = len(crons) + + if total == 0: + # Pool not initialized + return (0, 0) + + busy = sum(1 for c in crons if c.status == "busy") + idle = sum(1 for c in crons if c.status == "idle") + + scaled_up = 0 + scaled_down = 0 + + # Scale up if busy/total > threshold and not at max + if busy / total > SCALE_UP_THRESHOLD and total < MAX_POOL_SIZE: + scaled_up = await scale_up(1, session, client) + + # Scale down if idle > min_size * threshold and not at min + elif idle > MIN_POOL_SIZE * SCALE_DOWN_THRESHOLD and total > MIN_POOL_SIZE: + scaled_down = await scale_down(1, session, client) + + return (scaled_up, scaled_down) + + +def get_pool_status(session: Session) -> dict: + """Get current pool status. + + Returns: + Dict with pool stats + """ + crons = session.exec(select(OpenClawCronPoolDB)).all() + total = len(crons) + idle = sum(1 for c in crons if c.status == "idle") + busy = sum(1 for c in crons if c.status == "busy") + stuck = sum(1 for c in crons if c.status == "stuck") + + return { + "total": total, + "idle": idle, + "busy": busy, + "stuck": stuck, + "min_size": MIN_POOL_SIZE, + "max_size": MAX_POOL_SIZE, + } + + +async def sync_cron_states(session: Session, client: OpenClawClient) -> int: + """Sync cron states from OpenClaw to platform DB. + + Returns: + Number of crons synced + """ + # Get all crons from OpenClaw + openclaw_crons = await client.list_crons() + + synced = 0 + for oc_cron in openclaw_crons: + # Find corresponding DB record + db_cron = session.exec( + select(OpenClawCronPoolDB).where(OpenClawCronPoolDB.openclaw_cron_id == oc_cron.id) + ).first() + + if db_cron is None: + # New cron, add to DB + db_cron = OpenClawCronPoolDB( + openclaw_cron_id=oc_cron.id, + status="idle" if oc_cron.enabled else "disabled", + last_active_at=utc_now(), + ) + session.add(db_cron) + synced += 1 + else: + # Update existing record + if oc_cron.state: + new_status = oc_cron.state.get("status", "idle") + if db_cron.status != new_status: + db_cron.status = new_status + db_cron.updated_at = utc_now() + synced += 1 + + session.commit() + return synced + + +def detect_stuck_crons(session: Session) -> list[OpenClawCronPoolDB]: + """Detect stuck crons (busy but not active for > 10 minutes). + + Returns: + List of stuck crons + """ + threshold = utc_now() - timedelta(minutes=STUCK_THRESHOLD_MINUTES) + stuck = session.exec( + select(OpenClawCronPoolDB).where( + OpenClawCronPoolDB.status == "busy", + OpenClawCronPoolDB.last_active_at < threshold, + ) + ).all() + return list(stuck) + + +async def handle_stuck_cron(cron: OpenClawCronPoolDB, session: Session, client: OpenClawClient) -> None: + """Handle a stuck cron: mark as stuck, requeue task, delete cron, create new one. + + Args: + cron: Stuck cron + session: Database session + client: OpenClaw client + """ + from agenteval.intelligent_eval.task_queue import complete_task + + _logger.warning(f"Handling stuck cron {cron.openclaw_cron_id}") + + # Mark as stuck + cron.status = "stuck" + cron.updated_at = utc_now() + + # Requeue task if any + if cron.current_eval_id: + from agenteval.storage.db import IntelligentEvalTaskQueueDB + + task = session.exec( + select(IntelligentEvalTaskQueueDB).where( + IntelligentEvalTaskQueueDB.eval_id == cron.current_eval_id, + IntelligentEvalTaskQueueDB.status == "assigned", + IntelligentEvalTaskQueueDB.assigned_cron_id == cron.openclaw_cron_id, + ) + ).first() + + if task: + # Mark task as failed + complete_task(task.id, False, "Cron stuck", session) + + # Create new task for retry + new_task = IntelligentEvalTaskQueueDB( + eval_id=cron.current_eval_id, + status="pending", + priority=1, # High priority + reason="cron_stuck_retry", + ) + session.add(new_task) + + # Delete stuck cron + try: + await client.delete_cron(cron.openclaw_cron_id) + session.delete(cron) + except Exception as exc: + _logger.error(f"Failed to delete stuck cron {cron.openclaw_cron_id}: {exc}") + + # Create new cron to replace + await scale_up(1, session, client) + + session.commit() diff --git a/backend/agenteval/intelligent_eval/openclaw_client.py b/backend/agenteval/intelligent_eval/openclaw_client.py new file mode 100644 index 0000000..7cd1d23 --- /dev/null +++ b/backend/agenteval/intelligent_eval/openclaw_client.py @@ -0,0 +1,171 @@ +"""OpenClaw CLI client for managing cron jobs. + +Wraps `openclaw automations` commands to create, delete, and list cron jobs. +""" + +import asyncio +import json +import logging +from dataclasses import dataclass +from typing import Optional + +_logger = logging.getLogger("agenteval") + + +@dataclass +class OpenClawCron: + """OpenClaw cron job info.""" + + id: str + name: str + schedule: str + enabled: bool + state: Optional[dict] = None + + +class OpenClawClient: + """Client for OpenClaw CLI commands.""" + + def __init__(self, openclaw_bin: str = "openclaw"): + self.openclaw_bin = openclaw_bin + + async def _run_command(self, *args: str) -> tuple[int, str, str]: + """Run an OpenClaw CLI command. + + Returns: + (returncode, stdout, stderr) + """ + cmd = [self.openclaw_bin] + list(args) + _logger.debug(f"Running OpenClaw command: {' '.join(cmd)}") + + process = await asyncio.create_subprocess_exec( + *cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + stdout, stderr = await process.communicate() + + return ( + process.returncode or 0, + stdout.decode("utf-8"), + stderr.decode("utf-8"), + ) + + async def create_cron( + self, + *, + name: str, + schedule: str, + skill: str, + state: Optional[dict] = None, + ) -> str: + """Create a cron job. + + Args: + name: Cron job name + schedule: Cron schedule expression (e.g., "* * * * *") + skill: Skill to execute + state: Initial state (JSON) + + Returns: + Cron job ID + + Raises: + RuntimeError: If creation fails + """ + args = [ + "automations", + "create", + schedule, + f"--name={name}", + f"--skill={skill}", + ] + + if state: + args.append(f"--state={json.dumps(state)}") + + returncode, stdout, stderr = await self._run_command(*args) + + if returncode != 0: + raise RuntimeError(f"Failed to create cron: {stderr}") + + # Parse cron ID from output + # Expected output format: "Created automation " + lines = stdout.strip().split("\n") + for line in lines: + if "Created automation" in line: + cron_id = line.split()[-1] + _logger.info(f"Created OpenClaw cron {cron_id}: {name}") + return cron_id + + raise RuntimeError(f"Failed to parse cron ID from output: {stdout}") + + async def delete_cron(self, cron_id: str) -> None: + """Delete a cron job. + + Args: + cron_id: Cron job ID + + Raises: + RuntimeError: If deletion fails + """ + returncode, stdout, stderr = await self._run_command( + "automations", + "remove", + cron_id, + ) + + if returncode != 0: + raise RuntimeError(f"Failed to delete cron {cron_id}: {stderr}") + + _logger.info(f"Deleted OpenClaw cron {cron_id}") + + async def list_crons(self) -> list[OpenClawCron]: + """List all cron jobs. + + Returns: + List of cron jobs + + Raises: + RuntimeError: If listing fails + """ + returncode, stdout, stderr = await self._run_command( + "automations", + "list", + "--json", + ) + + if returncode != 0: + raise RuntimeError(f"Failed to list crons: {stderr}") + + try: + data = json.loads(stdout) + crons = [] + for item in data: + crons.append( + OpenClawCron( + id=item["id"], + name=item.get("name", ""), + schedule=item.get("schedule", ""), + enabled=item.get("enabled", True), + state=item.get("state"), + ) + ) + return crons + except (json.JSONDecodeError, KeyError) as e: + raise RuntimeError(f"Failed to parse cron list: {e}") from e + + async def get_cron_state(self, cron_id: str) -> Optional[dict]: + """Get cron job state. + + Args: + cron_id: Cron job ID + + Returns: + State dict, or None if not found + """ + crons = await self.list_crons() + for cron in crons: + if cron.id == cron_id: + return cron.state + return None diff --git a/backend/agenteval/web/app.py b/backend/agenteval/web/app.py index 8fc2116..5db0690 100644 --- a/backend/agenteval/web/app.py +++ b/backend/agenteval/web/app.py @@ -19,6 +19,7 @@ from agenteval.web.routers import ( files, intelligent_evals, model_configs, + openclaw_cron_pool, proxy, reports, runs, @@ -109,6 +110,9 @@ app.include_router(exploration.router, prefix="/api/exploration", tags=["explora app.include_router( intelligent_evals.router, prefix="/api/intelligent-evals", tags=["intelligent-evals"], dependencies=_api_deps ) +app.include_router( + openclaw_cron_pool.router, prefix="/api/openclaw", tags=["openclaw-cron-pool"], dependencies=_api_deps +) app.include_router(reports.router, prefix="/api/reports", tags=["reports"], dependencies=_api_deps) app.include_router(stats.router, prefix="/api/stats", tags=["stats"], dependencies=_api_deps) app.include_router(files.router, prefix="/api/files", tags=["files"], dependencies=_api_deps) diff --git a/backend/agenteval/web/routers/openclaw_cron_pool.py b/backend/agenteval/web/routers/openclaw_cron_pool.py new file mode 100644 index 0000000..532bd07 --- /dev/null +++ b/backend/agenteval/web/routers/openclaw_cron_pool.py @@ -0,0 +1,65 @@ +"""API routes for OpenClaw cron pool management.""" + +from fastapi import APIRouter, Depends, HTTPException +from pydantic import BaseModel, Field +from sqlmodel import Session + +from agenteval.intelligent_eval import cron_pool +from agenteval.intelligent_eval.openclaw_client import OpenClawClient +from agenteval.web.deps import get_db + +router = APIRouter() + + +class ScaleRequest(BaseModel): + target_size: int = Field(ge=1, le=50) + + +@router.get("/cron-pool") +async def get_cron_pool_status(session: Session = Depends(get_db)) -> dict: + """Get cron pool status.""" + status = cron_pool.get_pool_status(session) + return {"pool": status} + + +@router.post("/cron-pool/scale") +async def scale_cron_pool(request: ScaleRequest, session: Session = Depends(get_db)) -> dict: + """Manually scale cron pool to target size.""" + current_status = cron_pool.get_pool_status(session) + current_size = current_status["total"] + target_size = request.target_size + + client = OpenClawClient() + + if target_size > current_size: + # Scale up + count = target_size - current_size + created = await cron_pool.scale_up(count, session, client) + return {"success": True, "scaled_up": created, "current_size": current_size + created} + elif target_size < current_size: + # Scale down + count = current_size - target_size + deleted = await cron_pool.scale_down(count, session, client) + return {"success": True, "scaled_down": deleted, "current_size": current_size - deleted} + else: + return {"success": True, "current_size": current_size, "message": "already at target size"} + + +@router.post("/cron-pool/sync") +async def sync_cron_states(session: Session = Depends(get_db)) -> dict: + """Sync cron states from OpenClaw to platform DB.""" + client = OpenClawClient() + synced = await cron_pool.sync_cron_states(session, client) + return {"success": True, "synced": synced} + + +@router.post("/cron-pool/auto-scale") +async def auto_scale_pool(session: Session = Depends(get_db)) -> dict: + """Trigger auto-scaling based on current load.""" + client = OpenClawClient() + scaled_up, scaled_down = await cron_pool.auto_scale(session, client) + return { + "success": True, + "scaled_up": scaled_up, + "scaled_down": scaled_down, + } diff --git a/tests/integration/test_cron_pool_api.py b/tests/integration/test_cron_pool_api.py new file mode 100644 index 0000000..047422d --- /dev/null +++ b/tests/integration/test_cron_pool_api.py @@ -0,0 +1,185 @@ +"""Integration tests for cron pool API.""" + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from fastapi.testclient import TestClient +from sqlmodel import Session, SQLModel, create_engine + +from agenteval.intelligent_eval.openclaw_client import OpenClawClient +from agenteval.storage.db import 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 OpenClawCronPoolDB # noqa: F401 + + 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 = MagicMock(spec=OpenClawClient) + + counter = {"value": 0} + + async def create_cron_impl(**kwargs): + counter["value"] += 1 + return f"cron-{counter['value']}" + + client.create_cron = AsyncMock(side_effect=create_cron_impl) + client.delete_cron = AsyncMock() + client.list_crons = AsyncMock(return_value=[]) + return client + + +def test_get_cron_pool_status_empty(client: TestClient): + """Test getting pool status when pool is empty.""" + response = client.get("/api/openclaw/cron-pool") + assert response.status_code == 200 + + data = response.json() + assert data["pool"]["total"] == 0 + assert data["pool"]["idle"] == 0 + assert data["pool"]["busy"] == 0 + assert data["pool"]["stuck"] == 0 + assert data["pool"]["min_size"] == 5 + assert data["pool"]["max_size"] == 20 + + +def test_get_cron_pool_status_with_crons(client: TestClient, db_session: Session): + """Test getting pool status with existing crons.""" + # Create crons + for i in range(10): + status = ["idle", "busy", "stuck"][i % 3] + cron = OpenClawCronPoolDB( + openclaw_cron_id=f"cron-{i}", + status=status, + last_active_at=utc_now(), + ) + db_session.add(cron) + db_session.commit() + + response = client.get("/api/openclaw/cron-pool") + assert response.status_code == 200 + + data = response.json() + assert data["pool"]["total"] == 10 + assert data["pool"]["idle"] == 4 + assert data["pool"]["busy"] == 3 + assert data["pool"]["stuck"] == 3 + + +@patch("agenteval.web.routers.openclaw_cron_pool.OpenClawClient") +def test_scale_up_pool(mock_client_class, client: TestClient, db_session: Session, mock_openclaw_client): + """Test manually scaling up the pool.""" + mock_client_class.return_value = mock_openclaw_client + + # Create 5 crons + for i in range(5): + cron = OpenClawCronPoolDB( + openclaw_cron_id=f"existing-cron-{i}", + status="idle", + last_active_at=utc_now(), + ) + db_session.add(cron) + db_session.commit() + + # Scale up to 10 + response = client.post("/api/openclaw/cron-pool/scale", json={"target_size": 10}) + assert response.status_code == 200 + + data = response.json() + assert data["success"] is True + assert data["scaled_up"] == 5 + assert data["current_size"] == 10 + + # Verify crons created + from sqlmodel import select + + crons = db_session.exec(select(OpenClawCronPoolDB)).all() + assert len(crons) == 10 + + +@patch("agenteval.web.routers.openclaw_cron_pool.OpenClawClient") +def test_scale_down_pool(mock_client_class, client: TestClient, db_session: Session, mock_openclaw_client): + """Test manually scaling down the pool.""" + mock_client_class.return_value = mock_openclaw_client + + # Create 15 idle crons + for i in range(15): + cron = OpenClawCronPoolDB( + openclaw_cron_id=f"cron-{i}", + status="idle", + last_active_at=utc_now(), + ) + db_session.add(cron) + db_session.commit() + + # Scale down to 8 + response = client.post("/api/openclaw/cron-pool/scale", json={"target_size": 8}) + assert response.status_code == 200 + + data = response.json() + assert data["success"] is True + assert data["scaled_down"] == 7 + assert data["current_size"] == 8 + + # Verify crons deleted + from sqlmodel import select + + crons = db_session.exec(select(OpenClawCronPoolDB)).all() + assert len(crons) == 8 + + +@patch("agenteval.web.routers.openclaw_cron_pool.OpenClawClient") +def test_scale_pool_no_change(mock_client_class, client: TestClient, db_session: Session): + """Test scaling pool to same size (no change).""" + mock_client_class.return_value = MagicMock() + + # Create 10 crons + for i in range(10): + cron = OpenClawCronPoolDB( + openclaw_cron_id=f"cron-{i}", + status="idle", + last_active_at=utc_now(), + ) + db_session.add(cron) + db_session.commit() + + # Scale to same size + response = client.post("/api/openclaw/cron-pool/scale", json={"target_size": 10}) + assert response.status_code == 200 + + data = response.json() + assert data["success"] is True + assert data["current_size"] == 10 + assert "already at target size" in data["message"] diff --git a/tests/unit/test_cron_pool.py b/tests/unit/test_cron_pool.py new file mode 100644 index 0000000..628ec7c --- /dev/null +++ b/tests/unit/test_cron_pool.py @@ -0,0 +1,295 @@ +"""Unit tests for cron pool management.""" + +from datetime import timedelta +from unittest.mock import AsyncMock, MagicMock + +import pytest +from sqlmodel import Session, select + +from agenteval.intelligent_eval import cron_pool +from agenteval.intelligent_eval.openclaw_client import OpenClawClient, OpenClawCron +from agenteval.storage.db import OpenClawCronPoolDB, utc_now + + +@pytest.fixture() +def mock_openclaw_client(): + """Mock OpenClaw client.""" + client = MagicMock(spec=OpenClawClient) + + # Make create_cron return unique IDs + counter = {"value": 0} + + async def create_cron_impl(**kwargs): + counter["value"] += 1 + return f"cron-{counter['value']}" + + client.create_cron = AsyncMock(side_effect=create_cron_impl) + client.delete_cron = AsyncMock() + client.list_crons = AsyncMock(return_value=[]) + return client + + +async def test_initialize_pool_empty(db_session: Session, mock_openclaw_client): + """Test pool initialization when pool is empty.""" + created = await cron_pool.initialize_pool(db_session, mock_openclaw_client) + assert created == cron_pool.MIN_POOL_SIZE + + # Verify crons created in DB + crons = db_session.exec(select(OpenClawCronPoolDB)).all() + assert len(crons) == cron_pool.MIN_POOL_SIZE + assert all(c.status == "idle" for c in crons) + + +async def test_initialize_pool_already_exists(db_session: Session, mock_openclaw_client): + """Test pool initialization when pool already exists.""" + # Create existing cron + existing = OpenClawCronPoolDB( + openclaw_cron_id="existing-cron", + status="idle", + last_active_at=utc_now(), + ) + db_session.add(existing) + db_session.commit() + + created = await cron_pool.initialize_pool(db_session, mock_openclaw_client) + assert created == 0 + + # Verify no new crons created + crons = db_session.exec(select(OpenClawCronPoolDB)).all() + assert len(crons) == 1 + + +async def test_scale_up(db_session: Session, mock_openclaw_client): + """Test scaling up the pool.""" + created = await cron_pool.scale_up(3, db_session, mock_openclaw_client) + assert created == 3 + + crons = db_session.exec(select(OpenClawCronPoolDB)).all() + assert len(crons) == 3 + + +async def test_scale_up_max_limit(db_session: Session, mock_openclaw_client): + """Test scaling up respects max pool size.""" + # Fill pool to max + for i in range(cron_pool.MAX_POOL_SIZE): + cron = OpenClawCronPoolDB( + openclaw_cron_id=f"cron-{i}", + status="idle", + last_active_at=utc_now(), + ) + db_session.add(cron) + db_session.commit() + + created = await cron_pool.scale_up(5, db_session, mock_openclaw_client) + assert created == 0 + + +async def test_scale_down(db_session: Session, mock_openclaw_client): + """Test scaling down the pool.""" + # Create 10 idle crons + for i in range(10): + cron = OpenClawCronPoolDB( + openclaw_cron_id=f"cron-{i}", + status="idle", + last_active_at=utc_now(), + ) + db_session.add(cron) + db_session.commit() + + deleted = await cron_pool.scale_down(3, db_session, mock_openclaw_client) + assert deleted == 3 + + crons = db_session.exec(select(OpenClawCronPoolDB)).all() + assert len(crons) == 7 + + +async def test_scale_down_min_limit(db_session: Session, mock_openclaw_client): + """Test scaling down respects min pool size.""" + # Create exactly MIN_POOL_SIZE crons + for i in range(cron_pool.MIN_POOL_SIZE): + cron = OpenClawCronPoolDB( + openclaw_cron_id=f"cron-{i}", + status="idle", + last_active_at=utc_now(), + ) + db_session.add(cron) + db_session.commit() + + deleted = await cron_pool.scale_down(3, db_session, mock_openclaw_client) + assert deleted == 0 + + +async def test_scale_down_only_idle(db_session: Session, mock_openclaw_client): + """Test scaling down only deletes idle crons.""" + # Create 5 idle and 5 busy crons + for i in range(5): + idle_cron = OpenClawCronPoolDB( + openclaw_cron_id=f"idle-{i}", + status="idle", + last_active_at=utc_now(), + ) + busy_cron = OpenClawCronPoolDB( + openclaw_cron_id=f"busy-{i}", + status="busy", + last_active_at=utc_now(), + ) + db_session.add_all([idle_cron, busy_cron]) + db_session.commit() + + deleted = await cron_pool.scale_down(3, db_session, mock_openclaw_client) + assert deleted == 3 + + # Verify only idle crons deleted + remaining = db_session.exec(select(OpenClawCronPoolDB)).all() + assert len(remaining) == 7 + assert sum(1 for c in remaining if c.status == "busy") == 5 + assert sum(1 for c in remaining if c.status == "idle") == 2 + + +async def test_auto_scale_up(db_session: Session, mock_openclaw_client): + """Test auto-scaling up when busy/total > 0.8.""" + # Create 10 crons, 9 busy (90% > 80%) + for i in range(10): + status = "busy" if i < 9 else "idle" + cron = OpenClawCronPoolDB( + openclaw_cron_id=f"existing-cron-{i}", + status=status, + last_active_at=utc_now(), + ) + db_session.add(cron) + db_session.commit() + + scaled_up, scaled_down = await cron_pool.auto_scale(db_session, mock_openclaw_client) + assert scaled_up == 1 + assert scaled_down == 0 + + +async def test_auto_scale_down(db_session: Session, mock_openclaw_client): + """Test auto-scaling down when idle > min_size * 2.""" + # Create 15 idle crons (15 > 5 * 2) + for i in range(15): + cron = OpenClawCronPoolDB( + openclaw_cron_id=f"cron-{i}", + status="idle", + last_active_at=utc_now(), + ) + db_session.add(cron) + db_session.commit() + + scaled_up, scaled_down = await cron_pool.auto_scale(db_session, mock_openclaw_client) + assert scaled_up == 0 + assert scaled_down == 1 + + +def test_get_pool_status(db_session: Session): + """Test getting pool status.""" + # Create mixed crons + for i in range(10): + status = ["idle", "busy", "stuck"][i % 3] + cron = OpenClawCronPoolDB( + openclaw_cron_id=f"cron-{i}", + status=status, + last_active_at=utc_now(), + ) + db_session.add(cron) + db_session.commit() + + status = cron_pool.get_pool_status(db_session) + assert status["total"] == 10 + assert status["idle"] == 4 # 0, 3, 6, 9 + assert status["busy"] == 3 # 1, 4, 7 + assert status["stuck"] == 3 # 2, 5, 8 + assert status["min_size"] == cron_pool.MIN_POOL_SIZE + assert status["max_size"] == cron_pool.MAX_POOL_SIZE + + +async def test_sync_cron_states(db_session: Session, mock_openclaw_client): + """Test syncing cron states from OpenClaw.""" + # Mock OpenClaw returning 2 crons + mock_openclaw_client.list_crons.return_value = [ + OpenClawCron(id="cron-1", name="worker-1", schedule="* * * * *", enabled=True, state={"status": "busy"}), + OpenClawCron(id="cron-2", name="worker-2", schedule="* * * * *", enabled=True, state={"status": "idle"}), + ] + + synced = await cron_pool.sync_cron_states(db_session, mock_openclaw_client) + assert synced == 2 + + # Verify crons added to DB + crons = db_session.exec(select(OpenClawCronPoolDB)).all() + assert len(crons) == 2 + + +def test_detect_stuck_crons(db_session: Session): + """Test detecting stuck crons.""" + # Create crons with different last_active_at + now = utc_now() + active_cron = OpenClawCronPoolDB( + openclaw_cron_id="active", + status="busy", + last_active_at=now, + ) + stuck_cron = OpenClawCronPoolDB( + openclaw_cron_id="stuck", + status="busy", + last_active_at=now - timedelta(minutes=15), # 15 minutes ago + ) + idle_cron = OpenClawCronPoolDB( + openclaw_cron_id="idle", + status="idle", + last_active_at=now - timedelta(minutes=20), + ) + db_session.add_all([active_cron, stuck_cron, idle_cron]) + db_session.commit() + + stuck = cron_pool.detect_stuck_crons(db_session) + assert len(stuck) == 1 + assert stuck[0].openclaw_cron_id == "stuck" + + +async def test_handle_stuck_cron(db_session: Session, mock_openclaw_client): + """Test handling a stuck cron.""" + from agenteval.storage.db import IntelligentEvalTaskQueueDB + + # 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() + + # Handle stuck cron + await cron_pool.handle_stuck_cron(cron, db_session, mock_openclaw_client) + + # Verify task marked as failed + db_session.refresh(task) + assert task.status == "failed" + assert task.error == "Cron stuck" + + # Verify new task created for retry + 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" + assert new_task.priority == 1 + + # Verify cron deleted + deleted_cron = db_session.exec( + select(OpenClawCronPoolDB).where(OpenClawCronPoolDB.openclaw_cron_id == "stuck-cron") + ).first() + assert deleted_cron is None