backend/agenteval/evaluation/analysis.py

@@ -13,7 +13,7 @@ from typing import Any, Awaitable, Callable, Optional
 
 from sqlmodel import Session
 
-from agenteval.evaluation.report import load_campaign_report
+from agenteval.evaluation.report import generate_campaign_report
 from agenteval.exploration.summary import summarize_campaign_exploration
 from agenteval.model_gateway import ModelGateway
 from agenteval.models import Campaign, ModelCapability, RunStatus
@@ -28,6 +28,7 @@ from agenteval.storage.repository import (
     CampaignAnalysisRepository,
     CampaignRepository,
     RunRepository,
+    ScenarioRepository,
 )
 from agenteval.utils.llm import extract_reply_text, parse_json_from_llm_text
 
@@ -305,7 +306,8 @@ async def execute_campaign_analysis(
         try:
             client = chat_client or gateway_chat_client(runtime)
             runs = RunRepository(session).list_by_campaign(campaign_id)
-            report = load_campaign_report(session, campaign)
+            scenario_names = {s.id: s.name for s in ScenarioRepository(session).list_all()}
+            report = generate_campaign_report(campaign, runs, scenario_names=scenario_names)
             result = await analyze_campaign(
                 campaign=campaign,
                 report=report,

backend/agenteval/evaluation/campaign_runner.py

@@ -180,7 +180,7 @@ async def advance_campaign(
         summary = campaign.summary or CampaignSummary()
         summary.scheduler = SchedulerState(spawned_indices=sorted(spawned), errors=errors)
         campaign.summary = summary
-        repo.save_scheduler_state(campaign_id, summary)
+        repo.update(campaign)
 
     return result
 

backend/agenteval/evaluation/comparison.py

@@ -15,13 +15,15 @@ from typing import Any, Optional
 from sqlmodel import Session
 
 from agenteval.evaluation.analysis import ChatClient, gateway_chat_client, resolve_analysis_model
-from agenteval.evaluation.report import load_campaign_report
+from agenteval.evaluation.report import generate_campaign_report
-from agenteval.models import Campaign
+from agenteval.models import Campaign, EvalRun
 from agenteval.storage.db import get_session, iso_utc, utc_now
 from agenteval.storage.repository import (
     CampaignAnalysisRepository,
     CampaignPeriodComparisonRepository,
     CampaignRepository,
+    RunRepository,
+    ScenarioRepository,
 )
 from agenteval.utils.llm import parse_json_from_llm_text
 
@@ -152,14 +154,21 @@ def compute_metric_diff(
     return {"overall": overall, "scenarios": scenarios}
 
 
+def build_campaign_report_dict(campaign: Campaign, session: Session) -> dict[str, Any]:
+    """为周期对比现算一期活动的报告 dict（读路径，不重算聚合口径）。"""
+    runs: list[EvalRun] = RunRepository(session).list_by_campaign(campaign.id)
+    scenario_names = {s.id: s.name for s in ScenarioRepository(session).list_all()}
+    return generate_campaign_report(campaign, runs, scenario_names=scenario_names)
+
+
 def build_comparison_payload(campaign: Campaign, session: Session) -> dict[str, Any]:
     """GET 返回体：自动基线信息 + 机械 diff（无基线时两者均为 null）。"""
     baseline = resolve_auto_baseline(campaign, session)
     if baseline is None:
         return {"auto_baseline": None, "metric_diff": None}
     diff = compute_metric_diff(
-        load_campaign_report(session, baseline),
+        build_campaign_report_dict(baseline, session),
-        load_campaign_report(session, campaign),
+        build_campaign_report_dict(campaign, session),
     )
     return {
         "auto_baseline": {
@@ -349,8 +358,8 @@ async def execute_campaign_comparison(
         )
         try:
             diff = compute_metric_diff(
-                load_campaign_report(session, baseline),
+                build_campaign_report_dict(baseline, session),
-                load_campaign_report(session, campaign),
+                build_campaign_report_dict(campaign, session),
             )
             valid_scenario_ids = {s["scenario_id"] for s in diff["scenarios"]}
             result = await narrate_period_comparison(

backend/agenteval/evaluation/report.py

@@ -9,8 +9,6 @@ from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Optional
 
-from sqlmodel import Session
-
 from agenteval.evaluation.case_verdict import CaseEvidence, resolve_case_verdicts
 from agenteval.evaluation.metrics import aggregate_runs
 from agenteval.evaluation.report_render import render_html, render_json, render_markdown
@@ -361,12 +359,6 @@ def generate_campaign_report(
     }
 
 
-def load_campaign_report(session: Session, campaign: Campaign) -> dict[str, Any]:
-    """取数 + 聚合一步完成：报告 / 分析 / 对比 / 导出共用的活动报告 dict 取法。"""
-    runs = RunRepository(session).list_by_campaign(campaign.id)
-    return generate_campaign_report(campaign, runs, scenario_names=ScenarioRepository(session).name_map())
-
-
 def save_report(run_id: str, fmt: str = "html", output_dir: Optional[Path] = None) -> Path:
     """Generate a run report and save it to disk in the requested format."""
     output_dir = output_dir or DATA_DIR / "reports"

backend/agenteval/exploration/errors.py

@@ -1,26 +0,0 @@
-"""Domain errors for exploratory evaluation (不依赖 fastapi).
-
-领域违规用类型化异常表达，router 统一映射 HTTP 状态码：
-NotFound → 404，Guardrail → 409（reason 即给常驻代理的反馈），
-Channel → 502。
-"""
-
-
-class ExplorationNotFoundError(Exception):
-    """探索域实体（活动 / 评测对象 / 会话）不存在。"""
-
-
-class ExplorationGuardrailError(Exception):
-    """平台账本规则违规（状态机 / 预算），拒绝本身即反馈。"""
-
-    def __init__(self, reason: str):
-        self.reason = reason
-        super().__init__(reason)
-
-
-class ExplorationChannelError(Exception):
-    """评测对象通道收发失败。"""
-
-    def __init__(self, reason: str):
-        self.reason = reason
-        super().__init__(reason)

backend/agenteval/exploration/lifecycle.py

@@ -1,187 +0,0 @@
-"""Exploration session lifecycle (探索会话生命周期).
-
-会话三态的领域操作：open_session / conduct_turn / close_session。
-平台账本（预算、状态机、轮次记账）全部在此执行，违规抛类型化领域
-异常（errors.py），由 router 映射 HTTP 状态码；通道收发沿用
-ChannelFactory 既有接缝。
-"""
-
-import json
-from typing import Any, Optional
-
-from sqlmodel import Session
-
-from agenteval.channels.factory import ChannelFactory
-from agenteval.config import get_settings
-from agenteval.exploration.errors import (
-    ExplorationChannelError,
-    ExplorationGuardrailError,
-    ExplorationNotFoundError,
-)
-from agenteval.exploration.judge import start_judge_review
-from agenteval.exploration.models import (
-    ExplorationBudget,
-    ExplorationMessage,
-    ExplorationSession,
-    ExplorationSessionStatus,
-    ExplorationTrigger,
-    normalize_experience,
-    resolve_budget,
-)
-from agenteval.models import CampaignStatus
-from agenteval.storage.db import as_utc, utc_now
-from agenteval.storage.repository import (
-    CampaignRepository,
-    ExplorationMessageRepository,
-    ExplorationSessionRepository,
-    TargetRepository,
-)
-
-
-def coerce_reply_text(content: Any) -> str:
-    """Flatten a reply payload to text; tutu returns msgBody as a parsed object,
-    and str(dict) would leak a Python repr into the view."""
-    if isinstance(content, str):
-        return content
-    if isinstance(content, dict):
-        for key in ("content", "text", "message"):
-            value = content.get(key)
-            if isinstance(value, str) and value:
-                return value
-    if content is None:
-        return ""
-    return json.dumps(content, ensure_ascii=False)
-
-
-def _check_creation_guardrails(
-    campaign,
-    triggered_by: ExplorationTrigger,
-    budget: ExplorationBudget,
-    repo: ExplorationSessionRepository,
-) -> None:
-    if campaign.status != CampaignStatus.RUNNING:
-        raise ExplorationGuardrailError("活动不在进行中，无法创建探索会话")
-    if campaign.time_scale != 1 and triggered_by != ExplorationTrigger.MANUAL:
-        raise ExplorationGuardrailError("加速调试线仅允许手动创建探索会话（时间压缩与拟真相冲突）")
-    sessions = repo.list_by_campaign(campaign.id)
-    if len(sessions) >= budget.max_sessions:
-        raise ExplorationGuardrailError(f"探索会话数超出预算：本活动窗口最多 {budget.max_sessions} 个会话")
-    if sessions:
-        latest = max(s.created_at for s in sessions if s.created_at)
-        elapsed = (utc_now() - as_utc(latest)).total_seconds()
-        if elapsed < budget.min_interval_seconds:
-            wait_minutes = budget.min_interval_seconds // 60
-            raise ExplorationGuardrailError(f"相邻探索会话间隔不足：最小间隔 {wait_minutes} 分钟，请稍后再试")
-
-
-def open_session(
-    db_session: Session,
-    *,
-    campaign_id: str,
-    persona: dict[str, Any],
-    goal: str,
-    seed_ref: Optional[dict[str, Any]] = None,
-    triggered_by: ExplorationTrigger = ExplorationTrigger.AUTO,
-) -> ExplorationSession:
-    """按账本规则开一个 running 会话；违规抛领域异常。"""
-    campaign = CampaignRepository(db_session).get(campaign_id)
-    if not campaign:
-        raise ExplorationNotFoundError("campaign not found")
-    if not TargetRepository(db_session).get(campaign.target_id):
-        raise ExplorationNotFoundError("campaign target not found")
-
-    budget = resolve_budget(campaign)
-    repo = ExplorationSessionRepository(db_session)
-    _check_creation_guardrails(campaign, triggered_by, budget, repo)
-
-    return repo.create(ExplorationSession(
-        campaign_id=campaign.id,
-        target_id=campaign.target_id,
-        persona=persona,
-        goal=goal,
-        seed_ref=seed_ref,
-        triggered_by=triggered_by,
-    ))
-
-
-async def conduct_turn(db_session: Session, *, session_id: str, content: str) -> dict[str, Any]:
-    """一轮完整问答：状态机拒收 → 轮次预算记账 → 通道往返 → 双条落库。
-
-    消息已送达即消耗一轮预算（平台账本）：先落用户消息，再等回复。
-    """
-    repo = ExplorationSessionRepository(db_session)
-    session_obj = repo.get(session_id)
-    if not session_obj:
-        raise ExplorationNotFoundError("exploration session not found")
-    if session_obj.status != ExplorationSessionStatus.RUNNING:
-        raise ExplorationGuardrailError("探索会话不在进行中，拒收消息")
-
-    campaign = CampaignRepository(db_session).get(session_obj.campaign_id)
-    budget = resolve_budget(campaign) if campaign else ExplorationBudget()
-    if session_obj.turn_count >= budget.max_turns:
-        raise ExplorationGuardrailError(f"会话轮数超出预算：单会话最多 {budget.max_turns} 轮")
-
-    target = TargetRepository(db_session).get(session_obj.target_id)
-    if not target:
-        raise ExplorationNotFoundError("session target not found")
-
-    channel = ChannelFactory.create(target)
-    sent_at = utc_now()
-    try:
-        send_result = await channel.send(content)
-    except Exception as exc:  # channel adapters raise transport-specific errors
-        raise ExplorationChannelError(f"评测对象通道发送失败: {exc}") from exc
-    if not send_result.ok:
-        raise ExplorationChannelError(f"评测对象通道发送失败: {send_result.error}")
-
-    message_repo = ExplorationMessageRepository(db_session)
-    round_index = session_obj.turn_count + 1
-    message_repo.save_message(
-        ExplorationMessage(
-            session_id=session_obj.id, round_index=round_index, role="user", content=content, created_at=sent_at
-        )
-    )
-    session_obj.turn_count = round_index
-    repo.update(session_obj)
-
-    try:
-        reply = await channel.poll_reply(
-            send_result.question_msg_id,
-            timeout=get_settings().poll_reply_timeout,
-        )
-    except Exception as exc:
-        raise ExplorationChannelError(f"等待评测对象回复失败: {exc}") from exc
-    if reply is None:
-        raise ExplorationChannelError("等待评测对象回复超时")
-
-    received_at = utc_now()
-    latency_ms = int((received_at - sent_at).total_seconds() * 1000)
-    reply_text = coerce_reply_text(reply.content)
-    message_repo.save_message(
-        ExplorationMessage(
-            session_id=session_obj.id,
-            round_index=round_index,
-            role="assistant",
-            content=reply_text,
-            latency_ms=latency_ms,
-            created_at=received_at,
-        )
-    )
-    return {"reply": reply_text, "latency_ms": latency_ms, "turn_count": round_index}
-
-
-def close_session(db_session: Session, *, session_id: str, experience: dict[str, Any]) -> ExplorationSession:
-    """关闭会话：体验记录归一化 → 状态迁移 → 触发 judge 抽样复核。"""
-    repo = ExplorationSessionRepository(db_session)
-    session_obj = repo.get(session_id)
-    if not session_obj:
-        raise ExplorationNotFoundError("exploration session not found")
-    if session_obj.status != ExplorationSessionStatus.RUNNING:
-        raise ExplorationGuardrailError("探索会话不在进行中，无法关闭")
-
-    session_obj.experience = normalize_experience(experience)
-    session_obj.status = ExplorationSessionStatus.COMPLETED
-    session_obj.closed_at = utc_now()
-    updated = repo.update(session_obj)
-    start_judge_review(session_obj.id)
-    return updated

backend/agenteval/exploration/patrol.py

@@ -1,100 +0,0 @@
-"""Exploration patrol (探索巡检).
-
-常驻智能体的无状态巡检：筛出参与探索的正式线活动（running、
-time_scale == 1、有种子集），汇报水位以来的新增结果与剩余探索预算，
-构建完响应后推进水位，保证下次只报增量。
-"""
-
-from datetime import datetime
-from typing import Any
-
-from sqlmodel import Session
-
-from agenteval.evaluation.report import generate_campaign_report
-from agenteval.exploration.models import resolve_budget
-from agenteval.models import Campaign, CampaignStatus, EvalRun
-from agenteval.storage.db import as_utc, iso_utc, utc_now
-from agenteval.storage.repository import (
-    CampaignRepository,
-    ExplorationSessionRepository,
-    RunRepository,
-    ScenarioRepository,
-    TargetRepository,
-)
-
-
-def _new_runs_since(runs: list[EvalRun], watermark: datetime | None) -> list[EvalRun]:
-    fresh = []
-    for run in runs:
-        if run.completed_at is None:
-            continue
-        if watermark is not None and as_utc(run.completed_at) <= watermark:
-            continue
-        fresh.append(run)
-    return fresh
-
-
-def patrol_report(db_session: Session) -> dict[str, Any]:
-    """一次巡检的完整读模型：patrolled_at + 逐活动条目列表。
-
-    水位取构建响应之后的时刻：查询与持久化之间完成的结果不会在下次重复上报。
-    """
-    campaign_repo = CampaignRepository(db_session)
-    run_repo = RunRepository(db_session)
-    exploration_repo = ExplorationSessionRepository(db_session)
-    scenario_names = ScenarioRepository(db_session).name_map()
-    target_names = {t.id: t.name for t in TargetRepository(db_session).list_all()}
-
-    patrolled_at = utc_now()
-    entries: list[dict[str, Any]] = []
-    patrolled_campaigns: list[Campaign] = []
-    for campaign in campaign_repo.list_all():
-        if campaign.status != CampaignStatus.RUNNING:
-            continue
-        if campaign.time_scale != 1:
-            continue
-        if campaign.exploration_seeds is None:
-            continue
-
-        watermark = as_utc(campaign.last_patrolled_at) if campaign.last_patrolled_at else None
-        fresh = _new_runs_since(run_repo.list_by_campaign(campaign.id), watermark)
-        new_results = None
-        if fresh:
-            report = generate_campaign_report(campaign, fresh, scenario_names=scenario_names)
-            new_results = {
-                "summary": report["summary"],
-                "capability_summary": report["capability_summary"],
-            }
-
-        budget = resolve_budget(campaign)
-        sessions = exploration_repo.list_by_campaign(campaign.id)
-        seconds_since_last_session = None
-        if sessions:
-            latest = max(as_utc(s.created_at) for s in sessions if s.created_at)
-            seconds_since_last_session = int((patrolled_at - latest).total_seconds())
-
-        entries.append(
-            {
-                "campaign_id": campaign.id,
-                "campaign_name": campaign.name,
-                "target_id": campaign.target_id,
-                "target_name": target_names.get(campaign.target_id),
-                "last_patrolled_at": iso_utc(campaign.last_patrolled_at),
-                "new_results": new_results,
-                "budget": {
-                    "max_sessions": budget.max_sessions,
-                    "sessions_used": len(sessions),
-                    "remaining_sessions": max(0, budget.max_sessions - len(sessions)),
-                    "max_turns": budget.max_turns,
-                    "min_interval_seconds": budget.min_interval_seconds,
-                    "seconds_since_last_session": seconds_since_last_session,
-                },
-            }
-        )
-        patrolled_campaigns.append(campaign)
-
-    watermark_at = utc_now()
-    for campaign in patrolled_campaigns:
-        campaign_repo.touch_patrol_watermark(campaign.id, watermark_at)
-
-    return {"patrolled_at": iso_utc(watermark_at), "campaigns": entries}

backend/agenteval/storage/repository.py

@@ -1,12 +1,11 @@
 """Repository layer for database access."""
 
-from datetime import datetime
 from typing import Generic, Optional, TypeVar
 
 from sqlmodel import Session, select
 
 from agenteval.exploration.models import ExplorationMessage, ExplorationSession
-from agenteval.models import Campaign, CampaignStatus, CampaignSummary, Case, EvalResult, EvalRun, EvalTarget, Scenario
+from agenteval.models import Campaign, Case, EvalResult, EvalRun, EvalTarget, Scenario
 from agenteval.services.model_configs import ModelConfigService
 from agenteval.storage.db import (
     CampaignAnalysisDB,
@@ -248,10 +247,6 @@ class ScenarioRepository(BaseRepository[Scenario, ScenarioDB]):
             raise
         return True
 
-    def name_map(self) -> dict[str, str]:
-        """scenario_id → 名称映射：报告 / 时间线 / 列表等读路径共用的场景名取法。"""
-        return {sid: name for sid, name in self.session.exec(select(ScenarioDB.id, ScenarioDB.name)).all()}
-
 
 class RunRepository(BaseRepository[EvalRun, EvalRunDB]):
     """Repository for evaluation runs."""
@@ -412,38 +407,6 @@ class CampaignRepository(BaseRepository[Campaign, CampaignDB]):
         self.session.refresh(existing)
         return self._from_db(existing)
 
-    def touch_patrol_watermark(self, campaign_id: str, at: datetime) -> None:
-        """窄口径原子更新：只写巡检水位，不覆写并发的 status / summary 变更。"""
-        db = self.session.get(CampaignDB, campaign_id)
-        if not db:
-            return
-        db.last_patrolled_at = at
-        self.session.add(db)
-        self.session.commit()
-        self.session.refresh(db)
-
-    def mark_cancelled(self, campaign_id: str, at: datetime) -> Optional[Campaign]:
-        """窄口径终态迁移：只写 status + completed_at，不抹掉水位与调度进度。"""
-        db = self.session.get(CampaignDB, campaign_id)
-        if not db:
-            return None
-        db.status = CampaignStatus.CANCELLED.value
-        db.completed_at = at
-        self.session.add(db)
-        self.session.commit()
-        self.session.refresh(db)
-        return self._from_db(db)
-
-    def save_scheduler_state(self, campaign_id: str, summary: CampaignSummary) -> None:
-        """窄口径调度进度持久化：只写 summary 列，不覆写并发的状态 / 水位变更。"""
-        db = self.session.get(CampaignDB, campaign_id)
-        if not db:
-            return
-        db.set_summary(summary.model_dump(mode="json"))
-        self.session.add(db)
-        self.session.commit()
-        self.session.refresh(db)
-
 
 class CampaignAnalysisRepository:
     """Repository for campaign analysis rows (one per campaign, upserted)."""

backend/agenteval/web/routers/campaigns.py

@@ -14,6 +14,7 @@ from sqlmodel import Session
 from agenteval.evaluation.analysis import resolve_analysis_model, start_campaign_analysis
 from agenteval.evaluation.campaign_runner import campaign_progress, request_cancel, start_campaign
 from agenteval.evaluation.comparison import (
+    build_campaign_report_dict,
     build_comparison_payload,
     compute_metric_diff,
     resolve_auto_baseline,
@@ -21,7 +22,7 @@ from agenteval.evaluation.comparison import (
 )
 from agenteval.evaluation.report import (
     build_campaign_timeline,
-    load_campaign_report,
+    generate_campaign_report,
     summarize_campaign_progress,
 )
 from agenteval.evaluation.report_render import render_campaign_markdown
@@ -117,7 +118,9 @@ async def cancel_campaign(campaign_id: str, session: Session = Depends(get_db))
     if campaign.status not in (CampaignStatus.PLANNED, CampaignStatus.RUNNING):
         raise HTTPException(status_code=400, detail="campaign is not in a cancellable state")
 
-    campaign = repo.mark_cancelled(campaign_id, utc_now()) or campaign
+    campaign.status = CampaignStatus.CANCELLED
+    campaign.completed_at = utc_now()
+    repo.update(campaign)
     request_cancel(campaign_id)
     settle_campaign_sessions(campaign_id, session)
     return campaign.model_dump()
@@ -128,7 +131,9 @@ async def get_campaign_report(campaign_id: str, session: Session = Depends(get_d
     campaign = CampaignRepository(session).get(campaign_id)
     if not campaign:
         raise HTTPException(status_code=404, detail="campaign not found")
-    report = load_campaign_report(session, campaign)
+    runs = RunRepository(session).list_by_campaign(campaign_id)
+    scenario_names = {s.id: s.name for s in ScenarioRepository(session).list_all()}
+    report = generate_campaign_report(campaign, runs, scenario_names=scenario_names)
     exploration = summarize_campaign_exploration(session, campaign_id)
     if exploration is not None:
         report["exploration"] = exploration
@@ -140,8 +145,8 @@ async def get_campaign_report_markdown(campaign_id: str, session: Session = Depe
     campaign = CampaignRepository(session).get(campaign_id)
     if not campaign:
         raise HTTPException(status_code=404, detail="campaign not found")
-    report = load_campaign_report(session, campaign)
+    runs = RunRepository(session).list_by_campaign(campaign_id)
-    scenario_names = ScenarioRepository(session).name_map()
+    scenario_names = {s.id: s.name for s in ScenarioRepository(session).list_all()}
     analysis_row = CampaignAnalysisRepository(session).get_by_campaign(campaign_id)
     analysis = analysis_row.get_result() if analysis_row and analysis_row.status == "completed" else None
     target = TargetRepository(session).get(campaign.target_id)
@@ -153,8 +158,8 @@ async def get_campaign_report_markdown(campaign_id: str, session: Session = Depe
         baseline = CampaignRepository(session).get(cmp_row.baseline_campaign_id)
         metric_diff = (
             compute_metric_diff(
-                load_campaign_report(session, baseline),
+                build_campaign_report_dict(baseline, session),
-                load_campaign_report(session, campaign),
+                build_campaign_report_dict(campaign, session),
             )
             if baseline is not None
             else None
@@ -173,7 +178,7 @@ async def get_campaign_report_markdown(campaign_id: str, session: Session = Depe
         }
 
     md = render_campaign_markdown(
-        report,
+        generate_campaign_report(campaign, runs, scenario_names=scenario_names),
         analysis=analysis,
         comparison=comparison,
         exploration=summarize_campaign_exploration(session, campaign_id),
@@ -193,7 +198,7 @@ async def get_campaign_timeline(campaign_id: str, session: Session = Depends(get
     if not campaign:
         raise HTTPException(status_code=404, detail="campaign not found")
     runs = RunRepository(session).list_by_campaign(campaign_id)
-    scenario_names = ScenarioRepository(session).name_map()
+    scenario_names = {s.id: s.name for s in ScenarioRepository(session).list_all()}
     return {"entries": build_campaign_timeline(campaign, runs, scenario_names=scenario_names)}
 
 
@@ -249,8 +254,8 @@ async def get_campaign_comparison(campaign_id: str, session: Session = Depends(g
     baseline = CampaignRepository(session).get(row.baseline_campaign_id)
     metric_diff = (
         compute_metric_diff(
-            load_campaign_report(session, baseline),
+            build_campaign_report_dict(baseline, session),
-            load_campaign_report(session, campaign),
+            build_campaign_report_dict(campaign, session),
         )
         if baseline is not None
         else None

backend/agenteval/web/routers/exploration.py

@@ -1,28 +1,45 @@
 """API routes for exploratory evaluation sessions (探索式评测, v0.9).
 
-领域逻辑（账本规则、状态机、巡检）在 exploration/lifecycle.py 与
+The virtual user (OpenClaw) drives these sessions through plain HTTP: create a
-patrol.py；本层只做 HTTP 翻译：解析请求、调用领域操作、把领域异常
+running session against a campaign, converse with the target through its real
-映射为状态码（NotFound→404、Guardrail→409、Channel→502）。预算违规
+channel, then close with a structured self-reported experience record.
-与状态违规的拒绝文案由领域层给出，拒绝本身就是对常驻智能体的反馈。
+
+Guardrails are a platform ledger — enforced here on every call, never trusted
+to client self-discipline. Budget overruns and state violations are rejected
+with 409 plus a readable reason, so the rejection itself is feedback to the
+resident agent.
 """
 
+import json
+from datetime import datetime
 from typing import Any
 
 from fastapi import APIRouter, Depends, HTTPException
 from pydantic import BaseModel, Field
 from sqlmodel import Session
 
-from agenteval.exploration import lifecycle, patrol
+from agenteval.channels.factory import ChannelFactory
-from agenteval.exploration.errors import (
+from agenteval.config import get_settings
-    ExplorationChannelError,
+from agenteval.evaluation.report import generate_campaign_report
-    ExplorationGuardrailError,
+from agenteval.exploration.judge import start_judge_review
-    ExplorationNotFoundError,
+from agenteval.exploration.models import (
+    ExplorationBudget,
+    ExplorationMessage,
+    ExplorationSession,
+    ExplorationSessionStatus,
+    ExplorationTrigger,
+    normalize_experience,
+    resolve_budget,
 )
-from agenteval.exploration.models import ExplorationTrigger
+from agenteval.models import Campaign, CampaignStatus, EvalRun
+from agenteval.storage.db import as_utc, iso_utc, utc_now
 from agenteval.storage.repository import (
     CampaignRepository,
     ExplorationMessageRepository,
     ExplorationSessionRepository,
+    RunRepository,
+    ScenarioRepository,
+    TargetRepository,
 )
 from agenteval.web.deps import get_db
 
@@ -45,18 +62,133 @@ class CloseSessionRequest(BaseModel):
     experience: dict[str, Any]
 
 
-def _translate(exc: Exception) -> HTTPException:
+def _coerce_reply_text(content: Any) -> str:
-    if isinstance(exc, ExplorationNotFoundError):
+    """Flatten a reply payload to text; tutu returns msgBody as a parsed object,
-        return HTTPException(status_code=404, detail=str(exc))
+    and str(dict) would leak a Python repr into the view."""
-    if isinstance(exc, ExplorationGuardrailError):
+    if isinstance(content, str):
-        return HTTPException(status_code=409, detail=exc.reason)
+        return content
-    return HTTPException(status_code=502, detail=exc.reason)
+    if isinstance(content, dict):
+        for key in ("content", "text", "message"):
+            value = content.get(key)
+            if isinstance(value, str) and value:
+                return value
+    if content is None:
+        return ""
+    return json.dumps(content, ensure_ascii=False)
+
+
+def _check_creation_guardrails(
+    campaign: Campaign,
+    triggered_by: ExplorationTrigger,
+    budget: ExplorationBudget,
+    repo: ExplorationSessionRepository,
+) -> None:
+    if campaign.status != CampaignStatus.RUNNING:
+        raise HTTPException(status_code=409, detail="活动不在进行中，无法创建探索会话")
+    if campaign.time_scale != 1 and triggered_by != ExplorationTrigger.MANUAL:
+        raise HTTPException(
+            status_code=409,
+            detail="加速调试线仅允许手动创建探索会话（时间压缩与拟真相冲突）",
+        )
+    sessions = repo.list_by_campaign(campaign.id)
+    if len(sessions) >= budget.max_sessions:
+        raise HTTPException(
+            status_code=409,
+            detail=f"探索会话数超出预算：本活动窗口最多 {budget.max_sessions} 个会话",
+        )
+    if sessions:
+        latest = max(s.created_at for s in sessions if s.created_at)
+        elapsed = (utc_now() - as_utc(latest)).total_seconds()
+        if elapsed < budget.min_interval_seconds:
+            wait_minutes = budget.min_interval_seconds // 60
+            raise HTTPException(
+                status_code=409,
+                detail=f"相邻探索会话间隔不足：最小间隔 {wait_minutes} 分钟，请稍后再试",
+            )
+
+
+def _new_runs_since(runs: list[EvalRun], watermark: datetime | None) -> list[EvalRun]:
+    fresh = []
+    for run in runs:
+        if run.completed_at is None:
+            continue
+        if watermark is not None and as_utc(run.completed_at) <= watermark:
+            continue
+        fresh.append(run)
+    return fresh
 
 
 @router.get("/patrol")
-async def patrol_endpoint(session: Session = Depends(get_db)) -> dict:
+async def patrol(session: Session = Depends(get_db)) -> dict:
-    """Stateless patrol for the resident agent — see exploration/patrol.py."""
+    """Stateless patrol for the resident agent.
-    return patrol.patrol_report(session)
+
+    Reports every running production-line (time_scale == 1) campaign that
+    participates in exploration (has a seed set), with new results since the
+    last watermark and the remaining exploration budget. Advances each
+    patrolled campaign's watermark after building the response, so the next
+    call only reports increments.
+    """
+    campaign_repo = CampaignRepository(session)
+    run_repo = RunRepository(session)
+    exploration_repo = ExplorationSessionRepository(session)
+    scenario_names = {s.id: s.name for s in ScenarioRepository(session).list_all()}
+    target_names = {t.id: t.name for t in TargetRepository(session).list_all()}
+
+    patrolled_at = utc_now()
+    entries: list[dict[str, Any]] = []
+    patrolled_campaigns: list[Campaign] = []
+    for campaign in campaign_repo.list_all():
+        if campaign.status != CampaignStatus.RUNNING:
+            continue
+        if campaign.time_scale != 1:
+            continue
+        if campaign.exploration_seeds is None:
+            continue
+
+        watermark = as_utc(campaign.last_patrolled_at) if campaign.last_patrolled_at else None
+        fresh = _new_runs_since(run_repo.list_by_campaign(campaign.id), watermark)
+        new_results = None
+        if fresh:
+            report = generate_campaign_report(campaign, fresh, scenario_names=scenario_names)
+            new_results = {
+                "summary": report["summary"],
+                "capability_summary": report["capability_summary"],
+            }
+
+        budget = resolve_budget(campaign)
+        sessions = exploration_repo.list_by_campaign(campaign.id)
+        seconds_since_last_session = None
+        if sessions:
+            latest = max(as_utc(s.created_at) for s in sessions if s.created_at)
+            seconds_since_last_session = int((patrolled_at - latest).total_seconds())
+
+        entries.append(
+            {
+                "campaign_id": campaign.id,
+                "campaign_name": campaign.name,
+                "target_id": campaign.target_id,
+                "target_name": target_names.get(campaign.target_id),
+                "last_patrolled_at": iso_utc(campaign.last_patrolled_at),
+                "new_results": new_results,
+                "budget": {
+                    "max_sessions": budget.max_sessions,
+                    "sessions_used": len(sessions),
+                    "remaining_sessions": max(0, budget.max_sessions - len(sessions)),
+                    "max_turns": budget.max_turns,
+                    "min_interval_seconds": budget.min_interval_seconds,
+                    "seconds_since_last_session": seconds_since_last_session,
+                },
+            }
+        )
+        patrolled_campaigns.append(campaign)
+
+    # 水位取构建响应之后的时刻：查询与持久化之间完成的结果不会在下次重复上报。
+    watermark_at = utc_now()
+    for campaign in patrolled_campaigns:
+        campaign.last_patrolled_at = watermark_at
+        campaign_repo.update(campaign)
+
+    return {"patrolled_at": iso_utc(watermark_at), "campaigns": entries}
 
 
 @router.post("/sessions")
@@ -64,18 +196,25 @@ async def create_session(
     request: CreateSessionRequest,
     session: Session = Depends(get_db),
 ) -> dict:
-    try:
+    campaign = CampaignRepository(session).get(request.campaign_id)
-        session_obj = lifecycle.open_session(
+    if not campaign:
-            session,
+        raise HTTPException(status_code=404, detail="campaign not found")
-            campaign_id=request.campaign_id,
+    if not TargetRepository(session).get(campaign.target_id):
-            persona=request.persona,
+        raise HTTPException(status_code=404, detail="campaign target not found")
-            goal=request.goal,
+
-            seed_ref=request.seed_ref,
+    budget = resolve_budget(campaign)
-            triggered_by=request.triggered_by,
+    repo = ExplorationSessionRepository(session)
-        )
+    _check_creation_guardrails(campaign, request.triggered_by, budget, repo)
-    except (ExplorationNotFoundError, ExplorationGuardrailError) as exc:
+
-        raise _translate(exc) from exc
+    session_obj = ExplorationSession(
-    return session_obj.model_dump(mode="json")
+        campaign_id=campaign.id,
+        target_id=campaign.target_id,
+        persona=request.persona,
+        goal=request.goal,
+        seed_ref=request.seed_ref,
+        triggered_by=request.triggered_by,
+    )
+    return repo.create(session_obj).model_dump(mode="json")
 
 
 @router.get("/campaigns/{campaign_id}/sessions")
@@ -106,10 +245,72 @@ async def send_session_message(
     request: SendMessageRequest,
     session: Session = Depends(get_db),
 ) -> dict:
+    repo = ExplorationSessionRepository(session)
+    session_obj = repo.get(session_id)
+    if not session_obj:
+        raise HTTPException(status_code=404, detail="exploration session not found")
+    if session_obj.status != ExplorationSessionStatus.RUNNING:
+        raise HTTPException(status_code=409, detail="探索会话不在进行中，拒收消息")
+
+    campaign = CampaignRepository(session).get(session_obj.campaign_id)
+    budget = resolve_budget(campaign) if campaign else ExplorationBudget()
+    if session_obj.turn_count >= budget.max_turns:
+        raise HTTPException(
+            status_code=409,
+            detail=f"会话轮数超出预算：单会话最多 {budget.max_turns} 轮",
+        )
+
+    target = TargetRepository(session).get(session_obj.target_id)
+    if not target:
+        raise HTTPException(status_code=404, detail="session target not found")
+
+    channel = ChannelFactory.create(target)
+    sent_at = utc_now()
     try:
-        return await lifecycle.conduct_turn(session, session_id=session_id, content=request.content)
+        send_result = await channel.send(request.content)
-    except (ExplorationNotFoundError, ExplorationGuardrailError, ExplorationChannelError) as exc:
+    except Exception as exc:  # channel adapters raise transport-specific errors
-        raise _translate(exc) from exc
+        raise HTTPException(status_code=502, detail=f"评测对象通道发送失败: {exc}")
+    if not send_result.ok:
+        raise HTTPException(
+            status_code=502,
+            detail=f"评测对象通道发送失败: {send_result.error}",
+        )
+
+    # 消息已送达即消耗一轮预算（平台账本）：先落用户消息，再等回复。
+    message_repo = ExplorationMessageRepository(session)
+    round_index = session_obj.turn_count + 1
+    message_repo.save_message(
+        ExplorationMessage(
+            session_id=session_obj.id, round_index=round_index, role="user", content=request.content, created_at=sent_at
+        )
+    )
+    session_obj.turn_count = round_index
+    repo.update(session_obj)
+
+    try:
+        reply = await channel.poll_reply(
+            send_result.question_msg_id,
+            timeout=get_settings().poll_reply_timeout,
+        )
+    except Exception as exc:
+        raise HTTPException(status_code=502, detail=f"等待评测对象回复失败: {exc}")
+    if reply is None:
+        raise HTTPException(status_code=502, detail="等待评测对象回复超时")
+
+    received_at = utc_now()
+    latency_ms = int((received_at - sent_at).total_seconds() * 1000)
+    reply_text = _coerce_reply_text(reply.content)
+    message_repo.save_message(
+        ExplorationMessage(
+            session_id=session_obj.id,
+            round_index=round_index,
+            role="assistant",
+            content=reply_text,
+            latency_ms=latency_ms,
+            created_at=received_at,
+        )
+    )
+    return {"reply": reply_text, "latency_ms": latency_ms, "turn_count": round_index}
 
 
 @router.post("/sessions/{session_id}/close")
@@ -118,8 +319,16 @@ async def close_session(
     request: CloseSessionRequest,
     session: Session = Depends(get_db),
 ) -> dict:
-    try:
+    repo = ExplorationSessionRepository(session)
-        updated = lifecycle.close_session(session, session_id=session_id, experience=request.experience)
+    session_obj = repo.get(session_id)
-    except (ExplorationNotFoundError, ExplorationGuardrailError) as exc:
+    if not session_obj:
-        raise _translate(exc) from exc
+        raise HTTPException(status_code=404, detail="exploration session not found")
+    if session_obj.status != ExplorationSessionStatus.RUNNING:
+        raise HTTPException(status_code=409, detail="探索会话不在进行中，无法关闭")
+
+    session_obj.experience = normalize_experience(request.experience)
+    session_obj.status = ExplorationSessionStatus.COMPLETED
+    session_obj.closed_at = utc_now()
+    updated = repo.update(session_obj)
+    start_judge_review(session_obj.id)
     return updated.model_dump(mode="json")

backend/agenteval/web/routers/runs.py

@@ -68,7 +68,7 @@ async def _run_evaluation(
 
 @router.get("")
 async def list_runs(session: Session = Depends(get_db)) -> list[dict]:
-    scenario_names = ScenarioRepository(session).name_map()
+    scenario_names = {s.id: s.name for s in ScenarioRepository(session).list_all()}
     target_names = {t.id: t.name for t in TargetRepository(session).list_all()}
     return [
         {

backend/agenteval/web/routers/stats.py

@@ -36,10 +36,11 @@ def _settled(runs: list[EvalRun]) -> list[EvalRun]:
 @router.get("/dashboard")
 def dashboard(session: Session = Depends(get_db)) -> dict:
     targets = TargetRepository(session).list_all()
-    scenario_names = ScenarioRepository(session).name_map()
+    scenarios = ScenarioRepository(session).list_all()
     runs = RunRepository(session).list_all()
     model_configs = ModelConfigRepository(session).list_all()
 
+    scenario_names = {s.id: s.name for s in scenarios}
     target_names = {t.id: t.name for t in targets}
 
     settled_runs = _settled(runs)
@@ -81,7 +82,7 @@ def dashboard(session: Session = Depends(get_db)) -> dict:
 
     return {
         "targets_count": len(targets),
-        "scenarios_count": len(scenario_names),
+        "scenarios_count": len(scenarios),
         "runs_count": len(runs),
         "model_configs_count": len(model_configs),
         "today_runs": today_runs,

tests/integration/test_exploration_api.py

@@ -55,9 +55,9 @@ def seeded_db(db_session, monkeypatch):
     app.dependency_overrides[get_db] = _test_get_db
 
     # 隔离后台 judge 复核任务：API 测试不真正派发后台任务
-    from agenteval.exploration import lifecycle as exploration_lifecycle
+    from agenteval.web.routers import exploration as exploration_router
 
-    monkeypatch.setattr(exploration_lifecycle, "start_judge_review", lambda session_id: None)
+    monkeypatch.setattr(exploration_router, "start_judge_review", lambda session_id: None)
 
     target = EvalTarget(
         id="t-1",
@@ -75,15 +75,15 @@ def seeded_db(db_session, monkeypatch):
 
 
 def _stub_channel_factory(monkeypatch, channel) -> None:
-    """Point the exploration lifecycle's ChannelFactory at a test channel."""
+    """Point the exploration router's ChannelFactory at a test channel."""
-    from agenteval.exploration import lifecycle as lifecycle_module
+    from agenteval.web.routers import exploration as exploration_module
 
     class _StubFactory:
         @staticmethod
         def create(target):
             return channel
 
-    monkeypatch.setattr(lifecycle_module, "ChannelFactory", _StubFactory)
+    monkeypatch.setattr(exploration_module, "ChannelFactory", _StubFactory)
 
 
 @pytest.fixture()
@@ -178,7 +178,6 @@ async def test_full_lifecycle_create_message_close(seeded_db, mock_channel, clie
 async def test_dict_reply_content_is_flattened_to_text(seeded_db, monkeypatch, client):
     """通道回复 content 为对象（如 tutu msgBody）时应提取文本而非存 str(dict)。"""
     from agenteval.channels.base import Reply
-
     from tests.unit.mock_channel import MockChannel
 
     class _DictReplyChannel(MockChannel):
@@ -322,7 +321,7 @@ async def test_poll_timeout_consumes_turn_budget(seeded_db, monkeypatch, client)
     """消息已送达但等不到回复：账本仍计一轮（超时不可绕过轮数预算）。"""
     from types import SimpleNamespace
 
-    from agenteval.exploration import lifecycle as exploration_module
+    from agenteval.web.routers import exploration as exploration_module
 
     from tests.unit.mock_channel import MockChannel
 
@@ -368,10 +367,10 @@ async def test_close_twice_rejected(seeded_db, mock_channel, client):
 
 
 async def test_close_triggers_judge_review(seeded_db, mock_channel, client, monkeypatch):
-    from agenteval.exploration import lifecycle as exploration_lifecycle
+    from agenteval.web.routers import exploration as exploration_router
 
     started: list[str] = []
-    monkeypatch.setattr(exploration_lifecycle, "start_judge_review", started.append)
+    monkeypatch.setattr(exploration_router, "start_judge_review", started.append)
 
     session_id = (await _create_session(client)).json()["id"]
     resp = await client.post(

tests/integration/test_exploration_settlement.py

@@ -48,10 +48,10 @@ def seeded_db(db_session, monkeypatch):
     target + scenario needed by both the loop path and the cancel path."""
     from agenteval.channels import factory as factory_module
     from agenteval.evaluation import engine as engine_module
-    from agenteval.exploration import lifecycle as exploration_lifecycle
     from agenteval.storage import db as db_module
     from agenteval.storage import repository as repo_module
     from agenteval.web import app as app_module
+    from agenteval.web.routers import exploration as exploration_router
 
     monkeypatch.setattr(app_module, "init_db", lambda: None)
 
@@ -68,7 +68,7 @@ def seeded_db(db_session, monkeypatch):
 
     channel = MockChannel(reply_delay=0.0)
     monkeypatch.setattr(factory_module.ChannelFactory, "create", lambda target: channel)
-    monkeypatch.setattr(exploration_lifecycle, "start_judge_review", lambda session_id: None)
+    monkeypatch.setattr(exploration_router, "start_judge_review", lambda session_id: None)
 
     from agenteval.web.deps import get_db
 

tests/unit/test_campaign_narrow_updates.py

@@ -1,102 +0,0 @@
-"""CampaignRepository 窄口径原子更新直测（架构保养候选 1）。
-
-全行 update 逼着调用方承诺「持有最新完整快照」，巡检水位 / 取消 /
-调度循环并发写入会互相覆盖。三个窄方法各自只动自己的列：
-touch_patrol_watermark（水位）、mark_cancelled（终态迁移）、
-save_scheduler_state（调度进度）。
-"""
-
-from datetime import datetime, timezone
-
-import pytest
-from agenteval.models import Campaign, CampaignPlanEntry, CampaignStatus, CampaignSummary, SchedulerState
-from agenteval.storage.db import as_utc
-from agenteval.storage.repository import CampaignRepository
-from sqlmodel import Session, SQLModel, create_engine
-
-T0 = datetime(2026, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
-T1 = datetime(2026, 1, 1, 1, 0, 0, tzinfo=timezone.utc)
-
-
-@pytest.fixture()
-def db_session(tmp_path):
-    from agenteval.storage.db import (  # noqa: F401
-        CampaignDB,
-        EvalResultDB,
-        EvalRunDB,
-        EvalTargetDB,
-        FileCategoryDB,
-        FileRecordDB,
-        ScenarioDB,
-        TurnDB,
-    )
-    engine = create_engine(
-        f"sqlite:///{tmp_path / 'narrow.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(db_session, status=CampaignStatus.RUNNING) -> Campaign:
-    campaign = CampaignRepository(db_session).create(Campaign(
-        name="cycle", target_id="t-1", window_seconds=3600,
-        plan=[CampaignPlanEntry(scenario_id="s-a", offset_seconds=0, count=1)],
-        status=status, started_at=T0,
-        summary=CampaignSummary(scheduler=SchedulerState(spawned_indices=[0])),
-    ))
-    return campaign
-
-
-def test_touch_patrol_watermark_only_touches_watermark(db_session):
-    campaign = _seed(db_session)
-    CampaignRepository(db_session).touch_patrol_watermark(campaign.id, T1)
-
-    fresh = CampaignRepository(db_session).get(campaign.id)
-    assert as_utc(fresh.last_patrolled_at) == T1
-    # 其余列原样：窄口径不覆写并发写入的 status / summary
-    assert fresh.status == CampaignStatus.RUNNING
-    assert fresh.summary.scheduler.spawned_indices == [0]
-    assert fresh.name == "cycle" and as_utc(fresh.started_at) == T0
-
-
-def test_mark_cancelled_only_touches_status_and_completed_at(db_session):
-    campaign = _seed(db_session)
-    CampaignRepository(db_session).touch_patrol_watermark(campaign.id, T1)
-
-    updated = CampaignRepository(db_session).mark_cancelled(campaign.id, T1)
-    assert updated.status == CampaignStatus.CANCELLED
-    assert as_utc(updated.completed_at) == T1
-
-    fresh = CampaignRepository(db_session).get(campaign.id)
-    assert fresh.status == CampaignStatus.CANCELLED
-    assert as_utc(fresh.completed_at) == T1
-    # 水位与调度进度不被终态迁移抹掉
-    assert as_utc(fresh.last_patrolled_at) == T1
-    assert fresh.summary.scheduler.spawned_indices == [0]
-
-
-def test_save_scheduler_state_only_touches_summary(db_session):
-    campaign = _seed(db_session)
-    CampaignRepository(db_session).touch_patrol_watermark(campaign.id, T1)
-
-    new_summary = CampaignSummary(scheduler=SchedulerState(spawned_indices=[0, 1, 2]))
-    CampaignRepository(db_session).save_scheduler_state(campaign.id, new_summary)
-
-    fresh = CampaignRepository(db_session).get(campaign.id)
-    assert fresh.summary.scheduler.spawned_indices == [0, 1, 2]
-    # 状态与水位不被调度进度覆写
-    assert fresh.status == CampaignStatus.RUNNING
-    assert as_utc(fresh.last_patrolled_at) == T1
-
-
-def test_narrow_updates_ignore_missing_campaign(db_session):
-    repo = CampaignRepository(db_session)
-    repo.touch_patrol_watermark("missing", T1)
-    repo.save_scheduler_state("missing", CampaignSummary())
-    assert repo.mark_cancelled("missing", T1) is None

tests/unit/test_campaign_report_loader.py

@@ -1,96 +0,0 @@
-"""活动报告读模型 load_campaign_report 与 ScenarioRepository.name_map 直测。
-
-读模型是报告 / 分析 / 对比 / 导出共用的「取数 + 聚合」单一出口：
-返回结果必须与手写三件套（list_by_campaign + scenario_names + generate_campaign_report）完全一致。
-"""
-
-from datetime import datetime, timedelta, timezone
-
-import pytest
-from agenteval.evaluation.report import generate_campaign_report, load_campaign_report
-from agenteval.models import Campaign, CampaignPlanEntry, CampaignStatus, Case, CaseType, EvalRun, RunStatus, Scenario
-from agenteval.storage.repository import CampaignRepository, RunRepository, ScenarioRepository
-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,
-        FileCategoryDB,
-        FileRecordDB,
-        ScenarioDB,
-        TurnDB,
-    )
-    engine = create_engine(
-        f"sqlite:///{tmp_path / 'loader.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(session) -> Campaign:
-    ScenarioRepository(session).create(Scenario(
-        id="s-a", name="夜间问诊",
-        cases=[Case(id="c1", type=CaseType.SINGLE, messages=["hi"])],
-    ))
-    campaign = CampaignRepository(session).create(Campaign(
-        name="cycle", target_id="t-1", window_seconds=12, time_scale=1.0,
-        plan=[CampaignPlanEntry(scenario_id="s-a", offset_seconds=0, count=1)],
-    ))
-    campaign.status = CampaignStatus.RUNNING
-    campaign.started_at = T0
-    campaign = CampaignRepository(session).update(campaign)
-    RunRepository(session).create(EvalRun(
-        target_id="t-1", scenario_id="s-a", campaign_id=campaign.id,
-        status=RunStatus.COMPLETED, started_at=T0 + timedelta(seconds=1),
-        summary={"total_cases": 1, "passed_cases": 1, "pass_rate": 1.0, "avg_latency_ms": 100},
-    ))
-    return campaign
-
-
-def test_load_campaign_report_matches_manual_trio(db_session):
-    campaign = _seed(db_session)
-    report = load_campaign_report(db_session, campaign)
-
-    runs = RunRepository(db_session).list_by_campaign(campaign.id)
-    scenario_names = {s.id: s.name for s in ScenarioRepository(db_session).list_all()}
-    expected = generate_campaign_report(campaign, runs, scenario_names=scenario_names)
-
-    assert report == expected
-    cap = next(c for c in report["capability_summary"] if c["scenario_id"] == "s-a")
-    assert cap["scenario_name"] == "夜间问诊"
-
-
-def test_load_campaign_report_empty_campaign(db_session):
-    campaign = CampaignRepository(db_session).create(Campaign(
-        name="empty", target_id="t-1", window_seconds=12,
-        plan=[CampaignPlanEntry(scenario_id="s-x", offset_seconds=0, count=1)],
-    ))
-    report = load_campaign_report(db_session, campaign)
-    assert report["summary"]["total_runs"] == 0
-    assert report["capability_summary"] == []
-
-
-def test_scenario_name_map(db_session):
-    _seed(db_session)
-    ScenarioRepository(db_session).create(Scenario(
-        id="s-b", name="缴费引导",
-        cases=[Case(id="c2", type=CaseType.SINGLE, messages=["缴费"])],
-    ))
-    assert ScenarioRepository(db_session).name_map() == {"s-a": "夜间问诊", "s-b": "缴费引导"}
-
-
-def test_scenario_name_map_empty(db_session):
-    assert ScenarioRepository(db_session).name_map() == {}

tests/unit/test_exploration_lifecycle.py

@@ -1,218 +0,0 @@
-"""探索生命周期领域模块直测（架构保养候选 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.repository import (
-    CampaignRepository,
-    ExplorationSessionRepository,
-    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