Merge branch 'master' of http://192.168.0.200/knowledge-base-bj/deerflow-server into xuhd

2026-06-01 10:49:53 +08:00
commit 22b93e32eb
@@ -512,11 +512,17 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
                _mark_legacy_sync_fresh,
                _sync_legacy_agents,
            )
+            from app.gateway.services import _mark_legacy_sync_fresh as _services_mark_sync_fresh

            agent_store = getattr(app.state, "agent_store", None)
            if agent_store is not None:
                await _sync_legacy_agents(agent_store)
                _mark_legacy_sync_fresh()
+                # ``services.py`` keeps an independent TTL gate (used by
+                # ``start_run`` → ``_enforce_agent_access``). Without this
+                # ack, the first ``/runs/stream`` after startup would
+                # re-run the heavy sync even though we just did it.
+                _services_mark_sync_fresh()
                logger.info("Legacy folder-backed agents synced at startup")
        except Exception:
            logger.exception("Legacy agent sync failed at startup (non-fatal)")
@@ -2,21 +2,35 @@

 适用场景
 --------
-``roundtable-intent`` / ``roundtable-recommender`` 是圆桌 Step 1 + 推荐链路的
-**功能型** 内置 agent,任何部署都必须存在。但仓库根 ``.gitignore`` 把整个
-``.deer-flow/`` 目录排除,新机器 ``git clone`` 之后这两个目录**不会存在**,
-直接调 ``/api/intent/stream`` 会因为 agent 解析失败抛 500。
+``roundtable-intent`` / ``roundtable-recommender`` / ``roundtable-coordinator``
+是圆桌 Step 1 + 推荐 + Step 2 总控链路的**功能型** 内置 agent,任何部署都必须
+存在。但仓库根 ``.gitignore`` 把整个 ``.deer-flow/`` 目录排除,新机器
+``git clone`` 之后这些目录**不会存在**,直接调对应 endpoint 会因为 agent
+解析失败抛 500。

-本模块在三个路由入口(``intent.init``、``intent.stream``、``recommend.stream``)
-被调用,检查磁盘文件是否齐全;缺失则用下方嵌入的标准模板写回去。
-``_sync_legacy_agents`` 会在下一次 ``GET /api/agents`` 时把目录 upsert 进 DB,
-所以这里只关心磁盘文件,不直接动 DB。
+本模块在四个路由入口(``intent.init``、``intent.stream``、``recommend.stream``、
+``multi_agent.init``)被调用,检查磁盘文件是否齐全;缺失则用下方嵌入的标准模板
+写回去。``_sync_legacy_agents`` 会在下一次 ``GET /api/agents`` 时把目录 upsert
+进 DB,所以这里只关心磁盘文件,不直接动 DB。

 为什么嵌入字符串而不是用 importlib.resources 资源文件
 ------------------------------------------------------
-SOUL 体量不大(两个加起来 ~200 行),嵌入到 Python 文件里是**单一可信源**——改
+SOUL 体量不大(三个加起来 ~400 行),嵌入到 Python 文件里是**单一可信源**——改
 SOUL 走代码 review、commit 走 git 历史,且不会因为 wheel 打包漏配
 ``package-data`` 导致部署机拿不到。
