From 2a0bcdd1857acb33588f9f8e95c0262a3276a857 Mon Sep 17 00:00:00 2001 From: sinohqb Date: Fri, 21 Aug 2026 03:50:44 +0800 Subject: [PATCH] refactor(intelligent-eval): drop ADR-0008 transitional wrappers, callers use domain MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 架构审查候选④：ADR-0008 收敛调度域时为保测试兼容留下的过渡 wrapper 使命结束。 删除 8 个浅封装：task_queue 的 _is_slot_due / _calculate_session_deficit （零调用死函数）+ _calculate_priority / _get_attention_reason，decision 的 _parse_time_slot（零调用死函数）+ _get_current_slot / _count_sessions_in_slot / _has_high_severity_issues。调用方直接使用 domain 模块。 5 个隔着 wrapper 测 domain 行为的测试迁到新文件 test_intelligent_eval_domain.py，直接锁定 domain，覆盖零丢失。 删除测试通过：复杂度直接消失，时段/欠账/优先级知识只剩 domain 一处。 870 tests passed，零行为变化。 --- .../agenteval/intelligent_eval/decision.py | 40 ++---- .../agenteval/intelligent_eval/task_queue.py | 34 +---- tests/unit/test_intelligent_eval_domain.py | 134 ++++++++++++++++++ .../unit/test_intelligent_eval_task_queue.py | 130 +---------------- 4 files changed, 152 insertions(+), 186 deletions(-) create mode 100644 tests/unit/test_intelligent_eval_domain.py diff --git a/backend/agenteval/intelligent_eval/decision.py b/backend/agenteval/intelligent_eval/decision.py index 4611b34..e9b0a35 100644 --- a/backend/agenteval/intelligent_eval/decision.py +++ b/backend/agenteval/intelligent_eval/decision.py @@ -6,9 +6,7 @@ Worker analyzes current situation and decides what to do: - start_analysis: Start analysis (all sessions completed) """ -from datetime import datetime, timedelta from enum import Enum -from typing import Optional from sqlmodel import Session, select @@ -45,32 +43,6 @@ class Decision: } -def _parse_time_slot(time_slot: str) -> Optional[tuple[float, float]]: - """Thin wrapper delegating to :func:`agenteval.intelligent_eval.domain.parse_time_slot`.""" - from agenteval.intelligent_eval.domain import parse_time_slot as _impl - return _impl(time_slot) - - -def _get_current_slot(time_distribution: list[dict], current_offset: timedelta) -> Optional[dict]: - """Thin wrapper delegating to :func:`agenteval.intelligent_eval.domain.get_current_slot`.""" - from agenteval.intelligent_eval.domain import get_current_slot as _impl - return _impl(time_distribution, current_offset) - - -def _count_sessions_in_slot( - eval_id: str, slot: dict, eval_started_at: datetime, session: Session -) -> int: - """Thin wrapper delegating to :func:`agenteval.intelligent_eval.domain.count_sessions_in_slot`.""" - from agenteval.intelligent_eval.domain import count_sessions_in_slot as _impl - return _impl(eval_id, slot, eval_started_at, session) - - -def _has_high_severity_issues(eval_id: str, session: Session) -> bool: - """Thin wrapper delegating to :func:`agenteval.intelligent_eval.domain.has_high_severity_issues`.""" - from agenteval.intelligent_eval.domain import has_high_severity_issues as _impl - return _impl(eval_id, session) - - def make_decision(eval_db: IntelligentEvalDB, session: Session) -> Decision: """Make a decision based on current evaluation state. @@ -81,6 +53,12 @@ def make_decision(eval_db: IntelligentEvalDB, session: Session) -> Decision: Returns: Decision object """ + from agenteval.intelligent_eval.domain import ( + count_sessions_in_slot, + get_current_slot, + has_high_severity_issues, + ) + # Check if eval is still executing if eval_db.status != IntelligentEvalStatus.EXECUTING.value: return Decision( @@ -124,7 +102,7 @@ def make_decision(eval_db: IntelligentEvalDB, session: Session) -> Decision: ) # Get current time slot - current_slot = _get_current_slot(time_distribution, current_offset) + current_slot = get_current_slot(time_distribution, current_offset) if current_slot is None: return Decision( @@ -136,7 +114,7 @@ def make_decision(eval_db: IntelligentEvalDB, session: Session) -> Decision: # Check if current slot has deficit slot_name = current_slot.get("time_slot", "") expected_sessions = current_slot.get("sessions", 0) - current_sessions = _count_sessions_in_slot(eval_db.id, current_slot, eval_db.started_at, session) + current_sessions = count_sessions_in_slot(eval_db.id, current_slot, eval_db.started_at, session) deficit = expected_sessions - current_sessions @@ -153,7 +131,7 @@ def make_decision(eval_db: IntelligentEvalDB, session: Session) -> Decision: ) # Check if any high severity issues found - if _has_high_severity_issues(eval_db.id, session): + if has_high_severity_issues(eval_db.id, session): return Decision( DecisionType.EXECUTE_SESSION, "发现高严重度问题，需要深入挖掘", diff --git a/backend/agenteval/intelligent_eval/task_queue.py b/backend/agenteval/intelligent_eval/task_queue.py index b6ffb57..3b834fd 100644 --- a/backend/agenteval/intelligent_eval/task_queue.py +++ b/backend/agenteval/intelligent_eval/task_queue.py @@ -20,34 +20,6 @@ from agenteval.storage.db import ( ) -def _is_slot_due(slot: dict, current_offset: timedelta) -> bool: - """Thin wrapper delegating to :func:`agenteval.intelligent_eval.domain.is_slot_due`.""" - from agenteval.intelligent_eval.domain import is_slot_due as _impl - - return _impl(slot, current_offset) - - -def _calculate_session_deficit(eval_db: IntelligentEvalDB, session: Session) -> int: - """Thin wrapper delegating to :func:`agenteval.intelligent_eval.domain.calculate_session_deficit`.""" - from agenteval.intelligent_eval.domain import calculate_session_deficit as _impl - - return _impl(eval_db, session) - - -def _calculate_priority(eval_db: IntelligentEvalDB, session: Session) -> int: - """Thin wrapper delegating to :func:`agenteval.intelligent_eval.domain.calculate_priority`.""" - from agenteval.intelligent_eval.domain import calculate_priority as _impl - - return _impl(eval_db, session) - - -def _get_attention_reason(eval_db: IntelligentEvalDB, session: Session) -> Optional[str]: - """Thin wrapper delegating to :func:`agenteval.intelligent_eval.domain.get_attention_reason`.""" - from agenteval.intelligent_eval.domain import get_attention_reason as _impl - - return _impl(eval_db, session) - - def _has_pending_task(eval_id: str, session: Session) -> bool: """Check if eval already has a pending/assigned task (去重).""" existing = session.exec( @@ -65,6 +37,8 @@ def scan_and_enqueue_tasks(session: Session) -> int: Returns: Number of tasks enqueued """ + from agenteval.intelligent_eval.domain import calculate_priority, get_attention_reason + # Get all executing evals evals = session.exec( select(IntelligentEvalDB).where(IntelligentEvalDB.status == IntelligentEvalStatus.EXECUTING.value) @@ -74,7 +48,7 @@ def scan_and_enqueue_tasks(session: Session) -> int: for eval_db in evals: # Check if eval needs attention - reason = _get_attention_reason(eval_db, session) + reason = get_attention_reason(eval_db, session) if reason is None: continue @@ -83,7 +57,7 @@ def scan_and_enqueue_tasks(session: Session) -> int: continue # Calculate priority - priority = _calculate_priority(eval_db, session) + priority = calculate_priority(eval_db, session) # Create task task = IntelligentEvalTaskQueueDB( diff --git a/tests/unit/test_intelligent_eval_domain.py b/tests/unit/test_intelligent_eval_domain.py new file mode 100644 index 0000000..4d12176 --- /dev/null +++ b/tests/unit/test_intelligent_eval_domain.py @@ -0,0 +1,134 @@ +"""Unit tests for the intelligent-eval scheduling domain (domain.py). + +从 test_intelligent_eval_task_queue.py 迁移：原先隔着 task_queue 的过渡 +wrapper 测试，ADR-0008 残留 wrapper 删除后直接锁定 domain 模块—— +时段解析、欠账、优先级、注意原因的唯一实现。 +""" + +from datetime import timedelta + +from agenteval.intelligent_eval import domain +from agenteval.intelligent_eval.models import IntelligentEvalStatus +from agenteval.storage.db import IntelligentEvalDB, IntelligentEvalSessionDB, utc_now +from sqlmodel import Session + + +def test_is_slot_due(): + """Test time slot due detection.""" + # Slot "8-10h" should be due after 8 hours + slot = {"time_slot": "8-10h", "sessions": 2} + assert domain.is_slot_due(slot, timedelta(hours=7)) is False + assert domain.is_slot_due(slot, timedelta(hours=8)) is True + assert domain.is_slot_due(slot, timedelta(hours=9)) is True + + # Invalid slot format + assert domain.is_slot_due({"time_slot": "invalid"}, timedelta(hours=1)) is False + assert domain.is_slot_due({}, timedelta(hours=1)) is False + + +def test_is_slot_due_minute_format(): + """Minute-level slots (1h window: "0-20min") must be parsed and due on time.""" + # 0-20min: due at 0min, not before + assert domain.is_slot_due({"time_slot": "0-20min"}, timedelta(minutes=-1)) is False + assert domain.is_slot_due({"time_slot": "0-20min"}, timedelta(minutes=0)) is True + assert domain.is_slot_due({"time_slot": "0-20min"}, timedelta(minutes=5)) is True + # 20-40min: not due before 20min, due at 20min + assert domain.is_slot_due({"time_slot": "20-40min"}, timedelta(minutes=19)) is False + assert domain.is_slot_due({"time_slot": "20-40min"}, timedelta(minutes=20)) is True + + # parse both formats consistently (hours) + assert domain.parse_time_slot("8-10h") == (8.0, 10.0) + assert domain.parse_time_slot("0-20min") == (0.0, 1.0 / 3) + assert domain.parse_time_slot("20-40min") == (1.0 / 3, 2.0 / 3) + assert domain.parse_time_slot("bad") is None + + +def test_calculate_session_deficit(db_session: Session): + """Test session deficit calculation.""" + # Create eval with plan + eval_db = IntelligentEvalDB( + name="test", + target_id="target1", + status=IntelligentEvalStatus.EXECUTING.value, + started_at=utc_now() - timedelta(hours=9), + ) + eval_db.set_plan({ + "time_distribution": [ + {"time_slot": "0-2h", "sessions": 1}, + {"time_slot": "8-10h", "sessions": 2}, + ], + "estimated_sessions": 3, + }) + db_session.add(eval_db) + db_session.commit() + + # No sessions yet, should have 3 (1 from 0-2h, 2 from 8-10h) + deficit = domain.calculate_session_deficit(eval_db, db_session) + assert deficit == 3 + + # Add 1 session + session_db = IntelligentEvalSessionDB( + eval_id=eval_db.id, + target_id="target1", + status="completed", + ) + db_session.add(session_db) + db_session.commit() + + # Should have 3, has 1, deficit = 2 + deficit = domain.calculate_session_deficit(eval_db, db_session) + assert deficit == 2 + + +def test_calculate_priority(db_session: Session): + """Test task priority calculation.""" + # Eval with due slot and deficit + eval_db = IntelligentEvalDB( + name="test", + target_id="target1", + status=IntelligentEvalStatus.EXECUTING.value, + started_at=utc_now() - timedelta(hours=9), + ) + eval_db.set_plan({ + "time_distribution": [{"time_slot": "8-10h", "sessions": 2}], + "estimated_sessions": 2, + }) + db_session.add(eval_db) + db_session.commit() + + priority = domain.calculate_priority(eval_db, db_session) + # Base 100 - 50 (slot due) - 20 (deficit 2 * 10) - 20 (wait 9h / 10min = 54, capped at 20) + assert priority == 10 + + +def test_get_attention_reason(db_session: Session): + """Test attention reason detection.""" + # Eval with due slot + eval_db = IntelligentEvalDB( + name="test", + target_id="target1", + status=IntelligentEvalStatus.EXECUTING.value, + started_at=utc_now() - timedelta(hours=9), + ) + eval_db.set_plan({ + "time_distribution": [{"time_slot": "8-10h", "sessions": 2}], + "estimated_sessions": 2, + }) + db_session.add(eval_db) + db_session.commit() + + reason = domain.get_attention_reason(eval_db, db_session) + assert reason == "slot_due" + + # Add all sessions as completed + for _ in range(2): + session_db = IntelligentEvalSessionDB( + eval_id=eval_db.id, + target_id="target1", + status="completed", + ) + db_session.add(session_db) + db_session.commit() + + reason = domain.get_attention_reason(eval_db, db_session) + assert reason == "all_sessions_completed" diff --git a/tests/unit/test_intelligent_eval_task_queue.py b/tests/unit/test_intelligent_eval_task_queue.py index 69277a6..9b968bf 100644 --- a/tests/unit/test_intelligent_eval_task_queue.py +++ b/tests/unit/test_intelligent_eval_task_queue.py @@ -1,4 +1,8 @@ -"""Unit tests for intelligent eval task queue.""" +"""Unit tests for intelligent eval task queue. + +时段/欠账/优先级等调度域的测试已随 ADR-0008 过渡 wrapper 删除迁移到 +test_intelligent_eval_domain.py，直接锁定 domain 模块。 +""" from datetime import timedelta @@ -6,136 +10,12 @@ from agenteval.intelligent_eval import task_queue from agenteval.intelligent_eval.models import IntelligentEvalStatus from agenteval.storage.db import ( IntelligentEvalDB, - IntelligentEvalSessionDB, IntelligentEvalTaskQueueDB, utc_now, ) from sqlmodel import Session, select -def test_is_slot_due(): - """Test time slot due detection.""" - # Slot "8-10h" should be due after 8 hours - slot = {"time_slot": "8-10h", "sessions": 2} - assert task_queue._is_slot_due(slot, timedelta(hours=7)) is False - assert task_queue._is_slot_due(slot, timedelta(hours=8)) is True - assert task_queue._is_slot_due(slot, timedelta(hours=9)) is True - - # Invalid slot format - assert task_queue._is_slot_due({"time_slot": "invalid"}, timedelta(hours=1)) is False - assert task_queue._is_slot_due({}, timedelta(hours=1)) is False - - -def test_is_slot_due_minute_format(): - """Minute-level slots (1h window: "0-20min") must be parsed and due on time.""" - from agenteval.intelligent_eval import domain - - # 0-20min: due at 0min, not before - assert task_queue._is_slot_due({"time_slot": "0-20min"}, timedelta(minutes=-1)) is False - assert task_queue._is_slot_due({"time_slot": "0-20min"}, timedelta(minutes=0)) is True - assert task_queue._is_slot_due({"time_slot": "0-20min"}, timedelta(minutes=5)) is True - # 20-40min: not due before 20min, due at 20min - assert task_queue._is_slot_due({"time_slot": "20-40min"}, timedelta(minutes=19)) is False - assert task_queue._is_slot_due({"time_slot": "20-40min"}, timedelta(minutes=20)) is True - - # parse both formats consistently (hours) - assert domain.parse_time_slot("8-10h") == (8.0, 10.0) - assert domain.parse_time_slot("0-20min") == (0.0, 1.0 / 3) - assert domain.parse_time_slot("20-40min") == (1.0 / 3, 2.0 / 3) - assert domain.parse_time_slot("bad") is None - - -def test_calculate_session_deficit(db_session: Session): - """Test session deficit calculation.""" - # Create eval with plan - eval_db = IntelligentEvalDB( - name="test", - target_id="target1", - status=IntelligentEvalStatus.EXECUTING.value, - started_at=utc_now() - timedelta(hours=9), - ) - eval_db.set_plan({ - "time_distribution": [ - {"time_slot": "0-2h", "sessions": 1}, - {"time_slot": "8-10h", "sessions": 2}, - ], - "estimated_sessions": 3, - }) - db_session.add(eval_db) - db_session.commit() - - # No sessions yet, should have 3 (1 from 0-2h, 2 from 8-10h) - deficit = task_queue._calculate_session_deficit(eval_db, db_session) - assert deficit == 3 - - # Add 1 session - session_db = IntelligentEvalSessionDB( - eval_id=eval_db.id, - target_id="target1", - status="completed", - ) - db_session.add(session_db) - db_session.commit() - - # Should have 3, has 1, deficit = 2 - deficit = task_queue._calculate_session_deficit(eval_db, db_session) - assert deficit == 2 - - -def test_calculate_priority(db_session: Session): - """Test task priority calculation.""" - # Eval with due slot and deficit - eval_db = IntelligentEvalDB( - name="test", - target_id="target1", - status=IntelligentEvalStatus.EXECUTING.value, - started_at=utc_now() - timedelta(hours=9), - ) - eval_db.set_plan({ - "time_distribution": [{"time_slot": "8-10h", "sessions": 2}], - "estimated_sessions": 2, - }) - db_session.add(eval_db) - db_session.commit() - - priority = task_queue._calculate_priority(eval_db, db_session) - # Base 100 - 50 (slot due) - 20 (deficit 2 * 10) - 20 (wait 9h / 10min = 54, capped at 20) - assert priority == 10 - - -def test_get_attention_reason(db_session: Session): - """Test attention reason detection.""" - # Eval with due slot - eval_db = IntelligentEvalDB( - name="test", - target_id="target1", - status=IntelligentEvalStatus.EXECUTING.value, - started_at=utc_now() - timedelta(hours=9), - ) - eval_db.set_plan({ - "time_distribution": [{"time_slot": "8-10h", "sessions": 2}], - "estimated_sessions": 2, - }) - db_session.add(eval_db) - db_session.commit() - - reason = task_queue._get_attention_reason(eval_db, db_session) - assert reason == "slot_due" - - # Add all sessions as completed - for _ in range(2): - session_db = IntelligentEvalSessionDB( - eval_id=eval_db.id, - target_id="target1", - status="completed", - ) - db_session.add(session_db) - db_session.commit() - - reason = task_queue._get_attention_reason(eval_db, db_session) - assert reason == "all_sessions_completed" - - def test_has_pending_task(db_session: Session): """Test pending task detection (去重).""" eval_db = IntelligentEvalDB(