"""圆桌规划 · Step 2「多智能体研讨」Gateway 路由。

前端页面：``RoundtablePlanningPage`` Step 2
前端模块：``frontend-web/src/roundtable-planning/api/multi-agent.ts``

本模块把原 ``newpython/OUT.py`` 演示编排进 FastAPI Gateway，通过 **HTTP loopback**
（``127.0.0.1`` + 调用方 ``Authorization``）复用既有 ``/api/threads``、
``/api/agents``、``/api/threads/{id}/runs/stream``，不再硬编码 token。

对外接口
--------
1. ``POST /api/multi-agent/init``（SSE）
   - 为每个研讨席位（``agent_names``）并行创建 thread，按完成顺序推送
     ``seat_ready``；
   - 动态创建**当次会话**的总控协调智能体（``main_agent_name``，如
     ``roundtable-coordinator-xxx``），写入席位列表 SOUL + ``agent_orchestration`` skill；
   - 结束帧 ``init_done`` 携带 ``thread_ids: { agent_name: thread_id, ... }``，
     含协调智能体自身 thread。

2. ``POST /api/multi-agent/run/stream``（SSE）
   - ``agent_type=leader``：总控一轮——可能派活、可能澄清、可能共识（空 dispatch）；
   - ``agent_type=special``：单席位子智能体执行一轮交付。

跨席位「广播」语义
------------------
通过 ``_append_thread_message`` 向其它 thread 的 checkpoint ``messages`` 追加
**只读上下文**的 human 消息（非触发 run），使各席位 LangGraph 状态里能看到
总控派活/用户干预/子 agent 完成摘要。前端编排循环仍靠显式多次 ``/run/stream``
驱动各 agent 真正推理。

Leader 终态帧（前端 ``isFinalStatusFrame``）
-------------------------------------------
- ``status: [[agent_name, task], ...]`` — 派活列表，前端并行 special run；
- ``status: []`` — 无 tool_calls，视为共识，可进 Step 3；
- ``status: "clarification"`` — 总控 ``ask_clarification``，前端暂停编排展示澄清条；
- ``status: "子智能体 X 完成..."`` — special 完成（字符串状态）；
- ``status: "error"`` — 失败。

共享导出
--------
``intent.py`` / ``recommend.py`` 从此模块 import 底层流式与消息解析函数，
勿在 intent/recommend 中重复实现 loopback 逻辑。

相关文档：``frontend-web/docs/multi-agent-backend-dev.md``
"""

from __future__ import annotations

import asyncio
import json
import logging
import os
import re
import time
import traceback
from typing import Any, AsyncIterator

import httpx
from deerflow.config import get_app_config
from deerflow.tools.builtins.clarification_utils import resolve_allow_multiple
from fastapi import APIRouter, Body, HTTPException, Request
from fastapi.responses import StreamingResponse

logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/multi-agent", tags=["multi-agent"])


# ---------------------------------------------------------------------------
# 共享:异常诊断器(intent/recommend/multi-agent 共用)
# ---------------------------------------------------------------------------


def _diagnose_exception(exc: BaseException, *, context: str = "") -> dict[str, Any]:
    """把后端异常转成结构化诊断 dict,直接给前端展示用。

    设计目标:让运维 / 前端开发不用翻 backend 日志就能看到根因和处置建议。
    返回字段:
      - code:           短稳定分类(loopback_connect_failed / upstream_connect_failed / ...)
      - message:        一行人类可读
      - exception_type: 表层异常类名
      - exception_repr: repr(exc)
      - root_type / root_repr: 沿 __cause__ 链找到的最深异常(若与表层不同)
      - hint:           针对常见失败模式的可执行建议(中文)
      - trace_tail:     traceback 最后 ~12 行(避免响应体过大)
      - context:        调用方标签(如 "intent_init")
    """
    exc_type = type(exc).__name__
    exc_repr = repr(exc)

    # 沿 __cause__ 链找最深异常:httpx 经常把 ConnectError 包成 RemoteProtocolError 等。
    root: BaseException = exc
    seen: set[int] = {id(root)}
    while getattr(root, "__cause__", None) is not None and id(root.__cause__) not in seen:
        root = root.__cause__  # type: ignore[assignment]
        seen.add(id(root))

    code = "unexpected"
    hint = "查看 trace_tail 最后一帧定位代码位置;若信息不足请到 backend 日志找完整 traceback。"

    if isinstance(root, httpx.ConnectError):
        target = str(root).strip().lower()
        # loopback 走真 TCP 到 127.0.0.1:<gateway port>;失败 = Gateway 进程没起 / 端口被占。
        # outbound 失败(其它 host)= LLM 提供商不可达。
        if "127.0.0.1" in target or "localhost" in target or "loopback" in target:
            code = "loopback_connect_failed"
            hint = (
                "本地 Gateway loopback 连接失败:确认后端进程正在监听配置的端口。"
                "端口配置优先级(高→低):env DEER_FLOW_GATEWAY_PORT > config.yaml 的 "
                "gateway.port > 默认 8001。若部署机改了端口,在 config.yaml 加 "
                "`gateway:\\n  port: <端口>` 即可。"
            )
        else:
            code = "upstream_connect_failed"
            hint = (
                "外网/上游服务连接失败:大概率是 LLM 提供商不可达"
                "(如 api.deepseek.com 在内网被防火墙拦截)。"
                "把 config.yaml 里所有 base_url 改成内网可达的 LLM endpoint,"
                "或加白名单/正向代理(HTTPS_PROXY)。"
            )
    elif isinstance(root, httpx.TimeoutException):
        code = "upstream_timeout"
        hint = (
            "上游请求超时:网络抖动 / 模型推理过慢。"
            "先 curl <base_url>/v1/models 确认可达,再看模型是否过载或 max_tokens 过大。"
        )
    elif isinstance(root, httpx.HTTPStatusError):
        code = "upstream_http_error"
        status = getattr(getattr(root, "response", None), "status_code", "?")
        hint = (
            f"上游接口返回 HTTP {status}:可能是 API key 失效、模型名不对、"
            "或目标服务自身报错。查看 backend 日志里 response.text 拿详细错误体。"
        )
    elif isinstance(root, FileNotFoundError):
        code = "file_not_found"
        hint = (
            f"找不到路径:{root}。常见原因:内置 agent 目录(.deer-flow/agents/<id>/)缺失,"
            "或 SOUL.md 没拷过来;详见后端开发文档 §11.7。"
        )
    elif isinstance(root, KeyError):
        code = "key_error"
        hint = (
            f"缺少 key:{root}。若 key 是 agent_name/agent_id 类:大概率内置 agent 目录缺失;"
            "若是配置 key:检查 config.yaml schema。"
        )
    elif isinstance(root, ValueError) and "model" in str(root).lower():
        code = "no_models_configured"
        hint = "config.yaml 没有任何 models[],或所有条目都解析失败。检查 yaml 语法与 $ENV 引用。"

    trace_lines = traceback.format_exception(type(exc), exc, exc.__traceback__)
    trace_tail = "".join(trace_lines).splitlines()[-12:]

    return {
        "code": code,
        "message": f"{exc_type}: {exc}" if str(exc) else exc_type,
        "exception_type": exc_type,
        "exception_repr": exc_repr,
        "root_type": type(root).__name__ if root is not exc else None,
        "root_repr": repr(root) if root is not exc else None,
        "hint": hint,
        "trace_tail": trace_tail,
        "context": context or None,
    }


