feat(exploration): session lifecycle endpoints with platform hard guardrails

v0.9 ticket 01. Independent exploration_sessions/exploration_messages
entities (never merged into EvalRun, keeping ADR-0001/0002 semantics
intact): create/message/close APIs forward virtual-user messages through
the target's real channel, persist both parties' rows with latency, and
close with a whitelist-normalized experience record. Budget enforcement
is a platform ledger — sessions per window, turns per session, and
session interval overruns return 409 with readable reasons; accelerated
lines accept manual sessions only. Messages delivered but unanswered
still consume a turn so timeouts cannot bypass the budget.

backend/agenteval/exploration/__init__.py

@@ -0,0 +1 @@
+"""Exploratory evaluation (探索式评测): virtual-user sessions inside campaigns."""

backend/agenteval/exploration/models.py

@@ -0,0 +1,99 @@
+"""Domain models and budget defaults for exploratory evaluation (v0.9).
+
+Exploration sessions are an independent entity — never merged into EvalRun —
+so pass-rate semantics (ADR-0002) and scenario comparability (ADR-0001) stay
+untouched. Budget enforcement is a platform ledger: the defaults here are the
+hard floor/ceiling, overridable per-campaign (ticket 02 wires the override).
+"""
+
+from datetime import datetime
+from enum import Enum
+from typing import Any, Optional
+
+from pydantic import BaseModel, Field
+
+from agenteval.models import Campaign
+
+DEFAULT_MAX_SESSIONS_PER_WINDOW = 8
+DEFAULT_MAX_TURNS_PER_SESSION = 12
+DEFAULT_MIN_SESSION_INTERVAL_SECONDS = 30 * 60
+
+VALID_EMOTIONS = ("positive", "neutral", "confused", "frustrated")
+
+
+class ExplorationSessionStatus(str, Enum):
+    RUNNING = "running"
+    COMPLETED = "completed"
+    FAILED = "failed"
+    EXPIRED = "expired"
+
+
+class ExplorationTrigger(str, Enum):
+    AUTO = "auto"
+    MANUAL = "manual"
+
+
+class ExplorationMessage(BaseModel):
+    """One chat message (role/content) inside an exploration session."""
+
+    id: Optional[str] = None
+    session_id: str
+    round_index: int = 0
+    role: str = "user"
+    content: str = ""
+    latency_ms: Optional[int] = None
+    created_at: Optional[datetime] = None
+
+
+class ExplorationSession(BaseModel):
+    """A virtual-user exploration session belonging to one campaign."""
+
+    id: Optional[str] = None
+    campaign_id: str
+    target_id: str
+    persona: dict[str, Any] = Field(default_factory=dict)
+    goal: str = ""
+    seed_ref: Optional[dict[str, Any]] = None
+    status: ExplorationSessionStatus = ExplorationSessionStatus.RUNNING
+    triggered_by: ExplorationTrigger = ExplorationTrigger.AUTO
+    experience: Optional[dict[str, Any]] = None
+    judge_review: Optional[dict[str, Any]] = None
+    turn_count: int = 0
+    error: Optional[str] = None
+    created_at: Optional[datetime] = None
+    closed_at: Optional[datetime] = None
+
+
+class ExplorationBudget(BaseModel):
+    max_sessions: int = DEFAULT_MAX_SESSIONS_PER_WINDOW
+    max_turns: int = DEFAULT_MAX_TURNS_PER_SESSION
+    min_interval_seconds: int = DEFAULT_MIN_SESSION_INTERVAL_SECONDS
+
+
+def resolve_budget(campaign: Campaign) -> ExplorationBudget:
+    """Effective exploration budget for a campaign.
+
+    Currently platform defaults only; the campaign-level override field lands
+    in ticket 02 and plugs in here.
+    """
+    return ExplorationBudget()
+
+
+def normalize_experience(raw: dict[str, Any]) -> dict[str, Any]:
+    """Whitelist-normalize a self-reported experience record (v0.7 precedent).
+
+    Invalid values are coerced to safe defaults rather than rejected, so a
+    sloppy virtual user still leaves a usable evidence row.
+    """
+    raw_blockers = raw.get("blockers") if isinstance(raw.get("blockers"), list) else []
+    raw_misled = raw.get("misled") if isinstance(raw.get("misled"), list) else []
+    blockers = [str(item) for item in raw_blockers if isinstance(item, (str, int, float))]
+    misled = [str(item) for item in raw_misled if isinstance(item, (str, int, float))]
+    emotion = raw.get("emotion")
+    return {
+        "goal_achieved": bool(raw.get("goal_achieved")),
+        "blockers": blockers,
+        "misled": misled,
+        "emotion": emotion if emotion in VALID_EMOTIONS else "neutral",
+        "notes": str(raw.get("notes") or ""),
+    }

