AgentEvalTool/backend/agenteval/intelligent_eval/lifecycle/plan_management.py

"""Plan management phase: create, plan, approve/reject, cancel, delete, reads."""

from typing import Any, Optional

from sqlmodel import Session

from agenteval.intelligent_eval.lifecycle._core import (
    IntelligentEvalNotFoundError,
    get_or_raise,
    resolve_write,
    transition,
)
from agenteval.intelligent_eval.models import IntelligentEval, IntelligentEvalStatus
from agenteval.intelligent_eval.repository import IntelligentEvalRepository
from agenteval.storage.db import IntelligentEvalDB, utc_now
from agenteval.storage.repository import TargetRepository


def create_eval(
    session: Session,
    *,
    name: str,
    target_id: str,
    goal: str,
    seeds: dict[str, Any],
    intent: str,
    role_description: str,
    time_window_hours: int = 24,
) -> IntelligentEval:
    """创建智能评估并直接进入 planning 状态（draft → planning 一步完成）。"""
    from agenteval.intelligent_eval.config_snapshot import save_snapshot

    if TargetRepository(session).get(target_id) is None:
        raise IntelligentEvalNotFoundError(f"target {target_id} not found")

    repo = IntelligentEvalRepository(session)
    ev = repo._create(
        IntelligentEval(
            name=name,
            target_id=target_id,
            status=IntelligentEvalStatus.PLANNING,
            goal=goal,
            seeds=seeds,
            intent=intent,
            role_description=role_description,
            time_window_hours=time_window_hours,
            created_at=utc_now(),
            updated_at=utc_now(),
        )
    )

    # Save config snapshot
    eval_db = session.get(IntelligentEvalDB, ev.id)
    if eval_db:
        save_snapshot(eval_db, "created", "user", session)

    return ev


def submit_plan(session: Session, eval_id: str, plan: dict[str, Any]) -> IntelligentEval:
    """OpenClaw 提交粗计划：planning → pending_approval。"""
    from agenteval.intelligent_eval.config_snapshot import save_snapshot

    repo = IntelligentEvalRepository(session)
    result = repo._submit_plan_if_planning(eval_id, plan)
    ev = resolve_write(
        eval_id,
        result,
        expected=IntelligentEvalStatus.PLANNING,
        target=IntelligentEvalStatus.PENDING_APPROVAL,
    )

    # Save config snapshot
    eval_db = session.get(IntelligentEvalDB, eval_id)
    if eval_db:
        save_snapshot(eval_db, "plan_submitted", "openclaw", session)

    return ev


def approve(session: Session, eval_id: str) -> IntelligentEval:
    """用户批准：pending_approval → executing。"""
    repo = IntelligentEvalRepository(session)
    ev = get_or_raise(repo, eval_id)
    return transition(repo, ev, IntelligentEvalStatus.EXECUTING)


def reject(session: Session, eval_id: str, feedback: str) -> IntelligentEval:
    """用户打回：pending_approval → planning（附反馈）。"""
    repo = IntelligentEvalRepository(session)
    result = repo._reject_plan_if_pending(eval_id, feedback)
    return resolve_write(
        eval_id,
        result,
        expected=IntelligentEvalStatus.PENDING_APPROVAL,
        target=IntelligentEvalStatus.PLANNING,
    )


def cancel(session: Session, eval_id: str) -> IntelligentEval:
    """用户取消：pending_approval / executing → cancelled。"""
    repo = IntelligentEvalRepository(session)
    ev = get_or_raise(repo, eval_id)
    return transition(repo, ev, IntelligentEvalStatus.CANCELLED)


def delete_eval(session: Session, eval_id: str) -> IntelligentEval:
    """逻辑删除：completed / cancelled / failed → deleted。幂等：已删除直接返回。"""
    repo = IntelligentEvalRepository(session)
    ev = repo.get_including_deleted(eval_id)
    if ev is None:
        raise IntelligentEvalNotFoundError(f"intelligent eval {eval_id} not found")
    if ev.status is IntelligentEvalStatus.DELETED:
        return ev
    return transition(repo, ev, IntelligentEvalStatus.DELETED)


def get_eval(session: Session, eval_id: str) -> IntelligentEval:
    repo = IntelligentEvalRepository(session)
    return get_or_raise(repo, eval_id)


def list_evals(session: Session) -> list[IntelligentEval]:
    return IntelligentEvalRepository(session).list_all()


def list_evals_page(
    session: Session, offset: int, limit: int, status: Optional[str] = None
) -> tuple[list[IntelligentEval], int, dict[str, int]]:
    """Return one page of evals (newest first), total count, and per-status counts."""
    repo = IntelligentEvalRepository(session)
    return repo.list_page(offset, limit, status), repo.count(), repo.count_by_status()


def eval_status_counts(session: Session) -> dict[str, int]:
    """Count evaluations per status (for the list page stat bar)."""
    return IntelligentEvalRepository(session).count_by_status()