# ---------------------------------------------------------------------------
# Loopback 辅助函数（Gateway → Gateway，同进程）
#
# 圆桌三路由（intent / recommend / multi_agent）需要从一个 FastAPI handler 里
# 调用本机 Gateway 的另一个 endpoint（`/api/threads`、`/api/agents`、
# `/api/threads/{id}/runs/stream`）。这里用真 TCP loopback 到本机的 127.0.0.1
# + Gateway 监听端口。
#
# ⚠️ 历史踩坑(三个) ⚠️
# 1. 直接走系统代理:开发机若设了 HTTP_PROXY / HTTPS_PROXY,httpx 默认会把
#    127.0.0.1 也走代理 → 上游代理不可达,40s 超时(典型日志 headers_in=42.160s)。
#    → 修法:client 用 `trust_env=False`,强制忽略代理 env vars。
#
# 2. 用 request.url.netloc 拼 base URL:Nginx 反代部署时,客户端打到 nginx 的
#    public host(如 example.com),nginx 把请求转给 gateway 时 Host header
#    通常是 `gateway:8001` 或 public host;后者 httpx 会真的去 DNS 解析,
#    在 gateway 容器内可能不可达 → 连接失败。
#    → 修法:**写死 127.0.0.1**,完全无视 request.url.netloc。
#
# 3. 一度尝试用 httpx.ASGITransport(app=request.app) 进程内直调,看似一劳永逸,
#    但 httpx 0.28 的 ASGITransport 对 streaming response **整段 buffer**,
#    一直收齐所有 body chunks 才一次性把响应给消费方 → SSE 完全失效,前端表现
#    为"调度阶段很长 + 流式结果一次性出现"。已弃用。
#    → 详见:git log 6e3a07b(那次改 ASGI 引入)及随后回退。
#
# 端口策略(优先级,从高到低):
#   1. 环境变量 DEER_FLOW_GATEWAY_PORT
#      ── 部署侧紧急覆盖通道,适合 docker-compose / k8s 注入或临时调试
#   2. config.yaml 的 `gateway.port`
#      ── 部署侧**推荐**的配置方式,改完不需要重启二进制(get_app_config 有 mtime
#         热加载,但仍建议跑一次 reload 以确保所有 worker 拿到新值)
#   3. request.url.port
#      ── 浏览器直连 gateway:8001 时拿得到;走 Nginx 80/443 反代时会落到下一档
#   4. 默认 8001(与 `make dev` / `make gateway` / `scripts/start-all.sh` 写死的一致)
#
# 端口在哪里改(给运维 / 部署侧的提示):
#   - yaml 配置:打开 config.yaml,加 / 改:
#       gateway:
#         port: 9999
#   - 环境变量:
#       export DEER_FLOW_GATEWAY_PORT=9999
#
# 修改这两个函数时务必跑一次 SSE 流式 smoke 验证(见 multi-agent-backend-dev.md
# §13 测试建议),确认 chunks 是逐个到达而不是一次喷出。
# ---------------------------------------------------------------------------


# 写死的最终 fallback。**不要**为了改端口而修改这个常量——优先用 config.yaml /
# env 这两层(它们在部署侧不需要改源码即可生效)。这个值只在 yaml 缺失 + env
# 没设 + 请求里也拿不到合法 port 时才会用到。
_DEFAULT_GATEWAY_PORT = 8001


