接口联调

offline-backend-20260512/backend/app/gateway/routers/multi_agent.py

@@ -21,6 +21,7 @@ import asyncio
 import json
 import logging
 import re
+import time
 from typing import Any, AsyncIterator
 
 import httpx
@@ -127,14 +128,68 @@ async def _broadcast(
 
 _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 _extract_name_task(text: str) -> tuple[str, str]:
-    match = _TOOL_CALL_PATTERN.search(text)
+
+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 match.group(1), match.group(2)
+        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
@@ -195,7 +250,22 @@ def _run_payload(
 
 
 @router.post("/init")
-async def init_multi_agent(request: Request, payload: dict = Body(...)) -> dict[str, Any]:
+async def init_multi_agent(request: Request, payload: dict = Body(...)) -> StreamingResponse:
+    """Streaming /init.
+
+    Emits text/event-stream frames as each seat/coordinator becomes ready, so
+    the frontend can visualize the parallel bring-up rather than blocking on
+    one bulk JSON. Frames:
+
+      data: {"event":"seat_ready","agent_name":"...","thread_id":"...","display_name":"...","description":"..."}
+      data: {"event":"coordinator_ready","main_agent_name":"...","thread_id":"..."}
+      data: {"event":"init_done","main_agent_name":"...","thread_ids":{...}}
+      data: {"event":"error","detail":"..."}                 # on failure
+
+    Validation errors (missing agent_names, bad id pattern) still raise
+    HTTPException synchronously so the connection fails fast with a real 4xx
+    instead of a 200 SSE that just carries an error frame.
+    """
     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()
@@ -204,61 +274,112 @@ async def init_multi_agent(request: Request, payload: dict = Body(...)) -> dict[
     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)
 
-    async with httpx.AsyncClient(timeout=None) as client:
-        # Optimization A: fetch every sub-agent's metadata and create its
-        # thread concurrently. With 6 seats this turns ~12 sequential
-        # loopback round-trips into one wave of 12 parallel ones.
-        async def _prepare_sub_agent(sub_id: str) -> tuple[str, str, str]:
-            info = await _get_agent(client, base, headers, sub_id)
-            sub_name = str(info["name"]).lower()
-            thread_id = await _create_thread(client, base, headers)
-            return sub_name, thread_id, info.get("description", "") or ""
+    def _frame(payload: dict[str, Any]) -> str:
+        return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
 
-        prepared = await asyncio.gather(*(_prepare_sub_agent(sid) for sid in agent_names))
+    async def stream_generator() -> AsyncIterator[str]:
+        async with httpx.AsyncClient(timeout=None) 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
 
-        agent_threads: dict[str, str] = {sub_name: thread_id for sub_name, thread_id, _ in prepared}
-        descriptions: list[str] = [
-            f"agent_name:{sub_name},{description}" for sub_name, _, description in prepared
-        ]
+                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
 
-        soul = (
-            "#核心定位：\n"
-            "用途：根据用户的任务目标,调用其他智能体干活,只负责协调汇总,"
-            "当你觉得已有信息支持回答用户问题时候,停止调度,并输出最后的结果\n"
-            "#调度方法：\n"
-            "通过使用agent_orchestration技能调用其他智能体干活,使用技能时传入agent_name和需要做的活\n"
-            "#可调度智能体：\n"
-            f"{descriptions}"
-        )
+                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
+                ]
 
-        agent_threads[main_agent_name] = await _create_thread(client, base, headers)
-        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:
-            raise HTTPException(status_code=500, detail=f"create coordinator agent failed: {result}")
+                soul = (
+                    "#核心定位：\n"
+                    "用途：根据用户的任务目标,调用其他智能体干活,只负责协调汇总,"
+                    "当你觉得已有信息支持回答用户问题时候,停止调度,并输出最后的结果\n"
+                    "#调度方法：\n"
+                    "通过使用agent_orchestration技能调用其他智能体干活,使用技能时传入agent_name和需要做的活。"
+                    "注意：agent_name 必须严格使用下方列出的英文 id（例如 roundtable-intelligence），"
+                    "不要把括号里的中文显示名拼进 agent_name 里。\n"
+                    "#可调度智能体：\n"
+                    + "\n".join(descriptions)
+                )
 
-        # Optimization C: the sub-agent SOULs already declare "你只在被点到时
-        # 发言,由总控协调智能体派活", so the old intro broadcast (one human
-        # message per thread, 6 × GET-state + 6 × POST-state) is redundant.
-        # Skipping it removes the slowest stage of init entirely.
+                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
 
-    return {
-        "status": "success",
-        "main_agent_name": main_agent_name,
-        "thread_ids": agent_threads,
-    }
+                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",
+        },
+    )
 
 
 # ---------------------------------------------------------------------------
