"""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 # 发布状态: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 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 # 发布开关(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))}", ) # ── 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) 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, }, ) 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) 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")