From 30b9cac22485ffa38064c964fcd6d7986cabd00e Mon Sep 17 00:00:00 2001
From: sinohqb <huangqiaobin2012@gmail.com>
Date: Wed, 12 Aug 2026 10:01:56 +0800
Subject: [PATCH] feat(intelligent-eval): implement worker skill and APIs
 (ticket 03)

- Create agenteval-intelligent-worker SKILL.md with decision logic
- Implement heartbeat API (POST /api/openclaw/crons/{id}/heartbeat)
- Implement decision log API (POST /api/intelligent-evals/{id}/decision-logs)
- Skill includes idle/busy state management and cron state handling
- Deployment script already syncs skills automatically
- Add 6 integration tests

All 784 tests passing.
---
 .../web/routers/intelligent_evals.py          |  46 ++++
 .../web/routers/openclaw_cron_pool.py         |  39 ++-
 .../agenteval-intelligent-worker/SKILL.md     | 245 ++++++++++++++++++
 tests/integration/test_worker_skill_api.py    | 213 +++++++++++++++
 4 files changed, 542 insertions(+), 1 deletion(-)
 create mode 100644 backend/plugins/openclaw/skills/agenteval-intelligent-worker/SKILL.md
 create mode 100644 tests/integration/test_worker_skill_api.py

diff --git a/backend/agenteval/web/routers/intelligent_evals.py b/backend/agenteval/web/routers/intelligent_evals.py
index 30d3f69..0fa2cf4 100644
--- a/backend/agenteval/web/routers/intelligent_evals.py
+++ b/backend/agenteval/web/routers/intelligent_evals.py
@@ -298,3 +298,49 @@ async def complete_task(
     if not completed:
         raise HTTPException(status_code=404, detail="task not found")
     return {"success": True}
+
+
+class DecisionLogRequest(BaseModel):
+    decision_type: str = Field(min_length=1)  # execute_session / wait / start_analysis
+    reason: str = Field(min_length=1)
+    context: dict[str, Any] = Field(default_factory=dict)
+    cron_id: str = Field(min_length=1)
+
+
+@router.post("/{eval_id}/decision-logs")
+async def create_decision_log(
+    eval_id: str,
+    request: DecisionLogRequest,
+    session: Session = Depends(get_db),
+) -> dict:
+    """Create a decision log entry for an intelligent eval."""
+    from agenteval.storage.db import IntelligentEvalDecisionLogDB
+
+    # Verify eval exists
+    from agenteval.storage.db import IntelligentEvalDB
+
+    eval_db = session.get(IntelligentEvalDB, eval_id)
+    if eval_db is None:
+        raise HTTPException(status_code=404, detail=f"intelligent eval {eval_id} not found")
+
+    # Create decision log
+    log = IntelligentEvalDecisionLogDB(
+        eval_id=eval_id,
+        decision_type=request.decision_type,
+        reason=request.reason,
+        cron_id=request.cron_id,
+    )
+    log.set_context(request.context)
+    session.add(log)
+    session.commit()
+    session.refresh(log)
+
+    return {
+        "id": log.id,
+        "eval_id": log.eval_id,
+        "decision_type": log.decision_type,
+        "reason": log.reason,
+        "context": log.get_context(),
+        "cron_id": log.cron_id,
+        "created_at": log.created_at.isoformat() if log.created_at else None,
+    }
diff --git a/backend/agenteval/web/routers/openclaw_cron_pool.py b/backend/agenteval/web/routers/openclaw_cron_pool.py
index 532bd07..0cdef94 100644
--- a/backend/agenteval/web/routers/openclaw_cron_pool.py
+++ b/backend/agenteval/web/routers/openclaw_cron_pool.py
@@ -1,11 +1,14 @@
 """API routes for OpenClaw cron pool management."""
 
+from datetime import datetime
+
 from fastapi import APIRouter, Depends, HTTPException
 from pydantic import BaseModel, Field
-from sqlmodel import Session
+from sqlmodel import Session, select
 
 from agenteval.intelligent_eval import cron_pool
 from agenteval.intelligent_eval.openclaw_client import OpenClawClient
+from agenteval.storage.db import OpenClawCronPoolDB, utc_now
 from agenteval.web.deps import get_db
 
 router = APIRouter()
@@ -15,6 +18,11 @@ class ScaleRequest(BaseModel):
     target_size: int = Field(ge=1, le=50)
 
 
+class HeartbeatRequest(BaseModel):
+    status: str  # idle / busy
+    current_eval_id: str | None = None
+
+
 @router.get("/cron-pool")
 async def get_cron_pool_status(session: Session = Depends(get_db)) -> dict:
     """Get cron pool status."""
@@ -63,3 +71,32 @@ async def auto_scale_pool(session: Session = Depends(get_db)) -> dict:
         "scaled_up": scaled_up,
         "scaled_down": scaled_down,
     }
