diff --git a/backend/agenteval/intelligent_eval/alerts.py b/backend/agenteval/intelligent_eval/alerts.py new file mode 100644 index 0000000..ecc2c4f --- /dev/null +++ b/backend/agenteval/intelligent_eval/alerts.py @@ -0,0 +1,217 @@ +"""Alert rules and notifications for cron pool (告警规则和通知). + +Alert rules: +- Pool utilization > 90% for 10 minutes +- Task backlog > 50 +- Stuck rate > 10% + +Notifications: +- Log alerts +- Optional webhook notifications +""" + +import json +import logging +from datetime import datetime, timedelta +from typing import Any, Optional + +import httpx +from sqlmodel import Session, SQLModel, Field, select + +from agenteval.intelligent_eval.metrics import ( + calculate_pool_utilization, + calculate_stuck_rate, + calculate_task_backlog, +) +from agenteval.storage.db import utc_now + +_logger = logging.getLogger("agenteval") + + +class AlertHistoryDB(SQLModel, table=True): + """Alert history database record.""" + + __tablename__ = "cron_pool_alert_history" + + id: Optional[str] = Field(default=None, primary_key=True) + alert_type: str = Field(index=True) # pool_utilization / task_backlog / stuck_rate + severity: str = Field(index=True) # warning / critical + message: str + metric_value: float + threshold: float + created_at: datetime = Field(default_factory=utc_now) + resolved_at: Optional[datetime] = None + webhook_sent: bool = False + + +class AlertRule: + """Alert rule definition.""" + + def __init__( + self, + name: str, + metric_func, + threshold: float, + severity: str, + duration_minutes: int = 0, + ): + self.name = name + self.metric_func = metric_func + self.threshold = threshold + self.severity = severity + self.duration_minutes = duration_minutes + self.triggered_at: Optional[datetime] = None + + +class AlertManager: + """Manages alert rules and notifications.""" + + def __init__(self, session: Session, webhook_url: Optional[str] = None): + self.session = session + self.webhook_url = webhook_url + self.rules = [ + AlertRule( + name="pool_utilization", + metric_func=calculate_pool_utilization, + threshold=0.9, + severity="warning", + duration_minutes=10, + ), + AlertRule( + name="task_backlog", + metric_func=calculate_task_backlog, + threshold=50, + severity="warning", + duration_minutes=0, + ), + AlertRule( + name="stuck_rate", + metric_func=calculate_stuck_rate, + threshold=0.1, + severity="critical", + duration_minutes=0, + ), + ] + + def check_rules(self) -> list[AlertHistoryDB]: + """Check all alert rules and create alerts if triggered. + + Returns: + List of newly created alerts + """ + alerts = [] + + for rule in self.rules: + metric_value = rule.metric_func(self.session) + + # Check if threshold exceeded + if metric_value > rule.threshold: + # Check duration requirement + if rule.duration_minutes > 0: + if rule.triggered_at is None: + rule.triggered_at = utc_now() + continue + + duration = (utc_now() - rule.triggered_at).total_seconds() / 60 + if duration < rule.duration_minutes: + continue + else: + rule.triggered_at = utc_now() + + # Create alert + alert = self._create_alert(rule, metric_value) + alerts.append(alert) + + # Send webhook notification + if self.webhook_url: + self._send_webhook(alert) + else: + # Reset trigger time + rule.triggered_at = None + + return alerts + + def _create_alert(self, rule: AlertRule, metric_value: float) -> AlertHistoryDB: + """Create an alert history record.""" + message = f"{rule.name}: {metric_value:.2f} exceeds threshold {rule.threshold}" + + alert = AlertHistoryDB( + id=f"alert-{utc_now().timestamp()}", + alert_type=rule.name, + severity=rule.severity, + message=message, + metric_value=metric_value, + threshold=rule.threshold, + ) + + self.session.add(alert) + self.session.commit() + self.session.refresh(alert) + + _logger.warning(f"Alert triggered: {message}") + + return alert + + def _send_webhook(self, alert: AlertHistoryDB) -> None: + """Send webhook notification.""" + if not self.webhook_url: + return + + try: + payload = { + "alert_id": alert.id, + "alert_type": alert.alert_type, + "severity": alert.severity, + "message": alert.message, + "metric_value": alert.metric_value, + "threshold": alert.threshold, + "timestamp": alert.created_at.isoformat(), + } + + response = httpx.post(self.webhook_url, json=payload, timeout=5.0) + response.raise_for_status() + + alert.webhook_sent = True + self.session.commit() + + _logger.info(f"Webhook sent for alert {alert.id}") + except Exception as e: + _logger.error(f"Failed to send webhook: {e}") + + def get_alert_history(self, limit: int = 100) -> list[AlertHistoryDB]: + """Get alert history. + + Returns: + List of alerts, ordered by created_at descending + """ + alerts = self.session.exec( + select(AlertHistoryDB) + .order_by(AlertHistoryDB.created_at.desc()) + .limit(limit) + ).all() + return list(alerts) + + def get_unresolved_alerts(self) -> list[AlertHistoryDB]: + """Get unresolved alerts. + + Returns: + List of unresolved alerts + """ + alerts = self.session.exec( + select(AlertHistoryDB).where(AlertHistoryDB.resolved_at.is_(None)) + ).all() + return list(alerts) + + def resolve_alert(self, alert_id: str) -> bool: + """Resolve an alert. + + Returns: + True if alert was resolved, False if not found + """ + alert = self.session.get(AlertHistoryDB, alert_id) + if alert is None: + return False + + alert.resolved_at = utc_now() + self.session.commit() + return True diff --git a/backend/agenteval/intelligent_eval/metrics.py b/backend/agenteval/intelligent_eval/metrics.py new file mode 100644 index 0000000..fd7cc7d --- /dev/null +++ b/backend/agenteval/intelligent_eval/metrics.py @@ -0,0 +1,118 @@ +"""Metrics calculation for cron pool monitoring (监控指标). + +Calculates: +- Pool utilization (busy/total) +- Task backlog (pending tasks count) +- Stuck rate (stuck/total) +- Average task processing time +- Eval completion rate +""" + +from datetime import datetime, timedelta +from typing import Optional + +from sqlmodel import Session, select + +from agenteval.intelligent_eval.models import IntelligentEvalStatus +from agenteval.storage.db import ( + IntelligentEvalDB, + IntelligentEvalTaskQueueDB, + OpenClawCronPoolDB, + utc_now, +) + + +def calculate_pool_utilization(session: Session) -> float: + """Calculate pool utilization (busy/total). + + Returns: + Utilization rate (0.0 to 1.0) + """ + crons = session.exec(select(OpenClawCronPoolDB)).all() + if not crons: + return 0.0 + + busy = sum(1 for c in crons if c.status == "busy") + return busy / len(crons) + + +def calculate_task_backlog(session: Session) -> int: + """Calculate task backlog (pending tasks count). + + Returns: + Number of pending tasks + """ + pending = session.exec( + select(IntelligentEvalTaskQueueDB).where(IntelligentEvalTaskQueueDB.status == "pending") + ).all() + return len(pending) + + +def calculate_stuck_rate(session: Session) -> float: + """Calculate stuck rate (stuck/total). + + Returns: + Stuck rate (0.0 to 1.0) + """ + crons = session.exec(select(OpenClawCronPoolDB)).all() + if not crons: + return 0.0 + + stuck = sum(1 for c in crons if c.status == "stuck") + return stuck / len(crons) + + +def calculate_avg_processing_time(session: Session) -> Optional[float]: + """Calculate average task processing time (in seconds). + + Returns: + Average processing time in seconds, or None if no completed tasks + """ + completed_tasks = session.exec( + select(IntelligentEvalTaskQueueDB).where( + IntelligentEvalTaskQueueDB.status == "completed", + IntelligentEvalTaskQueueDB.assigned_at.isnot(None), + IntelligentEvalTaskQueueDB.completed_at.isnot(None), + ) + ).all() + + if not completed_tasks: + return None + + total_seconds = 0 + for task in completed_tasks: + if task.assigned_at and task.completed_at: + duration = (task.completed_at - task.assigned_at).total_seconds() + total_seconds += duration + + return total_seconds / len(completed_tasks) + + +def calculate_eval_completion_rate(session: Session) -> float: + """Calculate evaluation completion rate. + + Returns: + Completion rate (0.0 to 1.0) + """ + all_evals = session.exec(select(IntelligentEvalDB)).all() + if not all_evals: + return 0.0 + + completed = sum(1 for e in all_evals if e.status == IntelligentEvalStatus.COMPLETED.value) + return completed / len(all_evals) + + +def get_all_metrics(session: Session) -> dict: + """Get all metrics. + + Returns: + Dict with all metrics + """ + return { + "pool_utilization": calculate_pool_utilization(session), + "task_backlog": calculate_task_backlog(session), + "stuck_rate": calculate_stuck_rate(session), + "avg_processing_time_seconds": calculate_avg_processing_time(session), + "eval_completion_rate": calculate_eval_completion_rate(session), + "timestamp": utc_now().isoformat(), + } diff --git a/backend/agenteval/web/routers/openclaw_cron_pool.py b/backend/agenteval/web/routers/openclaw_cron_pool.py index 0cdef94..ec1ebf1 100644 --- a/backend/agenteval/web/routers/openclaw_cron_pool.py +++ b/backend/agenteval/web/routers/openclaw_cron_pool.py @@ -100,3 +100,86 @@ async def report_heartbeat( session.commit() return {"success": True} + + +@router.get("/cron-pool/metrics") +async def get_cron_pool_metrics(session: Session = Depends(get_db)) -> dict: + """Get cron pool metrics.""" + from agenteval.intelligent_eval.metrics import get_all_metrics + + metrics = get_all_metrics(session) + return {"metrics": metrics} + + +@router.post("/cron-pool/check-alerts") +async def check_alerts(session: Session = Depends(get_db)) -> dict: + """Check alert rules and create alerts if triggered.""" + from agenteval.intelligent_eval.alerts import AlertManager + + manager = AlertManager(session) + alerts = manager.check_rules() + + return { + "success": True, + "alerts_triggered": len(alerts), + "alerts": [ + { + "id": alert.id, + "alert_type": alert.alert_type, + "severity": alert.severity, + "message": alert.message, + "metric_value": alert.metric_value, + "threshold": alert.threshold, + "created_at": alert.created_at.isoformat(), + } + for alert in alerts + ], + } + + +@router.get("/cron-pool/alerts") +async def get_alert_history( + limit: int = 100, + unresolved_only: bool = False, + session: Session = Depends(get_db), +) -> dict: + """Get alert history.""" + from agenteval.intelligent_eval.alerts import AlertManager + + manager = AlertManager(session) + + if unresolved_only: + alerts = manager.get_unresolved_alerts() + else: + alerts = manager.get_alert_history(limit=limit) + + return { + "alerts": [ + { + "id": alert.id, + "alert_type": alert.alert_type, + "severity": alert.severity, + "message": alert.message, + "metric_value": alert.metric_value, + "threshold": alert.threshold, + "created_at": alert.created_at.isoformat(), + "resolved_at": alert.resolved_at.isoformat() if alert.resolved_at else None, + "webhook_sent": alert.webhook_sent, + } + for alert in alerts + ] + } + + +@router.post("/cron-pool/alerts/{alert_id}/resolve") +async def resolve_alert(alert_id: str, session: Session = Depends(get_db)) -> dict: + """Resolve an alert.""" + from agenteval.intelligent_eval.alerts import AlertManager + + manager = AlertManager(session) + resolved = manager.resolve_alert(alert_id) + + if not resolved: + raise HTTPException(status_code=404, detail=f"alert {alert_id} not found") + + return {"success": True} diff --git a/migrations/versions/c8f3e9a2b4d1_add_alert_history_table.py b/migrations/versions/c8f3e9a2b4d1_add_alert_history_table.py new file mode 100644 index 0000000..059b18a --- /dev/null +++ b/migrations/versions/c8f3e9a2b4d1_add_alert_history_table.py @@ -0,0 +1,48 @@ +"""add alert history table + +Revision ID: c8f3e9a2b4d1 +Revises: b72debf55c3b +Create Date: 2026-08-12 10:00:00.000000 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +import sqlmodel + + +# revision identifiers, used by Alembic. +revision: str = 'c8f3e9a2b4d1' +down_revision: Union[str, Sequence[str], None] = 'b72debf55c3b' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Upgrade schema.""" + op.create_table( + 'cron_pool_alert_history', + sa.Column('id', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('alert_type', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('severity', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('message', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('metric_value', sa.Float(), nullable=False), + sa.Column('threshold', sa.Float(), nullable=False), + sa.Column('created_at', sa.DateTime(), nullable=False), + sa.Column('resolved_at', sa.DateTime(), nullable=True), + sa.Column('webhook_sent', sa.Boolean(), nullable=False, default=False), + sa.PrimaryKeyConstraint('id') + ) + with op.batch_alter_table('cron_pool_alert_history', schema=None) as batch_op: + batch_op.create_index('ix_cron_pool_alert_history_alert_type', ['alert_type'], unique=False) + batch_op.create_index('ix_cron_pool_alert_history_severity', ['severity'], unique=False) + + +def downgrade() -> None: + """Downgrade schema.""" + with op.batch_alter_table('cron_pool_alert_history', schema=None) as batch_op: + batch_op.drop_index('ix_cron_pool_alert_history_severity') + batch_op.drop_index('ix_cron_pool_alert_history_alert_type') + + op.drop_table('cron_pool_alert_history') diff --git a/tests/integration/test_campaigns_api.py b/tests/integration/test_campaigns_api.py index 93bc811..95f1880 100644 --- a/tests/integration/test_campaigns_api.py +++ b/tests/integration/test_campaigns_api.py @@ -488,6 +488,7 @@ def test_exploration_config_migration_on_existing_db(tmp_path, monkeypatch): connection.execute(text("DROP TABLE IF EXISTS intelligent_eval_decision_logs")) connection.execute(text("DROP TABLE IF EXISTS intelligent_eval_task_queue")) connection.execute(text("DROP TABLE IF EXISTS openclaw_cron_pool")) + connection.execute(text("DROP TABLE IF EXISTS cron_pool_alert_history")) connection.execute(text("DROP TABLE IF EXISTS exploration_sessions")) connection.execute(text("DROP TABLE IF EXISTS exploration_messages")) connection.execute(text("ALTER TABLE campaigns DROP COLUMN exploration_seeds")) diff --git a/tests/integration/test_exploration_api.py b/tests/integration/test_exploration_api.py index 2e9e6da..cff2b52 100644 --- a/tests/integration/test_exploration_api.py +++ b/tests/integration/test_exploration_api.py @@ -477,6 +477,7 @@ def test_exploration_migration_on_existing_db(tmp_path, monkeypatch): connection.execute(text("DROP TABLE IF EXISTS intelligent_eval_decision_logs")) connection.execute(text("DROP TABLE IF EXISTS intelligent_eval_task_queue")) connection.execute(text("DROP TABLE IF EXISTS openclaw_cron_pool")) + connection.execute(text("DROP TABLE IF EXISTS cron_pool_alert_history")) connection.execute(text("DROP TABLE IF EXISTS exploration_sessions")) connection.execute(text("DROP TABLE IF EXISTS exploration_messages")) connection.execute(text("ALTER TABLE campaigns DROP COLUMN exploration_seeds")) diff --git a/tests/integration/test_exploration_patrol_api.py b/tests/integration/test_exploration_patrol_api.py index 5a4f1dc..d10df04 100644 --- a/tests/integration/test_exploration_patrol_api.py +++ b/tests/integration/test_exploration_patrol_api.py @@ -196,6 +196,7 @@ async def test_patrol_migration_column_on_existing_db(tmp_path, monkeypatch): connection.execute(text("DROP TABLE IF EXISTS intelligent_eval_decision_logs")) connection.execute(text("DROP TABLE IF EXISTS intelligent_eval_task_queue")) connection.execute(text("DROP TABLE IF EXISTS openclaw_cron_pool")) + connection.execute(text("DROP TABLE IF EXISTS cron_pool_alert_history")) connection.execute(text("DROP TABLE IF EXISTS exploration_sessions")) connection.execute(text("DROP TABLE IF EXISTS exploration_messages")) connection.execute(text("ALTER TABLE campaigns DROP COLUMN exploration_seeds")) diff --git a/tests/integration/test_metrics_alerts_api.py b/tests/integration/test_metrics_alerts_api.py new file mode 100644 index 0000000..b43647a --- /dev/null +++ b/tests/integration/test_metrics_alerts_api.py @@ -0,0 +1,278 @@ +"""Integration tests for metrics and alerts API.""" + +from datetime import timedelta + +import pytest +from fastapi.testclient import TestClient +from sqlmodel import Session, SQLModel, create_engine, select +from unittest.mock import patch, MagicMock + +from agenteval.intelligent_eval.models import IntelligentEvalStatus +from agenteval.storage.db import ( + IntelligentEvalDB, + 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 + IntelligentEvalDB, + IntelligentEvalTaskQueueDB, + OpenClawCronPoolDB, + ) + from agenteval.intelligent_eval.alerts import AlertHistoryDB + + 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]()) + + +def test_get_metrics_api(client: TestClient, db_session: Session): + """Test getting metrics via API.""" + # Create some test data + for i in range(5): + cron = OpenClawCronPoolDB( + openclaw_cron_id=f"cron-{i}", + status="busy" if i < 3 else "idle", + last_active_at=utc_now(), + ) + db_session.add(cron) + + db_session.commit() + + response = client.get("/api/openclaw/cron-pool/metrics") + assert response.status_code == 200 + + data = response.json() + assert "metrics" in data + assert "pool_utilization" in data["metrics"] + assert "task_backlog" in data["metrics"] + assert "stuck_rate" in data["metrics"] + assert "avg_processing_time_seconds" in data["metrics"] + assert "eval_completion_rate" in data["metrics"] + + assert data["metrics"]["pool_utilization"] == 0.6 + + +def test_check_alerts_api_no_alerts(client: TestClient, db_session: Session): + """Test checking alerts when no rules are triggered.""" + # Create healthy state + for i in range(5): + cron = OpenClawCronPoolDB( + openclaw_cron_id=f"cron-{i}", + status="idle", + last_active_at=utc_now(), + ) + db_session.add(cron) + + db_session.commit() + + response = client.post("/api/openclaw/cron-pool/check-alerts") + assert response.status_code == 200 + + data = response.json() + assert data["success"] is True + assert data["alerts_triggered"] == 0 + + +def test_check_alerts_api_with_alerts(client: TestClient, db_session: Session): + """Test checking alerts when rules are triggered.""" + # Create high backlog + for i in range(60): + task = IntelligentEvalTaskQueueDB( + eval_id=f"eval-{i}", + status="pending", + priority=1, + reason="slot_due", + ) + db_session.add(task) + + db_session.commit() + + response = client.post("/api/openclaw/cron-pool/check-alerts") + assert response.status_code == 200 + + data = response.json() + assert data["success"] is True + assert data["alerts_triggered"] > 0 + + # Should have task_backlog alert + backlog_alerts = [a for a in data["alerts"] if a["alert_type"] == "task_backlog"] + assert len(backlog_alerts) == 1 + assert backlog_alerts[0]["metric_value"] == 60 + + +def test_get_alert_history_api(client: TestClient, db_session: Session): + """Test getting alert history via API.""" + # Create some alerts + from agenteval.intelligent_eval.alerts import AlertHistoryDB + + for i in range(3): + alert = AlertHistoryDB( + id=f"alert-{i}", + alert_type="task_backlog", + severity="warning", + message=f"Test alert {i}", + metric_value=50 + i, + threshold=50, + ) + db_session.add(alert) + + db_session.commit() + + response = client.get("/api/openclaw/cron-pool/alerts") + assert response.status_code == 200 + + data = response.json() + assert "alerts" in data + assert len(data["alerts"]) == 3 + + +def test_get_alert_history_unresolved_only(client: TestClient, db_session: Session): + """Test getting unresolved alerts only.""" + from agenteval.intelligent_eval.alerts import AlertHistoryDB + + # Create resolved and unresolved alerts + alert1 = AlertHistoryDB( + id="alert-1", + alert_type="task_backlog", + severity="warning", + message="Test alert 1", + metric_value=60, + threshold=50, + ) + alert2 = AlertHistoryDB( + id="alert-2", + alert_type="stuck_rate", + severity="critical", + message="Test alert 2", + metric_value=0.2, + threshold=0.1, + resolved_at=utc_now(), + ) + + db_session.add_all([alert1, alert2]) + db_session.commit() + + response = client.get("/api/openclaw/cron-pool/alerts?unresolved_only=true") + assert response.status_code == 200 + + data = response.json() + assert len(data["alerts"]) == 1 + assert data["alerts"][0]["id"] == "alert-1" + + +def test_resolve_alert_api(client: TestClient, db_session: Session): + """Test resolving an alert via API.""" + from agenteval.intelligent_eval.alerts import AlertHistoryDB + + alert = AlertHistoryDB( + id="alert-1", + alert_type="task_backlog", + severity="warning", + message="Test alert", + metric_value=60, + threshold=50, + ) + db_session.add(alert) + db_session.commit() + + response = client.post("/api/openclaw/cron-pool/alerts/alert-1/resolve") + assert response.status_code == 200 + + data = response.json() + assert data["success"] is True + + # Verify alert is resolved + db_session.refresh(alert) + assert alert.resolved_at is not None + + +def test_resolve_nonexistent_alert_api(client: TestClient): + """Test resolving a non-existent alert.""" + response = client.post("/api/openclaw/cron-pool/alerts/nonexistent/resolve") + assert response.status_code == 404 + + +def test_end_to_end_metrics_and_alerts(client: TestClient, db_session: Session): + """Test end-to-end metrics and alerts flow.""" + # Create high utilization state + for i in range(19): + cron = OpenClawCronPoolDB( + openclaw_cron_id=f"busy-{i}", + status="busy", + last_active_at=utc_now(), + ) + db_session.add(cron) + + cron = OpenClawCronPoolDB( + openclaw_cron_id="idle-0", + status="idle", + last_active_at=utc_now(), + ) + db_session.add(cron) + + # Create high backlog + for i in range(60): + task = IntelligentEvalTaskQueueDB( + eval_id=f"eval-{i}", + status="pending", + priority=1, + reason="slot_due", + ) + db_session.add(task) + + db_session.commit() + + # Get metrics + response = client.get("/api/openclaw/cron-pool/metrics") + assert response.status_code == 200 + metrics_data = response.json()["metrics"] + assert metrics_data["pool_utilization"] == 0.95 + assert metrics_data["task_backlog"] == 60 + + # Check alerts + response = client.post("/api/openclaw/cron-pool/check-alerts") + assert response.status_code == 200 + alerts_data = response.json() + assert alerts_data["alerts_triggered"] > 0 + + # Get alert history + response = client.get("/api/openclaw/cron-pool/alerts") + assert response.status_code == 200 + history_data = response.json() + assert len(history_data["alerts"]) > 0 + + # Resolve first alert + if history_data["alerts"]: + alert_id = history_data["alerts"][0]["id"] + response = client.post(f"/api/openclaw/cron-pool/alerts/{alert_id}/resolve") + assert response.status_code == 200 diff --git a/tests/unit/test_alerts.py b/tests/unit/test_alerts.py new file mode 100644 index 0000000..cf6d245 --- /dev/null +++ b/tests/unit/test_alerts.py @@ -0,0 +1,273 @@ +"""Unit tests for alert rules and notifications.""" + +from datetime import timedelta +from unittest.mock import MagicMock, patch + +import pytest +from sqlmodel import Session, select + +from agenteval.intelligent_eval.alerts import AlertHistoryDB, AlertManager +from agenteval.storage.db import ( + IntelligentEvalTaskQueueDB, + OpenClawCronPoolDB, + utc_now, +) + + +def test_alert_manager_check_rules_no_alerts(db_session: Session): + """Test alert manager when no rules are triggered.""" + # Create healthy state: low utilization, low backlog, no stuck + for i in range(5): + cron = OpenClawCronPoolDB( + openclaw_cron_id=f"cron-{i}", + status="idle", + last_active_at=utc_now(), + ) + db_session.add(cron) + + db_session.commit() + + manager = AlertManager(db_session) + alerts = manager.check_rules() + + assert len(alerts) == 0 + + +def test_alert_manager_task_backlog_alert(db_session: Session): + """Test alert manager triggers task backlog alert.""" + # Create high backlog: 60 pending tasks + for i in range(60): + task = IntelligentEvalTaskQueueDB( + eval_id=f"eval-{i}", + status="pending", + priority=1, + reason="slot_due", + ) + db_session.add(task) + + db_session.commit() + + manager = AlertManager(db_session) + alerts = manager.check_rules() + + # Should trigger task_backlog alert + backlog_alerts = [a for a in alerts if a.alert_type == "task_backlog"] + assert len(backlog_alerts) == 1 + assert backlog_alerts[0].metric_value == 60 + assert backlog_alerts[0].threshold == 50 + assert backlog_alerts[0].severity == "warning" + + +def test_alert_manager_stuck_rate_alert(db_session: Session): + """Test alert manager triggers stuck rate alert.""" + # Create high stuck rate: 3 stuck out of 10 + for i in range(3): + cron = OpenClawCronPoolDB( + openclaw_cron_id=f"stuck-{i}", + status="stuck", + last_active_at=utc_now(), + ) + db_session.add(cron) + + for i in range(7): + cron = OpenClawCronPoolDB( + openclaw_cron_id=f"active-{i}", + status="busy", + last_active_at=utc_now(), + ) + db_session.add(cron) + + db_session.commit() + + manager = AlertManager(db_session) + alerts = manager.check_rules() + + # Should trigger stuck_rate alert + stuck_alerts = [a for a in alerts if a.alert_type == "stuck_rate"] + assert len(stuck_alerts) == 1 + assert stuck_alerts[0].metric_value == 0.3 + assert stuck_alerts[0].threshold == 0.1 + assert stuck_alerts[0].severity == "critical" + + +def test_alert_manager_pool_utilization_with_duration(db_session: Session): + """Test alert manager respects duration requirement for pool utilization.""" + # Create high utilization: 19 busy out of 20 + for i in range(19): + cron = OpenClawCronPoolDB( + openclaw_cron_id=f"busy-{i}", + status="busy", + last_active_at=utc_now(), + ) + db_session.add(cron) + + cron = OpenClawCronPoolDB( + openclaw_cron_id="idle-0", + status="idle", + last_active_at=utc_now(), + ) + db_session.add(cron) + + db_session.commit() + + manager = AlertManager(db_session) + + # First check: should not trigger (duration not met) + alerts1 = manager.check_rules() + utilization_alerts1 = [a for a in alerts1 if a.alert_type == "pool_utilization"] + assert len(utilization_alerts1) == 0 + + # Simulate time passing (10 minutes) + # In real scenario, this would be checked over time + # For testing, we just verify the logic exists + + +def test_alert_manager_get_alert_history(db_session: Session): + """Test getting alert history.""" + # Create some alerts + for i in range(5): + alert = AlertHistoryDB( + id=f"alert-{i}", + alert_type="task_backlog", + severity="warning", + message=f"Test alert {i}", + metric_value=50 + i, + threshold=50, + ) + db_session.add(alert) + + db_session.commit() + + manager = AlertManager(db_session) + alerts = manager.get_alert_history(limit=3) + + assert len(alerts) == 3 + # Should be ordered by created_at descending + assert alerts[0].id == "alert-4" + + +def test_alert_manager_get_unresolved_alerts(db_session: Session): + """Test getting unresolved alerts.""" + # Create resolved and unresolved alerts + alert1 = AlertHistoryDB( + id="alert-1", + alert_type="task_backlog", + severity="warning", + message="Test alert 1", + metric_value=60, + threshold=50, + ) + alert2 = AlertHistoryDB( + id="alert-2", + alert_type="stuck_rate", + severity="critical", + message="Test alert 2", + metric_value=0.2, + threshold=0.1, + resolved_at=utc_now(), + ) + alert3 = AlertHistoryDB( + id="alert-3", + alert_type="pool_utilization", + severity="warning", + message="Test alert 3", + metric_value=0.95, + threshold=0.9, + ) + + db_session.add_all([alert1, alert2, alert3]) + db_session.commit() + + manager = AlertManager(db_session) + unresolved = manager.get_unresolved_alerts() + + assert len(unresolved) == 2 + assert all(a.resolved_at is None for a in unresolved) + + +def test_alert_manager_resolve_alert(db_session: Session): + """Test resolving an alert.""" + alert = AlertHistoryDB( + id="alert-1", + alert_type="task_backlog", + severity="warning", + message="Test alert", + metric_value=60, + threshold=50, + ) + db_session.add(alert) + db_session.commit() + + manager = AlertManager(db_session) + resolved = manager.resolve_alert("alert-1") + + assert resolved is True + + db_session.refresh(alert) + assert alert.resolved_at is not None + + +def test_alert_manager_resolve_nonexistent_alert(db_session: Session): + """Test resolving a non-existent alert.""" + manager = AlertManager(db_session) + resolved = manager.resolve_alert("nonexistent") + + assert resolved is False + + +def test_alert_manager_webhook_notification(db_session: Session): + """Test webhook notification.""" + # Create high backlog to trigger alert + for i in range(60): + task = IntelligentEvalTaskQueueDB( + eval_id=f"eval-{i}", + status="pending", + priority=1, + reason="slot_due", + ) + db_session.add(task) + + db_session.commit() + + # Mock webhook + with patch("httpx.post") as mock_post: + mock_post.return_value.status_code = 200 + mock_post.return_value.raise_for_status = MagicMock() + + manager = AlertManager(db_session, webhook_url="https://example.com/webhook") + alerts = manager.check_rules() + + # Should have triggered alerts + assert len(alerts) > 0 + + # Should have called webhook + assert mock_post.called + + +def test_alert_manager_webhook_failure(db_session: Session): + """Test webhook notification failure.""" + # Create high backlog to trigger alert + for i in range(60): + task = IntelligentEvalTaskQueueDB( + eval_id=f"eval-{i}", + status="pending", + priority=1, + reason="slot_due", + ) + db_session.add(task) + + db_session.commit() + + # Mock webhook failure + with patch("httpx.post") as mock_post: + mock_post.side_effect = Exception("Webhook failed") + + manager = AlertManager(db_session, webhook_url="https://example.com/webhook") + alerts = manager.check_rules() + + # Should still create alerts even if webhook fails + assert len(alerts) > 0 + + # Webhook should not be marked as sent + for alert in alerts: + assert alert.webhook_sent is False diff --git a/tests/unit/test_metrics_calculation.py b/tests/unit/test_metrics_calculation.py new file mode 100644 index 0000000..e79ca36 --- /dev/null +++ b/tests/unit/test_metrics_calculation.py @@ -0,0 +1,213 @@ +"""Unit tests for metrics calculation.""" + +from datetime import timedelta + +import pytest +from sqlmodel import Session + +from agenteval.intelligent_eval import metrics +from agenteval.intelligent_eval.models import IntelligentEvalStatus +from agenteval.storage.db import ( + IntelligentEvalDB, + IntelligentEvalTaskQueueDB, + OpenClawCronPoolDB, + utc_now, +) + + +def test_calculate_pool_utilization_empty(db_session: Session): + """Test pool utilization when no crons exist.""" + utilization = metrics.calculate_pool_utilization(db_session) + assert utilization == 0.0 + + +def test_calculate_pool_utilization(db_session: Session): + """Test pool utilization calculation.""" + # Create 10 crons: 6 busy, 4 idle + for i in range(6): + cron = OpenClawCronPoolDB( + openclaw_cron_id=f"busy-{i}", + status="busy", + last_active_at=utc_now(), + ) + db_session.add(cron) + + for i in range(4): + cron = OpenClawCronPoolDB( + openclaw_cron_id=f"idle-{i}", + status="idle", + last_active_at=utc_now(), + ) + db_session.add(cron) + + db_session.commit() + + utilization = metrics.calculate_pool_utilization(db_session) + assert utilization == 0.6 + + +def test_calculate_task_backlog_empty(db_session: Session): + """Test task backlog when no tasks exist.""" + backlog = metrics.calculate_task_backlog(db_session) + assert backlog == 0 + + +def test_calculate_task_backlog(db_session: Session): + """Test task backlog calculation.""" + # Create 5 pending tasks + for i in range(5): + task = IntelligentEvalTaskQueueDB( + eval_id=f"eval-{i}", + status="pending", + priority=1, + reason="slot_due", + ) + db_session.add(task) + + # Create 3 assigned tasks (not counted) + for i in range(3): + task = IntelligentEvalTaskQueueDB( + eval_id=f"eval-assigned-{i}", + status="assigned", + priority=1, + reason="slot_due", + ) + db_session.add(task) + + db_session.commit() + + backlog = metrics.calculate_task_backlog(db_session) + assert backlog == 5 + + +def test_calculate_stuck_rate_empty(db_session: Session): + """Test stuck rate when no crons exist.""" + rate = metrics.calculate_stuck_rate(db_session) + assert rate == 0.0 + + +def test_calculate_stuck_rate(db_session: Session): + """Test stuck rate calculation.""" + # Create 10 crons: 2 stuck, 8 active + for i in range(2): + cron = OpenClawCronPoolDB( + openclaw_cron_id=f"stuck-{i}", + status="stuck", + last_active_at=utc_now(), + ) + db_session.add(cron) + + for i in range(8): + cron = OpenClawCronPoolDB( + openclaw_cron_id=f"active-{i}", + status="busy", + last_active_at=utc_now(), + ) + db_session.add(cron) + + db_session.commit() + + rate = metrics.calculate_stuck_rate(db_session) + assert rate == 0.2 + + +def test_calculate_avg_processing_time_empty(db_session: Session): + """Test average processing time when no completed tasks.""" + avg_time = metrics.calculate_avg_processing_time(db_session) + assert avg_time is None + + +def test_calculate_avg_processing_time(db_session: Session): + """Test average processing time calculation.""" + now = utc_now() + + # Create 3 completed tasks with different processing times + task1 = IntelligentEvalTaskQueueDB( + eval_id="eval-1", + status="completed", + priority=1, + reason="slot_due", + assigned_at=now - timedelta(minutes=10), + completed_at=now - timedelta(minutes=5), + ) + task2 = IntelligentEvalTaskQueueDB( + eval_id="eval-2", + status="completed", + priority=1, + reason="slot_due", + assigned_at=now - timedelta(minutes=20), + completed_at=now - timedelta(minutes=10), + ) + task3 = IntelligentEvalTaskQueueDB( + eval_id="eval-3", + status="completed", + priority=1, + reason="slot_due", + assigned_at=now - timedelta(minutes=30), + completed_at=now - timedelta(minutes=15), + ) + + db_session.add_all([task1, task2, task3]) + db_session.commit() + + avg_time = metrics.calculate_avg_processing_time(db_session) + # Average: (5 + 10 + 15) / 3 = 10 minutes = 600 seconds + assert avg_time == 600.0 + + +def test_calculate_eval_completion_rate_empty(db_session: Session): + """Test eval completion rate when no evals exist.""" + rate = metrics.calculate_eval_completion_rate(db_session) + assert rate == 0.0 + + +def test_calculate_eval_completion_rate(db_session: Session): + """Test eval completion rate calculation.""" + # Create 10 evals: 7 completed, 3 executing + for i in range(7): + eval_db = IntelligentEvalDB( + name=f"eval-completed-{i}", + target_id="target1", + status=IntelligentEvalStatus.COMPLETED.value, + ) + db_session.add(eval_db) + + for i in range(3): + eval_db = IntelligentEvalDB( + name=f"eval-executing-{i}", + target_id="target1", + status=IntelligentEvalStatus.EXECUTING.value, + ) + db_session.add(eval_db) + + db_session.commit() + + rate = metrics.calculate_eval_completion_rate(db_session) + assert rate == 0.7 + + +def test_get_all_metrics(db_session: Session): + """Test getting all metrics.""" + # Create some test data + for i in range(5): + cron = OpenClawCronPoolDB( + openclaw_cron_id=f"cron-{i}", + status="busy" if i < 3 else "idle", + last_active_at=utc_now(), + ) + db_session.add(cron) + + db_session.commit() + + all_metrics = metrics.get_all_metrics(db_session) + + assert "pool_utilization" in all_metrics + assert "task_backlog" in all_metrics + assert "stuck_rate" in all_metrics + assert "avg_processing_time_seconds" in all_metrics + assert "eval_completion_rate" in all_metrics + assert "timestamp" in all_metrics + + assert all_metrics["pool_utilization"] == 0.6 + assert all_metrics["task_backlog"] == 0 + assert all_metrics["stuck_rate"] == 0.0