"""HTTP surface of the assistant evaluation console — admin-gated throughout. GET /api/v1/assistant/eval/overview GET /api/v1/assistant/eval/trend GET /api/v1/assistant/eval/suites POST /api/v1/assistant/eval/suites POST /api/v1/assistant/eval/suites/seed PATCH /api/v1/assistant/eval/suites/{suite_id} GET /api/v1/assistant/eval/cases POST /api/v1/assistant/eval/cases POST /api/v1/assistant/eval/cases/mine POST /api/v1/assistant/eval/cases/bulk-status PATCH /api/v1/assistant/eval/cases/{case_id} DELETE /api/v1/assistant/eval/cases/{case_id} POST /api/v1/assistant/eval/cases/{case_id}/variants GET /api/v1/assistant/eval/runs POST /api/v1/assistant/eval/runs GET /api/v1/assistant/eval/runs/{run_id} GET /api/v1/assistant/eval/runs/{run_id}/results GET /api/v1/assistant/eval/runs/{run_id}/results/{result_id} GET /api/v1/assistant/eval/runs/{run_id}/compare POST /api/v1/assistant/eval/runs/{run_id}/resume POST /api/v1/assistant/eval/runs/{run_id}/cancel POST /api/v1/assistant/eval/runs/{run_id}/promote-failures **Read gate is ``global_admin`` + ``auditor``, not ``require_admin_read()``.** That factory also admits ``management``, and these rows carry full assistant replies produced under a simulated employee identity. It mirrors the frontend's ``canReadAssistantTranscripts`` exactly, which is deliberately narrower than audit access for the same reason. **Starting a run is NOT behind ``require_browser_session``.** A run cannot send a card, file a request, or change anything an employee sees — it is read-only with respect to the business and spends provider tokens. So it gets the sensitive-endpoint rate limit and a single-flight guard rather than the step-up-auth gate reserved for actions that are catastrophic if the credential is stolen. That line is what lets the MCP twin exist without being an end-run around a browser-only gate. """ from __future__ import annotations import uuid from typing import Any, NoReturn from fastapi import APIRouter, Depends, Query, Request from pydantic import BaseModel, Field from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.assistant.evaluation import service from app.assistant.evaluation.constants import ( CASE_STATUSES, CATEGORIES, DEFAULT_CONCURRENCY, DEFAULT_REPEATS, FAILURE_TYPES, JUDGE_DIMENSIONS, LANGS, MAX_CONCURRENCY, MAX_REPEATS, ORIGINS, RECOMMENDED_REPEATS, RESUMABLE_STATUSES, RUN_RUNNING, SUITE_ACTIVE, SUITE_STATUSES, TRIGGER_MANUAL, VERDICTS, ) from app.assistant.evaluation.models import ( AssistantEvalCase, AssistantEvalResult, AssistantEvalRun, AssistantEvalSuite, ) from app.assistant.evaluation.runner import EvalRunError, cancel_run, execute_run from app.audit.service import AuditService from app.auth.models import AuthUser from app.core.background import spawn from app.core.database import get_db from app.core.dependencies import ( check_sensitive_endpoint_rate_limit, require_admin, require_roles, ) from app.core.exceptions import ConflictError, NotFoundError from app.core.rbac import AUDITOR_ROLE, GLOBAL_ADMIN_ROLE from app.core.response import paginated_response, success_response router = APIRouter(prefix="/api/v1/assistant/eval", tags=["assistant-eval"]) #: Read gate — see the module docstring for why this is not ``require_admin_read``. _read_gate = require_roles(GLOBAL_ADMIN_ROLE, AUDITOR_ROLE) def _trace_id(request: Request) -> str: return getattr(request.state, "trace_id", None) or "assistant-eval" async def _audit( db: AsyncSession, request: Request, user: AuthUser, *, action: str, resource_id: str | None, detail: dict[str, Any] | None = None, ) -> None: await AuditService(db).log( trace_id=_trace_id(request), actor_type="user", actor_user_id=user.id, action=action, status="success", resource_type="assistant_eval", resource_id=resource_id, detail=detail, ) def _raise(exc: EvalRunError) -> NoReturn: """Map a domain refusal onto the envelope the frontend already handles.""" if exc.code.endswith("NOT_FOUND"): raise NotFoundError(str(exc)) raise ConflictError(str(exc)) async def _get_suite(db: AsyncSession, suite_id: uuid.UUID) -> AssistantEvalSuite: suite = await db.get(AssistantEvalSuite, suite_id) if suite is None: raise NotFoundError("题库不存在。") return suite async def _get_case(db: AsyncSession, case_id: uuid.UUID) -> AssistantEvalCase: case = await db.get(AssistantEvalCase, case_id) if case is None: raise NotFoundError("用例不存在。") return case async def _get_run(db: AsyncSession, run_id: uuid.UUID) -> AssistantEvalRun: run = await db.get(AssistantEvalRun, run_id) if run is None: raise NotFoundError("评测运行不存在。") return run # ── overview / trend ──────────────────────────────────────────────── @router.get("/overview", summary="Latest state of every evaluation suite") async def overview( request: Request, _user: AuthUser = Depends(_read_gate), db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: """Suites, their latest terminal run, and whether one is in flight now.""" from app.domains.support.copilot import active_llm_identity, copilot_available suites = await service.list_suites(db) runs, _ = await service.list_runs(db, page=1, page_size=200) latest: dict[str, dict[str, Any]] = {} running: dict[str, dict[str, Any]] = {} for run in runs: if run["status"] == RUN_RUNNING: running.setdefault(run["suite_id"], run) elif run["pass_rate"] is not None: latest.setdefault(run["suite_id"], run) for suite in suites: suite["latest_run"] = latest.get(suite["id"]) suite["running_run"] = running.get(suite["id"]) provider, model = await active_llm_identity(db) return success_response( data={ "suites": suites, "assistant_available": await copilot_available(db), "provider": provider, "model": model, "dimensions": list(JUDGE_DIMENSIONS), "categories": list(CATEGORIES), "failure_types": list(FAILURE_TYPES), "origins": list(ORIGINS), "langs": list(LANGS), "defaults": { "repeats": DEFAULT_REPEATS, "recommended_repeats": RECOMMENDED_REPEATS, "max_repeats": MAX_REPEATS, "concurrency": DEFAULT_CONCURRENCY, "max_concurrency": MAX_CONCURRENCY, }, }, request=request, ) @router.get("/trend", summary="Pass-rate trend for one suite") async def trend( request: Request, suite_id: uuid.UUID = Query(...), limit: int = Query(default=20, ge=1, le=100), _user: AuthUser = Depends(_read_gate), db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: return success_response( data=await service.trend(db, suite_id=suite_id, limit=limit), request=request ) # ── suites ────────────────────────────────────────────────────────── @router.get("/suites", summary="Every evaluation suite") async def list_suites( request: Request, _user: AuthUser = Depends(_read_gate), db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: return success_response(data=await service.list_suites(db), request=request) class SuiteCreateBody(BaseModel): slug: str = Field(min_length=2, max_length=64, pattern=r"^[a-z][a-z0-9_-]*$") name: str = Field(min_length=1, max_length=128) description: str | None = Field(default=None, max_length=4000) default_persona_person_id: uuid.UUID | None = None @router.post("/suites", status_code=201, summary="Create a suite") async def create_suite( body: SuiteCreateBody, request: Request, user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: clash = ( await db.execute( select(AssistantEvalSuite.id).where(AssistantEvalSuite.slug == body.slug) ) ).first() if clash: raise ConflictError(f"题库标识 {body.slug} 已存在。") suite = AssistantEvalSuite( slug=body.slug, name=body.name, description=body.description, status=SUITE_ACTIVE, is_builtin=False, default_persona_person_id=body.default_persona_person_id, created_by_person_id=user.id, ) db.add(suite) await db.flush() await _audit( db, request, user, action="assistant_eval.suite_created", resource_id=str(suite.id), detail={"slug": body.slug}, ) await db.commit() return success_response( data={"id": str(suite.id), "slug": suite.slug}, request=request ) # NOTE: must stay registered BEFORE ``PATCH /suites/{suite_id}`` — FastAPI # matches in registration order and the path-param route would otherwise # swallow ``seed`` as a (non-UUID) suite id. @router.post("/suites/seed", summary="Create any missing built-in suites and cases") async def seed_suites( request: Request, user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: result = await service.seed_builtin_suites(db) await _audit( db, request, user, action="assistant_eval.seeded", resource_id=None, detail=result, ) await db.commit() return success_response(data=result, request=request) class SuiteUpdateBody(BaseModel): name: str | None = Field(default=None, min_length=1, max_length=128) description: str | None = Field(default=None, max_length=4000) status: str | None = None default_persona_person_id: uuid.UUID | None = None #: Present-but-null cannot be distinguished from absent in pydantic without #: this. Clearing a pinned persona is a real operation and the UI needs it. clear_persona: bool = False @router.patch("/suites/{suite_id}", summary="Edit a suite") async def update_suite( suite_id: uuid.UUID, body: SuiteUpdateBody, request: Request, user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: suite = await _get_suite(db, suite_id) if body.name: suite.name = body.name if body.description is not None: suite.description = body.description if body.status in SUITE_STATUSES: suite.status = body.status if body.clear_persona: suite.default_persona_person_id = None elif body.default_persona_person_id is not None: suite.default_persona_person_id = body.default_persona_person_id await _audit( db, request, user, action="assistant_eval.suite_updated", resource_id=str(suite.id), detail={"status": suite.status}, ) await db.commit() return success_response(data={"id": str(suite.id)}, request=request) # ── cases ─────────────────────────────────────────────────────────── @router.get("/cases", summary="Question bank") async def list_cases( request: Request, suite_id: uuid.UUID | None = Query(default=None), status: str | None = Query(default=None), category: str | None = Query(default=None), origin: str | None = Query(default=None), q: str | None = Query(default=None, max_length=200), page: int = Query(default=1, ge=1), page_size: int = Query(default=50, ge=1, le=200), _user: AuthUser = Depends(_read_gate), db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: rows, total = await service.list_cases( db, suite_id=suite_id, status=status, category=category, origin=origin, q=q, page=page, page_size=page_size, ) return paginated_response(rows, total, page, page_size, request) class TurnBody(BaseModel): role: str = Field(pattern="^(user|assistant)$") content: str = Field(min_length=1, max_length=4000) class CaseCreateBody(BaseModel): suite_id: uuid.UUID question: str = Field(min_length=1, max_length=4000) category: str lang: str = "zh" setup_turns: list[TurnBody] | None = None expectations: dict[str, Any] | None = None persona_person_id: uuid.UUID | None = None status: str = "active" # NOTE: registered before ``PATCH /cases/{case_id}`` for the same reason the # suite seed route is — ``mine`` and ``bulk-status`` are not uuids. @router.post("/cases/mine", summary="Mine draft cases out of production traffic") async def mine_cases( request: Request, suite_id: uuid.UUID = Query(...), days: int = Query(default=30, ge=1, le=365), limit: int = Query(default=50, ge=1, le=200), _rate: None = Depends(check_sensitive_endpoint_rate_limit), user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: """Every mined case lands as ``draft``; a person promotes it. The questions are real employees' words. They are redacted on the way in (mail addresses, phone numbers, ids, credential-shaped tokens), but the redaction is a floor — the review is where somebody reads them. """ await _get_suite(db, suite_id) result = await service.mine_candidates( db, suite_id=suite_id, days=days, limit=limit, actor_person_id=user.id ) await _audit( db, request, user, action="assistant_eval.cases_mined", resource_id=str(suite_id), detail=result, ) await db.commit() return success_response(data=result, request=request) class BulkStatusBody(BaseModel): case_ids: list[uuid.UUID] = Field(min_length=1, max_length=500) status: str @router.post("/cases/bulk-status", summary="Promote or retire many cases at once") async def bulk_status( body: BulkStatusBody, request: Request, user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: if body.status not in CASE_STATUSES: raise ConflictError(f"未知的用例状态:{body.status}") rows = ( ( await db.execute( select(AssistantEvalCase).where(AssistantEvalCase.id.in_(body.case_ids)) ) ) .scalars() .all() ) for case in rows: await service.update_case( db, case=case, actor_person_id=user.id, status=body.status ) await _audit( db, request, user, action="assistant_eval.cases_bulk_status", resource_id=None, detail={"status": body.status, "count": len(rows)}, ) await db.commit() return success_response( data={"updated": len(rows), "status": body.status}, request=request ) @router.post("/cases", status_code=201, summary="Add a case") async def create_case( body: CaseCreateBody, request: Request, user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: try: case = await service.create_case( db, suite_id=body.suite_id, question=body.question, category=body.category, lang=body.lang, setup_turns=[turn.model_dump() for turn in (body.setup_turns or [])], expectations=body.expectations, persona_person_id=body.persona_person_id, status=body.status, actor_person_id=user.id, ) except EvalRunError as exc: _raise(exc) await _audit( db, request, user, action="assistant_eval.case_created", resource_id=str(case.id), detail={"suite_id": str(body.suite_id), "category": case.category}, ) await db.commit() return success_response(data=service.serialize_case(case), request=request) class CaseUpdateBody(BaseModel): question: str | None = Field(default=None, min_length=1, max_length=4000) category: str | None = None lang: str | None = None setup_turns: list[TurnBody] | None = None expectations: dict[str, Any] | None = None persona_person_id: uuid.UUID | None = None clear_persona: bool = False status: str | None = None @router.patch("/cases/{case_id}", summary="Edit a case") async def update_case( case_id: uuid.UUID, body: CaseUpdateBody, request: Request, user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: case = await _get_case(db, case_id) fields: dict[str, Any] = {} if body.question is not None: fields["question"] = body.question if body.category is not None: fields["category"] = body.category if body.lang is not None: fields["lang"] = body.lang if body.setup_turns is not None: fields["setup_turns"] = [turn.model_dump() for turn in body.setup_turns] if body.expectations is not None: fields["expectations"] = body.expectations if body.clear_persona: fields["persona_person_id"] = None elif body.persona_person_id is not None: fields["persona_person_id"] = body.persona_person_id if body.status is not None: fields["status"] = body.status await service.update_case(db, case=case, actor_person_id=user.id, **fields) await _audit( db, request, user, action="assistant_eval.case_updated", resource_id=str(case.id), detail={"fields": sorted(fields)}, ) await db.commit() return success_response(data=service.serialize_case(case), request=request) @router.delete("/cases/{case_id}", summary="Delete (or retire) a case") async def delete_case( case_id: uuid.UUID, request: Request, user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: case = await _get_case(db, case_id) result = await service.delete_case(db, case) await _audit( db, request, user, action="assistant_eval.case_deleted", resource_id=str(case_id), detail=result, ) await db.commit() return success_response(data=result, request=request) @router.post("/cases/{case_id}/variants", summary="Generate paraphrases of a case") async def generate_variants( case_id: uuid.UUID, request: Request, count: int = Query(default=4, ge=1, le=8), _rate: None = Depends(check_sensitive_endpoint_rate_limit), user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: case = await _get_case(db, case_id) result = await service.generate_variants( db, case=case, count=count, actor_person_id=user.id ) await _audit( db, request, user, action="assistant_eval.variants_generated", resource_id=str(case_id), detail=result, ) await db.commit() return success_response(data=result, request=request) # ── runs ──────────────────────────────────────────────────────────── @router.get("/runs", summary="Evaluation runs") async def list_runs( request: Request, suite_id: uuid.UUID | None = Query(default=None), page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=100), _user: AuthUser = Depends(_read_gate), db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: rows, total = await service.list_runs( db, suite_id=suite_id, page=page, page_size=page_size ) return paginated_response(rows, total, page, page_size, request) class RunCreateBody(BaseModel): suite_id: uuid.UUID repeats: int = Field(default=DEFAULT_REPEATS, ge=1, le=MAX_REPEATS) concurrency: int = Field(default=DEFAULT_CONCURRENCY, ge=1, le=MAX_CONCURRENCY) @router.post("/runs", status_code=201, summary="Start an evaluation run") async def start_run( body: RunCreateBody, request: Request, _rate: None = Depends(check_sensitive_endpoint_rate_limit), user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: """Create the run, then execute it in the background and return at once. A run is minutes to hours of provider calls; holding the request open for it would only ever end as an nginx timeout over work that was still running. The row is committed first, so the console can follow it from the moment this returns, and execution is resumable if this process goes away. """ try: run = await service.start_run( db, suite_id=body.suite_id, repeats=body.repeats, concurrency=body.concurrency, trigger=TRIGGER_MANUAL, actor_person_id=user.id, trace_id=_trace_id(request), ) except EvalRunError as exc: _raise(exc) await _audit( db, request, user, action="assistant_eval.run_started", resource_id=str(run.id), detail={ "suite_id": str(body.suite_id), "repeats": run.repeats, "concurrency": run.concurrency, "cases": run.cases_total, }, ) await db.commit() run_id = run.id spawn(execute_run(run_id), name="assistant.eval_run") return success_response( data={"id": str(run_id), "status": RUN_RUNNING}, request=request ) @router.get("/runs/{run_id}", summary="One run") async def get_run( run_id: uuid.UUID, request: Request, _user: AuthUser = Depends(_read_gate), db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: run = await _get_run(db, run_id) suite = await db.get(AssistantEvalSuite, run.suite_id) return success_response( data=service.serialize_run(run, suite_name=suite.name if suite else None), request=request, ) @router.get("/runs/{run_id}/results", summary="Attempts within a run") async def list_results( run_id: uuid.UUID, request: Request, verdict: str | None = Query(default=None), category: str | None = Query(default=None), failure_type: str | None = Query(default=None), page: int = Query(default=1, ge=1), page_size: int = Query(default=50, ge=1, le=200), _user: AuthUser = Depends(_read_gate), db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: await _get_run(db, run_id) rows, total = await service.list_results( db, run_id=run_id, verdict=verdict if verdict in VERDICTS else None, category=category if category in CATEGORIES else None, failure_type=failure_type if failure_type in FAILURE_TYPES else None, page=page, page_size=page_size, ) return paginated_response(rows, total, page, page_size, request) @router.get( "/runs/{run_id}/results/{result_id}", summary="One attempt, in full" ) async def get_result( run_id: uuid.UUID, result_id: uuid.UUID, request: Request, _user: AuthUser = Depends(_read_gate), db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: result = await db.get(AssistantEvalResult, result_id) if result is None or result.run_id != run_id: raise NotFoundError("这次尝试不存在。") return success_response( data=service.serialize_result(result, full=True), request=request ) @router.get("/runs/{run_id}/compare", summary="This run against the previous one") async def compare( run_id: uuid.UUID, request: Request, _user: AuthUser = Depends(_read_gate), db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: run = await _get_run(db, run_id) baseline = await service.previous_terminal_run(db, run=run) if baseline is None: return success_response( data={"comparable": False, "reason": "no_baseline"}, request=request ) return success_response( data=await service.compare_runs(db, run=run, baseline=baseline), request=request, ) @router.post("/runs/{run_id}/resume", summary="Continue an interrupted run") async def resume_run( run_id: uuid.UUID, request: Request, _rate: None = Depends(check_sensitive_endpoint_rate_limit), user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: """Put a partial run back into flight; attempts already scored are kept. ⚠️ A whitelist, not "anything that is not running". A ``cancelled`` run is one an operator deliberately stopped — usually to stop provider spend — and resuming it silently spends exactly what they stopped. A ``succeeded`` run has nothing pending, so resuming it only clears its notes and re-runs the aggregation. Only the two statuses that mean 「有活没干完」 may be resumed. """ run = await _get_run(db, run_id) if run.status == RUN_RUNNING: raise ConflictError("这次运行还在跑。") if run.status not in RESUMABLE_STATUSES: raise ConflictError( f"状态为 {run.status} 的运行不能续跑 —— 只有中断和失败的可以。" ) run.status = RUN_RUNNING run.finished_at = None run.notes = None await _audit( db, request, user, action="assistant_eval.run_resumed", resource_id=str(run.id), detail={"attempts_done": run.attempts_total}, ) await db.commit() spawn(execute_run(run_id), name="assistant.eval_run") return success_response( data={"id": str(run_id), "status": RUN_RUNNING}, request=request ) @router.post("/runs/{run_id}/cancel", summary="Stop a running evaluation") async def cancel( run_id: uuid.UUID, request: Request, user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: run = await _get_run(db, run_id) if run.status != RUN_RUNNING: raise ConflictError("这次运行已经结束了。") await cancel_run(db, run) await _audit( db, request, user, action="assistant_eval.run_cancelled", resource_id=str(run.id), detail={"attempts_done": run.attempts_total}, ) await db.commit() return success_response( data=service.serialize_run(run), request=request ) class PromoteBody(BaseModel): suite_id: uuid.UUID @router.post( "/runs/{run_id}/promote-failures", summary="Turn this run's failures into draft cases", ) async def promote_failures( run_id: uuid.UUID, body: PromoteBody, request: Request, user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: """A failure becomes a permanent case, so the fix has something to prove. Without this the loop stops at "we found a problem": the question that exposed it is thrown away, and the same defect can come back later with nothing watching for it. """ await _get_run(db, run_id) await _get_suite(db, body.suite_id) result = await service.promote_failures( db, run_id=run_id, suite_id=body.suite_id, actor_person_id=user.id ) await _audit( db, request, user, action="assistant_eval.failures_promoted", resource_id=str(run_id), detail=result, ) await db.commit() return success_response(data=result, request=request) __all__ = ["router"]