+
+
+@router.post("/crons/{cron_id}/heartbeat")
+async def report_heartbeat(
+    cron_id: str,
+    request: HeartbeatRequest,
+    session: Session = Depends(get_db),
+) -> dict:
+    """Report cron heartbeat.
+
+    Updates the cron's last_active_at timestamp and current status.
+    """
+    # Find cron by openclaw_cron_id
+    cron = session.exec(
+        select(OpenClawCronPoolDB).where(OpenClawCronPoolDB.openclaw_cron_id == cron_id)
+    ).first()
+
+    if cron is None:
+        raise HTTPException(status_code=404, detail=f"cron {cron_id} not found")
+
+    # Update heartbeat
+    cron.last_active_at = utc_now()
+    cron.status = request.status
+    cron.current_eval_id = request.current_eval_id
+    cron.updated_at = utc_now()
+
+    session.commit()
+
+    return {"success": True}
diff --git a/backend/plugins/openclaw/skills/agenteval-intelligent-worker/SKILL.md b/backend/plugins/openclaw/skills/agenteval-intelligent-worker/SKILL.md
new file mode 100644
index 0000000..aa1f84a
--- /dev/null
+++ b/backend/plugins/openclaw/skills/agenteval-intelligent-worker/SKILL.md
@@ -0,0 +1,245 @@
+---
+name: agenteval-intelligent-worker
+description: 智能评估工作单元：从平台任务队列取任务，执行决策逻辑，上报心跳和决策日志
+---
+
+你是智能评估的工作单元（Worker），每分钟被 cron 唤醒一次。你的职责是：从平台任务队列取任务 → 执行决策逻辑 → 上报结果。
+
+所有操作必须走 AgentEvalTool 标准 HTTP API（禁止直接调 CLI 或操作数据库）。
+
+平台可能启用了 API Key 鉴权。每次执行命令前先读取密钥（文件不存在则为空，不影响未启用鉴权的环境）：
+
+```bash
+KEY=$(cat ~/.openclaw/agenteval-api-key 2>/dev/null)
+```
+
+以下所有 curl 命令都必须带 `-H "X-API-Key: $KEY"`。
+
+## 你的 Cron State
+
+OpenClaw 的 cron state 是一个 JSON 对象，用于在多次唤醒之间保持状态。你的 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": "..."
+    }
+  ]
+}
+```
+
+**读取 state**：OpenClaw 会在每次唤醒时注入 `trigger.state`（只读）。
+**更新 state**：在脚本结束时输出 JSON 到 stdout，格式：`{"state": {...}}`。
+
+## 工作流程
+
+### 第一步：读取当前状态
+
+从 `trigger.state` 读取你的当前状态：
+
+- `status`: "idle" 或 "busy"
+- `eval_id`: 当前处理的评估 ID（如果 busy）
+- `cron_id`: 你的 cron ID（从环境变量 `OPENCLAW_CRON_ID` 读取）
+
+### 第二步：上报心跳
+
+每次唤醒时，无论状态如何，都要上报心跳：
+
+```bash
+CRON_ID="${OPENCLAW_CRON_ID}"
+
+curl -s -X POST "http://agenteval:8000/api/openclaw/crons/${CRON_ID}/heartbeat" \
+  -H "X-API-Key: $KEY" \
+  -H "Content-Type: application/json" \
+  -d "{
+    \"status\": \"${STATUS}\",
+    \"current_eval_id\": \"${EVAL_ID}\"
+  }"
+```
+
+### 第三步：根据状态执行
+
+#### 如果 status == "idle"：
+
+1. 从平台取任务：
+
+```bash
+TASK_RESPONSE=$(curl -s -H "X-API-Key: $KEY" \
+  http://agenteval:8000/api/intelligent-evals/tasks/next)
+
+TASK=$(echo "$TASK_RESPONSE" | python3 -c "import sys, json; print(json.dumps(json.load(sys.stdin).get('task')))")
+
+if [ "$TASK" == "null" ]; then
+  # 无任务，本节拍结束
+  echo '{"state": {"status": "idle", "last_decision_at": "'$(date -u +%Y-%m-%dT%H:%M:%SZ)'"}}'
+  exit 0
+fi
+
+TASK_ID=$(echo "$TASK" | python3 -c "import sys, json; print(json.load(sys.stdin)['id'])")
+EVAL_ID=$(echo "$TASK" | python3 -c "import sys, json; print(json.load(sys.stdin)['eval_id'])")
+```
+
+2. 认领任务：
+
+```bash
+curl -s -X POST "http://agenteval:8000/api/intelligent-evals/tasks/${TASK_ID}/assign?cron_id=${CRON_ID}" \
+  -H "X-API-Key: $KEY"
+```
+
+3. 更新 state 为 busy：
+
+```bash
+echo '{
+  "state": {
+    "status": "busy",
+    "eval_id": "'${EVAL_ID}'",
+    "task_id": "'${TASK_ID}'",
+    "started_at": "'$(date -u +%Y-%m-%dT%H:%M:%SZ)'",
+    "last_decision_at": "'$(date -u +%Y-%m-%dT%H:%M:%SZ)'",
+    "completed_sessions": 0,
+    "decisions_history": []
+  }
+}'
+```
+
+#### 如果 status == "busy"：
+
+1. 读取评估详情：
+
+```bash
+EVAL_ID=$(echo "$TRIGGER_STATE" | python3 -c "import sys, json; print(json.load(sys.stdin)['eval_id'])")
+
+EVAL=$(curl -s -H "X-API-Key: $KEY" \
+  http://agenteval:8000/api/intelligent-evals/${EVAL_ID})
+```
+
+2. 执行决策逻辑（见下文「决策逻辑」）
+
+3. 根据决策结果调用相应的 skill：
+   - `execute_session` → 调用 `agenteval-intelligent-evaluator` skill
+   - `start_analysis` → 调用 `agenteval-intelligent-analyst` skill
+   - `wait` → 本节拍结束
+
+4. 上报决策日志：
+
+```bash
+curl -s -X POST "http://agenteval:8000/api/intelligent-evals/${EVAL_ID}/decision-logs" \
+  -H "X-API-Key: $KEY" \
+  -H "Content-Type: application/json" \
+  -d "{
+    \"decision_type\": \"${DECISION}\",
+    \"reason\": \"${REASON}\",
+    \"context\": ${CONTEXT},
+    \"cron_id\": \"${CRON_ID}\"
+  }"
+```
+
+5. 更新 state：
+
+```bash
+# 追加决策历史
+NEW_HISTORY=$(echo "$TRIGGER_STATE" | python3 -c "
+import sys, json
+state = json.load(sys.stdin)
+state['decisions_history'].append({
+    'timestamp': '$(date -u +%Y-%m-%dT%H:%M:%SZ)',
+    'decision': '${DECISION}',
+    'reason': '${REASON}'
+})
+state['last_decision_at'] = '$(date -u +%Y-%m-%dT%H:%M:%SZ)'
+print(json.dumps(state))
+")
+
+echo '{"state": '$NEW_HISTORY'}'
+```
+
+6. 检查评估是否完成：
+
+```bash
+# 读取评估状态
+EVAL_STATUS=$(echo "$EVAL" | python3 -c "import sys, json; print(json.load(sys.stdin)['status'])")
+
+if [ "$EVAL_STATUS" == "completed" ] || [ "$EVAL_STATUS" == "failed" ] || [ "$EVAL_STATUS" == "cancelled" ]; then
+  # 评估已完成，标记任务完成
+  TASK_ID=$(echo "$TRIGGER_STATE" | python3 -c "import sys, json; print(json.load(sys.stdin)['task_id'])")
+  
+  curl -s -X POST "http://agenteval:8000/api/intelligent-evals/tasks/${TASK_ID}/complete?success=true" \
+    -H "X-API-Key: $KEY"
+  
+  # 归还 cron，更新 state 为 idle
+  echo '{
+    "state": {
+      "status": "idle",
+      "eval_id": null,
+      "task_id": null,
+      "last_decision_at": "'$(date -u +%Y-%m-%dT%H:%M:%SZ)'"
+    }
+  }'
+fi
+```
+
+## 决策逻辑
+
+你需要根据当前评估的状态，自主决定"现在该做什么"。决策依据：
+
+1. **读取评估详情**：
+   - `status`: 评估状态（executing / completed / failed / cancelled）
+   - `plan.time_distribution`: 时间分布计划
+   - `started_at`: 评估开始时间
+
+2. **读取会话列表**：
+
+```bash
+SESSIONS=$(curl -s -H "X-API-Key: $KEY" \
+  http://agenteval:8000/api/intelligent-evals/${EVAL_ID}/sessions)
+```
+
+3. **分析当前情况**：
+   - 计算当前时间偏移：`current_offset = now - started_at`
+   - 判断当前处于哪个时段（早高峰/午间/晚间）
+   - 统计当前时段已完成的会话数
+   - 检查是否有严重问题（severity == "high"）
+
+4. **决策规则**：
+
+   - **如果评估状态不是 executing** → 返回 "wait"，原因 "评估已完成或取消"
+   
+   - **如果当前时段有欠账**（计划 2 个会话，实际 1 个）→ 返回 "execute_session"，原因 "时段 X 欠账 Y 个会话"
+   
+   - **如果发现严重问题**（某个会话的 verdict 包含 high severity）→ 返回 "execute_session"，原因 "发现严重问题，需要深入挖掘"
+   
+   - **如果所有会话已完成** → 返回 "start_analysis"，原因 "所有会话已完成，开始分析"
+   
+   - **否则** → 返回 "wait"，原因 "当前时段无欠账，等待下一时段"
+
+5. **输出决策**：
+   - 决策类型：`execute_session` / `wait` / `start_analysis`
+   - 决策原因：一句话说明为什么做这个决策
+   - 决策上下文：JSON 对象，包含当前时段、已完成会话数、欠账数等
+
+## 错误处理
+
+- 如果 API 调用失败（网络错误、404、500 等），记录错误到 decisions_history，但不改变 state
+- 如果连续 3 次 API 调用失败，将 state 的 status 改为 "idle"，放弃当前任务
+- 如果评估状态为 "cancelled"，立即标记任务完成，归还 cron
+
+## 调试
+
+- 所有 API 调用的响应都应该记录到 decisions_history
+- 使用 `echo` 输出调试信息到 stderr（不会影响 state）
+- 可以在 state 中添加自定义字段（如 `debug_info`）用于调试
+
+## 注意事项
+
+- 请将 <eval_id>、<session_id> 等占位符替换为实际值
+- 所有时间戳使用 ISO 8601 格式（UTC）
+- State 大小限制为 16KB，注意不要存储过多历史记录（最多保留最近 50 条决策）
+- 如果 decisions_history 超过 50 条，删除最旧的记录
diff --git a/tests/integration/test_worker_skill_api.py b/tests/integration/test_worker_skill_api.py
new file mode 100644
index 0000000..96a5ad6
--- /dev/null
+++ b/tests/integration/test_worker_skill_api.py
@@ -0,0 +1,213 @@
+"""Integration tests for worker skill APIs (heartbeat, decision logs)."""
+
+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,
+    IntelligentEvalDecisionLogDB,
+    OpenClawCronPoolDB,
+    utc_now,
+)
+from agenteval.web.app import app
+from agenteval.web.deps import get_db
+
+
+@pytest.fixture()
+def client(tmp_path):
+    """Create a TestClient with a fresh database."""
+    from agenteval.storage.db import (  # noqa: F401
+        IntelligentEvalDB,
+        IntelligentEvalDecisionLogDB,
+        OpenClawCronPoolDB,
+    )
+
+    engine = create_engine(
+        f"sqlite:///{tmp_path / 'test.db'}",
+        connect_args={"check_same_thread": False},
+    )
+    SQLModel.metadata.create_all(engine)
+    session = Session(engine)
+
+    def override_get_db():
+        try:
+            yield session
+        finally:
+            pass
+
+    app.dependency_overrides[get_db] = override_get_db
+    client = TestClient(app)
+    yield client
+    app.dependency_overrides.clear()
+    session.close()
+    engine.dispose()
+
+
+@pytest.fixture()
+def db_session(client):
+    """Get the database session from the client fixture."""
+    return next(app.dependency_overrides[get_db]())
+
+
+def test_heartbeat_idle(client: TestClient, db_session: Session):
+    """Test heartbeat from idle cron."""
+    # Create cron
+    cron = OpenClawCronPoolDB(
+        openclaw_cron_id="cron-123",
+        status="idle",
+        last_active_at=utc_now() - timedelta(minutes=5),
+    )
+    db_session.add(cron)
+    db_session.commit()
+
+    old_active_at = cron.last_active_at
+
+    # Report heartbeat
+    response = client.post(
+        "/api/openclaw/crons/cron-123/heartbeat",
+        json={"status": "idle", "current_eval_id": None},
+    )
+    assert response.status_code == 200
+    assert response.json() == {"success": True}
+
+    # Verify heartbeat updated
+    db_session.refresh(cron)
+    assert cron.last_active_at > old_active_at
+    assert cron.status == "idle"
+    assert cron.current_eval_id is None
+
+
+def test_heartbeat_busy(client: TestClient, db_session: Session):
+    """Test heartbeat from busy cron."""
+    # Create cron
+    cron = OpenClawCronPoolDB(
+        openclaw_cron_id="cron-456",
+        status="busy",
+        current_eval_id="eval-789",
+        last_active_at=utc_now() - timedelta(minutes=2),
+    )
+    db_session.add(cron)
+    db_session.commit()
+
+    # Report heartbeat
+    response = client.post(
+        "/api/openclaw/crons/cron-456/heartbeat",
+        json={"status": "busy", "current_eval_id": "eval-789"},
+    )
+    assert response.status_code == 200
+
+    # Verify heartbeat updated
+    db_session.refresh(cron)
+    assert cron.status == "busy"
+    assert cron.current_eval_id == "eval-789"
+
+
+def test_heartbeat_not_found(client: TestClient):
+    """Test heartbeat from non-existent cron."""
+    response = client.post(
+        "/api/openclaw/crons/nonexistent/heartbeat",
+        json={"status": "idle", "current_eval_id": None},
+    )
+    assert response.status_code == 404
+
+
+def test_create_decision_log(client: TestClient, db_session: Session):
+    """Test creating a decision log."""
+    # Create eval
+    eval_db = IntelligentEvalDB(
+        name="test",
+        target_id="target1",
+        status=IntelligentEvalStatus.EXECUTING.value,
+    )
+    db_session.add(eval_db)
+    db_session.commit()
+
+    # Create decision log
+    response = client.post(
+        f"/api/intelligent-evals/{eval_db.id}/decision-logs",
+        json={
+            "decision_type": "execute_session",
+            "reason": "时段 8-10h 欠账 2 个会话",
+            "context": {
+                "current_slot": "8-10h",
+                "deficit": 2,
+                "completed_sessions": 1,
+            },
+            "cron_id": "cron-123",
+        },
+    )
+    assert response.status_code == 200
+
+    data = response.json()
+    assert data["eval_id"] == eval_db.id
+    assert data["decision_type"] == "execute_session"
+    assert data["reason"] == "时段 8-10h 欠账 2 个会话"
+    assert data["context"]["current_slot"] == "8-10h"
+    assert data["cron_id"] == "cron-123"
+
+    # Verify log saved to DB
+    log = db_session.exec(
+        select(IntelligentEvalDecisionLogDB).where(IntelligentEvalDecisionLogDB.eval_id == eval_db.id)
+    ).first()
+    assert log is not None
+    assert log.decision_type == "execute_session"
+
+
+def test_create_decision_log_eval_not_found(client: TestClient):
+    """Test creating decision log for non-existent eval."""
+    response = client.post(
+        "/api/intelligent-evals/nonexistent/decision-logs",
+        json={
+            "decision_type": "wait",
+            "reason": "test",
+            "context": {},
+            "cron_id": "cron-123",
+        },
+    )
+    assert response.status_code == 404
+
+
+def test_decision_log_multiple_entries(client: TestClient, db_session: Session):
+    """Test creating multiple decision logs for same eval."""
+    # Create eval
+    eval_db = IntelligentEvalDB(
+        name="test",
+        target_id="target1",
+        status=IntelligentEvalStatus.EXECUTING.value,
+    )
+    db_session.add(eval_db)
+    db_session.commit()
+
+    # Create 3 decision logs
+    decisions = [
+        ("execute_session", "时段到期"),
+        ("wait", "当前时段无欠账"),
+        ("start_analysis", "所有会话完成"),
+    ]
+
+    for decision_type, reason in decisions:
+        response = client.post(
+            f"/api/intelligent-evals/{eval_db.id}/decision-logs",
+            json={
+                "decision_type": decision_type,
+                "reason": reason,
+                "context": {},
+                "cron_id": "cron-123",
+            },
+        )
+        assert response.status_code == 200
+
+    # Verify all logs saved
+    logs = db_session.exec(
+        select(IntelligentEvalDecisionLogDB)
+        .where(IntelligentEvalDecisionLogDB.eval_id == eval_db.id)
+        .order_by(IntelligentEvalDecisionLogDB.created_at)
+    ).all()
+    assert len(logs) == 3
+    assert logs[0].decision_type == "execute_session"
+    assert logs[1].decision_type == "wait"
+    assert logs[2].decision_type == "start_analysis"