AgentEvalTool/backend/agenteval/services/runs.py

"""Orchestration for single evaluation runs.

Owns the full lifecycle of a web-started run: creating the DB row, launching
and tracking the background task (via a :class:`TaskRegistry`), cancelling it
(with a DB fallback when no live task exists), and assembling the ``/logs``
payload. The router translates the domain errors raised here into HTTP codes.
"""

import asyncio

from sqlmodel import Session

from agenteval.config import get_settings
from agenteval.evaluation.case_verdict import build_case_evidence, resolve_case_verdicts
from agenteval.evaluation.engine import EvalEngine
from agenteval.models import EvalRun, RunStatus, RunSummary, RunTrigger
from agenteval.storage.db import get_session, iso_utc
from agenteval.storage.repository import RunRepository, ScenarioRepository, TargetRepository
from agenteval.task_registry import TaskRegistry
from agenteval.utils.llm import extract_reply_text
from agenteval.utils.webhook import send_run_webhook

run_registry = TaskRegistry()


class RunNotFoundError(LookupError):
    pass


class RunNotCancellableError(ValueError):
    pass


class RunStartError(LookupError):
    """Target or scenario referenced by a start request does not exist."""


async def execute_run(
    run_id: str,
    target_id: str,
    scenario_id: str,
    *,
    cancel_token: asyncio.Event,
    on_progress,
) -> None:
    """Drive one evaluation run to completion, then fire the webhook.

    ``on_progress`` is called as ``on_progress(event, data)`` for every engine
    progress event; the web layer wires the WebSocket broadcast in here so the
    service stays transport-agnostic.
    """
    session = get_session()
    try:
        target = TargetRepository(session).get(target_id)
        scenario = ScenarioRepository(session).get(scenario_id)
        existing_run = RunRepository(session).get(run_id)
        if not target or not scenario:
            return

        engine = EvalEngine(
            target=target,
            scenario=scenario,
            session=session,
            cancel_token=cancel_token,
            max_concurrent_cases=get_settings().max_concurrent_cases,
        )
        await engine.run(progress_callback=on_progress, existing_run=existing_run)
        # Fire webhook after run completes (non-blocking, best-effort)
        completed_run = RunRepository(session).get(run_id)
        if completed_run:
            await send_run_webhook(
                run_id=run_id,
                status=completed_run.status.value,
                summary=completed_run.summary.model_dump(mode="json") if completed_run.summary else {},
            )
    finally:
        session.close()


def start_run(session: Session, target_id: str, scenario_id: str, triggered_by: RunTrigger, on_progress) -> EvalRun:
    """Create the run row and launch its background task.

    ``on_progress`` receives the run id and returns the progress callback for
    that run, letting the caller keep per-run wiring (e.g. WS channels).
    Raises :class:`RunStartError` when the target or scenario is missing.
    """
    target = TargetRepository(session).get(target_id)
    scenario = ScenarioRepository(session).get(scenario_id)
    if not target or not scenario:
        raise RunStartError("target or scenario not found")

    run = EvalRun(
        target_id=target_id,
        scenario_id=scenario_id,
        scenario_version=scenario.version or 1,
        triggered_by=triggered_by,
    )
    run = RunRepository(session).create(run)

    run_registry.launch(
        run.id,
        lambda cancel_token: execute_run(
            run.id,
            target_id,
            scenario_id,
            cancel_token=cancel_token,
            on_progress=on_progress(run.id),
        ),
    )
    return run


def cancel_run(session: Session, run_id: str) -> EvalRun:
    """Cancel a pending/running run.

    Signals the live task when one exists; otherwise (e.g. process restarted)
    marks the DB row failed directly. Raises :class:`RunNotFoundError` /
    :class:`RunNotCancellableError`.
    """
    repo = RunRepository(session)
    run = repo.get(run_id)
    if not run:
        raise RunNotFoundError("run not found")
    if run.status not in (RunStatus.PENDING, RunStatus.RUNNING):
        raise RunNotCancellableError("run is not in a cancellable state")

    signalled = run_registry.cancel(run_id)
    if not signalled:
        run.status = RunStatus.FAILED
        run.summary = {
            "error": {"code": "cancelled_by_user", "message": "评测已手动停止"},
        }
        repo.update(run)
    return run


def build_run_logs(session: Session, run_id: str) -> dict:
    """Assemble the ``/logs`` payload for an existing run.

    Per-case verdicts come from :func:`resolve_case_verdicts`, which prefers
    the engine's stored case_outcomes and approximates only for legacy runs.
    """
    repo = RunRepository(session)
    run = repo.get(run_id)
    if not run:
        raise RunNotFoundError("run not found")

    turns = repo.get_turns(run_id)
    results = repo.get_results(run_id)

    turns_data = [
        {
            "id": t.id,
            "case_id": t.case_id,
            "round_index": t.round_index,
            "latency_ms": t.latency_ms,
            "sent_text": t.get_sent_message().get("msgBody", {}).get("content", ""),
            "reply_text": extract_reply_text(t.get_reply()),
            "sent_at": iso_utc(t.sent_at),
            "received_at": iso_utc(t.received_at),
        }
        for t in turns
    ]
    results_data = [
        {
            "case_id": r.case_id,
            "rule_type": r.rule_type,
            "passed": r.passed,
            "score": r.score,
            "reason": r.reason,
        }
        for r in results
    ]

    summary = run.summary or RunSummary()
    errored_case_ids = {e.get("case_id") for e in summary.case_errors}
    verdicts = resolve_case_verdicts(
        case_outcomes=summary.case_outcomes,
        evidence=build_case_evidence(turns, results),
        errored_case_ids=errored_case_ids,
    )
    case_verdicts = {cid: {"passed": v.passed, "connectivity": v.connectivity} for cid, v in verdicts.items()}

    scenario_snapshot: dict = {}
    scenario = ScenarioRepository(session).get(run.scenario_id)
    if scenario:
        for case in scenario.cases:
            scenario_snapshot[case.id] = {
                "id": case.id,
                "type": case.type.value if hasattr(case.type, "value") else str(case.type),
                "messages": list(case.messages),
                "prompt": case.prompt,
                "turns": case.turns,
                "expectations": {
                    "intent": case.expectations.intent,
                    "keywords_include": list(case.expectations.keywords_include),
                    "keywords_exclude": list(case.expectations.keywords_exclude),
                    "response_time_max_ms": case.expectations.response_time_max_ms,
                    "coherence_min_score": case.expectations.coherence_min_score,
                },
                "eval_rules": [{"type": r.type, "params": dict(r.params), "weight": r.weight} for r in case.eval_rules],
                "rule_logic": case.rule_logic.value if hasattr(case.rule_logic, "value") else str(case.rule_logic),
                "rule_pass_threshold": case.rule_pass_threshold,
            }

    return {
        "turns": turns_data,
        "results": results_data,
        "case_verdicts": case_verdicts,
        "scenario_snapshot": scenario_snapshot,
    }