7eae6de52d
合并两个不可分割的深化： Phase 2 — 智能作业结算统一（ADR-0012） - intelligence_jobs.execute(job_kind, campaign_id, ...) 作为结算的 唯一实现：建行 → 认领 → 校验 → generating → 落账，一处编排、 一处截断（500 字符）。两个 executor 退化为 ensure_queued / validate / work_fn 三个小 adapter。 - analysis.validate_analysis_request() 共享校验入口（活动终态 → 模型），路由捕获映射 400、executor 捕获落 failed 行，与 validate_comparison_request 先例同构。 - campaign_runner._auto_start_analysis 的跳过守卫收敛至 auto_intelligence_eligible 单一判断点。 - comparison.py 删除零调用的 build_comparison_payload； load_comparison_view 投影归位至 campaign_read_model。 - 新增 characterization 测试（认领竞争、重复触发、截断、恢复上限）。 Phase 3 — storage/repository.py 拆分 - AsyncJobRepository 及两个子类迁至 storage/async_job_repository.py（Phase 2 的 intelligence_jobs 与 comparison 必须 import 自该路径，故与 Phase 2 同 commit）。 - ExplorationSession / ExplorationMessage 迁至 storage/exploration_repository.py；repository.py 由 1180 行降至 约 814 行，grep 确认无残留符号。 - exploration 子模块与路由 import 全部更新；测试 import 跟随。 刻意不做：CAS 共享原语、app.py 五 registry 关停顺序归一 （ADR-0006 精神，等真实需求出现再议）。
219 lines
8.8 KiB
Python
219 lines
8.8 KiB
Python
"""探索生命周期领域模块直测（架构保养候选 3）。
|
||
|
||
账本规则与状态机从 HTTP 层落入 exploration/lifecycle.py 与 patrol.py，
|
||
用类型化领域异常（NotFound / Guardrail / Channel）表达违规，
|
||
脱离 TestClient 即可单测。
|
||
"""
|
||
|
||
from datetime import datetime, timedelta, timezone
|
||
from uuid import uuid4
|
||
|
||
import pytest
|
||
from agenteval.exploration import lifecycle, patrol
|
||
from agenteval.exploration.errors import ExplorationGuardrailError, ExplorationNotFoundError
|
||
from agenteval.exploration.models import ExplorationSessionStatus, ExplorationTrigger
|
||
from agenteval.models import (
|
||
Campaign,
|
||
CampaignPlanEntry,
|
||
CampaignStatus,
|
||
EvalRun,
|
||
EvalTarget,
|
||
ExplorationBudgetConfig,
|
||
ExplorationSeeds,
|
||
RunStatus,
|
||
)
|
||
from agenteval.storage.exploration_repository import ExplorationSessionRepository
|
||
from agenteval.storage.repository import (
|
||
CampaignRepository,
|
||
RunRepository,
|
||
TargetRepository,
|
||
)
|
||
from sqlmodel import Session, SQLModel, create_engine
|
||
|
||
T0 = datetime(2026, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
|
||
|
||
|
||
@pytest.fixture()
|
||
def db_session(tmp_path):
|
||
from agenteval.storage.db import ( # noqa: F401
|
||
CampaignDB,
|
||
EvalResultDB,
|
||
EvalRunDB,
|
||
EvalTargetDB,
|
||
ExplorationMessageDB,
|
||
ExplorationSessionDB,
|
||
FileCategoryDB,
|
||
FileRecordDB,
|
||
ScenarioDB,
|
||
TurnDB,
|
||
)
|
||
engine = create_engine(
|
||
f"sqlite:///{tmp_path / 'lifecycle.db'}",
|
||
connect_args={"check_same_thread": False},
|
||
)
|
||
SQLModel.metadata.create_all(engine)
|
||
session = Session(engine)
|
||
try:
|
||
yield session
|
||
finally:
|
||
session.close()
|
||
engine.dispose()
|
||
|
||
|
||
def _seed_campaign(
|
||
db_session,
|
||
*,
|
||
status=CampaignStatus.RUNNING,
|
||
time_scale=1.0,
|
||
budget=None,
|
||
seeds=None,
|
||
) -> Campaign:
|
||
target = TargetRepository(db_session).create(EvalTarget(id=f"t-{uuid4().hex[:8]}", name="数字员工"))
|
||
campaign = CampaignRepository(db_session).create(Campaign(
|
||
name="cycle", target_id=target.id, window_seconds=3600, time_scale=time_scale,
|
||
plan=[CampaignPlanEntry(scenario_id="s-a", offset_seconds=0, count=1)],
|
||
status=status, started_at=T0,
|
||
exploration_budget=budget, exploration_seeds=seeds,
|
||
))
|
||
return campaign
|
||
|
||
|
||
def _open(db_session, campaign, triggered_by=ExplorationTrigger.AUTO, goal="查询账单"):
|
||
return lifecycle.open_session(
|
||
db_session,
|
||
campaign_id=campaign.id,
|
||
persona={"name": "急性子用户"},
|
||
goal=goal,
|
||
triggered_by=triggered_by,
|
||
)
|
||
|
||
|
||
# ── 账本四规则 ────────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_open_rejects_non_running_campaign(db_session):
|
||
campaign = _seed_campaign(db_session, status=CampaignStatus.PLANNED)
|
||
with pytest.raises(ExplorationGuardrailError, match="活动不在进行中"):
|
||
_open(db_session, campaign)
|
||
|
||
|
||
def test_open_rejects_auto_on_accelerated_line_but_allows_manual(db_session):
|
||
campaign = _seed_campaign(db_session, time_scale=10.0)
|
||
with pytest.raises(ExplorationGuardrailError, match="加速调试线仅允许手动"):
|
||
_open(db_session, campaign)
|
||
session_obj = _open(db_session, campaign, triggered_by=ExplorationTrigger.MANUAL)
|
||
assert session_obj.status == ExplorationSessionStatus.RUNNING
|
||
|
||
|
||
def test_open_enforces_max_sessions_budget(db_session):
|
||
campaign = _seed_campaign(db_session, budget=ExplorationBudgetConfig(max_sessions=1))
|
||
_open(db_session, campaign, triggered_by=ExplorationTrigger.MANUAL)
|
||
with pytest.raises(ExplorationGuardrailError, match="探索会话数超出预算"):
|
||
_open(db_session, campaign, triggered_by=ExplorationTrigger.MANUAL)
|
||
|
||
|
||
def test_open_enforces_min_interval(db_session):
|
||
campaign = _seed_campaign(db_session) # 默认间隔 30 分钟
|
||
_open(db_session, campaign)
|
||
with pytest.raises(ExplorationGuardrailError, match="相邻探索会话间隔不足"):
|
||
_open(db_session, campaign)
|
||
|
||
|
||
def test_open_missing_campaign_or_target_raises_not_found(db_session):
|
||
with pytest.raises(ExplorationNotFoundError):
|
||
lifecycle.open_session(
|
||
db_session, campaign_id="missing", persona={}, goal="x",
|
||
)
|
||
campaign = _seed_campaign(db_session)
|
||
campaign.target_id = "missing-target"
|
||
CampaignRepository(db_session).update(campaign)
|
||
with pytest.raises(ExplorationNotFoundError):
|
||
lifecycle.open_session(
|
||
db_session, campaign_id=campaign.id, persona={}, goal="x",
|
||
)
|
||
|
||
|
||
# ── 关闭状态机 ────────────────────────────────────────────────────────────
|
||
|
||
|
||
def test_close_completes_session_and_normalizes_experience(db_session, monkeypatch):
|
||
judged: list[str] = []
|
||
monkeypatch.setattr(lifecycle, "start_judge_review", lambda sid: judged.append(sid))
|
||
campaign = _seed_campaign(db_session)
|
||
session_obj = _open(db_session, campaign)
|
||
|
||
updated = lifecycle.close_session(
|
||
db_session, session_id=session_obj.id,
|
||
experience={"goal_achieved": True, "blockers": ["入口难找"], "emotion": "weird"},
|
||
)
|
||
assert updated.status == ExplorationSessionStatus.COMPLETED
|
||
assert updated.closed_at is not None
|
||
assert updated.experience["emotion"] == "neutral" # 非法情绪归一
|
||
assert updated.experience["blockers"] == ["入口难找"]
|
||
assert judged == [session_obj.id]
|
||
|
||
|
||
def test_close_rejects_non_running_session(db_session, monkeypatch):
|
||
monkeypatch.setattr(lifecycle, "start_judge_review", lambda sid: None)
|
||
campaign = _seed_campaign(db_session)
|
||
session_obj = _open(db_session, campaign)
|
||
lifecycle.close_session(db_session, session_id=session_obj.id, experience={"goal_achieved": False})
|
||
with pytest.raises(ExplorationGuardrailError, match="不在进行中"):
|
||
lifecycle.close_session(db_session, session_id=session_obj.id, experience={"goal_achieved": False})
|
||
|
||
|
||
def test_close_missing_session_raises_not_found(db_session):
|
||
with pytest.raises(ExplorationNotFoundError):
|
||
lifecycle.close_session(db_session, session_id="missing", experience={})
|
||
|
||
|
||
# ── 对话轮账本（拒收路径；通道往返由集成测试覆盖）────────────────────────
|
||
|
||
|
||
async def test_turn_rejects_non_running_session(db_session):
|
||
campaign = _seed_campaign(db_session)
|
||
session_obj = _open(db_session, campaign)
|
||
lifecycle.close_session(db_session, session_id=session_obj.id, experience={"goal_achieved": True})
|
||
with pytest.raises(ExplorationGuardrailError, match="拒收消息"):
|
||
await lifecycle.conduct_turn(db_session, session_id=session_obj.id, content="你好")
|
||
|
||
|
||
async def test_turn_enforces_turn_budget(db_session):
|
||
campaign = _seed_campaign(db_session, budget=ExplorationBudgetConfig(max_turns=1, min_interval_seconds=1))
|
||
session_obj = _open(db_session, campaign)
|
||
repo = ExplorationSessionRepository(db_session)
|
||
session_obj.turn_count = 1
|
||
repo.update(session_obj)
|
||
with pytest.raises(ExplorationGuardrailError, match="会话轮数超出预算"):
|
||
await lifecycle.conduct_turn(db_session, session_id=session_obj.id, content="你好")
|
||
|
||
|
||
async def test_turn_missing_session_raises_not_found(db_session):
|
||
with pytest.raises(ExplorationNotFoundError):
|
||
await lifecycle.conduct_turn(db_session, session_id="missing", content="你好")
|
||
|
||
|
||
# ── 巡检：筛选 + 增量 + 水位 ──────────────────────────────────────────────
|
||
|
||
|
||
def test_patrol_filters_and_advances_watermark(db_session):
|
||
eligible = _seed_campaign(db_session, seeds=ExplorationSeeds(personas=["急性子用户"], goals=["查账单"]))
|
||
# 不符合巡检条件：加速线 / 无种子集
|
||
_seed_campaign(db_session, time_scale=10.0, seeds=ExplorationSeeds(personas=["慢用户"], goals=["y"]))
|
||
RunRepository(db_session).create(EvalRun(
|
||
target_id="t-1", scenario_id="s-a", campaign_id=eligible.id,
|
||
status=RunStatus.COMPLETED, started_at=T0, completed_at=T0 + timedelta(seconds=10),
|
||
summary={"total_cases": 1, "passed_cases": 1, "pass_rate": 1.0, "avg_latency_ms": 100},
|
||
))
|
||
|
||
first = patrol.patrol_report(db_session)
|
||
assert [e["campaign_id"] for e in first["campaigns"]] == [eligible.id]
|
||
assert first["campaigns"][0]["new_results"] is not None
|
||
|
||
# 水位推进后无增量；且只动水位列（状态不被覆写）
|
||
second = patrol.patrol_report(db_session)
|
||
assert second["campaigns"][0]["new_results"] is None
|
||
fresh = CampaignRepository(db_session).get(eligible.id)
|
||
assert fresh.last_patrolled_at is not None
|
||
assert fresh.status == CampaignStatus.RUNNING
|