backend/agenteval/storage/db.py

@@ -251,6 +251,75 @@ class CampaignPeriodComparisonDB(SQLModel, table=True):
         self.result = _json_dumps(result)
 
 
+class ExplorationSessionDB(SQLModel, table=True):
+    """One virtual-user exploration session (探索会话) within a campaign.
+
+    Independent entity, deliberately NOT an EvalRun: exploration outcomes feed
+    the 体验判定 evidence line and must not pollute pass-rate semantics
+    (ADR-0001 comparability / ADR-0002).
+    """
+
+    __tablename__ = "exploration_sessions"
+
+    id: Optional[str] = Field(default_factory=new_uuid, primary_key=True)
+    campaign_id: str = Field(index=True, foreign_key="campaigns.id")
+    target_id: str = Field(foreign_key="eval_targets.id")
+    persona: str = "{}"
+    goal: str = ""
+    seed_ref: Optional[str] = None
+    status: str = "running"
+    triggered_by: str = "auto"
+    experience: Optional[str] = None
+    judge_review: Optional[str] = None
+    turn_count: int = 0
+    error: Optional[str] = None
+    created_at: Optional[datetime] = Field(default_factory=utc_now)
+    closed_at: Optional[datetime] = None
+
+    def get_persona(self) -> dict[str, Any]:
+        return _json_loads(self.persona)
+
+    def set_persona(self, persona: dict[str, Any]) -> None:
+        self.persona = _json_dumps(persona)
+
+    def get_seed_ref(self) -> Optional[dict[str, Any]]:
+        return _json_loads(self.seed_ref) if self.seed_ref else None
+
+    def set_seed_ref(self, seed_ref: dict[str, Any]) -> None:
+        self.seed_ref = _json_dumps(seed_ref)
+
+    def get_experience(self) -> Optional[dict[str, Any]]:
+        return _json_loads(self.experience) if self.experience else None
+
+    def set_experience(self, experience: dict[str, Any]) -> None:
+        self.experience = _json_dumps(experience)
+
+    def get_judge_review(self) -> Optional[dict[str, Any]]:
+        return _json_loads(self.judge_review) if self.judge_review else None
+
+    def set_judge_review(self, judge_review: dict[str, Any]) -> None:
+        self.judge_review = _json_dumps(judge_review)
+
+
+class ExplorationMessageDB(SQLModel, table=True):
+    """One chat message inside an exploration session (role/content form).
+
+    Parallel to ``eval_runs`` turns but never shared with them. A completed
+    question-answer pair contributes two rows (user + assistant); the reply
+    row carries ``latency_ms``.
+    """
+
+    __tablename__ = "exploration_messages"
+
+    id: Optional[str] = Field(default_factory=new_uuid, primary_key=True)
+    session_id: str = Field(index=True, foreign_key="exploration_sessions.id")
+    round_index: int = 0
+    role: str = "user"
+    content: str = ""
+    latency_ms: Optional[int] = None
+    created_at: Optional[datetime] = Field(default_factory=utc_now)
+
+
 class EvalRunDB(SQLModel, table=True):
     """Database table for evaluation runs."""
 

backend/agenteval/storage/repository.py

