AgentEvalTool/backend/agenteval/exploration/lifecycle.py

"""Exploration session lifecycle (探索会话生命周期).

会话三态的领域操作：open_session / conduct_turn / close_session。
平台账本（预算、状态机、轮次记账）全部在此执行，违规抛类型化领域
异常（errors.py），由 router 映射 HTTP 状态码；通道收发沿用
ChannelFactory 既有接缝。
"""

from typing import Any, Optional

from sqlmodel import Session

from agenteval.channels.base import ExchangeStatus, SendResult
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.exploration_repository import ExplorationMessageRepository, ExplorationSessionRepository
from agenteval.storage.repository import (
    CampaignRepository,
    TargetRepository,
)


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")

    message_repo = ExplorationMessageRepository(db_session)
    round_index = session_obj.turn_count + 1
    sent_at = utc_now()
    channel = ChannelFactory.create(target)

    async def record_sent(_send_result: SendResult) -> None:
        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)

    outcome = await channel.exchange(
        content,
        timeout=get_settings().poll_reply_timeout,
        on_sent=record_sent,
    )
    if outcome.status is ExchangeStatus.SEND_FAILED:
        raise ExplorationChannelError(f"评测对象通道发送失败: {outcome.reason}")
    if outcome.status is ExchangeStatus.POLL_FAILED:
        raise ExplorationChannelError(f"等待评测对象回复失败: {outcome.reason}")
    if outcome.status is ExchangeStatus.REPLY_TIMEOUT:
        raise ExplorationChannelError("等待评测对象回复超时")

    received_at = utc_now()
    latency_ms = (
        outcome.latency_ms if outcome.latency_ms is not None else int((received_at - sent_at).total_seconds() * 1000)
    )
    reply_text = outcome.reply_text or ""
    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