d0487b54b4
冒烟发现 analyst_nudge 不考虑时间窗口：首会话完成后 10 分钟即催促， 但后续时段尚未到期，导致报告提前收敛、漏掉计划内会话。催促闸门新增 "窗口未结束且会话数未达计划则跳过"；worker 触发指令同步明确仅当所有 时段会话都达终态才转 analyst。
501 lines
17 KiB
Python
501 lines
17 KiB
Python
"""Watchdog contract tests for ADR-0011 (原 Phase 0 刻画测试).
|
||
|
||
Phase 0 时本文件锁定「无 watchdog」的旧行为；Phase 1 全部落地后均为新契约：
|
||
- 会话过期（expire_stale_running_sessions）：60min 无新轮次 → expired；
|
||
- planning 双闸（enforce_planning_gates）：触发 ≥5 次或 ≥30min → failed；
|
||
- executing 兜底（enforce_executing_ceiling + analyst 催促）：超窗判败。
|
||
"""
|
||
import json
|
||
from datetime import timedelta
|
||
|
||
from agenteval.intelligent_eval.lifecycle import (
|
||
PLANNING_MAX_ATTEMPTS,
|
||
SESSION_IDLE_EXPIRE_MINUTES,
|
||
enforce_executing_ceiling,
|
||
enforce_planning_gates,
|
||
enforce_trigger_failure_gates,
|
||
expire_stale_running_sessions,
|
||
)
|
||
from agenteval.intelligent_eval.models import IntelligentEvalStatus
|
||
from agenteval.intelligent_eval.task_queue import (
|
||
requeue_stale_assigned_tasks,
|
||
scan_and_enqueue_tasks,
|
||
settle_tasks_for_finished_evals,
|
||
)
|
||
from agenteval.storage.db import (
|
||
IntelligentEvalDB,
|
||
IntelligentEvalDecisionLogDB,
|
||
IntelligentEvalMessageDB,
|
||
IntelligentEvalSessionDB,
|
||
utc_now,
|
||
)
|
||
from sqlmodel import Session, select
|
||
|
||
|
||
def _run_scan_cycle(session: Session) -> None:
|
||
"""Simulate one platform scan iteration (与 app.py 扫描循环同序)。"""
|
||
requeue_stale_assigned_tasks(session)
|
||
expire_stale_running_sessions(session)
|
||
enforce_planning_gates(session)
|
||
enforce_executing_ceiling(session)
|
||
enforce_trigger_failure_gates(session)
|
||
scan_and_enqueue_tasks(session)
|
||
settle_tasks_for_finished_evals(session)
|
||
|
||
|
||
def _make_executing_eval(db_session: Session, **overrides) -> IntelligentEvalDB:
|
||
ev = IntelligentEvalDB(
|
||
name=overrides.pop("name", "wd-eval"),
|
||
target_id="t1",
|
||
status=IntelligentEvalStatus.EXECUTING.value,
|
||
started_at=overrides.pop("started_at", utc_now() - timedelta(hours=3)),
|
||
**overrides,
|
||
)
|
||
db_session.add(ev)
|
||
db_session.commit()
|
||
return ev
|
||
|
||
|
||
def test_planning_eval_stuck_too_long_fails(db_session: Session):
|
||
"""ADR-0011 闸二：planning 超 30 分钟未提交粗计划 → failed（原因落 plan_feedback）。"""
|
||
ev = IntelligentEvalDB(
|
||
name="stuck-planning",
|
||
target_id="t1",
|
||
status=IntelligentEvalStatus.PLANNING.value,
|
||
created_at=utc_now() - timedelta(hours=5),
|
||
updated_at=utc_now() - timedelta(hours=5),
|
||
)
|
||
db_session.add(ev)
|
||
db_session.commit()
|
||
|
||
_run_scan_cycle(db_session)
|
||
|
||
db_session.refresh(ev)
|
||
assert ev.status == IntelligentEvalStatus.FAILED.value
|
||
assert "30" in ev.plan_feedback
|
||
assert ev.completed_at is not None
|
||
|
||
logs = db_session.exec(
|
||
select(IntelligentEvalDecisionLogDB).where(
|
||
IntelligentEvalDecisionLogDB.eval_id == ev.id,
|
||
IntelligentEvalDecisionLogDB.decision_type == "planning_gate_failed",
|
||
)
|
||
).all()
|
||
assert len(logs) == 1
|
||
|
||
|
||
def test_planning_eval_fails_at_max_planner_triggers(db_session: Session):
|
||
"""ADR-0011 闸一：planner 触发次数达上限仍未提交 → failed（即使时长未到）。"""
|
||
from agenteval.intelligent_eval.lifecycle import record_planner_triggers
|
||
|
||
ev = IntelligentEvalDB(
|
||
name="attempt-gated",
|
||
target_id="t1",
|
||
status=IntelligentEvalStatus.PLANNING.value,
|
||
created_at=utc_now(),
|
||
updated_at=utc_now(),
|
||
)
|
||
db_session.add(ev)
|
||
db_session.commit()
|
||
|
||
for _ in range(PLANNING_MAX_ATTEMPTS):
|
||
assert record_planner_triggers(db_session) == 1
|
||
# 触发冷却 10min：把刚落账的日志时间改老，模拟跨冷却的多次触发
|
||
logs = db_session.exec(
|
||
select(IntelligentEvalDecisionLogDB).where(
|
||
IntelligentEvalDecisionLogDB.eval_id == ev.id,
|
||
IntelligentEvalDecisionLogDB.decision_type == "planner_trigger",
|
||
)
|
||
).all()
|
||
for log in logs:
|
||
log.created_at = utc_now() - timedelta(minutes=30)
|
||
db_session.commit()
|
||
|
||
_run_scan_cycle(db_session)
|
||
|
||
db_session.refresh(ev)
|
||
assert ev.status == IntelligentEvalStatus.FAILED.value
|
||
assert "触发" in ev.plan_feedback
|
||
|
||
|
||
def test_planner_trigger_respects_cooldown(db_session: Session):
|
||
"""ADR-0011 孤儿 agent 双管：同一评估 10min 冷却内不重复触发 planner。"""
|
||
from agenteval.intelligent_eval.lifecycle import record_planner_triggers
|
||
|
||
ev = IntelligentEvalDB(
|
||
name="cooldown-planning",
|
||
target_id="t1",
|
||
status=IntelligentEvalStatus.PLANNING.value,
|
||
created_at=utc_now(),
|
||
updated_at=utc_now(),
|
||
)
|
||
db_session.add(ev)
|
||
db_session.commit()
|
||
|
||
assert record_planner_triggers(db_session) == 1
|
||
db_session.commit()
|
||
# 冷却期内：不再计次数（返回 0 → 平台不触发 docker exec）
|
||
assert record_planner_triggers(db_session) == 0
|
||
|
||
|
||
def test_fresh_planning_eval_not_gated(db_session: Session):
|
||
"""刚进 planning 且触发次数未达上限的评估不受影响。"""
|
||
ev = IntelligentEvalDB(
|
||
name="fresh-planning",
|
||
target_id="t1",
|
||
status=IntelligentEvalStatus.PLANNING.value,
|
||
created_at=utc_now(),
|
||
updated_at=utc_now(),
|
||
)
|
||
db_session.add(ev)
|
||
db_session.commit()
|
||
|
||
_run_scan_cycle(db_session)
|
||
|
||
db_session.refresh(ev)
|
||
assert ev.status == IntelligentEvalStatus.PLANNING.value
|
||
|
||
|
||
def test_executing_eval_past_window_stays_executing(db_session: Session):
|
||
"""CURRENT BEHAVIOR: an executing eval far beyond time_window_hours is never
|
||
force-finished by the platform (ADR-0011 adds the +2h ceiling)."""
|
||
ev = IntelligentEvalDB(
|
||
name="overdue-executing",
|
||
target_id="t1",
|
||
status=IntelligentEvalStatus.EXECUTING.value,
|
||
time_window_hours=1,
|
||
started_at=utc_now() - timedelta(hours=10),
|
||
plan=json.dumps({"time_distribution": [{"time_slot": "0-1h", "sessions": 1}]}),
|
||
)
|
||
db_session.add(ev)
|
||
db_session.commit()
|
||
|
||
_run_scan_cycle(db_session)
|
||
|
||
db_session.refresh(ev)
|
||
assert ev.status == IntelligentEvalStatus.FAILED.value
|
||
assert "宽限" in ev.plan_feedback
|
||
assert ev.completed_at is not None
|
||
|
||
logs = db_session.exec(
|
||
select(IntelligentEvalDecisionLogDB).where(
|
||
IntelligentEvalDecisionLogDB.eval_id == ev.id,
|
||
IntelligentEvalDecisionLogDB.decision_type == "executing_timeout",
|
||
)
|
||
).all()
|
||
assert len(logs) == 1
|
||
|
||
|
||
def test_executing_within_window_plus_grace_not_gated(db_session: Session):
|
||
"""时间窗 + 宽限内的 executing 评估不受影响。"""
|
||
ev = IntelligentEvalDB(
|
||
name="in-window",
|
||
target_id="t1",
|
||
status=IntelligentEvalStatus.EXECUTING.value,
|
||
time_window_hours=24,
|
||
started_at=utc_now() - timedelta(hours=10),
|
||
plan=json.dumps({"time_distribution": [{"time_slot": "0-1h", "sessions": 1}]}),
|
||
)
|
||
db_session.add(ev)
|
||
db_session.commit()
|
||
|
||
_run_scan_cycle(db_session)
|
||
|
||
db_session.refresh(ev)
|
||
assert ev.status == IntelligentEvalStatus.EXECUTING.value
|
||
|
||
|
||
def _make_terminal_session(
|
||
db_session: Session,
|
||
eval_id: str,
|
||
*,
|
||
status: str = "completed",
|
||
closed_minutes_ago: int = 30,
|
||
) -> IntelligentEvalSessionDB:
|
||
row = IntelligentEvalSessionDB(
|
||
eval_id=eval_id,
|
||
target_id="t1",
|
||
status=status,
|
||
goal="完成退货",
|
||
created_at=utc_now() - timedelta(minutes=closed_minutes_ago + 10),
|
||
closed_at=utc_now() - timedelta(minutes=closed_minutes_ago),
|
||
)
|
||
db_session.add(row)
|
||
db_session.commit()
|
||
return row
|
||
|
||
|
||
def test_analyst_nudge_when_all_sessions_terminal(db_session: Session):
|
||
"""ADR-0011：全部会话终态满 10min 且评估仍 executing → 需要 analyst 催促。"""
|
||
from agenteval.intelligent_eval.lifecycle import evals_needing_analyst_nudge
|
||
|
||
ev = _make_executing_eval(db_session)
|
||
_make_terminal_session(db_session, ev.id, closed_minutes_ago=30)
|
||
|
||
assert evals_needing_analyst_nudge(db_session) == [ev.id]
|
||
|
||
|
||
def test_analyst_nudge_skipped_before_delay(db_session: Session):
|
||
"""末会话终态未满 10min 不催促。"""
|
||
from agenteval.intelligent_eval.lifecycle import evals_needing_analyst_nudge
|
||
|
||
ev = _make_executing_eval(db_session)
|
||
_make_terminal_session(db_session, ev.id, closed_minutes_ago=5)
|
||
|
||
assert evals_needing_analyst_nudge(db_session) == []
|
||
|
||
|
||
def test_analyst_nudge_skipped_with_running_session(db_session: Session):
|
||
"""存在非终态会话时不催促（worker 还在干活）。"""
|
||
from agenteval.intelligent_eval.lifecycle import evals_needing_analyst_nudge
|
||
|
||
ev = _make_executing_eval(db_session)
|
||
_make_terminal_session(db_session, ev.id, closed_minutes_ago=30)
|
||
_make_terminal_session(db_session, ev.id, status="running", closed_minutes_ago=0)
|
||
|
||
assert evals_needing_analyst_nudge(db_session) == []
|
||
|
||
|
||
def test_analyst_nudge_skipped_while_future_slot_pending(db_session: Session):
|
||
"""冒烟教训：窗口内还有未到期时段（会话数未达计划）时不催 analyst。"""
|
||
from agenteval.intelligent_eval.lifecycle import evals_needing_analyst_nudge
|
||
|
||
ev = _make_executing_eval(
|
||
db_session,
|
||
started_at=utc_now() - timedelta(minutes=20),
|
||
plan=json.dumps(
|
||
{
|
||
"estimated_sessions": 2,
|
||
"time_distribution": [
|
||
{"time_slot": "0-30min", "sessions": 1},
|
||
{"time_slot": "30-60min", "sessions": 1},
|
||
],
|
||
}
|
||
),
|
||
)
|
||
_make_terminal_session(db_session, ev.id, closed_minutes_ago=15)
|
||
|
||
assert evals_needing_analyst_nudge(db_session) == []
|
||
|
||
|
||
def test_analyst_nudge_fires_after_window_elapsed_with_deficit(db_session: Session):
|
||
"""窗口已结束但会话欠账：不再等未来时段，催促 analyst 按现有证据收敛。"""
|
||
from agenteval.intelligent_eval.lifecycle import evals_needing_analyst_nudge
|
||
|
||
ev = _make_executing_eval(
|
||
db_session,
|
||
started_at=utc_now() - timedelta(minutes=70),
|
||
plan=json.dumps(
|
||
{
|
||
"estimated_sessions": 2,
|
||
"time_distribution": [
|
||
{"time_slot": "0-30min", "sessions": 1},
|
||
{"time_slot": "30-60min", "sessions": 1},
|
||
],
|
||
}
|
||
),
|
||
)
|
||
_make_terminal_session(db_session, ev.id, closed_minutes_ago=30)
|
||
|
||
assert evals_needing_analyst_nudge(db_session) == [ev.id]
|
||
|
||
|
||
def test_analyst_nudge_capped_and_cooled_down(db_session: Session):
|
||
"""催促上限 3 次；两次催促间隔未满 10min 不重复。"""
|
||
from agenteval.intelligent_eval.lifecycle import (
|
||
ANALYST_NUDGE_MAX,
|
||
evals_needing_analyst_nudge,
|
||
record_analyst_nudge,
|
||
)
|
||
|
||
ev = _make_executing_eval(db_session)
|
||
_make_terminal_session(db_session, ev.id, closed_minutes_ago=30)
|
||
|
||
record_analyst_nudge(db_session, ev.id)
|
||
db_session.commit()
|
||
# 冷却期内不再催促
|
||
assert evals_needing_analyst_nudge(db_session) == []
|
||
|
||
# 直接把已有催促日志时间改老，模拟冷却结束，补满到上限
|
||
logs = db_session.exec(
|
||
select(IntelligentEvalDecisionLogDB).where(
|
||
IntelligentEvalDecisionLogDB.eval_id == ev.id,
|
||
IntelligentEvalDecisionLogDB.decision_type == "analyst_nudge",
|
||
)
|
||
).all()
|
||
for log in logs:
|
||
log.created_at = utc_now() - timedelta(minutes=30)
|
||
db_session.commit()
|
||
for _ in range(ANALYST_NUDGE_MAX - 1):
|
||
assert evals_needing_analyst_nudge(db_session) == [ev.id]
|
||
record_analyst_nudge(db_session, ev.id)
|
||
db_session.commit()
|
||
more = db_session.exec(
|
||
select(IntelligentEvalDecisionLogDB).where(
|
||
IntelligentEvalDecisionLogDB.eval_id == ev.id,
|
||
IntelligentEvalDecisionLogDB.decision_type == "analyst_nudge",
|
||
)
|
||
).all()
|
||
for log in more:
|
||
log.created_at = utc_now() - timedelta(minutes=30)
|
||
db_session.commit()
|
||
|
||
# 达到上限后不再催促（等待 executing 超窗兜底判败）
|
||
assert evals_needing_analyst_nudge(db_session) == []
|
||
|
||
|
||
def test_idle_running_session_expires(db_session: Session):
|
||
"""ADR-0011：running 会话 60 分钟无新轮次 → expired（终态，写 closed_at）。"""
|
||
ev = _make_executing_eval(db_session)
|
||
session_row = IntelligentEvalSessionDB(
|
||
eval_id=ev.id,
|
||
target_id="t1",
|
||
status="running",
|
||
goal="完成退货",
|
||
created_at=utc_now() - timedelta(minutes=SESSION_IDLE_EXPIRE_MINUTES + 30),
|
||
)
|
||
db_session.add(session_row)
|
||
db_session.commit()
|
||
|
||
_run_scan_cycle(db_session)
|
||
|
||
db_session.refresh(session_row)
|
||
assert session_row.status == "expired"
|
||
assert session_row.closed_at is not None
|
||
|
||
logs = db_session.exec(
|
||
select(IntelligentEvalDecisionLogDB).where(
|
||
IntelligentEvalDecisionLogDB.eval_id == ev.id,
|
||
IntelligentEvalDecisionLogDB.decision_type == "session_expired",
|
||
)
|
||
).all()
|
||
assert len(logs) == 1
|
||
|
||
|
||
def test_active_running_session_not_expired(db_session: Session):
|
||
"""最近有消息的 running 会话不过期（按最后一条消息时间判定）。"""
|
||
ev = _make_executing_eval(db_session)
|
||
session_row = IntelligentEvalSessionDB(
|
||
eval_id=ev.id,
|
||
target_id="t1",
|
||
status="running",
|
||
goal="咨询物流",
|
||
turn_count=3,
|
||
created_at=utc_now() - timedelta(hours=5),
|
||
)
|
||
db_session.add(session_row)
|
||
db_session.commit()
|
||
db_session.add(
|
||
IntelligentEvalMessageDB(
|
||
session_id=session_row.id,
|
||
role="assistant",
|
||
content="在的",
|
||
created_at=utc_now() - timedelta(minutes=5),
|
||
)
|
||
)
|
||
db_session.commit()
|
||
|
||
_run_scan_cycle(db_session)
|
||
|
||
db_session.refresh(session_row)
|
||
assert session_row.status == "running"
|
||
|
||
|
||
def test_expired_session_unblocks_submit_report(db_session: Session):
|
||
"""ADR-0011：会话过期后（全部终态）submit_report 不再被拒。"""
|
||
from agenteval.intelligent_eval import lifecycle
|
||
|
||
ev = _make_executing_eval(db_session)
|
||
session_row = IntelligentEvalSessionDB(
|
||
eval_id=ev.id,
|
||
target_id="t1",
|
||
status="running",
|
||
goal="完成退货",
|
||
created_at=utc_now() - timedelta(minutes=SESSION_IDLE_EXPIRE_MINUTES + 30),
|
||
)
|
||
db_session.add(session_row)
|
||
db_session.commit()
|
||
|
||
_run_scan_cycle(db_session)
|
||
|
||
done = lifecycle.submit_report(db_session, ev.id, {"summary": "done"})
|
||
assert done.status is IntelligentEvalStatus.COMPLETED
|
||
|
||
|
||
def _record_failures(db_session: Session, eval_id: str, channel: str, count: int) -> None:
|
||
from agenteval.intelligent_eval.lifecycle import record_trigger_failures
|
||
|
||
for i in range(count):
|
||
record_trigger_failures(
|
||
db_session, channel=channel, eval_ids=[eval_id], error=f"模拟失败 {i + 1}"
|
||
)
|
||
db_session.commit()
|
||
|
||
|
||
def test_planning_eval_fails_after_repeated_planner_trigger_failures(db_session: Session):
|
||
"""ADR-0011：planner 连续触发失败 3 次 → planning 评估判失败。"""
|
||
ev = IntelligentEvalDB(
|
||
name="planner-fail",
|
||
target_id="t1",
|
||
status=IntelligentEvalStatus.PLANNING.value,
|
||
created_at=utc_now(),
|
||
updated_at=utc_now(),
|
||
)
|
||
db_session.add(ev)
|
||
db_session.commit()
|
||
_record_failures(db_session, ev.id, "planner", 3)
|
||
|
||
_run_scan_cycle(db_session)
|
||
|
||
db_session.refresh(ev)
|
||
assert ev.status == IntelligentEvalStatus.FAILED.value
|
||
assert "触发失败" in ev.plan_feedback
|
||
|
||
|
||
def test_executing_eval_fails_after_repeated_worker_trigger_failures(db_session: Session):
|
||
"""ADR-0011：worker 连续触发失败 3 次且期间无任务被认领 → 评估判失败。"""
|
||
ev = _make_executing_eval(db_session)
|
||
_record_failures(db_session, ev.id, "worker", 3)
|
||
|
||
_run_scan_cycle(db_session)
|
||
|
||
db_session.refresh(ev)
|
||
assert ev.status == IntelligentEvalStatus.FAILED.value
|
||
assert "worker" in ev.plan_feedback
|
||
|
||
|
||
def test_trigger_failures_below_threshold_keep_eval_alive(db_session: Session):
|
||
"""失败次数未达上限不判死。"""
|
||
ev = _make_executing_eval(db_session)
|
||
_record_failures(db_session, ev.id, "worker", 2)
|
||
|
||
_run_scan_cycle(db_session)
|
||
|
||
db_session.refresh(ev)
|
||
assert ev.status == IntelligentEvalStatus.EXECUTING.value
|
||
|
||
|
||
def test_task_pickup_breaks_consecutive_worker_failures(db_session: Session):
|
||
"""失败之后有任务被认领 → 「连续」被打断，不判死。"""
|
||
from agenteval.storage.db import IntelligentEvalTaskQueueDB
|
||
|
||
ev = _make_executing_eval(db_session)
|
||
_record_failures(db_session, ev.id, "worker", 3)
|
||
# 失败落账之后任务被认领（触发实际生效过）→ 连续性中断
|
||
db_session.add(
|
||
IntelligentEvalTaskQueueDB(
|
||
eval_id=ev.id,
|
||
status="completed",
|
||
priority=1,
|
||
reason="slot_due",
|
||
assigned_cron_id="manual-run-ok",
|
||
assigned_at=utc_now(),
|
||
)
|
||
)
|
||
db_session.commit()
|
||
|
||
_run_scan_cycle(db_session)
|
||
|
||
db_session.refresh(ev)
|
||
assert ev.status == IntelligentEvalStatus.EXECUTING.value
|