From 58c2ad0227fad891dbb2f21045bb4520b3aecbf6 Mon Sep 17 00:00:00 2001 From: sinohqb Date: Fri, 21 Aug 2026 03:35:05 +0800 Subject: [PATCH] refactor(intelligent-eval): unify watchdog failures behind fail_eval seam MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 架构审查候选③（状态机归一）：_TRANSITIONS 成为评估状态机的唯一真相。 三个 watchdog（planning 双闸 / executing 兜底 / 触发失败闸门）原先直接 row.status = FAILED 绕过转换表、手工重复写字段。收编到新的公开接缝 fail_eval(session, eval_id, reason, decision_type, context)：表校验 → repo CAS 条件写（status/plan_feedback/updated_at/completed_at 一条 SQL） → append_decision_log 留痕。 CAS 冲突（如用户在扫描间隙抢先取消）跳过并留日志，不当故障。API 路径 本已在表内，不动。新增 4 个契约测试，870 tests passed，零行为变化。 --- .../agenteval/intelligent_eval/lifecycle.py | 101 ++++++++++-------- ...elligent_eval_watchdog_characterization.py | 70 ++++++++++++ 2 files changed, 129 insertions(+), 42 deletions(-) diff --git a/backend/agenteval/intelligent_eval/lifecycle.py b/backend/agenteval/intelligent_eval/lifecycle.py index 338f901..5830d62 100644 --- a/backend/agenteval/intelligent_eval/lifecycle.py +++ b/backend/agenteval/intelligent_eval/lifecycle.py @@ -10,6 +10,7 @@ 非法转换抛 IntelligentEvalTransitionError，路由层映射为 409。 """ +import logging from datetime import timedelta from typing import Any, Optional @@ -558,7 +559,7 @@ def enforce_planning_gates(session: Session) -> int: """ from sqlmodel import select - from agenteval.intelligent_eval.decision_logs import append_decision_log, count_decisions + from agenteval.intelligent_eval.decision_logs import count_decisions from agenteval.storage.db import IntelligentEvalDB now = utc_now().replace(tzinfo=None) @@ -577,22 +578,15 @@ def enforce_planning_gates(session: Session) -> int: reason = f"planning 超过 {PLANNING_TIMEOUT_MINUTES} 分钟未提交粗计划，按 ADR-0011 双闸判失败" else: continue - row.status = IntelligentEvalStatus.FAILED.value - row.plan_feedback = reason - row.updated_at = utc_now() - row.completed_at = utc_now() - failed += 1 - append_decision_log( - row.id, - "planning_gate_failed", - f"平台兜底：{reason}", - "platform", - {"platform_supplemented": True, "attempts": attempts, "age_minutes": int(age_minutes)}, + if fail_eval( session, - ) + row.id, + reason, + "planning_gate_failed", + {"attempts": attempts, "age_minutes": int(age_minutes)}, + ): + failed += 1 - if failed: - session.commit() return failed @@ -619,7 +613,6 @@ def enforce_executing_ceiling(session: Session) -> int: """ from sqlmodel import select - from agenteval.intelligent_eval.decision_logs import append_decision_log from agenteval.storage.db import IntelligentEvalDB now = utc_now().replace(tzinfo=None) @@ -640,22 +633,15 @@ def enforce_executing_ceiling(session: Session) -> int: f"executing 超过 {row.time_window_hours}h 时间窗 + {EXECUTING_OVERRUN_GRACE_HOURS}h 宽限" "仍未完成，按 ADR-0011 判失败" ) - row.status = IntelligentEvalStatus.FAILED.value - row.plan_feedback = reason - row.updated_at = utc_now() - row.completed_at = utc_now() - failed += 1 - append_decision_log( - row.id, - "executing_timeout", - f"平台兜底：{reason}", - "platform", - {"platform_supplemented": True, "time_window_hours": row.time_window_hours}, + if fail_eval( session, - ) + row.id, + reason, + "executing_timeout", + {"time_window_hours": row.time_window_hours}, + ): + failed += 1 - if failed: - session.commit() return failed @@ -792,21 +778,50 @@ def record_trigger_failures(session: Session, *, channel: str, eval_ids: list[st return recorded -def _fail_eval(session: Session, row, reason: str, decision_type: str, context: dict[str, Any]) -> None: +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 冲突 + ——状态已被他人收敛，属正常竞争结局，仅留日志不当故障）。 + """ 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() + 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( - row.id, + eval_id, decision_type, f"平台兜底：{reason}", "platform", {"platform_supplemented": True, **context}, session, ) + return True def enforce_trigger_failure_gates(session: Session) -> int: @@ -852,8 +867,10 @@ def enforce_trigger_failure_gates(session: Session) -> int: if len(failures) < TRIGGER_FAILURE_MAX: continue reason = f"planner 连续触发失败 {len(failures)} 次，按 ADR-0011 判失败" - _fail_eval(session, row, reason, "trigger_failure_gate", {"channel": "planner", "failures": len(failures)}) - failed += 1 + 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( @@ -868,9 +885,9 @@ def enforce_trigger_failure_gates(session: Session) -> int: if len(failures) < TRIGGER_FAILURE_MAX: continue reason = f"worker 连续触发失败 {len(failures)} 次且期间无任务被认领，按 ADR-0011 判失败" - _fail_eval(session, row, reason, "trigger_failure_gate", {"channel": "worker", "failures": len(failures)}) - failed += 1 + if fail_eval( + session, row.id, reason, "trigger_failure_gate", {"channel": "worker", "failures": len(failures)} + ): + failed += 1 - if failed: - session.commit() return failed diff --git a/tests/unit/test_intelligent_eval_watchdog_characterization.py b/tests/unit/test_intelligent_eval_watchdog_characterization.py index 17d4183..43fdc6f 100644 --- a/tests/unit/test_intelligent_eval_watchdog_characterization.py +++ b/tests/unit/test_intelligent_eval_watchdog_characterization.py @@ -498,3 +498,73 @@ def test_task_pickup_breaks_consecutive_worker_failures(db_session: Session): db_session.refresh(ev) assert ev.status == IntelligentEvalStatus.EXECUTING.value + + +# --------------------------------------------------------------------------- +# fail_eval 统一接缝契约（状态机归一）：watchdog 判失败必须穿过表校验 + CAS +# --------------------------------------------------------------------------- + + +def test_fail_eval_writes_fields_and_log(db_session: Session): + """fail_eval：CAS 条件写 status/plan_feedback/completed_at + 决策日志留痕。""" + from agenteval.intelligent_eval.lifecycle import fail_eval + + ev = _make_executing_eval(db_session) + assert fail_eval(db_session, ev.id, "测试判败", "executing_timeout", {"k": 1}) is True + + db_session.refresh(ev) + assert ev.status == IntelligentEvalStatus.FAILED.value + assert ev.plan_feedback == "测试判败" + assert ev.completed_at is not None + logs = db_session.exec( + select(IntelligentEvalDecisionLogDB).where(IntelligentEvalDecisionLogDB.eval_id == ev.id) + ).all() + assert len(logs) == 1 + assert logs[0].decision_type == "executing_timeout" + assert logs[0].cron_id == "platform" + + +def test_fail_eval_blocked_by_transition_table(db_session: Session): + """终态评估不允许再判 failed——_TRANSITIONS 是唯一真相。""" + from agenteval.intelligent_eval.lifecycle import fail_eval + + ev = _make_executing_eval(db_session) + ev.status = IntelligentEvalStatus.COMPLETED.value + db_session.add(ev) + db_session.commit() + + assert fail_eval(db_session, ev.id, "不该生效", "executing_timeout", {}) is False + db_session.refresh(ev) + assert ev.status == IntelligentEvalStatus.COMPLETED.value + logs = db_session.exec( + select(IntelligentEvalDecisionLogDB).where(IntelligentEvalDecisionLogDB.eval_id == ev.id) + ).all() + assert logs == [] + + +def test_fail_eval_skips_on_cas_conflict(db_session: Session, monkeypatch): + """CAS 冲突（用户抢先取消等并发收敛）→ 跳过不当故障，不写日志。""" + from agenteval.intelligent_eval.lifecycle import fail_eval + from agenteval.intelligent_eval.repository import ( + CompareAndSetResult, + CompareAndSetStatus, + IntelligentEvalRepository, + ) + + ev = _make_executing_eval(db_session) + monkeypatch.setattr( + IntelligentEvalRepository, + "_compare_and_set_fields", + lambda self, eval_id, *, expected_status, values: CompareAndSetResult(status=CompareAndSetStatus.CONFLICT), + ) + + assert fail_eval(db_session, ev.id, "竞争失败", "executing_timeout", {}) is False + db_session.refresh(ev) + assert ev.status == IntelligentEvalStatus.EXECUTING.value + + +def test_fail_eval_skips_unknown_eval(db_session: Session): + """评估不存在 → 跳过（NOT_FOUND 语义）。""" + from agenteval.intelligent_eval.lifecycle import fail_eval + + assert fail_eval(db_session, "no-such-eval", "r", "executing_timeout", {}) is False