feat(intelligent-eval): task queue monitor (方案③可视化)
All checks were successful
CI / test (push) Successful in 4m1s

方案③的定时触发(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 通过
This commit is contained in:
sinohqb 2026-08-17 13:57:00 +08:00
parent b0969ae582
commit b4f9c887f4
6 changed files with 364 additions and 13 deletions

View File

@ -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 消费端 APInext/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}

View File

@ -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)

View File

@ -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<TaskQueueList>('/intelligent-evals/tasks', { params }),
get: (id: string) => api.get<IntelligentEval>(`/intelligent-evals/${id}`),
create: (data: CreateIntelligentEvalPayload) => api.post<IntelligentEval>('/intelligent-evals', data),
submitPlan: (id: string, plan: Record<string, unknown>) =>

View File

@ -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<TaskQueueStatus, { label: string; color: string }> = {
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<TaskQueueItem[] | null>(null)
const [stats, setStats] = useState<{ pending: number; assigned: number; completed: number; failed: number; unresolved: number } | null>(null)
const [statusFilter, setStatusFilter] = useState<TaskQueueStatus | 'all'>('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<TaskQueueItem> = [
{
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) => (
<Space size={6} direction="vertical" style={{ gap: 2 }}>
<span style={{ fontWeight: 500 }}>{name ?? t.eval_id.slice(0, 8)}</span>
{t.eval_status && (
<span style={{ fontSize: 12, color: colors.textSecondary }}>{t.eval_status}</span>
)}
</Space>
),
},
{
title: '状态', dataIndex: 'status', key: 'status', width: 100,
render: (s: TaskQueueStatus) => {
const meta = STATUS_META[s]
return <Tag color={meta.color}>{meta.label}</Tag>
},
},
{
title: '优先级', dataIndex: 'priority', key: 'priority', width: 80,
render: (p: number) => <Tag>{p}</Tag>,
},
{
title: '原因', dataIndex: 'reason', key: 'reason',
render: (r: string) => (
<Tooltip title={r}>
<span style={{ display: 'inline-block', maxWidth: 220, overflow: 'hidden', textOverflow: 'ellipsis', whiteSpace: 'nowrap' }}>
{r}
</span>
</Tooltip>
),
},
{
title: '认领方', dataIndex: 'assigned_cron_id', key: 'assigned_cron_id', width: 140,
render: (v: string | null, t) => {
if (!v) return <span style={{ color: colors.textSecondary }}></span>
return (
<Tooltip title={t.assigned_at ? `认领于 ${formatDateTime(t.assigned_at)}` : undefined}>
<span style={{ fontFamily: 'monospace', fontSize: 12 }}>{v}</span>
</Tooltip>
)
},
},
{
title: '完成时间', dataIndex: 'completed_at', key: 'completed_at', width: 170,
render: (v: string | null, t) => {
if (t.status === 'failed' && t.error) {
return <Tooltip title={t.error}><span style={{ color: statusColors.failed }}></span></Tooltip>
}
return v ? formatDateTime(v) : '—'
},
},
]
return (
<div style={{ height: '100%', display: 'flex', flexDirection: 'column', gap: 16 }}>
<div style={{ display: 'flex', justifyContent: 'space-between', alignItems: 'center', flexWrap: 'wrap', gap: 8 }}>
<Space size={16} wrap>
<Statistic title="待处理" value={stats?.unresolved ?? 0} valueStyle={{ color: colors.warning }} />
<Statistic title="待认领" value={stats?.pending ?? 0} />
<Statistic title="执行中" value={stats?.assigned ?? 0} valueStyle={{ color: colors.primary }} />
<Statistic title="已完成" value={stats?.completed ?? 0} valueStyle={{ color: statusColors.completed }} />
<Statistic title="失败" value={stats?.failed ?? 0} valueStyle={{ color: statusColors.failed }} />
</Space>
<Space>
<Select
value={statusFilter}
onChange={(v) => setStatusFilter(v)}
style={{ width: 110 }}
options={[{ value: 'all', label: '全部状态' }, ...STATUS_OPTIONS]}
/>
<Button icon={<ReloadOutlined />} onClick={() => void loadData()} />
</Space>
</div>
<Alert
type="info"
showIcon
message="平台每 60 秒扫描 executing 评估并入队,有任务时通过 docker exec 触发 OpenClaw worker 执行(取代外部 Channel 的方案③)。"
style={{ fontSize: 12 }}
/>
<div style={{ flex: 1, overflowY: 'auto' }}>
<Table
rowKey="id"
size="small"
loading={loading}
dataSource={tasks ?? []}
columns={columns}
pagination={(tasks?.length ?? 0) > 50 ? { pageSize: 50, showTotal: (t) => `${t}` } : false}
locale={{ emptyText: <Empty description="暂无任务" /> }}
/>
</div>
</div>
)
}

View File