def _resolve_gateway_port(request: Request) -> int:
    """按四档 fallback 解析 Gateway 监听端口。详见模块顶部「端口策略」。"""
    # 第 1 档:环境变量。允许非法值(garbage),只警告并继续下一档,不抛异常。
    env_port = os.environ.get("DEER_FLOW_GATEWAY_PORT", "").strip()
    if env_port:
        try:
            port = int(env_port)
            if 1 <= port <= 65535:
                return port
            raise ValueError("out of range")
        except ValueError:
            logger.warning(
                "DEER_FLOW_GATEWAY_PORT=%r 不是合法端口,fallback 到 config.yaml / request / %d",
                env_port,
                _DEFAULT_GATEWAY_PORT,
            )

    # 第 2 档:config.yaml 的 gateway.port。GatewayConfig.port 的 pydantic 校验
    # 已经保证它在 1-65535 范围内,无需再次校验。读 config 失败时降级,不阻断业务。
    try:
        configured = get_app_config().gateway.port
        if configured:
            return configured
    except Exception as exc:
        # 极端场景:config.yaml 不存在或解析失败。这种情况下 gateway 本身也起不来,
        # loopback 端口"猜错"也已经无所谓。仅记 debug 日志方便排查。
        logger.debug("read gateway.port from app config failed: %s", exc)

    # 第 3 档:从当前请求拿 port。浏览器直连 gateway 时是真实监听端口;
    # 走 Nginx 反代时通常是 80/443(public port),那就跳过,落到最终默认。
    inbound_port = request.url.port
    if inbound_port is not None and inbound_port not in (80, 443):
        return inbound_port

    # 第 4 档:写死默认。
    return _DEFAULT_GATEWAY_PORT


def _loopback_base(request: Request) -> str:
    """返回 `http://127.0.0.1:<port>`,供 loopback 调用拼 URL。

    强制 127.0.0.1 是为了避免 Nginx 反代场景下读到经过 Host 改写的 netloc。
    端口按 `_resolve_gateway_port` 的四档 fallback 解析。
    """
    return f"http://127.0.0.1:{_resolve_gateway_port(request)}"


def _make_loopback_client(request: Request) -> httpx.AsyncClient:
    """构造同进程 loopback 用的 httpx.AsyncClient(真 TCP)。

    所有圆桌路由(intent / recommend / multi_agent)的 loopback 调用必须走这里,
    保持配置一致。
    - `trust_env=False`:**关键**,忽略 HTTP_PROXY / HTTPS_PROXY,防代理拦截 127.0.0.1
    - `timeout=None`:SSE 长连接需要,LangGraph runs/stream 可能跑数十秒
    - `request` 参数当前未被使用,保留是为了:(a) 未来若需要从 request 取部署
      上下文(如多租户端口隔离)可以无侵入扩展;(b) 与调用方现有签名兼容。
    """
    return httpx.AsyncClient(
        timeout=None,
        trust_env=False,
    )


def _auth_headers(request: Request) -> dict[str, str]:
    token = request.headers.get("authorization")
    if not token:
        raise HTTPException(status_code=401, detail="Missing Authorization header")
    return {"Authorization": token, "Content-Type": "application/json"}


async def _create_thread(client: httpx.AsyncClient, base: str, headers: dict[str, str]) -> str:
    """``POST /api/threads`` → 新 thread_id（intent/recommend/init 席位均用）。"""
    resp = await client.post(f"{base}/api/threads", json={}, headers=headers)
    resp.raise_for_status()
    return resp.json()["thread_id"]


async def _get_agent(client: httpx.AsyncClient, base: str, headers: dict[str, str], agent_id: str) -> dict[str, Any]:
    resp = await client.get(f"{base}/api/agents/{agent_id}", headers=headers)
    resp.raise_for_status()
    return resp.json()


async def _create_agent(
    client: httpx.AsyncClient,
    base: str,
    headers: dict[str, str],
    *,
    name: str,
    description: str,
    model: str | None,
    skills: list[str] | None,
    soul: str,
    agent_id: str | None = None,
) -> tuple[bool, dict[str, Any] | str]:
    payload: dict[str, Any] = {"name": name, "description": description, "soul": soul}
    if agent_id is not None:
        payload["id"] = agent_id
    if model is not None:
        payload["model"] = model
    if skills is not None:
        payload["skills"] = skills
    resp = await client.post(f"{base}/api/agents", json=payload, headers=headers)
    if resp.status_code == 201:
        return True, resp.json()
    return False, resp.text


async def _append_thread_message(
    client: httpx.AsyncClient,
    base: str,
    headers: dict[str, str],
    thread_id: str,
    content: str,
) -> None:
    """向指定 thread 的 checkpoint 追加一条 human 消息（广播上下文，不触发 LLM run）。

    失败只打 warning，不阻断主流程——广播是 best-effort，席位真正干活仍靠
    前端对该席位 thread 发起的 ``/run/stream``。
    """
    state_url = f"{base}/api/threads/{thread_id}/state"
    get_resp = await client.get(state_url, headers=headers)
    if get_resp.status_code != 200:
        logger.warning("get state %s failed: %s %s", thread_id, get_resp.status_code, get_resp.text[:200])
        return
    state = get_resp.json()
    values = state.setdefault("values", {})
    messages = values.setdefault("messages", [])
    if not isinstance(messages, list):
        messages = []
        values["messages"] = messages
    messages.append({"type": "human", "content": content})
    post_resp = await client.post(state_url, json=state, headers=headers)
    if post_resp.status_code >= 400:
        logger.warning("update state %s failed: %s %s", thread_id, post_resp.status_code, post_resp.text[:200])


async def _broadcast(
    client: httpx.AsyncClient,
    base: str,
    headers: dict[str, str],
    message: str,
    agent_threads: dict[str, str],
    skip: list[str],
) -> None:
    """并行向 ``agent_threads`` 中除 ``skip`` 外的所有 thread 追加同一条 human 消息。"""
    targets = [tid for name, tid in agent_threads.items() if name not in skip]
    if not targets:
        return
    await asyncio.gather(
        *(_append_thread_message(client, base, headers, tid, message) for tid in targets),
        return_exceptions=True,
    )


_TOOL_CALL_PATTERN = re.compile(r'\.py\s+"([^"]*)"\s+"([^"]*)"')

