AgentEvalTool/backend/agenteval/web/app.py

"""FastAPI web backend for AgentEvalTool."""

import logging
from contextlib import asynccontextmanager
from pathlib import Path

from fastapi import Depends, FastAPI, WebSocket, WebSocketDisconnect
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, JSONResponse

from agenteval.config import get_settings
from agenteval.storage.db import get_session, init_db
from agenteval.storage.repository import (
    CampaignAnalysisRepository,
    CampaignPeriodComparisonRepository,
    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,
    exploration,
    files,
    intelligent_evals,
    model_configs,
    proxy,
    reports,
    runs,
    scenarios,
    stats,
    targets,
)
from agenteval.web.websocket import ws_manager


@asynccontextmanager
async def lifespan(_: FastAPI):
    init_db()
    # 评测任务是进程内 asyncio 任务，重启后不会恢复——清理僵尸运行（尽力而为，不阻断启动）
    try:
        session = get_session()
        try:
            count = RunRepository(session).mark_orphans_failed()
            if count:
                logging.getLogger("agenteval").warning("启动清理：%d 个中断的运行已标记为 failed", count)
            llm_orphans = CampaignAnalysisRepository(session).mark_orphans_failed()
            llm_orphans += CampaignPeriodComparisonRepository(session).mark_orphans_failed()
            if llm_orphans:
                logging.getLogger("agenteval").warning("启动清理：%d 条中断的分析/对比已标记为 failed", llm_orphans)
            # 据库恢复所有未完成的评估活动，重建其调度循环（不重复派生已到点条目）
            from agenteval.evaluation.campaign_runner import resume_running_campaigns

            resumed = resume_running_campaigns(session)
            if resumed:
                logging.getLogger("agenteval").warning("启动恢复：%d 个进行中的评估活动已续跑", resumed)
        finally:
            session.close()
    except Exception as exc:
        logging.getLogger("agenteval").warning("启动清理失败（忽略）: %s", exc)
    yield
    # 优雅停止所有进程内任务：先停活动调度循环，再停在跑的评测运行，
    # 最后停三条 LLM 任务链（分析 / 周期对比 / judge 复核）。
    try:
        from agenteval.evaluation.analysis import analysis_registry
        from agenteval.evaluation.campaign_runner import shutdown_all
        from agenteval.evaluation.comparison import comparison_registry
        from agenteval.exploration.judge import judge_registry
        from agenteval.web.routers.runs import run_registry

        await shutdown_all()
        await run_registry.shutdown_all()
        await analysis_registry.shutdown_all()
        await comparison_registry.shutdown_all()
        await judge_registry.shutdown_all()
    except Exception as exc:
        logging.getLogger("agenteval").warning("活动调度停止失败（忽略）: %s", exc)


app = FastAPI(
    title="AgentEvalTool",
    description="智能体质量评估工具集平台 Web API",
    version=get_version(),
    lifespan=lifespan,
)

settings = get_settings()

app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.allowed_origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

_api_deps = [Depends(require_api_key)]

# Login endpoints must stay open — they are how the client obtains credentials.
app.include_router(auth.router, prefix="/api/auth", tags=["auth"])
app.include_router(targets.router, prefix="/api/targets", tags=["targets"], dependencies=_api_deps)
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(intelligent_evals.router, prefix="/api/intelligent-evals", tags=["intelligent-evals"], 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)
app.include_router(
    model_configs.router,
    prefix="/api/model-configs",
    tags=["model-configs"],
    dependencies=_api_deps,
)


@app.websocket("/openclaw")
async def openclaw_ws_root(websocket: WebSocket):
    url = proxy.get_ws_upstream()
    if websocket.query_params:
        url += f"?{websocket.query_params}"
    await proxy.ws_bridge(websocket, url)


@app.websocket("/openclaw/{path:path}")
async def openclaw_ws_proxy(websocket: WebSocket, path: str):
    url = f"{proxy.get_ws_upstream()}/{path}"
    if websocket.query_params:
        url += f"?{websocket.query_params}"
    await proxy.ws_bridge(websocket, url)


app.include_router(proxy.router, prefix="/openclaw", tags=["proxy"])

_default_dist = Path(__file__).resolve().parent.parent.parent.parent / "frontend" / "web" / "dist"
WEB_DIST = Path(settings.frontend_dist_path) if settings.frontend_dist_path else _default_dist
if WEB_DIST.exists():
    from fastapi.staticfiles import StaticFiles

    app.mount("/assets", StaticFiles(directory=WEB_DIST / "assets"), name="assets")


@app.get("/api/health", tags=["health"])
def health() -> dict:
    return {"status": "ok", **get_build_info()}


@app.websocket("/ws/runs/{run_id}")
async def websocket_run_progress(websocket: WebSocket, run_id: str):
    await ws_manager.connect(run_id, websocket)
    try:
        while True:
            await websocket.receive_text()
    except WebSocketDisconnect:
        ws_manager.disconnect(run_id, websocket)


@app.get("/{full_path:path}")
def serve_spa(full_path: str):
    """Serve the React SPA for all non-API routes."""
    index_file = WEB_DIST / "index.html"
    if index_file.exists():
        return FileResponse(index_file)
    return JSONResponse({"detail": "frontend not built"}, status_code=404)