AgentEvalTool/backend/agenteval/intelligent_eval/lifecycle/watchdog.py

"""Watchdog gates (ADR-0011): platform-side failure convergence.

planning 双闸、executing 超窗兜底、analyst 催促、触发冷却与触发失败判死。
所有平台侧「置 failed」都穿过 :func:`fail_eval` —— 转换表是唯一真相，
条件写挡住并发竞争（如用户在扫描间隙抢先取消）。
"""

import logging
from datetime import timedelta
from typing import Any

from sqlmodel import Session, func, select

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.lifecycle._core import _TRANSITIONS
from agenteval.intelligent_eval.models import IntelligentEvalStatus
from agenteval.intelligent_eval.repository import CompareAndSetStatus, IntelligentEvalRepository
from agenteval.storage.db import (
    IntelligentEvalDB,
    IntelligentEvalDecisionLogDB,
    IntelligentEvalSessionDB,
    IntelligentEvalTaskQueueDB,
    as_utc,
    utc_now,
)

# ---------------------------------------------------------------------------
# planning 双闸：触发 ≥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 兜底：超窗判败 + 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,
    )


# ---------------------------------------------------------------------------
# 触发失败持久化：连续 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