deerflow-code/offline-backend-20260512/backend/app/gateway/routers/parallel_agent_tasks.py

"""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")