refactor(exploration): move ledger and state machine into domain modules

架构保养候选 3：探索生命周期的账本规则与状态机从 HTTP 层落入
exploration/lifecycle.py（open/conduct/close）与 patrol.py（巡检读模型），
违规改用类型化领域异常（NotFound/Guardrail/Channel），router 瘦回纯
HTTP 翻译（404/409/502 映射），领域层不再依赖 fastapi，可脱离
TestClient 直测（新增 12 个单元测试）。

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 generate_campaign_report
+from agenteval.evaluation.report import load_campaign_report
 from agenteval.exploration.summary import summarize_campaign_exploration
 from agenteval.model_gateway import ModelGateway
 from agenteval.models import Campaign, ModelCapability, RunStatus
@@ -28,7 +28,6 @@ from agenteval.storage.repository import (
     CampaignAnalysisRepository,
     CampaignRepository,
     RunRepository,
-    ScenarioRepository,
 )
 from agenteval.utils.llm import extract_reply_text, parse_json_from_llm_text
 
@@ -306,8 +305,7 @@ async def execute_campaign_analysis(
         try:
             client = chat_client or gateway_chat_client(runtime)
             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)
+            report = load_campaign_report(session, campaign)
             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.update(campaign)
+        repo.save_scheduler_state(campaign_id, summary)
 
     return result
 

backend/agenteval/evaluation/comparison.py

@@ -15,15 +15,13 @@ 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 generate_campaign_report
-from agenteval.models import Campaign, EvalRun
+from agenteval.evaluation.report import load_campaign_report
+from agenteval.models import Campaign
 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
 