+
+为什么 ``roundtable-coordinator`` 是单例而不是每次会话新建
+------------------------------------------------------------
+历史实现每次 Step 2 init 都用 ``roundtable-coordinator-<timestamp36>`` 这种
+带时间戳的临时 id 创建新 agent,把"本次可调度席位列表"写进动态拼接的 SOUL。
+长期跑下来 ``.deer-flow/agents/`` 累积了几十上百个完全相同的 coordinator
+壳,管理页非常杂乱。
+
+改造后 coordinator 是 **单例**(id 固定为 ``roundtable-coordinator``):
+- SOUL 是**通用协调规则**,不含本次席位列表;
+- 本次圆桌可调度的席位清单由 ``multi_agent.init`` 在创建好 coordinator
+  thread 后,以 human 消息的形式 append 到 thread 历史,让 leader run 自然
+  读到。这样 SOUL 长期稳定,席位列表随会话动态注入。
 """

 from __future__ import annotations
@@ -476,17 +490,91 @@ RECOMMENDER_AGENT_SOUL = """# 智能体推荐专员
 """


-_BUILTIN_FUNCTIONAL_AGENTS: tuple[tuple[str, str, str, str], ...] = (
-    (INTENT_AGENT_ID, INTENT_AGENT_NAME, INTENT_AGENT_DESCRIPTION, INTENT_AGENT_SOUL),
-    (RECOMMENDER_AGENT_ID, RECOMMENDER_AGENT_NAME, RECOMMENDER_AGENT_DESCRIPTION, RECOMMENDER_AGENT_SOUL),
+COORDINATOR_AGENT_ID = "roundtable-coordinator"
+COORDINATOR_AGENT_NAME = "总控协调"
+COORDINATOR_AGENT_DESCRIPTION = (
+    "圆桌总控协调智能体,基于「任务目标 + 当前会话已选席位列表」决定派活给哪些子智能体、"
+    "汇总它们的交付,达成共识后输出最终方案。本身不调研、不写文档,只负责协调与综合。"
+)
+
+# Coordinator SOUL 是**通用版本**——只规定协调规则、工具用法、终止条件,
+# **不**包含本次会话的席位列表。席位列表由 multi_agent.init 在创建 coordinator
+# thread 时以 human 消息的形式 append 进对话历史,leader run 每次都能看到。
+# 这样 SOUL 单例 + 席位列表动态注入,避免每次 init 重写 SOUL 文件。
+COORDINATOR_AGENT_SOUL = """# 圆桌总控协调智能体
+
+## 核心定位
+你是多智能体圆桌研讨 **Step 2** 的「总控协调智能体」。你不亲自调研、不写文档、不联网搜索,你的**唯一职责**是:
+
+1. **理解用户任务目标**(由 Step 1 的任务理解专员整理交给你);
+2. **从本次圆桌已上线的席位中挑合适的派活**(通过 `agent_orchestration` 技能);
+3. **汇总子智能体交付**,达成共识后输出最终结论。
+
+## 调度方法
+通过 `agent_orchestration` 技能调用其他智能体干活,使用技能时传入 `agent_name` 和需要做的活。
+
+**⚠️ 关于 `agent_name` 的硬性规则**:
+- 必须严格使用**本次圆桌已上线席位的英文 id**(例如 `roundtable-intelligence`,或形如 `02714cbe6e64406a8a4659f0e3d9c0b8` 这样的 UUID)
+- 本次可调度的席位清单会以「【圆桌系统初始化】本次可调度席位:...」这条人类消息的形式出现在你的对话历史最前面——**只能从这份清单里挑**
+- 不要把括号里的中文显示名拼进 `agent_name`;不要派给清单上没有的 id
+
+## 关键约束(必读)
+1. **不要对同一子智能体重复派同样的活**:每次派活前回顾对话历史,如果某个子智能体已经针对你正要派的任务完成过交付(看到形如「子智能体[名称]完成XX工作,交付内容为:...」的 human 消息),就直接基于已有交付推进,**不要重复派活**,除非你确实有**具体的、不一样的**补充问题
+2. **看到交付要会用**:子智能体的交付内容会作为 human 消息追加到你的对话历史里,你能完整看见。当各方信息已足够形成结论时,**直接综合输出最终方案**(此时不再调用任何工具),让讨论收敛
+3. **避免无意义循环**:如果连续两轮都没有新信息流入,立刻停止派活并基于现有素材给出最终结论
+4. **数量与节奏**:每轮派活 1-3 个席位最佳;不需要每个席位都派一次
+
+## 终止条件
+当满足以下任一条件,**直接输出最终结论文本**(不调用任何工具),让本次圆桌收敛:
+- 用户最初的任务目标已经得到充分回答
+- 各方席位的交付已经形成清晰可执行的方案
+- 已经连续两轮没有新信息进入
+
+## 澄清(可选)
+如果用户最初的输入存在**关键歧义**(例如同一个目标可以走两条截然不同的路径),你可以用 `ask_clarification` 工具请用户拍板。**但请克制使用**——大多数情况下你应该基于现有信息开始派活,而不是反过来追问用户。
+
+## 输出风格
+- 派活说明:简短(1-2 句话)告诉子智能体你期望它做什么、关注哪些点
+- 最终综合:结构化,明确各席位的核心结论、冲突点、最终建议
+- 适度 emoji:📋 汇总、⚡ 派活、✅ 完成
+
+## 禁止事项
+- ❌ 亲自调研、写文档、写代码、联网搜索(这些是子智能体的活)
+- ❌ 派活给本次清单**没有**的 agent_name
+- ❌ 把括号里的中文显示名拼进 `agent_name`
+- ❌ 已经达成共识后仍然继续派活
+"""
+
+
+# (id, name, description, soul, skills) — skills 为 None 表示不写 skills 字段。
+_BUILTIN_FUNCTIONAL_AGENTS: tuple[tuple[str, str, str, str, list[str] | None], ...] = (
+    (INTENT_AGENT_ID, INTENT_AGENT_NAME, INTENT_AGENT_DESCRIPTION, INTENT_AGENT_SOUL, None),
+    (RECOMMENDER_AGENT_ID, RECOMMENDER_AGENT_NAME, RECOMMENDER_AGENT_DESCRIPTION, RECOMMENDER_AGENT_SOUL, None),
+    (
+        COORDINATOR_AGENT_ID,
+        COORDINATOR_AGENT_NAME,
+        COORDINATOR_AGENT_DESCRIPTION,
+        COORDINATOR_AGENT_SOUL,
+        ["agent_orchestration"],
+    ),
 )


-def _ensure_one(agent_id: str, name: str, description: str, soul: str) -> bool:
+def _ensure_one(
+    agent_id: str,
+    name: str,
+    description: str,
+    soul: str,
+    skills: list[str] | None = None,
+) -> bool:
    """Create the on-disk files for one functional agent if missing.

    Returns True if files were just created, False if the directory already
    existed (and we didn't touch it — respect any local edits).
+
+    ``skills`` 写入 config.yaml 顶层 ``skills`` 字段,与 routers/agents.py 的
+    ``_write_agent_config`` 输出形状对齐;coordinator 需要 ``agent_orchestration``
+    才能派活,intent / recommender 不需要任何 skill(传 None 即可)。
    """
    agent_dir = get_paths().agent_dir(agent_id)
    if agent_dir.exists():
@@ -496,8 +584,13 @@ def _ensure_one(agent_id: str, name: str, description: str, soul: str) -> bool:

    # newline="\n" 强制 LF;否则 Windows 下 open() 把 \n 翻成 \r\n,跨 OS 部署时与
    # Linux 部署机字节级不一致(功能上 YAML/Markdown 都兼容,但 git diff / 哈希对不齐)。
+    config_payload: dict[str, object] = {"id": agent_id, "name": name, "description": description}
+    if skills:
+        # 与 routers/agents.py::_write_agent_config 行为一致:None 不写、空列表也不写,
+        # 仅当确有 skill 名时才追加顶层 skills 字段。
+        config_payload["skills"] = list(skills)
    config_yaml = yaml.dump(
-        {"id": agent_id, "name": name, "description": description},
+        config_payload,
        default_flow_style=False,
        allow_unicode=True,
    )
@@ -520,9 +613,9 @@ def ensure_roundtable_functional_agents() -> list[str]:
    actionable error if the agent really cannot be resolved.
    """
    created: list[str] = []
-    for agent_id, name, description, soul in _BUILTIN_FUNCTIONAL_AGENTS:
+    for agent_id, name, description, soul, skills in _BUILTIN_FUNCTIONAL_AGENTS:
        try:
-            if _ensure_one(agent_id, name, description, soul):
+            if _ensure_one(agent_id, name, description, soul, skills):
                created.append(agent_id)
        except Exception:
            logger.exception("Failed to seed roundtable functional agent '%s'", agent_id)
@@ -435,14 +435,19 @@ async def list_agents(
            agents = await store.list_all()
        else:
            agents = await store.list_visible(user_id)
-        # List endpoint stays lean: skip per-agent SOUL.md read, config.yaml
-        # parse, and dir self-heal — none of those feed UI fields the gallery /
-        # recommend dialog / selector actually render. Detail endpoints (single
-        # agent + create/update responses) still load them.
-        responses = [
-            _record_to_response(a, include_soul=False, include_extras=False)
-            for a in agents
-        ]
+        # Load lightweight extras (skills / model / tool_groups from config.yaml)
+        # so the gallery cards can render skill badges. SOUL.md is still skipped
+        # (only the edit dialog reads it). The per-agent config.yaml reads run in
+        # worker threads concurrently, so the event loop stays responsive even
+        # with many agents.
+        responses = list(
+            await asyncio.gather(
+                *(
+                    _record_to_response_async(a, include_soul=False, include_extras=True)
+                    for a in agents
+                )
+            )
+        )

        assignments = await tag_store.list_assignments("agent", [r.id for r in responses])
        favorite_ids = await store.list_favorite_ids(user_id)
@@ -57,6 +57,10 @@ import uuid
 from typing import Any, AsyncIterator

 import httpx
+from app.gateway.routers._roundtable_seed import (
+    COORDINATOR_AGENT_ID,
+    ensure_roundtable_functional_agents,
+)
 from deerflow.config import get_app_config
 from deerflow.tools.builtins.clarification_utils import resolve_allow_multiple
 from deerflow.runtime.user_context import get_effective_user_id
@@ -443,82 +447,11 @@ async def _get_agents_batch_direct(
    return out


-async def _create_coordinator_direct(
-    request: Request,
-    *,
-    agent_id: str,
-    name: str,
-    description: str,
-    soul: str,
-    model: str | None,
-    skills: list[str] | None,
-) -> tuple[bool, dict[str, Any] | str]:
-    """同进程版的协调 agent 创建。
-
-    完整复刻 ``routers/agents.create_agent_endpoint`` 的关键副作用:
-      - ``agents_api.enabled`` master switch
-      - 显示名校验(空白则 422 → 这里转 (False, msg))
-      - 重复 id 检查(409 → (False, msg))
-      - 写 ``config.yaml`` + ``SOUL.md``(用 ``asyncio.to_thread`` 防同步 IO 阻塞)
-      - ``agent_store.create`` 入表
-
-    返回 ``(True, record)`` 或 ``(False, error_text)``;调用方决定要不要发 SSE error 帧。
-    """
-    from app.gateway.deps import get_agent_store
-    from app.gateway.routers.agents import (
-        _require_agents_api_enabled,
-        _write_agent_config,
-        _write_agent_soul,
-    )
-    from deerflow.config.paths import get_paths
-
-    _require_agents_api_enabled()
-    normalized_name = (name or "").strip()
-    if not normalized_name:
-        return False, "Agent name cannot be empty"
-
-    store = get_agent_store(request)
-    if await store.get_any(agent_id) is not None:
-        return False, f"Agent id '{agent_id}' already exists"
-
-    agent_dir = get_paths().agent_dir(agent_id)
-    if agent_dir.exists():
-        return False, f"Agent id '{agent_id}' already exists"
-
-    try:
-        # config.yaml / SOUL.md 是同步 IO —— 全部丢线程池,不阻塞 init SSE 的发帧节奏。
-        await asyncio.to_thread(
-            _write_agent_config,
-            agent_id,
-            name=normalized_name,
-            description=description or "",
-            skills=skills,
-            model=model,
-            tool_groups=None,
-        )
-        await asyncio.to_thread(_write_agent_soul, agent_id, soul)
-        record = await store.create(
-            {
-                "id": agent_id,
-                "name": normalized_name,
-                "description": description or "",
-                "user_id": _resolve_user_id(request),
-                "published": False,
-            }
-        )
-        return True, record
-    except Exception as exc:
-        # 与 routers/agents 失败回滚一致 —— 文件已写出但 DB 失败时清掉 dir,避免下次
-        # `agent_dir.exists()` 误判为"已存在"导致后续 init 永远 409。
-        try:
-            if agent_dir.exists():
-                import shutil
-
-                await asyncio.to_thread(shutil.rmtree, agent_dir)
-        except Exception:
-            logger.exception("[init-direct] cleanup after coordinator create failure")
-        logger.exception("[init-direct] coordinator create failed")
-        return False, str(exc)
+# NOTE: 历史上这里曾有 _create_coordinator_direct(),每次 init 都创建一个带时间戳的
+# roundtable-coordinator-<ts36>。2026-05 改造为单例(见 _roundtable_seed.py 与本
+# 模块 init_multi_agent 的 docstring),不再需要"每次创建"的直调辅助;改造后由
+# ensure_roundtable_functional_agents() 在路由入口落地静态 SOUL,席位列表通过
+# _append_thread_message 注入到 coordinator thread 的对话历史中。


 async def _append_thread_message(
@@ -718,31 +651,44 @@ async def init_multi_agent(request: Request, payload: dict = Body(...)) -> Strea

        {
          "agent_names": ["roundtable-intelligence", ...],  // 用户选的席位 id
-          "main_agent_name": "roundtable-coordinator-xxx",  // 当次总控 id，前端生成
-          "model": "<可选>"                                  // 可选,写入 coordinator agent;不传由 lead_agent 回退到 config.yaml models[0]
+          "main_agent_name": "<可选,已废弃>",                 // 历史兼容字段,新代码忽略,统一用 COORDINATOR_AGENT_ID
+          "model": "<可选>"                                  // 可选;不传由 lead_agent 回退到 config.yaml models[0]
        }

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

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

-    协调智能体 SOUL 在运行时根据各席位 ``GET /api/agents/{id}`` 的 name/description 拼接，
-    强调 ``agent_name`` 必须用英文 id（``roundtable-intelligence``），禁止带中文括号后缀。
+    协调智能体改造说明
+    ----------------
+    协调智能体现在是 **单例**(id 固定为 ``roundtable-coordinator``,SOUL 见
+    ``_roundtable_seed.COORDINATOR_AGENT_SOUL``),由 ``ensure_roundtable_functional_agents()``
+    在首次调用时自动落地。每次 init **不再创建新的 coordinator agent**(避免
+    ``.deer-flow/agents/`` 累积 ``roundtable-coordinator-<ts36>`` 残留)。
+
+    本次圆桌的「可调度席位列表」由 init 在创建好 coordinator thread 之后,以一条
+    human 消息追加到 thread 历史里,leader run 每轮都能从对话历史里读到。这样
+    SOUL 长期稳定,席位列表随会话动态注入。前端为兼容旧 client 仍可传
+    ``main_agent_name`` 字段,但后端会忽略并使用 ``COORDINATOR_AGENT_ID``。
    """
+    # 调用前自检:与 intent/recommend 一样,缺失时就地补建,避免 /run/stream 报 500。
+    ensure_roundtable_functional_agents()
+
    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")
+    # main_agent_name 是历史字段,2026-05 改造后忽略,统一用单例 id。保留接收但不再用前端值,
+    # 避免下游再次创建带时间戳的临时 coordinator。
+    main_agent_name = COORDINATOR_AGENT_ID

    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")

    def _frame(payload: dict[str, Any]) -> str:
        return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
@@ -794,56 +740,50 @@ async def init_multi_agent(request: Request, payload: dict = Body(...)) -> Strea
            t_seats_done = time.perf_counter()

            agent_threads: dict[str, str] = {sid: thread_id for sid, thread_id, _, _ in prepared}
-            descriptions: list[str] = [
+            seat_lines: 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,"
-                "或形如 02714cbe6e64406a8a4659f0e3d9c0b8 这样的 UUID）,"
-                "不要把括号里的中文显示名拼进 agent_name 里。\n"
-                "#关键约束(必读)：\n"
-                "1. **不要对同一子智能体重复派同样的活**：每次派活前回顾对话历史,"
-                "如果某个子智能体已经针对你正要派的任务完成过交付("
-                "看到形如\"子智能体[名称]完成XX工作,交付内容为:...\"的 human 消息),"
-                "就直接基于已有交付推进,**不要重复派活**,除非你确实有**具体的、不一样的**补充问题。\n"
-                "2. **看到交付要会用**：子智能体的交付内容会作为 human 消息追加到你的对话历史里,"
-                "你能完整看见。当各方信息已足够形成结论时,**直接综合输出最终方案**(此时不再调用任何工具),"
-                "让讨论收敛。\n"
-                "3. **避免无意义循环**：如果连续两轮都没有新信息流入,立刻停止派活并基于现有素材给出最终结论。\n"
-                "#可调度智能体：\n"
-                + "\n".join(descriptions)
+            # 协调智能体改造为**单例**(id = roundtable-coordinator):
+            # - SOUL 已经由 ensure_roundtable_functional_agents() 在路由入口落地,
+            #   通用协调规则常驻磁盘,无需每次重写。
+            # - 本次"可调度席位清单"以 human 消息形式 append 到 coordinator
+            #   thread 历史里,leader run 每轮都能在对话历史最前面看到。
+            # 这样席位列表随会话动态变化,但磁盘上不再累积 roundtable-coordinator-<ts36>。
+            #
+            # 注意:model 字段在 lead_agent run 时通过 context.model_name 注入,
+            # 不需要写回到 agent config.yaml(否则会污染单例的配置)。
+            seat_introduction = (
+                "【圆桌系统初始化】本次研讨可调度的席位清单如下,请严格按 agent_name "
+                "通过 agent_orchestration 技能派活,不要派给不在清单内的 agent:\n"
+                + "\n".join(seat_lines)
            )

-            # 3) coordinator thread + agent 并行创建。thread 与 agent 之间无依赖,
-            #    asyncio.gather 比串行省一个直调往返的开销。
-            coord_thread_task = asyncio.create_task(
-                _create_thread_direct(request, metadata=_ROUNDTABLE_THREAD_METADATA)
-            )
-            coord_agent_task = asyncio.create_task(
-                _create_coordinator_direct(
-                    request,
-                    agent_id=main_agent_name,
-                    name=main_agent_name,
-                    description="你是总控控制协调智能体,负责协调其他智能体干活",
-                    soul=soul,
-                    model=model,
-                    skills=["agent_orchestration"],
-                )
-            )
-            coord_thread = await coord_thread_task
+            coord_thread = await _create_thread_direct(request, metadata=_ROUNDTABLE_THREAD_METADATA)
            agent_threads[main_agent_name] = coord_thread
-            ok, result = await coord_agent_task
-            if not ok and isinstance(result, str) and "exists" not in result:
-                yield _frame({"event": "error", "detail": f"create coordinator agent failed: {result}"})
-                return
+
+            # 把席位列表注入 coordinator thread 的对话历史。HTTP loopback 调用
+            # 走 /api/threads/{tid}/state(读+写),开销 ~10-30ms,在 init 整体
+            # 几百 ms 的尺度上微不足道,但比起手撕 langgraph checkpoint 内部结构
+            # 风险低、与 _broadcast / _append_thread_message 一套语义,长期可维护。
+            try:
+                base = _loopback_base(request)
+                headers = _auth_headers(request)
+                async with _make_loopback_client(request) as seed_client:
+                    await _append_thread_message(
+                        seed_client,
+                        base,
+                        headers,
+                        coord_thread,
+                        seat_introduction,
+                    )
+            except Exception:
+                # 注入失败不阻断 init:leader run 仍能跑,只是看不到席位清单,
+                # 可能派活给错的 id。会被 _strip_display_suffix / _validate_agent_id
+                # 兜底,前端能看到错误帧;运维通过 backend 日志找到根因再修复。
+                logger.exception("[init-direct] failed to seed coordinator thread with seat list")
+
            t_coord_done = time.perf_counter()
            logger.info(
                "[init-direct] meta=%.3fs seats=%.3fs coord=%.3fs total=%.3fs seats_count=%d",
@@ -11,6 +11,7 @@ import asyncio
 import json
 import logging
 import re
+import time
 from collections.abc import Mapping
 from typing import Any

@@ -100,14 +101,15 @@ _DEFAULT_ASSISTANT_ID = "lead_agent"

 # Built-in LangGraph graph IDs registered in langgraph.json.
 # These are NOT stored in the agents DB and must bypass the custom-agent access check.
-_BUILTIN_GRAPH_IDS: frozenset[str] = frozenset({"lead_agent", "canvas_agent", "ai_writing"})
-
-_RAG_CITATION_PROMPT = """当前对话已开启参考文献模式。
-如果本轮调用了搜索、知识库、检索类技能或工具，请基于本轮所有检索结果整体回答。
-你必须把同一轮内多次检索得到的资料视为同一批参考文献，按出现顺序从 [1] 开始连续编号，不要让每个检索词各自从 [1] 重新编号。
-回答正文中的关键结论必须标注对应编号，例如 [1]、[2]，多个来源可写为 [1][3][5]。
-不要引用不存在的编号，不要使用技能内部自带的编号，不要把参考文献完整列表堆在正文末尾。
-如果没有任何可用检索结果，请直接说明未检索到可靠参考资料。"""
+_BUILTIN_GRAPH_IDS: frozenset[str] = frozenset({"lead_agent", "canvas_agent", "ai_writing"})
+
+_RAG_CITATION_PROMPT = """当前对话已开启参考文献模式。
+如果本轮调用了搜索、知识库、检索类技能或工具，请基于本轮所有检索结果整体回答。
+如果你已经知道需要多个检索词，请优先在同一步里并行调用检索工具，等这些检索都返回后再回答，不要查一个检索词就先写一版答案。
+你必须把同一轮内多次检索得到的资料视为同一个 referenceBatch，按系统稍后提供的合并批次从 [1] 开始连续编号，不要让每个检索词各自从 [1] 重新编号。
+回答正文中的关键结论必须标注对应编号，例如 [1]、[2]，多个来源可写为 [1][3][5]。
+不要引用不存在的编号，不要使用技能内部自带的编号，不要把参考文献完整列表堆在正文末尾。
+如果没有任何可用检索结果，请直接说明未检索到可靠参考资料。"""


 def _prepend_prompt_prefix(graph_input: dict[str, Any], prefix: str) -> dict[str, Any]:
@@ -136,7 +138,7 @@ def _prepend_prompt_prefix(graph_input: dict[str, Any], prefix: str) -> dict[str
    return {**graph_input, "messages": new_messages}


-def _prompt_prefix_opted_in(body: Any) -> bool:
+def _prompt_prefix_opted_in(body: Any) -> bool:
    """Detect whether the caller asked for the prompt prefix to be applied."""
    context = getattr(body, "context", None)
    if isinstance(context, Mapping) and context.get("prompt_prefix_enabled"):
@@ -144,44 +146,44 @@ def _prompt_prefix_opted_in(body: Any) -> bool:
    metadata = getattr(body, "metadata", None)
    if isinstance(metadata, Mapping) and metadata.get("prompt_prefix_enabled"):
        return True
-    return False
-
-
-def _rag_mode_requested(body: Any) -> bool:
-    context = getattr(body, "context", None)
-    if isinstance(context, Mapping) and context.get("rag_mode_enabled"):
-        return True
-    metadata = getattr(body, "metadata", None)
-    if isinstance(metadata, Mapping) and (
-        metadata.get("rag_mode_enabled") or metadata.get("rag_mode")
-    ):
-        return True
-    return False
-
-
-def _thread_rag_mode_enabled(record: Any) -> bool:
-    if not isinstance(record, Mapping):
-        return False
-    metadata = record.get("metadata")
-    return isinstance(metadata, Mapping) and bool(
-        metadata.get("rag_mode_enabled") or metadata.get("rag_mode")
-    )
-
-
-def _append_invisible_prompt(graph_input: dict[str, Any], prompt: str) -> dict[str, Any]:
-    messages = graph_input.get("messages")
-    if not isinstance(messages, list) or not messages:
-        return graph_input
-    new_messages = list(messages)
-    for idx, msg in enumerate(new_messages):
-        if not isinstance(msg, HumanMessage):
-            continue
-        extra = dict(getattr(msg, "additional_kwargs", None) or {})
-        existing = str(extra.get("prompt_prefix") or "").strip()
-        extra["prompt_prefix"] = f"{existing}\n\n{prompt}".strip() if existing else prompt
-        new_messages[idx] = msg.model_copy(update={"additional_kwargs": extra})
-        break
-    return {**graph_input, "messages": new_messages}
+    return False
+
+
+def _rag_mode_requested(body: Any) -> bool:
+    context = getattr(body, "context", None)
+    if isinstance(context, Mapping) and context.get("rag_mode_enabled"):
+        return True
+    metadata = getattr(body, "metadata", None)
+    if isinstance(metadata, Mapping) and (
+        metadata.get("rag_mode_enabled") or metadata.get("rag_mode")
+    ):
+        return True
+    return False
+
+
+def _thread_rag_mode_enabled(record: Any) -> bool:
+    if not isinstance(record, Mapping):
+        return False
+    metadata = record.get("metadata")
+    return isinstance(metadata, Mapping) and bool(
+        metadata.get("rag_mode_enabled") or metadata.get("rag_mode")
+    )
+
+
+def _append_invisible_prompt(graph_input: dict[str, Any], prompt: str) -> dict[str, Any]:
+    messages = graph_input.get("messages")
+    if not isinstance(messages, list) or not messages:
+        return graph_input
+    new_messages = list(messages)
+    for idx, msg in enumerate(new_messages):
+        if not isinstance(msg, HumanMessage):
+            continue
+        extra = dict(getattr(msg, "additional_kwargs", None) or {})
+        existing = str(extra.get("prompt_prefix") or "").strip()
+        extra["prompt_prefix"] = f"{existing}\n\n{prompt}".strip() if existing else prompt
+        new_messages[idx] = msg.model_copy(update={"additional_kwargs": extra})
+        break
+    return {**graph_input, "messages": new_messages}


 # Whitelist of run-context keys that the langgraph-compat layer forwards from
@@ -196,9 +198,9 @@ _CONTEXT_CONFIGURABLE_KEYS: frozenset[str] = frozenset(
        "model_name",
        "mode",
        "thinking_enabled",
-        "reasoning_effort",
-        "rag_mode_enabled",
-        "writing_mode",
+        "reasoning_effort",
+        "rag_mode_enabled",
+        "writing_mode",
        "writing_artifact_path",
        "canvas_artifact_title",
        "canvas_artifact_type",
@@ -356,7 +358,17 @@ def _collect_requested_agent_ids(body: Any) -> set[str]:


 async def _sync_legacy_agent_records(store) -> None:
-    """Import existing folder-backed agents as built-in records."""
+    """Import existing folder-backed agents as built-in records.
+
+    Heavy: ``list_custom_agents()`` scans ``.deer-flow/agents/`` and reads
+    ``config.yaml`` + ``SOUL.md`` per directory, then ``store.ensure_builtin``
+    does one DB round-trip per agent. On a machine with N agent dirs this is
+    O(N) file IO + O(N) DB calls — historically the worst per-request hotspot
+    in ``start_run`` (observed 30s+ on machines with 100+ residual coordinator
+    shells + slow MySQL). **Never call this directly from request handlers** —
+    use :func:`_maybe_sync_legacy_agent_records` which applies the same TTL
+    gating as ``routers/agents._maybe_sync_legacy_agents``.
+    """
    from deerflow.config.agents_config import list_custom_agents

    for agent_cfg in list_custom_agents():
@@ -373,6 +385,69 @@ async def _sync_legacy_agent_records(store) -> None:
        )


+# Throttle per-request legacy-sync. Matches ``routers/agents._LEGACY_SYNC_TTL_SECONDS``
+# semantics — the real sync runs once at startup (lifespan); this is a safety
+# net for environments that bypass the lifespan hook or for freshly dropped
+# agent directories. Lifespan registers itself via the same state by calling
+# the agents router's ``_mark_legacy_sync_fresh()`` after its boot-time sync,
+# so this module's TTL state is independent (a separate gate kept here so we
+# don't import from ``routers/agents`` and create a circular import).
+_LEGACY_SYNC_TTL_SECONDS = 3600.0
+_legacy_sync_state: dict[str, Any] = {"last_at": 0.0, "lock": None}
+
+
+# Anything slower than this triggers the WARNING-level breakdown in
+# ``start_run``. 500ms gives plenty of headroom for cold caches + a few DB
+# round-trips on healthy systems while still catching the pathological
+# multi-second cases that brought us here.
+_START_RUN_SLOW_THRESHOLD_SECONDS = 0.5
+
+
+def _mark_legacy_sync_fresh() -> None:
+    """Tell this module's TTL gate that a sync just completed elsewhere.
+
+    Called by the Gateway lifespan hook after it runs the boot-time legacy
+    agent sync (``routers/agents._sync_legacy_agents``). Without this, the
+    first request after startup would race past the TTL gate here and trigger
+    a redundant rescan even though the data is already fresh.
+    """
+    _legacy_sync_state["last_at"] = time.monotonic()
+
+
+async def _maybe_sync_legacy_agent_records(store) -> None:
+    """TTL-gated wrapper around :func:`_sync_legacy_agent_records`.
+
+    Steady-state behavior: returns in microseconds inside the TTL window — no
+    filesystem walk, no DB round-trips. Outside the TTL window we serialize
+    callers under an :class:`asyncio.Lock` so a burst of concurrent
+    ``/runs/stream`` requests fans into a single rescan instead of N parallel
+    fan-outs. Pattern mirrors ``routers/agents._maybe_sync_legacy_agents``.
+    """
+    now = time.monotonic()
+    if now - float(_legacy_sync_state["last_at"]) < _LEGACY_SYNC_TTL_SECONDS:
+        return
+    lock = _legacy_sync_state["lock"]
+    if lock is None:
+        lock = asyncio.Lock()
+        _legacy_sync_state["lock"] = lock
+    async with lock:
+        # Re-check under the lock — a sibling coroutine may have just refreshed.
+        if time.monotonic() - float(_legacy_sync_state["last_at"]) < _LEGACY_SYNC_TTL_SECONDS:
+            return
+        t_sync_start = time.perf_counter()
+        await _sync_legacy_agent_records(store)
+        elapsed = time.perf_counter() - t_sync_start
+        _legacy_sync_state["last_at"] = time.monotonic()
+        # When this fires it's a "first request of the hour" warmup — log it
+        # so operators can see the spike isn't an isolated bug. Anything
+        # over 2s usually means a lot of agent directories + slow IO/DB.
+        logger.info(
+            "[start-run-timing] legacy-agent sync warmed cache in %.3fs (next sync in %.0fs)",
+            elapsed,
+            _LEGACY_SYNC_TTL_SECONDS,
+        )
+
+
 async def _enforce_agent_access(body: Any, request: Request) -> None:
    agent_ids = _collect_requested_agent_ids(body)
    if not agent_ids:
@@ -383,7 +458,7 @@ async def _enforce_agent_access(body: Any, request: Request) -> None:
    from deerflow.runtime.user_context import get_effective_user_id

    store = get_agent_store(request)
-    await _sync_legacy_agent_records(store)
+    await _maybe_sync_legacy_agent_records(store)
    user_id = get_effective_user_id()
    for agent_id in agent_ids:
        try:
@@ -417,14 +492,27 @@ async def start_run(
        Target thread.
    request : Request
        FastAPI request — used to retrieve singletons from ``app.state``.
+
+    Timing observability
+    --------------------
+    Each phase records ``perf_counter`` deltas. When the total wallclock from
+    function entry to ``run_agent`` task creation exceeds
+    ``_START_RUN_SLOW_THRESHOLD_SECONDS`` we emit a ``[start-run-timing]``
+    breakdown at WARNING level. Stays at DEBUG when fast, so steady-state
+    logs are not polluted. Investigate spikes by grep'ing
+    ``[start-run-timing]`` together with ``[upstream-timing]`` (multi-agent
+    router) and ``[leader-timing] / [special-timing]`` (圆桌 leader/special).
    """
+    t_entry = time.perf_counter()
    bridge = get_stream_bridge(request)
    run_mgr = get_run_manager(request)
    run_ctx = get_run_context(request)

    disconnect = DisconnectMode.cancel if body.on_disconnect == "cancel" else DisconnectMode.continue_

+    t_pre_enforce = time.perf_counter()
    await _enforce_agent_access(body, request)
+    t_post_enforce = time.perf_counter()

    try:
        record = await run_mgr.create_or_reject(
@@ -439,33 +527,34 @@ async def start_run(
        raise HTTPException(status_code=409, detail=str(exc)) from exc
    except UnsupportedStrategyError as exc:
        raise HTTPException(status_code=501, detail=str(exc)) from exc
+    t_post_create = time.perf_counter()

    # Upsert thread metadata so the thread appears in /threads/search,
    # even for threads that were never explicitly created via POST /threads
    # (e.g. stateless runs).
-    requested_rag_mode = _rag_mode_requested(body)
-    effective_rag_mode = requested_rag_mode
-    try:
-        existing = await run_ctx.thread_store.get(thread_id)
-        effective_rag_mode = requested_rag_mode or _thread_rag_mode_enabled(existing)
-        if existing is None:
-            metadata = dict(body.metadata or {})
-            if effective_rag_mode:
-                metadata["rag_mode_enabled"] = True
-                metadata["rag_mode"] = True
-            await run_ctx.thread_store.create(
-                thread_id,
-                assistant_id=body.assistant_id,
-                metadata=metadata,
-            )
-        else:
-            await run_ctx.thread_store.update_status(thread_id, "running")
-            if requested_rag_mode and not _thread_rag_mode_enabled(existing):
-                await run_ctx.thread_store.update_metadata(
-                    thread_id,
-                    {"rag_mode_enabled": True, "rag_mode": True},
-                )
-                effective_rag_mode = True
+    requested_rag_mode = _rag_mode_requested(body)
+    effective_rag_mode = requested_rag_mode
+    try:
+        existing = await run_ctx.thread_store.get(thread_id)
+        effective_rag_mode = requested_rag_mode or _thread_rag_mode_enabled(existing)
+        if existing is None:
+            metadata = dict(body.metadata or {})
+            if effective_rag_mode:
+                metadata["rag_mode_enabled"] = True
+                metadata["rag_mode"] = True
+            await run_ctx.thread_store.create(
+                thread_id,
+                assistant_id=body.assistant_id,
+                metadata=metadata,
+            )
+        else:
+            await run_ctx.thread_store.update_status(thread_id, "running")
+            if requested_rag_mode and not _thread_rag_mode_enabled(existing):
+                await run_ctx.thread_store.update_metadata(
+                    thread_id,
+                    {"rag_mode_enabled": True, "rag_mode": True},
+                )
+                effective_rag_mode = True
            # SDK 的 ``client.threads.create()`` 不传 assistant_id —— 先建出来空的 thread，
            # 真正提交 run 时才知道用哪个 assistant。这里回填一次，避免历史列表 / 按
            # assistant 筛选时拿不到关联。
@@ -473,6 +562,7 @@ async def start_run(
                await run_ctx.thread_store.update_assistant_id(thread_id, body.assistant_id)
    except Exception:
        logger.warning("Failed to upsert thread_meta for %s (non-fatal)", sanitize_log_param(thread_id))
+    t_post_thread_upsert = time.perf_counter()

    agent_factory = resolve_agent_factory(body.assistant_id)
    graph_input = normalize_input(body.input)
@@ -485,21 +575,21 @@ async def start_run(
    # applies once. We intentionally do NOT gate on thread-row existence: the
    # thread_meta row is frequently pre-created, which previously skipped the
    # prefix for genuine first turns.
-    if _prompt_prefix_opted_in(body):
-        prefix_settings = load_system_settings().prompt_prefix
-        if prefix_settings.enabled and prefix_settings.prompt_prefix.strip():
-            graph_input = _prepend_prompt_prefix(graph_input, prefix_settings.prompt_prefix.strip())
-    if effective_rag_mode:
-        graph_input = _append_invisible_prompt(graph_input, _RAG_CITATION_PROMPT)
-    config = build_run_config(thread_id, body.config, body.metadata, assistant_id=body.assistant_id)
+    if _prompt_prefix_opted_in(body):
+        prefix_settings = load_system_settings().prompt_prefix
+        if prefix_settings.enabled and prefix_settings.prompt_prefix.strip():
+            graph_input = _prepend_prompt_prefix(graph_input, prefix_settings.prompt_prefix.strip())
+    if effective_rag_mode:
+        graph_input = _append_invisible_prompt(graph_input, _RAG_CITATION_PROMPT)
+    config = build_run_config(thread_id, body.config, body.metadata, assistant_id=body.assistant_id)

    # Merge DeerFlow-specific context overrides into both ``configurable`` and ``context``.
    # The ``context`` field is a custom extension for the langgraph-compat layer
    # that carries agent configuration (model_name, thinking_enabled, etc.).
    # Only agent-relevant keys are forwarded; unknown keys (e.g. thread_id) are ignored.
-    body_context = dict(getattr(body, "context", None) or {})
-    if effective_rag_mode:
-        body_context["rag_mode_enabled"] = True
+    body_context = dict(getattr(body, "context", None) or {})
+    if effective_rag_mode:
+        body_context["rag_mode_enabled"] = True
    # Promote whitelisted top-level body fields into the context map so
    # callers can pass them either nested under ``context`` or as flat
    # top-level fields (``excluded_tools``, ``skill_stop_names``, …).
@@ -511,6 +601,7 @@ async def start_run(

    stream_modes = normalize_stream_modes(body.stream_mode)

+    t_pre_task = time.perf_counter()
    task = asyncio.create_task(
        run_agent(
            bridge,
@@ -535,6 +626,33 @@ async def start_run(
    # title from the checkpoint and calls thread_store.update_display_name
    # after the run completes.

+    total = t_pre_task - t_entry
+    if total >= _START_RUN_SLOW_THRESHOLD_SECONDS:
+        # Slow path: emit a phase breakdown so on-call can identify the
+        # culprit without re-running with extra instrumentation. The most
+        # common cause is a stale ``_maybe_sync_legacy_agent_records`` TTL
+        # window firing on a machine with hundreds of agent directories.
+        logger.warning(
+            "[start-run-timing] thread=%s total=%.3fs enforce_access=%.3fs "
+            "run_create=%.3fs thread_upsert=%.3fs build_payload=%.3fs",
+            sanitize_log_param(thread_id),
+            total,
+            t_post_enforce - t_pre_enforce,
+            t_post_create - t_post_enforce,
+            t_post_thread_upsert - t_post_create,
+            t_pre_task - t_post_thread_upsert,
+        )
+    else:
+        logger.debug(
+            "[start-run-timing] thread=%s total=%.3fs enforce=%.3fs create=%.3fs upsert=%.3fs build=%.3fs",
+            sanitize_log_param(thread_id),
+            total,
+            t_post_enforce - t_pre_enforce,
+            t_post_create - t_post_enforce,
+            t_post_thread_upsert - t_post_create,
+            t_pre_task - t_post_thread_upsert,
+        )
+
    return record


@@ -23,6 +23,19 @@ models:
    supports_vision: true
    supports_thinking: true

+  - name: deepseek-chat
+    display_name: DeepSeek Chat
+    use: langchain_openai:ChatOpenAI
+    model: deepseek-chat
+    api_key: sk-1520028668cb4cc388ed8315ff1e856e
+    base_url: https://api.deepseek.com
+    request_timeout: 600.0
+    max_retries: 2
+    max_tokens: 4096
+    temperature: 1.0
+    supports_vision: true
+    supports_thinking: true
+
  - name: deepseek-v4-flash
    display_name: DeepSeek V4 Flash
    use: langchain_openai:ChatOpenAI
@@ -86,6 +99,11 @@ tools:
    use: deerflow.community.ddg_search.tools:web_search_tool
    max_results: 5

+  - name: web_fetch
+    group: web
+    use: deerflow.community.jina_ai.tools:web_fetch_tool
+    timeout: 15
+
  - name: ls
    group: file:read
    use: deerflow.sandbox.tools:ls_tool
@@ -121,7 +139,7 @@ sandbox:
  allow_host_bash: true

 skills:
-  path: D:\cursor-work\offline-backend-20260512\backend\skills
+  path: skills
  security_scan:
    enabled: false

@@ -129,14 +147,14 @@ agents_api:
  enabled: true

 database:
-  backend: mysql
-  mysql_url: $MYSQL_DATABASE_URL
+  backend: sqlite
+  sqlite_dir: .deer-flow/data
  pool_size: 10
  echo_sql: false

 checkpointer:
  type: sqlite
-  connection_string: D:\cursor-work\offline-backend-20260512\backend\.deer-flow\data\deerflow.db
+  connection_string: D:\05271502\deerflow-server\offline-backend-20260512\backend\.deer-flow\data\deerflow.db

 # 记忆子系统:跨会话持久化用户画像与 agent 私有记忆。
 # 三层覆盖优先级(高->低):
@@ -0,0 +1,11 @@
+{
+  "mcpServers": {},
+  "skills": {
+    "web-research": { "enabled": true },
+    "deep-search": { "enabled": true },
+    "code-retrieval": { "enabled": true },
+    "news-briefing": { "enabled": true },
+    "workspace-search": { "enabled": true },
+    "policy-knowledge": { "enabled": true }
+  }
+}
@@ -55,6 +55,9 @@ async def pause_material_confirm(state: AIWritingState) -> dict:

    action = user_input.get("action", "confirm")

+    # 单输入框文本：可能是技能名 / 来源 / 检索意图，re_search 和 enable_skill_search 都用得到
+    user_query = (user_input.get("userQuery") or user_input.get("user_query") or "").strip()
+
    if action == "confirm":
        approved_ids = set(user_input.get("approvedMaterialIds") or [m["id"] for m in pkg["materials"]])
        filtered = [m for m in pkg["materials"] if m["id"] in approved_ids]
@@ -66,12 +69,22 @@ async def pause_material_confirm(state: AIWritingState) -> dict:
            "progress_events": [{"type": "resumed", "agent_name": "系统", "message": "用户已确认素材，开始规划大纲"}],
        }

-    # re_search：重新搜索
+    # enable_skill_search：用户选择"启用技能检索"，回到 researcher 走 skill 模式
+    if action == "enable_skill_search":
+        msg = f"已启用技能检索（输入：{user_query[:30]}）" if user_query else "已启用技能检索，正在筛选可用技能…"
+        return {
+            "last_user_intervention": {**user_input, "action": "enable_skill_search", "user_query": user_query},
+            "status": "researching",
+            "progress_events": [{"type": "resumed", "agent_name": "系统", "message": msg}],
+        }
+
+    # re_search：重新搜索（支持新版 user_query 单输入框 + 旧版 extraKeywords 数组）
    extra_kws = user_input.get("extraKeywords") or []
+    msg = f"按输入「{user_query[:30]}」重新搜索…" if user_query else "重新搜索中..."
    return {
-        "last_user_intervention": {**user_input, "_extra_keywords": extra_kws},
+        "last_user_intervention": {**user_input, "_extra_keywords": extra_kws, "user_query": user_query},
        "status": "researching",
-        "progress_events": [{"type": "resumed", "agent_name": "系统", "message": "重新搜索中..."}],
+        "progress_events": [{"type": "resumed", "agent_name": "系统", "message": msg}],
    }


@@ -10,6 +10,7 @@ from typing import List
 from langchain_core.runnables import RunnableConfig

 from ..search import get_search_provider, SearchResult
+from ..search.skill_search import SkillPicker
 from ..state import AIWritingState, Material, MaterialPackage
 from ..prompts.researcher_prompts import (
    EXTRACT_KEYWORDS_SYSTEM, EXTRACT_KEYWORDS_USER,
@@ -96,15 +97,24 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
    # 检查干预模式（PAUSE-1 素材确认 / 素材不足求助）
    intervention = state.get("last_user_intervention") or {}
    extra_keywords: list[str] = intervention.get("_extra_keywords") or intervention.get("extra_keywords") or []
+    # PAUSE-1 单输入框：用户写的自然语言（技能名 / 来源 / 检索意图），优先级最高
+    user_query: str = (intervention.get("user_query") or "").strip()
    action = intervention.get("action")

    # 素材不足求助点触发的「补充检索」：保留原大纲，检索完直接回写作、不走素材确认/大纲
    is_supplement = action == "supplement_search"
-    # PAUSE-1 素材确认卡用户点「追加检索」并填了新关键词：
+    # PAUSE-1 素材确认卡用户点「按输入重新搜索」并填了 user_query：
    #   - 跳过 LLM 重新生成检索词（节省一次调用 + 尊重用户意图）
-    #   - 只检索 extra_keywords 这几个新词，不重检索之前 LLM 出过的词
+    #   - 直接用 user_query 作为唯一检索词
    #   - 保留 existing material_package，按 URL 去重合并
-    is_append_only = action == "re_search" and bool(extra_keywords)
+    is_user_query_search = action == "re_search" and bool(user_query)
+    # 兼容旧版"追加关键词"调用方（如果还有 client 传 extra_keywords 数组而非 user_query）
+    is_append_only = action == "re_search" and not user_query and bool(extra_keywords)
+    # PAUSE-1 素材确认卡用户点「启用技能检索」：
+    #   - 不再走 web/intranet 检索，改用 SkillSearchProvider
+    #   - 顺序调用候选技能 + 早停（第一个技能够就停，不够再加下一个）
+    #   - 旧 material_package 完全丢弃（覆盖式替换，让用户对比 web vs skill 结果）
+    is_skill_search = action == "enable_skill_search"

    # 用户指定的检索关键词数量（1~10），决定生成多少个检索词、检索多少个方向
    keyword_count = _resolve_keyword_count(state.get("keyword_count"))
@@ -126,10 +136,17 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
        writer({"type": "research_progress", "message": message})

    # Step 1: 提取关键词
-    kw_result: dict = {}  # 走追加分支时跳过 LLM 调用，但下方 notice 拼接需要兜底值
-    if is_append_only:
-        # 追加模式：直接用用户填的关键词，不再调 LLM
-        keywords: list[str] = _clean_keywords(extra_keywords, intent)
+    kw_result: dict = {}  # 走"用户输入 / 追加 / 技能"分支时跳过 LLM 调用，notice 拼接需要兜底值
+    if is_skill_search:
+        # 技能模式：每个 skill 会自己拼 query（user_query + skill.description），这里不用 LLM 生成关键词
+        keywords: list[str] = [user_query or intent]
+    elif is_user_query_search:
+        # 用户输入分支：把整段自然语言作为唯一检索词，尊重原意，不再做关键词拆分
+        keywords = [user_query]
+        _emit(f"按你的输入「{user_query[:30]}」检索…")
+    elif is_append_only:
+        # 兼容旧版追加模式：直接用用户填的关键词，不再调 LLM
+        keywords = _clean_keywords(extra_keywords, intent)
        _emit(f"追加检索 {len(keywords)} 个关键词…")
    else:
        _emit("正在生成检索词…")
@@ -150,78 +167,172 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
        if extra_keywords:
            keywords = list(dict.fromkeys(keywords + extra_keywords))

-    # Step 2: 并发搜索
-    provider = get_search_provider(ai_writing_config)
+    # Step 2: 搜索
+    # 技能检索：技能不是素材，是"检索策略 / 视角"。流程：
+    #   a. SkillPicker 选出按相关度排序的 skills（推 list_skills + select_skills 事件）
+    #   b. 按顺序遍历 skills：每个 skill 用其 description 让 LLM 生成针对性关键词
+    #   c. 调真实 web/intranet provider 执行检索，推 invoke_skill + skill_result 事件
+    #   d. 累积素材到 max_materials 即 break，推 stop_early
+    skill_steps: list[dict] = []
+    search_errors: list[str] = []

-    # 按用户指定的关键词数量截断。
-    # 追加模式：keywords 本身就是用户填的，全部检索不截断（用户填 N 个就搜 N 个）。
-    # 其他模式：受 keyword_count 限制，但 extra_keywords 在内时上限放宽到 keywords 总数。
-    if is_append_only:
-        search_keywords = keywords
+    def _emit_skill_step(step: dict) -> None:
+        skill_steps.append(step)
+        writer({"type": "skill_search_step", "agent_name": "素材收集专家", **step})
+
+    if is_skill_search:
+        # ── 技能编排检索 ──
+        # 技能选择走流式 LLM（推 skill_select_chunk）+ <think> 分离，前端打字机展示
+        picker = SkillPicker(
+            config=config,
+            model_name=keyword_model,
+            emit_step=_emit_skill_step,
+            max_select=5,
+            user_query=user_query,
+        )
+        selected = await picker.pick(user_intent=intent)
+        search_errors.extend(picker.errors)
+
+        # 真实 web/intranet provider（按 config.yaml 配置）
+        web_provider = get_search_provider(ai_writing_config)
+
+        # 每个 skill 尽量多取（不分摊），第一个就达到 max_materials 即可早停；
+        # 这才贴近"第一个技能够了就停，否则下一个补充"的语义。
+        per_skill = max(1, max_materials)
+        raw_materials_pre: list[Material] = []
+        seen_urls_pre: set[str] = set()
+        stopped_early = False
+
+        for entry in selected:
+            sk = entry["skill"]
+            # 用 LLM 把 user_query + skill.description 转成针对性关键词
+            _emit_skill_step({
+                "type": "invoke_skill",
+                "message": f"使用技能「{sk.name}」开始检索…",
+                "skill_name": sk.name,
+            })
+            skill_query = f"{user_query} {sk.description or ''}".strip() or intent
+            try:
+                res = await web_provider.search(skill_query, max_results=per_skill)
+            except Exception as e:
+                logger.warning("skill[%s] search failed: %s", sk.name, e)
+                search_errors.append(f"skill[{sk.name}]: {e}")
+                res = []
+
+            # 去重累积
+            new_count = 0
+            for r in res or []:
+                if r.url and r.url in seen_urls_pre:
+                    continue
+                if r.url:
+                    seen_urls_pre.add(r.url)
+                raw_materials_pre.append(Material(
+                    id=f"m{len(raw_materials_pre)+1:03d}",
+                    title=r.title,
+                    content=r.content[:500],
+                    source=f"{r.source}（via 技能 {sk.name}）",
+                    published_at=r.published_at,
+                    relevance_score=r.relevance_score,
+                    source_type=r.source_type,
+                    url=r.url,
+                    recUuid=r.recUuid,
+                ))
+                new_count += 1
+
+            _emit_skill_step({
+                "type": "skill_result",
+                "message": f"技能「{sk.name}」检索到 {new_count} 条新素材",
+                "skill_name": sk.name,
+                "matched_count": new_count,
+            })
+
+            # 早停：累积素材已达上限
+            if len(raw_materials_pre) >= max_materials:
+                stopped_early = True
+                _emit_skill_step({
+                    "type": "stop_early",
+                    "message": f"已获取 {len(raw_materials_pre)} 条素材，达到上限，停止调用后续技能",
+                    "matched_count": len(raw_materials_pre),
+                })
+                break
+
+        # 技能模式不再走下面的并发/合并逻辑，all_results_nested 留空、跳过 web 通用搜索
+        all_results_nested: list = []
+        # 把已收集到的素材挂在 raw_materials_pre，下方合并阶段直接挪过去
    else:
-        search_limit = keyword_count
-        if extra_keywords:
-            search_limit = max(search_limit, len(keywords))
-        search_keywords = keywords[:search_limit]
-    total = len(search_keywords)
-    # 每个检索词分到的检索条数 = 总量上限 / 检索词数（向上取整）
-    per_keyword = _per_keyword_limit(max_materials, total)
-    _emit(f"检索词已生成，开始检索 {total} 个方向…")
+        # ── 常规 web/intranet 模式 ──
+        provider = get_search_provider(ai_writing_config)

-    # 并发搜索，但每完成一个就推送进度（计数在单线程 asyncio 下无需加锁）
-    completed = 0
+        # 按用户指定的关键词数量截断。
+        # 用户输入 / 追加模式：keywords 本身就是用户填的，全部检索不截断。
+        if is_user_query_search or is_append_only:
+            search_keywords = keywords
+        else:
+            search_limit = keyword_count
+            if extra_keywords:
+                search_limit = max(search_limit, len(keywords))
+            search_keywords = keywords[:search_limit]
+        total = len(search_keywords)
+        per_keyword = _per_keyword_limit(max_materials, total)
+        _emit(f"检索词已生成，开始检索 {total} 个方向…")

-    async def _search_one(kw: str) -> list[SearchResult]:
-        nonlocal completed
-        try:
-            return await provider.search(kw, max_results=per_keyword)
-        finally:
-            completed += 1
-            if completed < total:
-                _emit(f"已完成 {completed}/{total} 次检索，正在继续检索相关内容…")
-            else:
-                _emit(f"已完成全部 {total} 次检索")
+        completed = 0

-    search_tasks = [_search_one(kw) for kw in search_keywords]
-    all_results_nested: list[list[SearchResult]] = await asyncio.gather(*search_tasks, return_exceptions=True)
+        async def _search_one(kw: str) -> list[SearchResult]:
+            nonlocal completed
+            try:
+                return await provider.search(kw, max_results=per_keyword)
+            finally:
+                completed += 1
+                if completed < total:
+                    _emit(f"已完成 {completed}/{total} 次检索，正在继续检索相关内容…")
+                else:
+                    _emit(f"已完成全部 {total} 次检索")
+
+        search_tasks = [_search_one(kw) for kw in search_keywords]
+        all_results_nested = await asyncio.gather(*search_tasks, return_exceptions=True)
+        # provider 内部吞掉的检索异常（如内网 ES 连接失败），一并取出用于提示用户
+        search_errors.extend(getattr(provider, "errors", []))

    # 合并去重
    # - 首次检索 / 普通 re_search（不带 extra_keywords，用户只想 LLM 重出）：旧素材丢弃
    # - 素材不足求助 supplement_search：保留旧素材 + 追加新检索结果
    # - PAUSE-1 追加检索 is_append_only：保留旧素材 + 仅检索新关键词
-    seen_urls: set[str] = set()
-    raw_materials: list[Material] = []
+    # - 技能检索 is_skill_search：覆盖式替换，旧素材完全丢弃（编排循环已经把素材累积到 raw_materials_pre）
    existing_pkg = state.get("material_package")
-    preserve_existing = existing_pkg and (is_supplement or is_append_only)
-    if preserve_existing:
-        for m in existing_pkg.get("materials", []):
-            seen_urls.add(m.get("url", ""))
-            raw_materials.append(m)
+    if is_skill_search:
+        # 技能编排循环已经完成素材累积，直接用 raw_materials_pre 当 raw_materials
+        raw_materials: list[Material] = raw_materials_pre  # type: ignore[name-defined]
+    else:
+        seen_urls: set[str] = set()
+        raw_materials = []
+        preserve_existing = existing_pkg and (is_supplement or is_append_only)
+        if preserve_existing:
+            for m in existing_pkg.get("materials", []):
+                seen_urls.add(m.get("url", ""))
+                raw_materials.append(m)

-    search_errors: list[str] = []
-    for results in all_results_nested:
-        if isinstance(results, Exception):
-            logger.warning("search failed: %s", results)
-            search_errors.append(f"{type(results).__name__}: {results}")
-            continue
-        for r in results:
-            if r.url and r.url in seen_urls:
+        for results in all_results_nested:
+            if isinstance(results, Exception):
+                logger.warning("search failed: %s", results)
+                search_errors.append(f"{type(results).__name__}: {results}")
                continue
-            seen_urls.add(r.url)
-            raw_materials.append(Material(
-                id=f"m{len(raw_materials)+1:03d}",
-                title=r.title,
-                content=r.content[:500],
-                source=r.source,
-                published_at=r.published_at,
-                relevance_score=r.relevance_score,
-                source_type=r.source_type,
-                url=r.url,
-                recUuid=r.recUuid,
-            ))
-
-    # provider 内部吞掉的检索异常（如内网 ES 连接失败），一并取出用于提示用户
-    search_errors.extend(getattr(provider, "errors", []))
+            for r in results:
+                if r.url and r.url in seen_urls:
+                    continue
+                if r.url:
+                    seen_urls.add(r.url)
+                raw_materials.append(Material(
+                    id=f"m{len(raw_materials)+1:03d}",
+                    title=r.title,
+                    content=r.content[:500],
+                    source=r.source,
+                    published_at=r.published_at,
+                    relevance_score=r.relevance_score,
+                    source_type=r.source_type,
+                    url=r.url,
+                    recUuid=r.recUuid,
+                ))

    # Step 3: 相关度排序（如素材较多，让 LLM 排序）
    _emit("检索完成，正在筛选排序素材…")
@@ -264,6 +375,14 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
        old_kws = list(existing_pkg.get("keywords") or [])
        pkg_keywords = list(dict.fromkeys(old_kws + keywords))
    pkg = MaterialPackage(keywords=pkg_keywords, materials=materials, summary=summary)
+    # 技能检索时把过程步骤一并放进 pkg，前端在素材确认卡里能展示"使用了哪些技能"
+    if is_skill_search:
+        pkg["skill_search_used"] = True
+        pkg["skill_search_steps"] = skill_steps
+    elif existing_pkg and existing_pkg.get("skill_search_used") and (is_supplement or is_append_only):
+        # 保留之前一次技能检索的痕迹（不污染纯 web 检索）
+        pkg["skill_search_used"] = existing_pkg.get("skill_search_used", False)
+        pkg["skill_search_steps"] = existing_pkg.get("skill_search_steps", [])

    # 容错告警：检索接口 / 模型调用出错时以非致命 notice 提示用户，不打断写作流程
    notices: list[dict] = []
@@ -274,19 +393,24 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
    if search_errors:
        uniq = list(dict.fromkeys(search_errors))
        # 解析失败时收集到的接口原始响应体，前端折叠展示便于定位
-        raw_dumps = list(dict.fromkeys(getattr(provider, "raw_responses", []) or []))
+        # （技能模式下 provider 局部变量没暴露，所以 raw_dumps 兜底为空列表）
+        provider_obj = locals().get("provider") or locals().get("web_provider")
+        raw_dumps = list(dict.fromkeys(getattr(provider_obj, "raw_responses", []) or []))
        notices.append(_notice(
            "ES 波动导致检索异常",
            "；".join(uniq[:3]),
            raw="\n\n----------\n\n".join(raw_dumps[:3]),
        ))

-    # 进度事件文案：追加模式下让用户看到只搜了哪几个新词
-    step_start_msg = (
-        f"追加检索关键词：{', '.join(keywords[:3])}"
-        if is_append_only
-        else f"搜索关键词：{', '.join(keywords[:3])}"
-    )
+    # 进度事件文案：用户输入 / 追加模式 / 技能模式下让用户看到差异化提示
+    if is_skill_search:
+        step_start_msg = f"启用技能检索（{len(skill_steps)} 步）"
+    elif is_user_query_search:
+        step_start_msg = f"按你的输入检索：{user_query[:40]}"
+    elif is_append_only:
+        step_start_msg = f"追加检索关键词：{', '.join(keywords[:3])}"
+    else:
+        step_start_msg = f"搜索关键词：{', '.join(keywords[:3])}"
    return {
        "material_package": pkg,
        # 补充检索：直接回写作（沿用原大纲）；常规检索：进入素材确认暂停点
@@ -3,6 +3,7 @@ from .internet_search import InternetSearchProvider
 from .intranet_stub import IntranetSearchProvider
 from .demo_search import DemoSearchProvider
 from .duckduckgo_search import DuckDuckGoSearchProvider
+from .skill_search import SkillSearchProvider


 def get_search_provider(ai_writing_config: dict) -> BaseSearchProvider:
@@ -60,5 +61,6 @@ __all__ = [
    "BaseSearchProvider", "SearchResult",
    "InternetSearchProvider", "IntranetSearchProvider",
    "DemoSearchProvider", "DuckDuckGoSearchProvider",
+    "SkillSearchProvider",
    "get_search_provider",
 ]
@@ -0,0 +1,193 @@
+"""技能选择器：根据用户意图 + 输入，从启用的 Skills 里挑出按相关度排序的若干个。
+
+注意：这个模块**只负责选择技能，不直接产出素材**。
+真正的素材产出由 researcher 节点用每个 skill 的描述去编排真实 web/intranet 检索。
+
+流程（对应素材确认卡上的"技能检索"按钮）：
+1) list_skills 事件 —— 列出当前启用的全部 skill
+2) skill_select_chunk 事件 —— LLM 流式打字 + <think></think> 分离（前端打字机展示）
+3) select_skills 事件 —— LLM 给出的排序结果（结构化字段，写到 SkillStepsCard 里）
+
+下游 researcher 节点拿到 selected_skills 后，**按顺序**：
+- 用 skill.description + user_query 生成针对性 query
+- 调真实 web search provider 执行
+- 累积素材达到阈值就停（推 stop_early）
+"""
+from __future__ import annotations
+
+import logging
+from typing import Callable, List, Optional
+
+from langchain_core.messages import HumanMessage, SystemMessage
+from langchain_core.runnables import RunnableConfig
+
+from ..nodes._utils import (
+    ThinkTagParser,
+    _content_to_text,
+    get_model,
+    parse_json,
+    safe_get_stream_writer,
+)
+
+logger = logging.getLogger(__name__)
+
+
+_SELECT_SKILLS_SYSTEM = """你是一名情报研究员，需要从可用「技能」中挑出对当前检索任务最有帮助的若干个，并按相关度从高到低排序。
+技能是一段封装好的指令 / 知识 / 检索策略，每个技能有 name 和 description。
+
+【输出要求】
+- 只输出一个 JSON 对象，不要 markdown 代码块。
+- 必须使用英文双引号。
+- selected_skills 中的 name 必须严格来自下方"可用技能"列表，不要编造，按相关度从高到低排序。
+- 如果用户输入里明确点名了某个技能（按 name 完全匹配），把它排在第一位。
+- 至少选 1 个，最多选 {max_select} 个。
+
+【完整示例】
+{{"selected_skills": ["platform-sales-data", "industry-report"], "reason": "用户想查各平台销量，平台销量技能最相关"}}"""
+
+_SELECT_SKILLS_USER = """写作意图：{user_intent}
+用户补充输入：{user_query}
+
+可用技能（共 {skill_count} 个）：
+{skill_list}
+
+请从中按相关度从高到低选出最多 {max_select} 个技能。
+只输出 JSON：
+{{"selected_skills": ["name1", "name2"], "reason": "为什么选这些 + 排序依据"}}"""
+
+
+EmitStepFn = Callable[[dict], None]
+
+
+class SkillPicker:
+    """从启用的 Skills 中流式选择按相关度排序的若干个供下游编排使用。
+
+    构造参数：
+      config:     LangGraph runnable config（用于创建 model）
+      model_name: 指定使用的模型，None 则用 configurable.model_name
+      emit_step:  推送时间线步骤事件（list_skills / select_skills）
+      max_select: LLM 单次最多挑几个技能
+      user_query: 用户在素材确认卡输入框写的自然语言
+    """
+
+    def __init__(
+        self,
+        config: RunnableConfig,
+        *,
+        model_name: Optional[str] = None,
+        emit_step: Optional[EmitStepFn] = None,
+        max_select: int = 5,
+        user_query: str = "",
+    ) -> None:
+        self._config = config
+        self._model_name = model_name
+        self._emit_step = emit_step or (lambda _step: None)
+        self._max_select = max_select
+        self._user_query = (user_query or "").strip()
+        self._errors: list[str] = []
+
+    @property
+    def errors(self) -> list[str]:
+        return list(self._errors)
+
+    def _load_enabled_skills(self) -> list:
+        try:
+            from deerflow.skills import get_or_new_skill_storage
+            storage = get_or_new_skill_storage()
+            skills = storage.load_skills(enabled_only=True)
+        except Exception as e:
+            logger.exception("ai_writing.skill_search: 加载启用技能失败")
+            self._errors.append(f"skill list error: {e}")
+            skills = []
+        self._emit_step({
+            "type": "list_skills",
+            "message": f"发现 {len(skills)} 个可用技能",
+            "skill_names": [s.name for s in skills],
+        })
+        return skills
+
+    async def pick(self, user_intent: str) -> List[dict]:
+        """流式选择并排序候选技能，返回 [{skill, reason}, ...]，按相关度降序。
+
+        过程中通过 custom 通道推送 ``skill_select_chunk`` 事件给前端打字机；
+        结束后再推一条 ``select_skills`` 结构化事件。
+        """
+        skills = self._load_enabled_skills()
+        if not skills:
+            self._emit_step({
+                "type": "select_skills",
+                "message": "当前没有启用的技能可用",
+                "skill_names": [],
+            })
+            return []
+
+        skill_list_text = "\n".join(
+            f"- {s.name}: {s.description or '(无描述)'}" for s in skills
+        )
+
+        fallback = {"selected_skills": [s.name for s in skills[: self._max_select]], "reason": ""}
+
+        # 流式 LLM 调用：与 intent_parser / writer_outline 同套机制
+        writer = safe_get_stream_writer()
+        parser = ThinkTagParser()
+        full_text = ""
+
+        def _emit_chunk(text: str, is_thinking: bool) -> None:
+            writer({
+                "type": "skill_select_chunk",
+                "text": text,
+                "is_thinking": is_thinking,
+                "agent_name": "素材收集专家",
+            })
+
+        try:
+            model = get_model(self._config, model_name=self._model_name)
+            async for chunk in model.astream([
+                SystemMessage(content=_SELECT_SKILLS_SYSTEM.format(max_select=self._max_select)),
+                HumanMessage(content=_SELECT_SKILLS_USER.format(
+                    user_intent=user_intent or "(无明确写作意图)",
+                    user_query=self._user_query or "(无)",
+                    skill_count=len(skills),
+                    skill_list=skill_list_text,
+                    max_select=self._max_select,
+                )),
+            ]):
+                delta = _content_to_text(chunk.content)
+                if not delta:
+                    continue
+                for is_thinking, text in parser.feed(delta):
+                    if not is_thinking:
+                        full_text += text
+                    _emit_chunk(text, is_thinking)
+        except Exception as e:
+            logger.exception("ai_writing.skill_search: 流式选择技能失败")
+            self._errors.append(f"skill select error: {e}")
+
+        result = parse_json(full_text, fallback) if full_text else fallback
+
+        wanted_names = result.get("selected_skills") or []
+        if not isinstance(wanted_names, list):
+            wanted_names = []
+        by_name = {s.name: s for s in skills}
+        reason = result.get("reason", "") or ""
+        chosen: list[dict] = []
+        for name in wanted_names:
+            if not isinstance(name, str):
+                continue
+            sk = by_name.get(name.strip())
+            if sk and sk.name not in {c["skill"].name for c in chosen}:
+                chosen.append({"skill": sk, "reason": reason})
+            if len(chosen) >= self._max_select:
+                break
+
+        self._emit_step({
+            "type": "select_skills",
+            "message": f"按相关度排序，候选 {len(chosen)} 个技能（将按顺序使用）",
+            "skill_names": [c["skill"].name for c in chosen],
+            "reason": reason,
+        })
+        return chosen
+
+
+# 保留旧名字让 search/__init__.py 的导出不报错；研究节点不再用它
+SkillSearchProvider = SkillPicker
@@ -3,21 +3,43 @@ import operator

 # ── 素材 ──────────────────────────────────────────────────────────────────────

-class Material(TypedDict):
+class Material(TypedDict, total=False):
    id: str
    title: str
    content: str           # 截断到 500 字
    source: str
    published_at: str
    relevance_score: float
-    source_type: str       # "internet" | "intranet"
+    source_type: str       # "internet" | "intranet" | "skill"
    url: str
    recUuid: str
+    # skill 类素材附带的元信息（skill_name / container_path / category），其他类型为空
+    extra_metadata: dict

-class MaterialPackage(TypedDict):
+class SkillStep(TypedDict, total=False):
+    """技能检索过程的单步记录，用于前端时间线展示。
+
+    type=match_skills：根据用户输入流式分析有没有匹配技能（无 skill_names）
+    type=list_skills：列出可用技能
+    type=select_skills：LLM 从中选出的相关技能
+    type=invoke_skill：实际调用某一个技能检索
+    type=stop_early：达到素材阈值，提前停止后续技能调用
+    """
+    type: str
+    message: str
+    skill_names: List[str]
+    skill_name: str
+    matched_count: int
+    reason: str
+
+class MaterialPackage(TypedDict, total=False):
    keywords: List[str]
    materials: List[Material]
    summary: str
+    # 是否本次素材来自技能检索（覆盖式替换 web 检索）
+    skill_search_used: bool
+    # 技能检索的步骤记录（前端展示"使用了 X 技能"卡片）
+    skill_search_steps: List[SkillStep]

 # ── 大纲 ──────────────────────────────────────────────────────────────────────
 # 三级结构:章节(##) → 小节(###) → 子要点(bullet -)。
@@ -72,10 +94,13 @@ class ReviewResult(TypedDict):

 # ── 用户干预 ──────────────────────────────────────────────────────────────────

-class UserInterventionPayload(TypedDict):
+class UserInterventionPayload(TypedDict, total=False):
    action: str
    approved_material_ids: List[str]
    extra_keywords: List[str]
+    # PAUSE-1 素材确认 / enable_skill_search 共用：用户在单输入框里填的自然语言
+    # （可能是要使用的技能名、想检索的来源、或者一段补充检索意图）。
+    user_query: str
    edited_outline: Optional[Outline]
    outline_feedback: str
    user_revision_notes: str
@@ -703,12 +703,14 @@ def _get_cached_skills_prompt_section(

    display_pattern = (
        """
-**Skill Result Display Pattern:**
- If a skill has `<display mode="citation">`, use the skill's returned result order as reference numbers. Cite facts from the first result as `[[1]]`, the second as `[[2]]`, and so on. Use only numbers that correspond to returned results.
- If a skill has `<display mode="list">`, summarize the returned items as a concise list and keep item order stable.
- If a skill has `<display mode="table">`, summarize the result and output `<skill-display tool="SKILL_OR_TOOL_NAME" mode="table" />` where the frontend should render the real table. Do not hand-write the table rows yourself.
- If a skill has `<display mode="card">`, summarize the result and output `<skill-display tool="SKILL_OR_TOOL_NAME" mode="card" />` where the frontend should render the cards.
- If a skill has `<display mode="custom">`, follow the custom display instruction in the `<display>` tag when presenting the skill result.
+**Skill Result Display Pattern:**
+- If RAG/reference display mode is enabled, do not number each skill result independently. Treat all retrieval results from the same user turn as one combined `referenceBatch` and cite with `[1]`, `[2]`, etc. from the system-provided reference prompt.
+- In RAG/reference display mode, do not output `<skill-display .../>`; the frontend renders cards, tables, lists, and citation details in the right reference panel.
+- If RAG/reference display mode is not enabled and a skill has `<display mode="citation">`, use the skill's returned result order as reference numbers. Cite facts from the first result as `[1]`, the second as `[2]`, and so on. Use only numbers that correspond to returned results.
+- If a skill has `<display mode="list">`, summarize the returned items as a concise list and keep item order stable.
+- If a skill has `<display mode="table">`, summarize the result and output `<skill-display tool="SKILL_OR_TOOL_NAME" mode="table" />` where the frontend should render the real table. Do not hand-write the table rows yourself.
+- If a skill has `<display mode="card">`, summarize the result and output `<skill-display tool="SKILL_OR_TOOL_NAME" mode="card" />` where the frontend should render the cards.
+- If a skill has `<display mode="custom">`, follow the custom display instruction in the `<display>` tag when presenting the skill result.
 """
        if display_signature
        else ""
@@ -993,16 +995,16 @@ Writing mode is enabled for this run.
    network_policy_section = _build_network_policy_section()
    rag_mode_section = (
        """
-<rag_citation_mode>
-RAG/reference display mode is enabled.
-When a search, knowledge, skill, or MCP tool returns a result list:
- If the skill has a display adapter, follow that adapter first.
- For `citation`, cite returned items with `[[1]]`, `[[2]]`, etc. Only cite real returned items.
- For `table`, do not hand-write the table. Summarize and place `<skill-display tool="SKILL_OR_TOOL_NAME" mode="table" />` where the table should appear. The frontend will render the real tool result as a stable table.
- For `list` or `card`, summarize and place `<skill-display tool="SKILL_OR_TOOL_NAME" mode="list" />` or `<skill-display tool="SKILL_OR_TOOL_NAME" mode="card" />` where the structured result should appear.
- If no display adapter exists, default to citation mode with numbered citations.
-</rag_citation_mode>
-"""
+<rag_citation_mode>
+RAG/reference display mode is enabled.
+When a search, knowledge, skill, or MCP tool returns a result list:
+- If multiple search terms are needed, prefer calling the retrieval tools for all known terms in the same step, then answer after the tool results are returned.
+- Treat every retrieval result from the same user turn as one combined `referenceBatch`; do not restart numbering for each search term, tool call, skill, or display adapter.
+- Use only the system-provided merged reference numbers in the final answer, formatted as `[1]`, `[2]`, `[1][3][5]`.
+- Do not output `<skill-display .../>` in RAG/reference mode. The frontend renders configured cards, tables, lists, and citation details in the right reference panel.
+- If no display adapter exists, the frontend defaults the right reference panel to numbered citations.
+</rag_citation_mode>
+"""
        if rag_mode_enabled
        else ""
    )
@@ -80,13 +80,15 @@ class RagCitationMiddleware(AgentMiddleware[AgentState]):
            return None
        summary = build_reference_debug_summary(messages, [batch] if batch else None)
        _log_rag_debug(
-            "middleware reference-prompt enabled=true prompt=true messages=%s tool_messages=%s "
-            "raw_items=%s normalized_items=%s kept_items=%s latest_batch_sources=%s",
+            "middleware reference-prompt enabled=true prompt=true single_batch=true messages=%s "
+            "tool_messages=%s raw_items=%s normalized_items=%s kept_items=%s latest_batch_id=%s "
+            "latest_batch_sources=%s",
            summary["messages"],
            summary["tool_messages"],
            summary["raw_items"],
            summary["normalized_items"],
            summary["kept_items"],
+            (batch or {}).get("id"),
            len((batch or {}).get("sources") or []),
        )
        _log_rag_debug(
@@ -37,24 +37,45 @@ def build_reference_batches(messages: list[Any]) -> list[dict[str, Any]]:
    return batches


-def build_latest_reference_prompt(messages: list[Any]) -> tuple[str, dict[str, Any] | None]:
+def build_latest_reference_batch(messages: list[Any]) -> dict[str, Any] | None:
+    """Return the single combined citation batch for the latest human turn."""
    human_index = _find_latest_human_index(messages)
    if human_index is None:
-        return "", None
+        return None
    sources = _collect_sources(messages[human_index + 1 :])
    if not sources:
-        return "", None
+        return None
    assistant = _find_reference_batch_anchor(messages, human_index + 1, len(messages))
    message_id = _message_id(assistant) or _message_id(messages[human_index]) or str(human_index)
-    batch = {
+    return {
        "id": message_id,
        "assistant_message_id": message_id,
+        "source_count": len(sources),
        "sources": sources,
    }
+
+
+def build_latest_reference_prompt(messages: list[Any]) -> tuple[str, dict[str, Any] | None]:
+    batch = build_latest_reference_batch(messages)
+    if batch is None:
+        return "", None
+    return build_reference_prompt_for_batch(batch), batch
+
+
+def build_reference_prompt_for_batch(batch: dict[str, Any]) -> str:
+    """Build the authoritative model prompt from one already-combined batch."""
+    sources = [
+        source
+        for source in batch.get("sources", [])
+        if isinstance(source, dict)
+    ]
+    source_count = len(sources)
    lines = [
-        "本轮检索结果已由系统统一合并并重新编号。最终回答必须使用下面这套编号。",
-        "如果你仍需要补充检索，请先继续调用工具；最终回答只能引用最新合并后的编号。",
+        f"本轮所有检索结果已合并为同一个 referenceBatch，共 {source_count} 条，编号从 [1] 到 [{source_count}] 唯一连续。",
+        "下面这一批就是当前回答唯一可用的参考文献批次；不要按检索词、工具调用次数或工具原始编号重新分批编号。",
+        "如果你仍需要补充检索，请只继续调用工具，不要输出正文；最终回答只能引用补充检索后系统重新合并出的最新 referenceBatch。",
        "",
+        f"<reference_batch source_count=\"{source_count}\">",
    ]
    for source in sources:
        title = source.get("title") or f"参考文献 {source.get('index')}"
@@ -73,18 +94,20 @@ def build_latest_reference_prompt(messages: list[Any]) -> tuple[str, dict[str, A
        if content:
            lines.append(content)
        lines.append("")
+    lines.append("</reference_batch>")
+    lines.append("")
    lines.extend(
        [
            "引用规则：",
            "1. 回答正文中的关键结论必须标注对应编号，例如 [1]、[2]。",
            "2. 多个来源可写为 [1][3][5]。",
            "3. 只能引用上方存在的编号，不要编造编号。",
-            "4. 不要使用工具原始编号，也不要重新排序编号。",
+            "4. 不要使用工具原始编号，也不要重新排序、重新分批或重新从 [1] 开始编号。",
            "5. 不要在正文末尾堆完整参考文献列表，前端会展示右侧参考文献。",
            "6. 参考文献模式下不要输出 <skill-display .../> 占位符，也不要在正文里额外渲染卡片、表格或列表；前端会在右侧参考文献区域展示。",
        ]
    )
-    return "\n".join(lines).strip(), batch
+    return "\n".join(lines).strip()


 def _find_latest_human_index(messages: list[Any]) -> int | None:
@@ -0,0 +1,96 @@
+"""Tests for RAG citation batch construction."""
+
+from __future__ import annotations
+
+import json
+
+from deerflow.runtime.references import (
+    build_latest_reference_batch,
+    build_latest_reference_prompt,
+    build_reference_batches,
+)
+
+
+def _tool_message(call_id: str, query: str, start: int) -> dict:
+    return {
+        "type": "tool",
+        "name": "web_search",
+        "tool_call_id": call_id,
+        "content": json.dumps(
+            {
+                "query": query,
+                "results": [
+                    {
+                        "title": f"{query} result {start}",
+                        "content": f"{query} content {start}",
+                        "url": f"https://example.test/{start}",
+                    },
+                    {
+                        "title": f"{query} result {start + 1}",
+                        "content": f"{query} content {start + 1}",
+                        "url": f"https://example.test/{start + 1}",
+                    },
+                ],
+            },
+            ensure_ascii=False,
+        ),
+    }
+
+
+def test_latest_reference_prompt_uses_one_combined_batch_for_multi_search_turn() -> None:
+    messages = [
+        {"type": "human", "id": "human-1", "content": "search two terms"},
+        {
+            "type": "ai",
+            "id": "ai-search-1",
+            "content": "",
+            "tool_calls": [{"id": "call-1", "name": "web_search", "args": {"query": "Trump"}}],
+        },
+        _tool_message("call-1", "Trump", 1),
+        {
+            "type": "ai",
+            "id": "ai-search-2",
+            "content": "",
+            "tool_calls": [{"id": "call-2", "name": "web_search", "args": {"query": "White House"}}],
+        },
+        _tool_message("call-2", "White House", 3),
+    ]
+
+    batch = build_latest_reference_batch(messages)
+    assert batch is not None
+    assert batch["source_count"] == 4
+    assert [source["index"] for source in batch["sources"]] == [1, 2, 3, 4]
+    assert [source["title"] for source in batch["sources"]] == [
+        "Trump result 1",
+        "Trump result 2",
+        "White House result 3",
+        "White House result 4",
+    ]
+
+    prompt, prompt_batch = build_latest_reference_prompt(messages)
+    assert prompt_batch == batch
+    assert '<reference_batch source_count="4">' in prompt
+    assert "[1] Trump result 1" in prompt
+    assert "[4] White House result 4" in prompt
+    assert "重新分批" in prompt
+
+
+def test_reference_batches_keep_separate_user_turns_but_merge_searches_inside_turn() -> None:
+    messages = [
+        {"type": "human", "id": "human-1", "content": "first"},
+        {"type": "ai", "id": "ai-1", "content": "", "tool_calls": [{"id": "c1", "name": "web_search"}]},
+        _tool_message("c1", "first", 1),
+        {"type": "ai", "id": "answer-1", "content": "answer [1]"},
+        {"type": "human", "id": "human-2", "content": "second"},
+        {"type": "ai", "id": "ai-2", "content": "", "tool_calls": [{"id": "c2", "name": "web_search"}]},
+        _tool_message("c2", "second-a", 3),
+        {"type": "ai", "id": "ai-3", "content": "", "tool_calls": [{"id": "c3", "name": "web_search"}]},
+        _tool_message("c3", "second-b", 5),
+        {"type": "ai", "id": "answer-2", "content": "answer [1][4]"},
+    ]
+
+    batches = build_reference_batches(messages)
+    assert [len(batch["sources"]) for batch in batches] == [2, 4]
+    assert [source["index"] for source in batches[1]["sources"]] == [1, 2, 3, 4]
+    assert batches[0]["assistant_message_id"] == "answer-1"
+    assert batches[1]["assistant_message_id"] == "answer-2"
@@ -92,7 +92,7 @@ async def test_update_builtin_only_targets_builtins() -> None:

@pytest.mark.asyncio
 async def test_get_editable_or_raise_admin_can_edit_builtin() -> None:
-    """Route helper: admin gets edit access to built-ins; non-admin still 403."""
+    """Route helper: admin gets edit access to any agent; non-admin still 403."""
    from fastapi import HTTPException

    from app.gateway.routers.agents import _get_editable_or_raise
@@ -118,14 +118,16 @@ async def test_get_editable_or_raise_admin_can_edit_builtin() -> None:
        await _get_editable_or_raise(store, "builtin-agent", "user-a", is_admin=False)
    assert exc.value.status_code == 403

-    # Admin tries to edit someone else's PUBLISHED custom agent → still 403
-    # (admin override only applies to built-ins, not to others' custom agents).
+    # Admin edits someone else's custom agent → returns ('admin', record).
+    # Behavior changed: admins manage the full agent catalog (built-in + any
+    # user's agents), not just built-ins. Non-admins are still owner-only.
    await store.create(
        {"id": "published-other", "name": "Pub", "user_id": "user-x", "published": True}
    )
-    with pytest.raises(HTTPException) as exc:
-        await _get_editable_or_raise(store, "published-other", "user-a", is_admin=True)
-    assert exc.value.status_code == 403
+    record, mode = await _get_editable_or_raise(
+        store, "published-other", "user-a", is_admin=True
+    )
+    assert mode == "admin" and record["id"] == "published-other"


 def test_build_run_config_injects_custom_assistant_as_agent_id() -> None: