Merge branch 'dhk-1' of http://192.168.0.200/knowledge-base-bj/deerflow-server into dhk-1

2026-06-09 14:06:01 +08:00
commit 10795d55e6
@@ -47,6 +47,8 @@ _PUBLIC_EXACT_PATHS: frozenset[str] = frozenset(
        "/api/v1/auth/logout",
        "/api/v1/auth/setup-status",
        "/api/v1/auth/initialize",
+        # 并行多智能体面板专用的「绕过常规登录」取 token 端点（内网部署限制了标准登录时用）。
+        "/api/parallel-agents/auth/token",
    }
 )

@@ -59,6 +59,8 @@ _AUTH_EXEMPT_PATHS: frozenset[str] = frozenset(
        "/api/v1/auth/logout",
        "/api/v1/auth/register",
        "/api/v1/auth/initialize",
+        # 并行多智能体面板专用取 token 端点（公共 + CSRF 豁免）。
+        "/api/parallel-agents/auth/token",
    }
 )

@@ -1,14 +1,16 @@
 """Server-side parallel fan-out for the standalone parallel multi-agent panel.

 The Vue embeddable panel sends a single request — ``{task_name, brief, agent_ids}``
-— and this endpoint **fans out** to every agent in parallel (file writes disabled),
-then **merges** their LangGraph streams into ONE SSE tagged per ``agent_id``, so the
-client only consumes one stream and never parses LangGraph internals.
+— and this endpoint **fans out** to every agent in parallel, then **merges** their
+LangGraph streams into ONE SSE tagged per ``agent_id``, so the client only consumes
+one stream and never parses LangGraph internals.

-It reuses the roundtable single-agent machinery from ``multi_agent``:
-``_special_run(..., no_file=True)`` runs one seat under the ``seat_parallel`` run
-policy (which excludes ``write_file`` / ``str_replace`` / ``bash`` / ``web_search``
-/ ``agent_orchestration`` …), so parallel agents can't write files or dispatch.
+Each fan-out agent runs as a pure **executing seat**: ``excluded_tools`` =
+``_EXCLUDED_TOOLS`` (``write_file`` / ``str_replace`` / ``agent_orchestration`` /
+``bash``) so a seat can neither write files (parallel writes would clash) nor
+dispatch / run scripts (a weak model fed "总控 / 业务链条" framing would otherwise
+mimic the controller and "派活"). Read-only tools + ``ask_clarification`` stay on,
+so the step bar and clarification cards still work.

 Unified event protocol (one JSON object per ``data:`` frame)::

@@ -31,18 +33,24 @@ import json
 import logging
 from typing import Any, AsyncIterator

+import httpx
 from fastapi import APIRouter, Body, HTTPException, Request
 from fastapi.responses import StreamingResponse
+from pydantic import BaseModel, Field

 from app.gateway.routers.multi_agent import (
    _ROUNDTABLE_THREAD_METADATA,
    _auth_headers,
    _create_thread,
+    _flatten_content,
    _loopback_base,
    _make_loopback_client,
-    _special_run,
+    _parse_last_messages,
+    _run_payload,
+    _stream_upstream,
    _validate_agent_id,
 )
+from deerflow.tools.builtins.clarification_utils import resolve_allow_multiple

 logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api/parallel-agents", tags=["parallel-agents"])
@@ -50,6 +58,49 @@ router = APIRouter(prefix="/api/parallel-agents", tags=["parallel-agents"])
 # Soft cap on parallel degree — protects remote model concurrency + sandbox.
 MAX_PARALLEL_AGENTS = 8

+# 禁两类工具：①**写文件**（write_file / str_replace）—— 并行席位互相写文件会打架；
+# ②**派活 / 脚本逃逸**（agent_orchestration / bash）—— 本面板里每个智能体只做自己那
+# 份分析交付，绝不该扮演「总控」去调度别人。弱模型（deepseek 等）拿到带「总控 / 业务
+# 链条」语气的任务后，会概率性地用 agent_orchestration 真去派活，或绕道 bash 跑脚本
+# 调内部接口模仿总控调度（bug：单步单智能体链条里子智能体输出「总控调度指令…」并用
+# glob / bash 翻找派活脚本）。物理禁掉这两类即可根治。其余只读工具（ls / read_file /
+# 检索）与 ask_clarification（协作卡片）保留：既不互相打架，又能正常出步骤条 / 要澄清。
+_EXCLUDED_TOOLS: list[str] = [
+    "write_file",           # 并行不互相写文件打架
+    "str_replace",
+    "agent_orchestration",  # 子智能体不派活 / 不调度（防模仿总控）
+    "bash",                 # 禁跑脚本（防绕道脚本调内部接口模仿派活）
+]
+
+
+# ── 专用「绕过常规登录」取 token ──────────────────────────────────────────────
+# 内网部署对标准 /api/v1/auth/login/username 有额外操作限制（口令门 / 网关裁剪等），
+# 嵌入方拿不到 token。这里给并行多智能体面板单开一个**公共**端点：传用户名直接签发 token
+# （账号不存在则自动注册，与 username 登录同一套逻辑）。已在 auth_middleware / csrf_middleware
+# 的白名单里放行（公共 + CSRF 豁免）。⚠️ 这是有意的鉴权旁路，仅供内网部署的本功能使用。
+
+
+class _TokenRequest(BaseModel):
+    username: str = Field(min_length=1, max_length=128)
+
+
+class _TokenResponse(BaseModel):
+    access_token: str
+    token_type: str = "bearer"
+    user_id: str
+
+
+@router.post("/auth/token", response_model=_TokenResponse)
+async def issue_token(body: _TokenRequest) -> _TokenResponse:
+    """传用户名 → 直接签发 access_token（绕过常规登录，公共端点）。"""
+    # 懒加载，避免与 auth 路由的模块级循环依赖。
+    from app.gateway.auth import create_access_token
+    from app.gateway.routers.auth import _get_or_create_user_by_username
+
+    user, _created = await _get_or_create_user_by_username(body.username.strip())
+    token = create_access_token(str(user.id), token_version=user.token_version)
+    return _TokenResponse(access_token=token, token_type="bearer", user_id=str(user.id))
+

 # ── frame normalisation (mirrors multi-agent.ts) ─────────────────────────────

@@ -75,33 +126,136 @@ def _extract_ai_text(chunk: dict[str, Any]) -> str:
    return ""


-def _extract_tool_steps(chunk: dict[str, Any]) -> list[dict[str, Any]]:
-    """Tool-call steps from a chunk obj — both complete ``tool_calls`` and
-    incremental ``tool_call_chunks`` (we only need the name to surface a step)."""
-    steps: list[dict[str, Any]] = []
-    tool_calls = chunk.get("tool_calls")
-    if isinstance(tool_calls, list):
-        for tc in tool_calls:
-            if not isinstance(tc, dict):
-                continue
-            name = tc.get("name")
-            if not isinstance(name, str) or not name:
-                continue
-            ident = tc.get("id") or name
-            args = tc.get("args") if isinstance(tc.get("args"), dict) else None
-            steps.append({"key": f"tool:{ident}", "label": f"调用工具 {name}", "args": args})
-    tcc = chunk.get("tool_call_chunks")
-    if isinstance(tcc, list):
-        for c in tcc:
-            if not isinstance(c, dict):
-                continue
-            name = c.get("name")
-            if not isinstance(name, str) or not name:
-                continue
-            idx = c.get("index") if isinstance(c.get("index"), int) else 0
-            ident = c.get("id") or f"{name}:{idx}"
-            steps.append({"key": f"tool:{ident}", "label": f"调用工具 {name}", "args": None})
-    return steps
+def _ingest_tool_chunks(chunks: Any, acc: dict[int, dict[str, Any]]) -> None:
+    """累积 ``tool_call_chunks`` 的增量（name + args 字符串），按 index 归并
+    （mirror multi-agent.ts ``ingestToolCallChunks``）。"""
+    if not isinstance(chunks, list):
+        return
+    for c in chunks:
+        if not isinstance(c, dict):
+            continue
+        idx = c.get("index") if isinstance(c.get("index"), int) else 0
+        entry = acc.setdefault(idx, {"id": None, "name": "", "argsStr": ""})
+        cid = c.get("id")
+        if isinstance(cid, str) and cid:
+            if entry["id"] and entry["id"] != cid:  # 同 index 换了新调用 → 重置
+                entry["name"] = ""
+                entry["argsStr"] = ""
+            entry["id"] = cid
+        name = c.get("name")
+        if isinstance(name, str) and name:
+            entry["name"] = name
+        a = c.get("args")
+        if isinstance(a, str):
+            entry["argsStr"] += a
+
+
+def _synthetic_tool_calls(acc: dict[int, dict[str, Any]]) -> list[dict[str, Any]]:
+    """把累积的 argsStr 尽力 parse 成完整 args，产出 ``{id,name,args}``（未成完整 JSON 则跳过）。"""
+    out: list[dict[str, Any]] = []
+    for entry in acc.values():
+        if not entry["name"]:
+            continue
+        s = str(entry["argsStr"]).strip()
+        if not s:
+            continue
+        try:
+            args = json.loads(s)
+        except Exception:
+            continue
+        if isinstance(args, dict):
+            out.append({"id": entry["id"], "name": entry["name"], "args": args})
+    return out
+
+
+def _clarification_from_messages(messages: list[dict[str, Any]]) -> dict[str, Any] | None:
+    """从最终 messages 里找最后一个 ``ask_clarification`` 工具调用，抽出结构化澄清载荷
+    （字段对齐 multi_agent leader 的 clarification 帧）。"""
+    for msg in reversed(messages):
+        if not isinstance(msg, dict):
+            continue
+        if (msg.get("type") or "").lower() not in ("ai", "aimessage", "aimessagechunk"):
+            continue
+        for tc in msg.get("tool_calls") or []:
+            if isinstance(tc, dict) and (tc.get("name") or "").lower() == "ask_clarification":
+                cargs = tc.get("args") if isinstance(tc.get("args"), dict) else {}
+                return {
+                    "question": str(cargs.get("question") or ""),
+                    "clarification_type": str(cargs.get("clarification_type") or ""),
+                    "clarification_context": str(cargs.get("context") or ""),
+                    "options": cargs.get("options") or [],
+                    "allow_custom": bool(cargs.get("allow_custom", True)),
+                    "allow_multiple": resolve_allow_multiple(cargs),
+                }
+    return None
+
+
+async def _run_one_agent(
+    client: httpx.AsyncClient,
+    base: str,
+    headers: dict[str, str],
+    agent_id: str,
+    thread_id: str,
+    new_message: str,
+    model_name: str,
+) -> AsyncIterator[str | dict[str, Any]]:
+    """跑一个智能体一轮（禁写文件 + 禁派活 / 脚本，其余只读工具放开）。
+
+    与 ``multi_agent._special_run`` 同形（yield 原始上游 SSE 行 + 末尾一条终态 dict），
+    但 ``excluded_tools`` = ``_EXCLUDED_TOOLS``（write_file / str_replace /
+    agent_orchestration / bash），且不做 post-broadcast（独立面板各智能体互不广播）。
+    """
+    body = _run_payload(
+        agent_name=agent_id,
+        model_name=model_name,
+        thread_id=thread_id,
+        new_message=new_message,
+        excluded_tools=list(_EXCLUDED_TOOLS),
+        skill_stop_names=[],
+        thinking_enabled=True,
+        reasoning_effort="medium",
+        multitask_strategy="rollback",
+    )
+    captured: list[str] = []
+    async for line, done in _stream_upstream(client, base, headers, thread_id, body):
+        if line is not None:
+            yield line
+        elif done is not None:
+            captured = done
+    messages = _parse_last_messages(captured)
+    if messages is None:
+        # 流被抢占/截断,没有终态 values 帧 —— 静默收尾(调用方据无终态帧兜底)。
+        return
+    last_ai: dict[str, Any] = {}
+    for msg in reversed(messages):
+        if isinstance(msg, dict) and (msg.get("type") or "").lower() in ("ai", "aimessage", "aimessagechunk"):
+            last_ai = msg
+            break
+    content = _flatten_content(last_ai.get("content", ""))
+    # 收集**所有** AI 消息的 tool_calls（全量 args，按顺序）—— 用于在收口时补发所有步骤
+    # （含 ls / skill_list 这类无参/中间步骤），保证步骤条不丢步、且 args 完整。
+    final_tool_calls: list[dict[str, Any]] = []
+    for msg in messages:
+        if isinstance(msg, dict) and (msg.get("type") or "").lower() in ("ai", "aimessage", "aimessagechunk"):
+            for tc in msg.get("tool_calls") or []:
+                if isinstance(tc, dict):
+                    final_tool_calls.append(tc)
+    clarification = _clarification_from_messages(messages)
+    if clarification is not None:
+        # 智能体在向用户要澄清 —— 交给前端渲染「协作卡片」(对齐圆桌第一步)。
+        yield {
+            "type": "clarification",
+            "content": content,
+            "clarification": clarification,
+            "tool_calls": final_tool_calls,
+        }
+    else:
+        yield {
+            "status": f"子智能体 {agent_id} 完成",
+            "content": content,
+            "agent_name": agent_id,
+            "tool_calls": final_tool_calls,
+        }


 def _build_task_message(task_name: str, brief: str) -> str:
@@ -112,8 +266,15 @@ def _build_task_message(task_name: str, brief: str) -> str:
        parts.append(f"任务简介：\n{brief.strip()}")
    parts.append(
        "请你从你的专业职责出发，针对以上任务给出你的分析与交付。"
-        "本轮为**并行**研讨：请把你的产出**直接写在回复正文里**，"
-        "不要写文件、不要执行脚本/命令。"
+        "本轮为**并行**研讨：请把你的产出**直接写在回复正文里**（本轮禁止写文件）。"
+    )
+    # 角色边界护栏：本面板里每个智能体都是**执行席位**，不是总控 / 指挥。即便任务里带有
+    # 「总控 / 业务链条 / 调度 / 派活」字样，也只完成属于自己的那份交付，严禁模仿总控派活。
+    parts.append(
+        "【角色边界】你是本轮的**执行席位**，不是总控 / 指挥。即使任务里出现「总控 / 业务链条 / 调度 / "
+        "派活」等字样，你也只需完成**属于你自己**的那份分析与交付；严禁模仿总控口吻输出「总控调度指令 / "
+        "调度对象 / 分配给『…』的任务 / agent_name」之类调度内容，严禁罗列或转派其他席位的任务，严禁调用"
+        "任何派活 / 调度类工具，也不要用脚本（bash / python）去调用内部接口或翻找派活脚本。"
    )
    return "\n\n".join(parts)

@@ -156,28 +317,60 @@ async def run_parallel_agents_stream(request: Request, payload: dict = Body(...)
    ) -> None:
        await queue.put({"type": "agent_start", "agent_id": agent_id})
        last_event = ""
-        seen_steps: set[str] = set()
+        emitted_tools: set[str] = set()  # 已发过 step 的 tool_call id（去重）
+        tool_acc: dict[int, dict[str, Any]] = {}  # tool_call_chunks 累积
        buffer: list[str] = []
        done_sent = False
+
+        async def _emit_tool(tc: Any, allow_empty: bool = False) -> None:
+            """一个 tool_call（含完整 args）→ 一条 step 事件；按 id 去重。
+
+            ``allow_empty=False``（流式期间）：args 为空先不发 —— 流式帧里的 tool_calls 常先给
+            空 ``{}`` 占位、真实 args 走 tool_call_chunks 后到，过早发会把空 args 钉死。等累积出
+            完整 args（synthetic）或终态全量 tool_calls（``allow_empty=True``）再发。
+            """
+            if not isinstance(tc, dict):
+                return
+            name = tc.get("name")
+            if not isinstance(name, str) or not name:
+                return
+            args = tc.get("args") if isinstance(tc.get("args"), dict) else None
+            if args is None:
+                return
+            if not args and not allow_empty:  # 空 args 先不发，等 chunks / 终态补全
+                return
+            key_id = str(tc.get("id") or name)
+            if key_id in emitted_tools:
+                return
+            emitted_tools.add(key_id)
+            await queue.put(
+                {"type": "step", "agent_id": agent_id, "key": f"tool_calling:{name}", "label": name, "args": args}
+            )
+
        try:
            thread_id = await _create_thread(client, base, headers, metadata=_ROUNDTABLE_THREAD_METADATA)
-            gen = _special_run(
-                client,
-                base,
-                headers,
-                {agent_id: thread_id},
-                agent_id,
-                new_message,
-                [],
-                model_name,
-                display_name="",
-                no_file=True,  # seat_parallel policy → 禁 write_file/bash/str_replace…
+            gen = _run_one_agent(  # 只禁文件写入，其余工具放开（保留步骤条）
+                client, base, headers, agent_id, thread_id, new_message, model_name
            )
            async for chunk in gen:
                if isinstance(chunk, dict):
-                    # 终态帧 {status, content, agent_name}
-                    content = str(chunk.get("content") or "")
-                    await queue.put({"type": "agent_done", "agent_id": agent_id, "content": content})
+                    # 终态：先用最终 tool_calls（全量 args）补发漏掉的 step，再发收口/澄清。
+                    for tc in chunk.get("tool_calls") or []:
+                        await _emit_tool(tc, allow_empty=True)
+                    if chunk.get("type") == "clarification":
+                        clar = chunk.get("clarification") or {}
+                        await queue.put(
+                            {
+                                "type": "agent_clarification",
+                                "agent_id": agent_id,
+                                "content": str(chunk.get("content") or ""),
+                                **clar,
+                            }
+                        )
+                    else:
+                        await queue.put(
+                            {"type": "agent_done", "agent_id": agent_id, "content": str(chunk.get("content") or "")}
+                        )
                    done_sent = True
                    continue
                line = chunk.strip() if isinstance(chunk, str) else ""
@@ -200,12 +393,12 @@ async def run_parallel_agents_stream(request: Request, payload: dict = Body(...)
                        if text:
                            buffer.append(text)
                            await queue.put({"type": "delta", "agent_id": agent_id, "text": text})
-                        for step in _extract_tool_steps(obj):
-                            key = step["key"]
-                            if key in seen_steps:
-                                continue
-                            seen_steps.add(key)
-                            await queue.put({"type": "step", "agent_id": agent_id, **step})
+                        # 完整 tool_calls（部分模型直接给）+ 累积 chunks 后的合成调用，都尝试发 step。
+                        for tc in obj.get("tool_calls") or []:
+                            await _emit_tool(tc)
+                        _ingest_tool_chunks(obj.get("tool_call_chunks"), tool_acc)
+                        for tc in _synthetic_tool_calls(tool_acc):
+                            await _emit_tool(tc)
            if not done_sent:
                # 流被抢占/截断,没收到终态帧 —— 用已累积正文兜底收口本席位。
                await queue.put(
@@ -51,6 +51,8 @@ class ChainResponse(BaseModel):
    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）。
@@ -72,6 +74,7 @@ class ChainCreateRequest(BaseModel):
    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):
@@ -81,6 +84,7 @@ class ChainUpdateRequest(BaseModel):
    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

@@ -222,6 +226,7 @@ async def create_chain(request: Request, body: ChainCreateRequest) -> ChainRespo
            "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)
@@ -0,0 +1,56 @@
+"""Add coordinator_prompt column to roundtable_chains.
+
+Per-chain「总控编排提示」: an optional free-text instruction the user writes for the
+**coordinator (总控) agent** of a business chain — telling it how to dispatch / what to
+do (overall orchestration strategy / style / constraints). Injected into the coordinator
+per-stage dispatch prompt (``buildDagStageLeaderTask``) in **dag-mode only**. NULL = no
+prompt. The column is LONGTEXT on MySQL (mirrors ``stage_goals`` headroom) and plain TEXT
+elsewhere — matching ``PortableLongText`` on ``RoundtableChainRow.coordinator_prompt``.
+``create_all`` adds the column for brand-new DBs; this revision patches DBs that already
+have the table.
+
+Chained after ``20260608_05`` (parallel_agent_tasks) so the revision graph stays
+single-headed — otherwise ``alembic upgrade head`` aborts with "multiple heads" and the
+startup auto-migrate silently falls back to the create_all schema (which lacks this
+column → 500 on list).
+
+Revision ID: 20260608_06
+Revises: 20260608_05
+Create Date: 2026-06-09
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy import inspect
+from sqlalchemy.dialects.mysql import LONGTEXT
+
+
+revision = "20260608_06"
+down_revision = "20260608_05"
+branch_labels = None
+depends_on = None
+
+
+def _has_table(table_name: str) -> bool:
+    inspector = inspect(op.get_bind())
+    return table_name in inspector.get_table_names()
+
+
+def _has_column(table_name: str, column_name: str) -> bool:
+    if not _has_table(table_name):
+        return False
+    inspector = inspect(op.get_bind())
+    return any(col["name"] == column_name for col in inspector.get_columns(table_name))
+
+
+def upgrade() -> None:
+    if _has_table("roundtable_chains") and not _has_column("roundtable_chains", "coordinator_prompt"):
+        col_type = LONGTEXT() if op.get_bind().dialect.name == "mysql" else sa.Text()
+        op.add_column("roundtable_chains", sa.Column("coordinator_prompt", col_type, nullable=True))
+
+
+def downgrade() -> None:
+    if _has_column("roundtable_chains", "coordinator_prompt"):
+        op.drop_column("roundtable_chains", "coordinator_prompt")
@@ -45,6 +45,10 @@ class RoundtableChainRow(Base):
    # coordinator dispatch prompt (and seat tasks as fallback) in dag-mode only. NULL =
    # no goals (and always NULL when ``stages`` is NULL/linear). Stored as JSON string.
    stage_goals: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