# Mirrors the upstream agent-id validator in the LangGraph runtime. Reject early
# so callers get a clear 400 here instead of a stream that aborts with HTTP 422
# from the loopback /api/threads/.../runs/stream endpoint.
_AGENT_ID_PATTERN = re.compile(r"^[A-Za-z0-9_-]+$")


def _validate_agent_id(value: str, field: str) -> str:
    if not _AGENT_ID_PATTERN.match(value):
        raise HTTPException(
            status_code=400,
            detail=(
                f"{field} '{value}' is not a valid agent id. "
                f"Must match ^[A-Za-z0-9_-]+$ (use the hyphen-case id, not the display name)."
            ),
        )
    return value


def _extract_dispatch(tool_call: dict[str, Any]) -> tuple[str, str]:
    """Return ``(agent_name, task)`` parsed from a leader tool call.

    Two formats are accepted:

    1. **Structured** — ``{"name": "agent_orchestration", "args": {"agent_name": ..., "task": ...}}``.
       This is what most LLMs emit when told to "use the agent_orchestration skill",
       and the form the roundtable coordinator's SOUL prompt elicits.

    2. **Legacy bash argv** — ``{"name": "bash", "args": {"command": '... .py "X" "Y" ...'}}``.
       Kept as a fallback so older skills that dispatch through bash still work.
    """
    name = (tool_call.get("name") or "").lower()
    args = tool_call.get("args")

    # Format 1: structured agent_orchestration call.
    if name == "agent_orchestration" and isinstance(args, dict):
        agent_name = str(args.get("agent_name") or "").strip()
        task = str(args.get("task") or "").strip()
        if agent_name and task:
            return _strip_display_suffix(agent_name), task

    # Format 2: legacy bash + .py "X" "Y".
    match = _TOOL_CALL_PATTERN.search(str(args))
    if match:
        return _strip_display_suffix(match.group(1)), match.group(2)

    return "", ""


def _strip_display_suffix(agent_name: str) -> str:
    """Tolerate models that paste the human-readable suffix into agent_name.

    The coordinator SOUL lists each seat as ``agent_name: roundtable-x（中文）：...``
    so weaker models sometimes echo the entire ``roundtable-x（中文）`` label
    back as the dispatch target. Strip everything from the first non-id
    character onward so we recover the bare hyphen-case id; if what's left
    still violates the id pattern we hand it back untouched and let the
    upstream validator reject it.
    """
    head = re.split(r"[^A-Za-z0-9_-]", agent_name, maxsplit=1)[0]
    return head or agent_name


def _flatten_content(content: Any) -> str:
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts: list[str] = []
        for item in content:
            if isinstance(item, dict):
                parts.append(str(item.get("text", "")))
            else:
                parts.append(str(item))
        return "".join(parts)
    return str(content)


def _run_payload(
    *,
    agent_name: str,
    model_name: str,
    thread_id: str,
    new_message: str,
    excluded_tools: list[str],
    skill_stop_names: list[str],
) -> dict[str, Any]:
    """构造 ``POST /api/threads/{thread_id}/runs/stream`` 的请求体。

    - ``context.agent_name``：本次 run 使用的 agent（内置 id 或 init 创建的 coordinator id）。
    - ``excluded_tools``：按路由裁剪工具面（intent 仅澄清、special 禁派活等）。
    - ``skill_stop_names``：leader 侧配合 SkillStopMiddleware 在 ``agent_orchestration``
      调用后截断图，以便网关从 AIMessage.tool_calls 解析派活列表。
    """
    return {
        "input": {
            "messages": [
                {
                    "type": "human",
                    "content": [{"type": "text", "text": new_message}],
                    "additional_kwargs": {},
                }
            ]
        },
        "config": {"recursion_limit": 1000},
        "context": {
            "agent_name": agent_name,
            "model_name": model_name,
            "mode": "pro",
            "reasoning_effort": "medium",
            "thinking_enabled": True,
            "is_plan_mode": False,
            "subagent_enabled": False,
            "thread_id": thread_id,
        },
        "stream_mode": ["messages-tuple", "values"],
        "stream_subgraphs": True,
        "stream_resumable": True,
        "assistant_id": "lead_agent",
        "on_disconnect": "continue",
        "excluded_tools": excluded_tools,
        "skill_stop_names": skill_stop_names,
    }


# ---------------------------------------------------------------------------
# POST /api/multi-agent/init — Step 2 席位 + 总控初始化（SSE）
# ---------------------------------------------------------------------------


