diff --git a/backend/agenteval/intelligent_eval/decision_logs.py b/backend/agenteval/intelligent_eval/decision_logs.py index 2d40b79..824b05b 100644 --- a/backend/agenteval/intelligent_eval/decision_logs.py +++ b/backend/agenteval/intelligent_eval/decision_logs.py @@ -7,7 +7,12 @@ from typing import Any from sqlmodel import Session, select -from agenteval.storage.db import IntelligentEvalDB, IntelligentEvalDecisionLogDB +from agenteval.intelligent_eval.models import IntelligentEvalStatus +from agenteval.storage.db import ( + IntelligentEvalDB, + IntelligentEvalDecisionLogDB, + IntelligentEvalSessionDB, +) def _log_to_dict(log: IntelligentEvalDecisionLogDB) -> dict: @@ -22,7 +27,12 @@ def _log_to_dict(log: IntelligentEvalDecisionLogDB) -> dict: } -def create_decision_log( +def _require_eval(eval_id: str, session: Session) -> None: + if session.get(IntelligentEvalDB, eval_id) is None: + raise LookupError(f"intelligent eval {eval_id} not found") + + +def _append_row( eval_id: str, decision_type: str, reason: str, @@ -30,30 +40,6 @@ def create_decision_log( context: dict[str, Any], session: Session, ) -> dict: - """Create a decision log entry, or return the existing one if the - (eval_id, decision_type, context) tuple is already recorded. - - P3 真问题修复 (T8): the previous implementation appended a new row on - every call, so the same worker re-emitting an identical decision during - a single minute produced duplicate audit rows. Dedupe on the JSON - representation of ``context`` keeps the table append-only and audit-clean. - - Raises ``LookupError`` if eval not found. - """ - eval_db = session.get(IntelligentEvalDB, eval_id) - if eval_db is None: - raise LookupError(f"intelligent eval {eval_id} not found") - - # Dedupe: same (eval, decision_type, context) → return existing. - for existing in session.exec( - select(IntelligentEvalDecisionLogDB).where( - IntelligentEvalDecisionLogDB.eval_id == eval_id, - IntelligentEvalDecisionLogDB.decision_type == decision_type, - ) - ).all(): - if existing.get_context() == context: - return _log_to_dict(existing) - log = IntelligentEvalDecisionLogDB( eval_id=eval_id, decision_type=decision_type, @@ -67,6 +53,69 @@ def create_decision_log( return _log_to_dict(log) +def create_decision_log( + eval_id: str, + decision_type: str, + reason: str, + cron_id: str, + context: dict[str, Any], + session: Session, +) -> dict: + """Agent-reporting entry: create a decision log, or return the existing + one if the (eval_id, decision_type, context) tuple is already recorded. + + P3 真问题修复 (T8 / Gitea #6): agent 会在同一分钟内重复上报相同决策, + 按 context JSON 去重保表干净。**去重只服务 agent 上报路径**——平台落账 + 每次都是新事实,用 ``append_decision_log``,调用方无需知道去重存在。 + + Raises ``LookupError`` if eval not found. + """ + _require_eval(eval_id, session) + + # Dedupe: same (eval, decision_type, context) → return existing. + for existing in session.exec( + select(IntelligentEvalDecisionLogDB).where( + IntelligentEvalDecisionLogDB.eval_id == eval_id, + IntelligentEvalDecisionLogDB.decision_type == decision_type, + ) + ).all(): + if existing.get_context() == context: + return _log_to_dict(existing) + + return _append_row(eval_id, decision_type, reason, cron_id, context, session) + + +def append_decision_log( + eval_id: str, + decision_type: str, + reason: str, + cron_id: str, + context: dict[str, Any], + session: Session, +) -> dict: + """Platform-bookkeeping entry: 纯追加,不去重。 + + 平台每次落账都是新事实(attempt 递增、task_id 不同、闸门判定), + 相同 context 也总是追加一行。 + + Raises ``LookupError`` if eval not found. + """ + _require_eval(eval_id, session) + return _append_row(eval_id, decision_type, reason, cron_id, context, session) + + +def count_decisions(eval_id: str, decision_type: str, session: Session) -> int: + """该评估某类型决策日志的条数(平台闸门计数与 attempt 落账的计数原语)。""" + return len( + session.exec( + select(IntelligentEvalDecisionLogDB).where( + IntelligentEvalDecisionLogDB.eval_id == eval_id, + IntelligentEvalDecisionLogDB.decision_type == decision_type, + ) + ).all() + ) + + def list_decision_logs(eval_id: str, session: Session) -> list[dict]: """List decision logs for an eval. Raises ``LookupError`` if eval not found.""" eval_db = session.get(IntelligentEvalDB, eval_id) @@ -80,3 +129,108 @@ def list_decision_logs(eval_id: str, session: Session) -> list[dict]: ).all() return [_log_to_dict(log) for log in logs] + + +def supplement_decision_logs(session: Session) -> int: + """Platform audit backfill for decision logs. + + 方案③的决策日志由 OpenClaw agent 上报(LLM 自主,尽力而为)——异常路径 + (如卡死恢复后重试)agent 可能跳过上报,导致决策过程页面为空。这里按评估 + 状态推导决策并补录: + + - EXECUTING:欠账(completed < estimated)补 execute_session,所有会话 + 完成后补 start_analysis。 + - COMPLETED:历史评估/异常路径可能完全没有决策日志,回填 execute_session + (按 plan 时段逐条)+ start_analysis,让旧报告也有决策过程可看。 + + 只补"该类型缺失"的,不重复;且只记录状态,不改变 agent 的实际执行。 + + Returns: + 补录的决策日志条数。 + """ + evals = session.exec( + 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() + completed = sum(1 for s in sessions if s.status == "completed") + types = { + x.decision_type + for x in session.exec( + select(IntelligentEvalDecisionLogDB).where(IntelligentEvalDecisionLogDB.eval_id == ev.id) + ).all() + } + if ev.status == IntelligentEvalStatus.EXECUTING.value: + if "execute_session" not in types and completed < estimated: + _append_row( + 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: + _append_row( + 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: + _append_row( + 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: + _append_row( + ev.id, + "execute_session", + "平台兜底:执行会话(历史回填)", + "platform", + {"platform_supplemented": True, "completed": completed, "estimated": estimated}, + session, + ) + added += 1 + if "start_analysis" not in types and sessions: + _append_row( + ev.id, + "start_analysis", + "平台兜底:所有会话已完成开始分析(历史回填)", + "platform", + {"platform_supplemented": True, "completed": completed, "estimated": estimated}, + session, + ) + added += 1 + return added diff --git a/backend/agenteval/intelligent_eval/lifecycle.py b/backend/agenteval/intelligent_eval/lifecycle.py index 5339582..338f901 100644 --- a/backend/agenteval/intelligent_eval/lifecycle.py +++ b/backend/agenteval/intelligent_eval/lifecycle.py @@ -412,7 +412,7 @@ def expire_stale_running_sessions(session: Session) -> int: from sqlalchemy import func from sqlmodel import select - from agenteval.intelligent_eval.decision_logs import create_decision_log + from agenteval.intelligent_eval.decision_logs import append_decision_log from agenteval.storage.db import IntelligentEvalMessageDB, IntelligentEvalSessionDB now = utc_now() @@ -435,7 +435,7 @@ def expire_stale_running_sessions(session: Session) -> int: row.status = IntelligentEvalSessionStatus.EXPIRED.value row.closed_at = now expired += 1 - create_decision_log( + append_decision_log( row.eval_id, "session_expired", f"平台兜底:会话 {SESSION_IDLE_EXPIRE_MINUTES} 分钟无新轮次,置为过期(不完整证据)", @@ -486,7 +486,6 @@ def record_planner_triggers(session: Session) -> int: 触发次数即日志条数(含触发失败——attempt 在触发动作发生前落账),供 enforce_planning_gates 计闸,也让决策过程页可见平台尝试过多少次。 - context 带递增 attempt 序号避免 create_decision_log 去重合并。 ADR-0011:同一评估 10min 冷却内的扫描节拍不重复触发(孤儿 agent 双管), 冷却中的评估不计次数。 @@ -495,8 +494,8 @@ def record_planner_triggers(session: Session) -> int: """ from sqlmodel import select - from agenteval.intelligent_eval.decision_logs import create_decision_log - from agenteval.storage.db import IntelligentEvalDB, IntelligentEvalDecisionLogDB + from agenteval.intelligent_eval.decision_logs import append_decision_log, count_decisions + from agenteval.storage.db import IntelligentEvalDB now = utc_now().replace(tzinfo=None) planning = session.exec( @@ -506,18 +505,8 @@ def record_planner_triggers(session: Session) -> int: for row in planning: if _in_trigger_cooldown(session, row.id, "planner_trigger", now): continue - attempt = ( - len( - session.exec( - select(IntelligentEvalDecisionLogDB).where( - IntelligentEvalDecisionLogDB.eval_id == row.id, - IntelligentEvalDecisionLogDB.decision_type == "planner_trigger", - ) - ).all() - ) - + 1 - ) - create_decision_log( + attempt = count_decisions(row.id, "planner_trigger", session) + 1 + append_decision_log( row.id, "planner_trigger", f"平台第 {attempt} 次触发 planner 生成粗计划", @@ -541,25 +530,11 @@ def worker_trigger_candidates(session: Session) -> list[str]: def record_worker_triggers(session: Session, eval_ids: list[str]) -> None: """为本次 worker 触发覆盖的评估落账 worker_trigger(供冷却与可见性)。""" - from sqlmodel import select - - from agenteval.intelligent_eval.decision_logs import create_decision_log - from agenteval.storage.db import IntelligentEvalDecisionLogDB + from agenteval.intelligent_eval.decision_logs import append_decision_log, count_decisions for eval_id in eval_ids: - # attempt 递增保证 context 唯一——否则 create_decision_log 去重后冷却时间不前进 - attempt = ( - len( - session.exec( - select(IntelligentEvalDecisionLogDB).where( - IntelligentEvalDecisionLogDB.eval_id == eval_id, - IntelligentEvalDecisionLogDB.decision_type == "worker_trigger", - ) - ).all() - ) - + 1 - ) - create_decision_log( + attempt = count_decisions(eval_id, "worker_trigger", session) + 1 + append_decision_log( eval_id, "worker_trigger", f"平台第 {attempt} 次触发 worker 执行会话/分析", @@ -583,8 +558,8 @@ def enforce_planning_gates(session: Session) -> int: """ from sqlmodel import select - from agenteval.intelligent_eval.decision_logs import create_decision_log - from agenteval.storage.db import IntelligentEvalDB, IntelligentEvalDecisionLogDB + from agenteval.intelligent_eval.decision_logs import append_decision_log, count_decisions + from agenteval.storage.db import IntelligentEvalDB now = utc_now().replace(tzinfo=None) planning = session.exec( @@ -593,14 +568,7 @@ def enforce_planning_gates(session: Session) -> int: failed = 0 for row in planning: - attempts = len( - session.exec( - select(IntelligentEvalDecisionLogDB).where( - IntelligentEvalDecisionLogDB.eval_id == row.id, - IntelligentEvalDecisionLogDB.decision_type == "planner_trigger", - ) - ).all() - ) + attempts = count_decisions(row.id, "planner_trigger", session) entered_at = row.updated_at or row.created_at age_minutes = (now - entered_at).total_seconds() / 60 if entered_at else 0 if attempts >= PLANNING_MAX_ATTEMPTS: @@ -614,7 +582,7 @@ def enforce_planning_gates(session: Session) -> int: row.updated_at = utc_now() row.completed_at = utc_now() failed += 1 - create_decision_log( + append_decision_log( row.id, "planning_gate_failed", f"平台兜底:{reason}", @@ -651,7 +619,7 @@ def enforce_executing_ceiling(session: Session) -> int: """ from sqlmodel import select - from agenteval.intelligent_eval.decision_logs import create_decision_log + from agenteval.intelligent_eval.decision_logs import append_decision_log from agenteval.storage.db import IntelligentEvalDB now = utc_now().replace(tzinfo=None) @@ -677,7 +645,7 @@ def enforce_executing_ceiling(session: Session) -> int: row.updated_at = utc_now() row.completed_at = utc_now() failed += 1 - create_decision_log( + append_decision_log( row.id, "executing_timeout", f"平台兜底:{reason}", @@ -725,11 +693,8 @@ def evals_needing_analyst_nudge(session: Session) -> list[str]: """ from sqlmodel import select - from agenteval.storage.db import ( - IntelligentEvalDB, - IntelligentEvalDecisionLogDB, - IntelligentEvalSessionDB, - ) + from agenteval.intelligent_eval.decision_logs import count_decisions + from agenteval.storage.db import IntelligentEvalDB, IntelligentEvalSessionDB now = utc_now().replace(tzinfo=None) executing = session.exec( @@ -753,13 +718,7 @@ def evals_needing_analyst_nudge(session: Session) -> list[str]: last_closed = max(closed_moments) if (now - last_closed).total_seconds() < ANALYST_NUDGE_DELAY_MINUTES * 60: continue - nudges = session.exec( - select(IntelligentEvalDecisionLogDB).where( - IntelligentEvalDecisionLogDB.eval_id == row.id, - IntelligentEvalDecisionLogDB.decision_type == "analyst_nudge", - ) - ).all() - if len(nudges) >= ANALYST_NUDGE_MAX: + if count_decisions(row.id, "analyst_nudge", session) >= ANALYST_NUDGE_MAX: continue # 冷却复用(ADR-0011):最近的 analyst_nudge 或 worker_trigger 都会 # 唤起 agent(最长 10min),冷却期内不再催促以免堆叠孤儿 agent @@ -776,23 +735,10 @@ def evals_needing_analyst_nudge(session: Session) -> list[str]: def record_analyst_nudge(session: Session, eval_id: str) -> None: """落账一次 analyst 催促(analyst_nudge 决策日志,attempt 递增供上限计数)。""" - from sqlmodel import select + from agenteval.intelligent_eval.decision_logs import append_decision_log, count_decisions - from agenteval.intelligent_eval.decision_logs import create_decision_log - from agenteval.storage.db import IntelligentEvalDecisionLogDB - - attempt = ( - len( - session.exec( - select(IntelligentEvalDecisionLogDB).where( - IntelligentEvalDecisionLogDB.eval_id == eval_id, - IntelligentEvalDecisionLogDB.decision_type == "analyst_nudge", - ) - ).all() - ) - + 1 - ) - create_decision_log( + attempt = count_decisions(eval_id, "analyst_nudge", session) + 1 + append_decision_log( eval_id, "analyst_nudge", f"平台第 {attempt} 次催促 analyst 汇总报告(所有会话已终态)", @@ -824,31 +770,17 @@ def eval_ids_with_pending_worker_tasks(session: Session) -> list[str]: def record_trigger_failures(session: Session, *, channel: str, eval_ids: list[str], error: str) -> int: """触发失败落账:为每个受影响评估写一条 trigger_failed 决策日志。 - channel 区分 "worker" / "planner";context 带递增序号防 create_decision_log - 去重合并。失败原因截断落 reason,供决策过程页可见。 + channel 区分 "worker" / "planner";失败原因截断落 reason,供决策过程页可见。 Returns: 落账条数。 """ - from sqlmodel import select - - from agenteval.intelligent_eval.decision_logs import create_decision_log - from agenteval.storage.db import IntelligentEvalDecisionLogDB + from agenteval.intelligent_eval.decision_logs import append_decision_log, count_decisions recorded = 0 for eval_id in eval_ids: - attempt = ( - len( - session.exec( - select(IntelligentEvalDecisionLogDB).where( - IntelligentEvalDecisionLogDB.eval_id == eval_id, - IntelligentEvalDecisionLogDB.decision_type == "trigger_failed", - ) - ).all() - ) - + 1 - ) - create_decision_log( + attempt = count_decisions(eval_id, "trigger_failed", session) + 1 + append_decision_log( eval_id, "trigger_failed", f"平台触发 {channel} 失败(第 {attempt} 次):{error[:200]}", @@ -861,13 +793,13 @@ def record_trigger_failures(session: Session, *, channel: str, eval_ids: list[st def _fail_eval(session: Session, row, reason: str, decision_type: str, context: dict[str, Any]) -> None: - from agenteval.intelligent_eval.decision_logs import create_decision_log + from agenteval.intelligent_eval.decision_logs import append_decision_log row.status = IntelligentEvalStatus.FAILED.value row.plan_feedback = reason row.updated_at = utc_now() row.completed_at = utc_now() - create_decision_log( + append_decision_log( row.id, decision_type, f"平台兜底:{reason}", diff --git a/backend/agenteval/intelligent_eval/scheduler.py b/backend/agenteval/intelligent_eval/scheduler.py new file mode 100644 index 0000000..8dd39ad --- /dev/null +++ b/backend/agenteval/intelligent_eval/scheduler.py @@ -0,0 +1,311 @@ +"""Intelligent-eval scheduler runtime(触发式执行的平台节奏)。 + +从 web/app.py 抽出(ADR-0009 / ADR-0011):平台掌控节奏——每 60s 扫描 +executing 评估,watchdog 兜底、任务入队、决策日志补录,然后按需触发 +无状态 OpenClaw agent(planner/worker)执行。web 层只负责 start/stop +本 runtime,不再知道扫描循环的细节。 + +对外 interface 只有三个入口:``scheduler_runtime.start()`` / +``.stop()`` / ``scan_once()``(单次扫描节拍,即测试面)。 +""" + +import asyncio +import logging +import subprocess +import time +from typing import Callable, Optional + +from agenteval.storage.db import get_session + +_logger = logging.getLogger("agenteval") + +SCAN_INTERVAL_SECONDS = 60 + +# 触发指令必须带"立即执行"语义:`openclaw agent` 无 cron state,若只发 +# skill 名,agent 会按 skill 的"跨节拍"设计决策后等下一拍而死锁。 +WORKER_MSG = ( + "执行 agenteval-intelligent-worker skill,立即完成当前任务,不要等待下一节拍:" + "仅执行当前时间对应时段(time_distribution 中当前 offset 所在时段)内欠账的会话" + "——按评估 started_at 与当前时间精确判断当前时段,只创建该时段计划内的会话," + "绝不创建未来时段的会话,未来时段到期后平台会再次触发你;" + "仅当所有时段计划内的会话都已达到终态(completed/failed/expired)时," + "才调用 agenteval-intelligent-analyst skill 完成分析并生成报告;" + "若还有未到期的未来时段,保持等待、本次不做任何操作。" +) + +PLANNER_MSG = ( + "执行 agenteval-intelligent-planner skill,立即完成当前任务,不要等待下一节拍:" + "为 planning 状态的智能评估读取输入、产出粗计划并提交平台审批。" +) + + +async def _trigger_openclaw_agent( + role: str, + session_prefix: str, + message: str, + on_failure: Callable[[str], None], +) -> None: + """触发原语:docker exec 唤醒一个无状态 headless OpenClaw agent。 + + 方案③(避免外部 IM channel):`--deliver` 默认 false,不经 cron + delivery channel。ADR-0011 孤儿 agent 双管之一:容器内命令包 + `timeout 600`——agent 进程最多跑 10 分钟即被强杀。 + + 必须用独立 session:`--agent main` 复用 main 持久 session,多次触发 + 累积上下文缓存(~12 万 token)后 LLM 不再执行 skill 的 API 步骤, + 直接幻觉输出而不取任务。 + + 失败(子进程缺失/超时/非零退出)调用 ``on_failure(error)`` 落账。 + """ + cmd = [ + "docker", + "exec", + "openclaw-eval", + # ADR-0011:容器内 timeout 强杀,agent 最长存活 10min + "timeout", + "600", + "openclaw", + "agent", + "--agent", + "main", + "--session-id", + f"{session_prefix}-{int(time.time())}", + "-m", + message, + "--json", + ] + try: + proc = await asyncio.to_thread( + subprocess.run, + cmd, + capture_output=True, + text=True, + timeout=660, # 容器内 timeout 600 + 启动/回收余量 + ) + _logger.info("%s 触发完成 exit=%s", role, proc.returncode) + if proc.returncode != 0: + _logger.warning("%s 触发 stderr: %s", role, proc.stderr[-300:]) + on_failure(f"exit={proc.returncode} stderr={proc.stderr[-200:]}") + except Exception as exc: + _logger.warning("%s 触发失败(忽略): %s", role, exc) + on_failure(str(exc)) + + +def _record_worker_trigger_failure(error: str) -> None: + """ADR-0011:worker 触发失败落账到受影响评估(有待认领任务的评估)。""" + from agenteval.intelligent_eval.lifecycle import ( + eval_ids_with_pending_worker_tasks, + record_trigger_failures, + ) + + session = get_session() + try: + eval_ids = eval_ids_with_pending_worker_tasks(session) + if eval_ids: + record_trigger_failures(session, channel="worker", eval_ids=eval_ids, error=error) + except Exception as exc: + _logger.warning("触发失败落账失败(忽略): %s", exc) + finally: + session.close() + + +def _record_planner_trigger_failure(error: str) -> None: + """ADR-0011:planner 触发失败落账到所有 planning 评估。""" + from sqlmodel import select + + from agenteval.intelligent_eval.lifecycle import record_trigger_failures + from agenteval.intelligent_eval.models import IntelligentEvalStatus + from agenteval.storage.db import IntelligentEvalDB + + session = get_session() + try: + eval_ids = [ + row.id + for row in session.exec( + select(IntelligentEvalDB).where(IntelligentEvalDB.status == IntelligentEvalStatus.PLANNING.value) + ).all() + ] + if eval_ids: + record_trigger_failures(session, channel="planner", eval_ids=eval_ids, error=error) + except Exception as exc: + _logger.warning("触发失败落账失败(忽略): %s", exc) + finally: + session.close() + + +async def trigger_worker(nudge_eval_ids: Optional[list[str]] = None) -> bool: + """触发 OpenClaw headless agent 执行 worker skill。 + + 平台只当"触发闹钟"——有冷却期外的候选评估时,用 docker exec 唤醒 + agent,worker skill 由 agent 自主执行(决策/建会话/真实对话/上报)。 + per-eval 触发冷却 10min(worker_trigger 决策日志记录)——agent 还在跑 + 的评估不重复触发,避免 60s 扫描节奏堆叠并发 agent。 + + nudge_eval_ids:analyst 兜底催促的评估(所有会话终态、无任务队列条目), + 与候选任务评估合并去重后一同落账触发。 + + Returns: + True 若确实触发了 agent(存在冷却期外的服务对象)。 + """ + from agenteval.intelligent_eval.lifecycle import ( + record_worker_triggers, + worker_trigger_candidates, + ) + + session = get_session() + try: + trigger_ids = sorted(set(worker_trigger_candidates(session)) | set(nudge_eval_ids or [])) + if not trigger_ids: + return False + # 触发前落账(含后续失败也计入冷却),防止 agent 在跑期间重复触发 + record_worker_triggers(session, trigger_ids) + finally: + session.close() + + await _trigger_openclaw_agent("Worker", "agenteval-worker", WORKER_MSG, _record_worker_trigger_failure) + return True + + +async def trigger_planner() -> bool: + """触发 OpenClaw headless agent 执行 planner skill。 + + 方案③只自动化了 executing(worker)→ completed(analyst),planning + 阶段由本函数补上:对 planning 状态评估触发 planner skill,planner 自会 + 拉取 planning 评估列表、读取四件套、产出粗计划并 PUT /plan 提交。 + 处理完评估离开 planning 后不再触发。 + + Returns: + True 若确实触发了 agent(存在 planning 评估)。 + """ + # ADR-0011:触发前先落账 planner_trigger(含后续失败也计入双闸), + # 0 个 planning 评估时不触发。 + from agenteval.intelligent_eval.lifecycle import record_planner_triggers + + session = get_session() + try: + planning_count = record_planner_triggers(session) + finally: + session.close() + if planning_count == 0: + return False + + await _trigger_openclaw_agent("Planner", "agenteval-planner", PLANNER_MSG, _record_planner_trigger_failure) + return True + + +def _fire_and_forget(coro, name: str) -> None: + """ADR-0011:触发改为 fire-and-forget——agent 最长跑 10min,await 会把 + 60s 扫描节奏拖到 10min+。派生 asyncio task,异常在完成回调中记录 + (触发失败已由触发函数自身落账 trigger_failed)。""" + + def _done(task: asyncio.Task) -> None: + if task.cancelled(): + return + exc = task.exception() + if exc: + _logger.warning("%s 触发任务异常(忽略): %s", name, exc) + + asyncio.create_task(coro).add_done_callback(_done) + + +def scan_once() -> None: + """单次扫描节拍:watchdog 兜底 → 入队 → 任务回收 → 决策日志补录 → + analyst 催促 → fire-and-forget 触发 worker/planner。失败不阻断 + (下次节拍继续)。""" + try: + session = get_session() + try: + from agenteval.intelligent_eval.decision_logs import supplement_decision_logs + from agenteval.intelligent_eval.lifecycle import ( + enforce_executing_ceiling, + enforce_planning_gates, + enforce_trigger_failure_gates, + expire_stale_running_sessions, + ) + from agenteval.intelligent_eval.task_queue import ( + requeue_stale_assigned_tasks, + scan_and_enqueue_tasks, + settle_tasks_for_finished_evals, + ) + + r = requeue_stale_assigned_tasks(session) + if r: + _logger.info("卡死恢复:%d 个 assigned 任务重新入队", r) + x = expire_stale_running_sessions(session) + if x: + _logger.info("会话过期:%d 个 running 会话 60 分钟无新轮次,置为 expired", x) + g = enforce_planning_gates(session) + if g: + _logger.info("planning 双闸:%d 个评估超限判失败", g) + c = enforce_executing_ceiling(session) + if c: + _logger.info("executing 兜底:%d 个评估超窗判失败", c) + f = enforce_trigger_failure_gates(session) + if f: + _logger.info("触发失败判死:%d 个评估连续触发失败超限判失败", f) + n = scan_and_enqueue_tasks(session) + if n: + _logger.info("智能评估扫描:入队 %d 个 Worker 任务", n) + # 清理:评估已结束(非 executing)的 pending/assigned 任务回收为 completed + settled = settle_tasks_for_finished_evals(session) + if settled: + _logger.info("任务回收:%d 个已结束评估的任务标记完成", settled) + # 审计兜底:agent 未上报决策日志时,平台按评估状态补录 + added = supplement_decision_logs(session) + if added: + _logger.info("决策日志兜底:补录 %d 条", added) + finally: + session.close() + except Exception as exc: + _logger.warning("智能评估扫描失败(忽略): %s", exc) + # analyst 兜底(ADR-0011):末会话终态 10min 后平台催促 analyst 汇总报告, + # 催促落账后把评估交给 worker 触发(worker 在全终态时会转 analyst 路径) + nudge_ids: list[str] = [] + try: + from agenteval.intelligent_eval.lifecycle import ( + evals_needing_analyst_nudge, + record_analyst_nudge, + ) + + session = get_session() + try: + nudge_ids = evals_needing_analyst_nudge(session) + for eval_id in nudge_ids: + record_analyst_nudge(session, eval_id) + finally: + session.close() + if nudge_ids: + _logger.info("analyst 兜底:催促 %d 个评估汇总报告", len(nudge_ids)) + except Exception as exc: + _logger.warning("analyst 兜底失败(忽略): %s", exc) + # ADR-0011:触发 fire-and-forget,不阻塞 60s 扫描节奏(agent 最长跑 10min) + _fire_and_forget(trigger_worker(nudge_eval_ids=nudge_ids), "Worker") + _fire_and_forget(trigger_planner(), "Planner") + + +class IntelligentEvalScheduler: + """扫描循环的 runtime 单例:持有任务句柄,web 层只做 start/stop。""" + + def __init__(self) -> None: + self._task: Optional[asyncio.Task] = None + + async def _loop(self) -> None: + while True: + scan_once() + await asyncio.sleep(SCAN_INTERVAL_SECONDS) + + def start(self) -> None: + if self._task is None or self._task.done(): + self._task = asyncio.create_task(self._loop()) + + async def stop(self) -> None: + if self._task is not None: + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + self._task = None + + +scheduler_runtime = IntelligentEvalScheduler() diff --git a/backend/agenteval/intelligent_eval/task_queue.py b/backend/agenteval/intelligent_eval/task_queue.py index 4ac560e..b6ffb57 100644 --- a/backend/agenteval/intelligent_eval/task_queue.py +++ b/backend/agenteval/intelligent_eval/task_queue.py @@ -224,7 +224,7 @@ def requeue_stale_assigned_tasks(session: Session) -> int: Returns: 重新入队的任务数(不含被判 failed 的)。 """ - from agenteval.intelligent_eval.decision_logs import create_decision_log + from agenteval.intelligent_eval.decision_logs import append_decision_log threshold = utc_now() - timedelta(minutes=STALE_ASSIGNED_MINUTES) stale = session.exec( @@ -247,7 +247,7 @@ def requeue_stale_assigned_tasks(session: Session) -> int: task.completed_at = utc_now() task.updated_at = utc_now() abandoned += 1 - create_decision_log( + append_decision_log( task.eval_id, "task_abandoned", f"平台兜底:任务卡死重试 {task.attempts} 次仍未完成,置为失败", diff --git a/backend/agenteval/web/app.py b/backend/agenteval/web/app.py index 35f3ff4..be18879 100644 --- a/backend/agenteval/web/app.py +++ b/backend/agenteval/web/app.py @@ -1,17 +1,15 @@ """FastAPI web backend for AgentEvalTool.""" -import asyncio import logging -import time from contextlib import asynccontextmanager from pathlib import Path -from typing import Optional from fastapi import Depends, FastAPI, WebSocket, WebSocketDisconnect from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse, JSONResponse from agenteval.config import get_settings +from agenteval.intelligent_eval.scheduler import scheduler_runtime from agenteval.storage.db import get_session, init_db from agenteval.version import get_build_info, get_version from agenteval.web.deps import require_api_key @@ -32,416 +30,10 @@ from agenteval.web.routers import ( from agenteval.web.websocket import ws_manager -def _supplement_decision_logs(session) -> int: - """Platform audit backfill for decision logs. - - 方案③的决策日志由 OpenClaw agent 上报(LLM 自主,尽力而为)——异常路径 - (如卡死恢复后重试)agent 可能跳过上报,导致决策过程页面为空。这里按评估 - 状态推导决策并补录: - - - EXECUTING:欠账(completed < estimated)补 execute_session,所有会话 - 完成后补 start_analysis。 - - COMPLETED:历史评估/异常路径可能完全没有决策日志,回填 execute_session - (按 plan 时段逐条)+ start_analysis,让旧报告也有决策过程可看。 - - 只补"该类型缺失"的,不重复;且只记录状态,不改变 agent 的实际执行。 - - Returns: - 补录的决策日志条数。 - """ - from sqlmodel import select - - from agenteval.intelligent_eval.decision_logs import create_decision_log - from agenteval.intelligent_eval.models import IntelligentEvalStatus - from agenteval.storage.db import IntelligentEvalDB, IntelligentEvalDecisionLogDB, IntelligentEvalSessionDB - - evals = session.exec( - 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() - completed = sum(1 for s in sessions if s.status == "completed") - types = { - x.decision_type - for x in session.exec( - select(IntelligentEvalDecisionLogDB).where(IntelligentEvalDecisionLogDB.eval_id == ev.id) - ).all() - } - 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 - - -async def _trigger_intelligent_worker(nudge_eval_ids: Optional[list[str]] = None) -> bool: - """Trigger OpenClaw's headless agent to run the intelligent-eval worker skill. - - 方案③(避免外部 IM channel):平台只当"触发闹钟"——有 pending 任务时, - 用 `docker exec openclaw-eval openclaw agent --agent main -m - "agenteval-intelligent-worker"` 唤醒 agent,worker skill 由 OpenClaw agent - 自主执行(决策/建会话/真实对话/上报,逻辑零改动)。`--deliver` 默认 false - 因此不经 cron delivery channel。 - - ADR-0011 孤儿 agent 双管: - - 容器内命令包 `timeout 600`——agent 进程最多跑 10 分钟即被强杀; - - per-eval 触发冷却 10min(worker_trigger 决策日志记录)——agent 还在跑 - 的评估不重复触发,避免 60s 扫描节奏堆叠并发 agent。 - nudge_eval_ids:analyst 兜底催促的评估(所有会话终态、无任务队列条目), - 与候选任务评估合并去重后一同落账触发。 - - Returns: - True 若确实触发了 agent(存在冷却期外的服务对象)。 - """ - from agenteval.intelligent_eval.lifecycle import ( - record_worker_triggers, - worker_trigger_candidates, - ) - - session = get_session() - try: - trigger_ids = sorted(set(worker_trigger_candidates(session)) | set(nudge_eval_ids or [])) - if not trigger_ids: - return False - # 触发前落账(含后续失败也计入冷却),防止 agent 在跑期间重复触发 - record_worker_triggers(session, trigger_ids) - finally: - session.close() - - import subprocess - - _logger = logging.getLogger("agenteval") - # 触发指令必须带"立即执行"语义:`openclaw agent` 无 cron state,若只发 - # skill 名,agent 会按 worker skill 的"跨节拍"设计决策后等下一拍而死锁。 - # 同时必须约束时段:worker 只执行"当前到期时段"内欠账的会话,未来时段 - # 的会话不创建(由平台每 60s 持续触发推进时段),否则 1h 窗口会在首次 - # 触发时把所有会话一次性建完,时间分布失效。 - worker_msg = ( - "执行 agenteval-intelligent-worker skill,立即完成当前任务,不要等待下一节拍:" - "仅执行当前时间对应时段(time_distribution 中当前 offset 所在时段)内欠账的会话" - "——按评估 started_at 与当前时间精确判断当前时段,只创建该时段计划内的会话," - "绝不创建未来时段的会话,未来时段到期后平台会再次触发你;" - "仅当所有时段计划内的会话都已达到终态(completed/failed/expired)时," - "才调用 agenteval-intelligent-analyst skill 完成分析并生成报告;" - "若还有未到期的未来时段,保持等待、本次不做任何操作。" - ) - try: - proc = await asyncio.to_thread( - subprocess.run, - [ - "docker", - "exec", - "openclaw-eval", - # ADR-0011:容器内 timeout 强杀,agent 最长存活 10min(孤儿 agent 双管之一) - "timeout", - "600", - "openclaw", - "agent", - "--agent", - "main", - # 必须用独立 session:`--agent main` 复用 main 持久 session,多次触发 - # 累积上下文缓存(~12 万 token)后 LLM 不再执行 worker skill 的 API - # 步骤,直接幻觉输出(如"评估 pending_approval")而不取任务建会话。 - "--session-id", - f"agenteval-worker-{int(time.time())}", - "-m", - worker_msg, - "--json", - ], - capture_output=True, - text=True, - timeout=660, # 容器内 timeout 600 + 启动/回收余量 - ) - _logger.info("Worker 触发完成 exit=%s", proc.returncode) - if proc.returncode != 0: - _logger.warning("Worker 触发 stderr: %s", proc.stderr[-300:]) - _record_worker_trigger_failure(f"exit={proc.returncode} stderr={proc.stderr[-200:]}") - except Exception as exc: - _logger.warning("Worker 触发失败(忽略): %s", exc) - _record_worker_trigger_failure(str(exc)) - return True - - -def _record_worker_trigger_failure(error: str) -> None: - """ADR-0011:worker 触发失败落账到受影响评估(有待认领任务的评估)。""" - from agenteval.intelligent_eval.lifecycle import ( - eval_ids_with_pending_worker_tasks, - record_trigger_failures, - ) - - session = get_session() - try: - eval_ids = eval_ids_with_pending_worker_tasks(session) - if eval_ids: - record_trigger_failures(session, channel="worker", eval_ids=eval_ids, error=error) - except Exception as exc: - logging.getLogger("agenteval").warning("触发失败落账失败(忽略): %s", exc) - finally: - session.close() - - -async def _trigger_intelligent_planner() -> bool: - """Trigger OpenClaw's headless agent to run the planner skill. - - 方案③只自动化了 executing(worker)→ completed(analyst),**缺少 planning - 阶段的 planner 触发**——新建或被打回的评估会永远停在 planning。这里对 - planning 状态评估触发 `agenteval-intelligent-planner` skill:planner 自会 - 拉取 planning 评估列表、读取四件套、产出粗计划并 PUT /plan 提交(planner - skill 定义见 OpenClaw workspace skills)。处理完评估离开 planning 后不再触发。 - - Returns: - True 若确实触发了 agent(存在 planning 评估)。 - """ - # ADR-0011:触发前先落账 planner_trigger(含后续失败也计入双闸), - # 0 个 planning 评估时不触发。 - from agenteval.intelligent_eval.lifecycle import record_planner_triggers - - session = get_session() - try: - planning_count = record_planner_triggers(session) - finally: - session.close() - if planning_count == 0: - return False - - import subprocess - - _logger = logging.getLogger("agenteval") - # 同 worker:`openclaw agent` 无 cron state,须带"立即完成"语义,否则 planner - # 会"决策后等下一拍"而死锁。 - planner_msg = ( - "执行 agenteval-intelligent-planner skill,立即完成当前任务,不要等待下一节拍:" - "为 planning 状态的智能评估读取输入、产出粗计划并提交平台审批。" - ) - try: - proc = await asyncio.to_thread( - subprocess.run, - [ - "docker", - "exec", - "openclaw-eval", - # ADR-0011:容器内 timeout 强杀,agent 最长存活 10min(孤儿 agent 双管之一) - "timeout", - "600", - "openclaw", - "agent", - "--agent", - "main", - # 同 worker:独立 session 避免 main 持久 session 上下文缓存污染 - "--session-id", - f"agenteval-planner-{int(time.time())}", - "-m", - planner_msg, - "--json", - ], - capture_output=True, - text=True, - timeout=660, # 容器内 timeout 600 + 启动/回收余量 - ) - _logger.info("Planner 触发完成 exit=%s", proc.returncode) - if proc.returncode != 0: - _logger.warning("Planner 触发 stderr: %s", proc.stderr[-300:]) - _record_planner_trigger_failure(f"exit={proc.returncode} stderr={proc.stderr[-200:]}") - except Exception as exc: - _logger.warning("Planner 触发失败(忽略): %s", exc) - _record_planner_trigger_failure(str(exc)) - return True - - -def _record_planner_trigger_failure(error: str) -> None: - """ADR-0011:planner 触发失败落账到所有 planning 评估。""" - from sqlmodel import select - - from agenteval.intelligent_eval.lifecycle import record_trigger_failures - from agenteval.intelligent_eval.models import IntelligentEvalStatus - from agenteval.storage.db import IntelligentEvalDB - - session = get_session() - try: - eval_ids = [ - row.id - for row in session.exec( - select(IntelligentEvalDB).where(IntelligentEvalDB.status == IntelligentEvalStatus.PLANNING.value) - ).all() - ] - if eval_ids: - record_trigger_failures(session, channel="planner", eval_ids=eval_ids, error=error) - except Exception as exc: - logging.getLogger("agenteval").warning("触发失败落账失败(忽略): %s", exc) - finally: - session.close() - - -def _fire_and_forget(coro, name: str) -> None: - """ADR-0011:触发改为 fire-and-forget——agent 最长跑 10min,await 会把 - 60s 扫描节奏拖到 10min+。派生 asyncio task,异常在完成回调中记录 - (触发失败已由触发函数自身落账 trigger_failed)。""" - - def _done(task: asyncio.Task) -> None: - if task.cancelled(): - return - exc = task.exception() - if exc: - logging.getLogger("agenteval").warning("%s 触发任务异常(忽略): %s", name, exc) - - asyncio.create_task(coro).add_done_callback(_done) - - -async def _intelligent_eval_scan_loop() -> None: - """Scan intelligent evals, enqueue tasks, and trigger planner/worker. - - v1.1.0 缺陷修复:`scan_and_enqueue_tasks` 此前没有调度点,OpenClaw Worker - 每分钟唤醒却永远取不到任务。平台启动后每 60s: - ① 有 planning 评估则触发 OpenClaw planner skill 产出粗计划(planning→待审批); - ② 扫描 executing 的评估入队;③ 若有 pending 任务则触发 worker skill 执行会话/ - 分析(方案③,免外部 channel)。失败不阻断(下次循环继续)。 - """ - _logger = logging.getLogger("agenteval") - while True: - try: - session = get_session() - try: - from agenteval.intelligent_eval.lifecycle import ( - enforce_executing_ceiling, - enforce_planning_gates, - enforce_trigger_failure_gates, - expire_stale_running_sessions, - ) - from agenteval.intelligent_eval.task_queue import ( - requeue_stale_assigned_tasks, - scan_and_enqueue_tasks, - settle_tasks_for_finished_evals, - ) - - r = requeue_stale_assigned_tasks(session) - if r: - _logger.info("卡死恢复:%d 个 assigned 任务重新入队", r) - x = expire_stale_running_sessions(session) - if x: - _logger.info("会话过期:%d 个 running 会话 60 分钟无新轮次,置为 expired", x) - g = enforce_planning_gates(session) - if g: - _logger.info("planning 双闸:%d 个评估超限判失败", g) - c = enforce_executing_ceiling(session) - if c: - _logger.info("executing 兜底:%d 个评估超窗判失败", c) - f = enforce_trigger_failure_gates(session) - if f: - _logger.info("触发失败判死:%d 个评估连续触发失败超限判失败", f) - n = scan_and_enqueue_tasks(session) - if n: - _logger.info("智能评估扫描:入队 %d 个 Worker 任务", n) - # 清理:评估已结束(非 executing)的 pending/assigned 任务回收为 completed - settled = settle_tasks_for_finished_evals(session) - if settled: - _logger.info("任务回收:%d 个已结束评估的任务标记完成", settled) - # 审计兜底:agent 未上报决策日志时,平台按评估状态补录 - added = _supplement_decision_logs(session) - if added: - _logger.info("决策日志兜底:补录 %d 条", added) - finally: - session.close() - except Exception as exc: - logging.getLogger("agenteval").warning("智能评估扫描失败(忽略): %s", exc) - # analyst 兜底(ADR-0011):末会话终态 10min 后平台催促 analyst 汇总报告, - # 催促落账后把评估交给 worker 触发(worker 在全终态时会转 analyst 路径) - nudge_ids: list[str] = [] - try: - from agenteval.intelligent_eval.lifecycle import ( - evals_needing_analyst_nudge, - record_analyst_nudge, - ) - - session = get_session() - try: - nudge_ids = evals_needing_analyst_nudge(session) - for eval_id in nudge_ids: - record_analyst_nudge(session, eval_id) - finally: - session.close() - if nudge_ids: - logging.getLogger("agenteval").info("analyst 兜底:催促 %d 个评估汇总报告", len(nudge_ids)) - except Exception as exc: - logging.getLogger("agenteval").warning("analyst 兜底失败(忽略): %s", exc) - # ADR-0011:触发 fire-and-forget,不阻塞 60s 扫描节奏(agent 最长跑 10min) - _fire_and_forget(_trigger_intelligent_worker(nudge_eval_ids=nudge_ids), "Worker") - _fire_and_forget(_trigger_intelligent_planner(), "Planner") - await asyncio.sleep(60) - - @asynccontextmanager async def lifespan(_: FastAPI): init_db() - scan_task = asyncio.create_task(_intelligent_eval_scan_loop()) + scheduler_runtime.start() # 恢复耐久 Campaign/智能作业,并清理无法安全重放的中断运行(尽力而为,不阻断启动)。 try: session = get_session() @@ -474,12 +66,8 @@ async def lifespan(_: FastAPI): except Exception as exc: logging.getLogger("agenteval").warning("启动清理失败(忽略): %s", exc) yield - # 优雅停止后台扫描,再停其他进程内任务。 - scan_task.cancel() - try: - await scan_task - except asyncio.CancelledError: - pass + # 优雅停止智能评估扫描调度,再停其他进程内任务。 + await scheduler_runtime.stop() # 优雅停止所有进程内任务:先停活动调度循环,再停在跑的评测运行, # 最后停活动智能作业(分析 / 周期对比)和 judge 复核。 try: diff --git a/tests/integration/test_decision_log_immutability.py b/tests/integration/test_decision_log_immutability.py index f4597be..00dda82 100644 --- a/tests/integration/test_decision_log_immutability.py +++ b/tests/integration/test_decision_log_immutability.py @@ -132,3 +132,45 @@ def test_decision_log_dedupes_identical_entries( ) ).all() assert len(rows) == 1, f"expected dedup, got {len(rows)} rows" + + +def test_append_decision_log_always_appends_identical_context(db_session: Session): + """平台落账入口不去重:相同 context 也总是追加(每次落账都是新事实)。""" + from agenteval.intelligent_eval.decision_logs import append_decision_log + + eval_db = _make_eval(db_session) + ctx = {"platform_supplemented": True, "attempt": 1} + for _ in range(3): + append_decision_log(eval_db.id, "worker_trigger", "落账", "platform", ctx, db_session) + + rows = db_session.exec( + select(IntelligentEvalDecisionLogDB).where( + IntelligentEvalDecisionLogDB.eval_id == eval_db.id + ) + ).all() + assert len(rows) == 3, f"append 不应去重,得到 {len(rows)} 行" + + +def test_append_decision_log_raises_for_unknown_eval(db_session: Session): + """append 与 create 一样校验评估存在性。""" + from agenteval.intelligent_eval.decision_logs import append_decision_log + + with pytest.raises(LookupError): + append_decision_log("no-such-eval", "worker_trigger", "r", "platform", {}, db_session) + + +def test_count_decisions_counts_by_eval_and_type(db_session: Session): + """count_decisions 按 eval + 类型计数(闸门与 attempt 落账的计数原语)。""" + from agenteval.intelligent_eval.decision_logs import append_decision_log, count_decisions + + eval_db = _make_eval(db_session) + other = _make_eval(db_session) + for _ in range(2): + append_decision_log(eval_db.id, "planner_trigger", "r", "platform", {"n": 1}, db_session) + append_decision_log(eval_db.id, "worker_trigger", "r", "platform", {"n": 2}, db_session) + append_decision_log(other.id, "planner_trigger", "r", "platform", {"n": 3}, db_session) + + assert count_decisions(eval_db.id, "planner_trigger", db_session) == 2 + assert count_decisions(eval_db.id, "worker_trigger", db_session) == 1 + assert count_decisions(other.id, "planner_trigger", db_session) == 1 + assert count_decisions(eval_db.id, "trigger_failed", db_session) == 0 diff --git a/tests/integration/test_intelligent_eval_scan_scheduler.py b/tests/integration/test_intelligent_eval_scan_scheduler.py index a422049..389d7b8 100644 --- a/tests/integration/test_intelligent_eval_scan_scheduler.py +++ b/tests/integration/test_intelligent_eval_scan_scheduler.py @@ -1,9 +1,12 @@ -"""Lifespan scan-loop test (v1.1.0 defect fix). +"""Lifespan scheduler wiring test (v1.1.0 defect fix). `scan_and_enqueue_tasks` previously had no scheduler — the OpenClaw Worker -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. +wakes every minute but could never pull a task. The lifespan now starts the +intelligent-eval scheduler runtime that scans executing evals every 60s. +This test verifies that on application startup the scan is actually invoked. + +编排逻辑本身(watchdog/入队/触发)的测试在 +``tests/unit/test_intelligent_eval_scheduler.py``,直接驱动 ``scan_once``。 """ from unittest.mock import MagicMock @@ -13,6 +16,7 @@ from fastapi.testclient import TestClient def test_lifespan_starts_scan_loop(monkeypatch): """Lifespan startup must invoke the intelligent-eval scan loop once.""" + import agenteval.intelligent_eval.scheduler as scheduler_mod import agenteval.intelligent_eval.task_queue as tq import agenteval.web.app as app_mod @@ -25,7 +29,7 @@ def test_lifespan_starts_scan_loop(monkeypatch): # The scan loop calls get_session() to open a DB session; replace it with a # no-op mock so the test does not touch the real SQLite file. - monkeypatch.setattr(app_mod, "get_session", lambda: MagicMock()) + monkeypatch.setattr(scheduler_mod, "get_session", lambda: MagicMock()) monkeypatch.setattr(tq, "scan_and_enqueue_tasks", fake_scan) with TestClient(app_mod.app) as client: @@ -33,303 +37,3 @@ def test_lifespan_starts_scan_loop(monkeypatch): # The background task runs immediately (before its first 60s sleep). assert calls, "scan_and_enqueue_tasks should have been invoked on startup" - - -def _patch_worker_candidates(monkeypatch, candidates: list[str]): - """ADR-0011:worker 触发的服务对象由冷却过滤后的候选决定(替代原 _has_pending_task)。""" - from unittest.mock import MagicMock - - import agenteval.intelligent_eval.lifecycle as lifecycle_mod - import agenteval.web.app as app_mod - - monkeypatch.setattr(lifecycle_mod, "worker_trigger_candidates", lambda session: candidates) - monkeypatch.setattr(lifecycle_mod, "record_worker_triggers", lambda session, ids: None) - monkeypatch.setattr(app_mod, "get_session", lambda: MagicMock()) - - -def test_trigger_worker_skips_when_no_pending(monkeypatch): - """No pending task → no docker exec invocation.""" - import asyncio - import subprocess - from unittest.mock import MagicMock - - import agenteval.web.app as app_mod - - calls: list = [] - _patch_worker_candidates(monkeypatch, []) - - def fake_run(cmd, **kwargs): - calls.append(cmd) - return MagicMock(returncode=0, stderr="") - - monkeypatch.setattr(subprocess, "run", fake_run) - - assert asyncio.run(app_mod._trigger_intelligent_worker()) is False - assert not calls, "should not invoke docker exec when queue is empty" - - -def test_trigger_worker_calls_docker_exec(monkeypatch): - """Pending task present → invoke `docker exec openclaw-eval openclaw agent`.""" - import asyncio - import subprocess - from unittest.mock import MagicMock - - import agenteval.web.app as app_mod - - calls: list = [] - _patch_worker_candidates(monkeypatch, ["ev-1"]) - - def fake_run(cmd, **kwargs): - calls.append(cmd) - return MagicMock(returncode=0, stderr="") - - monkeypatch.setattr(subprocess, "run", fake_run) - - assert asyncio.run(app_mod._trigger_intelligent_worker()) is True - assert calls, "docker exec should be invoked" - joined = " ".join(calls[0]) - assert "docker" in joined and "openclaw" in joined - assert "agenteval-intelligent-worker" in joined - assert "--agent" in joined and "main" in joined - # 独立 session:避免 main 持久 session 上下文缓存污染导致 worker 幻觉不执行 - assert "--session-id" in joined - assert "agenteval-worker-" in joined - # ADR-0011 孤儿 agent 双管:容器内 timeout 强杀 - assert "timeout" in joined and "600" in joined - - -def test_trigger_worker_msg_has_execute_semantics(monkeypatch): - """The trigger message must instruct immediate execution (no cron state). - - `openclaw agent` has no cron state; a bare skill name makes the worker - skill "decide then wait for the next tick", deadlocking. The message must - say "立即完成当前任务 / 不要等待下一节拍". - """ - import asyncio - import subprocess - from unittest.mock import MagicMock - - import agenteval.web.app as app_mod - - calls: list = [] - _patch_worker_candidates(monkeypatch, ["ev-1"]) - - def fake_run(cmd, **kwargs): - calls.append(cmd) - return MagicMock(returncode=0, stderr="") - - monkeypatch.setattr(subprocess, "run", fake_run) - - asyncio.run(app_mod._trigger_intelligent_worker()) - joined = " ".join(calls[0]) - assert "不要等待下一节拍" in joined - assert "立即完成当前任务" in joined - assert "agenteval-intelligent-worker" in joined - assert "agenteval-intelligent-analyst" in joined - - -def test_trigger_planner_skips_when_no_planning(monkeypatch): - """No planning eval → no docker exec invocation.""" - import asyncio - import subprocess - from unittest.mock import MagicMock - - import agenteval.intelligent_eval.lifecycle as lifecycle_mod - import agenteval.web.app as app_mod - - calls: list = [] - # ADR-0011:planner 触发前置计数落账(record_planner_triggers),0 时不触发 - monkeypatch.setattr(lifecycle_mod, "record_planner_triggers", lambda session: 0) - monkeypatch.setattr(app_mod, "get_session", lambda: MagicMock()) - - def fake_run(cmd, **kwargs): - calls.append(cmd) - return MagicMock(returncode=0, stderr="") - - monkeypatch.setattr(subprocess, "run", fake_run) - - assert asyncio.run(app_mod._trigger_intelligent_planner()) is False - assert not calls, "should not invoke docker exec when no planning eval" - - -def test_trigger_planner_calls_docker_exec(monkeypatch): - """Planning eval present → invoke `docker exec openclaw-eval openclaw agent` planner skill.""" - import asyncio - import subprocess - from unittest.mock import MagicMock - - import agenteval.intelligent_eval.lifecycle as lifecycle_mod - import agenteval.web.app as app_mod - - calls: list = [] - monkeypatch.setattr(lifecycle_mod, "record_planner_triggers", lambda session: 1) - monkeypatch.setattr(app_mod, "get_session", lambda: MagicMock()) - - def fake_run(cmd, **kwargs): - calls.append(cmd) - return MagicMock(returncode=0, stderr="") - - monkeypatch.setattr(subprocess, "run", fake_run) - - assert asyncio.run(app_mod._trigger_intelligent_planner()) is True - assert calls, "docker exec should be invoked" - joined = " ".join(calls[0]) - assert "docker" in joined and "openclaw" in joined - assert "agenteval-intelligent-planner" in joined - # 独立 session(同 worker 防上下文缓存污染) - assert "--session-id" in joined - assert "agenteval-planner-" in joined - assert "立即完成当前任务" in joined - assert "不要等待下一节拍" in joined - - -def test_supplement_execute_session_log(monkeypatch, db_session): - """Executing eval with deficit and no execute_session log → platform backfills.""" - import agenteval.web.app as app_mod - from agenteval.intelligent_eval.models import IntelligentEvalStatus - 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(), - ) - ev.set_plan({"time_distribution": [{"time_slot": "0-1h", "sessions": 1}], "estimated_sessions": 1}) - db_session.add(ev) - db_session.commit() - - assert app_mod._supplement_decision_logs(db_session) == 1 - # second call: already backfilled → 0 - assert app_mod._supplement_decision_logs(db_session) == 0 - - -def test_supplement_start_analysis_log(monkeypatch, db_session): - """Executing eval with all sessions completed and no start_analysis → backfill.""" - 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-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) - 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() - - 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 - - -def test_fire_and_forget_does_not_block_and_logs_errors(monkeypatch): - """ADR-0011:触发派生 asyncio task 后台执行,异常在完成回调记录而不抛出。""" - import asyncio - import logging - - import agenteval.web.app as app_mod - - async def boom(): - await asyncio.sleep(0) - raise RuntimeError("触发爆炸") - - async def main(): - app_mod._fire_and_forget(boom(), "Worker") - await asyncio.sleep(0.01) # 让被派生的任务及其完成回调执行 - return "loop-continues" - - # 直接挂 handler 到 agenteval logger:不依赖 caplog(全量跑时 alembic 迁移 - # 用例的 fileConfig 会 disable_existing_loggers,把 agenteval logger 关掉) - records: list[logging.LogRecord] = [] - handler = logging.Handler() - handler.emit = records.append - logger = logging.getLogger("agenteval") - was_disabled = logger.disabled - logger.disabled = False - logger.addHandler(handler) - try: - assert asyncio.run(main()) == "loop-continues" - finally: - logger.removeHandler(handler) - logger.disabled = was_disabled - assert any("触发爆炸" in r.getMessage() for r in records) diff --git a/tests/unit/test_intelligent_eval_scheduler.py b/tests/unit/test_intelligent_eval_scheduler.py new file mode 100644 index 0000000..d79963c --- /dev/null +++ b/tests/unit/test_intelligent_eval_scheduler.py @@ -0,0 +1,331 @@ +"""Unit tests for the intelligent-eval scheduler runtime. + +从 tests/integration/test_intelligent_eval_scan_scheduler.py 迁移: +scheduler 从 web/app.py 抽出后,编排逻辑直接通过其 interface +(``scan_once`` / ``trigger_worker`` / ``trigger_planner``)驱动, +不再隔着 TestClient 与 60s 循环。lifespan 启停接线测试留在 integration。 +""" + +import asyncio +import logging +import subprocess +from unittest.mock import MagicMock + + +def _patch_worker_candidates(monkeypatch, candidates: list[str]): + """ADR-0011:worker 触发的服务对象由冷却过滤后的候选决定(替代原 _has_pending_task)。""" + import agenteval.intelligent_eval.lifecycle as lifecycle_mod + import agenteval.intelligent_eval.scheduler as scheduler_mod + + monkeypatch.setattr(lifecycle_mod, "worker_trigger_candidates", lambda session: candidates) + monkeypatch.setattr(lifecycle_mod, "record_worker_triggers", lambda session, ids: None) + monkeypatch.setattr(scheduler_mod, "get_session", lambda: MagicMock()) + + +def test_trigger_worker_skips_when_no_pending(monkeypatch): + """No pending task → no docker exec invocation.""" + import agenteval.intelligent_eval.scheduler as scheduler_mod + + calls: list = [] + _patch_worker_candidates(monkeypatch, []) + + def fake_run(cmd, **kwargs): + calls.append(cmd) + return MagicMock(returncode=0, stderr="") + + monkeypatch.setattr(subprocess, "run", fake_run) + + assert asyncio.run(scheduler_mod.trigger_worker()) is False + assert not calls, "should not invoke docker exec when queue is empty" + + +def test_trigger_worker_calls_docker_exec(monkeypatch): + """Pending task present → invoke `docker exec openclaw-eval openclaw agent`.""" + import agenteval.intelligent_eval.scheduler as scheduler_mod + + calls: list = [] + _patch_worker_candidates(monkeypatch, ["ev-1"]) + + def fake_run(cmd, **kwargs): + calls.append(cmd) + return MagicMock(returncode=0, stderr="") + + monkeypatch.setattr(subprocess, "run", fake_run) + + assert asyncio.run(scheduler_mod.trigger_worker()) is True + assert calls, "docker exec should be invoked" + joined = " ".join(calls[0]) + assert "docker" in joined and "openclaw" in joined + assert "agenteval-intelligent-worker" in joined + assert "--agent" in joined and "main" in joined + # 独立 session:避免 main 持久 session 上下文缓存污染导致 worker 幻觉不执行 + assert "--session-id" in joined + assert "agenteval-worker-" in joined + # ADR-0011 孤儿 agent 双管:容器内 timeout 强杀 + assert "timeout" in joined and "600" in joined + + +def test_trigger_worker_msg_has_execute_semantics(monkeypatch): + """The trigger message must instruct immediate execution (no cron state). + + `openclaw agent` has no cron state; a bare skill name makes the worker + skill "decide then wait for the next tick", deadlocking. The message must + say "立即完成当前任务 / 不要等待下一节拍". + """ + import agenteval.intelligent_eval.scheduler as scheduler_mod + + calls: list = [] + _patch_worker_candidates(monkeypatch, ["ev-1"]) + + def fake_run(cmd, **kwargs): + calls.append(cmd) + return MagicMock(returncode=0, stderr="") + + monkeypatch.setattr(subprocess, "run", fake_run) + + asyncio.run(scheduler_mod.trigger_worker()) + joined = " ".join(calls[0]) + assert "不要等待下一节拍" in joined + assert "立即完成当前任务" in joined + assert "agenteval-intelligent-worker" in joined + assert "agenteval-intelligent-analyst" in joined + + +def test_trigger_planner_skips_when_no_planning(monkeypatch): + """No planning eval → no docker exec invocation.""" + import agenteval.intelligent_eval.lifecycle as lifecycle_mod + import agenteval.intelligent_eval.scheduler as scheduler_mod + + calls: list = [] + # ADR-0011:planner 触发前置计数落账(record_planner_triggers),0 时不触发 + monkeypatch.setattr(lifecycle_mod, "record_planner_triggers", lambda session: 0) + monkeypatch.setattr(scheduler_mod, "get_session", lambda: MagicMock()) + + def fake_run(cmd, **kwargs): + calls.append(cmd) + return MagicMock(returncode=0, stderr="") + + monkeypatch.setattr(subprocess, "run", fake_run) + + assert asyncio.run(scheduler_mod.trigger_planner()) is False + assert not calls, "should not invoke docker exec when no planning eval" + + +def test_trigger_planner_calls_docker_exec(monkeypatch): + """Planning eval present → invoke `docker exec openclaw-eval openclaw agent` planner skill.""" + import agenteval.intelligent_eval.lifecycle as lifecycle_mod + import agenteval.intelligent_eval.scheduler as scheduler_mod + + calls: list = [] + monkeypatch.setattr(lifecycle_mod, "record_planner_triggers", lambda session: 1) + monkeypatch.setattr(scheduler_mod, "get_session", lambda: MagicMock()) + + def fake_run(cmd, **kwargs): + calls.append(cmd) + return MagicMock(returncode=0, stderr="") + + monkeypatch.setattr(subprocess, "run", fake_run) + + assert asyncio.run(scheduler_mod.trigger_planner()) is True + assert calls, "docker exec should be invoked" + joined = " ".join(calls[0]) + assert "docker" in joined and "openclaw" in joined + assert "agenteval-intelligent-planner" in joined + # 独立 session(同 worker 防上下文缓存污染) + assert "--session-id" in joined + assert "agenteval-planner-" in joined + assert "立即完成当前任务" in joined + assert "不要等待下一节拍" in joined + + +def test_scan_once_orchestrates_full_tick(monkeypatch): + """scan_once 一个节拍按原顺序执行 watchdog → 入队 → 回收 → 补录 → 催促 → 触发。""" + import agenteval.intelligent_eval.decision_logs as dl_mod + import agenteval.intelligent_eval.lifecycle as lifecycle_mod + import agenteval.intelligent_eval.scheduler as scheduler_mod + import agenteval.intelligent_eval.task_queue as tq_mod + + calls: list[str] = [] + monkeypatch.setattr(scheduler_mod, "get_session", lambda: MagicMock()) + + for name in ("requeue_stale_assigned_tasks", "scan_and_enqueue_tasks", "settle_tasks_for_finished_evals"): + monkeypatch.setattr(tq_mod, name, lambda session, _n=name: calls.append(_n) or 0) + for name in ( + "expire_stale_running_sessions", + "enforce_planning_gates", + "enforce_executing_ceiling", + "enforce_trigger_failure_gates", + ): + monkeypatch.setattr(lifecycle_mod, name, lambda session, _n=name: calls.append(_n) or 0) + monkeypatch.setattr(dl_mod, "supplement_decision_logs", lambda session: calls.append("supplement") or 0) + monkeypatch.setattr(lifecycle_mod, "evals_needing_analyst_nudge", lambda session: calls.append("nudge") or []) + + fired: list[str] = [] + + def fake_fire_and_forget(coro, name): + fired.append(name) + coro.close() # 不真正执行触发(避免 docker exec) + + monkeypatch.setattr(scheduler_mod, "_fire_and_forget", fake_fire_and_forget) + + scheduler_mod.scan_once() + + assert calls == [ + "requeue_stale_assigned_tasks", + "expire_stale_running_sessions", + "enforce_planning_gates", + "enforce_executing_ceiling", + "enforce_trigger_failure_gates", + "scan_and_enqueue_tasks", + "settle_tasks_for_finished_evals", + "supplement", + "nudge", + ] + assert fired == ["Worker", "Planner"] + + +def test_supplement_execute_session_log(monkeypatch, db_session): + """Executing eval with deficit and no execute_session log → platform backfills.""" + from agenteval.intelligent_eval.decision_logs import supplement_decision_logs + from agenteval.intelligent_eval.models import IntelligentEvalStatus + from agenteval.storage.db import IntelligentEvalDB, utc_now + + ev = IntelligentEvalDB( + name="supp-eval", + 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) + db_session.commit() + + assert supplement_decision_logs(db_session) == 1 + # second call: already backfilled → 0 + assert supplement_decision_logs(db_session) == 0 + + +def test_supplement_start_analysis_log(monkeypatch, db_session): + """Executing eval with all sessions completed and no start_analysis → backfill.""" + from agenteval.intelligent_eval.decision_logs import supplement_decision_logs + from agenteval.intelligent_eval.models import IntelligentEvalStatus + 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(), + ) + 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() + + assert 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.""" + from agenteval.intelligent_eval.decision_logs import supplement_decision_logs + 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 supplement_decision_logs(db_session) == 3 + # 幂等:二次调用不再补 + assert 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.""" + from agenteval.intelligent_eval.decision_logs import create_decision_log, supplement_decision_logs + 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() + + # 已有 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 supplement_decision_logs(db_session) == 0 + + +def test_fire_and_forget_does_not_block_and_logs_errors(monkeypatch): + """ADR-0011:触发派生 asyncio task 后台执行,异常在完成回调记录而不抛出。""" + import agenteval.intelligent_eval.scheduler as scheduler_mod + + async def boom(): + await asyncio.sleep(0) + raise RuntimeError("触发爆炸") + + async def main(): + scheduler_mod._fire_and_forget(boom(), "Worker") + await asyncio.sleep(0.01) # 让被派生的任务及其完成回调执行 + return "loop-continues" + + # 直接挂 handler 到 agenteval logger:不依赖 caplog(全量跑时 alembic 迁移 + # 用例的 fileConfig 会 disable_existing_loggers,把 agenteval logger 关掉) + records: list[logging.LogRecord] = [] + handler = logging.Handler() + handler.emit = records.append + logger = logging.getLogger("agenteval") + was_disabled = logger.disabled + logger.disabled = False + logger.addHandler(handler) + try: + assert asyncio.run(main()) == "loop-continues" + finally: + logger.removeHandler(handler) + logger.disabled = was_disabled + assert any("触发爆炸" in r.getMessage() for r in records)