feat: restore legacy data and add typed task result views

backend/src/api/routes/tasks.py

@@ -4,7 +4,7 @@
 # @PURPOSE: Defines the FastAPI router for task-related endpoints, allowing clients to create, list, and get the status of tasks.
 # @LAYER: UI (API)
 # @RELATION: Depends on the TaskManager. It is included by the main app.
-from typing import List, Dict, Any, Optional
+from typing import List, Dict, Any, Optional
 from fastapi import APIRouter, Depends, HTTPException, status, Query
 from pydantic import BaseModel
 from ...core.logger import belief_scope
@@ -13,9 +13,15 @@ from ...core.task_manager import TaskManager, Task, TaskStatus, LogEntry
 from ...core.task_manager.models import LogFilter, LogStats
 from ...dependencies import get_task_manager, has_permission, get_current_user
 
-router = APIRouter()
-
-class CreateTaskRequest(BaseModel):
+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]
 
@@ -79,18 +85,36 @@ async def create_task(
 # @PRE: task_manager must be available.
 # @POST: Returns a list of tasks.
 # @RETURN: List[Task] - List of tasks.
-async def list_tasks(
-    limit: int = 10,
-    offset: int = 0,
-    status: Optional[TaskStatus] = None,
-    task_manager: TaskManager = Depends(get_task_manager),
-    _ = Depends(has_permission("tasks", "READ"))
-):
-    """
-    Retrieve a list of tasks with pagination and optional status filter.
-    """
-    with belief_scope("list_tasks"):
-        return task_manager.get_tasks(limit=limit, offset=offset, status=status)
+async def list_tasks(
+    limit: int = 10,
+    offset: int = 0,
+    status_filter: Optional[TaskStatus] = Query(None, alias="status"),
+    task_type: Optional[str] = Query(None, description="Task category: llm_validation, backup, migration"),
+    plugin_id: Optional[List[str]] = 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"))
+):
+    """
+    Retrieve a list of tasks with pagination and optional status filter.
+    """
+    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
+        )
 # [/DEF:list_tasks:Function]
 
 @router.get("/{task_id}", response_model=Task)
@@ -276,4 +300,4 @@ async def clear_tasks(
         task_manager.clear_tasks(status)
         return
 # [/DEF:clear_tasks:Function]
-# [/DEF:TasksRouter:Module]
+# [/DEF:TasksRouter:Module]