@router.post("/init")
async def init_multi_agent(request: Request, payload: dict = Body(...)) -> StreamingResponse:
    """流式初始化圆桌 Step 2 所需的 thread 映射与协调智能体。

    请求体::

        {
          "agent_names": ["roundtable-intelligence", ...],  // 用户选的席位 id
          "main_agent_name": "roundtable-coordinator-xxx",  // 当次总控 id，前端生成
          "model": "<可选>"                                  // 可选,写入 coordinator agent;不传由 lead_agent 回退到 config.yaml models[0]
        }

    SSE 事件（每条 ``data: {...}\\n\\n``）::

        seat_ready       — 某一席位 thread 已创建（``asyncio.as_completed`` 顺序，先完先发）
        coordinator_ready — 总控 agent + thread 已就绪
        init_done        — ``thread_ids`` 全量映射，前端存 ``threadIds`` 并开始编排
        error            — 失败详情（同步校验失败则直接 HTTP 4xx，不进 SSE）

    协调智能体 SOUL 在运行时根据各席位 ``GET /api/agents/{id}`` 的 name/description 拼接，
    强调 ``agent_name`` 必须用英文 id（``roundtable-intelligence``），禁止带中文括号后缀。
    """
    agent_names_in = payload.get("agent_names") or []
    agent_names: list[str] = [str(name).lower() for name in agent_names_in]
    main_agent_name: str = str(payload.get("main_agent_name") or "main-agent").lower()
    model: str | None = payload.get("model")

    if not agent_names:
        raise HTTPException(status_code=400, detail="agent_names is required")

    for name in agent_names:
        _validate_agent_id(name, "agent_names entry")
    _validate_agent_id(main_agent_name, "main_agent_name")

    base = _loopback_base(request)
    headers = _auth_headers(request)

    def _frame(payload: dict[str, Any]) -> str:
        return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"

    async def stream_generator() -> AsyncIterator[str]:
        # TCP loopback 到 127.0.0.1:<gateway port>。client 配置见 _make_loopback_client：
        # trust_env=False 防 HTTP_PROXY 拦截,timeout=None 容纳 SSE 长连接。
        async with _make_loopback_client(request) as client:
            try:
                # Per-seat prepare task — same body as before, but we no longer
                # await them with asyncio.gather. Instead, asyncio.as_completed
                # surfaces them in REAL completion order so we can emit a
                # seat_ready frame the moment each finishes.
                async def _prepare_sub_agent(sub_id: str) -> tuple[str, str, str, str]:
                    info = await _get_agent(client, base, headers, sub_id)
                    display_name = str(info.get("name") or sub_id)
                    description = info.get("description", "") or ""
                    thread_id = await _create_thread(client, base, headers)
                    return sub_id.lower(), thread_id, display_name, description

                tasks = [asyncio.create_task(_prepare_sub_agent(sid)) for sid in agent_names]
                prepared: list[tuple[str, str, str, str]] = []
                try:
                    for fut in asyncio.as_completed(tasks):
                        sid, thread_id, display_name, description = await fut
                        prepared.append((sid, thread_id, display_name, description))
                        yield _frame({
                            "event": "seat_ready",
                            "agent_name": sid,
                            "thread_id": thread_id,
                            "display_name": display_name,
                            "description": description,
                        })
                except Exception:
                    # Cancel any still-pending sibling so we don't leak tasks
                    # past the AsyncClient close.
                    for t in tasks:
                        if not t.done():
                            t.cancel()
                    raise

                agent_threads: dict[str, str] = {sid: thread_id for sid, thread_id, _, _ in prepared}
                descriptions: list[str] = [
                    f"- agent_name: {sid}（{display_name}）：{description}"
                    for sid, _, display_name, description in prepared
                ]

                # 动态 SOUL：把本次 init 的席位列表写进总控 prompt，供 agent_orchestration 派活。
                soul = (
                    "#核心定位：\n"
                    "用途：根据用户的任务目标,调用其他智能体干活,只负责协调汇总,"
                    "当你觉得已有信息支持回答用户问题时候,停止调度,并输出最后的结果\n"
                    "#调度方法：\n"
                    "通过使用agent_orchestration技能调用其他智能体干活,使用技能时传入agent_name和需要做的活。"
                    "注意：agent_name 必须严格使用下方列出的英文 id（例如 roundtable-intelligence），"
                    "不要把括号里的中文显示名拼进 agent_name 里。\n"
                    "#可调度智能体：\n"
                    + "\n".join(descriptions)
                )

                coord_thread = await _create_thread(client, base, headers)
                agent_threads[main_agent_name] = coord_thread
                ok, result = await _create_agent(
                    client,
                    base,
                    headers,
                    name=main_agent_name,
                    description="你是总控控制协调智能体,负责协调其他智能体干活",
                    model=model,
                    skills=["agent_orchestration"],
                    soul=soul,
                    agent_id=main_agent_name,
                )
                if not ok and isinstance(result, str) and "exists" not in result:
                    yield _frame({"event": "error", "detail": f"create coordinator agent failed: {result}"})
                    return

                yield _frame({
                    "event": "coordinator_ready",
                    "main_agent_name": main_agent_name,
                    "thread_id": coord_thread,
                })
                yield _frame({
                    "event": "init_done",
                    "main_agent_name": main_agent_name,
                    "thread_ids": agent_threads,
                })
            except HTTPException as exc:
                logger.warning("multi-agent init HTTPException: %s %s", exc.status_code, exc.detail)
                yield _frame({"event": "error", "detail": f"[{exc.status_code}] {exc.detail}"})
            except Exception as exc:
                logger.exception("multi-agent init unexpected failure")
                yield _frame({"event": "error", "detail": f"unexpected: {exc!r}"})

    return StreamingResponse(
        stream_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",
            "Connection": "keep-alive",
        },
    )


# ---------------------------------------------------------------------------
# 上游 LangGraph SSE 透传（intent / recommend / leader / special 共用）
# ---------------------------------------------------------------------------