@@ -4,6 +4,7 @@ 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.services.model_configs import ModelConfigService
 from agenteval.storage.db import (
@@ -13,6 +14,8 @@ from agenteval.storage.db import (
     EvalResultDB,
     EvalRunDB,
     EvalTargetDB,
+    ExplorationMessageDB,
+    ExplorationSessionDB,
     ScenarioDB,
     TurnDB,
     get_session,
@@ -282,11 +285,7 @@ class RunRepository(BaseRepository[EvalRun, EvalRunDB]):
         )
 
     def list_by_campaign(self, campaign_id: str) -> list[EvalRun]:
-        statement = (
-            select(EvalRunDB)
-            .where(EvalRunDB.campaign_id == campaign_id)
-            .order_by(EvalRunDB.started_at)
-        )
+        statement = select(EvalRunDB).where(EvalRunDB.campaign_id == campaign_id).order_by(EvalRunDB.started_at)
         return [self._from_db(r) for r in self.session.exec(statement).all()]
 
     def mark_orphans_failed(self) -> int:
@@ -441,9 +440,7 @@ class CampaignPeriodComparisonRepository:
         self.session = session or get_session()
 
     def get_by_campaign(self, campaign_id: str) -> Optional[CampaignPeriodComparisonDB]:
-        statement = select(CampaignPeriodComparisonDB).where(
-            CampaignPeriodComparisonDB.campaign_id == campaign_id
-        )
+        statement = select(CampaignPeriodComparisonDB).where(CampaignPeriodComparisonDB.campaign_id == campaign_id)
         return self.session.exec(statement).first()
 
     def upsert(
@@ -510,3 +507,113 @@ class ResultRepository:
         self.session.commit()
         self.session.refresh(db)
         return _result_from_db(db)
+
+
+class ExplorationSessionRepository(BaseRepository[ExplorationSession, ExplorationSessionDB]):
+    """Repository for virtual-user exploration sessions (探索会话)."""
+
+    _table = ExplorationSessionDB
+    _order_by = "created_at"
+
+    def _copy_mutable(self, db: ExplorationSessionDB, session_obj: ExplorationSession) -> None:
+        db.goal = session_obj.goal
+        db.status = session_obj.status.value
+        db.triggered_by = session_obj.triggered_by.value
+        db.turn_count = session_obj.turn_count
+        db.error = session_obj.error
+        db.closed_at = session_obj.closed_at
+        db.set_persona(session_obj.persona)
+        if session_obj.seed_ref is not None:
+            db.set_seed_ref(session_obj.seed_ref)
+        if session_obj.experience is not None:
+            db.set_experience(session_obj.experience)
+        if session_obj.judge_review is not None:
+            db.set_judge_review(session_obj.judge_review)
+
+    def _to_db(self, session_obj: ExplorationSession) -> ExplorationSessionDB:
+        db = ExplorationSessionDB(
+            id=session_obj.id,
+            campaign_id=session_obj.campaign_id,
+            target_id=session_obj.target_id,
+            created_at=session_obj.created_at,
+        )
+        self._copy_mutable(db, session_obj)
+        return db
+
+    def _from_db(self, db: ExplorationSessionDB) -> ExplorationSession:
+        return ExplorationSession(
+            id=db.id,
+            campaign_id=db.campaign_id,
+            target_id=db.target_id,
+            persona=db.get_persona(),
+            goal=db.goal,
+            seed_ref=db.get_seed_ref(),
+            status=db.status,
+            triggered_by=db.triggered_by,
+            experience=db.get_experience(),
+            judge_review=db.get_judge_review(),
+            turn_count=db.turn_count,
+            error=db.error,
+            created_at=db.created_at,
+            closed_at=db.closed_at,
+        )
+
+    def update(self, session_obj: ExplorationSession) -> Optional[ExplorationSession]:
+        existing = self.session.get(ExplorationSessionDB, session_obj.id)
+        if not existing:
+            return None
+        self._copy_mutable(existing, session_obj)
+        self.session.add(existing)
+        self.session.commit()
+        self.session.refresh(existing)
+        return self._from_db(existing)
+
+    def list_by_campaign(self, campaign_id: str) -> list[ExplorationSession]:
+        statement = (
+            select(ExplorationSessionDB)
+            .where(ExplorationSessionDB.campaign_id == campaign_id)
+            .order_by(ExplorationSessionDB.created_at)
+        )
+        return [self._from_db(r) for r in self.session.exec(statement).all()]
+
+
+class ExplorationMessageRepository:
+    """Append-only repository for exploration session chat rows."""
+
+    def __init__(self, session: Optional[Session] = None):
+        self.session = session or get_session()
+
+    def save_message(self, message: ExplorationMessage) -> ExplorationMessage:
+        db = ExplorationMessageDB(
+            id=message.id,
+            session_id=message.session_id,
+            round_index=message.round_index,
+            role=message.role,
+            content=message.content,
+            latency_ms=message.latency_ms,
+            created_at=message.created_at,
+        )
+        self.session.add(db)
+        self.session.commit()
+        self.session.refresh(db)
+        message.id = db.id
+        return message
+
+    def list_by_session(self, session_id: str) -> list[ExplorationMessage]:
+        statement = (
+            select(ExplorationMessageDB)
+            .where(ExplorationMessageDB.session_id == session_id)
+            .order_by(ExplorationMessageDB.created_at)
+        )
+        return [
+            ExplorationMessage(
+                id=r.id,
+                session_id=r.session_id,
+                round_index=r.round_index,
+                role=r.role,
+                content=r.content,
+                latency_ms=r.latency_ms,
+                created_at=r.created_at,
+            )
+            for r in self.session.exec(statement).all()
+        ]

backend/agenteval/web/app.py

@@ -13,7 +13,19 @@ from agenteval.storage.db import get_session, init_db
 from agenteval.storage.repository import RunRepository
 from agenteval.version import get_build_info, get_version
 from agenteval.web.deps import require_api_key
-from agenteval.web.routers import auth, campaigns, files, model_configs, proxy, reports, runs, scenarios, stats, targets
+from agenteval.web.routers import (
+    auth,
+    campaigns,
+    exploration,
+    files,
+    model_configs,
+    proxy,
+    reports,
+    runs,
+    scenarios,
+    stats,
+    targets,
+)
 from agenteval.web.websocket import ws_manager
 
 
@@ -74,6 +86,7 @@ app.include_router(targets.router, prefix="/api/targets", tags=["targets"], depe
 app.include_router(scenarios.router, prefix="/api/scenarios", tags=["scenarios"], dependencies=_api_deps)
 app.include_router(runs.router, prefix="/api/runs", tags=["runs"], dependencies=_api_deps)
 app.include_router(campaigns.router, prefix="/api/campaigns", tags=["campaigns"], dependencies=_api_deps)
+app.include_router(exploration.router, prefix="/api/exploration", tags=["exploration"], dependencies=_api_deps)
 app.include_router(reports.router, prefix="/api/reports", tags=["reports"], dependencies=_api_deps)
 app.include_router(stats.router, prefix="/api/stats", tags=["stats"], dependencies=_api_deps)
 app.include_router(files.router, prefix="/api/files", tags=["files"], dependencies=_api_deps)

backend/agenteval/web/routers/exploration.py

@@ -0,0 +1,208 @@
+"""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.
+"""
+
+from datetime import timezone
+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.exploration.models import (
+    ExplorationBudget,
+    ExplorationMessage,
+    ExplorationSession,
+    ExplorationSessionStatus,
+    ExplorationTrigger,
+    normalize_experience,
+    resolve_budget,
+)
+from agenteval.models import Campaign, CampaignStatus
+from agenteval.storage.db import utc_now
+from agenteval.storage.repository import (
+    CampaignRepository,
+    ExplorationMessageRepository,
+    ExplorationSessionRepository,
+    TargetRepository,
+)
+from agenteval.web.deps import get_db
+
+router = APIRouter()
+
+
+class CreateSessionRequest(BaseModel):
+    campaign_id: str
+    persona: dict[str, Any]
+    goal: str = Field(min_length=1)
+    seed_ref: dict[str, Any] | None = None
+    triggered_by: ExplorationTrigger = ExplorationTrigger.AUTO
+
+
+class SendMessageRequest(BaseModel):
+    content: str = Field(min_length=1)
+
+
+class CloseSessionRequest(BaseModel):
+    experience: dict[str, Any]
+
+
+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)
+        if latest.tzinfo is None:  # SQLite round-trip drops tzinfo; stored times are UTC
+            latest = latest.replace(tzinfo=timezone.utc)
+        elapsed = (utc_now() - 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} 分钟，请稍后再试",
+            )
+
+
+@router.post("/sessions")
+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")
+
+
+@router.post("/sessions/{session_id}/messages")
+async def send_session_message(
+    session_id: str,
+    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 = str(reply.content)
+    message_repo.save_message(
+        ExplorationMessage(
+            session_id=session_obj.id,
+            round_index=round_index,
+            role="assistant",
+            content=reply_text,
+            latency_ms=latency_ms,
+            created_at=received_at,
+        )
+    )
+    return {"reply": reply_text, "latency_ms": latency_ms, "turn_count": round_index}
+
+
+@router.post("/sessions/{session_id}/close")
+async def close_session(
+    session_id: str,
+    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()
+    return repo.update(session_obj).model_dump(mode="json")

migrations/versions/0e4a7c91d2b3_add_exploration_tables.py

@@ -0,0 +1,69 @@
+"""add exploration_sessions and exploration_messages tables
+
+Revision ID: 0e4a7c91d2b3
+Revises: f2a9b7c34d18
+Create Date: 2026-08-03
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+import sqlmodel  # noqa: F401
+from alembic import op
+
+revision: str = "0e4a7c91d2b3"
+down_revision: Union[str, Sequence[str], None] = "f2a9b7c34d18"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "exploration_sessions",
+        sa.Column("id", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+        sa.Column("campaign_id", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+        sa.Column("target_id", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+        sa.Column("persona", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+        sa.Column("goal", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+        sa.Column("seed_ref", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
+        sa.Column("status", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+        sa.Column("triggered_by", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+        sa.Column("experience", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
+        sa.Column("judge_review", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
+        sa.Column("turn_count", sa.Integer(), nullable=False),
+        sa.Column("error", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
+        sa.Column("created_at", sa.DateTime(), nullable=True),
+        sa.Column("closed_at", sa.DateTime(), nullable=True),
+        sa.PrimaryKeyConstraint("id"),
+        sa.ForeignKeyConstraint(["campaign_id"], ["campaigns.id"]),
+        sa.ForeignKeyConstraint(["target_id"], ["eval_targets.id"]),
+    )
+    op.create_index(
+        "ix_exploration_sessions_campaign_id",
+        "exploration_sessions",
+        ["campaign_id"],
+    )
+    op.create_table(
+        "exploration_messages",
+        sa.Column("id", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+        sa.Column("session_id", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+        sa.Column("round_index", sa.Integer(), nullable=False),
+        sa.Column("role", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+        sa.Column("content", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+        sa.Column("latency_ms", sa.Integer(), nullable=True),
+        sa.Column("created_at", sa.DateTime(), nullable=True),
+        sa.PrimaryKeyConstraint("id"),
+        sa.ForeignKeyConstraint(["session_id"], ["exploration_sessions.id"]),
+    )
+    op.create_index(
+        "ix_exploration_messages_session_id",
+        "exploration_messages",
+        ["session_id"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_exploration_messages_session_id")
+    op.drop_table("exploration_messages")
+    op.drop_index("ix_exploration_sessions_campaign_id")
+    op.drop_table("exploration_sessions")

tests/conftest.py

@@ -35,6 +35,8 @@ def db_session(tmp_db_path: Path) -> Session:
         EvalResultDB,
         EvalRunDB,
         EvalTargetDB,
+        ExplorationMessageDB,
+        ExplorationSessionDB,
         ModelConfigDB,
         ScenarioDB,
         ScenarioModelBindingDB,
@@ -49,6 +51,7 @@ def db_session(tmp_db_path: Path) -> Session:
 
     # Sanity check: verify the scenarios table has all expected columns.
     from sqlalchemy import inspect as sa_inspect
+
     cols = [c["name"] for c in sa_inspect(engine).get_columns("scenarios")]
     assert "llm_config" in cols, f"scenarios table missing llm_config; cols={cols}"
 

tests/integration/test_exploration_api.py

@@ -0,0 +1,375 @@
+"""Integration tests for exploration session lifecycle + platform guardrails (v0.9 票据 01).
+
+Uses ``httpx.AsyncClient`` with ``app=`` to drive the FastAPI app in-process.
+Channel I/O is stubbed with MockChannel; tests cover the full lifecycle
+(create → message → close), the three budget guardrails (409), trigger-source
+gating per line tier, and experience-record normalization.
+"""
+
+from datetime import timedelta
+
+import pytest
+from agenteval.models import Campaign, ChannelType, EvalTarget, PlatformType, TargetStatus
+from agenteval.storage.db import ExplorationSessionDB, utc_now
+from agenteval.storage.repository import CampaignRepository, ExplorationSessionRepository, TargetRepository
+from agenteval.web.app import app
+from httpx import ASGITransport, AsyncClient
+
+
+def _make_campaign(campaign_id: str, *, time_scale: float = 1.0, status: str = "running") -> Campaign:
+    return Campaign(
+        id=campaign_id,
+        name=f"campaign-{campaign_id}",
+        target_id="t-1",
+        window_seconds=86400,
+        time_scale=time_scale,
+        plan=[{"scenario_id": "s-1", "offset_seconds": 0, "count": 1}],
+        status=status,
+        started_at=utc_now(),
+    )
+
+
+@pytest.fixture()
+def seeded_db(db_session, monkeypatch):
+    """Patch get_session/get_db to the test session and seed target + campaign."""
+    from agenteval.storage import db as db_module
+    from agenteval.storage import repository as repo_module
+    from agenteval.web import app as app_module
+
+    monkeypatch.setattr(app_module, "init_db", lambda: None)
+
+    def _test_get_session():
+        return db_session
+
+    monkeypatch.setattr(db_module, "get_session", _test_get_session)
+    monkeypatch.setattr(repo_module, "get_session", _test_get_session)
+
+    from agenteval.web.deps import get_db
+
+    def _test_get_db():
+        try:
+            yield db_session
+        finally:
+            pass
+
+    app.dependency_overrides[get_db] = _test_get_db
+
+    target = EvalTarget(
+        id="t-1",
+        name="mock-target",
+        platform=PlatformType.AI_DIGITAL_EMPLOYEE,
+        channel_type=ChannelType.TUTU_API,
+        channel_config={"base_url": "http://mock", "token": "x"},
+        status=TargetStatus.ACTIVE,
+    )
+    TargetRepository(db_session).create(target)
+    CampaignRepository(db_session).create(_make_campaign("c-1"))
+
+    yield db_session
+    app.dependency_overrides.clear()
+
+
+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
+
+    class _StubFactory:
+        @staticmethod
+        def create(target):
+            return channel
+
+    monkeypatch.setattr(exploration_module, "ChannelFactory", _StubFactory)
+
+
+@pytest.fixture()
+def mock_channel(monkeypatch):
+    """Stub ChannelFactory in the exploration router with a MockChannel."""
+    from tests.unit.mock_channel import MockChannel
+
+    channel = MockChannel(reply_text="您好，请问有什么可以帮您？")
+    _stub_channel_factory(monkeypatch, channel)
+    return channel
+
+
+@pytest.fixture()
+async def client():
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as c:
+        yield c
+
+
+def _session_payload(**overrides) -> dict:
+    payload = {
+        "campaign_id": "c-1",
+        "persona": {"name": "急性子用户", "traits": ["急躁", "目标导向"]},
+        "goal": "查询本月账单并完成缴费",
+        "triggered_by": "auto",
+    }
+    payload.update(overrides)
+    return payload
+
+
+def _rewind_latest_session(db_session, minutes: int = 31) -> None:
+    """Move the newest session's created_at back so the interval guardrail passes."""
+    repo = ExplorationSessionRepository(db_session)
+    sessions = repo.list_by_campaign("c-1")
+    latest = max(sessions, key=lambda s: s.created_at)
+    row = db_session.get(ExplorationSessionDB, latest.id)
+    row.created_at = utc_now() - timedelta(minutes=minutes)
+    db_session.add(row)
+    db_session.commit()
+
+
+async def _create_session(client, **overrides):
+    return await client.post("/api/exploration/sessions", json=_session_payload(**overrides))
+
+
+# ---------------------------------------------------------------- lifecycle
+
+
+async def test_full_lifecycle_create_message_close(seeded_db, mock_channel, client):
+    resp = await _create_session(client)
+    assert resp.status_code == 200, resp.text
+    session = resp.json()
+    assert session["status"] == "running"
+    assert session["campaign_id"] == "c-1"
+    assert session["target_id"] == "t-1"
+    assert session["persona"]["name"] == "急性子用户"
+    session_id = session["id"]
+
+    resp = await client.post(
+        f"/api/exploration/sessions/{session_id}/messages",
+        json={"content": "我要查这个月的账单"},
+    )
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert body["reply"] == "您好，请问有什么可以帮您？"
+    assert isinstance(body["latency_ms"], int)
+    assert body["turn_count"] == 1
+
+    repo = ExplorationSessionRepository(seeded_db)
+    assert repo.get(session_id).turn_count == 1
+
+    resp = await client.post(
+        f"/api/exploration/sessions/{session_id}/close",
+        json={
+            "experience": {
+                "goal_achieved": True,
+                "blockers": [],
+                "misled": [],
+                "emotion": "positive",
+                "notes": "顺利完成",
+            }
+        },
+    )
+    assert resp.status_code == 200, resp.text
+    closed = resp.json()
+    assert closed["status"] == "completed"
+    assert closed["experience"]["goal_achieved"] is True
+    assert closed["experience"]["emotion"] == "positive"
+    assert closed["closed_at"] is not None
+
+
+async def test_create_requires_existing_running_campaign(seeded_db, client):
+    resp = await _create_session(client, campaign_id="nope")
+    assert resp.status_code == 404
+
+    CampaignRepository(seeded_db).create(_make_campaign("c-done", status="completed"))
+    resp = await _create_session(client, campaign_id="c-done")
+    assert resp.status_code == 409
+    assert "进行中" in resp.json()["detail"]
+
+
+async def test_accelerated_line_accepts_manual_only(seeded_db, mock_channel, client):
+    CampaignRepository(seeded_db).create(_make_campaign("c-fast", time_scale=24.0))
+
+    resp = await _create_session(client, campaign_id="c-fast", triggered_by="auto")
+    assert resp.status_code == 409
+    assert "手动" in resp.json()["detail"]
+
+    resp = await _create_session(client, campaign_id="c-fast", triggered_by="manual")
+    assert resp.status_code == 200
+
+
+async def test_session_budget_guardrail(seeded_db, mock_channel, client):
+    for _ in range(8):
+        resp = await _create_session(client)
+        assert resp.status_code == 200, resp.text
+        _rewind_latest_session(seeded_db)
+
+    resp = await _create_session(client)
+    assert resp.status_code == 409
+    assert "预算" in resp.json()["detail"]
+
+
+async def test_session_interval_guardrail(seeded_db, mock_channel, client):
+    resp = await _create_session(client)
+    assert resp.status_code == 200
+
+    resp = await _create_session(client)
+    assert resp.status_code == 409
+    assert "间隔" in resp.json()["detail"]
+
+    _rewind_latest_session(seeded_db)
+    resp = await _create_session(client)
+    assert resp.status_code == 200
+
+
+async def test_turn_budget_guardrail(seeded_db, mock_channel, client):
+    session_id = (await _create_session(client)).json()["id"]
+    for i in range(12):
+        resp = await client.post(
+            f"/api/exploration/sessions/{session_id}/messages",
+            json={"content": f"第 {i + 1} 轮问题"},
+        )
+        assert resp.status_code == 200, resp.text
+
+    resp = await client.post(
+        f"/api/exploration/sessions/{session_id}/messages",
+        json={"content": "第 13 轮问题"},
+    )
+    assert resp.status_code == 409
+    assert "轮数" in resp.json()["detail"]
+
+
+async def test_message_rejected_when_session_not_running(seeded_db, mock_channel, client):
+    session_id = (await _create_session(client)).json()["id"]
+    resp = await client.post(
+        f"/api/exploration/sessions/{session_id}/close",
+        json={"experience": {"goal_achieved": False}},
+    )
+    assert resp.status_code == 200
+
+    resp = await client.post(
+        f"/api/exploration/sessions/{session_id}/messages",
+        json={"content": "还在吗？"},
+    )
+    assert resp.status_code == 409
+    assert "进行中" in resp.json()["detail"]
+
+
+async def test_message_unknown_session_returns_404(seeded_db, mock_channel, client):
+    resp = await client.post("/api/exploration/sessions/nope/messages", json={"content": "hi"})
+    assert resp.status_code == 404
+
+
+async def test_channel_failure_returns_502_without_consuming_turn(seeded_db, monkeypatch, client):
+    from tests.unit.mock_channel import MockChannel
+
+    channel = MockChannel(send_ok=False)
+    _stub_channel_factory(monkeypatch, channel)
+
+    session_id = (await _create_session(client)).json()["id"]
+    resp = await client.post(
+        f"/api/exploration/sessions/{session_id}/messages",
+        json={"content": "你好"},
+    )
+    assert resp.status_code == 502
+    assert ExplorationSessionRepository(seeded_db).get(session_id).turn_count == 0
+
+
+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 tests.unit.mock_channel import MockChannel
+
+    channel = MockChannel(missing_reply=True)
+    _stub_channel_factory(monkeypatch, channel)
+    monkeypatch.setattr(exploration_module, "get_settings", lambda: SimpleNamespace(poll_reply_timeout=0.05))
+
+    session_id = (await _create_session(client)).json()["id"]
+    resp = await client.post(
+        f"/api/exploration/sessions/{session_id}/messages",
+        json={"content": "有人在吗"},
+    )
+    assert resp.status_code == 502
+    assert ExplorationSessionRepository(seeded_db).get(session_id).turn_count == 1
+
+
+async def test_close_normalizes_experience(seeded_db, mock_channel, client):
+    session_id = (await _create_session(client)).json()["id"]
+    resp = await client.post(
+        f"/api/exploration/sessions/{session_id}/close",
+        json={
+            "experience": {
+                "blockers": [42, {"not": "a string"}],
+                "misled": "不是列表",
+                "emotion": "暴怒!!!",
+            }
+        },
+    )
+    assert resp.status_code == 200, resp.text
+    experience = resp.json()["experience"]
+    assert experience["goal_achieved"] is False
+    assert experience["blockers"] == ["42"]
+    assert experience["misled"] == []
+    assert experience["emotion"] == "neutral"
+
+
+async def test_close_twice_rejected(seeded_db, mock_channel, client):
+    session_id = (await _create_session(client)).json()["id"]
+    payload = {"experience": {"goal_achieved": True}}
+    assert (await client.post(f"/api/exploration/sessions/{session_id}/close", json=payload)).status_code == 200
+    resp = await client.post(f"/api/exploration/sessions/{session_id}/close", json=payload)
+    assert resp.status_code == 409
+
+
+# ---------------------------------------------------------------- migration
+
+
+def test_exploration_migration_on_existing_db(tmp_path, monkeypatch):
+    """Alembic migration applies on an existing DB at the previous head.
+
+    Brand-new DBs take the create_all path (exercised by every test above via
+    the db_session fixture, which creates the new tables from metadata).
+    """
+    from pathlib import Path
+
+    from agenteval.storage import db as db_module
+    from alembic import command
+    from alembic.config import Config
+    from sqlalchemy import create_engine, inspect
+
+    database_url = f"sqlite:///{tmp_path / 'exploration.db'}"
+    monkeypatch.setattr(db_module, "DATABASE_URL", database_url)
+    config = Config(str(Path(__file__).resolve().parents[2] / "alembic.ini"))
+
+    # 既有库先例：基线迁移假设 create_all 建好的表已存在，先 stamp 基线前状态
+    from sqlmodel import SQLModel
+
+    engine = create_engine(database_url)
+    SQLModel.metadata.create_all(engine)
+    engine.dispose()
+    with create_engine(database_url).begin() as connection:
+        from sqlalchemy import text
+
+        connection.execute(text("DROP TABLE IF EXISTS exploration_sessions"))
+        connection.execute(text("DROP TABLE IF EXISTS exploration_messages"))
+        connection.execute(text("DROP TABLE IF EXISTS alembic_version"))
+
+    command.stamp(config, "f2a9b7c34d18")
+    command.upgrade(config, "head")
+
+    inspector = inspect(create_engine(database_url))
+    tables = set(inspector.get_table_names())
+    assert "exploration_sessions" in tables
+    assert "exploration_messages" in tables
+    session_cols = {c["name"] for c in inspector.get_columns("exploration_sessions")}
+    assert {
+        "campaign_id",
+        "target_id",
+        "persona",
+        "goal",
+        "seed_ref",
+        "status",
+        "triggered_by",
+        "experience",
+        "judge_review",
+        "turn_count",
+        "closed_at",
+    } <= session_cols
+    message_cols = {c["name"] for c in inspector.get_columns("exploration_messages")}
+    assert {"session_id", "round_index", "role", "content", "latency_ms"} <= message_cols