"""探索生命周期领域模块直测（架构保养候选 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