async def _stream_upstream(
    client: httpx.AsyncClient,
    base: str,
    headers: dict[str, str],
    thread_id: str,
    body: dict[str, Any],
) -> AsyncIterator[tuple[str | None, list[str] | None]]:
    """消费 ``/api/threads/{thread_id}/runs/stream``，逐行 yield 原始 SSE。

    产出元组：
    - ``(line, None)`` — 可原样转发给浏览器（含 ``event:`` / ``data:`` 行）；
    - ``(None, captured)`` — 流结束，``captured`` 为所有 ``data:`` 行文本列表，
      供 ``_parse_last_messages`` 从倒数 values 快照提取 ``messages``。

    日志 ``[upstream-timing]``：排查首 token 慢；headers_in 长尾若 >> 网络往返(loopback
    本机往返通常 <10ms),怀疑 HTTP_PROXY 拦了回环——检查 client 是否走 _make_loopback_client
    (trust_env=False)。
    """

    url = f"{base}/api/threads/{thread_id}/runs/stream"
    captured: list[str] = []
    t_send = time.perf_counter()
    async with client.stream("POST", url, json=body, headers=headers) as resp:
        t_headers = time.perf_counter()
        if resp.status_code != 200:
            body_text = (await resp.aread()).decode("utf-8", "replace")[:1000]
            raise HTTPException(
                status_code=502,
                detail=f"upstream {url} returned HTTP {resp.status_code}: {body_text}",
            )
        agent_for_log = (body.get("context") or {}).get("agent_name") or thread_id
        logger.info(
            "[upstream-timing] %s headers_in=%.3fs",
            agent_for_log,
            t_headers - t_send,
        )
        first_line = True
        async for line in resp.aiter_lines():
            if not line:
                continue
            if first_line:
                first_line = False
                logger.info(
                    "[upstream-timing] %s first_line=%.3fs (LangGraph runtime startup + LLM first byte)",
                    agent_for_log,
                    time.perf_counter() - t_headers,
                )
            yield line, None
            if line.startswith("data:"):
                captured.append(line)
    yield None, captured


def _parse_last_messages(captured: list[str]) -> list[dict[str, Any]]:
    """Return the full ``messages`` list from the most recent ``values`` snapshot.

    The upstream sends the final ``values`` frame second-to-last (the very last
    line is the ``end`` event with ``null``). We tolerate trailing non-JSON
    frames (e.g. ``data: null``) by scanning backwards for the first frame that
    successfully parses as a state object with a ``messages`` key.
    """
    if not captured:
        raise HTTPException(status_code=502, detail="upstream produced no data lines")

    for raw_line in reversed(captured):
        raw = raw_line[5:].strip() if raw_line.startswith("data:") else raw_line.strip()
        if not raw or raw == "null":
            continue
        try:
            parsed = json.loads(raw)
        except json.JSONDecodeError:
            continue
        if isinstance(parsed, dict) and "messages" in parsed:
            messages = parsed.get("messages") or []
            if isinstance(messages, list):
                return messages

    raise HTTPException(status_code=502, detail="no parseable values frame in upstream stream")


def _find_trailing_ai_text(messages: list[dict[str, Any]], *, exclude_text: str = "") -> str:
    """Return the text content of the latest AIMessage with no tool_calls.

    Used by the clarification branch: ClarificationMiddleware ends the run
    after emitting the structured ToolMessage, but the model often produces a
    richer human-readable follow-up AIMessage right before the interrupt.
    Prefer that text over the brief preamble that accompanied the
    ask_clarification tool_call. When the only trailing AIMessage matches the
    preamble (``exclude_text``), return empty so the caller can fall back to
    the question itself.
    """
    for msg in reversed(messages):
        if not isinstance(msg, dict):
            continue
        if (msg.get("type") or "").lower() not in ("ai", "aimessage", "aimessagechunk"):
            continue
        if msg.get("tool_calls"):
            continue
        text = _flatten_content(msg.get("content", "")).strip()
        if not text:
            continue
        if exclude_text and text == exclude_text.strip():
            continue
        return text
    return ""


def _find_dispatch_message(messages: list[dict[str, Any]]) -> dict[str, Any]:
    """Locate the latest AIMessage that carries dispatch tool calls.

    The very last message can be a skill-stop ToolMessage injected by
    ``SkillStopMiddleware`` when it intercepts the dispatch, or a clarification
    ToolMessage. Either way, the actionable tool_calls live on the AIMessage
    *before* it, so we walk backwards until we hit one with ``tool_calls``.
    Returns the latest AIMessage when none of them carry tool_calls so the
    caller can still extract its text content.
    """
    last_ai: dict[str, Any] = {}
    for msg in reversed(messages):
        if not isinstance(msg, dict):
            continue
        if (msg.get("type") or "").lower() in ("ai", "aimessage", "aimessagechunk"):
            if not last_ai:
                last_ai = msg
            if msg.get("tool_calls"):
                return msg
    return last_ai