@@ -277,37 +398,112 @@ async def _stream_upstream(
 
     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_message(captured: list[str]) -> dict[str, Any]:
-    if len(captured) < 2:
-        raise HTTPException(
-            status_code=502,
-            detail=f"upstream produced no usable data lines (got {len(captured)})",
-        )
-    raw = captured[-2][5:].strip()
-    try:
-        parsed = json.loads(raw)
-    except json.JSONDecodeError as exc:
-        raise HTTPException(status_code=502, detail=f"failed to parse upstream frame: {exc}") from exc
-    messages = parsed.get("messages") or []
-    if not messages:
-        return {}
-    return messages[-1] or {}
+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(
@@ -320,48 +516,144 @@ async def _leader_run(
     skill_stop_names: list[str],
     model_name: str,
 ) -> AsyncIterator[str | dict[str, Any]]:
-    await _broadcast(client, base, headers, "总控智能体收到用户消息:" + new_message, agent_threads, [agent_name])
+    # Fine-grained timing for the leader path. The user reported the
+    # coordinator "stalls before output" — we log each major await so we can
+    # see whether the wait is in pre-broadcast scheduling, upstream stream
+    # setup, first-token latency, or post-stream parsing.
+    t0 = time.perf_counter()
+
+    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=skill_stop_names,
+        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 []
+
+    # Clarification branch: the coordinator called ask_clarification. The
+    # ClarificationMiddleware intercepts with Command(goto=END) and stashes
+    # the structured question in the tool_call args; sometimes a richer
+    # follow-up AIMessage is also produced after the ToolMessage. Emit a
+    # dedicated frame so the frontend can render it as a question instead of
+    # treating an empty dispatch as "consensus reached".
+    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)),
+            "agent_name": agent_name,
+        }
+        return
 
-    last_msg = _parse_last_message(captured)
-    leader_text = _flatten_content(last_msg.get("content", ""))
-    tool_calls = last_msg.get("tool_calls") or []
     if not tool_calls:
         yield {"status": [], "content": leader_text, "agent_name": agent_name}
         return
 
+    # Run all dispatch broadcasts concurrently — they target disjoint thread
+    # 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_name_task(str(tool_call.get("args", "")))
+            sub_name, task_text = _extract_dispatch(tool_call)
             if not sub_name:
                 continue
-            await _broadcast(
-                client,
-                base,
-                headers,
-                f"总控智能体给{sub_name}派活,内容为:" + task_text,
-                agent_threads,
-                [agent_name, sub_name],
+            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}
 
 
@@ -375,38 +667,90 @@ async def _special_run(
     skill_stop_names: list[str],
     model_name: str,
 ) -> AsyncIterator[str | dict[str, Any]]:
+    # Same instrumentation as _leader_run so all 6 seats are diagnosable.
+    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"],
+        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,
+    )
 
-    last_msg = _parse_last_message(captured)
+    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", ""))
-    await _broadcast(
-        client,
-        base,
-        headers,
-        f"子智能体{agent_name}完成{new_message}工作,交付内容为:" + content,
-        agent_threads,
-        [agent_name],
+    # 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,
+    )
 
 
 @router.post("/run/stream")
@@ -423,6 +767,9 @@ async def run_multi_agent_stream(request: Request, payload: dict = Body(...)) ->
             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)

offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/skill_stop_middleware.py

@@ -78,21 +78,35 @@ class SkillStopMiddleware(AgentMiddleware[SkillStopMiddlewareState]):
         """Return the matched skill name if this tool call invokes one of the
         configured target skills, otherwise ``None``.
 
-        Matching today is bash-based: the lead agent reaches into a skill by
-        running ``python3 scripts/agent_orchestration.py "<skill-name>" "<task>"``,
-        so we only look at ``bash`` calls and extract the skill name from the
-        argv-style quoted form.
+        Two invocation patterns are recognized:
+
+        1. **Structured tool call** — the model emits a tool call whose ``name``
+           literally matches one of the configured skill names (e.g. the
+           multi-agent leader calling
+           ``agent_orchestration({"agent_name": "...", "task": "..."})``).
+           This is the natural form most LLMs produce when told to "use a
+           skill", and it's the form that drives the roundtable dispatch.
+
+        2. **Bash + argv** — the legacy form where the lead agent reaches into
+           a skill by running ``python3 scripts/<skill-name>.py "<arg1>" "<arg2>"``.
+           Kept for backward compatibility with skills that use bash for
+           dispatch.
         """
         name = tool_call.get("name") or ""
-        skill_name = ""
+        lowered_targets = {s.lower() for s in self._skill_names}
+
+        # Pattern 1: structured tool call whose tool name IS the skill name.
+        if name and (name in self._skill_names or name.lower() in lowered_targets):
+            logger.info("SkillStopMiddleware: matched structured skill call name=%s", name)
+            return name
+
+        # Pattern 2: bash invocation with argv-style skill name.
         if name == "bash" and "args" in tool_call:
             skill_name, _ = extract_name_task(str(tool_call["args"]))
