From 1aa453ef0ab73e0e86ff74adc805e2e536766824 Mon Sep 17 00:00:00 2001 From: sinohqb Date: Wed, 12 Aug 2026 02:13:21 +0800 Subject: [PATCH] feat(intelligent-eval): add cron pool data model and task queue API Implement Ticket 01 of intelligent eval cron pool architecture (ADR-0007): - Add 4 new tables: task_queue, cron_pool, config_snapshots, decision_logs - Implement task enqueueing logic with priority calculation - Implement task assignment and completion APIs - Add unit tests (9) and integration tests (7) - Update CONTEXT.md with new vocabulary - Add ADR-0007 documenting cron pool architecture decision All 760 tests passing. --- .../issues/01-data-model-and-task-queue.md | 19 + .../issues/02-cron-pool-management.md | 18 + .../issues/03-worker-skill.md | 19 + .../issues/04-decision-logic-and-e2e.md | 19 + .../issues/05-config-snapshot-management.md | 17 + .../issues/06-fault-tolerance.md | 17 + .../issues/07-monitoring-and-alerting.md | 17 + .../issues/08-config-snapshot-page.md | 17 + .../issues/09-decision-process-page.md | 18 + .../issues/10-cron-pool-monitoring-page.md | 19 + .scratch/intelligent-eval-cron-pool/spec.md | 705 ++++++++++++++++++ CONTEXT.md | 20 + .../agenteval/intelligent_eval/task_queue.py | 248 ++++++ backend/agenteval/storage/db.py | 137 ++++ .../web/routers/intelligent_evals.py | 63 ++ docs/adr/0007-intelligent-eval-cron-pool.md | 118 +++ ...b_add_intelligent_eval_cron_pool_tables.py | 130 ++++ tests/integration/test_campaigns_api.py | 4 + tests/integration/test_exploration_api.py | 4 + .../test_exploration_patrol_api.py | 4 + .../test_intelligent_eval_task_queue_api.py | 216 ++++++ .../unit/test_intelligent_eval_task_queue.py | 277 +++++++ 22 files changed, 2106 insertions(+) create mode 100644 .scratch/intelligent-eval-cron-pool/issues/01-data-model-and-task-queue.md create mode 100644 .scratch/intelligent-eval-cron-pool/issues/02-cron-pool-management.md create mode 100644 .scratch/intelligent-eval-cron-pool/issues/03-worker-skill.md create mode 100644 .scratch/intelligent-eval-cron-pool/issues/04-decision-logic-and-e2e.md create mode 100644 .scratch/intelligent-eval-cron-pool/issues/05-config-snapshot-management.md create mode 100644 .scratch/intelligent-eval-cron-pool/issues/06-fault-tolerance.md create mode 100644 .scratch/intelligent-eval-cron-pool/issues/07-monitoring-and-alerting.md create mode 100644 .scratch/intelligent-eval-cron-pool/issues/08-config-snapshot-page.md create mode 100644 .scratch/intelligent-eval-cron-pool/issues/09-decision-process-page.md create mode 100644 .scratch/intelligent-eval-cron-pool/issues/10-cron-pool-monitoring-page.md create mode 100644 .scratch/intelligent-eval-cron-pool/spec.md create mode 100644 backend/agenteval/intelligent_eval/task_queue.py create mode 100644 docs/adr/0007-intelligent-eval-cron-pool.md create mode 100644 migrations/versions/b72debf55c3b_add_intelligent_eval_cron_pool_tables.py create mode 100644 tests/integration/test_intelligent_eval_task_queue_api.py create mode 100644 tests/unit/test_intelligent_eval_task_queue.py diff --git a/.scratch/intelligent-eval-cron-pool/issues/01-data-model-and-task-queue.md b/.scratch/intelligent-eval-cron-pool/issues/01-data-model-and-task-queue.md new file mode 100644 index 0000000..8d16630 --- /dev/null +++ b/.scratch/intelligent-eval-cron-pool/issues/01-data-model-and-task-queue.md @@ -0,0 +1,19 @@ +# 01 — 数据模型 + 任务入队 API + +**What to build:** 创建任务队列和 Cron 池的数据表，实现任务入队逻辑和取任务 API。同时创建配置快照和决策日志表，为后续的可观测性功能打基础。 + +**Blocked by:** None — 可立即开始 + +**Status:** ready-for-agent + +- [ ] 创建 `intelligent_eval_task_queue` 表（任务队列） +- [ ] 创建 `openclaw_cron_pool` 表（Cron 池状态） +- [ ] 创建 `intelligent_eval_config_snapshots` 表（配置快照） +- [ ] 创建 `intelligent_eval_decision_logs` 表（决策日志） +- [ ] 实现任务入队逻辑：每分钟扫描 executing 评估 → 判断是否需要处理 → 生成任务 +- [ ] 实现任务优先级计算（时段到期 > 欠账多 > 等待时间长） +- [ ] 实现 `GET /api/intelligent-evals/tasks/next` API（供 OpenClaw 调用） +- [ ] 实现任务去重逻辑（同一评估已有 pending 任务时不重复创建） +- [ ] 单元测试：任务入队逻辑、优先级计算、去重逻辑 +- [ ] 集成测试：创建评估 → 批准 → 任务自动入队 +- [ ] **可演示**：创建一个智能评估并批准执行，看到任务自动入队 + 配置快照自动保存 diff --git a/.scratch/intelligent-eval-cron-pool/issues/02-cron-pool-management.md b/.scratch/intelligent-eval-cron-pool/issues/02-cron-pool-management.md new file mode 100644 index 0000000..af2f29c --- /dev/null +++ b/.scratch/intelligent-eval-cron-pool/issues/02-cron-pool-management.md @@ -0,0 +1,18 @@ +# 02 — Cron 池管理（平台侧） + +**What to build:** 封装 OpenClaw CLI 客户端，实现 Cron 池的创建、删除、扩容、缩容逻辑，提供池状态查询和手动扩缩容 API。 + +**Blocked by:** 01 — 数据模型 + 任务入队 API + +**Status:** ready-for-agent + +- [ ] 封装 OpenClaw CLI 客户端（`openclaw automations create/remove/list`） +- [ ] 实现池初始化逻辑（启动时创建 min_size 个 cron） +- [ ] 实现池扩容逻辑（busy/total > 0.8 且 total < max_size 时扩容） +- [ ] 实现池缩容逻辑（idle > min_size * 2 且 total > min_size 时缩容） +- [ ] 实现 `GET /api/openclaw/cron-pool` API（查询池状态） +- [ ] 实现 `POST /api/openclaw/cron-pool/scale` API（手动扩缩容） +- [ ] 实现 cron 状态同步（从 OpenClaw 侧同步到平台 DB） +- [ ] 单元测试：扩容/缩容逻辑、状态同步 +- [ ] 集成测试：手动触发扩容 → cron 数量增加；手动触发缩容 → cron 数量减少 +- [ ] **可演示**：手动触发扩容，看到 cron 数量从 5 增加到 10；查询池状态 API 返回正确的统计信息 diff --git a/.scratch/intelligent-eval-cron-pool/issues/03-worker-skill.md b/.scratch/intelligent-eval-cron-pool/issues/03-worker-skill.md new file mode 100644 index 0000000..471f6f3 --- /dev/null +++ b/.scratch/intelligent-eval-cron-pool/issues/03-worker-skill.md @@ -0,0 +1,19 @@ +# 03 — Worker Skill（OpenClaw 侧） + +**What to build:** 创建 OpenClaw Worker Skill，实现从平台取任务、上报心跳、上报任务完成、上报决策日志的功能。 + +**Blocked by:** 01 — 数据模型 + 任务入队 API + +**Status:** ready-for-agent + +- [ ] 创建 `agenteval-intelligent-worker` skill 文件（SKILL.md） +- [ ] 实现取任务逻辑（调 `GET /api/intelligent-evals/tasks/next`） +- [ ] 实现心跳上报（调 `POST /api/openclaw/crons/{id}/heartbeat`） +- [ ] 实现任务完成上报（调 `POST /api/intelligent-evals/tasks/{id}/complete`） +- [ ] 实现决策日志上报（调 `POST /api/intelligent-evals/{id}/decision-logs`） +- [ ] 实现 cron state 的读写（使用 OpenClaw 的 state 机制） +- [ ] 实现 idle/busy 状态切换逻辑 +- [ ] 编写 skill 文档（如何使用、如何调试） +- [ ] 部署脚本：将 skill 同步到 OpenClaw workspace +- [ ] 端到端测试：部署一个 cron → 每分钟唤醒 → 从平台取任务 → 上报心跳 +- [ ] **可演示**：部署一个 cron，每分钟唤醒，从平台取任务（即使任务是空的）+ 上报决策日志 diff --git a/.scratch/intelligent-eval-cron-pool/issues/04-decision-logic-and-e2e.md b/.scratch/intelligent-eval-cron-pool/issues/04-decision-logic-and-e2e.md new file mode 100644 index 0000000..1791a41 --- /dev/null +++ b/.scratch/intelligent-eval-cron-pool/issues/04-decision-logic-and-e2e.md @@ -0,0 +1,19 @@ +# 04 — 决策逻辑 + 端到端执行 + +**What to build:** 实现 Worker 的决策逻辑（分析当前情况 → 决定执行会话/等待/开始分析），适配 evaluator 和 analyst skill，实现决策日志持久化，完成端到端执行流程。 + +**Blocked by:** 03 — Worker Skill（OpenClaw 侧） + +**Status:** ready-for-agent + +- [ ] 实现 worker 的决策逻辑（分析当前时段、已完成会话数、是否有严重问题） +- [ ] 实现决策类型：execute_session / wait / start_analysis +- [ ] 适配 `agenteval-intelligent-evaluator` skill（从 worker 调用，传入 session_config） +- [ ] 适配 `agenteval-intelligent-analyst` skill（从 worker 调用） +- [ ] 实现决策日志持久化（保存到 `intelligent_eval_decision_logs` 表） +- [ ] 实现 cron state 的决策历史记录（decisions_history 字段） +- [ ] 实现评估完成检测（所有会话完成 + 报告已提交） +- [ ] 实现任务完成后归还 cron 逻辑 +- [ ] 单元测试：决策逻辑、评估完成检测 +- [ ] 端到端测试：创建评估 → 批准 → 自动执行 → 完成 → 决策日志完整 +- [ ] **可演示**：端到端执行一个智能评估，看到完整的决策日志（每次 cron 唤醒的决策类型和原因） diff --git a/.scratch/intelligent-eval-cron-pool/issues/05-config-snapshot-management.md b/.scratch/intelligent-eval-cron-pool/issues/05-config-snapshot-management.md new file mode 100644 index 0000000..6f67c8e --- /dev/null +++ b/.scratch/intelligent-eval-cron-pool/issues/05-config-snapshot-management.md @@ -0,0 +1,17 @@ +# 05 — 配置快照管理 + +**What to build:** 实现配置快照的自动保存和查询功能，支持快照对比（diff）。 + +**Blocked by:** 01 — 数据模型 + 任务入队 API + +**Status:** ready-for-agent + +- [ ] 实现配置快照自动保存（创建评估时、提交计划时、修改配置时） +- [ ] 实现快照类型：created / plan_submitted / config_updated +- [ ] 实现 `GET /api/intelligent-evals/{id}/config-snapshots` API（查询快照列表） +- [ ] 实现 `GET /api/intelligent-evals/{id}/config-snapshots/{snapshot_id}` API（查询单个快照） +- [ ] 实现配置对比功能（两个快照的 diff，返回差异字段） +- [ ] 实现 `POST /api/intelligent-evals/{id}/config-snapshots/compare` API（对比两个快照） +- [ ] 单元测试：快照保存、快照对比 +- [ ] 集成测试：创建评估 → 修改配置 → 自动生成新快照 → 对比两个快照 +- [ ] **可演示**：修改评估配置，看到新的快照自动生成，可以对比两个版本的差异 diff --git a/.scratch/intelligent-eval-cron-pool/issues/06-fault-tolerance.md b/.scratch/intelligent-eval-cron-pool/issues/06-fault-tolerance.md new file mode 100644 index 0000000..951d955 --- /dev/null +++ b/.scratch/intelligent-eval-cron-pool/issues/06-fault-tolerance.md @@ -0,0 +1,17 @@ +# 06 — 故障恢复 + 状态对账 + +**What to build:** 实现 cron 卡死检测、任务重新入队、状态对账、OpenClaw 重启恢复等故障恢复机制。 + +**Blocked by:** 04 — 决策逻辑 + 端到端执行 + +**Status:** ready-for-agent + +- [ ] 实现卡死检测（10 分钟未活跃的 cron 标记为 stuck） +- [ ] 实现任务重新入队（cron 卡死后，任务重新分配给其他 cron） +- [ ] 实现状态对账（每 5 分钟检查平台 DB 与 OpenClaw state 一致性） +- [ ] 实现 OpenClaw 重启恢复（cron state 持久化，重启后继续处理） +- [ ] 实现平台重启恢复（扫描 assigned 任务，检查 cron 是否还活跃） +- [ ] 实现卡死 cron 的清理逻辑（删除卡死的 cron，创建新 cron 补充） +- [ ] 单元测试：卡死检测、任务重新入队、状态对账 +- [ ] 集成测试：模拟 cron 卡死 → 任务自动重新分配；重启 OpenClaw → 任务继续执行 +- [ ] **可演示**：模拟 cron 卡死，看到任务自动重新分配；重启 OpenClaw，看到任务继续执行 diff --git a/.scratch/intelligent-eval-cron-pool/issues/07-monitoring-and-alerting.md b/.scratch/intelligent-eval-cron-pool/issues/07-monitoring-and-alerting.md new file mode 100644 index 0000000..279c2f2 --- /dev/null +++ b/.scratch/intelligent-eval-cron-pool/issues/07-monitoring-and-alerting.md @@ -0,0 +1,17 @@ +# 07 — 监控和告警 + +**What to build:** 实现关键指标计算、告警规则、告警通知功能。 + +**Blocked by:** 04 — 决策逻辑 + 端到端执行 + +**Status:** ready-for-agent + +- [ ] 实现关键指标计算（池使用率、任务积压、卡死率、平均处理时间、评估完成率） +- [ ] 实现 `GET /api/openclaw/cron-pool/metrics` API（查询指标） +- [ ] 实现告警规则（池使用率 > 90% 持续 10 分钟、任务积压 > 50、卡死率 > 10%） +- [ ] 实现告警通知（日志 + 可选的 webhook） +- [ ] 实现告警历史记录（保存到 DB） +- [ ] 实现 `GET /api/openclaw/cron-pool/alerts` API（查询告警历史） +- [ ] 单元测试：指标计算、告警规则 +- [ ] 集成测试：触发告警条件 → 收到告警通知 +- [ ] **可演示**：触发告警条件（如手动让 20 个 cron 都 busy），收到告警 diff --git a/.scratch/intelligent-eval-cron-pool/issues/08-config-snapshot-page.md b/.scratch/intelligent-eval-cron-pool/issues/08-config-snapshot-page.md new file mode 100644 index 0000000..464e1d2 --- /dev/null +++ b/.scratch/intelligent-eval-cron-pool/issues/08-config-snapshot-page.md @@ -0,0 +1,17 @@ +# 08 — 配置快照页面（前端） + +**What to build:** 在智能评估详情页新增「配置历史」Tab，显示快照列表、快照详情、快照对比功能。 + +**Blocked by:** 05 — 配置快照管理 + +**Status:** ready-for-agent + +- [ ] 在智能评估详情页新增「配置历史」Tab +- [ ] 实现快照列表（时间、版本、修改人、快照类型） +- [ ] 实现快照详情展示（四件套、粗计划、时间窗口） +- [ ] 实现快照对比功能（选择两个快照，显示 diff） +- [ ] 实现快照恢复功能（可选，将配置回滚到某个快照） +- [ ] 实现快照导出功能（导出为 JSON） +- [ ] 前端类型检查（`tsc --noEmit`） +- [ ] 端到端测试：在 UI 上看到配置历史，可以对比两个版本 +- [ ] **可演示**：在 UI 上看到配置历史，可以对比两个版本，可以导出快照 diff --git a/.scratch/intelligent-eval-cron-pool/issues/09-decision-process-page.md b/.scratch/intelligent-eval-cron-pool/issues/09-decision-process-page.md new file mode 100644 index 0000000..5db2f1b --- /dev/null +++ b/.scratch/intelligent-eval-cron-pool/issues/09-decision-process-page.md @@ -0,0 +1,18 @@ +# 09 — 决策过程页面（前端） + +**What to build:** 在智能评估详情页新增「决策过程」Tab，显示决策时间线、任务分配历史、cron state 快照。 + +**Blocked by:** 04 — 决策逻辑 + 端到端执行 + +**Status:** ready-for-agent + +- [ ] 在智能评估详情页新增「决策过程」Tab +- [ ] 实现决策时间线（每次 cron 唤醒的时间、决策类型、原因） +- [ ] 实现任务分配历史（哪个 cron 处理了这个评估，何时分配/完成） +- [ ] 实现 cron state 快照展示（每个时间点的状态） +- [ ] 实现决策过滤器（按决策类型、时间范围筛选） +- [ ] 实现决策详情展开（显示决策时的完整上下文） +- [ ] 实现决策日志导出功能（导出为 JSON 或 CSV） +- [ ] 前端类型检查（`tsc --noEmit`） +- [ ] 端到端测试：在 UI 上看到完整的决策过程，可以追踪每一步的原因 +- [ ] **可演示**：在 UI 上看到完整的决策过程，可以追踪每一步的原因，可以导出决策日志 diff --git a/.scratch/intelligent-eval-cron-pool/issues/10-cron-pool-monitoring-page.md b/.scratch/intelligent-eval-cron-pool/issues/10-cron-pool-monitoring-page.md new file mode 100644 index 0000000..b78fa8a --- /dev/null +++ b/.scratch/intelligent-eval-cron-pool/issues/10-cron-pool-monitoring-page.md @@ -0,0 +1,19 @@ +# 10 — Cron 池监控页面（前端） + +**What to build:** 创建独立的「Cron 池监控」页面，显示池状态、任务队列、cron 列表、任务历史。 + +**Blocked by:** 07 — 监控和告警 + +**Status:** ready-for-agent + +- [ ] 创建「Cron 池监控」页面（独立页面，不在评估详情页） +- [ ] 实现池状态卡片（total/idle/busy/stuck） +- [ ] 实现任务队列列表（pending/assigned/completed，支持筛选） +- [ ] 实现 cron 列表（ID、状态、当前任务、最后活跃时间、完成任务数） +- [ ] 实现任务历史列表（完成时间、成功率、平均处理时间） +- [ ] 实现告警历史列表（时间、级别、消息） +- [ ] 实现实时刷新（每 5 秒轮询） +- [ ] 实现手动扩缩容按钮（调用 `POST /api/openclaw/cron-pool/scale`） +- [ ] 前端类型检查（`tsc --noEmit`） +- [ ] 端到端测试：在 UI 上实时看到池状态和任务执行情况 +- [ ] **可演示**：在 UI 上实时看到池状态和任务执行情况，可以手动扩缩容 diff --git a/.scratch/intelligent-eval-cron-pool/spec.md b/.scratch/intelligent-eval-cron-pool/spec.md new file mode 100644 index 0000000..ca89d67 --- /dev/null +++ b/.scratch/intelligent-eval-cron-pool/spec.md @@ -0,0 +1,705 @@ +# 智能评估 Cron 池实现规范 + +**版本**: v1.1 +**日期**: 2026-08-11 +**状态**: draft +**决策依据**: ADR-0007 + +## Problem Statement + +当前智能评估的 OpenClaw 集成采用"多 Session"模式：每个评估对应多个无状态的 OpenClaw session，通过 API 传递上下文。这导致： +1. 上下文割裂：planner 的决策逻辑，evaluator 不知道 +2. 无法追踪：哪个 OpenClaw session 做了什么，平台无视图 +3. 重复读取：每个 session 都要重新读取评估详情 + +同时，如果改为"一个评估一个 cron"，会导致 cron 爆炸（100 个评估 = 100 个 cron）。 + +## Solution + +采用 **Cron 池模式**：OpenClaw 维护一个 cron 池（5-20 个），平台维护任务队列，cron 从队列取任务执行，完成后归还。 + +### 核心概念 + +- **Cron 池（Cron Pool）**：OpenClaw 侧的工作单元池，每个 cron 可以处理任意评估 +- **任务队列（Task Queue）**：平台侧的待处理评估队列，按优先级排序 +- **工作单元（Worker）**：一个 cron + 其 state，表示一个可用的执行单元 +- **任务（Task）**：一个需要处理的评估，包含 eval_id、优先级、原因 + +--- + +## 数据模型 + +### 1. 平台侧：任务队列 + +```python +class EvalTaskQueueDB(SQLModel, table=True): + """智能评估任务队列""" + __tablename__ = "intelligent_eval_task_queue" + + id: str = Field(primary_key=True) # 任务 ID + eval_id: str = Field(foreign_key="intelligent_evals.id", index=True) # 关联的评估 + + # 任务状态 + status: str = Field(index=True) # pending / assigned / completed / failed + priority: int = Field(index=True) # 优先级（越小越优先） + reason: str # 需要处理的原因（如 "slot_due", "all_sessions_completed"） + + # 分配信息 + assigned_cron_id: str | None = None # 分配的 cron ID + assigned_at: datetime | None = None # 分配时间 + + # 完成信息 + completed_at: datetime | None = None + error: str | None = None + + # 时间戳 + created_at: datetime + updated_at: datetime + + # 索引 + __table_args__ = ( + Index("idx_status_priority", "status", "priority"), + Index("idx_eval_status", "eval_id", "status"), + ) +``` + +### 2. 平台侧：Cron 池状态 + +```python +class OpenClawCronPoolDB(SQLModel, table=True): + """OpenClaw Cron 池状态（平台侧记录）""" + __tablename__ = "openclaw_cron_pool" + + id: str = Field(primary_key=True) # 平台侧 ID + openclaw_cron_id: str = Field(unique=True, index=True) # OpenClaw 侧的 cron ID + + # 状态 + status: str = Field(index=True) # idle / busy / stuck + current_eval_id: str | None = Field(foreign_key="intelligent_evals.id") # 当前处理的评估 + + # 心跳 + last_active_at: datetime # 上次活跃时间 + last_task_at: datetime | None = None # 上次取任务时间 + + # 统计 + total_tasks_completed: int = 0 + total_tasks_failed: int = 0 + + # 时间戳 + created_at: datetime + updated_at: datetime + + # 索引 + __table_args__ = ( + Index("idx_status_last_active", "status", "last_active_at"), + ) +``` + +### 3. OpenClaw 侧：Cron State + +```json +{ + "status": "idle | busy", + "eval_id": "uuid | null", + "started_at": "ISO8601 | null", + "last_decision_at": "ISO8601", + "completed_sessions": 0, + "decisions_history": [ + { + "timestamp": "ISO8601", + "decision": "execute_session | wait | start_analysis", + "reason": "..." + } + ] +} +``` + +--- + +## API 契约 + +### 1. 平台侧 API（供 OpenClaw 调用） + +#### 获取下一个任务 + +```http +GET /api/intelligent-evals/tasks/next +X-API-Key: + +Response 200: +{ + "task": { + "id": "task_uuid", + "eval_id": "eval_uuid", + "priority": 1, + "reason": "slot_due", + "eval": { + "id": "eval_uuid", + "name": "...", + "status": "executing", + "plan": {...}, + "started_at": "..." + } + } +} + +Response 200 (无任务): +{ + "task": null +} +``` + +#### 标记任务完成 + +```http +POST /api/intelligent-evals/tasks/{task_id}/complete +X-API-Key: +Content-Type: application/json + +{ + "cron_id": "openclaw_cron_id", + "success": true, + "error": null +} + +Response 200: +{ + "success": true +} +``` + +#### 上报 Cron 心跳 + +```http +POST /api/openclaw/crons/{cron_id}/heartbeat +X-API-Key: +Content-Type: application/json + +{ + "status": "busy", + "current_eval_id": "eval_uuid" +} + +Response 200: +{ + "success": true +} +``` + +### 2. 平台侧管理 API（供前端/管理员调用） + +#### 查看 Cron 池状态 + +```http +GET /api/openclaw/cron-pool +X-API-Key: + +Response 200: +{ + "pool": { + "total": 10, + "idle": 3, + "busy": 7, + "stuck": 0, + "min_size": 5, + "max_size": 20 + }, + "crons": [ + { + "id": "platform_id", + "openclaw_cron_id": "openclaw_id", + "status": "busy", + "current_eval_id": "eval_uuid", + "last_active_at": "...", + "total_tasks_completed": 15 + } + ] +} +``` + +#### 手动扩容/缩容 + +```http +POST /api/openclaw/cron-pool/scale +X-API-Key: +Content-Type: application/json + +{ + "target_size": 15 +} + +Response 200: +{ + "success": true, + "current_size": 15 +} +``` + +--- + +## 状态机 + +### 1. 任务状态机 + +``` +pending → assigned → completed + → failed +``` + +- **pending**：任务已创建，等待分配 +- **assigned**：任务已分配给某个 cron +- **completed**：任务已完成 +- **failed**：任务失败（cron 卡死、评估取消等） + +### 2. Cron 状态机 + +``` +idle → busy → idle + → stuck → (平台介入) → idle +``` + +- **idle**：空闲，可以取任务 +- **busy**：正在处理任务 +- **stuck**：卡死（10 分钟未活跃） + +### 3. 评估状态机（不变） + +``` +draft → planning → pending_approval → executing → completed + → cancelled + → failed +``` + +--- + +## 核心流程 + +### 1. 任务入队流程 + +```python +# 平台侧：定时扫描（每分钟） +async def scan_and_enqueue_tasks(): + """扫描所有 executing 评估，生成任务""" + evals = get_executing_evals() + + for eval in evals: + # 判断是否需要立即处理 + if needs_attention(eval): + # 检查是否已有 pending 任务（去重） + existing = get_pending_task(eval.id) + if existing: + continue + + # 创建任务 + task = EvalTaskQueueDB( + eval_id=eval.id, + status="pending", + priority=calculate_priority(eval), + reason=get_attention_reason(eval), + created_at=utc_now() + ) + session.add(task) + + session.commit() + +def needs_attention(eval) -> bool: + """判断评估是否需要立即处理""" + # 1. 检查是否有时段到期 + current_offset = utc_now() - eval.started_at + for slot in eval.plan["time_distribution"]: + if is_slot_due(slot, current_offset): + return True + + # 2. 检查是否所有会话完成（需要开始分析） + sessions = get_sessions(eval.id) + if all(s.status == "completed" for s in sessions): + return True + + return False + +def calculate_priority(eval) -> int: + """计算优先级（越小越优先）""" + priority = 100 + + # 时段到期的优先 + if has_due_slot(eval): + priority -= 50 + + # 欠账多的优先 + deficit = calculate_session_deficit(eval) + priority -= deficit * 10 + + # 等待时间长的优先 + wait_minutes = (utc_now() - eval.started_at).total_seconds() / 60 + priority -= min(wait_minutes / 10, 20) + + return max(priority, 1) +``` + +### 2. Cron 取任务流程 + +```python +# OpenClaw 侧：Worker Skill 每分钟执行 +async def worker_tick(): + """Cron 每分钟唤醒""" + # 1. 读取自己的 state + state = get_cron_state() + + if state["status"] == "idle": + # 2. 调平台 API 取任务 + task = await fetch_next_task() + + if task is None: + # 无任务，本节拍结束 + return + + # 3. 更新 state 为 busy + update_cron_state({ + "status": "busy", + "eval_id": task["eval_id"], + "started_at": utc_now() + }) + + # 4. 处理任务 + await process_task(task) + + elif state["status"] == "busy": + # 5. 继续处理当前任务 + eval_id = state["eval_id"] + await continue_processing(eval_id) + +async def process_task(task): + """处理任务""" + eval_id = task["eval_id"] + + # 读取评估详情 + eval = await fetch_eval(eval_id) + + # 决策逻辑 + decision = make_decision(eval) + + if decision == "execute_session": + # 调用 evaluator skill + await execute_session(eval) + elif decision == "start_analysis": + # 调用 analyst skill + await start_analysis(eval) + elif decision == "wait": + # 等待，更新 state + update_cron_state({"last_decision_at": utc_now()}) + + # 检查是否完成 + if is_eval_completed(eval): + # 标记任务完成 + await complete_task(task["id"], success=True) + + # 归还 cron + update_cron_state({"status": "idle", "eval_id": None}) +``` + +### 3. 池管理流程 + +```python +# 平台侧：池管理器（每分钟执行） +async def manage_pool(): + """管理 cron 池""" + pool = get_pool_status() + + # 1. 扩容 + if pool["busy"] / pool["total"] > 0.8 and pool["total"] < MAX_POOL_SIZE: + await scale_up(1) + + # 2. 缩容 + if pool["idle"] > MIN_POOL_SIZE * 2 and pool["total"] > MIN_POOL_SIZE: + await scale_down(1) + + # 3. 检测卡死的 cron + stuck_crons = get_stuck_crons() # 10 分钟未活跃 + for cron in stuck_crons: + await handle_stuck_cron(cron) + +async def scale_up(count: int): + """扩容""" + for _ in range(count): + # 调 OpenClaw CLI 创建 cron + cron_id = await openclaw_client.create_cron( + name=f"intelligent-eval-worker-{uuid()}", + schedule="* * * * *", + skill="agenteval-intelligent-worker", + state={"status": "idle"} + ) + + # 记录到平台 DB + pool_db = OpenClawCronPoolDB( + openclaw_cron_id=cron_id, + status="idle", + last_active_at=utc_now() + ) + session.add(pool_db) + + session.commit() + +async def handle_stuck_cron(cron): + """处理卡死的 cron""" + # 1. 标记 cron 为 stuck + cron.status = "stuck" + + # 2. 如果有正在处理的任务，标记为 failed + if cron.current_eval_id: + task = get_assigned_task(cron.current_eval_id) + if task: + task.status = "failed" + task.error = "Cron stuck" + + # 3. 重新入队（让其他 cron 处理） + new_task = EvalTaskQueueDB( + eval_id=cron.current_eval_id, + status="pending", + priority=1, # 高优先级 + reason="cron_stuck_retry" + ) + session.add(new_task) + + # 4. 删除卡死的 cron + await openclaw_client.delete_cron(cron.openclaw_cron_id) + session.delete(cron) + + # 5. 创建新 cron 补充 + await scale_up(1) + + session.commit() +``` + +--- + +## 故障恢复策略 + +### 1. OpenClaw 重启 + +**场景**：OpenClaw Gateway 重启 + +**影响**： +- Cron state 持久化在 SQLite，不丢失 +- 正在执行的 session 可能中断 + +**恢复**： +1. OpenClaw 重启后，cron 自动恢复 +2. Cron 下次唤醒时，检查 state： + - 如果 busy → 继续处理（从 state 恢复上下文） + - 如果 idle → 正常取任务 + +### 2. 平台重启 + +**场景**：AgentEvalTool 平台重启 + +**影响**： +- 任务队列持久化在 DB，不丢失 +- 正在处理的任务状态可能不一致 + +**恢复**： +1. 平台重启后，扫描所有 assigned 任务 +2. 检查对应的 cron 是否还活跃： + - 如果 cron 活跃 → 任务继续 + - 如果 cron 不活跃 → 任务重新入队 + +### 3. Cron 卡死 + +**场景**：Cron 10 分钟未活跃（未上报心跳） + +**检测**： +```python +def detect_stuck_crons(): + """检测卡死的 cron""" + threshold = utc_now() - timedelta(minutes=10) + stuck = session.query(OpenClawCronPoolDB).filter( + OpenClawCronPoolDB.status == "busy", + OpenClawCronPoolDB.last_active_at < threshold + ).all() + return stuck +``` + +**处理**： +1. 标记 cron 为 stuck +2. 任务重新入队 +3. 删除卡死的 cron +4. 创建新 cron 补充 + +### 4. 状态不一致 + +**场景**：平台 DB 与 OpenClaw state 不一致 + +**检测**： +```python +async def reconcile_state(): + """定期对账（每 5 分钟）""" + # 1. 检查平台 DB 中 busy 的 cron，OpenClaw 侧是否存在 + busy_crons = get_busy_crons() + for cron in busy_crons: + openclaw_state = await openclaw_client.get_cron_state(cron.openclaw_cron_id) + if openclaw_state is None: + # OpenClaw 侧不存在，标记为 stuck + await handle_stuck_cron(cron) + + # 2. 检查 OpenClaw 侧 busy 的 cron，平台 DB 是否记录 + openclaw_crons = await openclaw_client.list_crons() + for oc_cron in openclaw_crons: + platform_cron = get_cron_by_openclaw_id(oc_cron.id) + if platform_cron is None: + # 平台 DB 不存在，补充记录 + await sync_cron_to_db(oc_cron) +``` + +--- + +## 监控和告警 + +### 1. 关键指标 + +```python +# 池使用率 +pool_utilization = busy_crons / total_crons + +# 任务积压 +pending_tasks_count = count_pending_tasks() + +# Cron 卡死率 +stuck_rate = stuck_crons / total_crons + +# 平均任务处理时间 +avg_task_duration = avg(completed_at - assigned_at) + +# 评估完成率 +eval_completion_rate = completed_evals / total_evals +``` + +### 2. 告警规则 + +| 指标 | 阈值 | 告警级别 | +|------|------|---------| +| 池使用率 | > 90% 持续 10 分钟 | Warning | +| 任务积压 | > 50 个 | Warning | +| Cron 卡死率 | > 10% | Critical | +| 平均任务处理时间 | > 30 分钟 | Warning | +| 评估完成率 | < 80% | Info | + +### 3. 监控面板 + +前端新增「Cron 池监控」页面： +- 池状态（total/idle/busy/stuck） +- 任务队列（pending/assigned/completed） +- Cron 列表（ID、状态、当前任务、最后活跃时间） +- 任务历史（完成时间、成功率） + +--- + +## 性能优化 + +### 1. 数据库索引 + +```sql +-- 任务队列索引 +CREATE INDEX idx_status_priority ON intelligent_eval_task_queue(status, priority); +CREATE INDEX idx_eval_status ON intelligent_eval_task_queue(eval_id, status); + +-- Cron 池索引 +CREATE INDEX idx_status_last_active ON openclaw_cron_pool(status, last_active_at); +``` + +### 2. 缓存 + +```python +# 缓存池状态（1 分钟） +@cache(ttl=60) +def get_pool_status(): + return calculate_pool_status() + +# 缓存任务队列（30 秒） +@cache(ttl=30) +def get_pending_tasks(): + return fetch_pending_tasks() +``` + +### 3. 批量操作 + +```python +# 批量标记任务完成 +async def batch_complete_tasks(task_ids: list[str]): + session.query(EvalTaskQueueDB).filter( + EvalTaskQueueDB.id.in_(task_ids) + ).update({"status": "completed", "completed_at": utc_now()}) + session.commit() +``` + +--- + +## 测试策略 + +### 1. 单元测试 + +- 任务优先级计算 +- 池扩容/缩容逻辑 +- 卡死检测逻辑 +- 状态对账逻辑 + +### 2. 集成测试 + +- 任务入队 → cron 取任务 → 处理 → 完成 +- Cron 卡死 → 任务重新入队 +- 池满 → 新任务排队 +- OpenClaw 重启 → 状态恢复 + +### 3. 端到端测试 + +- 创建评估 → 分配 cron → 执行 → 完成 +- 并发 20 个评估 → 池扩容 → 全部完成 +- Cron 卡死 → 自动恢复 + +--- + +## 迁移策略 + +### 从当前方案迁移 + +1. **阶段 1**：部署 Cron 池，但保持旧的多 Session 模式 +2. **阶段 2**：新评估使用 Cron 池，旧评估继续用旧模式 +3. **阶段 3**：旧评估完成后，下线旧模式 + +### 数据迁移 + +无需数据迁移（新功能，不影响现有数据） + +--- + +## Out of Scope + +- 事件驱动唤醒（webhook）：后续优化 +- 优先级队列 UI：后续优化 +- 跨评估对比：v1.2 功能 +- 多对象绑定：v1.2 功能 + +--- + +## Open Questions + +1. **OpenClaw Gateway API 的具体端点？** + - 需要验证或通过 CLI 封装 + +2. **Cron state 的 16KB 限制是否够用？** + - 粗计划 JSON 大约 2-5KB + - 决策历史可能需要截断 + +3. **池大小的最佳配置？** + - 最小 5 个，最大 20 个，是否需要调整？ + +4. **任务优先级的权重？** + - 时段到期 -50，欠账多 -10/个，等待时间 -1/10分钟 + - 是否需要调整？ + +--- + +## References + +- ADR-0007: 智能评估 OpenClaw 集成采用 Cron 池模式 +- CONTEXT.md: 智能评估词汇 +- ADR-0003: 评估活动分期 +- [OpenClaw Cron Jobs Documentation](https://docs.openclaw.ai/automation/cron-jobs) diff --git a/CONTEXT.md b/CONTEXT.md index 97847eb..a847f7d 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -138,6 +138,26 @@ _Avoid_: 岗位（与模型用途的"岗位"概念冲突） 智能评估的模拟约束：模拟一个完整服务周期（如 24h）内的用户交互分布。OpenClaw 规划时考虑交互时机（早高峰、午间冷清、晚间投诉多），通过 cron 自唤醒在对应时间点执行。是模拟约束而非硬截止。 _Avoid_: 窗口（与静态评估的"服务周期窗口"混淆时需加前缀） +**Cron 池（Cron Pool）**: +OpenClaw 侧的工作单元池（5-20 个 cron），每个 cron 可以处理任意智能评估。池化管理避免"一个评估一个 cron"导致的 cron 爆炸，同时保持每个 cron 的自主决策能力。平台负责创建/删除 cron，OpenClaw 负责执行。（决策见 ADR-0007） +_Avoid_: 任务池、工作池（太泛） + +**工作单元（Worker）**: +Cron 池中的一个 cron + 其 state，表示一个可用的执行单元。Worker 每分钟唤醒，从平台任务队列取一个任务执行，完成后归还到池中。Worker 有完整的决策权：执行会话、等待、开始分析。 +_Avoid_: 执行器、处理器（失去自主性含义） + +**任务队列（Task Queue）**: +平台侧的待处理评估队列，持久化在 DB。每分钟扫描所有 executing 评估，判断哪些需要立即处理（时段到期、有欠账），按优先级排序（时段到期 > 欠账多 > 等待时间长）。Worker 从队列取任务。 +_Avoid_: 消息队列（与 MQ 混淆）、任务列表 + +**配置快照（Config Snapshot）**: +智能评估配置的历史版本（四件套、粗计划、时间窗口）。创建评估、提交计划、修改配置时自动保存，支持对比和回滚。用于追溯配置变更历史。 +_Avoid_: 版本（与场景版本混淆）、备份 + +**决策日志（Decision Log）**: +Worker 每次唤醒时的决策记录（执行会话/等待/开始分析 + 原因 + 上下文）。持久化在 DB，用于调试和审计 OpenClaw 的自主决策过程。 +_Avoid_: 日志（太泛）、审计日志（与安全审计混淆） + ## 模型配置 **模型能力（Capability）**: diff --git a/backend/agenteval/intelligent_eval/task_queue.py b/backend/agenteval/intelligent_eval/task_queue.py new file mode 100644 index 0000000..2e84ede --- /dev/null +++ b/backend/agenteval/intelligent_eval/task_queue.py @@ -0,0 +1,248 @@ +"""Task queue for intelligent evaluations (任务队列). + +Platform scans executing evals every minute and enqueues tasks for +OpenClaw workers to pick up. Tasks are prioritized by: +1. Time slot due (时段到期) +2. Session deficit (欠账多) +3. Wait time (等待时间长) +""" + +from datetime import datetime, timedelta +from typing import Optional + +from sqlmodel import Session, select + +from agenteval.intelligent_eval.models import IntelligentEval, IntelligentEvalStatus +from agenteval.storage.db import ( + IntelligentEvalDB, + IntelligentEvalSessionDB, + IntelligentEvalTaskQueueDB, + as_utc, + utc_now, +) + + +def _is_slot_due(slot: dict, current_offset: timedelta) -> bool: + """Check if a time slot is due (时段到期). + + Args: + slot: Time slot from plan.time_distribution (e.g., {"time_slot": "8-10h", "sessions": 2}) + current_offset: Time elapsed since eval started + + Returns: + True if the slot's start time has passed + """ + time_slot = slot.get("time_slot", "") + if not time_slot: + return False + + # Parse time slot (e.g., "8-10h" -> 8 hours) + try: + start_hour = int(time_slot.split("-")[0].replace("h", "")) + slot_start = timedelta(hours=start_hour) + return current_offset >= slot_start + except (ValueError, IndexError): + return False + + +def _calculate_session_deficit(eval_db: IntelligentEvalDB, session: Session) -> int: + """Calculate session deficit (欠账). + + Returns: + Number of sessions that should exist but don't + """ + if not eval_db.plan: + return 0 + + plan = eval_db.get_plan() + time_distribution = plan.get("time_distribution", []) + estimated_sessions = plan.get("estimated_sessions", 0) + + # Count current sessions + current_sessions = session.exec( + select(IntelligentEvalSessionDB).where(IntelligentEvalSessionDB.eval_id == eval_db.id) + ).all() + current_count = len(current_sessions) + + # Calculate how many sessions should exist by now + current_offset = utc_now() - as_utc(eval_db.started_at) if eval_db.started_at else timedelta(0) + should_have = 0 + + for slot in time_distribution: + if _is_slot_due(slot, current_offset): + should_have += slot.get("sessions", 0) + + # Deficit = should have - current + deficit = max(0, should_have - current_count) + return deficit + + +def _calculate_priority(eval_db: IntelligentEvalDB, session: Session) -> int: + """Calculate task priority (越小越优先). + + Priority rules: + - Base priority: 100 + - Time slot due: -50 + - Session deficit: -10 per session + - Wait time: -1 per 10 minutes (max -20) + """ + priority = 100 + + # Check if any time slot is due + if eval_db.plan and eval_db.started_at: + plan = eval_db.get_plan() + time_distribution = plan.get("time_distribution", []) + current_offset = utc_now() - as_utc(eval_db.started_at) + + for slot in time_distribution: + if _is_slot_due(slot, current_offset): + priority -= 50 + break + + # Session deficit + deficit = _calculate_session_deficit(eval_db, session) + priority -= deficit * 10 + + # Wait time + if eval_db.started_at: + wait_minutes = (utc_now() - as_utc(eval_db.started_at)).total_seconds() / 60 + priority -= min(int(wait_minutes / 10), 20) + + return max(priority, 1) + + +def _get_attention_reason(eval_db: IntelligentEvalDB, session: Session) -> Optional[str]: + """Determine why this eval needs attention. + + Returns: + Reason string, or None if no attention needed + """ + if not eval_db.plan or not eval_db.started_at: + return None + + plan = eval_db.get_plan() + time_distribution = plan.get("time_distribution", []) + current_offset = utc_now() - as_utc(eval_db.started_at) + + # Check if any time slot is due + for slot in time_distribution: + if _is_slot_due(slot, current_offset): + deficit = _calculate_session_deficit(eval_db, session) + if deficit > 0: + return "slot_due" + + # Check if all sessions completed (need analysis) + sessions = session.exec( + select(IntelligentEvalSessionDB).where(IntelligentEvalSessionDB.eval_id == eval_db.id) + ).all() + + if sessions and all(s.status == "completed" for s in sessions): + estimated_sessions = plan.get("estimated_sessions", 0) + if len(sessions) >= estimated_sessions: + return "all_sessions_completed" + + return None + + +def _has_pending_task(eval_id: str, session: Session) -> bool: + """Check if eval already has a pending task (去重).""" + existing = session.exec( + select(IntelligentEvalTaskQueueDB).where( + IntelligentEvalTaskQueueDB.eval_id == eval_id, + IntelligentEvalTaskQueueDB.status == "pending", + ) + ).first() + return existing is not None + + +def scan_and_enqueue_tasks(session: Session) -> int: + """Scan all executing evals and enqueue tasks. + + Returns: + Number of tasks enqueued + """ + # Get all executing evals + evals = session.exec( + select(IntelligentEvalDB).where(IntelligentEvalDB.status == IntelligentEvalStatus.EXECUTING.value) + ).all() + + enqueued = 0 + + for eval_db in evals: + # Check if eval needs attention + reason = _get_attention_reason(eval_db, session) + if reason is None: + continue + + # Check if already has pending task (去重) + if _has_pending_task(eval_db.id, session): + continue + + # Calculate priority + priority = _calculate_priority(eval_db, session) + + # Create task + task = IntelligentEvalTaskQueueDB( + eval_id=eval_db.id, + status="pending", + priority=priority, + reason=reason, + created_at=utc_now(), + updated_at=utc_now(), + ) + session.add(task) + enqueued += 1 + + session.commit() + return enqueued + + +def get_next_task(session: Session) -> Optional[IntelligentEvalTaskQueueDB]: + """Get next pending task (highest priority). + + Returns: + Task with lowest priority value (highest priority), or None + """ + task = session.exec( + select(IntelligentEvalTaskQueueDB) + .where(IntelligentEvalTaskQueueDB.status == "pending") + .order_by(IntelligentEvalTaskQueueDB.priority, IntelligentEvalTaskQueueDB.created_at) + .limit(1) + ).first() + return task + + +def assign_task(task_id: str, cron_id: str, session: Session) -> bool: + """Assign a task to a cron. + + Returns: + True if assigned successfully, False if task not found or already assigned + """ + task = session.get(IntelligentEvalTaskQueueDB, task_id) + if task is None or task.status != "pending": + return False + + task.status = "assigned" + task.assigned_cron_id = cron_id + task.assigned_at = utc_now() + task.updated_at = utc_now() + session.commit() + return True + + +def complete_task(task_id: str, success: bool, error: Optional[str], session: Session) -> bool: + """Mark a task as completed or failed. + + Returns: + True if completed successfully, False if task not found + """ + task = session.get(IntelligentEvalTaskQueueDB, task_id) + if task is None: + return False + + task.status = "completed" if success else "failed" + task.completed_at = utc_now() + task.error = error + task.updated_at = utc_now() + session.commit() + return True diff --git a/backend/agenteval/storage/db.py b/backend/agenteval/storage/db.py index 4a7587e..e2b782d 100644 --- a/backend/agenteval/storage/db.py +++ b/backend/agenteval/storage/db.py @@ -582,6 +582,143 @@ class IntelligentEvalMessageDB(SQLModel, table=True): created_at: Optional[datetime] = Field(default_factory=utc_now) +class IntelligentEvalTaskQueueDB(SQLModel, table=True): + """Task queue for intelligent evaluations (任务队列). + + Platform scans executing evals every minute and enqueues tasks for + OpenClaw workers to pick up. + """ + + __tablename__ = "intelligent_eval_task_queue" + + id: Optional[str] = Field(default_factory=new_uuid, primary_key=True) + eval_id: str = Field(index=True, foreign_key="intelligent_evals.id") + + # Task status + status: str = Field(index=True) # pending / assigned / completed / failed + priority: int = Field(index=True) # Lower is higher priority + reason: str # Why this task needs attention (e.g., "slot_due", "all_sessions_completed") + + # Assignment info + assigned_cron_id: Optional[str] = None + assigned_at: Optional[datetime] = None + + # Completion info + completed_at: Optional[datetime] = None + error: Optional[str] = None + + # Timestamps + created_at: Optional[datetime] = Field(default_factory=utc_now) + updated_at: Optional[datetime] = Field(default_factory=utc_now) + + __table_args__ = ( + sa.Index("idx_task_queue_status_priority", "status", "priority"), + sa.Index("idx_task_queue_eval_status", "eval_id", "status"), + ) + + +class OpenClawCronPoolDB(SQLModel, table=True): + """OpenClaw cron pool state (平台侧记录). + + Platform tracks which crons exist and their current state. + """ + + __tablename__ = "openclaw_cron_pool" + + id: Optional[str] = Field(default_factory=new_uuid, primary_key=True) + openclaw_cron_id: str = Field(unique=True, index=True) # OpenClaw's cron job ID + + # State + status: str = Field(index=True) # idle / busy / stuck + current_eval_id: Optional[str] = Field(foreign_key="intelligent_evals.id") + + # Heartbeat + last_active_at: datetime + last_task_at: Optional[datetime] = None + + # Stats + total_tasks_completed: int = 0 + total_tasks_failed: int = 0 + + # Timestamps + created_at: Optional[datetime] = Field(default_factory=utc_now) + updated_at: Optional[datetime] = Field(default_factory=utc_now) + + __table_args__ = (sa.Index("idx_cron_pool_status_last_active", "status", "last_active_at"),) + + +class IntelligentEvalConfigSnapshotDB(SQLModel, table=True): + """Config snapshot for intelligent evaluations (配置快照). + + Automatically saved when eval is created, plan is submitted, or config is updated. + """ + + __tablename__ = "intelligent_eval_config_snapshots" + + id: Optional[str] = Field(default_factory=new_uuid, primary_key=True) + eval_id: str = Field(index=True, foreign_key="intelligent_evals.id") + + # Snapshot type + snapshot_type: str # created / plan_submitted / config_updated + + # Config snapshot + goal: str + seeds: str # JSON + intent: str + role_description: str + time_window_hours: int + plan: Optional[str] = None # JSON, coarse plan snapshot + + # Metadata + created_at: Optional[datetime] = Field(default_factory=utc_now) + created_by: str = "user" # user / openclaw + + __table_args__ = (sa.Index("idx_config_snapshots_eval_created", "eval_id", "created_at"),) + + def get_seeds(self) -> dict[str, Any]: + return _json_loads(self.seeds) + + def set_seeds(self, seeds: dict[str, Any]) -> None: + self.seeds = _json_dumps(seeds) + + def get_plan(self) -> Optional[dict[str, Any]]: + return _json_loads(self.plan) if self.plan else None + + def set_plan(self, plan: dict[str, Any]) -> None: + self.plan = _json_dumps(plan) + + +class IntelligentEvalDecisionLogDB(SQLModel, table=True): + """Decision log for intelligent evaluations (决策日志). + + Records every decision made by OpenClaw workers (execute_session / wait / start_analysis). + """ + + __tablename__ = "intelligent_eval_decision_logs" + + id: Optional[str] = Field(default_factory=new_uuid, primary_key=True) + eval_id: str = Field(index=True, foreign_key="intelligent_evals.id") + + # Decision info + decision_type: str # execute_session / wait / start_analysis + reason: str # Why this decision was made + context: str # JSON, decision context (current time slot, completed sessions, etc.) + + # Execution info + cron_id: str # Which cron made this decision + + # Timestamp + created_at: Optional[datetime] = Field(default_factory=utc_now) + + __table_args__ = (sa.Index("idx_decision_logs_eval_created", "eval_id", "created_at"),) + + def get_context(self) -> dict[str, Any]: + return _json_loads(self.context) + + def set_context(self, context: dict[str, Any]) -> None: + self.context = _json_dumps(context) + + def init_db() -> None: SQLModel.metadata.create_all(engine) diff --git a/backend/agenteval/web/routers/intelligent_evals.py b/backend/agenteval/web/routers/intelligent_evals.py index 763c96c..30d3f69 100644 --- a/backend/agenteval/web/routers/intelligent_evals.py +++ b/backend/agenteval/web/routers/intelligent_evals.py @@ -235,3 +235,66 @@ async def list_messages(eval_id: str, session_id: str, session: Session = Depend if messages is None: raise HTTPException(status_code=404, detail=f"intelligent eval session {session_id} not found") return {"messages": [m.model_dump(mode="json") for m in messages]} + + +@router.get("/tasks/next") +async def get_next_task(session: Session = Depends(get_db)) -> dict: + """Get next pending task for OpenClaw workers. + + Returns the highest-priority pending task, or None if no tasks available. + """ + from agenteval.intelligent_eval import task_queue + + task = task_queue.get_next_task(session) + if task is None: + return {"task": None} + + # Load eval details + from agenteval.storage.db import IntelligentEvalDB + + eval_db = session.get(IntelligentEvalDB, task.eval_id) + if eval_db is None: + return {"task": None} + + return { + "task": { + "id": task.id, + "eval_id": task.eval_id, + "priority": task.priority, + "reason": task.reason, + "eval": { + "id": eval_db.id, + "name": eval_db.name, + "status": eval_db.status, + "plan": eval_db.get_plan(), + "started_at": eval_db.started_at.isoformat() if eval_db.started_at else None, + }, + } + } + + +@router.post("/tasks/{task_id}/assign") +async def assign_task(task_id: str, cron_id: str, session: Session = Depends(get_db)) -> dict: + """Assign a task to a cron worker.""" + from agenteval.intelligent_eval import task_queue + + success = task_queue.assign_task(task_id, cron_id, session) + if not success: + raise HTTPException(status_code=404, detail="task not found or already assigned") + return {"success": True} + + +@router.post("/tasks/{task_id}/complete") +async def complete_task( + task_id: str, + success: bool, + error: str | None = None, + session: Session = Depends(get_db), +) -> dict: + """Mark a task as completed or failed.""" + from agenteval.intelligent_eval import task_queue + + completed = task_queue.complete_task(task_id, success, error, session) + if not completed: + raise HTTPException(status_code=404, detail="task not found") + return {"success": True} diff --git a/docs/adr/0007-intelligent-eval-cron-pool.md b/docs/adr/0007-intelligent-eval-cron-pool.md new file mode 100644 index 0000000..c0ffb93 --- /dev/null +++ b/docs/adr/0007-intelligent-eval-cron-pool.md @@ -0,0 +1,118 @@ +# ADR-0007: 智能评估 OpenClaw 集成采用 Cron 池模式 + +**状态**: 已接受 +**日期**: 2026-08-11 +**决策者**: 架构团队 +**相关**: ADR-0003（评估活动分期）、CONTEXT.md（智能评估词汇） + +## Context + +智能评估（Intelligent Evaluation）需要 OpenClaw 作为"虚拟用户大脑"自主规划和执行评测任务。关键争议在于：**一个智能评估如何与 OpenClaw 的执行单元关联**。 + +### 核心矛盾 + +1. **自主性 vs 可扩展性**： + - 保持 OpenClaw 自主性（每个评估独立决策）→ 一个评估一个 cron → cron 爆炸 + - 解决 cron 爆炸（共享 cron）→ OpenClaw 失去自主性 + +2. **耐久性 vs 灵活性**： + - 长 session（上下文连续）→ OpenClaw 重启丢失 + - 多 session（耐久）→ 上下文割裂 + +3. **平台控制 vs OpenClaw 自治**： + - 平台调度（精确控制）→ OpenClaw 退化为执行器 + - OpenClaw 自治（灵活）→ 平台失去控制 + +### 已否决的方案 + +- **一个评估 = 一个 OpenClaw Agent**：OpenClaw 不支持持久 agent 概念 +- **一个评估 = 一个长 Session**：长 session 超时，重启丢失 +- **平台调度 + Webhook 触发**：OpenClaw 失去自主性，退化为静态评估 +- **全局单 Cron + 批量处理**：单次唤醒耗时长，无并发 +- **事件驱动 + Cron 兜底**：OpenClaw 无法自主规划 + +## Decision + +采用 **Cron 池模式**： + +1. **OpenClaw 维护 Cron 池**（5-20 个，动态扩容/缩容） + - 每个 cron 是"工作单元"，可以处理任意评估 + - Cron state 存储 `{"status": "idle/busy", "eval_id": "..."}` + - 每分钟唤醒，自主决策"现在该做什么" + +2. **平台维护任务队列**（持久化在 DB） + - 扫描所有 executing 评估 + - 判断哪些需要立即处理（时段到期、有欠账） + - 按优先级排序，提供给 OpenClaw + +3. **Cron 每次唤醒时**： + - 如果 idle → 从平台队列取一个任务 + - 如果 busy → 继续处理当前评估 + - 处理完 → 归还 cron 到池中 + +4. **池管理**： + - 平台负责创建/删除 cron（通过 OpenClaw CLI 或 Gateway API） + - 负载高时扩容（busy/total > 0.8） + - 负载低时缩容（idle > min_size * 2） + +## Consequences + +### 优点 + +1. **保持 OpenClaw 自主性**：每个 cron 有完整的决策权（规划、执行、调整） +2. **可扩展性**：池化复用，最多 20 个 cron，支持 100+ 并发评估（排队） +3. **耐久性好**：Cron state + 任务队列都持久化，重启可恢复 +4. **资源可控**：限制并发评估数量（最多 20 个） +5. **弹性伸缩**：根据负载自动扩容/缩容 +6. **技术可行**：OpenClaw 支持动态创建/删除 cron（CLI + Gateway API） + +### 缺点 + +1. **复杂度高**：需要实现池管理、任务队列、超时检测、故障恢复 +2. **状态同步**：Cron state 与平台状态需要保持一致 +3. **调试困难**：需要追踪 cron 分配历史和决策日志 + +### 风险与缓解 + +| 风险 | 缓解措施 | +|------|---------| +| Cron 卡死 | 平台检测 10 分钟未活跃 → 标记评估 stuck → 分配新 cron | +| 池满（20 个都在用） | 新评估排队等待，前端提示"排队中" | +| OpenClaw 重启 | Cron state 持久化在 SQLite，重启后恢复 | +| 平台重启 | 任务队列持久化在 DB，重启后恢复 | +| 状态不一致 | 平台定期对账（每 5 分钟），发现不一致自动修复 | + +### 性能影响 + +- **响应延迟**：分钟级（cron 每分钟触发），对于 24 小时窗口的评估可接受 +- **资源消耗**：最多 20 个 cron 同时运行，每分钟 20 次唤醒 +- **数据库压力**：任务队列查询每分钟 20 次，需要索引优化 + +## Implementation Notes + +### 技术验证 + +OpenClaw 支持动态创建/删除 cron： +- CLI: `openclaw automations create/remove` +- Gateway API: 文档明确支持（具体端点待验证） +- State 持久化: SQLite，16KB 自定义 JSON + +### 关键设计决策 + +1. **池管理归属**：平台负责创建/删除 cron，OpenClaw 负责执行 +2. **状态权威**：平台 DB 是任务状态的权威，cron state 是执行上下文 +3. **超时机制**：评估 2 小时未完成 → 强制归还 cron；cron 10 分钟未活跃 → 标记卡死 +4. **公平性**：任务队列按优先级排序（时段到期 > 欠账多 > 等待时间长） + +### 后续优化 + +- **事件驱动**：平台状态变更时主动触发 OpenClaw（webhook），减少 cron 轮询压力 +- **优先级队列**：支持用户手动提升某个评估的优先级 +- **监控告警**：池使用率、任务积压、cron 卡死率 + +## References + +- [OpenClaw Cron Jobs Documentation](https://docs.openclaw.ai/automation/cron-jobs) +- [OpenClaw CLI Cron Commands](https://docs.openclaw.ai/cli/cron) +- CONTEXT.md「智能评估」章节 +- ADR-0003「评估活动分期」 diff --git a/migrations/versions/b72debf55c3b_add_intelligent_eval_cron_pool_tables.py b/migrations/versions/b72debf55c3b_add_intelligent_eval_cron_pool_tables.py new file mode 100644 index 0000000..03c148b --- /dev/null +++ b/migrations/versions/b72debf55c3b_add_intelligent_eval_cron_pool_tables.py @@ -0,0 +1,130 @@ +"""add intelligent eval cron pool tables + +Revision ID: b72debf55c3b +Revises: d4e7f9a1b2c3 +Create Date: 2026-08-12 01:45:36.671065 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +import sqlmodel + + +# revision identifiers, used by Alembic. +revision: str = 'b72debf55c3b' +down_revision: Union[str, Sequence[str], None] = 'd4e7f9a1b2c3' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Upgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + op.create_table('intelligent_eval_config_snapshots', + sa.Column('id', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('eval_id', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('snapshot_type', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('goal', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('seeds', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('intent', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('role_description', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('time_window_hours', sa.Integer(), nullable=False), + sa.Column('plan', sqlmodel.sql.sqltypes.AutoString(), nullable=True), + sa.Column('created_at', sa.DateTime(), nullable=True), + sa.Column('created_by', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.ForeignKeyConstraint(['eval_id'], ['intelligent_evals.id'], ), + sa.PrimaryKeyConstraint('id') + ) + with op.batch_alter_table('intelligent_eval_config_snapshots', schema=None) as batch_op: + batch_op.create_index('idx_config_snapshots_eval_created', ['eval_id', 'created_at'], unique=False) + batch_op.create_index(batch_op.f('ix_intelligent_eval_config_snapshots_eval_id'), ['eval_id'], unique=False) + + op.create_table('intelligent_eval_decision_logs', + sa.Column('id', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('eval_id', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('decision_type', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('reason', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('context', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('cron_id', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('created_at', sa.DateTime(), nullable=True), + sa.ForeignKeyConstraint(['eval_id'], ['intelligent_evals.id'], ), + sa.PrimaryKeyConstraint('id') + ) + with op.batch_alter_table('intelligent_eval_decision_logs', schema=None) as batch_op: + batch_op.create_index('idx_decision_logs_eval_created', ['eval_id', 'created_at'], unique=False) + batch_op.create_index(batch_op.f('ix_intelligent_eval_decision_logs_eval_id'), ['eval_id'], unique=False) + + op.create_table('intelligent_eval_task_queue', + sa.Column('id', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('eval_id', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('status', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('priority', sa.Integer(), nullable=False), + sa.Column('reason', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('assigned_cron_id', sqlmodel.sql.sqltypes.AutoString(), nullable=True), + sa.Column('assigned_at', sa.DateTime(), nullable=True), + sa.Column('completed_at', sa.DateTime(), nullable=True), + sa.Column('error', sqlmodel.sql.sqltypes.AutoString(), nullable=True), + sa.Column('created_at', sa.DateTime(), nullable=True), + sa.Column('updated_at', sa.DateTime(), nullable=True), + sa.ForeignKeyConstraint(['eval_id'], ['intelligent_evals.id'], ), + sa.PrimaryKeyConstraint('id') + ) + with op.batch_alter_table('intelligent_eval_task_queue', schema=None) as batch_op: + batch_op.create_index('idx_task_queue_eval_status', ['eval_id', 'status'], unique=False) + batch_op.create_index('idx_task_queue_status_priority', ['status', 'priority'], unique=False) + batch_op.create_index(batch_op.f('ix_intelligent_eval_task_queue_eval_id'), ['eval_id'], unique=False) + batch_op.create_index(batch_op.f('ix_intelligent_eval_task_queue_priority'), ['priority'], unique=False) + batch_op.create_index(batch_op.f('ix_intelligent_eval_task_queue_status'), ['status'], unique=False) + + op.create_table('openclaw_cron_pool', + sa.Column('id', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('openclaw_cron_id', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('status', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('current_eval_id', sqlmodel.sql.sqltypes.AutoString(), nullable=True), + sa.Column('last_active_at', sa.DateTime(), nullable=False), + sa.Column('last_task_at', sa.DateTime(), nullable=True), + sa.Column('total_tasks_completed', sa.Integer(), nullable=False), + sa.Column('total_tasks_failed', sa.Integer(), nullable=False), + sa.Column('created_at', sa.DateTime(), nullable=True), + sa.Column('updated_at', sa.DateTime(), nullable=True), + sa.ForeignKeyConstraint(['current_eval_id'], ['intelligent_evals.id'], ), + sa.PrimaryKeyConstraint('id') + ) + with op.batch_alter_table('openclaw_cron_pool', schema=None) as batch_op: + batch_op.create_index('idx_cron_pool_status_last_active', ['status', 'last_active_at'], unique=False) + batch_op.create_index(batch_op.f('ix_openclaw_cron_pool_openclaw_cron_id'), ['openclaw_cron_id'], unique=True) + batch_op.create_index(batch_op.f('ix_openclaw_cron_pool_status'), ['status'], unique=False) + + # ### end Alembic commands ### + + +def downgrade() -> None: + """Downgrade schema.""" + # ### commands auto generated by Alembic - please adjust! ### + with op.batch_alter_table('openclaw_cron_pool', schema=None) as batch_op: + batch_op.drop_index(batch_op.f('ix_openclaw_cron_pool_status')) + batch_op.drop_index(batch_op.f('ix_openclaw_cron_pool_openclaw_cron_id')) + batch_op.drop_index('idx_cron_pool_status_last_active') + + op.drop_table('openclaw_cron_pool') + with op.batch_alter_table('intelligent_eval_task_queue', schema=None) as batch_op: + batch_op.drop_index(batch_op.f('ix_intelligent_eval_task_queue_status')) + batch_op.drop_index(batch_op.f('ix_intelligent_eval_task_queue_priority')) + batch_op.drop_index(batch_op.f('ix_intelligent_eval_task_queue_eval_id')) + batch_op.drop_index('idx_task_queue_status_priority') + batch_op.drop_index('idx_task_queue_eval_status') + + op.drop_table('intelligent_eval_task_queue') + with op.batch_alter_table('intelligent_eval_decision_logs', schema=None) as batch_op: + batch_op.drop_index(batch_op.f('ix_intelligent_eval_decision_logs_eval_id')) + batch_op.drop_index('idx_decision_logs_eval_created') + + op.drop_table('intelligent_eval_decision_logs') + with op.batch_alter_table('intelligent_eval_config_snapshots', schema=None) as batch_op: + batch_op.drop_index(batch_op.f('ix_intelligent_eval_config_snapshots_eval_id')) + batch_op.drop_index('idx_config_snapshots_eval_created') + + op.drop_table('intelligent_eval_config_snapshots') + # ### end Alembic commands ### diff --git a/tests/integration/test_campaigns_api.py b/tests/integration/test_campaigns_api.py index 3808942..93bc811 100644 --- a/tests/integration/test_campaigns_api.py +++ b/tests/integration/test_campaigns_api.py @@ -484,6 +484,10 @@ def test_exploration_config_migration_on_existing_db(tmp_path, monkeypatch): connection.execute(text("DROP TABLE IF EXISTS intelligent_eval_messages")) connection.execute(text("DROP TABLE IF EXISTS intelligent_eval_sessions")) connection.execute(text("DROP TABLE IF EXISTS intelligent_evals")) + connection.execute(text("DROP TABLE IF EXISTS intelligent_eval_config_snapshots")) + 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 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 c95548b..2e9e6da 100644 --- a/tests/integration/test_exploration_api.py +++ b/tests/integration/test_exploration_api.py @@ -473,6 +473,10 @@ def test_exploration_migration_on_existing_db(tmp_path, monkeypatch): connection.execute(text("DROP TABLE IF EXISTS intelligent_eval_messages")) connection.execute(text("DROP TABLE IF EXISTS intelligent_eval_sessions")) connection.execute(text("DROP TABLE IF EXISTS intelligent_evals")) + connection.execute(text("DROP TABLE IF EXISTS intelligent_eval_config_snapshots")) + 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 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 f4858b7..5a4f1dc 100644 --- a/tests/integration/test_exploration_patrol_api.py +++ b/tests/integration/test_exploration_patrol_api.py @@ -192,6 +192,10 @@ async def test_patrol_migration_column_on_existing_db(tmp_path, monkeypatch): connection.execute(text("DROP TABLE IF EXISTS intelligent_eval_messages")) connection.execute(text("DROP TABLE IF EXISTS intelligent_eval_sessions")) connection.execute(text("DROP TABLE IF EXISTS intelligent_evals")) + connection.execute(text("DROP TABLE IF EXISTS intelligent_eval_config_snapshots")) + 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 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_intelligent_eval_task_queue_api.py b/tests/integration/test_intelligent_eval_task_queue_api.py new file mode 100644 index 0000000..67d9dc1 --- /dev/null +++ b/tests/integration/test_intelligent_eval_task_queue_api.py @@ -0,0 +1,216 @@ +"""Integration tests for intelligent eval task queue API.""" + +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 + + +@pytest.fixture() +def client(tmp_path): + """Create a TestClient with a fresh database.""" + from agenteval.storage.db import ( # noqa: F401 + IntelligentEvalDB, + IntelligentEvalSessionDB, + IntelligentEvalTaskQueueDB, + ) + + 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.""" + # The session is stored in the dependency override + return next(app.dependency_overrides[get_db]()) + + +def test_get_next_task_empty(client: TestClient): + """Test getting next task when queue is empty.""" + response = client.get("/api/intelligent-evals/tasks/next") + assert response.status_code == 200 + assert response.json() == {"task": None} + + +def test_get_next_task_with_pending_task(client: TestClient, db_session: Session): + """Test getting next task when there is a pending task.""" + # Create eval + eval_db = IntelligentEvalDB( + name="test", + target_id="target1", + status=IntelligentEvalStatus.EXECUTING.value, + started_at=utc_now() - timedelta(hours=9), + ) + eval_db.set_plan({ + "time_distribution": [{"time_slot": "8-10h", "sessions": 2}], + "estimated_sessions": 2, + }) + db_session.add(eval_db) + db_session.commit() + + # Create pending task + task = IntelligentEvalTaskQueueDB( + eval_id=eval_db.id, + status="pending", + priority=10, + reason="slot_due", + ) + db_session.add(task) + db_session.commit() + + # Get next task + response = client.get("/api/intelligent-evals/tasks/next") + assert response.status_code == 200 + + data = response.json() + assert data["task"] is not None + assert data["task"]["id"] == task.id + assert data["task"]["eval_id"] == eval_db.id + assert data["task"]["priority"] == 10 + assert data["task"]["reason"] == "slot_due" + assert data["task"]["eval"]["id"] == eval_db.id + assert data["task"]["eval"]["name"] == "test" + assert data["task"]["eval"]["status"] == IntelligentEvalStatus.EXECUTING.value + + +def test_assign_task(client: TestClient, db_session: Session): + """Test assigning a task to a cron.""" + # Create pending task + task = IntelligentEvalTaskQueueDB( + eval_id="eval1", + status="pending", + priority=1, + reason="slot_due", + ) + db_session.add(task) + db_session.commit() + + # Assign task + response = client.post(f"/api/intelligent-evals/tasks/{task.id}/assign?cron_id=cron1") + assert response.status_code == 200 + assert response.json() == {"success": True} + + # Verify assignment + db_session.refresh(task) + assert task.status == "assigned" + assert task.assigned_cron_id == "cron1" + + +def test_assign_task_not_found(client: TestClient): + """Test assigning a non-existent task.""" + response = client.post("/api/intelligent-evals/tasks/nonexistent/assign?cron_id=cron1") + assert response.status_code == 404 + + +def test_complete_task(client: TestClient, db_session: Session): + """Test completing a task.""" + # Create assigned task + task = IntelligentEvalTaskQueueDB( + eval_id="eval1", + status="assigned", + priority=1, + reason="slot_due", + assigned_cron_id="cron1", + ) + db_session.add(task) + db_session.commit() + + # Complete task + response = client.post(f"/api/intelligent-evals/tasks/{task.id}/complete?success=true") + assert response.status_code == 200 + assert response.json() == {"success": True} + + # Verify completion + db_session.refresh(task) + assert task.status == "completed" + assert task.completed_at is not None + + +def test_complete_task_with_error(client: TestClient, db_session: Session): + """Test completing a task with error.""" + # Create assigned task + task = IntelligentEvalTaskQueueDB( + eval_id="eval1", + status="assigned", + priority=1, + reason="slot_due", + assigned_cron_id="cron1", + ) + db_session.add(task) + db_session.commit() + + # Complete task with error + response = client.post(f"/api/intelligent-evals/tasks/{task.id}/complete?success=false&error=test_error") + assert response.status_code == 200 + assert response.json() == {"success": True} + + # Verify completion + db_session.refresh(task) + assert task.status == "failed" + assert task.error == "test_error" + + +def test_end_to_end_task_lifecycle(client: TestClient, db_session: Session): + """Test end-to-end task lifecycle: create eval -> scan -> enqueue -> assign -> complete.""" + # Create eval that needs attention + eval_db = IntelligentEvalDB( + name="test", + target_id="target1", + status=IntelligentEvalStatus.EXECUTING.value, + started_at=utc_now() - timedelta(hours=9), + ) + eval_db.set_plan({ + "time_distribution": [{"time_slot": "8-10h", "sessions": 2}], + "estimated_sessions": 2, + }) + db_session.add(eval_db) + db_session.commit() + + # Scan and enqueue tasks + from agenteval.intelligent_eval import task_queue + + enqueued = task_queue.scan_and_enqueue_tasks(db_session) + assert enqueued == 1 + + # Get next task + response = client.get("/api/intelligent-evals/tasks/next") + assert response.status_code == 200 + task_data = response.json()["task"] + assert task_data is not None + assert task_data["eval_id"] == eval_db.id + + # Assign task + response = client.post(f"/api/intelligent-evals/tasks/{task_data['id']}/assign?cron_id=cron1") + assert response.status_code == 200 + + # Complete task + response = client.post(f"/api/intelligent-evals/tasks/{task_data['id']}/complete?success=true") + assert response.status_code == 200 + + # Verify task completed + task = db_session.get(IntelligentEvalTaskQueueDB, task_data["id"]) + assert task.status == "completed" diff --git a/tests/unit/test_intelligent_eval_task_queue.py b/tests/unit/test_intelligent_eval_task_queue.py new file mode 100644 index 0000000..1e3f578 --- /dev/null +++ b/tests/unit/test_intelligent_eval_task_queue.py @@ -0,0 +1,277 @@ +"""Unit tests for intelligent eval task queue.""" + +from datetime import timedelta + +import pytest +from sqlmodel import Session, select + +from agenteval.intelligent_eval import task_queue +from agenteval.intelligent_eval.models import IntelligentEvalStatus +from agenteval.storage.db import ( + IntelligentEvalDB, + IntelligentEvalSessionDB, + IntelligentEvalTaskQueueDB, + utc_now, +) + + +def test_is_slot_due(): + """Test time slot due detection.""" + # Slot "8-10h" should be due after 8 hours + slot = {"time_slot": "8-10h", "sessions": 2} + assert task_queue._is_slot_due(slot, timedelta(hours=7)) is False + assert task_queue._is_slot_due(slot, timedelta(hours=8)) is True + assert task_queue._is_slot_due(slot, timedelta(hours=9)) is True + + # Invalid slot format + assert task_queue._is_slot_due({"time_slot": "invalid"}, timedelta(hours=1)) is False + assert task_queue._is_slot_due({}, timedelta(hours=1)) is False + + +def test_calculate_session_deficit(db_session: Session): + """Test session deficit calculation.""" + # Create eval with plan + eval_db = IntelligentEvalDB( + name="test", + target_id="target1", + status=IntelligentEvalStatus.EXECUTING.value, + started_at=utc_now() - timedelta(hours=9), + ) + eval_db.set_plan({ + "time_distribution": [ + {"time_slot": "0-2h", "sessions": 1}, + {"time_slot": "8-10h", "sessions": 2}, + ], + "estimated_sessions": 3, + }) + db_session.add(eval_db) + db_session.commit() + + # No sessions yet, should have 3 (1 from 0-2h, 2 from 8-10h) + deficit = task_queue._calculate_session_deficit(eval_db, db_session) + assert deficit == 3 + + # Add 1 session + session_db = IntelligentEvalSessionDB( + eval_id=eval_db.id, + target_id="target1", + status="completed", + ) + db_session.add(session_db) + db_session.commit() + + # Should have 3, has 1, deficit = 2 + deficit = task_queue._calculate_session_deficit(eval_db, db_session) + assert deficit == 2 + + +def test_calculate_priority(db_session: Session): + """Test task priority calculation.""" + # Eval with due slot and deficit + eval_db = IntelligentEvalDB( + name="test", + target_id="target1", + status=IntelligentEvalStatus.EXECUTING.value, + started_at=utc_now() - timedelta(hours=9), + ) + eval_db.set_plan({ + "time_distribution": [{"time_slot": "8-10h", "sessions": 2}], + "estimated_sessions": 2, + }) + db_session.add(eval_db) + db_session.commit() + + priority = task_queue._calculate_priority(eval_db, db_session) + # Base 100 - 50 (slot due) - 20 (deficit 2 * 10) - 20 (wait 9h / 10min = 54, capped at 20) + assert priority == 10 + + +def test_get_attention_reason(db_session: Session): + """Test attention reason detection.""" + # Eval with due slot + eval_db = IntelligentEvalDB( + name="test", + target_id="target1", + status=IntelligentEvalStatus.EXECUTING.value, + started_at=utc_now() - timedelta(hours=9), + ) + eval_db.set_plan({ + "time_distribution": [{"time_slot": "8-10h", "sessions": 2}], + "estimated_sessions": 2, + }) + db_session.add(eval_db) + db_session.commit() + + reason = task_queue._get_attention_reason(eval_db, db_session) + assert reason == "slot_due" + + # Add all sessions as completed + for _ in range(2): + session_db = IntelligentEvalSessionDB( + eval_id=eval_db.id, + target_id="target1", + status="completed", + ) + db_session.add(session_db) + db_session.commit() + + reason = task_queue._get_attention_reason(eval_db, db_session) + assert reason == "all_sessions_completed" + + +def test_has_pending_task(db_session: Session): + """Test pending task detection (去重).""" + eval_db = IntelligentEvalDB( + name="test", + target_id="target1", + status=IntelligentEvalStatus.EXECUTING.value, + ) + db_session.add(eval_db) + db_session.commit() + + # No pending task initially + assert task_queue._has_pending_task(eval_db.id, db_session) is False + + # Add pending task + task = IntelligentEvalTaskQueueDB( + eval_id=eval_db.id, + status="pending", + priority=1, + reason="slot_due", + ) + db_session.add(task) + db_session.commit() + + # Should detect pending task + assert task_queue._has_pending_task(eval_db.id, db_session) is True + + +def test_scan_and_enqueue_tasks(db_session: Session): + """Test task scanning and enqueueing.""" + # Create eval that needs attention + eval_db = IntelligentEvalDB( + name="test", + target_id="target1", + status=IntelligentEvalStatus.EXECUTING.value, + started_at=utc_now() - timedelta(hours=9), + ) + eval_db.set_plan({ + "time_distribution": [{"time_slot": "8-10h", "sessions": 2}], + "estimated_sessions": 2, + }) + db_session.add(eval_db) + db_session.commit() + + # Scan and enqueue + enqueued = task_queue.scan_and_enqueue_tasks(db_session) + assert enqueued == 1 + + # Verify task created + task = db_session.exec( + select(IntelligentEvalTaskQueueDB).where(IntelligentEvalTaskQueueDB.eval_id == eval_db.id) + ).first() + assert task is not None + assert task.status == "pending" + assert task.reason == "slot_due" + assert task.priority < 100 # Should have reduced priority + + # Scan again, should not create duplicate + enqueued = task_queue.scan_and_enqueue_tasks(db_session) + assert enqueued == 0 + + +def test_get_next_task(db_session: Session): + """Test getting next task (highest priority).""" + # Create tasks with different priorities + task1 = IntelligentEvalTaskQueueDB( + eval_id="eval1", + status="pending", + priority=50, + reason="slot_due", + ) + task2 = IntelligentEvalTaskQueueDB( + eval_id="eval2", + status="pending", + priority=10, # Higher priority (lower number) + reason="slot_due", + ) + task3 = IntelligentEvalTaskQueueDB( + eval_id="eval3", + status="assigned", # Not pending + priority=1, + reason="slot_due", + ) + db_session.add_all([task1, task2, task3]) + db_session.commit() + + # Should get task2 (priority 10) + next_task = task_queue.get_next_task(db_session) + assert next_task is not None + assert next_task.id == task2.id + + +def test_assign_task(db_session: Session): + """Test task assignment.""" + task = IntelligentEvalTaskQueueDB( + eval_id="eval1", + status="pending", + priority=1, + reason="slot_due", + ) + db_session.add(task) + db_session.commit() + + # Assign task + success = task_queue.assign_task(task.id, "cron1", db_session) + assert success is True + + # Verify assignment + db_session.refresh(task) + assert task.status == "assigned" + assert task.assigned_cron_id == "cron1" + assert task.assigned_at is not None + + # Try to assign again (should fail) + success = task_queue.assign_task(task.id, "cron2", db_session) + assert success is False + + +def test_complete_task(db_session: Session): + """Test task completion.""" + task = IntelligentEvalTaskQueueDB( + eval_id="eval1", + status="assigned", + priority=1, + reason="slot_due", + assigned_cron_id="cron1", + ) + db_session.add(task) + db_session.commit() + + # Complete task successfully + success = task_queue.complete_task(task.id, True, None, db_session) + assert success is True + + # Verify completion + db_session.refresh(task) + assert task.status == "completed" + assert task.completed_at is not None + assert task.error is None + + # Complete task with error + task2 = IntelligentEvalTaskQueueDB( + eval_id="eval2", + status="assigned", + priority=1, + reason="slot_due", + assigned_cron_id="cron1", + ) + db_session.add(task2) + db_session.commit() + + success = task_queue.complete_task(task2.id, False, "test error", db_session) + assert success is True + + db_session.refresh(task2) + assert task2.status == "failed" + assert task2.error == "test error"