da7dd434dd
- expire_stale_running_sessions: 使用单次 JOIN 查询替代 N+1 查询 将每个会话单独查询最后消息时间改为一次性获取所有 running 会话及其最后消息时间 - submit_report/evals_needing_analyst_nudge: 消除 session/sessions 参数名混淆 将局部变量 sessions 重命名为 eval_sessions，避免与数据库会话参数 session 混淆 这些改进提升了查询性能并增强了代码可读性。
867 lines
34 KiB
Python
867 lines
34 KiB
Python
"""Intelligent eval lifecycle (状态机 + 领域操作).
|
||
|
||
状态机：
|
||
draft → planning → pending_approval → executing → completed
|
||
→ cancelled
|
||
→ failed
|
||
pending_approval 可打回 → planning（附反馈）
|
||
executing 可取消 → cancelled
|
||
completed / cancelled / failed 可删除 → deleted（逻辑删除）
|
||
|
||
非法转换抛 IntelligentEvalTransitionError，路由层映射为 409。
|
||
"""
|
||
|
||
import logging
|
||
from datetime import timedelta
|
||
from typing import Any, Optional
|
||
|
||
from sqlmodel import Session, func, select
|
||
|
||
from agenteval.channels.base import ExchangeStatus, SendResult
|
||
from agenteval.channels.factory import ChannelFactory
|
||
from agenteval.config import get_settings
|
||
from agenteval.intelligent_eval.decision_logs import append_decision_log, count_decisions
|
||
from agenteval.intelligent_eval.domain import parse_time_slot
|
||
from agenteval.intelligent_eval.models import (
|
||
IntelligentEval,
|
||
IntelligentEvalMessage,
|
||
IntelligentEvalSession,
|
||
IntelligentEvalSessionStatus,
|
||
IntelligentEvalStatus,
|
||
)
|
||
from agenteval.intelligent_eval.repository import (
|
||
CompareAndSetStatus,
|
||
IntelligentEvalMessageRepository,
|
||
IntelligentEvalRepository,
|
||
IntelligentEvalSessionRepository,
|
||
)
|
||
from agenteval.storage.db import (
|
||
IntelligentEvalDB,
|
||
IntelligentEvalDecisionLogDB,
|
||
IntelligentEvalMessageDB,
|
||
IntelligentEvalSessionDB,
|
||
IntelligentEvalTaskQueueDB,
|
||
as_utc,
|
||
utc_now,
|
||
)
|
||
from agenteval.storage.repository import TargetRepository
|
||
|
||
# 合法转换表：当前状态 → 允许的目标状态集合
|
||
_TRANSITIONS: dict[IntelligentEvalStatus, set[IntelligentEvalStatus]] = {
|
||
IntelligentEvalStatus.DRAFT: {IntelligentEvalStatus.PLANNING},
|
||
# PLANNING → FAILED 仅由 planning 双闸 watchdog 触发（ADR-0011），非用户操作
|
||
IntelligentEvalStatus.PLANNING: {IntelligentEvalStatus.PENDING_APPROVAL, IntelligentEvalStatus.FAILED},
|
||
IntelligentEvalStatus.PENDING_APPROVAL: {
|
||
IntelligentEvalStatus.EXECUTING,
|
||
IntelligentEvalStatus.PLANNING,
|
||
IntelligentEvalStatus.CANCELLED,
|
||
},
|
||
IntelligentEvalStatus.EXECUTING: {
|
||
IntelligentEvalStatus.COMPLETED,
|
||
IntelligentEvalStatus.CANCELLED,
|
||
IntelligentEvalStatus.FAILED,
|
||
},
|
||
IntelligentEvalStatus.COMPLETED: {IntelligentEvalStatus.DELETED},
|
||
IntelligentEvalStatus.CANCELLED: {IntelligentEvalStatus.DELETED},
|
||
IntelligentEvalStatus.FAILED: {IntelligentEvalStatus.DELETED},
|
||
IntelligentEvalStatus.DELETED: set(),
|
||
}
|
||
|
||
|
||
class IntelligentEvalNotFoundError(Exception):
|
||
pass
|
||
|
||
|
||
class IntelligentEvalTransitionError(Exception):
|
||
def __init__(self, reason: str):
|
||
self.reason = reason
|
||
super().__init__(reason)
|
||
|
||
|
||
class IntelligentEvalChannelError(Exception):
|
||
pass
|
||
|
||
|
||
def _get_or_raise(repo: IntelligentEvalRepository, eval_id: str) -> IntelligentEval:
|
||
ev = repo.get(eval_id)
|
||
if ev is None:
|
||
raise IntelligentEvalNotFoundError(f"intelligent eval {eval_id} not found")
|
||
return ev
|
||
|
||
|
||
def _resolve_write(
|
||
eval_id: str,
|
||
result,
|
||
*,
|
||
expected: IntelligentEvalStatus,
|
||
target: IntelligentEvalStatus,
|
||
) -> IntelligentEval:
|
||
if result.status is CompareAndSetStatus.NOT_FOUND:
|
||
raise IntelligentEvalNotFoundError(f"intelligent eval {eval_id} not found")
|
||
if result.status is CompareAndSetStatus.CONFLICT:
|
||
raise IntelligentEvalTransitionError(
|
||
f"cannot transition from {expected.value} to {target.value}; state changed concurrently"
|
||
)
|
||
if result.evaluation is None:
|
||
raise RuntimeError(f"lifecycle write returned no evaluation: {eval_id}")
|
||
return result.evaluation
|
||
|
||
|
||
def _transition(repo: IntelligentEvalRepository, ev: IntelligentEval, target: IntelligentEvalStatus) -> IntelligentEval:
|
||
allowed = _TRANSITIONS.get(ev.status, set())
|
||
if target not in allowed:
|
||
raise IntelligentEvalTransitionError(f"cannot transition from {ev.status.value} to {target.value}")
|
||
result = repo._compare_and_set_status(
|
||
ev.id,
|
||
expected_status=ev.status,
|
||
new_status=target,
|
||
)
|
||
return _resolve_write(ev.id, result, expected=ev.status, target=target)
|
||
|
||
|
||
def create_eval(
|
||
session: Session,
|
||
*,
|
||
name: str,
|
||
target_id: str,
|
||
goal: str,
|
||
seeds: dict[str, Any],
|
||
intent: str,
|
||
role_description: str,
|
||
time_window_hours: int = 24,
|
||
) -> IntelligentEval:
|
||
"""创建智能评估并直接进入 planning 状态（draft → planning 一步完成）。"""
|
||
from agenteval.intelligent_eval.config_snapshot import save_snapshot
|
||
if TargetRepository(session).get(target_id) is None:
|
||
raise IntelligentEvalNotFoundError(f"target {target_id} not found")
|
||
|
||
repo = IntelligentEvalRepository(session)
|
||
ev = repo._create(
|
||
IntelligentEval(
|
||
name=name,
|
||
target_id=target_id,
|
||
status=IntelligentEvalStatus.PLANNING,
|
||
goal=goal,
|
||
seeds=seeds,
|
||
intent=intent,
|
||
role_description=role_description,
|
||
time_window_hours=time_window_hours,
|
||
created_at=utc_now(),
|
||
updated_at=utc_now(),
|
||
)
|
||
)
|
||
|
||
# Save config snapshot
|
||
eval_db = session.get(IntelligentEvalDB, ev.id)
|
||
if eval_db:
|
||
save_snapshot(eval_db, "created", "user", session)
|
||
|
||
return ev
|
||
|
||
|
||
def submit_plan(session: Session, eval_id: str, plan: dict[str, Any]) -> IntelligentEval:
|
||
"""OpenClaw 提交粗计划：planning → pending_approval。"""
|
||
from agenteval.intelligent_eval.config_snapshot import save_snapshot
|
||
repo = IntelligentEvalRepository(session)
|
||
result = repo._submit_plan_if_planning(eval_id, plan)
|
||
ev = _resolve_write(
|
||
eval_id,
|
||
result,
|
||
expected=IntelligentEvalStatus.PLANNING,
|
||
target=IntelligentEvalStatus.PENDING_APPROVAL,
|
||
)
|
||
|
||
# Save config snapshot
|
||
eval_db = session.get(IntelligentEvalDB, eval_id)
|
||
if eval_db:
|
||
save_snapshot(eval_db, "plan_submitted", "openclaw", session)
|
||
|
||
return ev
|
||
|
||
|
||
def approve(session: Session, eval_id: str) -> IntelligentEval:
|
||
"""用户批准：pending_approval → executing。"""
|
||
repo = IntelligentEvalRepository(session)
|
||
ev = _get_or_raise(repo, eval_id)
|
||
return _transition(repo, ev, IntelligentEvalStatus.EXECUTING)
|
||
|
||
|
||
def reject(session: Session, eval_id: str, feedback: str) -> IntelligentEval:
|
||
"""用户打回：pending_approval → planning（附反馈）。"""
|
||
repo = IntelligentEvalRepository(session)
|
||
result = repo._reject_plan_if_pending(eval_id, feedback)
|
||
return _resolve_write(
|
||
eval_id,
|
||
result,
|
||
expected=IntelligentEvalStatus.PENDING_APPROVAL,
|
||
target=IntelligentEvalStatus.PLANNING,
|
||
)
|
||
|
||
|
||
def cancel(session: Session, eval_id: str) -> IntelligentEval:
|
||
"""用户取消：pending_approval / executing → cancelled。"""
|
||
repo = IntelligentEvalRepository(session)
|
||
ev = _get_or_raise(repo, eval_id)
|
||
return _transition(repo, ev, IntelligentEvalStatus.CANCELLED)
|
||
|
||
|
||
def delete_eval(session: Session, eval_id: str) -> IntelligentEval:
|
||
"""逻辑删除：completed / cancelled / failed → deleted。幂等：已删除直接返回。"""
|
||
repo = IntelligentEvalRepository(session)
|
||
ev = repo.get_including_deleted(eval_id)
|
||
if ev is None:
|
||
raise IntelligentEvalNotFoundError(f"intelligent eval {eval_id} not found")
|
||
if ev.status is IntelligentEvalStatus.DELETED:
|
||
return ev
|
||
return _transition(repo, ev, IntelligentEvalStatus.DELETED)
|
||
|
||
|
||
def submit_report(session: Session, eval_id: str, report: dict[str, Any]) -> IntelligentEval:
|
||
"""OpenClaw 提交结构化报告：executing → completed。
|
||
|
||
报告结构校验在路由层（pydantic），此处只负责落库与状态迁移。提交前必须
|
||
满足：存在会话时须全部到达终态（completed/failed/expired）——否则拒绝，
|
||
杜绝"评估 completed 但 completed_sessions=0（进度 0%）"的不一致。
|
||
ADR-0011：卡死的 running 会话由 expire_stale_running_sessions 置 expired
|
||
（不完整证据），不再阻塞报告提交。
|
||
"""
|
||
repo = IntelligentEvalRepository(session)
|
||
eval_sessions = IntelligentEvalSessionRepository(session).list_by_eval(eval_id)
|
||
if any(s.status == IntelligentEvalSessionStatus.RUNNING for s in eval_sessions):
|
||
raise IntelligentEvalTransitionError("存在进行中的会话，不能提交报告")
|
||
# ADR-0011：submit 边界把 scores 归一到 {overall, dimensions} 单一规范结构
|
||
if report.get("scores"):
|
||
from agenteval.intelligent_eval.report import normalize_scores
|
||
|
||
report = {**report, "scores": normalize_scores(report["scores"])}
|
||
result = repo._submit_report_if_executing(eval_id, report)
|
||
return _resolve_write(
|
||
eval_id,
|
||
result,
|
||
expected=IntelligentEvalStatus.EXECUTING,
|
||
target=IntelligentEvalStatus.COMPLETED,
|
||
)
|
||
|
||
|
||
def get_eval(session: Session, eval_id: str) -> IntelligentEval:
|
||
repo = IntelligentEvalRepository(session)
|
||
return _get_or_raise(repo, eval_id)
|
||
|
||
|
||
def list_evals(session: Session) -> list[IntelligentEval]:
|
||
return IntelligentEvalRepository(session).list_all()
|
||
|
||
|
||
def list_evals_page(
|
||
session: Session, offset: int, limit: int, status: Optional[str] = None
|
||
) -> tuple[list[IntelligentEval], int, dict[str, int]]:
|
||
"""Return one page of evals (newest first), total count, and per-status counts."""
|
||
repo = IntelligentEvalRepository(session)
|
||
return repo.list_page(offset, limit, status), repo.count(), repo.count_by_status()
|
||
|
||
|
||
def eval_status_counts(session: Session) -> dict[str, int]:
|
||
"""Count evaluations per status (for the list page stat bar)."""
|
||
return IntelligentEvalRepository(session).count_by_status()
|
||
|
||
|
||
def _get_session_or_raise(repo: IntelligentEvalSessionRepository, session_id: str) -> IntelligentEvalSession:
|
||
obj = repo.get(session_id)
|
||
if obj is None:
|
||
raise IntelligentEvalNotFoundError(f"intelligent eval session {session_id} not found")
|
||
return obj
|
||
|
||
|
||
def open_session(
|
||
session: Session,
|
||
*,
|
||
eval_id: str,
|
||
persona: dict[str, Any],
|
||
goal: str,
|
||
dimension: str | None = None,
|
||
) -> IntelligentEvalSession:
|
||
"""创建虚拟用户会话：仅 executing 状态且未超粗计划预算的评估可创建。"""
|
||
eval_repo = IntelligentEvalRepository(session)
|
||
ev = _get_or_raise(eval_repo, eval_id)
|
||
if ev.status != IntelligentEvalStatus.EXECUTING:
|
||
raise IntelligentEvalTransitionError(f"评估不在执行中（当前 {ev.status.value}），无法创建会话")
|
||
# 预算硬闸门（ADR-0011）：会话数不得超过粗计划虚拟用户数
|
||
budget = (ev.plan or {}).get("estimated_sessions", 0)
|
||
if isinstance(budget, int) and budget > 0:
|
||
opened = session.exec(
|
||
select(func.count())
|
||
.select_from(IntelligentEvalSessionDB)
|
||
.where(IntelligentEvalSessionDB.eval_id == ev.id)
|
||
).one()
|
||
if opened >= budget:
|
||
raise IntelligentEvalTransitionError(f"会话数已达粗计划上限（{budget} 个虚拟用户），超出预算")
|
||
repo = IntelligentEvalSessionRepository(session)
|
||
status, created = repo._create_if_executing(
|
||
IntelligentEvalSession(
|
||
eval_id=ev.id,
|
||
target_id=ev.target_id,
|
||
persona=persona,
|
||
goal=goal,
|
||
dimension=dimension,
|
||
)
|
||
)
|
||
if status is CompareAndSetStatus.CONFLICT:
|
||
raise IntelligentEvalTransitionError("评估不在执行中，会话创建被拒绝")
|
||
if status is CompareAndSetStatus.NOT_FOUND or created is None:
|
||
raise IntelligentEvalNotFoundError(f"intelligent eval {eval_id} not found")
|
||
return created
|
||
|
||
|
||
async def conduct_turn(session: Session, *, eval_id: str, session_id: str, content: str) -> dict[str, Any]:
|
||
"""一轮完整问答：状态检查 → 通道往返 → 双条消息落库 → 轮次自增。"""
|
||
repo = IntelligentEvalSessionRepository(session)
|
||
obj = _get_owned_session_or_raise(repo, eval_id, session_id)
|
||
if obj.status != IntelligentEvalSessionStatus.RUNNING:
|
||
raise IntelligentEvalTransitionError("会话不在进行中，拒收消息")
|
||
|
||
target = TargetRepository(session).get(obj.target_id)
|
||
if not target:
|
||
raise IntelligentEvalNotFoundError("session target not found")
|
||
|
||
message_repo = IntelligentEvalMessageRepository(session)
|
||
sent_at = utc_now()
|
||
channel = ChannelFactory.create(target)
|
||
|
||
async def record_sent(_send_result: SendResult) -> None:
|
||
message = IntelligentEvalMessage(
|
||
session_id=obj.id,
|
||
role="user",
|
||
content=content,
|
||
created_at=sent_at,
|
||
)
|
||
status = message_repo._create_user_and_increment(message)
|
||
if status is CompareAndSetStatus.NOT_FOUND:
|
||
raise IntelligentEvalNotFoundError(f"intelligent eval session {obj.id} not found")
|
||
if status is CompareAndSetStatus.CONFLICT:
|
||
raise IntelligentEvalTransitionError("会话不在进行中，拒收消息")
|
||
|
||
outcome = await channel.exchange(
|
||
content,
|
||
timeout=get_settings().poll_reply_timeout,
|
||
on_sent=record_sent,
|
||
)
|
||
if outcome.status is ExchangeStatus.SEND_FAILED:
|
||
raise IntelligentEvalChannelError(f"评测对象通道发送失败: {outcome.reason}")
|
||
if outcome.status is ExchangeStatus.POLL_FAILED:
|
||
raise IntelligentEvalChannelError(f"等待评测对象回复失败: {outcome.reason}")
|
||
if outcome.status is ExchangeStatus.REPLY_TIMEOUT:
|
||
raise IntelligentEvalChannelError("等待评测对象回复超时")
|
||
|
||
received_at = utc_now()
|
||
latency_ms = (
|
||
outcome.latency_ms if outcome.latency_ms is not None else int((received_at - sent_at).total_seconds() * 1000)
|
||
)
|
||
reply_text = outcome.reply_text or ""
|
||
message_repo._create(
|
||
IntelligentEvalMessage(
|
||
session_id=obj.id,
|
||
role="assistant",
|
||
content=reply_text,
|
||
latency_ms=latency_ms,
|
||
created_at=received_at,
|
||
)
|
||
)
|
||
return {"reply": reply_text, "latency_ms": latency_ms, "turn_count": obj.turn_count + 1}
|
||
|
||
|
||
def close_session(
|
||
session: Session,
|
||
*,
|
||
eval_id: str,
|
||
session_id: str,
|
||
verdict: dict[str, Any],
|
||
) -> IntelligentEvalSession:
|
||
"""关闭会话并记录结论（verdict）；仅 running 会话可关闭。"""
|
||
repo = IntelligentEvalSessionRepository(session)
|
||
obj = _get_owned_session_or_raise(repo, eval_id, session_id)
|
||
status, closed = repo._close_if_running(obj.id, verdict)
|
||
if status is CompareAndSetStatus.NOT_FOUND:
|
||
raise IntelligentEvalNotFoundError(f"intelligent eval session {session_id} not found")
|
||
if status is CompareAndSetStatus.CONFLICT:
|
||
raise IntelligentEvalTransitionError("会话不在进行中，无法关闭")
|
||
if closed is None:
|
||
raise RuntimeError(f"session close returned no session: {session_id}")
|
||
return closed
|
||
|
||
|
||
def _get_owned_session_or_raise(
|
||
repo: IntelligentEvalSessionRepository,
|
||
eval_id: str,
|
||
session_id: str,
|
||
) -> IntelligentEvalSession:
|
||
obj = _get_session_or_raise(repo, session_id)
|
||
if obj.eval_id != eval_id:
|
||
raise IntelligentEvalNotFoundError(f"intelligent eval session {session_id} not found")
|
||
return obj
|
||
|
||
|
||
def list_sessions(session: Session, eval_id: str) -> list[IntelligentEvalSession]:
|
||
eval_repo = IntelligentEvalRepository(session)
|
||
_get_or_raise(eval_repo, eval_id)
|
||
return IntelligentEvalSessionRepository(session).list_by_eval(eval_id)
|
||
|
||
|
||
def list_messages(session: Session, *, eval_id: str, session_id: str) -> list[IntelligentEvalMessage]:
|
||
_get_owned_session_or_raise(IntelligentEvalSessionRepository(session), eval_id, session_id)
|
||
return IntelligentEvalMessageRepository(session).list_by_session(session_id)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 会话过期 watchdog（ADR-0011）：running 会话 60 分钟无新轮次 → expired
|
||
# ---------------------------------------------------------------------------
|
||
|
||
SESSION_IDLE_EXPIRE_MINUTES = 60
|
||
|
||
|
||
def expire_stale_running_sessions(session: Session) -> int:
|
||
"""把「60 分钟无新轮次」的 running 会话置为 expired。
|
||
|
||
worker/agent 中断会让会话永久卡在 running：conduct_turn 拒收非 running
|
||
会话，而 submit_report 又要求全部终态——不过期的话评估永远无法收尾。
|
||
活动判定取会话内最后一条消息的 created_at，无消息则退回会话 created_at。
|
||
expired 是终态（closed_at 一并写入），报告中按不完整证据对待。
|
||
|
||
Returns:
|
||
过期的会话数。
|
||
"""
|
||
now = utc_now()
|
||
# SQLite 读出为 naive datetime，阈值须同为 naive 才能在 Python 侧比较
|
||
threshold = now.replace(tzinfo=None) - timedelta(minutes=SESSION_IDLE_EXPIRE_MINUTES)
|
||
|
||
# 单次查询获取所有 running 会话及其最后消息时间（避免 N+1 查询）
|
||
stmt = (
|
||
select(
|
||
IntelligentEvalSessionDB,
|
||
func.max(IntelligentEvalMessageDB.created_at).label("last_message_at"),
|
||
)
|
||
.outerjoin(
|
||
IntelligentEvalMessageDB,
|
||
IntelligentEvalMessageDB.session_id == IntelligentEvalSessionDB.id,
|
||
)
|
||
.where(IntelligentEvalSessionDB.status == "running")
|
||
.group_by(IntelligentEvalSessionDB.id)
|
||
)
|
||
rows = session.exec(stmt).all()
|
||
|
||
expired = 0
|
||
for session_row, last_message_at in rows:
|
||
last_activity = last_message_at or session_row.created_at
|
||
if last_activity is None or last_activity >= threshold:
|
||
continue
|
||
session_row.status = IntelligentEvalSessionStatus.EXPIRED.value
|
||
session_row.closed_at = now
|
||
expired += 1
|
||
append_decision_log(
|
||
session_row.eval_id,
|
||
"session_expired",
|
||
f"平台兜底：会话 {SESSION_IDLE_EXPIRE_MINUTES} 分钟无新轮次，置为过期（不完整证据）",
|
||
"platform",
|
||
{"platform_supplemented": True, "session_id": session_row.id, "turn_count": session_row.turn_count},
|
||
session,
|
||
)
|
||
|
||
if expired:
|
||
session.commit()
|
||
return expired
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# planning 双闸 watchdog（ADR-0011）：触发 ≥5 次或 ≥30 分钟无提交 → failed
|
||
# ---------------------------------------------------------------------------
|
||
|
||
PLANNING_MAX_ATTEMPTS = 5
|
||
PLANNING_TIMEOUT_MINUTES = 30
|
||
|
||
# ADR-0011 孤儿 agent 双管之一：同一评估的触发冷却（agent 进程最长跑
|
||
# `timeout 600`，60s 扫描节奏下不冷却会堆叠并发 agent）。
|
||
TRIGGER_COOLDOWN_MINUTES = 10
|
||
|
||
|
||
def _last_decision_at(session: Session, eval_id: str, decision_type: str):
|
||
"""该评估某类决策日志的最近时间（无则 None）。"""
|
||
return session.exec(
|
||
select(func.max(IntelligentEvalDecisionLogDB.created_at)).where(
|
||
IntelligentEvalDecisionLogDB.eval_id == eval_id,
|
||
IntelligentEvalDecisionLogDB.decision_type == decision_type,
|
||
)
|
||
).one()
|
||
|
||
|
||
def _in_trigger_cooldown(session: Session, eval_id: str, decision_type: str, now) -> bool:
|
||
last = _last_decision_at(session, eval_id, decision_type)
|
||
return last is not None and (now - last).total_seconds() < TRIGGER_COOLDOWN_MINUTES * 60
|
||
|
||
|
||
def record_planner_triggers(session: Session) -> int:
|
||
"""为「冷却期外」的 planning 评估补一条 planner_trigger 决策日志。
|
||
|
||
触发次数即日志条数（含触发失败——attempt 在触发动作发生前落账），供
|
||
enforce_planning_gates 计闸，也让决策过程页可见平台尝试过多少次。
|
||
ADR-0011：同一评估 10min 冷却内的扫描节拍不重复触发（孤儿 agent 双管），
|
||
冷却中的评估不计次数。
|
||
|
||
Returns:
|
||
本次实际触发覆盖的评估数（0 表示无需触发 planner）。
|
||
"""
|
||
now = utc_now().replace(tzinfo=None)
|
||
planning = session.exec(
|
||
select(IntelligentEvalDB).where(IntelligentEvalDB.status == IntelligentEvalStatus.PLANNING.value)
|
||
).all()
|
||
served = 0
|
||
for row in planning:
|
||
if _in_trigger_cooldown(session, row.id, "planner_trigger", now):
|
||
continue
|
||
attempt = count_decisions(row.id, "planner_trigger", session) + 1
|
||
append_decision_log(
|
||
row.id,
|
||
"planner_trigger",
|
||
f"平台第 {attempt} 次触发 planner 生成粗计划",
|
||
"platform",
|
||
{"platform_supplemented": True, "attempt": attempt},
|
||
session,
|
||
)
|
||
served += 1
|
||
return served
|
||
|
||
|
||
def worker_trigger_candidates(session: Session) -> list[str]:
|
||
"""有待认领 worker 任务且冷却期外的评估 id（本次 worker 触发的服务对象）。"""
|
||
now = utc_now().replace(tzinfo=None)
|
||
return [
|
||
eval_id
|
||
for eval_id in eval_ids_with_pending_worker_tasks(session)
|
||
if not _in_trigger_cooldown(session, eval_id, "worker_trigger", now)
|
||
]
|
||
|
||
|
||
def record_worker_triggers(session: Session, eval_ids: list[str]) -> None:
|
||
"""为本次 worker 触发覆盖的评估落账 worker_trigger（供冷却与可见性）。"""
|
||
for eval_id in eval_ids:
|
||
attempt = count_decisions(eval_id, "worker_trigger", session) + 1
|
||
append_decision_log(
|
||
eval_id,
|
||
"worker_trigger",
|
||
f"平台第 {attempt} 次触发 worker 执行会话/分析",
|
||
"platform",
|
||
{"platform_supplemented": True, "attempt": attempt},
|
||
session,
|
||
)
|
||
|
||
|
||
def enforce_planning_gates(session: Session) -> int:
|
||
"""planning 双闸：触发次数或时长超限仍未提交粗计划 → 评估置 failed。
|
||
|
||
闸一：planner_trigger 决策日志 ≥ PLANNING_MAX_ATTEMPTS（planner 反复
|
||
触发却交不出计划，视为确定性失败，不再烧触发）。
|
||
闸二：进入 planning 超 PLANNING_TIMEOUT_MINUTES（updated_at 在进入
|
||
planning/被打回时刷新，即"最近一次进入 planning 的时刻"）。
|
||
失败原因写入 plan_feedback（前端可见）并补 planning_gate_failed 决策日志。
|
||
|
||
Returns:
|
||
被判失败的评估数。
|
||
"""
|
||
now = utc_now().replace(tzinfo=None)
|
||
planning = session.exec(
|
||
select(IntelligentEvalDB).where(IntelligentEvalDB.status == IntelligentEvalStatus.PLANNING.value)
|
||
).all()
|
||
|
||
failed = 0
|
||
for row in planning:
|
||
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:
|
||
reason = f"planner 已触发 {attempts} 次仍未提交粗计划，按 ADR-0011 双闸判失败"
|
||
elif age_minutes >= PLANNING_TIMEOUT_MINUTES:
|
||
reason = f"planning 超过 {PLANNING_TIMEOUT_MINUTES} 分钟未提交粗计划，按 ADR-0011 双闸判失败"
|
||
else:
|
||
continue
|
||
if fail_eval(
|
||
session,
|
||
row.id,
|
||
reason,
|
||
"planning_gate_failed",
|
||
{"attempts": attempts, "age_minutes": int(age_minutes)},
|
||
):
|
||
failed += 1
|
||
|
||
return failed
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# executing 兜底 watchdog（ADR-0011）：超窗判败 + analyst 催促
|
||
# ---------------------------------------------------------------------------
|
||
|
||
EXECUTING_OVERRUN_GRACE_HOURS = 2
|
||
ANALYST_NUDGE_DELAY_MINUTES = 10
|
||
ANALYST_NUDGE_MAX = 3
|
||
|
||
_TERMINAL_SESSION_STATUSES = ("completed", "failed", "expired")
|
||
|
||
|
||
def enforce_executing_ceiling(session: Session) -> int:
|
||
"""executing 总时长超过 time_window_hours + 宽限 → 评估置 failed。
|
||
|
||
所有自愈手段（任务重试、会话过期、analyst 催促）都失败后的最终收敛：
|
||
评估绝不无声卡死在 executing。原因写 plan_feedback + executing_timeout
|
||
决策日志，前端执行过程页可见。
|
||
|
||
Returns:
|
||
被判失败的评估数。
|
||
"""
|
||
now = utc_now().replace(tzinfo=None)
|
||
executing = session.exec(
|
||
select(IntelligentEvalDB).where(IntelligentEvalDB.status == IntelligentEvalStatus.EXECUTING.value)
|
||
).all()
|
||
|
||
failed = 0
|
||
for row in executing:
|
||
if row.started_at is None:
|
||
continue
|
||
deadline = row.started_at + timedelta(
|
||
hours=row.time_window_hours + EXECUTING_OVERRUN_GRACE_HOURS
|
||
)
|
||
if now < deadline:
|
||
continue
|
||
reason = (
|
||
f"executing 超过 {row.time_window_hours}h 时间窗 + {EXECUTING_OVERRUN_GRACE_HOURS}h 宽限"
|
||
"仍未完成，按 ADR-0011 判失败"
|
||
)
|
||
if fail_eval(
|
||
session,
|
||
row.id,
|
||
reason,
|
||
"executing_timeout",
|
||
{"time_window_hours": row.time_window_hours},
|
||
):
|
||
failed += 1
|
||
|
||
return failed
|
||
|
||
|
||
def _window_has_pending_future_slots(eval_db, sessions, now) -> bool:
|
||
"""窗口未结束且会话数未达计划 → 未来时段还要建会话，不该催 analyst。"""
|
||
if not eval_db.plan or not eval_db.started_at:
|
||
return False
|
||
plan = eval_db.get_plan()
|
||
slots = plan.get("time_distribution") or []
|
||
end_hours: list[float] = []
|
||
planned = 0
|
||
for slot in slots:
|
||
parsed = parse_time_slot(slot.get("time_slot", ""))
|
||
if parsed is None:
|
||
continue
|
||
end_hours.append(parsed[1])
|
||
planned += int(slot.get("sessions", 0) or 0)
|
||
if not end_hours:
|
||
return False
|
||
window_end = (as_utc(eval_db.started_at) + timedelta(hours=max(end_hours))).replace(tzinfo=None)
|
||
if now >= window_end:
|
||
return False
|
||
expected = max(planned, int(plan.get("estimated_sessions", 0) or 0))
|
||
return len(sessions) < expected
|
||
|
||
|
||
def evals_needing_analyst_nudge(session: Session) -> list[str]:
|
||
"""返回需要平台催促 analyst 的 executing 评估 id 列表。
|
||
|
||
条件（ADR-0011）：存在会话且全部终态、时间窗口内无未到期时段欠账、
|
||
末会话终态已满 ANALYST_NUDGE_DELAY_MINUTES、催促次数 < ANALYST_NUDGE_MAX、
|
||
距上次催促已满 ANALYST_NUDGE_DELAY_MINUTES（冷却，避免每分钟连发）。
|
||
"""
|
||
now = utc_now().replace(tzinfo=None)
|
||
executing = session.exec(
|
||
select(IntelligentEvalDB).where(IntelligentEvalDB.status == IntelligentEvalStatus.EXECUTING.value)
|
||
).all()
|
||
|
||
needing: list[str] = []
|
||
for row in executing:
|
||
eval_sessions = session.exec(
|
||
select(IntelligentEvalSessionDB).where(IntelligentEvalSessionDB.eval_id == row.id)
|
||
).all()
|
||
if not eval_sessions or any(s.status not in _TERMINAL_SESSION_STATUSES for s in eval_sessions):
|
||
continue
|
||
# 冒烟教训：窗口未结束且会话数未达计划时，未来时段到期后还要建会话，
|
||
# 此时催促 analyst 会让报告提前收敛（漏掉后续时段的证据）
|
||
if _window_has_pending_future_slots(row, eval_sessions, now):
|
||
continue
|
||
closed_moments = [s.closed_at for s in eval_sessions if s.closed_at is not None]
|
||
if not closed_moments:
|
||
continue
|
||
last_closed = max(closed_moments)
|
||
if (now - last_closed).total_seconds() < ANALYST_NUDGE_DELAY_MINUTES * 60:
|
||
continue
|
||
if count_decisions(row.id, "analyst_nudge", session) >= ANALYST_NUDGE_MAX:
|
||
continue
|
||
# 冷却复用（ADR-0011）：最近的 analyst_nudge 或 worker_trigger 都会
|
||
# 唤起 agent（最长 10min），冷却期内不再催促以免堆叠孤儿 agent
|
||
recent_triggers = []
|
||
for dtype in ("analyst_nudge", "worker_trigger"):
|
||
last = _last_decision_at(session, row.id, dtype)
|
||
if last is not None:
|
||
recent_triggers.append(last)
|
||
if recent_triggers and (now - max(recent_triggers)).total_seconds() < ANALYST_NUDGE_DELAY_MINUTES * 60:
|
||
continue
|
||
needing.append(row.id)
|
||
return needing
|
||
|
||
|
||
def record_analyst_nudge(session: Session, eval_id: str) -> None:
|
||
"""落账一次 analyst 催促（analyst_nudge 决策日志，attempt 递增供上限计数）。"""
|
||
attempt = count_decisions(eval_id, "analyst_nudge", session) + 1
|
||
append_decision_log(
|
||
eval_id,
|
||
"analyst_nudge",
|
||
f"平台第 {attempt} 次催促 analyst 汇总报告（所有会话已终态）",
|
||
"platform",
|
||
{"platform_supplemented": True, "attempt": attempt},
|
||
session,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 触发失败持久化（ADR-0011）：连续 3 次触发失败 → 评估 failed
|
||
# ---------------------------------------------------------------------------
|
||
|
||
TRIGGER_FAILURE_MAX = 3
|
||
|
||
|
||
def eval_ids_with_pending_worker_tasks(session: Session) -> list[str]:
|
||
"""当前有待认领 worker 任务的评估 id（触发失败的受影响方）。"""
|
||
rows = session.exec(
|
||
select(IntelligentEvalTaskQueueDB.eval_id).where(IntelligentEvalTaskQueueDB.status == "pending")
|
||
).all()
|
||
return sorted(set(rows))
|
||
|
||
|
||
def record_trigger_failures(session: Session, *, channel: str, eval_ids: list[str], error: str) -> int:
|
||
"""触发失败落账：为每个受影响评估写一条 trigger_failed 决策日志。
|
||
|
||
channel 区分 "worker" / "planner"；失败原因截断落 reason，供决策过程页可见。
|
||
|
||
Returns:
|
||
落账条数。
|
||
"""
|
||
recorded = 0
|
||
for eval_id in eval_ids:
|
||
attempt = count_decisions(eval_id, "trigger_failed", session) + 1
|
||
append_decision_log(
|
||
eval_id,
|
||
"trigger_failed",
|
||
f"平台触发 {channel} 失败（第 {attempt} 次）：{error[:200]}",
|
||
"platform",
|
||
{"platform_supplemented": True, "channel": channel, "attempt": attempt},
|
||
session,
|
||
)
|
||
recorded += 1
|
||
return recorded
|
||
|
||
|
||
def fail_eval(session: Session, eval_id: str, reason: str, decision_type: str, context: dict[str, Any]) -> bool:
|
||
"""watchdog 判失败的统一接缝：表校验 → CAS 条件写 → 决策日志留痕。
|
||
|
||
所有平台侧"置 failed"必须穿过这里——``_TRANSITIONS`` 是唯一真相，
|
||
条件写挡住并发竞争（如用户在扫描间隙抢先取消）。
|
||
|
||
Returns:
|
||
True 判失败成功；False 表示跳过（评估不存在、非法转换或 CAS 冲突
|
||
——状态已被他人收敛，属正常竞争结局，仅留日志不当故障）。
|
||
"""
|
||
logger = logging.getLogger("agenteval")
|
||
repo = IntelligentEvalRepository(session)
|
||
ev = repo.get(eval_id)
|
||
if ev is None:
|
||
logger.info("fail_eval 跳过：评估 %s 已不存在", eval_id)
|
||
return False
|
||
if IntelligentEvalStatus.FAILED not in _TRANSITIONS.get(ev.status, set()):
|
||
logger.warning("fail_eval 被状态表挡下：%s 当前 %s，不允许转 failed", eval_id, ev.status.value)
|
||
return False
|
||
now = utc_now()
|
||
result = repo._compare_and_set_fields(
|
||
eval_id,
|
||
expected_status=ev.status,
|
||
values={
|
||
"status": IntelligentEvalStatus.FAILED.value,
|
||
"plan_feedback": reason,
|
||
"updated_at": now,
|
||
"completed_at": now,
|
||
},
|
||
)
|
||
if result.status is not CompareAndSetStatus.APPLIED:
|
||
logger.info("fail_eval 跳过：评估 %s 状态已被并发收敛（%s）", eval_id, result.status)
|
||
return False
|
||
append_decision_log(
|
||
eval_id,
|
||
decision_type,
|
||
f"平台兜底：{reason}",
|
||
"platform",
|
||
{"platform_supplemented": True, **context},
|
||
session,
|
||
)
|
||
return True
|
||
|
||
|
||
def enforce_trigger_failure_gates(session: Session) -> int:
|
||
"""连续触发失败判死（ADR-0011）。
|
||
|
||
- planning 评估：planner 触发失败 ≥ TRIGGER_FAILURE_MAX 次即失败
|
||
（仍在 planning 说明 planner 从未成功，失败必然连续）。
|
||
- executing 评估：自最近一次任务被认领以来 worker 触发失败 ≥ 上限即失败
|
||
（期间无认领 = 触发从未生效，即"连续"）。
|
||
|
||
Returns:
|
||
被判失败的评估数。
|
||
"""
|
||
def _failures(eval_id: str, channel: str) -> list:
|
||
logs = session.exec(
|
||
select(IntelligentEvalDecisionLogDB).where(
|
||
IntelligentEvalDecisionLogDB.eval_id == eval_id,
|
||
IntelligentEvalDecisionLogDB.decision_type == "trigger_failed",
|
||
)
|
||
).all()
|
||
return [x for x in logs if x.get_context().get("channel") == channel]
|
||
|
||
failed = 0
|
||
stuck = session.exec(
|
||
select(IntelligentEvalDB).where(
|
||
IntelligentEvalDB.status.in_(
|
||
[IntelligentEvalStatus.PLANNING.value, IntelligentEvalStatus.EXECUTING.value]
|
||
)
|
||
)
|
||
).all()
|
||
for row in stuck:
|
||
if row.status == IntelligentEvalStatus.PLANNING.value:
|
||
failures = _failures(row.id, "planner")
|
||
if len(failures) < TRIGGER_FAILURE_MAX:
|
||
continue
|
||
reason = f"planner 连续触发失败 {len(failures)} 次，按 ADR-0011 判失败"
|
||
if fail_eval(
|
||
session, row.id, reason, "trigger_failure_gate", {"channel": "planner", "failures": len(failures)}
|
||
):
|
||
failed += 1
|
||
else:
|
||
last_assigned = session.exec(
|
||
select(func.max(IntelligentEvalTaskQueueDB.assigned_at)).where(
|
||
IntelligentEvalTaskQueueDB.eval_id == row.id
|
||
)
|
||
).one()
|
||
failures = [
|
||
x
|
||
for x in _failures(row.id, "worker")
|
||
if last_assigned is None or (x.created_at is not None and x.created_at > last_assigned)
|
||
]
|
||
if len(failures) < TRIGGER_FAILURE_MAX:
|
||
continue
|
||
reason = f"worker 连续触发失败 {len(failures)} 次且期间无任务被认领，按 ADR-0011 判失败"
|
||
if fail_eval(
|
||
session, row.id, reason, "trigger_failure_gate", {"channel": "worker", "failures": len(failures)}
|
||
):
|
||
failed += 1
|
||
|
||
return failed
|