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

"""CRUD API for roundtable-planning business chains (业务链条).

A business chain is a named, reusable, *ordered* list of agents the user
assembles in the chain editor. In Step 2 the coordinator dispatches to the seats
following this order. Bound to the requesting user (not the browser), so it
follows the account across devices/sessions — same model as roundtable drafts.

Routes (prefix ``/api/roundtable-chains``):
    GET    ""               list current user's chains (full records, incl. seats)
    POST   ""               create a chain
    GET    "/{chain_id}"    one chain
    PUT    "/{chain_id}"    partial update (title / description / seats)
    DELETE "/{chain_id}"    delete
"""

from __future__ import annotations

import logging
from datetime import datetime
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/roundtable-chains", tags=["roundtable-chains"])

# Seat-count bounds mirror the roundtable selection bounds (2 ≤ N ≤ 8).
MIN_SEATS = 2
MAX_SEATS = 8


# ── schemas ────────────────────────────────────────────────────────────────


class ChainSeatSchema(BaseModel):
    agent_id: str = Field(min_length=1, max_length=128)
    name: str = ""
    description: str = ""


class ChainResponse(BaseModel):
    id: str
    title: str = ""
    description: str | None = None
    seats: list[ChainSeatSchema] = []
    # Optional layered orchestration: list of stages, each a list of agent_ids
    # run in parallel (stages run serially). None/absent = linear chain.
    stages: list[list[str]] | None = None
    # Optional per-stage goals aligned to ``stages`` (dag-only). None = no goals.
    stage_goals: list[str | None] | None = None
    # 写给总控的「编排提示」（dag-only）：告诉总控怎么派活、做什么。None/空 = 不注入。
    coordinator_prompt: str | None = None
    # 发布状态：True = 公共（所有账号可在「公共」筛选查看 / 选用）。
    is_public: bool = False
    # 是否当前用户所有（仅本人可编辑 / 删除 / 发布；公共筛选里别人的链条 is_mine=False）。
    is_mine: bool = True
    created_at: datetime | str | None = None
    updated_at: datetime | str | None = None


class ChainListResponse(BaseModel):
    chains: list[ChainResponse]


class ChainCreateRequest(BaseModel):
    # Optional client-provided id so the browser can keep one stable id across
    # the create + subsequent updates without an extra round-trip.
    id: str | None = Field(default=None, max_length=64)
    title: str = Field(default="", max_length=512)
    description: str | None = Field(default=None, max_length=2048)
    seats: list[ChainSeatSchema] = []
    stages: list[list[str]] | None = None
    stage_goals: list[str | None] | None = None
    coordinator_prompt: str | None = Field(default=None, max_length=4000)


class ChainUpdateRequest(BaseModel):
    # All optional → callers send only what changed.
    title: str | None = Field(default=None, max_length=512)
    description: str | None = Field(default=None, max_length=2048)
    seats: list[ChainSeatSchema] | None = None
    stages: list[list[str]] | None = None
    stage_goals: list[str | None] | None = None
    coordinator_prompt: str | None = Field(default=None, max_length=4000)
    # 发布开关（True 发布为公共 / False 取消发布）；仅 owner 可改（update 本就 owner 限定）。
    is_public: bool | None = None


# ── 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, "roundtable_chain_store", None)
    if store is None:
        raise HTTPException(status_code=503, detail="Roundtable chain store not available")
    return store


def _validate_seats(seats: list[ChainSeatSchema]) -> None:
    """Enforce 2 ≤ N ≤ 8 and reject duplicate agent ids within a chain."""
    if len(seats) < MIN_SEATS or len(seats) > MAX_SEATS:
        raise HTTPException(
            status_code=400,
            detail=f"业务链条席位数须在 {MIN_SEATS}–{MAX_SEATS} 之间（当前 {len(seats)}）",
        )
    seen: set[str] = set()
    for s in seats:
        if s.agent_id in seen:
            raise HTTPException(status_code=400, detail=f"业务链条存在重复智能体: {s.agent_id}")
        seen.add(s.agent_id)


def _validate_stages(seats: list[ChainSeatSchema], stages: list[list[str]] | None) -> None:
    """Validate the optional layering. ``stages`` must partition the seats.

    Rules (None = linear chain, always valid):
      - at least one stage, no empty stage;
      - every agent_id in a stage must be one of the chain's seats;
      - no agent_id appears twice (within or across stages);
      - every seat must appear in exactly one stage (full coverage), so the
        stored layering and the seat list never drift apart.
    """
    if stages is None:
        return
    if len(stages) == 0:
        raise HTTPException(status_code=400, detail="编排分层不能为空（至少一个阶段）")
    seat_ids = {s.agent_id for s in seats}
    seen: set[str] = set()
    for idx, stage in enumerate(stages):
        if not stage:
            raise HTTPException(status_code=400, detail=f"第 {idx + 1} 阶段为空，请删除或放入席位")
        for agent_id in stage:
            if agent_id not in seat_ids:
                raise HTTPException(
                    status_code=400,
                    detail=f"编排分层包含不在席位列表中的智能体: {agent_id}",
                )
            if agent_id in seen:
                raise HTTPException(
                    status_code=400,
                    detail=f"智能体 {agent_id} 在编排分层中重复出现，每个席位只能跑一次",
                )
            seen.add(agent_id)
    if seen != seat_ids:
        missing = seat_ids - seen
        raise HTTPException(
            status_code=400,
            detail=f"编排分层未覆盖全部席位，缺少: {', '.join(sorted(missing))}",
        )


