"""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 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.models import ( ExplorationBudget, ExplorationMessage, ExplorationSession, ExplorationSessionStatus, ExplorationTrigger, normalize_experience, 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, ExplorationMessageRepository, ExplorationSessionRepository, RunRepository, ScenarioRepository, 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) elapsed = (utc_now() - as_utc(latest)).total_seconds() if elapsed < budget.min_interval_seconds: wait_minutes = budget.min_interval_seconds // 60 raise HTTPException( status_code=409, detail=f"相邻探索会话间隔不足：最小间隔 {wait_minutes} 分钟，请稍后再试", ) def _new_runs_since(runs: list[EvalRun], watermark: datetime | None) -> list[EvalRun]: fresh = [] for run in runs: if run.completed_at is None: continue if watermark is not None and as_utc(run.completed_at) <= watermark: continue fresh.append(run) return fresh @router.get("/patrol") async def patrol(session: Session = Depends(get_db)) -> dict: """Stateless patrol for the resident agent. Reports every running production-line (time_scale == 1) campaign that participates in exploration (has a seed set), with new results since the last watermark and the remaining exploration budget. Advances each patrolled campaign's watermark after building the response, so the next call only reports increments. """ campaign_repo = CampaignRepository(session) run_repo = RunRepository(session) exploration_repo = ExplorationSessionRepository(session) scenario_names = {s.id: s.name for s in ScenarioRepository(session).list_all()} target_names = {t.id: t.name for t in TargetRepository(session).list_all()} patrolled_at = utc_now() entries: list[dict[str, Any]] = [] patrolled_campaigns: list[Campaign] = [] for campaign in campaign_repo.list_all(): if campaign.status != CampaignStatus.RUNNING: continue if campaign.time_scale != 1: continue if campaign.exploration_seeds is None: continue watermark = as_utc(campaign.last_patrolled_at) if campaign.last_patrolled_at else None fresh = _new_runs_since(run_repo.list_by_campaign(campaign.id), watermark) new_results = None if fresh: report = generate_campaign_report(campaign, fresh, scenario_names=scenario_names) new_results = { "summary": report["summary"], "capability_summary": report["capability_summary"], } budget = resolve_budget(campaign) sessions = exploration_repo.list_by_campaign(campaign.id) seconds_since_last_session = None if sessions: latest = max(as_utc(s.created_at) for s in sessions if s.created_at) seconds_since_last_session = int((patrolled_at - latest).total_seconds()) entries.append( { "campaign_id": campaign.id, "campaign_name": campaign.name, "target_id": campaign.target_id, "target_name": target_names.get(campaign.target_id), "last_patrolled_at": iso_utc(campaign.last_patrolled_at), "new_results": new_results, "budget": { "max_sessions": budget.max_sessions, "sessions_used": len(sessions), "remaining_sessions": max(0, budget.max_sessions - len(sessions)), "max_turns": budget.max_turns, "min_interval_seconds": budget.min_interval_seconds, "seconds_since_last_session": seconds_since_last_session, }, } ) patrolled_campaigns.append(campaign) # 水位取构建响应之后的时刻：查询与持久化之间完成的结果不会在下次重复上报。 watermark_at = utc_now() for campaign in patrolled_campaigns: campaign.last_patrolled_at = watermark_at campaign_repo.update(campaign) return {"patrolled_at": iso_utc(watermark_at), "campaigns": entries} @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")