async def _leader_run(
    client: httpx.AsyncClient,
    base: str,
    headers: dict[str, str],
    agent_threads: dict[str, str],
    agent_name: str,
    new_message: str,
    skill_stop_names: list[str],
    model_name: str,
) -> AsyncIterator[str | dict[str, Any]]:
    """总控（leader）单轮：透传上游 SSE + 末尾 yield 一个 dict 状态帧（由路由层 ``json.dumps``）。

    时序要点：
    1. **预广播**（与 LLM 并行）：把用户/综合消息写入其它席位 thread，skip 总控自身；
    2. 跑总控 thread 的 stream，yield 每一行原始 SSE；
    3. 解析 messages → 澄清 / 空 dispatch（共识）/ 派活列表；
    4. 若有派活，**派活广播**后再 yield ``{status: dispatched, ...}``。

    ``skill_stop_names`` 会与 ``agent_orchestration`` 合并，确保 middleware 在工具调用后停图。
    """
    t0 = time.perf_counter()

    # 与上游 LLM 并行，减少「总控已开始想但席位 thread 里还没有用户消息」的窗口。
    pre_broadcast_task = asyncio.create_task(
        _broadcast(client, base, headers, "总控智能体收到用户消息:" + new_message, agent_threads, [agent_name])
    )
    thread_id = agent_threads[agent_name]
    effective_stop_names = list({"agent_orchestration", *skill_stop_names})
    body = _run_payload(
        agent_name=agent_name,
        model_name=model_name,
        thread_id=thread_id,
        new_message=new_message,
        excluded_tools=["web_search"],
        skill_stop_names=effective_stop_names,
    )
    t_payload_built = time.perf_counter()
    logger.info(
        "[leader-timing] %s payload_built=%.3fs (broadcast running in background)",
        agent_name,
        t_payload_built - t0,
    )

    captured: list[str] = []
    first_line_ts: float | None = None
    line_count = 0
    async for line, done in _stream_upstream(client, base, headers, thread_id, body):
        if line is not None:
            if first_line_ts is None:
                first_line_ts = time.perf_counter()
                logger.info(
                    "[leader-timing] %s first_upstream_line=%.3fs since_entry=%.3fs",
                    agent_name,
                    first_line_ts - t_payload_built,
                    first_line_ts - t0,
                )
            line_count += 1
            yield line
        elif done is not None:
            captured = done
    t_stream_done = time.perf_counter()
    logger.info(
        "[leader-timing] %s upstream_stream_complete=%.3fs lines=%d since_entry=%.3fs",
        agent_name,
        t_stream_done - (first_line_ts or t_payload_built),
        line_count,
        t_stream_done - t0,
    )

    messages = _parse_last_messages(captured)
    # Pre-broadcast must finish before we yield the final status frame on any
    # branch — the seats that get dispatched next rely on having seen the
    # original user message. The upstream LLM stream usually takes several
    # seconds, so the broadcast has typically already completed and this
    # await is effectively free.
    t_before_broadcast_await = time.perf_counter()
    try:
        await pre_broadcast_task
    except Exception as exc:
        logger.warning("pre-broadcast failed: %s", exc)
    t_after_broadcast_await = time.perf_counter()
    logger.info(
        "[leader-timing] %s pre_broadcast_await=%.3fs (>0 means broadcast was still in flight after LLM finished)",
        agent_name,
        t_after_broadcast_await - t_before_broadcast_await,
    )

    dispatch_msg = _find_dispatch_message(messages)
    leader_text = _flatten_content(dispatch_msg.get("content", ""))
    tool_calls = dispatch_msg.get("tool_calls") or []

    # 澄清分支：与 intent.py 相同字段；status 为字符串 "clarification"（非数组），
    # 前端据此暂停 runOrchestration 循环，展示 Step 2 澄清条。
    clarification_call = next(
        (tc for tc in tool_calls if (tc.get("name") or "").lower() == "ask_clarification"),
        None,
    )
    if clarification_call is not None:
        cargs = clarification_call.get("args") or {}
        question = str(cargs.get("question") or "").strip()
        follow_up = _find_trailing_ai_text(messages, exclude_text=leader_text)
        yield {
            "status": "clarification",
            "content": follow_up or question or leader_text,
            "question": question,
            "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),
            "agent_name": agent_name,
        }
        return

    if not tool_calls:
        # 空 status 数组 = 总控不再派活，前端视为共识（hasConsensus）。
        yield {"status": [], "content": leader_text, "agent_name": agent_name}
        return

    # 派活广播：告知各席位「总控给 X 派了什么活」，与即将发起的 special run 配合。
    # sets and we don't care about ordering, so serializing them just
    # multiplied latency before the frontend could start a sub-agent run.
    dispatched: list[list[str]] = []
    dispatch_broadcasts: list[asyncio.Task[None]] = []
    for tool_call in tool_calls:
        try:
            sub_name, task_text = _extract_dispatch(tool_call)
            if not sub_name:
                continue
            dispatch_broadcasts.append(
                asyncio.create_task(
                    _broadcast(
                        client,
                        base,
                        headers,
                        f"总控智能体给{sub_name}派活,内容为:" + task_text,
                        agent_threads,
                        [agent_name, sub_name],
                    )
                )
            )
            dispatched.append([sub_name, task_text])
        except Exception as exc:
            logger.warning("dispatch failed: %s", exc)
    if dispatch_broadcasts:
        t_before_dispatch = time.perf_counter()
        await asyncio.gather(*dispatch_broadcasts, return_exceptions=True)
        logger.info(
            "[leader-timing] %s dispatch_broadcasts_await=%.3fs count=%d (gates how soon the next sub agent can start)",
            agent_name,
            time.perf_counter() - t_before_dispatch,
            len(dispatch_broadcasts),
        )
    yield {"status": dispatched, "content": leader_text, "agent_name": agent_name}


