diff --git a/backend/agenteval/exploration/__init__.py b/backend/agenteval/exploration/__init__.py new file mode 100644 index 0000000..8eb6b3c --- /dev/null +++ b/backend/agenteval/exploration/__init__.py @@ -0,0 +1 @@ +"""Exploratory evaluation (探索式评测): virtual-user sessions inside campaigns.""" diff --git a/backend/agenteval/exploration/models.py b/backend/agenteval/exploration/models.py new file mode 100644 index 0000000..5283d44 --- /dev/null +++ b/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 ""), + } diff --git a/backend/agenteval/storage/db.py b/backend/agenteval/storage/db.py index f635994..d627877 100644 --- a/backend/agenteval/storage/db.py +++ b/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.""" diff --git a/backend/agenteval/storage/repository.py b/backend/agenteval/storage/repository.py index 4327ab5..43066f1 100644 --- a/backend/agenteval/storage/repository.py +++ b/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() + ] diff --git a/backend/agenteval/web/app.py b/backend/agenteval/web/app.py index f15d2ba..4c6cbd9 100644 --- a/backend/agenteval/web/app.py +++ b/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) diff --git a/backend/agenteval/web/routers/exploration.py b/backend/agenteval/web/routers/exploration.py new file mode 100644 index 0000000..4ce78df --- /dev/null +++ b/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") diff --git a/migrations/versions/0e4a7c91d2b3_add_exploration_tables.py b/migrations/versions/0e4a7c91d2b3_add_exploration_tables.py new file mode 100644 index 0000000..3689feb --- /dev/null +++ b/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") diff --git a/tests/conftest.py b/tests/conftest.py index 203c886..fd9cc31 100644 --- a/tests/conftest.py +++ b/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}" diff --git a/tests/integration/test_exploration_api.py b/tests/integration/test_exploration_api.py new file mode 100644 index 0000000..e79e690 --- /dev/null +++ b/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