From df76edcf55e7d3d63e5c0b96c8648c8ecbeeb133 Mon Sep 17 00:00:00 2001
From: sinohqb <huangqiaobin2012@gmail.com>
Date: Tue, 4 Aug 2026 03:46:51 +0800
Subject: [PATCH] refactor(exploration): move ledger and state machine into
 domain modules
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

架构保养候选 3：探索生命周期的账本规则与状态机从 HTTP 层落入
exploration/lifecycle.py（open/conduct/close）与 patrol.py（巡检读模型），
违规改用类型化领域异常（NotFound/Guardrail/Channel），router 瘦回纯
HTTP 翻译（404/409/502 映射），领域层不再依赖 fastapi，可脱离
TestClient 直测（新增 12 个单元测试）。
---
 backend/agenteval/exploration/errors.py       |  26 ++
 backend/agenteval/exploration/lifecycle.py    | 187 ++++++++++++
 backend/agenteval/exploration/patrol.py       | 100 ++++++
 backend/agenteval/web/routers/exploration.py  | 284 +++---------------
 tests/integration/test_exploration_api.py     |  16 +-
 .../test_exploration_settlement.py            |   4 +-
 tests/unit/test_exploration_lifecycle.py      | 218 ++++++++++++++
 7 files changed, 579 insertions(+), 256 deletions(-)
 create mode 100644 backend/agenteval/exploration/errors.py
 create mode 100644 backend/agenteval/exploration/lifecycle.py
 create mode 100644 backend/agenteval/exploration/patrol.py
 create mode 100644 tests/unit/test_exploration_lifecycle.py

diff --git a/backend/agenteval/exploration/errors.py b/backend/agenteval/exploration/errors.py
new file mode 100644
index 0000000..838d643
--- /dev/null
+++ b/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)
diff --git a/backend/agenteval/exploration/lifecycle.py b/backend/agenteval/exploration/lifecycle.py
new file mode 100644
index 0000000..a04f2ba
--- /dev/null
+++ b/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
diff --git a/backend/agenteval/exploration/patrol.py b/backend/agenteval/exploration/patrol.py
new file mode 100644
index 0000000..a84289e
--- /dev/null
+++ b/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}
diff --git a/backend/agenteval/web/routers/exploration.py b/backend/agenteval/web/routers/exploration.py
index f799ccb..a9e8b26 100644
--- a/backend/agenteval/web/routers/exploration.py
+++ b/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,132 +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 = ScenarioRepository(session).name_map()
-    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_repo.touch_patrol_watermark(campaign.id, watermark_at)
-
-    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")
@@ -195,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")
@@ -244,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")
@@ -318,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")
diff --git a/tests/integration/test_exploration_api.py b/tests/integration/test_exploration_api.py
index bd220ee..59b5516 100644
--- a/tests/integration/test_exploration_api.py
+++ b/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()
@@ -322,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
 
@@ -368,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(
diff --git a/tests/integration/test_exploration_settlement.py b/tests/integration/test_exploration_settlement.py
index 0932691..8cf57ab 100644
--- a/tests/integration/test_exploration_settlement.py
+++ b/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
 
diff --git a/tests/unit/test_exploration_lifecycle.py b/tests/unit/test_exploration_lifecycle.py
new file mode 100644
index 0000000..bd4982c
--- /dev/null
+++ b/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