From b0969ae58237c9e29c3736021502076ec330e21e Mon Sep 17 00:00:00 2001 From: sinohqb Date: Mon, 17 Aug 2026 13:06:27 +0800 Subject: [PATCH] feat(intelligent-eval): backfill decision logs for completed evals MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit COMPLETED 状态的评估（历史/异常路径）可能完全没有决策日志， 导致旧报告决策过程为空。扩展 _supplement_decision_logs 支持 COMPLETED：按时段补 execute_session + 补 start_analysis（历史回填）， scan loop 每分钟自动回填，无需一次性脚本。幂等，只补缺失类型。 --- backend/agenteval/web/app.py | 120 ++++++++++++++---- .../test_intelligent_eval_scan_scheduler.py | 89 ++++++++++++- 2 files changed, 177 insertions(+), 32 deletions(-) diff --git a/backend/agenteval/web/app.py b/backend/agenteval/web/app.py index 8e9cdd9..59fc057 100644 --- a/backend/agenteval/web/app.py +++ b/backend/agenteval/web/app.py @@ -47,9 +47,14 @@ def _supplement_decision_logs(session) -> int: 方案③的决策日志由 OpenClaw agent 上报（LLM 自主，尽力而为）——异常路径 （如卡死恢复后重试）agent 可能跳过上报，导致决策过程页面为空。这里按评估 - 状态推导决策并补录：欠账时补 execute_session，所有会话完成后补 - start_analysis。只补"该类型缺失"的，不重复；且只记录状态，不改变 agent - 的实际执行。 + 状态推导决策并补录： + + - EXECUTING：欠账（completed < estimated）补 execute_session，所有会话 + 完成后补 start_analysis。 + - COMPLETED：历史评估/异常路径可能完全没有决策日志，回填 execute_session + （按 plan 时段逐条）+ start_analysis，让旧报告也有决策过程可看。 + + 只补"该类型缺失"的，不重复；且只记录状态，不改变 agent 的实际执行。 Returns: 补录的决策日志条数。 @@ -61,39 +66,90 @@ def _supplement_decision_logs(session) -> int: from agenteval.storage.db import IntelligentEvalDB, IntelligentEvalDecisionLogDB, IntelligentEvalSessionDB evals = session.exec( - select(IntelligentEvalDB).where(IntelligentEvalDB.status == IntelligentEvalStatus.EXECUTING.value) + select(IntelligentEvalDB).where( + IntelligentEvalDB.status.in_( + [ + IntelligentEvalStatus.EXECUTING.value, + IntelligentEvalStatus.COMPLETED.value, + ] + ) + ) ).all() added = 0 for ev in evals: plan = ev.get_plan() if ev.plan else {} estimated = plan.get("estimated_sessions", 0) - sessions = session.exec( - select(IntelligentEvalSessionDB).where(IntelligentEvalSessionDB.eval_id == ev.id) - ).all() + sessions = session.exec(select(IntelligentEvalSessionDB).where(IntelligentEvalSessionDB.eval_id == ev.id)).all() completed = sum(1 for s in sessions if s.status == "completed") types = { - l.decision_type - for l in session.exec( - select(IntelligentEvalDecisionLogDB).where( - IntelligentEvalDecisionLogDB.eval_id == ev.id - ) + x.decision_type + for x in session.exec( + select(IntelligentEvalDecisionLogDB).where(IntelligentEvalDecisionLogDB.eval_id == ev.id) ).all() } - if "execute_session" not in types and completed < estimated: - create_decision_log( - ev.id, "execute_session", "平台兜底：时段欠账需执行会话", "platform", - {"platform_supplemented": True, "completed": completed, "estimated": estimated}, - session, - ) - added += 1 - elif "start_analysis" not in types and sessions and completed >= estimated: - create_decision_log( - ev.id, "start_analysis", "平台兜底：所有会话已完成开始分析", "platform", - {"platform_supplemented": True, "completed": completed, "estimated": estimated}, - session, - ) - added += 1 + if ev.status == IntelligentEvalStatus.EXECUTING.value: + if "execute_session" not in types and completed < estimated: + create_decision_log( + ev.id, + "execute_session", + "平台兜底：时段欠账需执行会话", + "platform", + {"platform_supplemented": True, "completed": completed, "estimated": estimated}, + session, + ) + added += 1 + elif "start_analysis" not in types and sessions and completed >= estimated: + create_decision_log( + ev.id, + "start_analysis", + "平台兜底：所有会话已完成开始分析", + "platform", + {"platform_supplemented": True, "completed": completed, "estimated": estimated}, + session, + ) + added += 1 + elif ev.status == IntelligentEvalStatus.COMPLETED.value: + # 历史回填：completed 评估决策日志全缺失时，按时段补 execute_session + if "execute_session" not in types: + slots = plan.get("time_distribution") or [] + if slots: + for slot in slots: + create_decision_log( + ev.id, + "execute_session", + f"平台兜底：时段{slot.get('time_slot', '')}执行会话（历史回填）", + "platform", + { + "platform_supplemented": True, + "time_slot": slot.get("time_slot"), + "sessions": slot.get("sessions"), + "completed": completed, + "estimated": estimated, + }, + session, + ) + added += 1 + else: + create_decision_log( + ev.id, + "execute_session", + "平台兜底：执行会话（历史回填）", + "platform", + {"platform_supplemented": True, "completed": completed, "estimated": estimated}, + session, + ) + added += 1 + if "start_analysis" not in types and sessions: + create_decision_log( + ev.id, + "start_analysis", + "平台兜底：所有会话已完成开始分析（历史回填）", + "platform", + {"platform_supplemented": True, "completed": completed, "estimated": estimated}, + session, + ) + added += 1 return added @@ -127,8 +183,16 @@ async def _trigger_intelligent_worker() -> bool: proc = await asyncio.to_thread( subprocess.run, [ - "docker", "exec", "openclaw-eval", "openclaw", "agent", - "--agent", "main", "-m", worker_msg, "--json", + "docker", + "exec", + "openclaw-eval", + "openclaw", + "agent", + "--agent", + "main", + "-m", + worker_msg, + "--json", ], capture_output=True, text=True, diff --git a/tests/integration/test_intelligent_eval_scan_scheduler.py b/tests/integration/test_intelligent_eval_scan_scheduler.py index 5564658..ccd36f9 100644 --- a/tests/integration/test_intelligent_eval_scan_scheduler.py +++ b/tests/integration/test_intelligent_eval_scan_scheduler.py @@ -5,6 +5,7 @@ wakes every minute but could never pull a task. The lifespan now starts an asyncio background task that scans executing evals every 60s. This test verifies that on application startup the scan is actually invoked. """ + from unittest.mock import MagicMock from fastapi.testclient import TestClient @@ -117,8 +118,10 @@ def test_supplement_execute_session_log(monkeypatch, db_session): from agenteval.storage.db import IntelligentEvalDB ev = IntelligentEvalDB( - name="supp-eval", target_id="t1", - status=IntelligentEvalStatus.EXECUTING.value, started_at=__import__("agenteval.storage.db", fromlist=["utc_now"]).utc_now(), + name="supp-eval", + target_id="t1", + status=IntelligentEvalStatus.EXECUTING.value, + started_at=__import__("agenteval.storage.db", fromlist=["utc_now"]).utc_now(), ) ev.set_plan({"time_distribution": [{"time_slot": "0-1h", "sessions": 1}], "estimated_sessions": 1}) db_session.add(ev) @@ -136,8 +139,10 @@ def test_supplement_start_analysis_log(monkeypatch, db_session): from agenteval.storage.db import IntelligentEvalDB, IntelligentEvalSessionDB, utc_now ev = IntelligentEvalDB( - name="supp-eval2", target_id="t1", - status=IntelligentEvalStatus.EXECUTING.value, started_at=utc_now(), + name="supp-eval2", + target_id="t1", + status=IntelligentEvalStatus.EXECUTING.value, + started_at=utc_now(), ) ev.set_plan({"time_distribution": [{"time_slot": "0-1h", "sessions": 1}], "estimated_sessions": 1}) db_session.add(ev) @@ -147,3 +152,79 @@ def test_supplement_start_analysis_log(monkeypatch, db_session): db_session.commit() assert app_mod._supplement_decision_logs(db_session) == 1 + + +def test_supplement_completed_backfill(monkeypatch, db_session): + """Completed eval with no decision logs → backfill execute_session + start_analysis.""" + import agenteval.web.app as app_mod + from agenteval.intelligent_eval.models import IntelligentEvalStatus + from agenteval.storage.db import ( + IntelligentEvalDB, + IntelligentEvalDecisionLogDB, + IntelligentEvalSessionDB, + utc_now, + ) + from sqlmodel import select + + ev = IntelligentEvalDB( + name="supp-completed", + target_id="t1", + status=IntelligentEvalStatus.COMPLETED.value, + started_at=utc_now(), + ) + ev.set_plan( + { + "time_distribution": [ + {"time_slot": "0-1h", "sessions": 1}, + {"time_slot": "1-2h", "sessions": 1}, + ], + "estimated_sessions": 2, + } + ) + db_session.add(ev) + db_session.commit() + for _ in range(2): + s = IntelligentEvalSessionDB(eval_id=ev.id, target_id=ev.target_id, status="completed", goal="g") + db_session.add(s) + db_session.commit() + + # 首轮：2 时段 → 2 条 execute_session + 1 条 start_analysis + assert app_mod._supplement_decision_logs(db_session) == 3 + # 幂等：二次调用不再补 + assert app_mod._supplement_decision_logs(db_session) == 0 + + logs = db_session.exec( + select(IntelligentEvalDecisionLogDB).where(IntelligentEvalDecisionLogDB.eval_id == ev.id) + ).all() + types = sorted(x.decision_type for x in logs) + assert types == ["execute_session", "execute_session", "start_analysis"] + assert all(x.cron_id == "platform" for x in logs) + assert all(x.get_context().get("platform_supplemented") for x in logs) + + +def test_supplement_completed_skips_when_already_logged(monkeypatch, db_session): + """Completed eval that already has worker-reported logs is not duplicated.""" + import agenteval.web.app as app_mod + from agenteval.intelligent_eval.models import IntelligentEvalStatus + from agenteval.storage.db import IntelligentEvalDB, IntelligentEvalSessionDB, utc_now + + ev = IntelligentEvalDB( + name="supp-completed2", + target_id="t1", + status=IntelligentEvalStatus.COMPLETED.value, + started_at=utc_now(), + ) + ev.set_plan({"time_distribution": [{"time_slot": "0-1h", "sessions": 1}], "estimated_sessions": 1}) + db_session.add(ev) + db_session.commit() + s = IntelligentEvalSessionDB(eval_id=ev.id, target_id=ev.target_id, status="completed", goal="g") + db_session.add(s) + db_session.commit() + + from agenteval.intelligent_eval.decision_logs import create_decision_log + + # 已有 worker 正常上报的两条日志 + create_decision_log(ev.id, "execute_session", "时段0-1h欠账1个会话，需要执行会话", "manual-run-1", {}, db_session) + create_decision_log(ev.id, "start_analysis", "所有会话已完成，开始分析", "manual-run-1", {}, db_session) + + assert app_mod._supplement_decision_logs(db_session) == 0