# Per-goal length cap to keep injected prompts bounded.
MAX_STAGE_GOAL_CHARS = 1000


def _validate_stage_goals(
    stages: list[list[str]] | None, stage_goals: list[str | None] | None
) -> None:
    """Validate the optional per-stage goals (None = no goals, always valid).

    Rules:
      - goals only make sense alongside a non-null layering (dag-only): reject
        goals sent without ``stages``;
      - length must match ``stages`` exactly (positional alignment);
      - each goal (when present) must be ≤ ``MAX_STAGE_GOAL_CHARS`` chars.
    """
    if stage_goals is None:
        return
    if stages is None:
        raise HTTPException(status_code=400, detail="阶段目标需与编排分层一起提交（线性链不支持阶段目标）")
    if len(stage_goals) != len(stages):
        raise HTTPException(
            status_code=400,
            detail=f"阶段目标数量（{len(stage_goals)}）与编排分层阶段数（{len(stages)}）不一致",
        )
    for idx, goal in enumerate(stage_goals):
        if goal is not None and len(goal) > MAX_STAGE_GOAL_CHARS:
            raise HTTPException(
                status_code=400,
                detail=f"第 {idx + 1} 阶段目标过长（上限 {MAX_STAGE_GOAL_CHARS} 字）",
            )


# ── routes ───────────────────────────────────────────────────────────────


@router.get("", response_model=ChainListResponse)
async def list_chains(request: Request, scope: str = "mine") -> ChainListResponse:
    """列出业务链条。

    ``scope`` 查询参数(筛选):
      - ``mine``（默认）：仅当前用户自己的链条；
      - ``public``：所有**已发布**的公共链条（任何账号均可查看 / 选入研讨）。
    其余值按 ``mine`` 处理。
    """
    normalized_scope = scope if scope in ("mine", "public") else "mine"
    store = _get_store(request)
    user_id = _current_user_id(request)
    rows = await store.list_chains(user_id, scope=normalized_scope)
    return ChainListResponse(chains=[ChainResponse(**r) for r in rows])


@router.post("", response_model=ChainResponse, status_code=201)
async def create_chain(request: Request, body: ChainCreateRequest) -> ChainResponse:
    _validate_seats(body.seats)
    _validate_stages(body.seats, body.stages)
    _validate_stage_goals(body.stages, body.stage_goals)
    store = _get_store(request)
    user_id = _current_user_id(request)
    row = await store.create_chain(
        user_id,
        {
            "id": (body.id or uuid4().hex),
            "title": body.title.strip(),
            "description": body.description,
            "seats": [s.model_dump() for s in body.seats],
            "stages": body.stages,
            "stage_goals": body.stage_goals,
            "coordinator_prompt": body.coordinator_prompt,
        },
    )
    return ChainResponse(**row)


@router.get("/{chain_id}", response_model=ChainResponse)
async def get_chain(request: Request, chain_id: str) -> ChainResponse:
    store = _get_store(request)
    user_id = _current_user_id(request)
    row = await store.get_chain(chain_id, user_id)
    if row is None:
        raise HTTPException(status_code=404, detail="Chain not found")
    return ChainResponse(**row)


@router.put("/{chain_id}", response_model=ChainResponse)
async def update_chain(request: Request, chain_id: str, body: ChainUpdateRequest) -> ChainResponse:
    if body.seats is not None:
        _validate_seats(body.seats)
    # A non-null stages layering can only be validated against seats; the editor
    # always saves both together. Reject a layering update that omits seats.
    if body.stages is not None:
        if body.seats is None:
            raise HTTPException(status_code=400, detail="更新编排分层时必须同时提交席位列表")
        _validate_stages(body.seats, body.stages)
    # Goals align to stages; reject goals sent without a layering, enforce length/size.
    _validate_stage_goals(body.stages, body.stage_goals)
    store = _get_store(request)
    user_id = _current_user_id(request)
    # Distinguish "omitted" from "explicitly sent"; only forward provided fields.
    fields = body.model_dump(exclude_unset=True)
    row = await store.update_chain(chain_id, user_id, **fields)
    if row is None:
        raise HTTPException(status_code=404, detail="Chain not found")
    return ChainResponse(**row)


@router.delete("/{chain_id}", status_code=204)
async def delete_chain(request: Request, chain_id: str) -> None:
    store = _get_store(request)
    user_id = _current_user_id(request)
    deleted = await store.delete_chain(chain_id, user_id)
    if not deleted:
        raise HTTPException(status_code=404, detail="Chain not found")