+    # 可选「总控编排提示」：用户为本链条写给**总控（协调）智能体**的一段指引，告诉总控该
+    # 怎么派活、做什么（整体编排策略 / 风格 / 约束）。**仅分层 DAG 模式生效**：注入到总控
+    # 每阶段派活提示（``buildDagStageLeaderTask``）。NULL = 不注入。纯文本，存为 LONGTEXT。
+    coordinator_prompt: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
    # 发布开关：True = 公共（所有账号可在「公共」筛选里查看 / 选用）；False = 仅本人可见。
    # 编辑 / 删除始终只限 owner；发布只是放开「查看 + 选入研讨」。索引以支持「公共」列表筛选。
    is_public: Mapped[bool] = mapped_column(
@@ -54,6 +54,8 @@ class RoundtableChainRepository:
            "stages": _loads(row.stages),
            # Per-stage goals aligned to ``stages``; None when absent.
            "stage_goals": _loads(getattr(row, "stage_goals", None)),
+            # 总控编排提示（写给总控的派活指引，dag-only）；None/空 = 不注入。纯文本,不 json。
+            "coordinator_prompt": getattr(row, "coordinator_prompt", None),
            # 发布状态 + 是否本人所有（前端据此显示「公共」徽标、决定能否编辑 / 发布）。
            "is_public": bool(getattr(row, "is_public", False)),
            "is_mine": (current_user_id is None) or (row.user_id == current_user_id),
@@ -105,6 +107,7 @@ class RoundtableChainRepository:
            seats=_dumps(data.get("seats") or []),
            stages=_dumps(data.get("stages")),
            stage_goals=_dumps(data.get("stage_goals")),
+            coordinator_prompt=(data.get("coordinator_prompt") or None),
            is_public=bool(data.get("is_public") or False),
            created_at=now,
            updated_at=now,
@@ -125,6 +128,7 @@ class RoundtableChainRepository:
        seats=_UNSET,
        stages=_UNSET,
        stage_goals=_UNSET,
+        coordinator_prompt=_UNSET,
        is_public=_UNSET,
    ) -> dict[str, Any] | None:
        async with self._sf() as session:
@@ -141,6 +145,8 @@ class RoundtableChainRepository:
                row.stages = _dumps(stages)
            if stage_goals is not _UNSET:
                row.stage_goals = _dumps(stage_goals)
+            if coordinator_prompt is not _UNSET:
+                row.coordinator_prompt = coordinator_prompt or None
            if is_public is not _UNSET and is_public is not None:
                row.is_public = bool(is_public)
            row.updated_at = datetime.now(UTC)