busya
d2b53c2a79
Core implementation of the v2 LLM dashboard validation pipeline: - LLM plugin with Path A (screenshots) and Path B (logs-only) execution - Validation task management (CRUD, schedule, run) - WebSocket task progress with Python 3.13 asyncio fix - Cross-task runs listing (GET /validation-tasks/runs/all) - RecordResponse schema for validation records - JSON prompt helper, per-dashboard status aggregation - Prompt templates with docs/git-commit/validation presets - Migration: v2 validation models + description column - Tests: plugin persistence, prompt templates, batch, payload, url
381 lines
14 KiB
Python
Executable File
381 lines
14 KiB
Python
Executable File
# #region TasksRouter [C:3] [TYPE Module] [SEMANTICS fastapi, task, api, search, create-task-request, resolve-task-request]
|
|
# @BRIEF Defines the FastAPI router for task-related endpoints, allowing clients to create, list, and get the status of tasks.
|
|
# @LAYER API
|
|
# @RELATION DEPENDS_ON -> [TaskManager]
|
|
# @RELATION DEPENDS_ON -> [ConfigManager]
|
|
# @RELATION DEPENDS_ON -> [LLMProviderService]
|
|
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
|
from pydantic import BaseModel
|
|
|
|
from ...core.config_manager import ConfigManager
|
|
from ...core.logger import belief_scope
|
|
from ...core.task_manager import Task, TaskManager, TaskStatus
|
|
from ...core.task_manager.models import LogFilter, LogStats
|
|
from ...dependencies import (
|
|
get_config_manager,
|
|
get_current_user,
|
|
get_task_manager,
|
|
has_permission,
|
|
)
|
|
from ...services.llm_prompt_templates import (
|
|
normalize_llm_settings,
|
|
resolve_bound_provider_id,
|
|
)
|
|
|
|
router = APIRouter()
|
|
|
|
TASK_TYPE_PLUGIN_MAP = {
|
|
"llm_validation": ["llm_dashboard_validation"],
|
|
"backup": ["superset-backup"],
|
|
"migration": ["superset-migration"],
|
|
}
|
|
|
|
|
|
class CreateTaskRequest(BaseModel):
|
|
plugin_id: str
|
|
params: dict[str, Any]
|
|
|
|
|
|
class ResolveTaskRequest(BaseModel):
|
|
resolution_params: dict[str, Any]
|
|
|
|
|
|
class ResumeTaskRequest(BaseModel):
|
|
passwords: dict[str, str]
|
|
|
|
|
|
# #region create_task [C:3] [TYPE Function]
|
|
# @BRIEF Create and start a new task for a given plugin.
|
|
# @PRE plugin_id must exist and params must be valid for that plugin.
|
|
# @POST A new task is created and started.
|
|
# @RELATION CALLS -> [TaskManager]
|
|
# @RELATION DEPENDS_ON -> [ConfigManager]
|
|
# @RELATION DEPENDS_ON -> [LLMProviderService]
|
|
@router.post("", response_model=Task, status_code=status.HTTP_201_CREATED)
|
|
async def create_task(
|
|
request: CreateTaskRequest,
|
|
task_manager: TaskManager = Depends(get_task_manager),
|
|
current_user=Depends(get_current_user),
|
|
config_manager: ConfigManager = Depends(get_config_manager),
|
|
):
|
|
# Deprecation guard: llm_dashboard_validation moved to POST /api/validation-tasks
|
|
if request.plugin_id == "llm_dashboard_validation":
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="llm_dashboard_validation is deprecated via POST /api/tasks. "
|
|
"Use POST /api/validation-tasks instead.",
|
|
)
|
|
|
|
# Dynamic permission check based on plugin_id
|
|
has_permission(f"plugin:{request.plugin_id}", "EXECUTE")(current_user)
|
|
with belief_scope("create_task"):
|
|
try:
|
|
# Special handling for LLM tasks to resolve provider config by task binding.
|
|
if request.plugin_id in {"llm_dashboard_validation", "llm_documentation"}:
|
|
from ...core.database import SessionLocal
|
|
from ...services.llm_provider import LLMProviderService
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
llm_service = LLMProviderService(db)
|
|
provider_id = request.params.get("provider_id")
|
|
if not provider_id:
|
|
llm_settings = normalize_llm_settings(
|
|
config_manager.get_config().settings.llm
|
|
)
|
|
binding_key = (
|
|
"dashboard_validation"
|
|
if request.plugin_id == "llm_dashboard_validation"
|
|
else "documentation"
|
|
)
|
|
provider_id = resolve_bound_provider_id(
|
|
llm_settings, binding_key
|
|
)
|
|
if provider_id:
|
|
request.params["provider_id"] = provider_id
|
|
if not provider_id:
|
|
providers = llm_service.get_all_providers()
|
|
active_provider = next(
|
|
(p for p in providers if p.is_active), None
|
|
)
|
|
if active_provider:
|
|
provider_id = active_provider.id
|
|
request.params["provider_id"] = provider_id
|
|
|
|
if provider_id:
|
|
db_provider = llm_service.get_provider(provider_id)
|
|
if not db_provider:
|
|
raise ValueError(f"LLM Provider {provider_id} not found")
|
|
if (
|
|
request.plugin_id == "llm_dashboard_validation"
|
|
and not bool(db_provider.is_multimodal)
|
|
):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail="Selected provider model is not multimodal for dashboard validation",
|
|
)
|
|
finally:
|
|
db.close()
|
|
|
|
task = await task_manager.create_task(
|
|
plugin_id=request.plugin_id, params=request.params
|
|
)
|
|
return task
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e))
|
|
|
|
|
|
# #endregion create_task
|
|
|
|
|
|
# #region list_tasks [C:2] [TYPE Function]
|
|
# @BRIEF Retrieve a list of tasks with pagination and optional status filter.
|
|
# @PRE task_manager must be available.
|
|
# @POST Returns a list of tasks.
|
|
# @RELATION CALLS -> [TaskManager]
|
|
# @RELATION BINDS_TO -> [EXT:method:TASK_TYPE_PLUGIN_MAP]
|
|
@router.get("", response_model=list[Task])
|
|
async def list_tasks(
|
|
limit: int = 10,
|
|
offset: int = 0,
|
|
status_filter: TaskStatus | None = Query(None, alias="status"),
|
|
task_type: str | None = Query(
|
|
None, description="Task category: llm_validation, backup, migration"
|
|
),
|
|
plugin_id: list[str] | None = Query(
|
|
None, description="Filter by plugin_id (repeatable query param)"
|
|
),
|
|
completed_only: bool = Query(
|
|
False, description="Return only completed tasks (SUCCESS/FAILED)"
|
|
),
|
|
task_manager: TaskManager = Depends(get_task_manager),
|
|
_=Depends(has_permission("tasks", "READ")),
|
|
):
|
|
with belief_scope("list_tasks"):
|
|
plugin_filters = list(plugin_id) if plugin_id else []
|
|
if task_type:
|
|
if task_type not in TASK_TYPE_PLUGIN_MAP:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Unsupported task_type '{task_type}'. Allowed: {', '.join(TASK_TYPE_PLUGIN_MAP.keys())}",
|
|
)
|
|
plugin_filters.extend(TASK_TYPE_PLUGIN_MAP[task_type])
|
|
|
|
return task_manager.get_tasks(
|
|
limit=limit,
|
|
offset=offset,
|
|
status=status_filter,
|
|
plugin_ids=plugin_filters or None,
|
|
completed_only=completed_only,
|
|
)
|
|
|
|
|
|
# #endregion list_tasks
|
|
|
|
|
|
# #region get_task [C:2] [TYPE Function]
|
|
# @BRIEF Retrieve the details of a specific task.
|
|
# @PRE task_id must exist.
|
|
# @POST Returns task details or raises 404.
|
|
# @RELATION CALLS -> [TaskManager]
|
|
@router.get("/{task_id}", response_model=Task)
|
|
async def get_task(
|
|
task_id: str,
|
|
task_manager: TaskManager = Depends(get_task_manager),
|
|
_=Depends(has_permission("tasks", "READ")),
|
|
):
|
|
with belief_scope("get_task"):
|
|
task = task_manager.get_task(task_id)
|
|
if not task:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND, detail="Task not found"
|
|
)
|
|
return task
|
|
|
|
|
|
# #endregion get_task
|
|
|
|
|
|
# #region get_task_logs [C:5] [TYPE Function]
|
|
# @BRIEF Retrieve logs for a specific task with optional filtering.
|
|
# @PRE task_id must exist.
|
|
# @POST Returns a list of log entries or raises 404.
|
|
# @RELATION CALLS -> [TaskManager]
|
|
# @RELATION DEPENDS_ON -> [LogFilter]
|
|
# @TEST_CONTRACT TaskLogQueryInput -> List[LogEntry]
|
|
# @TEST_SCENARIO existing_task_logs_filtered -> Returns filtered logs by level/source/search with pagination.
|
|
# @TEST_FIXTURE valid_task_with_mixed_logs -> backend/tests/fixtures/task_logs/valid_task_with_mixed_logs.json
|
|
# @TEST_EDGE missing_task -> Unknown task_id returns 404 Task not found.
|
|
# @TEST_EDGE invalid_level_type -> Non-string/invalid level query rejected by validation or yields empty result.
|
|
# @TEST_EDGE pagination_bounds -> offset=0 and limit=1000 remain within API bounds and do not overflow.
|
|
# @TEST_INVARIANT logs_only_for_existing_task -> VERIFIED_BY: [existing_task_logs_filtered, missing_task]
|
|
@router.get("/{task_id}/logs")
|
|
async def get_task_logs(
|
|
task_id: str,
|
|
level: str | None = Query(
|
|
None, description="Filter by log level (DEBUG, INFO, WARNING, ERROR)"
|
|
),
|
|
source: str | None = Query(None, description="Filter by source component"),
|
|
search: str | None = Query(None, description="Text search in message"),
|
|
offset: int = Query(0, ge=0, description="Number of logs to skip"),
|
|
limit: int = Query(
|
|
100, ge=1, le=1000, description="Maximum number of logs to return"
|
|
),
|
|
task_manager: TaskManager = Depends(get_task_manager),
|
|
):
|
|
with belief_scope("get_task_logs"):
|
|
task = task_manager.get_task(task_id)
|
|
if not task:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND, detail="Task not found"
|
|
)
|
|
|
|
log_filter = LogFilter(
|
|
level=level.upper() if level else None,
|
|
source=source,
|
|
search=search,
|
|
offset=offset,
|
|
limit=limit,
|
|
)
|
|
return task_manager.get_task_logs(task_id, log_filter)
|
|
|
|
|
|
# #endregion get_task_logs
|
|
|
|
|
|
# #region get_task_log_stats [C:2] [TYPE Function]
|
|
# @BRIEF Get statistics about logs for a task (counts by level and source).
|
|
# @PRE task_id must exist.
|
|
# @POST Returns log statistics or raises 404.
|
|
# @RELATION CALLS -> [TaskManager]
|
|
# @RELATION DEPENDS_ON -> [LogStats]
|
|
@router.get("/{task_id}/logs/stats", response_model=LogStats)
|
|
async def get_task_log_stats(
|
|
task_id: str,
|
|
task_manager: TaskManager = Depends(get_task_manager),
|
|
):
|
|
with belief_scope("get_task_log_stats"):
|
|
task = task_manager.get_task(task_id)
|
|
if not task:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND, detail="Task not found"
|
|
)
|
|
stats_payload = task_manager.get_task_log_stats(task_id)
|
|
if isinstance(stats_payload, LogStats):
|
|
return stats_payload
|
|
if isinstance(stats_payload, dict) and (
|
|
"total_count" in stats_payload
|
|
or "by_level" in stats_payload
|
|
or "by_source" in stats_payload
|
|
):
|
|
return LogStats(
|
|
total_count=int(stats_payload.get("total_count", 0) or 0),
|
|
by_level=dict(stats_payload.get("by_level") or {}),
|
|
by_source=dict(stats_payload.get("by_source") or {}),
|
|
)
|
|
flat_by_level = (
|
|
dict(stats_payload or {}) if isinstance(stats_payload, dict) else {}
|
|
)
|
|
return LogStats(
|
|
total_count=sum(int(value or 0) for value in flat_by_level.values()),
|
|
by_level={
|
|
str(key): int(value or 0) for key, value in flat_by_level.items()
|
|
},
|
|
by_source={},
|
|
)
|
|
|
|
|
|
# #endregion get_task_log_stats
|
|
|
|
|
|
# #region get_task_log_sources [C:2] [TYPE Function]
|
|
# @BRIEF Get unique sources for a task's logs.
|
|
# @PRE task_id must exist.
|
|
# @POST Returns list of unique source names or raises 404.
|
|
# @RELATION CALLS -> [TaskManager]
|
|
@router.get("/{task_id}/logs/sources", response_model=list[str])
|
|
async def get_task_log_sources(
|
|
task_id: str,
|
|
task_manager: TaskManager = Depends(get_task_manager),
|
|
):
|
|
with belief_scope("get_task_log_sources"):
|
|
task = task_manager.get_task(task_id)
|
|
if not task:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND, detail="Task not found"
|
|
)
|
|
return task_manager.get_task_log_sources(task_id)
|
|
|
|
|
|
# #endregion get_task_log_sources
|
|
|
|
|
|
# #region resolve_task [C:2] [TYPE Function]
|
|
# @BRIEF Resolve a task that is awaiting mapping.
|
|
# @PRE task must be in AWAITING_MAPPING status.
|
|
# @POST Task is resolved and resumes execution.
|
|
# @RELATION CALLS -> [TaskManager]
|
|
@router.post("/{task_id}/resolve", response_model=Task)
|
|
async def resolve_task(
|
|
task_id: str,
|
|
request: ResolveTaskRequest,
|
|
task_manager: TaskManager = Depends(get_task_manager),
|
|
_=Depends(has_permission("tasks", "WRITE")),
|
|
):
|
|
with belief_scope("resolve_task"):
|
|
try:
|
|
await task_manager.resolve_task(task_id, request.resolution_params)
|
|
return task_manager.get_task(task_id)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
|
|
|
|
|
|
# #endregion resolve_task
|
|
|
|
|
|
# #region resume_task [C:2] [TYPE Function]
|
|
# @BRIEF Resume a task that is awaiting input (e.g., passwords).
|
|
# @PRE task must be in AWAITING_INPUT status.
|
|
# @POST Task resumes execution with provided input.
|
|
# @RELATION CALLS -> [TaskManager]
|
|
@router.post("/{task_id}/resume", response_model=Task)
|
|
async def resume_task(
|
|
task_id: str,
|
|
request: ResumeTaskRequest,
|
|
task_manager: TaskManager = Depends(get_task_manager),
|
|
_=Depends(has_permission("tasks", "WRITE")),
|
|
):
|
|
with belief_scope("resume_task"):
|
|
try:
|
|
task_manager.resume_task_with_password(task_id, request.passwords)
|
|
return task_manager.get_task(task_id)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
|
|
|
|
|
|
# #endregion resume_task
|
|
|
|
|
|
# #region clear_tasks [C:2] [TYPE Function]
|
|
# @BRIEF Clear tasks matching the status filter.
|
|
# @PRE task_manager is available.
|
|
# @POST Tasks are removed from memory/persistence.
|
|
# @RELATION CALLS -> [TaskManager]
|
|
@router.delete("", status_code=status.HTTP_204_NO_CONTENT)
|
|
async def clear_tasks(
|
|
status: TaskStatus | None = None,
|
|
task_manager: TaskManager = Depends(get_task_manager),
|
|
_=Depends(has_permission("tasks", "WRITE")),
|
|
):
|
|
with belief_scope("clear_tasks", f"status={status}"):
|
|
task_manager.clear_tasks(status)
|
|
return
|
|
|
|
|
|
# #endregion clear_tasks
|
|
|
|
# #endregion TasksRouter
|