@@ -154,21 +152,14 @@ 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(
-        build_campaign_report_dict(baseline, session),
-        build_campaign_report_dict(campaign, session),
+        load_campaign_report(session, baseline),
+        load_campaign_report(session, campaign),
     )
     return {
         "auto_baseline": {
@@ -358,8 +349,8 @@ async def execute_campaign_comparison(
         )
         try:
             diff = compute_metric_diff(
-                build_campaign_report_dict(baseline, session),
-                build_campaign_report_dict(campaign, session),
+                load_campaign_report(session, baseline),
+                load_campaign_report(session, campaign),
             )
             valid_scenario_ids = {s["scenario_id"] for s in diff["scenarios"]}
             result = await narrate_period_comparison(

backend/agenteval/evaluation/report.py

@@ -9,6 +9,8 @@ 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
@@ -359,6 +361,12 @@ 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

@@ -0,0 +1,26 @@
+"""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

@@ -0,0 +1,187 @@
+"""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

@@ -0,0 +1,100 @@
+"""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,11 +1,12 @@
 """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, Case, EvalResult, EvalRun, EvalTarget, Scenario
+from agenteval.models import Campaign, CampaignStatus, CampaignSummary, Case, EvalResult, EvalRun, EvalTarget, Scenario
 from agenteval.services.model_configs import ModelConfigService
 from agenteval.storage.db import (
     CampaignAnalysisDB,
@@ -247,6 +248,10 @@ 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."""
@@ -407,6 +412,38 @@ 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,7 +14,6 @@ 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,
@@ -22,7 +21,7 @@ from agenteval.evaluation.comparison import (
 )
 from agenteval.evaluation.report import (
     build_campaign_timeline,
-    generate_campaign_report,
+    load_campaign_report,
     summarize_campaign_progress,
 )
 from agenteval.evaluation.report_render import render_campaign_markdown
@@ -118,9 +117,7 @@ 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.status = CampaignStatus.CANCELLED
-    campaign.completed_at = utc_now()
-    repo.update(campaign)
+    campaign = repo.mark_cancelled(campaign_id, utc_now()) or campaign
     request_cancel(campaign_id)
     settle_campaign_sessions(campaign_id, session)
     return campaign.model_dump()
@@ -131,9 +128,7 @@ 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")
-    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)
+    report = load_campaign_report(session, campaign)
     exploration = summarize_campaign_exploration(session, campaign_id)
     if exploration is not None:
         report["exploration"] = exploration
@@ -145,8 +140,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")
-    runs = RunRepository(session).list_by_campaign(campaign_id)
-    scenario_names = {s.id: s.name for s in ScenarioRepository(session).list_all()}
+    report = load_campaign_report(session, campaign)
+    scenario_names = ScenarioRepository(session).name_map()
     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)
@@ -158,8 +153,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(
-                build_campaign_report_dict(baseline, session),
-                build_campaign_report_dict(campaign, session),
+                load_campaign_report(session, baseline),
+                load_campaign_report(session, campaign),
             )
             if baseline is not None
             else None
@@ -178,7 +173,7 @@ async def get_campaign_report_markdown(campaign_id: str, session: Session = Depe
         }
 
     md = render_campaign_markdown(
-        generate_campaign_report(campaign, runs, scenario_names=scenario_names),
+        report,
         analysis=analysis,
         comparison=comparison,
         exploration=summarize_campaign_exploration(session, campaign_id),
@@ -198,7 +193,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 = {s.id: s.name for s in ScenarioRepository(session).list_all()}
+    scenario_names = ScenarioRepository(session).name_map()
     return {"entries": build_campaign_timeline(campaign, runs, scenario_names=scenario_names)}
 
 
@@ -254,8 +249,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(
-            build_campaign_report_dict(baseline, session),
-            build_campaign_report_dict(campaign, session),
+            load_campaign_report(session, baseline),
+            load_campaign_report(session, campaign),
         )
         if baseline is not None
         else None

backend/agenteval/web/routers/exploration.py

@@ -1,45 +1,28 @@
 """API routes for exploratory evaluation sessions (探索式评测, v0.9).
 
-The virtual user (OpenClaw) drives these sessions through plain HTTP: create a
-running session against a campaign, converse with the target through its real
-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.
+领域逻辑（账本规则、状态机、巡检）在 exploration/lifecycle.py 与
+patrol.py；本层只做 HTTP 翻译：解析请求、调用领域操作、把领域异常
+映射为状态码（NotFound→404、Guardrail→409、Channel→502）。预算违规
+与状态违规的拒绝文案由领域层给出，拒绝本身就是对常驻智能体的反馈。
 """
 
-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.channels.factory import ChannelFactory
-from agenteval.config import get_settings
-from agenteval.evaluation.report import generate_campaign_report
-from agenteval.exploration.judge import start_judge_review
-from agenteval.exploration.models import (
-    ExplorationBudget,
-    ExplorationMessage,
-    ExplorationSession,
-    ExplorationSessionStatus,
-    ExplorationTrigger,
-    normalize_experience,
-    resolve_budget,
+from agenteval.exploration import lifecycle, patrol
+from agenteval.exploration.errors import (
+    ExplorationChannelError,
+    ExplorationGuardrailError,
+    ExplorationNotFoundError,
 )
-from agenteval.models import Campaign, CampaignStatus, EvalRun
-from agenteval.storage.db import as_utc, iso_utc, utc_now
+from agenteval.exploration.models import ExplorationTrigger
 from agenteval.storage.repository import (
     CampaignRepository,
     ExplorationMessageRepository,
     ExplorationSessionRepository,
-    RunRepository,
-    ScenarioRepository,
-    TargetRepository,
 )
 from agenteval.web.deps import get_db
 
@@ -62,133 +45,18 @@ class CloseSessionRequest(BaseModel):
     experience: dict[str, Any]
 
 
-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: 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
+def _translate(exc: Exception) -> HTTPException:
+    if isinstance(exc, ExplorationNotFoundError):
+        return HTTPException(status_code=404, detail=str(exc))
+    if isinstance(exc, ExplorationGuardrailError):
+        return HTTPException(status_code=409, detail=exc.reason)
+    return HTTPException(status_code=502, detail=exc.reason)
 
 
 @router.get("/patrol")
-async def patrol(session: Session = Depends(get_db)) -> dict:
-    """Stateless patrol for the resident agent.
-
-    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}
+async def patrol_endpoint(session: Session = Depends(get_db)) -> dict:
+    """Stateless patrol for the resident agent — see exploration/patrol.py."""
+    return patrol.patrol_report(session)
 
 
 @router.post("/sessions")
@@ -196,25 +64,18 @@ async def create_session(
     request: CreateSessionRequest,
     session: Session = Depends(get_db),
 ) -> dict:
-    campaign = CampaignRepository(session).get(request.campaign_id)
-    if not campaign:
-        raise HTTPException(status_code=404, detail="campaign not found")
-    if not TargetRepository(session).get(campaign.target_id):
-        raise HTTPException(status_code=404, detail="campaign target not found")
-
-    budget = resolve_budget(campaign)
-    repo = ExplorationSessionRepository(session)
-    _check_creation_guardrails(campaign, request.triggered_by, budget, repo)
-
-    session_obj = ExplorationSession(
-        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")
+    try:
+        session_obj = lifecycle.open_session(
+            session,
+            campaign_id=request.campaign_id,
+            persona=request.persona,
+            goal=request.goal,
+            seed_ref=request.seed_ref,
+            triggered_by=request.triggered_by,
+        )
+    except (ExplorationNotFoundError, ExplorationGuardrailError) as exc:
+        raise _translate(exc) from exc
+    return session_obj.model_dump(mode="json")
 
 
 @router.get("/campaigns/{campaign_id}/sessions")
@@ -245,72 +106,10 @@ 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:
-        send_result = await channel.send(request.content)
-    except Exception as exc:  # channel adapters raise transport-specific errors
-        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}
+        return await lifecycle.conduct_turn(session, session_id=session_id, content=request.content)
+    except (ExplorationNotFoundError, ExplorationGuardrailError, ExplorationChannelError) as exc:
+        raise _translate(exc) from exc
 
 
 @router.post("/sessions/{session_id}/close")
@@ -319,16 +118,8 @@ async def close_session(
     request: CloseSessionRequest,
     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="探索会话不在进行中，无法关闭")
-
-    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)
+    try:
+        updated = lifecycle.close_session(session, session_id=session_id, experience=request.experience)
+    except (ExplorationNotFoundError, ExplorationGuardrailError) as exc:
+        raise _translate(exc) from exc
     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 = {s.id: s.name for s in ScenarioRepository(session).list_all()}
+    scenario_names = ScenarioRepository(session).name_map()
     target_names = {t.id: t.name for t in TargetRepository(session).list_all()}
     return [
         {

backend/agenteval/web/routers/stats.py

@@ -36,11 +36,10 @@ def _settled(runs: list[EvalRun]) -> list[EvalRun]:
 @router.get("/dashboard")
 def dashboard(session: Session = Depends(get_db)) -> dict:
     targets = TargetRepository(session).list_all()
-    scenarios = ScenarioRepository(session).list_all()
+    scenario_names = ScenarioRepository(session).name_map()
     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)
@@ -82,7 +81,7 @@ def dashboard(session: Session = Depends(get_db)) -> dict:
 
     return {
         "targets_count": len(targets),
-        "scenarios_count": len(scenarios),
+        "scenarios_count": len(scenario_names),
         "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.web.routers import exploration as exploration_router
+    from agenteval.exploration import lifecycle as exploration_lifecycle
 
-    monkeypatch.setattr(exploration_router, "start_judge_review", lambda session_id: None)
+    monkeypatch.setattr(exploration_lifecycle, "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 router's ChannelFactory at a test channel."""
-    from agenteval.web.routers import exploration as exploration_module
+    """Point the exploration lifecycle's ChannelFactory at a test channel."""
+    from agenteval.exploration import lifecycle as lifecycle_module
 
     class _StubFactory:
         @staticmethod
         def create(target):
             return channel
 
-    monkeypatch.setattr(exploration_module, "ChannelFactory", _StubFactory)
+    monkeypatch.setattr(lifecycle_module, "ChannelFactory", _StubFactory)
 
 
 @pytest.fixture()
@@ -178,6 +178,7 @@ 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):
@@ -321,7 +322,7 @@ async def test_poll_timeout_consumes_turn_budget(seeded_db, monkeypatch, client)
     """消息已送达但等不到回复：账本仍计一轮（超时不可绕过轮数预算）。"""
     from types import SimpleNamespace
 
-    from agenteval.web.routers import exploration as exploration_module
+    from agenteval.exploration import lifecycle as exploration_module
 
     from tests.unit.mock_channel import MockChannel
 
@@ -367,10 +368,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.web.routers import exploration as exploration_router
+    from agenteval.exploration import lifecycle as exploration_lifecycle
 
     started: list[str] = []
-    monkeypatch.setattr(exploration_router, "start_judge_review", started.append)
+    monkeypatch.setattr(exploration_lifecycle, "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_router, "start_judge_review", lambda session_id: None)
+    monkeypatch.setattr(exploration_lifecycle, "start_judge_review", lambda session_id: None)
 
     from agenteval.web.deps import get_db
 

tests/unit/test_campaign_narrow_updates.py

@@ -0,0 +1,102 @@
+"""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

@@ -0,0 +1,96 @@
+"""活动报告读模型 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

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