-        if not skill_name:
-            return None
-        lowered_targets = {s.lower() for s in self._skill_names}
-        if skill_name in self._skill_names or skill_name.lower() in lowered_targets:
-            logger.info("SkillStopMiddleware: matched skill_name=%s", skill_name)
-            return skill_name
+            if skill_name and (skill_name in self._skill_names or skill_name.lower() in lowered_targets):
+                logger.info("SkillStopMiddleware: matched bash skill call name=%s", skill_name)
+                return skill_name
+
         return None
 
     def _handle_skill_stop(self, request: ToolCallRequest, skill_name: str) -> Command:

offline-backend-20260512/backend/packages/harness/deerflow/tools/builtins/__init__.py

@@ -1,3 +1,4 @@
+from .agent_orchestration_tool import agent_orchestration_tool
 from .clarification_tool import ask_clarification_tool
 from .memory_tool import memory_tool
 from .present_file_tool import present_file_tool
@@ -10,6 +11,7 @@ __all__ = [
     "setup_agent",
     "present_file_tool",
     "ask_clarification_tool",
+    "agent_orchestration_tool",
     "memory_tool",
     "scheduled_task_tool",
     "view_image_tool",

offline-backend-20260512/backend/packages/harness/deerflow/tools/builtins/agent_orchestration_tool.py

@@ -0,0 +1,46 @@
+"""Placeholder tool that lets the roundtable leader agent dispatch sub-agents.
+
+Background
+----------
+The multi-agent ``/api/multi-agent/run/stream`` endpoint (``app.gateway.routers.multi_agent``)
+drives a *leader* agent whose only job is to delegate work to seat-specific sub-agents.
+Its SOUL.md tells the model to "use the agent_orchestration skill, passing agent_name
+and task". With no real tool registered under that name, the model used to emit
+``{"name": "agent_orchestration", "args": {"agent_name": ..., "task": ...}}`` calls
+that the runtime rejected with "agent_orchestration is not a valid tool", driving it
+into an unproductive retry loop (a single leader call could exceed 10 000 SSE frames).
+
+This module exposes a no-op tool with the exact signature the leader naturally
+emits. The actual dispatch is handled by ``SkillStopMiddleware`` which intercepts
+the call, halts the run, and lets ``multi_agent.py`` parse the AIMessage's
+``tool_calls`` to extract ``(agent_name, task)`` pairs for the frontend.
+"""
+
+from langchain.tools import tool
+
+
+@tool("agent_orchestration", parse_docstring=True)
+def agent_orchestration_tool(agent_name: str, task: str) -> str:
+    """Delegate a task to a named sub-agent in the multi-agent roundtable.
+
+    Use this tool ONLY when you are the roundtable coordinator and need to
+    dispatch work to one of the configured seat agents. Each call delegates
+    ONE task to ONE agent. Emit multiple calls in the same turn to dispatch
+    several seats in a single round of the discussion.
+
+    The call is intercepted by the runtime — you do not get a real result
+    back synchronously. The orchestrator will collect every dispatched
+    ``(agent_name, task)`` pair, run the corresponding sub-agents, then
+    re-prompt you with a synthesis instruction once they finish.
+
+    Args:
+        agent_name: Name of the sub-agent to dispatch. MUST exactly match one of
+            the names listed in your SOUL prompt's ``可调度智能体`` section.
+        task: The instruction to hand to that sub-agent. Be specific — the
+            sub-agent only sees this string (plus its own SOUL prompt), not the
+            broader conversation context.
+    """
+    # Real handling is in SkillStopMiddleware + the multi_agent router. This
+    # body only runs as a fallback if the middleware is not configured; it
+    # returns a placeholder so the model does not see a "not a valid tool" error.
+    return f"dispatched:{agent_name}"

offline-backend-20260512/backend/packages/harness/deerflow/tools/tools.py

@@ -6,7 +6,7 @@ from deerflow.config import get_app_config
 from deerflow.config.app_config import AppConfig
 from deerflow.reflection import resolve_variable
 from deerflow.sandbox.security import is_host_bash_allowed
-from deerflow.tools.builtins import ask_clarification_tool, memory_tool, present_file_tool, scheduled_task_tool, task_tool, view_image_tool
+from deerflow.tools.builtins import agent_orchestration_tool, ask_clarification_tool, memory_tool, present_file_tool, scheduled_task_tool, task_tool, view_image_tool
 from deerflow.tools.builtins.tool_search import reset_deferred_registry
 
 logger = logging.getLogger(__name__)
@@ -14,6 +14,7 @@ logger = logging.getLogger(__name__)
 BUILTIN_TOOLS = [
     present_file_tool,
     ask_clarification_tool,
+    agent_orchestration_tool,
 ]
 
 SUBAGENT_TOOLS = [