async def _special_run(
    client: httpx.AsyncClient,
    base: str,
    headers: dict[str, str],
    agent_threads: dict[str, str],
    agent_name: str,
    new_message: str,
    skill_stop_names: list[str],
    model_name: str,
) -> AsyncIterator[str | dict[str, Any]]:
    """子智能体（special）单轮：执行总控下发的 task 文本，交付 content。

    - ``excluded_tools`` 禁用派活/澄清等，避免席位抢总控活；
    - 终态 ``status`` 为描述性字符串（非数组），``content`` 为交付正文；
    - **后广播**在 yield 终态帧之后 await，保证其它席位 state 里有完成摘要，
      同时尽量让前端先收到结果关闭 loading。
    """
    t0 = time.perf_counter()
    thread_id = agent_threads[agent_name]
    body = _run_payload(
        agent_name=agent_name,
        model_name=model_name,
        thread_id=thread_id,
        new_message=new_message,
        excluded_tools=["web_search", "ask_clarification", "present_files", "view_image", "agent_orchestration"],
        skill_stop_names=skill_stop_names,
    )
    t_payload_built = time.perf_counter()
    logger.info(
        "[special-timing] %s payload_built=%.3fs",
        agent_name,
        t_payload_built - t0,
    )

    captured: list[str] = []
    first_line_ts: float | None = None
    line_count = 0
    async for line, done in _stream_upstream(client, base, headers, thread_id, body):
        if line is not None:
            if first_line_ts is None:
                first_line_ts = time.perf_counter()
                logger.info(
                    "[special-timing] %s first_upstream_line=%.3fs since_entry=%.3fs",
                    agent_name,
                    first_line_ts - t_payload_built,
                    first_line_ts - t0,
                )
            line_count += 1
            yield line
        elif done is not None:
            captured = done
    t_stream_done = time.perf_counter()
    logger.info(
        "[special-timing] %s upstream_stream_complete=%.3fs lines=%d since_entry=%.3fs",
        agent_name,
        t_stream_done - (first_line_ts or t_payload_built),
        line_count,
        t_stream_done - t0,
    )

    messages = _parse_last_messages(captured)
    # Sub-agent's final deliverable is the latest AIMessage's text content.
    last_msg: dict[str, Any] = {}
    for msg in reversed(messages):
        if isinstance(msg, dict) and (msg.get("type") or "").lower() in ("ai", "aimessage", "aimessagechunk"):
            last_msg = msg
            break
    content = _flatten_content(last_msg.get("content", ""))
    # Kick the "子智能体完成" broadcast off in the background and yield the
    # status frame first — the next leader cycle that depends on this
    # broadcast having landed only starts after a fresh /run/stream request
    # from the frontend, which is many seconds away (full leader LLM call).
    # Awaiting after the yield ensures the task finishes before the AsyncClient
    # closes; the SSE connection stays open a bit longer but the frontend has
    # already consumed the result.
    t_before_post = time.perf_counter()
    post_broadcast_task = asyncio.create_task(
        _broadcast(
            client,
            base,
            headers,
            f"子智能体{agent_name}完成{new_message}工作,交付内容为:" + content,
            agent_threads,
            [agent_name],
        )
    )
    yield {
        "status": f"子智能体{agent_name}完成{new_message}工作",
        "content": content,
        "agent_name": agent_name,
    }
    try:
        await post_broadcast_task
    except Exception as exc:
        logger.warning("post-broadcast failed: %s", exc)
    logger.info(
        "[special-timing] %s post_broadcast_total=%.3fs (broadcast time after yield — keeps SSE open this long)",
        agent_name,
        time.perf_counter() - t_before_post,
    )


# ---------------------------------------------------------------------------
# POST /api/multi-agent/run/stream — leader / special 单轮 SSE
# ---------------------------------------------------------------------------


@router.post("/run/stream")
async def run_multi_agent_stream(request: Request, payload: dict = Body(...)) -> StreamingResponse:
    """驱动总控或子智能体的一轮 LangGraph run。

    请求体::

        {
          "agent_type": "leader" | "special",
          "d_agent_thread_id": { "<agent_id>": "<thread_id>", ... },  // init_done.thread_ids
          "agent_name": "<本次运行的 agent_id>",
          "new_message": "<用户消息或派活 task 文本>",
          "model": "<可选>",       // 不传由 lead_agent 回退到 config.yaml models[0]
          "skill_stop_names": []   // 可选,leader 一般留空,由网关注入 agent_orchestration
        }

    响应：上游 SSE 行透传 + 末尾一条 ``data:`` JSON（leader 为 dict，special 同上）。
    客户端应使用 AbortSignal 支持「挂起协商」时取消 in-flight 请求。
    """
    agent_type: str = str(payload.get("agent_type") or "leader").lower()
    agent_threads: dict[str, str] = payload.get("d_agent_thread_id") or {}
    agent_name: str = str(payload.get("agent_name") or "").lower()
    new_message: str = str(payload.get("new_message") or "")
    skill_stop_names: list[str] = payload.get("skill_stop_names") or []
    # 不写死任何默认模型;空串会让 lead_agent._resolve_model_name() 回退到
    # config.yaml 的 models[0],与 /api/ai-writing 等其它路由一致。
    model_name: str = str(payload.get("model") or "")

    if not agent_name or agent_name not in agent_threads:
        raise HTTPException(
            status_code=400,
            detail=f"agent_name '{agent_name}' is missing from d_agent_thread_id",
        )
    _validate_agent_id(agent_name, "agent_name")
    for key in agent_threads:
        _validate_agent_id(key, "d_agent_thread_id key")

    base = _loopback_base(request)
    headers = _auth_headers(request)

    async def stream_generator() -> AsyncIterator[str]:
        # TCP loopback 到 127.0.0.1:<gateway port>。client 配置见 _make_loopback_client：
        # trust_env=False 防 HTTP_PROXY 拦截,timeout=None 容纳 SSE 长连接。
        async with _make_loopback_client(request) as client:
            try:
                if agent_type == "leader":
                    gen = _leader_run(
                        client, base, headers, agent_threads, agent_name, new_message, skill_stop_names, model_name,
                    )
                else:
                    gen = _special_run(
                        client, base, headers, agent_threads, agent_name, new_message, skill_stop_names, model_name,
                    )
                async for chunk in gen:
                    if isinstance(chunk, dict):
                        yield f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n"
                    else:
                        yield chunk if chunk.endswith("\n") else chunk + "\n"
            except HTTPException as exc:
                logger.warning("multi-agent stream HTTPException: %s %s", exc.status_code, exc.detail)
                err_frame = {
                    "status": "error",
                    "content": f"[{exc.status_code}] {exc.detail}",
                    "agent_name": agent_name,
                }
                yield f"data: {json.dumps(err_frame, ensure_ascii=False)}\n\n"
            except Exception as exc:
                logger.exception("multi-agent stream unexpected failure")
                err_frame = {
                    "status": "error",
                    "content": f"unexpected: {exc!r}",
                    "agent_name": agent_name,
                }
                yield f"data: {json.dumps(err_frame, ensure_ascii=False)}\n\n"

    return StreamingResponse(
        stream_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",
            "Connection": "keep-alive",
        },
    )