"""CRUD API for parallel multi-agent task records (并行多智能体任务). A task = one saved run of the standalone parallel multi-agent panel: a task name + brief dispatched to several agents in parallel, plus the resulting message list. Bound to the requesting user (not the browser), so it follows the account across devices/sessions — same model as roundtable drafts. Routes (prefix ``/api/parallel-agent-tasks``): GET "" list current user's tasks (lightweight meta, no messages) POST "" create a task GET "/{task_id}" one task (full, incl. messages) PUT "/{task_id}" partial update (title / brief / agents / messages) DELETE "/{task_id}" delete """ from __future__ import annotations import logging from datetime import datetime from typing import Any from uuid import uuid4 from fastapi import APIRouter, HTTPException, Request from pydantic import BaseModel, Field from deerflow.runtime.user_context import get_effective_user_id logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/parallel-agent-tasks", tags=["parallel-agent-tasks"]) # ── schemas ──────────────────────────────────────────────────────────────── class TaskAgentSchema(BaseModel): agent_id: str = Field(min_length=1, max_length=128) name: str = "" avatarType: str | None = None class TaskResponse(BaseModel): id: str title: str = "" brief: str | None = None agents: list[TaskAgentSchema] = [] # messages is an opaque dialogue list (mirrors the frontend Dialogue[] shape); # absent on list responses (lightweight), present on get. messages: list[dict[str, Any]] | None = None created_at: datetime | str | None = None updated_at: datetime | str | None = None class TaskListResponse(BaseModel): tasks: list[TaskResponse] class TaskCreateRequest(BaseModel): id: str | None = Field(default=None, max_length=64) title: str = Field(default="", max_length=512) brief: str | None = None agents: list[TaskAgentSchema] = [] messages: list[dict[str, Any]] = [] class TaskUpdateRequest(BaseModel): # All optional → callers send only what changed. title: str | None = Field(default=None, max_length=512) brief: str | None = None agents: list[TaskAgentSchema] | None = None messages: list[dict[str, Any]] | None = None class TaskEnsureRequest(BaseModel): """get-or-create:宿主传的 taskId 不固定,新 id 直接**初始化**一条空记录返回(不 404)。""" id: str = Field(min_length=1, max_length=64) title: str = Field(default="", max_length=512) brief: str | None = None agents: list[TaskAgentSchema] = [] # ── helpers ────────────────────────────────────────────────────────────── def _current_user_id(request: Request) -> str: user = getattr(request.state, "user", None) if user is not None: return str(user.id) return get_effective_user_id() def _get_store(request: Request): store = getattr(request.app.state, "parallel_agent_task_store", None) if store is None: raise HTTPException(status_code=503, detail="Parallel agent task store not available") return store # ── routes ─────────────────────────────────────────────────────────────── @router.get("", response_model=TaskListResponse) async def list_tasks(request: Request) -> TaskListResponse: store = _get_store(request) user_id = _current_user_id(request) rows = await store.list_tasks(user_id) return TaskListResponse(tasks=[TaskResponse(**r) for r in rows]) @router.post("", response_model=TaskResponse, status_code=201) async def create_task(request: Request, body: TaskCreateRequest) -> TaskResponse: store = _get_store(request) user_id = _current_user_id(request) row = await store.create_task( user_id, { "id": (body.id or uuid4().hex), "title": body.title.strip(), "brief": body.brief, "agents": [a.model_dump() for a in body.agents], "messages": body.messages, }, ) return TaskResponse(**row) @router.post("/ensure", response_model=TaskResponse) async def ensure_task(request: Request, body: TaskEnsureRequest) -> TaskResponse: """存在则返回(含 messages 供回显);不存在则用该 id **初始化**一条空记录返回。 宿主传的 taskId 不固定:新 id 不再 404,而是即时建好、可直接回显/保存。幂等。 """ store = _get_store(request) user_id = _current_user_id(request) existing = await store.get_task(body.id, user_id) if existing is not None: return TaskResponse(**existing) row = await store.create_task( user_id, { "id": body.id, "title": body.title.strip(), "brief": body.brief, "agents": [a.model_dump() for a in body.agents], "messages": [], }, ) return TaskResponse(**row) @router.get("/{task_id}", response_model=TaskResponse) async def get_task(request: Request, task_id: str) -> TaskResponse: store = _get_store(request) user_id = _current_user_id(request) row = await store.get_task(task_id, user_id) if row is None: raise HTTPException(status_code=404, detail="Task not found") return TaskResponse(**row) @router.put("/{task_id}", response_model=TaskResponse) async def update_task(request: Request, task_id: str, body: TaskUpdateRequest) -> TaskResponse: store = _get_store(request) user_id = _current_user_id(request) fields = body.model_dump(exclude_unset=True) # Normalize agents (pydantic models) to plain dicts for JSON storage. if "agents" in fields and fields["agents"] is not None: fields["agents"] = [a if isinstance(a, dict) else a.model_dump() for a in body.agents or []] row = await store.update_task(task_id, user_id, **fields) if row is None: raise HTTPException(status_code=404, detail="Task not found") return TaskResponse(**row) @router.delete("/{task_id}", status_code=204) async def delete_task(request: Request, task_id: str) -> None: store = _get_store(request) user_id = _current_user_id(request) deleted = await store.delete_task(task_id, user_id) if not deleted: raise HTTPException(status_code=404, detail="Task not found")