diff --git a/backend/agenteval/intelligent_eval/lifecycle.py b/backend/agenteval/intelligent_eval/lifecycle.py index ac8831e..c5324a2 100644 --- a/backend/agenteval/intelligent_eval/lifecycle.py +++ b/backend/agenteval/intelligent_eval/lifecycle.py @@ -20,6 +20,8 @@ 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, @@ -33,7 +35,15 @@ from agenteval.intelligent_eval.repository import ( IntelligentEvalRepository, IntelligentEvalSessionRepository, ) -from agenteval.storage.db import IntelligentEvalSessionDB, as_utc, utc_now +from agenteval.storage.db import ( + IntelligentEvalDB, + IntelligentEvalDecisionLogDB, + IntelligentEvalMessageDB, + IntelligentEvalSessionDB, + IntelligentEvalTaskQueueDB, + as_utc, + utc_now, +) from agenteval.storage.repository import TargetRepository # 合法转换表：当前状态 → 允许的目标状态集合 @@ -122,8 +132,6 @@ def create_eval( ) -> IntelligentEval: """创建智能评估并直接进入 planning 状态（draft → planning 一步完成）。""" from agenteval.intelligent_eval.config_snapshot import save_snapshot - from agenteval.storage.db import IntelligentEvalDB - if TargetRepository(session).get(target_id) is None: raise IntelligentEvalNotFoundError(f"target {target_id} not found") @@ -154,8 +162,6 @@ def create_eval( 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 - from agenteval.storage.db import IntelligentEvalDB - repo = IntelligentEvalRepository(session) result = repo._submit_plan_if_planning(eval_id, plan) ev = _resolve_write( @@ -423,12 +429,6 @@ def expire_stale_running_sessions(session: Session) -> int: Returns: 过期的会话数。 """ - from sqlalchemy import func - from sqlmodel import select - - from agenteval.intelligent_eval.decision_logs import append_decision_log - from agenteval.storage.db import IntelligentEvalMessageDB, IntelligentEvalSessionDB - now = utc_now() # SQLite 读出为 naive datetime，阈值须同为 naive 才能在 Python 侧比较 threshold = now.replace(tzinfo=None) - timedelta(minutes=SESSION_IDLE_EXPIRE_MINUTES) @@ -477,11 +477,6 @@ TRIGGER_COOLDOWN_MINUTES = 10 def _last_decision_at(session: Session, eval_id: str, decision_type: str): """该评估某类决策日志的最近时间（无则 None）。""" - from sqlalchemy import func - from sqlmodel import select - - from agenteval.storage.db import IntelligentEvalDecisionLogDB - return session.exec( select(func.max(IntelligentEvalDecisionLogDB.created_at)).where( IntelligentEvalDecisionLogDB.eval_id == eval_id, @@ -506,11 +501,6 @@ def record_planner_triggers(session: Session) -> int: Returns: 本次实际触发覆盖的评估数（0 表示无需触发 planner）。 """ - from sqlmodel import select - - 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( select(IntelligentEvalDB).where(IntelligentEvalDB.status == IntelligentEvalStatus.PLANNING.value) @@ -544,8 +534,6 @@ def worker_trigger_candidates(session: Session) -> list[str]: def record_worker_triggers(session: Session, eval_ids: list[str]) -> None: """为本次 worker 触发覆盖的评估落账 worker_trigger（供冷却与可见性）。""" - from agenteval.intelligent_eval.decision_logs import append_decision_log, count_decisions - for eval_id in eval_ids: attempt = count_decisions(eval_id, "worker_trigger", session) + 1 append_decision_log( @@ -570,11 +558,6 @@ def enforce_planning_gates(session: Session) -> int: Returns: 被判失败的评估数。 """ - from sqlmodel import select - - from agenteval.intelligent_eval.decision_logs import count_decisions - from agenteval.storage.db import IntelligentEvalDB - now = utc_now().replace(tzinfo=None) planning = session.exec( select(IntelligentEvalDB).where(IntelligentEvalDB.status == IntelligentEvalStatus.PLANNING.value) @@ -624,10 +607,6 @@ def enforce_executing_ceiling(session: Session) -> int: Returns: 被判失败的评估数。 """ - from sqlmodel import select - - from agenteval.storage.db import IntelligentEvalDB - now = utc_now().replace(tzinfo=None) executing = session.exec( select(IntelligentEvalDB).where(IntelligentEvalDB.status == IntelligentEvalStatus.EXECUTING.value) @@ -662,8 +641,6 @@ 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 - from agenteval.intelligent_eval.domain import parse_time_slot - plan = eval_db.get_plan() slots = plan.get("time_distribution") or [] end_hours: list[float] = [] @@ -690,11 +667,6 @@ def evals_needing_analyst_nudge(session: Session) -> list[str]: 末会话终态已满 ANALYST_NUDGE_DELAY_MINUTES、催促次数 < ANALYST_NUDGE_MAX、 距上次催促已满 ANALYST_NUDGE_DELAY_MINUTES（冷却，避免每分钟连发）。 """ - from sqlmodel import select - - 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( select(IntelligentEvalDB).where(IntelligentEvalDB.status == IntelligentEvalStatus.EXECUTING.value) @@ -734,8 +706,6 @@ def evals_needing_analyst_nudge(session: Session) -> list[str]: def record_analyst_nudge(session: Session, eval_id: str) -> None: """落账一次 analyst 催促（analyst_nudge 决策日志，attempt 递增供上限计数）。""" - from agenteval.intelligent_eval.decision_logs import append_decision_log, count_decisions - attempt = count_decisions(eval_id, "analyst_nudge", session) + 1 append_decision_log( eval_id, @@ -756,10 +726,6 @@ TRIGGER_FAILURE_MAX = 3 def eval_ids_with_pending_worker_tasks(session: Session) -> list[str]: """当前有待认领 worker 任务的评估 id（触发失败的受影响方）。""" - from sqlmodel import select - - from agenteval.storage.db import IntelligentEvalTaskQueueDB - rows = session.exec( select(IntelligentEvalTaskQueueDB.eval_id).where(IntelligentEvalTaskQueueDB.status == "pending") ).all() @@ -774,8 +740,6 @@ def record_trigger_failures(session: Session, *, channel: str, eval_ids: list[st Returns: 落账条数。 """ - from agenteval.intelligent_eval.decision_logs import append_decision_log, count_decisions - recorded = 0 for eval_id in eval_ids: attempt = count_decisions(eval_id, "trigger_failed", session) + 1 @@ -801,8 +765,6 @@ def fail_eval(session: Session, eval_id: str, reason: str, decision_type: str, c True 判失败成功；False 表示跳过（评估不存在、非法转换或 CAS 冲突 ——状态已被他人收敛，属正常竞争结局，仅留日志不当故障）。 """ - from agenteval.intelligent_eval.decision_logs import append_decision_log - logger = logging.getLogger("agenteval") repo = IntelligentEvalRepository(session) ev = repo.get(eval_id) @@ -848,15 +810,6 @@ def enforce_trigger_failure_gates(session: Session) -> int: Returns: 被判失败的评估数。 """ - from sqlalchemy import func - from sqlmodel import select - - from agenteval.storage.db import ( - IntelligentEvalDB, - IntelligentEvalDecisionLogDB, - IntelligentEvalTaskQueueDB, - ) - def _failures(eval_id: str, channel: str) -> list: logs = session.exec( select(IntelligentEvalDecisionLogDB).where( diff --git a/backend/agenteval/intelligent_eval/scheduler.py b/backend/agenteval/intelligent_eval/scheduler.py index 8dd39ad..1def1c5 100644 --- a/backend/agenteval/intelligent_eval/scheduler.py +++ b/backend/agenteval/intelligent_eval/scheduler.py @@ -15,7 +15,29 @@ import subprocess import time from typing import Callable, Optional -from agenteval.storage.db import get_session +from sqlmodel import select + +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, + eval_ids_with_pending_worker_tasks, + evals_needing_analyst_nudge, + expire_stale_running_sessions, + record_analyst_nudge, + record_planner_triggers, + record_trigger_failures, + record_worker_triggers, + worker_trigger_candidates, +) +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, get_session _logger = logging.getLogger("agenteval") @@ -93,11 +115,6 @@ async def _trigger_openclaw_agent( 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) @@ -111,12 +128,6 @@ def _record_worker_trigger_failure(error: str) -> None: 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 = [ @@ -147,11 +158,6 @@ async def trigger_worker(nudge_eval_ids: Optional[list[str]] = None) -> bool: 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 [])) @@ -179,8 +185,6 @@ async def trigger_planner() -> bool: """ # 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) @@ -215,19 +219,6 @@ def scan_once() -> None: 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) @@ -262,11 +253,6 @@ def scan_once() -> None: # 催促落账后把评估交给 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) diff --git a/tests/integration/test_intelligent_eval_scan_scheduler.py b/tests/integration/test_intelligent_eval_scan_scheduler.py index 389d7b8..f11cce9 100644 --- a/tests/integration/test_intelligent_eval_scan_scheduler.py +++ b/tests/integration/test_intelligent_eval_scan_scheduler.py @@ -30,7 +30,8 @@ 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(scheduler_mod, "get_session", lambda: MagicMock()) - monkeypatch.setattr(tq, "scan_and_enqueue_tasks", fake_scan) + # Patch the scheduler module's reference, not the task_queue module's + monkeypatch.setattr(scheduler_mod, "scan_and_enqueue_tasks", fake_scan) with TestClient(app_mod.app) as client: assert client.get("/api/health").status_code == 200 diff --git a/tests/unit/test_intelligent_eval_scheduler.py b/tests/unit/test_intelligent_eval_scheduler.py index d79963c..ff7e699 100644 --- a/tests/unit/test_intelligent_eval_scheduler.py +++ b/tests/unit/test_intelligent_eval_scheduler.py @@ -14,11 +14,11 @@ 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) + # Patch the scheduler module's references, not the lifecycle module's + monkeypatch.setattr(scheduler_mod, "worker_trigger_candidates", lambda session: candidates) + monkeypatch.setattr(scheduler_mod, "record_worker_triggers", lambda session, ids: None) monkeypatch.setattr(scheduler_mod, "get_session", lambda: MagicMock()) @@ -93,12 +93,12 @@ def test_trigger_worker_msg_has_execute_semantics(monkeypatch): 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) + # Patch the scheduler module's reference, not the lifecycle module's + monkeypatch.setattr(scheduler_mod, "record_planner_triggers", lambda session: 0) monkeypatch.setattr(scheduler_mod, "get_session", lambda: MagicMock()) def fake_run(cmd, **kwargs): @@ -113,11 +113,11 @@ def test_trigger_planner_skips_when_no_planning(monkeypatch): 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) + # Patch the scheduler module's reference, not the lifecycle module's + monkeypatch.setattr(scheduler_mod, "record_planner_triggers", lambda session: 1) monkeypatch.setattr(scheduler_mod, "get_session", lambda: MagicMock()) def fake_run(cmd, **kwargs): @@ -140,25 +140,25 @@ def test_trigger_planner_calls_docker_exec(monkeypatch): 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()) + # Patch the scheduler module's references for task_queue functions 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) + monkeypatch.setattr(scheduler_mod, name, lambda session, _n=name: calls.append(_n) or 0) + # Patch the scheduler module's references for lifecycle functions 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 []) + monkeypatch.setattr(scheduler_mod, name, lambda session, _n=name: calls.append(_n) or 0) + # Patch the scheduler module's reference for decision_logs function + monkeypatch.setattr(scheduler_mod, "supplement_decision_logs", lambda session: calls.append("supplement") or 0) + monkeypatch.setattr(scheduler_mod, "evals_needing_analyst_nudge", lambda session: calls.append("nudge") or []) fired: list[str] = []