@ -3,11 +3,14 @@ import {
Button, Drawer, Empty, Form, Input, InputNumber, Select, Space, Spin, Table, Tag, Tooltip, message,
} from 'antd'
import type { ColumnsType } from 'antd/es/table'
import { PlusOutlined, ReloadOutlined, EyeOutlined, FileTextOutlined } from '@ant-design/icons'
import {
PlusOutlined, ReloadOutlined, EyeOutlined, FileTextOutlined, UnorderedListOutlined,
} from '@ant-design/icons'
import FormDrawer from '../components/FormDrawer'
import PageWrapper from '../components/PageWrapper'
import EvalDetail from '../components/intelligent_eval/EvalDetail'
import EvalReport from '../components/intelligent_eval/EvalReport'
import TaskQueueMonitor from '../components/intelligent_eval/TaskQueueMonitor'
import { EVAL_STATUS } from '../components/intelligent_eval/status'
import { useResource } from '../hooks/useResource'
import {
@ -34,6 +37,7 @@ export default function IntelligentEvalsPage() {
const [drawerView, setDrawerView] = useState<DrawerView>('detail')
const [selectedId, setSelectedId] = useState<string | null>(null)
const [createOpen, setCreateOpen] = useState(false)
const [taskQueueOpen, setTaskQueueOpen] = useState(false)
const [submitting, setSubmitting] = useState(false)
const [form] = Form.useForm<CreateFormValues>()
@ -154,6 +158,9 @@ export default function IntelligentEvalsPage() {
extra={
<Space>
<Button icon={<ReloadOutlined />} onClick={() => void reloadList()} />
<Button icon={<UnorderedListOutlined />} onClick={() => setTaskQueueOpen(true)}>
</Button>
<Button type="primary" icon={<PlusOutlined />} onClick={() => setCreateOpen(true)}>
</Button>
@ -193,6 +200,16 @@ export default function IntelligentEvalsPage() {
)}
</Drawer>
<Drawer
title="任务队列(定时触发监控)"
open={taskQueueOpen}
onClose={() => setTaskQueueOpen(false)}
width={1000}
destroyOnClose
>
<TaskQueueMonitor />
</Drawer>
<FormDrawer
title="新建智能评估"
open={createOpen}

View File

@ -3,13 +3,12 @@
from datetime import timedelta
import pytest
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine, select
from agenteval.intelligent_eval.models import IntelligentEvalStatus
from agenteval.storage.db import IntelligentEvalDB, IntelligentEvalTaskQueueDB, utc_now
from agenteval.web.app import app
from agenteval.web.deps import get_db
from fastapi.testclient import TestClient
from sqlmodel import Session, SQLModel, create_engine
@pytest.fixture()
@ -65,10 +64,12 @@ def test_get_next_task_with_pending_task(client: TestClient, db_session: Session
status=IntelligentEvalStatus.EXECUTING.value,
started_at=utc_now() - timedelta(hours=9),
)
eval_db.set_plan({
eval_db.set_plan(
{
"time_distribution": [{"time_slot": "8-10h", "sessions": 2}],
"estimated_sessions": 2,
})
}
)
db_session.add(eval_db)
db_session.commit()
@ -183,10 +184,12 @@ def test_end_to_end_task_lifecycle(client: TestClient, db_session: Session):
status=IntelligentEvalStatus.EXECUTING.value,
started_at=utc_now() - timedelta(hours=9),
)
eval_db.set_plan({
eval_db.set_plan(
{
"time_distribution": [{"time_slot": "8-10h", "sessions": 2}],
"estimated_sessions": 2,
})
}
)
db_session.add(eval_db)
db_session.commit()
@ -214,3 +217,60 @@ def test_end_to_end_task_lifecycle(client: TestClient, db_session: Session):
# Verify task completed
task = db_session.get(IntelligentEvalTaskQueueDB, task_data["id"])
assert task.status == "completed"
def test_list_tasks(client: TestClient, db_session: Session):
"""Task list returns entries (newest first) with eval names and stats."""
eval_db = IntelligentEvalDB(
name="list-eval",
target_id="target1",
status=IntelligentEvalStatus.EXECUTING.value,
started_at=utc_now(),
)
eval_db.set_plan({"time_distribution": [{"time_slot": "0-1h", "sessions": 1}], "estimated_sessions": 1})
db_session.add(eval_db)
db_session.commit()
old = IntelligentEvalTaskQueueDB(eval_id=eval_db.id, status="completed", priority=5, reason="done")
new = IntelligentEvalTaskQueueDB(eval_id=eval_db.id, status="pending", priority=1, reason="slot_due")
db_session.add_all([old, new])
db_session.commit()
# 确保 old 早于 newcreated_at 由 default_factory 生成,顺序可能同秒)
old.created_at = utc_now() - timedelta(seconds=5)
db_session.commit()
response = client.get("/api/intelligent-evals/tasks")
assert response.status_code == 200
data = response.json()
assert data["stats"]["pending"] == 1
assert data["stats"]["completed"] == 1
assert data["stats"]["unresolved"] == 1
# newest first
assert [t["id"] for t in data["tasks"]] == [new.id, old.id]
task = data["tasks"][0]
assert task["eval_id"] == eval_db.id
assert task["eval_name"] == "list-eval"
assert task["eval_status"] == IntelligentEvalStatus.EXECUTING.value
assert task["priority"] == 1
assert task["reason"] == "slot_due"
def test_list_tasks_status_filter(client: TestClient, db_session: Session):
"""Status filter narrows the task list."""
db_session.add(IntelligentEvalTaskQueueDB(eval_id="eval1", status="pending", priority=1, reason="slot_due"))
db_session.add(IntelligentEvalTaskQueueDB(eval_id="eval1", status="failed", priority=1, reason="slot_due"))
db_session.commit()
response = client.get("/api/intelligent-evals/tasks?status=failed")
assert response.status_code == 200
data = response.json()
assert len(data["tasks"]) == 1
assert data["tasks"][0]["status"] == "failed"
assert data["stats"]["failed"] == 1
def test_list_tasks_not_shadowed_by_eval_id(client: TestClient):
"""GET /tasks must hit the task-list endpoint, not /{eval_id} with eval_id="tasks"."""
response = client.get("/api/intelligent-evals/tasks")
assert response.status_code == 200
assert "tasks" in response.json()