From b4f9c887f4ef51a04e0e1e15877e1a4a3751dc9b Mon Sep 17 00:00:00 2001 From: sinohqb Date: Mon, 17 Aug 2026 13:57:00 +0800 Subject: [PATCH] =?UTF-8?q?feat(intelligent-eval):=20task=20queue=20monito?= =?UTF-8?q?r=20(=E6=96=B9=E6=A1=88=E2=91=A2=E5=8F=AF=E8=A7=86=E5=8C=96)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 方案③的定时触发（scan loop 每 60s 入队 + 触发 worker）此前只有 Worker 消费端 API，无可查看的列表。新增： - GET /api/intelligent-evals/tasks：任务明细（含评估名/状态）+ 状态分布统计 （注册在 /{eval_id} 之前避免被捕获为 eval_id="tasks"） - 前端 TaskQueueMonitor 组件 + 智能评估页任务队列入口：5s 轮询 （usePolling），状态卡 + 状态筛选 + 明细表 测试：+3（列表/筛选/不被 {eval_id} 遮蔽），892 passed，tsc 通过 --- .../agenteval/intelligent_eval/task_queue.py | 67 +++++++- .../web/routers/intelligent_evals.py | 16 ++ frontend/web/src/api.ts | 33 ++++ .../intelligent_eval/TaskQueueMonitor.tsx | 160 ++++++++++++++++++ frontend/web/src/pages/IntelligentEvals.tsx | 19 ++- .../test_intelligent_eval_task_queue_api.py | 82 +++++++-- 6 files changed, 364 insertions(+), 13 deletions(-) create mode 100644 frontend/web/src/components/intelligent_eval/TaskQueueMonitor.tsx diff --git a/backend/agenteval/intelligent_eval/task_queue.py b/backend/agenteval/intelligent_eval/task_queue.py index 114f8e2..63bef84 100644 --- a/backend/agenteval/intelligent_eval/task_queue.py +++ b/backend/agenteval/intelligent_eval/task_queue.py @@ -10,7 +10,7 @@ OpenClaw workers to pick up. Tasks are prioritized by: from datetime import timedelta from typing import Optional -from sqlmodel import Session, select, update +from sqlmodel import Session, func, select, update from agenteval.intelligent_eval.models import IntelligentEvalStatus from agenteval.storage.db import ( @@ -23,24 +23,28 @@ from agenteval.storage.db import ( def _is_slot_due(slot: dict, current_offset: timedelta) -> bool: """Thin wrapper delegating to :func:`agenteval.intelligent_eval.domain.is_slot_due`.""" from agenteval.intelligent_eval.domain import is_slot_due as _impl + return _impl(slot, current_offset) def _calculate_session_deficit(eval_db: IntelligentEvalDB, session: Session) -> int: """Thin wrapper delegating to :func:`agenteval.intelligent_eval.domain.calculate_session_deficit`.""" from agenteval.intelligent_eval.domain import calculate_session_deficit as _impl + return _impl(eval_db, session) def _calculate_priority(eval_db: IntelligentEvalDB, session: Session) -> int: """Thin wrapper delegating to :func:`agenteval.intelligent_eval.domain.calculate_priority`.""" from agenteval.intelligent_eval.domain import calculate_priority as _impl + return _impl(eval_db, session) def _get_attention_reason(eval_db: IntelligentEvalDB, session: Session) -> Optional[str]: """Thin wrapper delegating to :func:`agenteval.intelligent_eval.domain.get_attention_reason`.""" from agenteval.intelligent_eval.domain import get_attention_reason as _impl + return _impl(eval_db, session) @@ -158,6 +162,7 @@ def complete_task(task_id: str, success: bool, error: Optional[str], session: Se session.commit() return result.rowcount > 0 + def requeue_stuck_task(eval_id: str, cron_id: str, session: Session) -> bool: """Mark the cron-stuck task as failed and enqueue a retry task. @@ -270,3 +275,63 @@ def requeue_stale_assigned_tasks(session: Session) -> int: if requeued: session.commit() return requeued + + +# --------------------------------------------------------------------------- +# 任务队列监控（方案③可视化）：列表 + 状态分布 +# --------------------------------------------------------------------------- + + +def list_tasks( + session: Session, + status: Optional[str] = None, + limit: int = 100, +) -> dict: + """List task-queue entries with their eval names, newest first. + + 方案③的"定时触发"（scan loop 每分钟扫描入队 + 触发 worker）此前只有 + Worker 消费端 API（next/assign/complete），没有可查看的列表。这里提供 + 任务明细 + 状态分布统计，供前端任务队列监控页展示。 + + Returns: + {"tasks": [...], "stats": {pending, assigned, completed, failed, unresolved}} + """ + stmt = select(IntelligentEvalTaskQueueDB).order_by(IntelligentEvalTaskQueueDB.created_at.desc()) + if status: + stmt = stmt.where(IntelligentEvalTaskQueueDB.status == status) + tasks = session.exec(stmt.limit(max(1, min(limit, 500)))).all() + + # 状态分布（全量统计，不受 limit 影响） + stats = {"pending": 0, "assigned": 0, "completed": 0, "failed": 0, "unresolved": 0} + rows = session.exec( + select( + IntelligentEvalTaskQueueDB.status, + func.count(IntelligentEvalTaskQueueDB.id), + ).group_by(IntelligentEvalTaskQueueDB.status) + ).all() + for status_val, cnt in rows: + if status_val in stats: + stats[status_val] = cnt + stats["unresolved"] = stats["pending"] + stats["assigned"] + + result = [] + for t in tasks: + ev = session.get(IntelligentEvalDB, t.eval_id) + result.append( + { + "id": t.id, + "eval_id": t.eval_id, + "eval_name": ev.name if ev else None, + "eval_status": ev.status if ev else None, + "status": t.status, + "priority": t.priority, + "reason": t.reason, + "assigned_cron_id": t.assigned_cron_id, + "assigned_at": t.assigned_at.isoformat() if t.assigned_at else None, + "completed_at": t.completed_at.isoformat() if t.completed_at else None, + "created_at": t.created_at.isoformat() if t.created_at else None, + "updated_at": t.updated_at.isoformat() if t.updated_at else None, + "error": t.error, + } + ) + return {"tasks": result, "stats": stats} diff --git a/backend/agenteval/web/routers/intelligent_evals.py b/backend/agenteval/web/routers/intelligent_evals.py index 4c1785e..4efe9ad 100644 --- a/backend/agenteval/web/routers/intelligent_evals.py +++ b/backend/agenteval/web/routers/intelligent_evals.py @@ -100,6 +100,22 @@ async def list_evals(session: Session = Depends(get_db)) -> dict: return {"intelligent_evals": [item.model_dump(mode="json") for item in reader.list_items(evals)]} +@router.get("/tasks") +async def list_tasks( + status: str | None = None, + limit: int = 100, + session: Session = Depends(get_db), +) -> dict: + """List task-queue entries with eval names (monitor UI). + + 注意：此端点必须注册在 ``/{eval_id}`` 之前，否则 ``/tasks`` 会被 + ``{eval_id}`` 捕获为 eval_id="tasks"。 + """ + from agenteval.intelligent_eval.task_queue import list_tasks as _list + + return _list(session, status=status, limit=limit) + + @router.get("/{eval_id}") async def get_eval(eval_id: str, session: Session = Depends(get_db)) -> dict: projection = IntelligentEvalReadModel(session).detail_by_id(eval_id) diff --git a/frontend/web/src/api.ts b/frontend/web/src/api.ts index 0ae6d0e..8a36cd0 100644 --- a/frontend/web/src/api.ts +++ b/frontend/web/src/api.ts @@ -753,8 +753,41 @@ export interface DecisionLog { created_at: string | null } +export type TaskQueueStatus = 'pending' | 'assigned' | 'completed' | 'failed' + +export interface TaskQueueStats { + pending: number + assigned: number + completed: number + failed: number + unresolved: number +} + +export interface TaskQueueItem { + id: string + eval_id: string + eval_name: string | null + eval_status: string | null + status: TaskQueueStatus + priority: number + reason: string + assigned_cron_id: string | null + assigned_at: string | null + completed_at: string | null + created_at: string | null + updated_at: string | null + error: string | null +} + +export interface TaskQueueList { + tasks: TaskQueueItem[] + stats: TaskQueueStats +} + export const intelligentEvalsApi = { list: () => api.get<{ intelligent_evals: IntelligentEval[] }>('/intelligent-evals'), + listTasks: (params?: { status?: TaskQueueStatus; limit?: number }) => + api.get('/intelligent-evals/tasks', { params }), get: (id: string) => api.get(`/intelligent-evals/${id}`), create: (data: CreateIntelligentEvalPayload) => api.post('/intelligent-evals', data), submitPlan: (id: string, plan: Record) => diff --git a/frontend/web/src/components/intelligent_eval/TaskQueueMonitor.tsx b/frontend/web/src/components/intelligent_eval/TaskQueueMonitor.tsx new file mode 100644 index 0000000..3ca373d --- /dev/null +++ b/frontend/web/src/components/intelligent_eval/TaskQueueMonitor.tsx @@ -0,0 +1,160 @@ +import { useEffect, useState } from 'react' +import { + Alert, Button, Empty, Select, Space, Statistic, Table, Tag, Tooltip, message, +} from 'antd' +import type { ColumnsType } from 'antd/es/table' +import { ReloadOutlined } from '@ant-design/icons' +import { intelligentEvalsApi, type TaskQueueItem, type TaskQueueStatus } from '../../api' +import { usePolling } from '../../hooks/usePolling' +import { colors, statusColors } from '../../tokens' +import { formatDateTime } from '../../utils/date' + +const STATUS_META: Record = { + pending: { label: '待处理', color: 'orange' }, + assigned: { label: '执行中', color: 'blue' }, + completed: { label: '已完成', color: 'green' }, + failed: { label: '失败', color: 'red' }, +} + +const STATUS_OPTIONS = (Object.keys(STATUS_META) as TaskQueueStatus[]).map((s) => ({ + value: s, + label: STATUS_META[s].label, +})) + +/** + * 任务队列监控（方案③可视化）。 + * + * 方案③的"定时触发"（scan loop 每 60s 扫描入队 + 触发 OpenClaw worker） + * 此前只有 Worker 消费端 API，无可查看的列表。这里展示任务队列明细与 + * 状态分布，5s 轮询，让平台侧的定时触发对用户可见。 + */ +export default function TaskQueueMonitor() { + const [tasks, setTasks] = useState(null) + const [stats, setStats] = useState<{ pending: number; assigned: number; completed: number; failed: number; unresolved: number } | null>(null) + const [statusFilter, setStatusFilter] = useState('all') + const [loading, setLoading] = useState(false) + + const loadData = async () => { + setLoading(true) + try { + const res = await intelligentEvalsApi.listTasks( + statusFilter === 'all' ? undefined : { status: statusFilter }, + ) + setTasks(res.data.tasks) + setStats(res.data.stats) + } catch { + message.error('加载任务队列失败') + } finally { + setLoading(false) + } + } + + // Initial fetch (usePolling owns the 5s interval + visibility pause). + useEffect(() => { + void loadData() + // eslint-disable-next-line react-hooks/exhaustive-deps + }, [statusFilter]) + + usePolling(() => { void loadData() }, 5000, true) + + const columns: ColumnsType = [ + { + title: '入队时间', dataIndex: 'created_at', key: 'created_at', width: 170, + render: (v: string | null) => (v ? formatDateTime(v) : '—'), + }, + { + title: '评估', dataIndex: 'eval_name', key: 'eval_name', + render: (name: string | null, t) => ( + + {name ?? t.eval_id.slice(0, 8)} + {t.eval_status && ( + {t.eval_status} + )} + + ), + }, + { + title: '状态', dataIndex: 'status', key: 'status', width: 100, + render: (s: TaskQueueStatus) => { + const meta = STATUS_META[s] + return {meta.label} + }, + }, + { + title: '优先级', dataIndex: 'priority', key: 'priority', width: 80, + render: (p: number) => {p}, + }, + { + title: '原因', dataIndex: 'reason', key: 'reason', + render: (r: string) => ( + + + {r} + + + ), + }, + { + title: '认领方', dataIndex: 'assigned_cron_id', key: 'assigned_cron_id', width: 140, + render: (v: string | null, t) => { + if (!v) return — + return ( + + {v} + + ) + }, + }, + { + title: '完成时间', dataIndex: 'completed_at', key: 'completed_at', width: 170, + render: (v: string | null, t) => { + if (t.status === 'failed' && t.error) { + return 失败 + } + return v ? formatDateTime(v) : '—' + }, + }, + ] + + return ( +
+
+ + + + + + + + +