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

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

@@ -53,6 +53,7 @@ from app.gateway.routers import (
     thread_shares,
     threads,
     uploads,
+    user_prompts,
     writing,
 )
 from deerflow.config import app_config as deerflow_app_config
@@ -502,6 +503,24 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
         except Exception:
             logger.exception("Skill ownership migration failed (non-fatal)")
 
+        # 一次性把 .deer-flow/agents/{id}/ 目录里历史遗留的 agent 导入 agents 表。
+        # 路由层原本每次请求都跑一遍这个 sync(扫文件系统 + N 个 agent 各一次 DB
+        # SELECT/UPDATE),把它挪到启动期 + 路由层只保留 TTL 兜底,显著降低
+        # /api/agents 列表延迟。
+        try:
+            from app.gateway.routers.agents import (
+                _mark_legacy_sync_fresh,
+                _sync_legacy_agents,
+            )
+
+            agent_store = getattr(app.state, "agent_store", None)
+            if agent_store is not None:
+                await _sync_legacy_agents(agent_store)
+                _mark_legacy_sync_fresh()
+                logger.info("Legacy folder-backed agents synced at startup")
+        except Exception:
+            logger.exception("Legacy agent sync failed at startup (non-fatal)")
+
         # 启动 skill curator 调度器:常驻后台任务,每小时检查一次需不需要进化
         # (per-user interval_hours 仍是真正决定要不要跑进化的闸)。
         try:
@@ -806,6 +825,8 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
     # Recommended Questions API is mounted at /api/recommended-questions
     app.include_router(recommended_questions.router)
 
+    app.include_router(user_prompts.router)
+
     # Suggestions API is mounted at /api/threads/{thread_id}/suggestions
     app.include_router(suggestions.router)
 

offline-backend-20260512/backend/app/gateway/deps.py

@@ -148,6 +148,10 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
 
         app.state.memory_config_store = make_user_memory_config_store(sf)
 
+        from deerflow.persistence.user_prompts import make_user_prompt_store
+
+        app.state.user_prompt_store = make_user_prompt_store(sf)
+
         # Run event store (has its own factory with config-driven backend selection)
         run_events_config = getattr(config, "run_events", None)
         app.state.run_event_store = make_run_event_store(run_events_config)

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

@@ -406,7 +406,7 @@ RECOMMENDER_AGENT_SOUL = """# 智能体推荐专员
 
 > 以下两个示例覆盖**不同任务类型**,目的是教你「思路要呼应任务本身」,**不要**把任何一个示例的具体职能链当模板。
 
-### 示例 A:合规改造任务(领域候选池)
+### 示例 A:合规改造任务(职能型候选池)
 
 **用户输入**:
 ```
@@ -415,11 +415,12 @@ RECOMMENDER_AGENT_SOUL = """# 智能体推荐专员
 关键假设:现有系统具备基础数字化能力;允许预定窗口停机
 
 候选智能体池:
-- compliance-research (合规调研):检索同类历史案例、行业最佳实践与监管口径
-- infra-capacity (基础设施评估):评估硬件容量与系统量化承载力
-- solution-architect (方案架构):产出主备方案与关键路径
-- risk-auditor (风险审查):合规红线检查与收敛建议
-- execution-planner (执行规划):里程碑与资源排布
+- roundtable-intelligence (情报收集):检索同类历史案例、行业最佳实践与外部情报
+- roundtable-environment (环境评估):评估硬件容量与系统量化承载力
+- roundtable-solution-design (方案设计):产出主备方案与关键路径
+- roundtable-risk-review (风险审查):合规红线检查与收敛建议
+- roundtable-execution-plan (执行规划):里程碑与资源排布
+- roundtable-summary (总结陈述):归纳冲突标记与盖印
 - my-cost-calc (成本测算):基于预算和资源单价计算总成本
 ```
 
@@ -430,10 +431,10 @@ RECOMMENDER_AGENT_SOUL = """# 智能体推荐专员
 ```json
 {
   "picks": [
-    {"agent_id": "compliance-research", "reason": "检索同类等保 2.0 改造案例与监管口径,直接支撑本次合规对标"},
-    {"agent_id": "solution-architect", "reason": "在 6 个月工期约束下产出主备改造方案与关键路径"},
-    {"agent_id": "risk-auditor", "reason": "对照等保 2.0 合规红线审查方案,避免后期返工"},
-    {"agent_id": "execution-planner", "reason": "把改造拆成可在 6 个月内交付的里程碑与资源排布"},
+    {"agent_id": "roundtable-intelligence", "reason": "检索同类等保 2.0 改造案例与监管口径,直接支撑本次合规对标"},
+    {"agent_id": "roundtable-solution-design", "reason": "在 6 个月工期约束下产出主备改造方案与关键路径"},
+    {"agent_id": "roundtable-risk-review", "reason": "对照等保 2.0 合规红线审查方案,避免后期返工"},
+    {"agent_id": "roundtable-execution-plan", "reason": "把改造拆成可在 6 个月内交付的里程碑与资源排布"},
     {"agent_id": "my-cost-calc", "reason": "校验整体方案落在 500 万元预算硬约束内"}
   ]
 }
@@ -452,7 +453,7 @@ RECOMMENDER_AGENT_SOUL = """# 智能体推荐专员
 - channel-planner (渠道规划师):按预算和受众设计跨平台投放组合
 - audience-insight (受众洞察):画像、关键词、情绪点研究
 - creative-copy (创意文案):产出落地页与社媒文案变体
-- ad-compliance (广告合规):审查文案与素材合规风险
+- roundtable-risk-review (风险审查):合规红线检查与收敛建议
 ```
 
 **你的输出**:
@@ -470,8 +471,8 @@ RECOMMENDER_AGENT_SOUL = """# 智能体推荐专员
 }
 ```
 
-> **关键观察**:两个示例都从候选池里按任务语义挑选,**不要**套用固定职能链(如"调研→设计→风险→执行")。
-> 候选池里的智能体能不能匹配任务,只看 description 和任务的实际语义。如果候选池里有"广告设计智能体"且任务是广告设计,**你必须挑它**,而不是机械复制示例 A 的链路。
+> **关键观察**:示例 A 选了 `roundtable-intelligence / solution-design / execution-plan` 等"职能型"智能体,示例 B **完全没用任何 `roundtable-*` 内置**,而是从用户自建的领域智能体里挑。
+> 这是因为:**候选池里的智能体能不能匹配任务,只看 description 和任务的实际语义,不看 id 是不是 `roundtable-` 开头**。如果候选池里有"广告设计智能体"且任务是广告设计,**你必须挑它**,而不是套用示例 A 的职能链。
 """
 
 

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

@@ -5,6 +5,7 @@ from __future__ import annotations
 import asyncio
 import logging
 import shutil
+import time
 from datetime import datetime
 from typing import Any
 from uuid import uuid4
@@ -58,6 +59,11 @@ class AgentResponse(BaseModel):
         description="Admin-curated homepage rank. Null = not featured. Lower sorts first.",
     )
 
+    is_favorite: bool = Field(
+        default=False,
+        description="Whether the requesting user has favorited (收藏) this agent.",
+    )
+
 
 class AgentsListResponse(BaseModel):
     """Response model for listing all visible agents."""
@@ -202,6 +208,11 @@ async def _sync_legacy_agents(store: AgentStore) -> None:
 
     This preserves old deployments where agents already existed under
     ``.deer-flow/agents/{name}`` before the agents table was introduced.
+
+    Heavy: scans the agents directory + parses each ``config.yaml`` + issues a
+    SELECT-then-INSERT/UPDATE per agent. Call once at startup via the lifespan,
+    and rely on :func:`_maybe_sync_legacy_agents` (TTL-gated) inside request
+    handlers as a safety net for environments that bypassed the lifespan hook.
     """
     try:
         for agent_cfg in list_custom_agents():
@@ -220,6 +231,44 @@ async def _sync_legacy_agents(store: AgentStore) -> None:
         logger.exception("Failed to sync legacy folder-backed agents")
 
 
+# Throttle the per-request legacy-sync fallback. The real sync runs once at
+# startup (lifespan); request handlers used to call ``_sync_legacy_agents`` on
+# every hit, which scanned the filesystem + did N DB upserts per agent each
+# time. After the first successful sync we skip the work for ``_LEGACY_SYNC_TTL_SECONDS``
+# — long enough to be effectively "startup-only" in steady state, short enough
+# that a freshly dropped agent directory gets picked up on the next bucket.
+_LEGACY_SYNC_TTL_SECONDS = 3600.0
+_legacy_sync_state: dict[str, Any] = {"last_at": 0.0, "lock": None}
+
+
+async def _maybe_sync_legacy_agents(store: AgentStore) -> None:
+    """TTL-gated wrapper around :func:`_sync_legacy_agents`.
+
+    Becomes a no-op in steady state — the heavy lifting happens in the lifespan
+    hook. Inside the TTL window we don't even take the lock; outside it we
+    serialize callers so a thundering herd doesn't fan out into N filesystem
+    scans + N*M DB queries.
+    """
+    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
+        await _sync_legacy_agents(store)
+        _legacy_sync_state["last_at"] = time.monotonic()
+
+
+def _mark_legacy_sync_fresh() -> None:
+    """Called by the lifespan hook after running :func:`_sync_legacy_agents`."""
+    _legacy_sync_state["last_at"] = time.monotonic()
+
+
 def _ensure_agent_dir_from_record(record: dict[str, Any]) -> None:
     """Lazily restore an agent's on-disk artifacts from its DB record.
 
@@ -257,14 +306,29 @@ def _record_to_response(
     record: dict[str, Any],
     *,
     include_soul: bool = False,
+    include_extras: bool = True,
 ) -> AgentResponse:
-    _ensure_agent_dir_from_record(record)
+    """Convert a DB row to an API response.
+
+    Args:
+        include_soul: Read SOUL.md from disk and embed in ``soul``. Set False
+            in list endpoints — the frontend (gallery / recommend / selector)
+            never reads ``soul``, only the edit dialog does. Skipping this
+            avoids one synchronous file read per agent.
+        include_extras: Self-heal a missing agent dir and read deprecated
+            ``model`` / ``tool_groups`` / ``skills`` from ``config.yaml``. Set
+            False in list endpoints to skip a stat + yaml parse per agent —
+            those fields are unused by current frontends and the dir restore
+            can happen lazily when the agent is actually opened.
+    """
     soul: str | None = None
+    extras: dict[str, Any] = {"skills": None, "model": None, "tool_groups": None}
+    if include_extras:
+        _ensure_agent_dir_from_record(record)
+        extras = _load_config_extras(record["id"])
     if include_soul:
         soul = load_agent_soul(record["id"]) or ""
 
-    extras = _load_config_extras(record["id"])
-
     return AgentResponse(
         id=record["id"],
         name=record["name"],
@@ -283,6 +347,25 @@ def _record_to_response(
     )
 
 
+async def _record_to_response_async(
+    record: dict[str, Any],
+    *,
+    include_soul: bool = False,
+    include_extras: bool = True,
+) -> AgentResponse:
+    """Async wrapper: when we actually touch disk, hop to a worker thread so
+    we don't block the asyncio loop. List endpoints (no soul, no extras) skip
+    this overhead and stay on the calling task."""
+    if not include_soul and not include_extras:
+        return _record_to_response(record, include_soul=False, include_extras=False)
+    return await asyncio.to_thread(
+        _record_to_response,
+        record,
+        include_soul=include_soul,
+        include_extras=include_extras,
+    )
+
+
 async def _get_owned_or_raise(store: AgentStore, agent_id: str, user_id: str) -> dict[str, Any]:
     owned = await store.get_owned(agent_id, user_id)
     if owned is not None:
@@ -344,7 +427,7 @@ async def list_agents(
     try:
         store = get_agent_store(request)
         tag_store: TagStore = get_tag_store(request)
-        await _sync_legacy_agents(store)
+        await _maybe_sync_legacy_agents(store)
         user_id = _current_user_id(request)
         if admin:
             if not _is_admin_user(request):
@@ -352,11 +435,20 @@ async def list_agents(
             agents = await store.list_all()
         else:
             agents = await store.list_visible(user_id)
-        responses = [_record_to_response(a, include_soul=True) for a in agents]
+        # 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
+        ]
 
         assignments = await tag_store.list_assignments("agent", [r.id for r in responses])
+        favorite_ids = await store.list_favorite_ids(user_id)
         for resp in responses:
             resp.tags = assignments.get(resp.id, [])
+            resp.is_favorite = resp.id in favorite_ids
         if search:
             needle = search.strip().lower()
             responses = [r for r in responses if needle in r.name.lower()]
@@ -396,11 +488,18 @@ async def get_agent_selector(request: Request) -> AgentSelectorResponse:
     _require_agents_api_enabled()
     try:
         store = get_agent_store(request)
-        await _sync_legacy_agents(store)
+        await _maybe_sync_legacy_agents(store)
 
         user_id = _current_user_id(request)
         records = await store.list_featured(user_id)
-        return AgentSelectorResponse(agents=[_record_to_response(r) for r in records])
+        # Selector dropdown only renders name/description/featured_order — same
+        # rationale as list_agents: skip soul + extras + self-heal.
+        return AgentSelectorResponse(
+            agents=[
+                _record_to_response(r, include_soul=False, include_extras=False)
+                for r in records
+            ]
+        )
     except HTTPException:
         raise
     except Exception as e:
@@ -420,12 +519,18 @@ async def get_agent(agent_id: str, request: Request) -> AgentResponse:
 
     try:
         store = get_agent_store(request)
-        await _sync_legacy_agents(store)
+        await _maybe_sync_legacy_agents(store)
         user_id = _current_user_id(request)
         record = await store.get_visible(agent_id, user_id)
         if record is None:
             raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found")
-        return _record_to_response(record, include_soul=True)
+        # Detail view legitimately reads SOUL.md + config.yaml — hop to a
+        # worker thread so the asyncio loop stays responsive under concurrent
+        # detail fetches.
+        response = await _record_to_response_async(record, include_soul=True, include_extras=True)
+        favorite_ids = await store.list_favorite_ids(user_id)
+        response.is_favorite = agent_id in favorite_ids
+        return response
     except HTTPException:
         raise
     except Exception as e:
@@ -478,7 +583,7 @@ async def create_agent_endpoint(request: AgentCreateRequest, http_request: Reque
             }
         )
         logger.info("Created agent '%s' (%s) at %s", name, agent_id, agent_dir)
-        return _record_to_response(record, include_soul=True)
+        return await _record_to_response_async(record, include_soul=True, include_extras=True)
 
     except HTTPException:
         raise
@@ -559,7 +664,7 @@ async def update_agent(agent_id: str, request: AgentUpdateRequest, http_request:
             raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found")
 
         logger.info("Updated agent '%s' (%s)", record["name"], agent_id)
-        return _record_to_response(record, include_soul=True)
+        return await _record_to_response_async(record, include_soul=True, include_extras=True)
 
     except HTTPException:
         raise
@@ -652,6 +757,10 @@ async def delete_agent(agent_id: str, request: Request) -> None:
             await get_tag_store(request).unassign_all("agent", agent_id)
         except Exception:  # noqa: BLE001 — tag cleanup must not block deletion
             pass
+        try:
+            await store.delete_favorites_for_agent(agent_id)
+        except Exception:  # noqa: BLE001 — favorite cleanup must not block deletion
+            pass
         logger.info("Deleted agent '%s' from %s", agent_id, agent_dir)
     except HTTPException:
         raise
@@ -660,6 +769,53 @@ async def delete_agent(agent_id: str, request: Request) -> None:
         raise HTTPException(status_code=500, detail=f"Failed to delete agent: {str(e)}") from e
 
 
+@router.post(
+    "/agents/{agent_id}/favorite",
+    status_code=204,
+    summary="Favorite Agent",
+    description="Bookmark a visible agent into the current user's 我的 tab. Idempotent.",
+)
+async def favorite_agent(agent_id: str, request: Request) -> None:
+    _require_agents_api_enabled()
+    agent_id = _validate_storage_id(agent_id)
+
+    try:
+        store = get_agent_store(request)
+        user_id = _current_user_id(request)
+        record = await store.get_visible(agent_id, user_id)
+        if record is None:
+            raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found")
+        await store.add_favorite(user_id, agent_id)
+        logger.info("User '%s' favorited agent '%s'", user_id, agent_id)
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error("Failed to favorite agent '%s': %s", agent_id, e, exc_info=True)
+        raise HTTPException(status_code=500, detail=f"Failed to favorite agent: {str(e)}") from e
+
+
+@router.delete(
+    "/agents/{agent_id}/favorite",
+    status_code=204,
+    summary="Unfavorite Agent",
+    description="Remove an agent from the current user's favorites. No-op when not favorited.",
+)
+async def unfavorite_agent(agent_id: str, request: Request) -> None:
+    _require_agents_api_enabled()
+    agent_id = _validate_storage_id(agent_id)
+
+    try:
+        store = get_agent_store(request)
+        user_id = _current_user_id(request)
+        await store.remove_favorite(user_id, agent_id)
+        logger.info("User '%s' unfavorited agent '%s'", user_id, agent_id)
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error("Failed to unfavorite agent '%s': %s", agent_id, e, exc_info=True)
+        raise HTTPException(status_code=500, detail=f"Failed to unfavorite agent: {str(e)}") from e
+
+
 @router.put(
     "/agents/{agent_id}/featured",
     response_model=AgentResponse,
@@ -683,7 +839,7 @@ async def set_agent_featured(
 
     try:
         store = get_agent_store(request)
-        await _sync_legacy_agents(store)
+        await _maybe_sync_legacy_agents(store)
         record = await store.get_any(agent_id)
         if record is None:
             raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found")
@@ -695,7 +851,7 @@ async def set_agent_featured(
         if updated is None:
             raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found")
         logger.info("Set agent '%s' featured_order=%s by admin", agent_id, body.order)
-        return _record_to_response(updated, include_soul=True)
+        return await _record_to_response_async(updated, include_soul=True, include_extras=True)
     except HTTPException:
         raise
     except Exception as e:

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

@@ -65,6 +65,7 @@ from fastapi.responses import StreamingResponse
 # identical in connection handling, payload shape, and error mapping.
 from app.gateway.routers._roundtable_seed import ensure_roundtable_functional_agents
 from app.gateway.routers.multi_agent import (
+    _ROUNDTABLE_THREAD_METADATA,
     _auth_headers,
     _create_thread,
     _diagnose_exception,
@@ -133,6 +134,70 @@ def _normalize_options(raw: object) -> tuple[str, ...]:
     return tuple(sorted(o for o in out if o))
 
 
+def _count_past_clarifications(messages: list[dict[str, Any]]) -> int:
+    """统计当前 thread 内已经发出过的 ``ask_clarification`` 次数。
+
+    用来给 ``stream_intent`` 做"3 轮硬上限"兜底:
+      - 0-1 次:正常流程,模型自行决定继续问还是收敛
+      - 2 次(即将进入第 3 轮):prompt 追加系统提示,要求本轮必须 INTENT_READY
+      - ≥3 次:工具白名单移除 ``ask_clarification``,模型物理上无法再问
+
+    只数"已经成型"的 tool_call,不区分用户是否回答过——LangGraph 把 ask
+    工具的调用永久记录在 AIMessage.tool_calls 里,数它即可。
+    """
+    if not isinstance(messages, list):
+        return 0
+    count = 0
+    for msg in messages:
+        if not isinstance(msg, dict):
+            continue
+        if (msg.get("type") or "").lower() not in ("ai", "aimessage", "aimessagechunk"):
+            continue
+        for tc in msg.get("tool_calls") or []:
+            if (tc.get("name") or "").lower() == "ask_clarification":
+                count += 1
+    return count
+
+
+async def _fetch_thread_messages(
+    client: httpx.AsyncClient,
+    base: str,
+    headers: dict[str, str],
+    thread_id: str,
+) -> list[dict[str, Any]]:
+    """读取当前 thread checkpoint 的 messages 列表(best-effort)。
+
+    用 ASGITransport loopback 调 ``GET /api/threads/{tid}/state``,失败时返回
+    空列表——计数兜底是个"刹车",失败时退化成原有的纯 SOUL 自律行为,不阻断主流程。
+    """
+    try:
+        resp = await client.get(f"{base}/api/threads/{thread_id}/state", headers=headers)
+    except Exception as exc:
+        logger.warning("intent: get thread state failed (%s) — falling back to soul-only flow", exc)
+        return []
+    if resp.status_code != 200:
+        logger.warning(
+            "intent: get thread state %s returned %s — falling back to soul-only flow",
+            thread_id,
+            resp.status_code,
+        )
+        return []
+    try:
+        payload = resp.json()
+    except Exception:
+        return []
+    values = payload.get("values") if isinstance(payload, dict) else None
+    if not isinstance(values, dict):
+        return []
+    messages = values.get("messages")
+    return messages if isinstance(messages, list) else []
+
+
+# 硬上限:超过这个轮数,无论模型怎么想,都强制移除 ask_clarification 工具。
+# 与 SOUL.md §0 「硬规则 2」对齐——"绝不存在第 4 轮"。
+_MAX_CLARIFICATION_ROUNDS = 3
+
+
 def _is_repeated_clarification(
     messages: list[dict[str, Any]],
     *,
@@ -222,7 +287,7 @@ async def init_intent(request: Request, payload: dict = Body(...)) -> dict[str,
     # 看不到逐 token 增量。
     try:
         async with _make_loopback_client(request) as client:
-            thread_id = await _create_thread(client, base, headers)
+            thread_id = await _create_thread(client, base, headers, metadata=_ROUNDTABLE_THREAD_METADATA)
     except HTTPException:
         # 上游已用结构化方式失败,原样传给前端不需要二次包装。
         raise
@@ -279,33 +344,70 @@ async def stream_intent(request: Request, payload: dict = Body(...)) -> Streamin
         # 因此既无须 trust_env=False 防代理，也不存在 loopback 40s 超时这种坑。
         async with _make_loopback_client(request) as client:
             try:
+                # —— 3 轮硬上限兜底 ——
+                # 在发起 run 之前先读一次 thread state,数当前 ask_clarification 已发出次数:
+                #   past=0/1 → 正常,不动 message,不动工具白名单
+                #   past=2(本轮即将变成第 3 轮)→ 给消息加前缀,强提示"必须 INTENT_READY"
+                #   past≥3 → 直接把 ask_clarification 也加进 excluded_tools,工具物理消失
+                # 这一步是个"刹车":SOUL 让模型自律,这里让系统兜底。
+                past_clarifications = _count_past_clarifications(
+                    await _fetch_thread_messages(client, base, headers, thread_id)
+                )
+
+                # 1. ask_clarification 的工具白名单——超过硬上限就移除。
+                base_excluded_tools = [
+                    "web_search",
+                    "present_files",
+                    "view_image",
+                    "agent_orchestration",
+                    "bash",
+                    "ls",
+                    "read_file",
+                    "write_file",
+                    "str_replace",
+                    "memory",
+                    "hindsight_recall",
+                    "hindsight_reflect",
+                    "hindsight_retain",
+                    "task",
+                    "write_todos",
+                ]
+                if past_clarifications >= _MAX_CLARIFICATION_ROUNDS:
+                    base_excluded_tools.append("ask_clarification")
+                    logger.info(
+                        "intent stream: past_clarifications=%d >= %d, force-disabling "
+                        "ask_clarification to break the loop",
+                        past_clarifications,
+                        _MAX_CLARIFICATION_ROUNDS,
+                    )
+
+                # 2. message 前缀——给模型一个明确的"本轮必须收敛"信号。
+                effective_message = message
+                if past_clarifications >= _MAX_CLARIFICATION_ROUNDS:
+                    effective_message = (
+                        f"[系统强制] 已澄清 {past_clarifications} 轮,触发硬上限。"
+                        f"本轮请直接输出 [INTENT_READY],缺的字段用 assumptions 兜底,"
+                        f"绝对不要再调任何工具,也不要再询问任何问题。\n\n"
+                        f"用户最新输入:{message}"
+                    )
+                elif past_clarifications >= _MAX_CLARIFICATION_ROUNDS - 1:
+                    effective_message = (
+                        f"[系统提示] 已澄清 {past_clarifications} 轮,这是最后一轮。"
+                        f"本轮请直接输出 [INTENT_READY],未明确的字段用 assumptions 兜底。\n\n"
+                        f"用户最新输入:{message}"
+                    )
+
                 body = _run_payload(
                     agent_name=INTENT_AGENT_ID,
                     model_name=model_name,
                     thread_id=thread_id,
-                    new_message=message,
-                    # Intent agent only ever needs ask_clarification. Block
+                    new_message=effective_message,
+                    # Intent agent only ever needs ask_clarification (and even that
+                    # is yanked once past_clarifications hits the cap above). Block
                     # everything else so the model can't drift into solutioning
-                    # (e.g. saving a design doc with write_file) — the model
-                    # tends to ignore SOUL prose, hard exclusion is reliable.
-                    # 仅保留 ask_clarification；其余工具见模块 docstring「工具白名单」。
-                    excluded_tools=[
-                        "web_search",
-                        "present_files",
-                        "view_image",
-                        "agent_orchestration",
-                        "bash",
-                        "ls",
-                        "read_file",
-                        "write_file",
-                        "str_replace",
-                        "memory",
-                        "hindsight_recall",
-                        "hindsight_reflect",
-                        "hindsight_retain",
-                        "task",
-                        "write_todos",
-                    ],
+                    # (e.g. saving a design doc with write_file) — SOUL prose is
+                    # not reliable on its own, hard exclusion is.
+                    excluded_tools=base_excluded_tools,
                     skill_stop_names=[],
                 )
 
@@ -338,7 +440,13 @@ async def stream_intent(request: Request, payload: dict = Body(...)) -> Streamin
                         except json.JSONDecodeError as exc:
                             logger.warning("intent summary JSON parse failed: %s", exc)
                             summary = None
-                        ready_text = msg_text
+                        # SOUL §0 硬规则 3 要求 ```json``` 反引号闭合后不得再输出任何字符,
+                        # 但模型仍偶尔会在后面接 "那么我帮你列一下大纲……" 之类的尾巴。
+                        # 终态帧只回到 marker 块结尾 + 一个换行,把废话从 content 里裁掉——
+                        # 草稿持久化 / 跨设备重放 / 前端 splitIntentReadyBlock 的 after
+                        # 段都因此干净。流式期间用户仍会逐 token 看到那段废话,前端
+                        # Step1Panel 已经把 `parts.after` 隐藏掉,体验闭环。
+                        ready_text = msg_text[: m.end()].rstrip() + "\n"
                         break
                 if summary and ready_text:
                     final_frame = {

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

@@ -53,18 +53,28 @@ import os
 import re
 import time
 import traceback
+import uuid
 from typing import Any, AsyncIterator
 
 import httpx
 from deerflow.config import get_app_config
 from deerflow.tools.builtins.clarification_utils import resolve_allow_multiple
+from deerflow.runtime.user_context import get_effective_user_id
+from deerflow.utils.time import now_iso
 from fastapi import APIRouter, Body, HTTPException, Request
 from fastapi.responses import StreamingResponse
+from langgraph.checkpoint.base import empty_checkpoint
 
 logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api/multi-agent", tags=["multi-agent"])
 
 
+# 圆桌（intent / recommend / 席位 / 协调）创建的 thread 都是功能型内部线程,
+# 不是用户的普通对话。打上这个 metadata,前端「最近的对话」列表(recent-chat-list.tsx)
+# 据 `metadata.thread_type` 过滤掉,与 scheduler / bootstrap 系统会话同一套机制。
+_ROUNDTABLE_THREAD_METADATA: dict[str, Any] = {"thread_type": "roundtable", "system": True}
+
+
 # ---------------------------------------------------------------------------
 # 共享:异常诊断器(intent/recommend/multi-agent 共用)
 # ---------------------------------------------------------------------------
@@ -285,9 +295,22 @@ def _auth_headers(request: Request) -> dict[str, str]:
     return {"Authorization": token, "Content-Type": "application/json"}
 
 
-async def _create_thread(client: httpx.AsyncClient, base: str, headers: dict[str, str]) -> str:
-    """``POST /api/threads`` → 新 thread_id（intent/recommend/init 席位均用）。"""
-    resp = await client.post(f"{base}/api/threads", json={}, headers=headers)
+async def _create_thread(
+    client: httpx.AsyncClient,
+    base: str,
+    headers: dict[str, str],
+    *,
+    metadata: dict[str, Any] | None = None,
+) -> str:
+    """``POST /api/threads`` → 新 thread_id（intent/recommend/init 席位均用）。
+
+    ``metadata`` 写入 thread 元数据,圆桌调用方传 ``_ROUNDTABLE_THREAD_METADATA``
+    把这些功能型线程标记成可被前端「最近的对话」过滤掉的系统会话。
+    """
+    body: dict[str, Any] = {}
+    if metadata:
+        body["metadata"] = metadata
+    resp = await client.post(f"{base}/api/threads", json=body, headers=headers)
     resp.raise_for_status()
     return resp.json()["thread_id"]
 
@@ -323,6 +346,181 @@ async def _create_agent(
     return False, resp.text
 
 
+# ---------------------------------------------------------------------------
+# 同进程直调 helpers(仅供 init_multi_agent 用,不要扩散到其它地方)
+#
+# init 阶段串了 N+1 个 thread create + N 个 GET /api/agents + 1 个 POST /api/agents,
+# 用 HTTP loopback 串走每次都得 TCP + 鉴权中间件 + Pydantic 反序列化一遍。每个
+# 调用 10-30ms,加起来对前端就是「点进 Step 2 后席位灯一颗一颗亮」的卡顿来源。
+#
+# 这里把这三个调用改成直接吃 app.state 里的 store/checkpointer —— 跑在同一个事
+# 件循环里,省掉 TCP + 序列化 + 中间件链。鉴权 / 校验 / 文件副作用按需在直调
+# helper 里显式做掉,不要"绕过"任何一项。
+#
+# 设计边界:这里只覆盖 init 阶段。SSE 长连接(/runs/stream / state)留在 HTTP
+# loopback,因为它们本就是别的进程模型(LangGraph 子树),抽出来风险大收益小。
+# ---------------------------------------------------------------------------
+
+
+def _resolve_user_id(request: Request) -> str:
+    """与 routers/agents._current_user_id 完全等价,直调路径里用。"""
+    user = getattr(request.state, "user", None)
+    if user is not None:
+        return str(user.id)
+    return get_effective_user_id()
+
+
+async def _create_thread_direct(request: Request, *, metadata: dict[str, Any] | None = None) -> str:
+    """同进程版的 ``POST /api/threads``。
+
+    完全镜像 ``routers/threads.create_thread`` 的写盘动作(thread_meta +
+    空 checkpoint),但跳过 TCP + Pydantic 反序列化。仅供 init 阶段批量建席位
+    线程用,失败时抛 HTTPException 由 SSE 帧上报。
+
+    ``metadata`` 写入 thread 元数据;init 传 ``_ROUNDTABLE_THREAD_METADATA`` 把
+    席位 / 协调线程标记成系统会话,从前端「最近的对话」里隐藏。
+    """
+    from app.gateway.deps import get_checkpointer, get_thread_store
+
+    checkpointer = get_checkpointer(request)
+    thread_store = get_thread_store(request)
+    thread_id = str(uuid.uuid4())
+    now = now_iso()
+
+    try:
+        await thread_store.create(thread_id, assistant_id=None, metadata=metadata or {})
+    except Exception:
+        logger.exception("[init-direct] thread_meta create failed for %s", thread_id)
+        raise HTTPException(status_code=500, detail="Failed to create thread")
+
+    config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
+    ckpt_metadata = {
+        "step": -1,
+        "source": "input",
+        "writes": None,
+        "parents": {},
+        "created_at": now,
+    }
+    try:
+        await checkpointer.aput(config, empty_checkpoint(), ckpt_metadata, {})
+    except Exception:
+        logger.exception("[init-direct] checkpoint create failed for %s", thread_id)
+        raise HTTPException(status_code=500, detail="Failed to create thread")
+    return thread_id
+
+
+async def _get_agents_batch_direct(
+    request: Request,
+    agent_ids: list[str],
+) -> dict[str, dict[str, Any]]:
+    """同进程版的 N 个并行 ``GET /api/agents/{id}`` 合并为 1 次 DB 查询。
+
+    返回 ``{agent_id: {id, name, description}}``。**严格模式**:任一席位对当前
+    用户不可见就抛 404 —— 与原 HTTP loopback 的 ``_get_agent`` 行为一致(原路径
+    单查 404 会 raise,在外层被 except 捕获并 yield error 帧),不要悄悄回退。
+    """
+    from app.gateway.deps import get_agent_store
+
+    store = get_agent_store(request)
+    user_id = _resolve_user_id(request)
+    records = await store.list_by_ids(agent_ids, user_id)
+    out: dict[str, dict[str, Any]] = {}
+    for r in records:
+        rid = str(r.get("id") or "").lower()
+        if not rid:
+            continue
+        out[rid] = {
+            "id": rid,
+            "name": str(r.get("name") or rid),
+            "description": str(r.get("description") or ""),
+        }
+    missing = [sid for sid in agent_ids if sid.lower() not in out]
+    if missing:
+        raise HTTPException(
+            status_code=404,
+            detail=f"Agent(s) not visible to current user: {', '.join(missing)}",
+        )
+    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)
+
+
 async def _append_thread_message(
     client: httpx.AsyncClient,
     base: str,
@@ -456,6 +654,8 @@ def _run_payload(
     new_message: str,
     excluded_tools: list[str],
     skill_stop_names: list[str],
+    thinking_enabled: bool = True,
+    reasoning_effort: str = "medium",
 ) -> dict[str, Any]:
     """构造 ``POST /api/threads/{thread_id}/runs/stream`` 的请求体。
 
@@ -463,6 +663,10 @@ def _run_payload(
     - ``excluded_tools``：按路由裁剪工具面（intent 仅澄清、special 禁派活等）。
     - ``skill_stop_names``：leader 侧配合 SkillStopMiddleware 在 ``agent_orchestration``
       调用后截断图，以便网关从 AIMessage.tool_calls 解析派活列表。
+    - ``thinking_enabled`` / ``reasoning_effort``：默认开 CoT + medium，适合 Step 1 澄清
+      与子智能体交付这类需要推理的场景。**推荐弹窗与总控派活**是短决策任务，调用方
+      应传 ``thinking_enabled=False`` + ``reasoning_effort="low"``，省掉模型先吐
+      reasoning tokens 才开口的时间，显著降低前端"调度中"首字节延迟。
     """
     return {
         "input": {
@@ -479,11 +683,17 @@ def _run_payload(
             "agent_name": agent_name,
             "model_name": model_name,
             "mode": "pro",
-            "reasoning_effort": "medium",
-            "thinking_enabled": True,
+            "reasoning_effort": reasoning_effort,
+            "thinking_enabled": thinking_enabled,
             "is_plan_mode": False,
             "subagent_enabled": False,
             "thread_id": thread_id,
+            # 圆桌四个 router(intent / recommend / leader / special)都是临时任务,
+            # 用户个人记忆与本次研讨无关。MemoryMiddleware 的 Hindsight recall 是
+            # 同步阻塞 HTTP 调用(单次 timeout 15s),如果记忆服务慢/不通会让前端
+            # "调度中"卡很久;这里默认跳过,builtin USER.md/MEMORY.md 仍通过系统
+            # 提示注入不受影响。
+            "memory_recall_disabled": True,
         },
         "stream_mode": ["messages-tuple", "values"],
         "stream_subgraphs": True,
@@ -534,111 +744,132 @@ async def init_multi_agent(request: Request, payload: dict = Body(...)) -> Strea
         _validate_agent_id(name, "agent_names entry")
     _validate_agent_id(main_agent_name, "main_agent_name")
 
-    base = _loopback_base(request)
-    headers = _auth_headers(request)
-
     def _frame(payload: dict[str, Any]) -> str:
         return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
 
     async def stream_generator() -> AsyncIterator[str]:
-        # TCP loopback 到 127.0.0.1:<gateway port>。client 配置见 _make_loopback_client：
-        # trust_env=False 防 HTTP_PROXY 拦截,timeout=None 容纳 SSE 长连接。
-        async with _make_loopback_client(request) as client:
-            try:
-                # Per-seat prepare task — same body as before, but we no longer
-                # await them with asyncio.gather. Instead, asyncio.as_completed
-                # surfaces them in REAL completion order so we can emit a
-                # seat_ready frame the moment each finishes.
-                async def _prepare_sub_agent(sub_id: str) -> tuple[str, str, str, str]:
-                    info = await _get_agent(client, base, headers, sub_id)
-                    display_name = str(info.get("name") or sub_id)
-                    description = info.get("description", "") or ""
-                    thread_id = await _create_thread(client, base, headers)
-                    return sub_id.lower(), thread_id, display_name, description
+        # init 阶段 fast path:全部走同进程直调(_create_thread_direct /
+        # _get_agents_batch_direct / _create_coordinator_direct),不再走 HTTP
+        # loopback —— TCP + 鉴权中间件 + Pydantic 反序列化每次 10-30ms,N+1 个席位
+        # 累计就是"点进 Step 2 后席位一颗颗亮"的卡顿来源。
+        try:
+            t_init_start = time.perf_counter()
+            # 1) 一次 DB 查询拿到所有席位的 name/description,替代之前的 N 个并行
+            #    GET /api/agents/{id}。SOUL 拼接需要描述,这里一次性拿齐。
+            agent_meta = await _get_agents_batch_direct(request, agent_names)
+            t_meta_done = time.perf_counter()
 
-                tasks = [asyncio.create_task(_prepare_sub_agent(sid)) for sid in agent_names]
-                prepared: list[tuple[str, str, str, str]] = []
-                try:
-                    for fut in asyncio.as_completed(tasks):
-                        sid, thread_id, display_name, description = await fut
-                        prepared.append((sid, thread_id, display_name, description))
-                        yield _frame({
-                            "event": "seat_ready",
-                            "agent_name": sid,
-                            "thread_id": thread_id,
-                            "display_name": display_name,
-                            "description": description,
-                        })
-                except Exception:
-                    # Cancel any still-pending sibling so we don't leak tasks
-                    # past the AsyncClient close.
-                    for t in tasks:
-                        if not t.done():
-                            t.cancel()
-                    raise
-
-                agent_threads: dict[str, str] = {sid: thread_id for sid, thread_id, _, _ in prepared}
-                descriptions: list[str] = [
-                    f"- agent_name: {sid}（{display_name}）：{description}"
-                    for sid, _, display_name, description in prepared
-                ]
-
-                # 动态 SOUL：把本次 init 的席位列表写进总控 prompt，供 agent_orchestration 派活。
-                soul = (
-                    "#核心定位：\n"
-                    "用途：根据用户的任务目标,调用其他智能体干活,只负责协调汇总,"
-                    "当你觉得已有信息支持回答用户问题时候,停止调度,并输出最后的结果\n"
-                    "#调度方法：\n"
-                    "通过使用agent_orchestration技能调用其他智能体干活,使用技能时传入agent_name和需要做的活。"
-                    "注意：agent_name 必须严格使用下方列出的英文 id（例如 roundtable-intelligence,"
-                    "或形如 02714cbe6e64406a8a4659f0e3d9c0b8 这样的 UUID）,"
-                    "不要把括号里的中文显示名拼进 agent_name 里。\n"
-                    "#关键约束(必读)：\n"
-                    "1. **不要对同一子智能体重复派同样的活**：每次派活前回顾对话历史,"
-                    "如果某个子智能体已经针对你正要派的任务完成过交付("
-                    "看到形如\"子智能体[名称]完成XX工作,交付内容为:...\"的 human 消息),"
-                    "就直接基于已有交付推进,**不要重复派活**,除非你确实有**具体的、不一样的**补充问题。\n"
-                    "2. **看到交付要会用**：子智能体的交付内容会作为 human 消息追加到你的对话历史里,"
-                    "你能完整看见。当各方信息已足够形成结论时,**直接综合输出最终方案**(此时不再调用任何工具),"
-                    "让讨论收敛。\n"
-                    "3. **避免无意义循环**：如果连续两轮都没有新信息流入,立刻停止派活并基于现有素材给出最终结论。\n"
-                    "#可调度智能体：\n"
-                    + "\n".join(descriptions)
+            # 2) 每个席位 thread 并行创建(直调 thread_store + checkpointer),
+            #    用 asyncio.as_completed 保持"先完先发 seat_ready"的体验。
+            async def _prepare_sub_agent(sub_id: str) -> tuple[str, str, str, str]:
+                thread_id = await _create_thread_direct(request, metadata=_ROUNDTABLE_THREAD_METADATA)
+                meta = agent_meta.get(sub_id.lower(), {})
+                return (
+                    sub_id.lower(),
+                    thread_id,
+                    str(meta.get("name") or sub_id),
+                    str(meta.get("description") or ""),
                 )
 
-                coord_thread = await _create_thread(client, base, headers)
-                agent_threads[main_agent_name] = coord_thread
-                ok, result = await _create_agent(
-                    client,
-                    base,
-                    headers,
+            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 partially-prepared seats
+                # don't keep running past the stream close.
+                for t in tasks:
+                    if not t.done():
+                        t.cancel()
+                raise
+            t_seats_done = time.perf_counter()
+
+            agent_threads: dict[str, str] = {sid: thread_id for sid, thread_id, _, _ in prepared}
+            descriptions: list[str] = [
+                f"- agent_name: {sid}（{display_name}）：{description}"
+                for sid, _, display_name, description in prepared
+            ]
+
+            # 动态 SOUL：把本次 init 的席位列表写进总控 prompt，供 agent_orchestration 派活。
+            soul = (
+                "#核心定位：\n"
+                "用途：根据用户的任务目标,调用其他智能体干活,只负责协调汇总,"
+                "当你觉得已有信息支持回答用户问题时候,停止调度,并输出最后的结果\n"
+                "#调度方法：\n"
+                "通过使用agent_orchestration技能调用其他智能体干活,使用技能时传入agent_name和需要做的活。"
+                "注意：agent_name 必须严格使用下方列出的英文 id（例如 roundtable-intelligence,"
+                "或形如 02714cbe6e64406a8a4659f0e3d9c0b8 这样的 UUID）,"
+                "不要把括号里的中文显示名拼进 agent_name 里。\n"
+                "#关键约束(必读)：\n"
+                "1. **不要对同一子智能体重复派同样的活**：每次派活前回顾对话历史,"
+                "如果某个子智能体已经针对你正要派的任务完成过交付("
+                "看到形如\"子智能体[名称]完成XX工作,交付内容为:...\"的 human 消息),"
+                "就直接基于已有交付推进,**不要重复派活**,除非你确实有**具体的、不一样的**补充问题。\n"
+                "2. **看到交付要会用**：子智能体的交付内容会作为 human 消息追加到你的对话历史里,"
+                "你能完整看见。当各方信息已足够形成结论时,**直接综合输出最终方案**(此时不再调用任何工具),"
+                "让讨论收敛。\n"
+                "3. **避免无意义循环**：如果连续两轮都没有新信息流入,立刻停止派活并基于现有素材给出最终结论。\n"
+                "#可调度智能体：\n"
+                + "\n".join(descriptions)
+            )
+
+            # 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"],
-                    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
+            )
+            coord_thread = await coord_thread_task
+            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
+            t_coord_done = time.perf_counter()
+            logger.info(
+                "[init-direct] meta=%.3fs seats=%.3fs coord=%.3fs total=%.3fs seats_count=%d",
+                t_meta_done - t_init_start,
+                t_seats_done - t_meta_done,
+                t_coord_done - t_seats_done,
+                t_coord_done - t_init_start,
+                len(prepared),
+            )
 
-                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}"})
+            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(),
@@ -825,6 +1056,11 @@ async def _leader_run(
         new_message=new_message,
         excluded_tools=["web_search"],
         skill_stop_names=effective_stop_names,
+        # 总控派活是"下一步派谁、什么任务"的短决策，不需要 CoT。
+        # 关掉 thinking + low reasoning 让"调度中"几乎瞬间转入"正在派活"，
+        # 子智能体仍保留 thinking（在 _special_run 里默认 True），保证交付质量。
+        thinking_enabled=False,
+        reasoning_effort="low",
     )
     t_payload_built = time.perf_counter()
     logger.info(

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

@@ -49,6 +49,7 @@ from fastapi.responses import StreamingResponse
 
 from app.gateway.routers._roundtable_seed import ensure_roundtable_functional_agents
 from app.gateway.routers.multi_agent import (
+    _ROUNDTABLE_THREAD_METADATA,
     _auth_headers,
     _create_thread,
     _flatten_content,
@@ -237,7 +238,7 @@ async def stream_recommend(request: Request, payload: dict = Body(...)) -> Strea
         async with _make_loopback_client(request) as client:
             try:
                 # 每次推荐独立 thread；prompt 已包含完整 intent + 候选列表，无需多轮。
-                thread_id = await _create_thread(client, base, headers)
+                thread_id = await _create_thread(client, base, headers, metadata=_ROUNDTABLE_THREAD_METADATA)
 
                 body = _run_payload(
                     agent_name=RECOMMENDER_AGENT_ID,
@@ -253,6 +254,12 @@ async def stream_recommend(request: Request, payload: dict = Body(...)) -> Strea
                         "ask_clarification",
                     ],
                     skill_stop_names=[],
+                    # 推荐是"从候选池挑 2-6 个"的短分类任务，不需要 CoT。
+                    # 关掉 thinking + 用 low reasoning，避免模型先吐数千 reasoning
+                    # tokens 才开始输出 [RECOMMEND_READY]，前端"等待推荐结果..."
+                    # 从 10-30s 降到 2-5s。
+                    thinking_enabled=False,
+                    reasoning_effort="low",
                 )
 
                 captured: list[str] = []

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

@@ -37,6 +37,10 @@ class ScheduledTaskCreateRequest(BaseModel):
     timezone: str | None = None
     execution_agent_name: str | None = None
     execution_context: dict[str, Any] | None = None
+    require_markdown: bool = Field(
+        default=False,
+        description="Require the result to be Markdown and exceed the minimum length, else mark the run failed.",
+    )
 
 
 class ScheduledTaskUpdateRequest(BaseModel):
@@ -48,6 +52,7 @@ class ScheduledTaskUpdateRequest(BaseModel):
     enabled: bool | None = None
     execution_agent_name: str | None = None
     execution_context: dict[str, Any] | None = None
+    require_markdown: bool | None = None
 
 
 class ScheduledTaskDeliveryProfileUpdateRequest(BaseModel):
@@ -555,6 +560,7 @@ async def create_scheduled_task(body: ScheduledTaskCreateRequest, request: Reque
             timezone=body.timezone,
             execution_agent_name=body.execution_agent_name,
             execution_context=body.execution_context,
+            require_markdown=body.require_markdown,
         )
     except DuplicateTaskNameError:
         raise HTTPException(status_code=409, detail="已存在同名定时任务,请换一个名称")

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

@@ -19,17 +19,23 @@ from pathlib import Path
 import uuid
 from typing import Any
 
-from fastapi import APIRouter, HTTPException, Request
-from langgraph.checkpoint.base import empty_checkpoint
-from pydantic import BaseModel, Field, field_validator
-
-from app.gateway.authz import require_permission
-from app.gateway.deps import get_checkpointer
-from app.gateway.utils import sanitize_log_param
+from fastapi import APIRouter, HTTPException, Request
+from langgraph.checkpoint.base import empty_checkpoint
+from pydantic import BaseModel, Field, field_validator
+from sqlalchemy import select
+
+from app.gateway.authz import require_permission
+from app.gateway.deps import get_checkpointer
+from app.gateway.utils import sanitize_log_param
+from deerflow.config.extensions_config import SkillDisplayConfig
 from deerflow.config.paths import Paths, get_paths
+from deerflow.persistence.engine import get_engine, get_session_factory
+from deerflow.persistence.skills.model import SkillDisplayAdapterRow
 from deerflow.runtime import serialize_channel_values
 from deerflow.runtime.references import (
+    adapt_reference_batches_with_displays,
     build_reference_batches as build_runtime_reference_batches,
+    build_reference_debug_payload,
     build_reference_debug_summary,
 )
 from deerflow.runtime.user_context import get_effective_user_id
@@ -81,11 +87,54 @@ def _rag_debug_file_path() -> str:
 def _log_rag_debug(message: str, *args: Any) -> None:
     logger.info(message, *args)
     _get_rag_debug_logger().info(message, *args)
-
-
-# ---------------------------------------------------------------------------
-# Response / request models
-# ---------------------------------------------------------------------------
+
+
+def _json_text_loads(value: str | None) -> dict[str, Any] | None:
+    if not value:
+        return None
+    try:
+        parsed = json.loads(value)
+        return parsed if isinstance(parsed, dict) else None
+    except Exception:
+        logger.debug("Failed to parse persisted skill display JSON", exc_info=True)
+        return None
+
+
+async def _load_persisted_skill_displays() -> dict[str, SkillDisplayConfig]:
+    engine = get_engine()
+    sf = get_session_factory()
+    if engine is None or sf is None:
+        return {}
+    try:
+        async with engine.begin() as conn:
+            await conn.run_sync(SkillDisplayAdapterRow.__table__.create, checkfirst=True)
+        async with sf() as session:
+            result = await session.execute(select(SkillDisplayAdapterRow))
+            displays: dict[str, SkillDisplayConfig] = {}
+            for row in result.scalars():
+                payload = _json_text_loads(row.display_config)
+                if payload is None:
+                    continue
+                try:
+                    display = SkillDisplayConfig.model_validate(payload)
+                except Exception:
+                    logger.debug(
+                        "Failed to validate persisted skill display for %s",
+                        row.skill_name,
+                        exc_info=True,
+                    )
+                    continue
+                if display.enabled and display.mode != "none":
+                    displays[row.skill_name] = display
+            return displays
+    except Exception:
+        logger.debug("Failed to load persisted skill display adapters", exc_info=True)
+        return {}
+
+
+# ---------------------------------------------------------------------------
+# Response / request models
+# ---------------------------------------------------------------------------
 
 
 class ThreadDeleteResponse(BaseModel):
@@ -183,10 +232,12 @@ class ReferenceSourceResponse(BaseModel):
     type: str = "result"
     snippet: str = ""
     source: str | None = None
+    author: str | None = None
     time: str | None = None
     url: str | None = None
     content: str | None = None
     score: float | None = None
+    tableColumns: list[dict[str, Any]] = Field(default_factory=list)
     raw: dict[str, Any] = Field(default_factory=dict)
 
 
@@ -794,11 +845,16 @@ async def get_thread_reference_batches(thread_id: str, request: Request) -> Refe
             _rag_debug_file_path(),
         )
         return ReferenceBatchesResponse(batches=[])
-    raw_batches = build_runtime_reference_batches(messages)
+    batches_before_display_adapt = build_runtime_reference_batches(messages)
+    raw_batches = batches_before_display_adapt
+    persisted_displays = await _load_persisted_skill_displays()
+    if persisted_displays:
+        raw_batches = adapt_reference_batches_with_displays(raw_batches, persisted_displays)
     debug_summary = build_reference_debug_summary(messages, raw_batches)
     _log_rag_debug(
         "backend reference-batches thread=%s messages=%s turns=%s tool_messages=%s raw_items=%s "
-        "normalized_items=%s kept_items=%s batches=%s batch_sources=%s batch_ids=%s turns_detail=%s log_file=%s",
+        "normalized_items=%s kept_items=%s batches=%s batch_sources=%s batch_ids=%s "
+        "display_adapters=%s turns_detail=%s log_file=%s",
         sanitize_log_param(thread_id),
         debug_summary["messages"],
         debug_summary["turns"],
@@ -809,9 +865,29 @@ async def get_thread_reference_batches(thread_id: str, request: Request) -> Refe
         len(raw_batches),
         ",".join(str(batch.get("source_count", 0)) for batch in debug_summary["batches"]),
         ",".join(str(batch.get("id") or "") for batch in debug_summary["batches"]),
+        len(persisted_displays),
         json.dumps(debug_summary["turns_detail"], ensure_ascii=False),
         _rag_debug_file_path(),
     )
+    _log_rag_debug(
+        "backend reference-batches-payload thread=%s %s",
+        sanitize_log_param(thread_id),
+        json.dumps(
+            {
+                "display_adapter_names": sorted(persisted_displays.keys()),
+                "before_display_adapt": build_reference_debug_payload(
+                    messages=messages,
+                    batches=batches_before_display_adapt,
+                ),
+                "after_display_adapt": build_reference_debug_payload(
+                    messages=messages,
+                    batches=raw_batches,
+                ),
+            },
+            ensure_ascii=False,
+            default=str,
+        ),
+    )
     return ReferenceBatchesResponse(
         batches=[
             ReferenceBatchResponse(

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

@@ -0,0 +1,108 @@
+"""CRUD API for a user's personal custom prompts (常用提示词).
+
+Prompts are bound to the requesting user (not stored in the browser), so they
+follow the account across devices/sessions.
+"""
+
+from __future__ import annotations
+
+import logging
+from datetime import datetime
+from uuid import uuid4
+
+from fastapi import APIRouter, HTTPException, Request
+from pydantic import BaseModel, Field
+
+from deerflow.runtime.user_context import get_effective_user_id
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/api/user-prompts", tags=["user-prompts"])
+
+
+class UserPromptResponse(BaseModel):
+    id: str
+    title: str = ""
+    content: str
+    sort_order: int = 0
+    created_at: datetime | str | None = None
+    updated_at: datetime | str | None = None
+
+
+class UserPromptsListResponse(BaseModel):
+    prompts: list[UserPromptResponse]
+
+
+class UserPromptCreateRequest(BaseModel):
+    title: str = Field(default="", max_length=255)
+    content: str = Field(..., min_length=1)
+    sort_order: int = Field(default=0)
+
+
+class UserPromptUpdateRequest(BaseModel):
+    title: str | None = Field(default=None, max_length=255)
+    content: str | None = Field(default=None, min_length=1)
+    sort_order: int | None = None
+
+
+def _current_user_id(request: Request) -> str:
+    user = getattr(request.state, "user", None)
+    if user is not None:
+        return str(user.id)
+    return get_effective_user_id()
+
+
+def _get_store(request: Request):
+    store = getattr(request.app.state, "user_prompt_store", None)
+    if store is None:
+        raise HTTPException(status_code=503, detail="User prompt store not available")
+    return store
+
+
+@router.get("", response_model=UserPromptsListResponse)
+async def list_user_prompts(request: Request) -> UserPromptsListResponse:
+    store = _get_store(request)
+    user_id = _current_user_id(request)
+    rows = await store.list_for_user(user_id)
+    return UserPromptsListResponse(prompts=[UserPromptResponse(**r) for r in rows])
+
+
+@router.post("", response_model=UserPromptResponse, status_code=201)
+async def create_user_prompt(request: Request, body: UserPromptCreateRequest) -> UserPromptResponse:
+    store = _get_store(request)
+    user_id = _current_user_id(request)
+    row = await store.create(
+        user_id,
+        {
+            "id": uuid4().hex,
+            "title": (body.title or "").strip(),
+            "content": body.content,
+            "sort_order": body.sort_order,
+        },
+    )
+    return UserPromptResponse(**row)
+
+
+@router.put("/{prompt_id}", response_model=UserPromptResponse)
+async def update_user_prompt(request: Request, prompt_id: str, body: UserPromptUpdateRequest) -> UserPromptResponse:
+    store = _get_store(request)
+    user_id = _current_user_id(request)
+    data: dict = {}
+    if body.title is not None:
+        data["title"] = body.title.strip()
+    if body.content is not None:
+        data["content"] = body.content
+    if body.sort_order is not None:
+        data["sort_order"] = body.sort_order
+    row = await store.update(prompt_id, user_id, data)
+    if row is None:
+        raise HTTPException(status_code=404, detail="Prompt not found")
+    return UserPromptResponse(**row)
+
+
+@router.delete("/{prompt_id}", status_code=204)
+async def delete_user_prompt(request: Request, prompt_id: str) -> None:
+    store = _get_store(request)
+    user_id = _current_user_id(request)
+    deleted = await store.delete(prompt_id, user_id)
+    if not deleted:
+        raise HTTPException(status_code=404, detail="Prompt not found")

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

@@ -137,7 +137,17 @@ class MemoryMiddleware(AgentMiddleware[MemoryMiddlewareState]):
 
         召回结果包在 ``<memory-context>`` 标签里并带系统提示,告诉模型这是
         背景信息而非用户的新输入。
+
+        **跳过开关**:若 ``runtime.context.memory_recall_disabled`` 为真,直接
+        返回 None。圆桌规划等临时性任务(intent / recommend / leader / special)
+        不需要从用户个人记忆里召回,Hindsight HTTP 同步阻塞会显著增加 LLM 首
+        字节延迟(前端"调度中"卡顿);由路由层通过 ``_run_payload`` 的 context
+        字段统一关闭。
         """
+        runtime = getattr(request, "runtime", None)
+        if runtime is not None and getattr(runtime, "context", None):
+            if runtime.context.get("memory_recall_disabled"):
+                return None
         if not self._injection_enabled:
             return None
         config = self._memory_config or get_memory_config()

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

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import json
 import logging
 import os
 from pathlib import Path
@@ -13,7 +14,11 @@ from langchain.agents.middleware import AgentMiddleware
 from langchain.agents.middleware.types import ModelCallResult, ModelRequest, ModelResponse
 from langchain_core.messages import SystemMessage
 
-from deerflow.runtime.references import build_latest_reference_prompt, build_reference_debug_summary
+from deerflow.runtime.references import (
+    build_latest_reference_prompt,
+    build_reference_debug_payload,
+    build_reference_debug_summary,
+)
 
 logger = logging.getLogger(__name__)
 _RAG_DEBUG_LOGGER_NAME = "rag_citation_debug"
@@ -84,6 +89,18 @@ class RagCitationMiddleware(AgentMiddleware[AgentState]):
             summary["kept_items"],
             len((batch or {}).get("sources") or []),
         )
+        _log_rag_debug(
+            "middleware reference-prompt-payload %s",
+            json.dumps(
+                build_reference_debug_payload(
+                    messages=messages,
+                    batch=batch,
+                    prompt=prompt,
+                ),
+                ensure_ascii=False,
+                default=str,
+            ),
+        )
         patched_messages = [
             *messages,
             SystemMessage(content=prompt),

offline-backend-20260512/backend/packages/harness/deerflow/config/agents_config.py

@@ -1,7 +1,9 @@
 """Configuration and loaders for custom agents."""
 
 import logging
+import os
 import re
+import threading
 from typing import Any
 
 import yaml
@@ -13,6 +15,25 @@ logger = logging.getLogger(__name__)
 
 SOUL_FILENAME = "SOUL.md"
 AGENT_ID_PATTERN = re.compile(r"^[A-Za-z0-9_-]+$")
+
+# Per-file mtime caches. These are read on every /api/agents list/detail request
+# (one read per agent) — caching keyed by (path, mtime_ns, size) is correct
+# because any edit to the file updates mtime, and an external tool re-writing
+# the same bytes doesn't actually change the result we'd return. Cache size is
+# unbounded but lookup is O(1) and entries are tiny (one parsed dict / string
+# per agent file); for the order of N≤10⁴ agents the memory cost is negligible.
+_config_cache: dict[str, tuple[tuple[int, int], "AgentConfig | None"]] = {}
+_soul_cache: dict[str, tuple[tuple[int, int], str | None]] = {}
+_cache_lock = threading.Lock()
+
+
+def _file_stat_key(path) -> tuple[int, int] | None:
+    """Return (mtime_ns, size) for cache invalidation, or None when missing."""
+    try:
+        st = os.stat(path)
+    except OSError:
+        return None
+    return (st.st_mtime_ns, st.st_size)
 # Backward-compatible alias. Historically this validated the display name;
 # after the DB-backed agent table, display names can be Chinese and duplicated.
 # The pattern now applies only to storage/runtime ids.
@@ -94,6 +115,18 @@ def load_agent_config(agent_id: str | None) -> AgentConfig | None:
         )
         return None
 
+    # mtime-based cache. The config file is rewritten every time the agent is
+    # edited via PUT /api/agents/{id}, which bumps mtime — so the next read
+    # picks up the change automatically. Missing config_file = key None below,
+    # bypassing the cache (cheap path: just return AgentConfig with id/name).
+    cache_path_key = str(config_file)
+    stat_key = _file_stat_key(config_file) if config_file.exists() else None
+    if stat_key is not None:
+        with _cache_lock:
+            cached = _config_cache.get(cache_path_key)
+        if cached is not None and cached[0] == stat_key:
+            return cached[1]
+
     data: dict[str, Any] = {}
     if config_file.exists():
         try:
@@ -112,7 +145,11 @@ def load_agent_config(agent_id: str | None) -> AgentConfig | None:
     known_fields = set(AgentConfig.model_fields.keys())
     data = {k: v for k, v in data.items() if k in known_fields}
 
-    return AgentConfig(**data)
+    result = AgentConfig(**data)
+    if stat_key is not None:
+        with _cache_lock:
+            _config_cache[cache_path_key] = (stat_key, result)
+    return result
 
 
 def load_agent_soul(agent_id: str | None) -> str | None:
@@ -131,8 +168,21 @@ def load_agent_soul(agent_id: str | None) -> str | None:
     soul_path = agent_dir / SOUL_FILENAME
     if not soul_path.exists():
         return None
+    # mtime-based cache, same rationale as load_agent_config: SOUL.md is
+    # rewritten on edit (PUT /api/agents/{id}), which bumps mtime.
+    cache_path_key = str(soul_path)
+    stat_key = _file_stat_key(soul_path)
+    if stat_key is not None:
+        with _cache_lock:
+            cached = _soul_cache.get(cache_path_key)
+        if cached is not None and cached[0] == stat_key:
+            return cached[1]
     content = soul_path.read_text(encoding="utf-8").strip()
-    return content or None
+    result = content or None
+    if stat_key is not None:
+        with _cache_lock:
+            _soul_cache[cache_path_key] = (stat_key, result)
+    return result
 
 
 def list_custom_agents() -> list[AgentConfig]:

offline-backend-20260512/backend/packages/harness/deerflow/config/extensions_config.py

@@ -75,6 +75,7 @@ class SkillDisplayConfig(BaseModel):
     summary_field: str | None = Field(default=None, description="Field used as short summary/snippet", alias="summaryField")
     content_field: str | None = Field(default=None, description="Field used as full item content", alias="contentField")
     source_field: str | None = Field(default=None, description="Field used as source label", alias="sourceField")
+    author_field: str | None = Field(default=None, description="Field used as author/byline label", alias="authorField")
     type_field: str | None = Field(default=None, description="Field used as source/type badge", alias="typeField")
     time_field: str | None = Field(default=None, description="Field used as publication/update time", alias="timeField")
     score_field: str | None = Field(default=None, description="Field used as relevance score", alias="scoreField")

offline-backend-20260512/backend/packages/harness/deerflow/persistence/agents/__init__.py

@@ -4,10 +4,11 @@ from __future__ import annotations
 
 from deerflow.persistence.agents.base import AgentStore
 from deerflow.persistence.agents.memory import MemoryAgentStore
-from deerflow.persistence.agents.model import AgentRow
+from deerflow.persistence.agents.model import AgentFavoriteRow, AgentRow
 from deerflow.persistence.agents.sql import AgentRepository
 
 __all__ = [
+    "AgentFavoriteRow",
     "AgentRepository",
     "AgentRow",
     "AgentStore",

offline-backend-20260512/backend/packages/harness/deerflow/persistence/agents/base.py

@@ -117,3 +117,25 @@ class AgentStore(ABC):
         ``False`` if the agent does not exist.
         """
         raise NotImplementedError
+
+    # --- Favorites ("收藏") ---
+
+    @abstractmethod
+    async def add_favorite(self, user_id: str, agent_id: str) -> None:
+        """Bookmark ``agent_id`` for ``user_id``. Idempotent."""
+        raise NotImplementedError
+
+    @abstractmethod
+    async def remove_favorite(self, user_id: str, agent_id: str) -> bool:
+        """Remove a bookmark. Returns ``False`` when none existed."""
+        raise NotImplementedError
+
+    @abstractmethod
+    async def list_favorite_ids(self, user_id: str) -> set[str]:
+        """Return the set of agent ids ``user_id`` has favorited."""
+        raise NotImplementedError
+
+    @abstractmethod
+    async def delete_favorites_for_agent(self, agent_id: str) -> None:
+        """Drop every user's favorite of ``agent_id`` (called on agent delete)."""
+        raise NotImplementedError

offline-backend-20260512/backend/packages/harness/deerflow/persistence/agents/memory.py

@@ -21,6 +21,7 @@ def _copy(data: dict[str, Any]) -> dict[str, Any]:
 class MemoryAgentStore(AgentStore):
     def __init__(self) -> None:
         self._agents: dict[str, dict[str, Any]] = {}
+        self._favorites: set[tuple[str, str]] = set()
 
     @staticmethod
     def _is_visible(agent: dict[str, Any], user_id: str) -> bool:
@@ -164,3 +165,20 @@ class MemoryAgentStore(AgentStore):
             return False
         self._agents.pop(agent_id, None)
         return True
+
+    # --- Favorites ("收藏") ---
+
+    async def add_favorite(self, user_id: str, agent_id: str) -> None:
+        self._favorites.add((user_id, agent_id))
+
+    async def remove_favorite(self, user_id: str, agent_id: str) -> bool:
+        if (user_id, agent_id) in self._favorites:
+            self._favorites.discard((user_id, agent_id))
+            return True
+        return False
+
+    async def list_favorite_ids(self, user_id: str) -> set[str]:
+        return {agent_id for (uid, agent_id) in self._favorites if uid == user_id}
+
+    async def delete_favorites_for_agent(self, agent_id: str) -> None:
+        self._favorites = {(uid, aid) for (uid, aid) in self._favorites if aid != agent_id}

offline-backend-20260512/backend/packages/harness/deerflow/persistence/agents/model.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 from datetime import UTC, datetime
 
-from sqlalchemy import BigInteger, Boolean, String
+from sqlalchemy import BigInteger, Boolean, String, UniqueConstraint
 from sqlalchemy.orm import Mapped, mapped_column
 
 from deerflow.persistence.base import Base
@@ -35,3 +35,20 @@ class AgentRow(Base):
     featured_order: Mapped[int | None] = mapped_column(BigInteger, nullable=True, index=True)
     created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC))
     updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
+
+
+class AgentFavoriteRow(Base):
+    """A user's bookmark of a (typically public-square) agent.
+
+    Lets a user collect agents from the 广场 (square) into their "我的" tab
+    without owning them. One row per (user, agent); the unique constraint makes
+    favoriting idempotent.
+    """
+
+    __tablename__ = "agent_favorites"
+    __table_args__ = (UniqueConstraint("user_id", "agent_id", name="uq_agent_favorites_user_agent"),)
+
+    id: Mapped[str] = mapped_column(String(64), primary_key=True)
+    user_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    agent_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC))

offline-backend-20260512/backend/packages/harness/deerflow/persistence/agents/sql.py

@@ -4,12 +4,14 @@ from __future__ import annotations
 
 from datetime import UTC, datetime
 from typing import Any
+from uuid import uuid4
 
-from sqlalchemy import and_, func, or_, select
+from sqlalchemy import and_, delete, func, or_, select
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from deerflow.persistence.agents.base import AgentStore
-from deerflow.persistence.agents.model import AgentRow
+from deerflow.persistence.agents.model import AgentFavoriteRow, AgentRow
 
 
 def _row_to_dict(row: AgentRow) -> dict[str, Any]:
@@ -230,3 +232,47 @@ class AgentRepository(AgentStore):
             await session.delete(row)
             await session.commit()
             return True
+
+    # --- Favorites ("收藏") ---
+
+    async def add_favorite(self, user_id: str, agent_id: str) -> None:
+        async with self._sf() as session:
+            existing = (
+                await session.execute(
+                    select(AgentFavoriteRow.id)
+                    .where(AgentFavoriteRow.user_id == user_id, AgentFavoriteRow.agent_id == agent_id)
+                    .limit(1)
+                )
+            ).scalar_one_or_none()
+            if existing is not None:
+                return
+            session.add(AgentFavoriteRow(id=uuid4().hex, user_id=user_id, agent_id=agent_id))
+            try:
+                await session.commit()
+            except IntegrityError:
+                # Lost a race against a concurrent favorite of the same agent —
+                # the unique constraint already guarantees the desired state.
+                await session.rollback()
+
+    async def remove_favorite(self, user_id: str, agent_id: str) -> bool:
+        async with self._sf() as session:
+            result = await session.execute(
+                delete(AgentFavoriteRow).where(
+                    AgentFavoriteRow.user_id == user_id,
+                    AgentFavoriteRow.agent_id == agent_id,
+                )
+            )
+            await session.commit()
+            return (result.rowcount or 0) > 0
+
+    async def list_favorite_ids(self, user_id: str) -> set[str]:
+        async with self._sf() as session:
+            result = await session.execute(
+                select(AgentFavoriteRow.agent_id).where(AgentFavoriteRow.user_id == user_id)
+            )
+            return {row[0] for row in result.all()}
+
+    async def delete_favorites_for_agent(self, agent_id: str) -> None:
+        async with self._sf() as session:
+            await session.execute(delete(AgentFavoriteRow).where(AgentFavoriteRow.agent_id == agent_id))
+            await session.commit()

offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260528_01_agent_favorites.py

@@ -0,0 +1,45 @@
+"""Add ``agent_favorites`` for the agent collection (收藏) feature.
+
+Lets a user bookmark agents from the public square (广场) so they surface in
+their "我的" tab without owning them. One row per (user, agent).
+
+Revision ID: 20260528_01
+Revises: 20260526_01
+Create Date: 2026-05-28
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy import inspect
+
+
+revision = "20260528_01"
+down_revision = "20260526_01"
+branch_labels = None
+depends_on = None
+
+
+def _has_table(table_name: str) -> bool:
+    inspector = inspect(op.get_bind())
+    return table_name in inspector.get_table_names()
+
+
+def upgrade() -> None:
+    if not _has_table("agent_favorites"):
+        op.create_table(
+            "agent_favorites",
+            sa.Column("id", sa.String(64), primary_key=True),
+            sa.Column("user_id", sa.String(64), nullable=False),
+            sa.Column("agent_id", sa.String(64), nullable=False),
+            sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+            sa.UniqueConstraint("user_id", "agent_id", name="uq_agent_favorites_user_agent"),
+        )
+        op.create_index("ix_agent_favorites_user_id", "agent_favorites", ["user_id"])
+        op.create_index("ix_agent_favorites_agent_id", "agent_favorites", ["agent_id"])
+
+
+def downgrade() -> None:
+    if _has_table("agent_favorites"):
+        op.drop_table("agent_favorites")

offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260528_02_user_custom_prompts.py

@@ -0,0 +1,45 @@
+"""Add ``user_custom_prompts`` for per-user saved prompts (常用提示词).
+
+Bound to the user (not the browser) so prompts follow the account.
+
+Revision ID: 20260528_02
+Revises: 20260528_01
+Create Date: 2026-05-28
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy import inspect
+
+
+revision = "20260528_02"
+down_revision = "20260528_01"
+branch_labels = None
+depends_on = None
+
+
+def _has_table(table_name: str) -> bool:
+    inspector = inspect(op.get_bind())
+    return table_name in inspector.get_table_names()
+
+
+def upgrade() -> None:
+    if not _has_table("user_custom_prompts"):
+        op.create_table(
+            "user_custom_prompts",
+            sa.Column("id", sa.String(64), primary_key=True),
+            sa.Column("user_id", sa.String(64), nullable=False),
+            sa.Column("title", sa.String(255), nullable=False, server_default=""),
+            sa.Column("content", sa.Text(), nullable=False),
+            sa.Column("sort_order", sa.Integer(), nullable=False, server_default="0"),
+            sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+            sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        )
+        op.create_index("ix_user_custom_prompts_user_id", "user_custom_prompts", ["user_id"])
+
+
+def downgrade() -> None:
+    if _has_table("user_custom_prompts"):
+        op.drop_table("user_custom_prompts")

offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py

@@ -15,7 +15,7 @@ its storage implementation lives in ``deerflow.runtime.events.store.db`` and
 there is no matching entity directory.
 """
 
-from deerflow.persistence.agents.model import AgentRow
+from deerflow.persistence.agents.model import AgentFavoriteRow, AgentRow
 from deerflow.persistence.memory_config.model import UserMemoryConfigRow
 from deerflow.persistence.feedback.model import FeedbackRow
 from deerflow.persistence.llm_metrics.model import LlmCallMetricRow
@@ -36,9 +36,11 @@ from deerflow.persistence.thread_meta.model import ThreadMetaRow
 from deerflow.persistence.thread_shares.model import ThreadShareRow
 from deerflow.persistence.tool_metrics.model import ToolCallMetricRow
 from deerflow.persistence.user.model import UserRow
+from deerflow.persistence.user_prompts.model import UserCustomPromptRow
 
 __all__ = [
     "FeedbackRow",
+    "AgentFavoriteRow",
     "AgentRow",
     "LlmCallMetricRow",
     "NotificationRow",
@@ -57,6 +59,7 @@ __all__ = [
     "ThreadMetaRow",
     "ThreadShareRow",
     "ToolCallMetricRow",
+    "UserCustomPromptRow",
     "UserMemoryConfigRow",
     "UserRow",
 ]

offline-backend-20260512/backend/packages/harness/deerflow/persistence/user_prompts/__init__.py

@@ -0,0 +1,18 @@
+"""Per-user custom prompt persistence."""
+
+from __future__ import annotations
+
+from deerflow.persistence.user_prompts.model import UserCustomPromptRow
+from deerflow.persistence.user_prompts.sql import UserCustomPromptRepository
+
+__all__ = [
+    "UserCustomPromptRepository",
+    "UserCustomPromptRow",
+    "make_user_prompt_store",
+]
+
+
+def make_user_prompt_store(session_factory):
+    if session_factory is None:
+        return None
+    return UserCustomPromptRepository(session_factory)

offline-backend-20260512/backend/packages/harness/deerflow/persistence/user_prompts/model.py

@@ -0,0 +1,23 @@
+"""ORM model for per-user custom prompts (常用提示词)."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+from sqlalchemy import Integer, String, Text
+from sqlalchemy.orm import Mapped, mapped_column
+
+from deerflow.persistence.base import Base
+from deerflow.persistence.types import BeijingDateTime
+
+
+class UserCustomPromptRow(Base):
+    __tablename__ = "user_custom_prompts"
+
+    id: Mapped[str] = mapped_column(String(64), primary_key=True)
+    user_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    title: Mapped[str] = mapped_column(String(255), nullable=False, default="")
+    content: Mapped[str] = mapped_column(Text, nullable=False)
+    sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC))
+    updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))

offline-backend-20260512/backend/packages/harness/deerflow/persistence/user_prompts/sql.py

@@ -0,0 +1,74 @@
+"""SQLAlchemy-backed per-user custom prompt repository."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from deerflow.persistence.user_prompts.model import UserCustomPromptRow
+
+
+def _row_to_dict(row: UserCustomPromptRow) -> dict[str, Any]:
+    data = row.to_dict()
+    for key in ("created_at", "updated_at"):
+        value = data.get(key)
+        if isinstance(value, datetime):
+            data[key] = value.isoformat()
+    return data
+
+
+class UserCustomPromptRepository:
+    def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
+        self._sf = session_factory
+
+    async def list_for_user(self, user_id: str) -> list[dict[str, Any]]:
+        stmt = (
+            select(UserCustomPromptRow)
+            .where(UserCustomPromptRow.user_id == user_id)
+            .order_by(UserCustomPromptRow.sort_order.asc(), UserCustomPromptRow.created_at.asc())
+        )
+        async with self._sf() as session:
+            result = await session.execute(stmt)
+            return [_row_to_dict(row) for row in result.scalars()]
+
+    async def create(self, user_id: str, data: dict[str, Any]) -> dict[str, Any]:
+        now = datetime.now(UTC)
+        row = UserCustomPromptRow(
+            id=data["id"],
+            user_id=user_id,
+            title=data.get("title") or "",
+            content=data["content"],
+            sort_order=data.get("sort_order", 0),
+            created_at=now,
+            updated_at=now,
+        )
+        async with self._sf() as session:
+            session.add(row)
+            await session.commit()
+            await session.refresh(row)
+            return _row_to_dict(row)
+
+    async def update(self, prompt_id: str, user_id: str, data: dict[str, Any]) -> dict[str, Any] | None:
+        async with self._sf() as session:
+            row = await session.get(UserCustomPromptRow, prompt_id)
+            if row is None or row.user_id != user_id:
+                return None
+            for key in ("title", "content", "sort_order"):
+                if key in data:
+                    setattr(row, key, data[key])
+            row.updated_at = datetime.now(UTC)
+            await session.commit()
+            await session.refresh(row)
+            return _row_to_dict(row)
+
+    async def delete(self, prompt_id: str, user_id: str) -> bool:
+        async with self._sf() as session:
+            row = await session.get(UserCustomPromptRow, prompt_id)
+            if row is None or row.user_id != user_id:
+                return False
+            await session.delete(row)
+            await session.commit()
+            return True

offline-backend-20260512/backend/packages/harness/deerflow/runtime/references.py

@@ -3,7 +3,14 @@
 from __future__ import annotations
 
 import json
+import re
 from typing import Any
+from urllib.parse import quote
+
+from deerflow.config.extensions_config import SkillDisplayConfig, get_extensions_config
+
+
+_SKILL_PATH_RE = re.compile(r"[\\/]skills[\\/](?:public|custom)[\\/]([^\\/]+)(?:[\\/]|$)")
 
 
 def build_reference_batches(messages: list[Any]) -> list[dict[str, Any]]:
@@ -16,12 +23,10 @@ def build_reference_batches(messages: list[Any]) -> list[dict[str, Any]]:
             if _message_type(messages[cursor]) == "human":
                 end = cursor
                 break
-        assistant = _find_last_assistant_answer(messages, index + 1, end)
-        if assistant is None:
-            continue
         sources = _collect_sources(messages[index + 1 : end])
         if sources:
-            message_id = _message_id(assistant) or str(index)
+            anchor = _find_reference_batch_anchor(messages, index + 1, end)
+            message_id = _message_id(anchor) or _message_id(message) or str(index)
             batches.append(
                 {
                     "id": message_id,
@@ -39,7 +44,7 @@ def build_latest_reference_prompt(messages: list[Any]) -> tuple[str, dict[str, A
     sources = _collect_sources(messages[human_index + 1 :])
     if not sources:
         return "", None
-    assistant = _find_last_assistant_answer(messages, human_index + 1, len(messages))
+    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 = {
         "id": message_id,
@@ -55,7 +60,7 @@ def build_latest_reference_prompt(messages: list[Any]) -> tuple[str, dict[str, A
         title = source.get("title") or f"参考文献 {source.get('index')}"
         meta = " · ".join(
             str(item)
-            for item in (source.get("source"), source.get("time"))
+            for item in (source.get("source"), source.get("author"), source.get("time"))
             if item
         )
         content = str(source.get("content") or source.get("snippet") or "").strip()
@@ -76,6 +81,7 @@ def build_latest_reference_prompt(messages: list[Any]) -> tuple[str, dict[str, A
             "3. 只能引用上方存在的编号，不要编造编号。",
             "4. 不要使用工具原始编号，也不要重新排序编号。",
             "5. 不要在正文末尾堆完整参考文献列表，前端会展示右侧参考文献。",
+            "6. 参考文献模式下不要输出 <skill-display .../> 占位符，也不要在正文里额外渲染卡片、表格或列表；前端会在右侧参考文献区域展示。",
         ]
     )
     return "\n".join(lines).strip(), batch
@@ -152,14 +158,191 @@ def build_reference_debug_summary(messages: list[Any], batches: list[dict[str, A
     }
 
 
+def build_reference_debug_payload(
+    *,
+    messages: list[Any] | None = None,
+    batch: dict[str, Any] | None = None,
+    batches: list[dict[str, Any]] | None = None,
+    prompt: str | None = None,
+) -> dict[str, Any]:
+    """Return a JSON-safe payload for comparing model prompts and UI batches."""
+    payload: dict[str, Any] = {}
+    if messages is not None:
+        payload["message_trace"] = [_message_debug_item(index, message) for index, message in enumerate(messages)]
+    if prompt is not None:
+        payload["model_reference_prompt"] = prompt
+    if batch is not None:
+        payload["batch"] = _batch_debug_item(batch)
+    if batches is not None:
+        payload["batches"] = [_batch_debug_item(item) for item in batches]
+    return payload
+
+
+def adapt_reference_batches_with_displays(
+    batches: list[dict[str, Any]],
+    displays: dict[str, SkillDisplayConfig | dict[str, Any]],
+) -> list[dict[str, Any]]:
+    """Apply persisted skill display configs to already-built reference batches.
+
+    The reference builder can run before the gateway has a fresh extensions
+    cache, while the skill adapter editor persists the user's latest mapping in
+    the database. This pass lets the API response use that DB-backed mapping
+    without changing citation indexes.
+    """
+    normalized_displays: dict[str, SkillDisplayConfig] = {}
+    for skill_name, display in displays.items():
+        try:
+            config = (
+                display
+                if isinstance(display, SkillDisplayConfig)
+                else SkillDisplayConfig.model_validate(display)
+            )
+        except Exception:
+            continue
+        if not config.enabled or config.mode == "none":
+            continue
+        normalized_displays[skill_name] = config
+
+    if not normalized_displays:
+        return batches
+
+    displays_by_normalized_name = {
+        _normalize_name(name): (name, display)
+        for name, display in normalized_displays.items()
+    }
+    adapted_batches: list[dict[str, Any]] = []
+    for batch in batches:
+        sources: list[dict[str, Any]] = []
+        for source in batch.get("sources", []):
+            if not isinstance(source, dict):
+                continue
+            skill_name = str(source.get("skillName") or "")
+            display = normalized_displays.get(skill_name)
+            if display is None:
+                normalized_source = _normalize_name(skill_name)
+                normalized_match = displays_by_normalized_name.get(normalized_source)
+                if normalized_match is None:
+                    for normalized_skill, candidate in displays_by_normalized_name.items():
+                        if (
+                            normalized_source.endswith(normalized_skill)
+                            or normalized_skill in normalized_source
+                        ):
+                            normalized_match = candidate
+                            break
+                if normalized_match is not None:
+                    skill_name, display = normalized_match
+            raw = source.get("raw")
+            if display is not None and isinstance(raw, dict):
+                index_value = int(source.get("index") or len(sources) + 1)
+                configured = _normalize_with_display_config(
+                    raw,
+                    index_value,
+                    skill_name,
+                    display,
+                )
+                if configured is not None:
+                    sources.append(
+                        {
+                            **source,
+                            **configured,
+                            "index": index_value,
+                            "skillName": skill_name,
+                            "raw": raw,
+                        }
+                    )
+                    continue
+            sources.append(source)
+        adapted_batches.append({**batch, "sources": sources})
+    return adapted_batches
+
+
+def _message_debug_item(index: int, message: Any) -> dict[str, Any]:
+    content = _message_content(message)
+    tool_calls = _extract_tool_calls(message)
+    return {
+        "position": index,
+        "type": _message_type(message),
+        "id": _message_id(message),
+        "name": _message_name(message) if _message_type(message) == "tool" else None,
+        "tool_call_id": _message_tool_call_id(message) if _message_type(message) == "tool" else None,
+        "tool_calls": [
+            {
+                "id": call.get("id"),
+                "name": call.get("name"),
+                "args": call.get("args"),
+            }
+            for call in tool_calls
+        ],
+        "content_length": len(content) if isinstance(content, str) else None,
+    }
+
+
+def _batch_debug_item(batch: dict[str, Any]) -> dict[str, Any]:
+    sources = [source for source in batch.get("sources", []) if isinstance(source, dict)]
+    return {
+        "id": batch.get("id"),
+        "assistant_message_id": batch.get("assistant_message_id"),
+        "source_count": len(sources),
+        "sources": [_source_debug_item(source) for source in sources],
+    }
+
+
+def _source_debug_item(source: dict[str, Any]) -> dict[str, Any]:
+    raw = source.get("raw")
+    raw_record = raw if isinstance(raw, dict) else {}
+    return {
+        "index": source.get("index"),
+        "id": source.get("id"),
+        "skillName": source.get("skillName"),
+        "mode": source.get("mode"),
+        "title": source.get("title"),
+        "source": source.get("source"),
+        "author": source.get("author"),
+        "time": source.get("time"),
+        "url": source.get("url"),
+        "snippet": source.get("snippet"),
+        "content": source.get("content"),
+        "raw_id_candidates": {
+            "recUuid": raw_record.get("recUuid"),
+            "rec_uuid": raw_record.get("rec_uuid"),
+            "uuid": raw_record.get("uuid"),
+            "docId": raw_record.get("docId"),
+            "doc_id": raw_record.get("doc_id"),
+            "id": raw_record.get("id"),
+            "url": raw_record.get("url"),
+        },
+        "raw": raw,
+    }
+
+
 def _collect_sources(messages: list[Any]) -> list[dict[str, Any]]:
     sources: list[dict[str, Any]] = []
+    pending_tool_contexts: list[dict[str, Any]] = []
+    tool_contexts_by_id: dict[str, dict[str, Any]] = {}
     for message in messages:
-        if _message_type(message) != "tool":
+        message_type = _message_type(message)
+        if message_type == "ai":
+            for tool_call in _extract_tool_calls(message):
+                context = _build_tool_context(tool_call)
+                pending_tool_contexts.append(context)
+                if context.get("id"):
+                    tool_contexts_by_id[str(context["id"])] = context
             continue
+        if message_type != "tool":
+            continue
+        tool_context = _resolve_tool_context(
+            message,
+            pending_tool_contexts,
+            tool_contexts_by_id,
+        )
         payload = _extract_tool_payload(message)
         for raw_item in _first_list(payload):
-            normalized = _normalize_reference_item(raw_item, len(sources) + 1)
+            normalized = _normalize_reference_item(
+                raw_item,
+                len(sources) + 1,
+                payload=payload,
+                tool_context=tool_context,
+            )
             if not normalized:
                 continue
             normalized["index"] = len(sources) + 1
@@ -167,6 +350,17 @@ def _collect_sources(messages: list[Any]) -> list[dict[str, Any]]:
     return sources
 
 
+def _find_reference_batch_anchor(messages: list[Any], start: int, end: int) -> Any | None:
+    answer = _find_last_assistant_answer(messages, start, end)
+    if answer is not None:
+        return answer
+    for index in range(end - 1, start - 1, -1):
+        candidate = messages[index]
+        if _message_type(candidate) == "ai" and _message_id(candidate):
+            return candidate
+    return None
+
+
 def _find_last_assistant_answer(messages: list[Any], start: int, end: int) -> Any | None:
     for index in range(end - 1, start - 1, -1):
         candidate = messages[index]
@@ -174,7 +368,6 @@ def _find_last_assistant_answer(messages: list[Any], start: int, end: int) -> An
             _message_type(candidate) == "ai"
             and _message_id(candidate)
             and _has_answer_content(candidate)
-            and not _has_tool_calls(candidate)
         ):
             return candidate
     return None
@@ -195,6 +388,20 @@ def _message_content(message: Any) -> Any:
     return message.get("content") if isinstance(message, dict) else getattr(message, "content", None)
 
 
+def _message_name(message: Any) -> str:
+    if isinstance(message, dict):
+        return str(message.get("name") or (message.get("additional_kwargs") or {}).get("name") or "tool")
+    return str(getattr(message, "name", None) or "tool")
+
+
+def _message_tool_call_id(message: Any) -> str | None:
+    if isinstance(message, dict):
+        value = message.get("tool_call_id") or message.get("toolCallId")
+    else:
+        value = getattr(message, "tool_call_id", None) or getattr(message, "toolCallId", None)
+    return str(value) if value else None
+
+
 def _has_tool_calls(message: Any) -> bool:
     if isinstance(message, dict):
         direct = message.get("tool_calls")
@@ -221,6 +428,102 @@ def _has_answer_content(message: Any) -> bool:
     return False
 
 
+def _extract_tool_calls(message: Any) -> list[dict[str, Any]]:
+    candidates: list[Any] = []
+    if isinstance(message, dict):
+        for key in ("tool_calls", "tool_call_chunks"):
+            value = message.get(key)
+            if isinstance(value, list):
+                candidates.extend(value)
+        extra = (message.get("additional_kwargs") or {}).get("tool_calls")
+        if isinstance(extra, list):
+            candidates.extend(extra)
+    else:
+        for key in ("tool_calls", "tool_call_chunks"):
+            value = getattr(message, key, None)
+            if isinstance(value, list):
+                candidates.extend(value)
+        extra = (getattr(message, "additional_kwargs", None) or {}).get("tool_calls")
+        if isinstance(extra, list):
+            candidates.extend(extra)
+
+    calls: list[dict[str, Any]] = []
+    seen_ids: set[str] = set()
+    for raw in candidates:
+        call = _normalize_tool_call(raw)
+        if not call:
+            continue
+        call_id = str(call.get("id") or "")
+        if call_id and call_id in seen_ids:
+            continue
+        if call_id:
+            seen_ids.add(call_id)
+        calls.append(call)
+    return calls
+
+
+def _normalize_tool_call(raw: Any) -> dict[str, Any] | None:
+    if not isinstance(raw, dict):
+        return None
+    fn = raw.get("function") if isinstance(raw.get("function"), dict) else {}
+    name = raw.get("name") or fn.get("name") or raw.get("tool_name")
+    if not name:
+        return None
+    return {
+        "id": raw.get("id") or raw.get("tool_call_id") or raw.get("call_id"),
+        "name": str(name),
+        "args": _normalize_tool_args(raw.get("args") or raw.get("arguments") or fn.get("arguments")),
+    }
+
+
+def _normalize_tool_args(value: Any) -> dict[str, Any]:
+    if isinstance(value, dict):
+        return value
+    if isinstance(value, str):
+        try:
+            parsed = json.loads(value)
+            if isinstance(parsed, dict):
+                return parsed
+        except Exception:
+            pass
+        return {"command": value}
+    return {}
+
+
+def _build_tool_context(tool_call: dict[str, Any]) -> dict[str, Any]:
+    tool_name = str(tool_call.get("name") or "tool")
+    args = tool_call.get("args") if isinstance(tool_call.get("args"), dict) else {}
+    return {
+        "id": tool_call.get("id"),
+        "tool_name": tool_name,
+        "args": args,
+        "skill_name": _infer_skill_name(tool_name, args),
+    }
+
+
+def _resolve_tool_context(
+    message: Any,
+    pending_tool_contexts: list[dict[str, Any]],
+    tool_contexts_by_id: dict[str, dict[str, Any]],
+) -> dict[str, Any]:
+    tool_call_id = _message_tool_call_id(message)
+    if tool_call_id and tool_call_id in tool_contexts_by_id:
+        context = tool_contexts_by_id[tool_call_id]
+        pending_tool_contexts[:] = [
+            item for item in pending_tool_contexts if str(item.get("id") or "") != tool_call_id
+        ]
+        return context
+    if pending_tool_contexts:
+        return pending_tool_contexts.pop(0)
+    tool_name = _message_name(message)
+    return {
+        "id": tool_call_id,
+        "tool_name": tool_name,
+        "args": {},
+        "skill_name": _infer_skill_name(tool_name, {}),
+    }
+
+
 def _extract_tool_payload(message: Any) -> Any:
     content = _message_content(message)
     if isinstance(content, dict):
@@ -313,17 +616,189 @@ def _pick(layers: list[dict[str, Any]], keys: tuple[str, ...]) -> str | None:
             value = layer.get(key)
             if isinstance(value, (str, int, float)) and str(value).strip():
                 return str(value).strip()
+            if isinstance(value, list):
+                text = _stringify(value, "")
+                if text:
+                    return text
     return None
 
 
-def _normalize_reference_item(raw: Any, fallback_index: int) -> dict[str, Any] | None:
+def _enabled_skill_displays() -> dict[str, SkillDisplayConfig]:
+    try:
+        return {
+            name: state.display
+            for name, state in get_extensions_config().skills.items()
+            if state.display is not None
+            and state.enabled
+            and state.display.enabled
+            and state.display.mode != "none"
+        }
+    except Exception:
+        return {}
+
+
+def _infer_skill_name(tool_name: str, args: dict[str, Any]) -> str | None:
+    displays = _enabled_skill_displays()
+    if tool_name in displays:
+        return tool_name
+    normalized_tool = _normalize_name(tool_name)
+    for skill_name in displays:
+        normalized_skill = _normalize_name(skill_name)
+        if (
+            normalized_tool == normalized_skill
+            or normalized_tool.endswith(normalized_skill)
+            or normalized_skill in normalized_tool
+        ):
+            return skill_name
+    for value in args.values():
+        if not isinstance(value, str):
+            continue
+        match = _SKILL_PATH_RE.search(value)
+        if match:
+            return match.group(1)
+    return None
+
+
+def _resolve_display_match(
+    raw: dict[str, Any],
+    payload: Any,
+    tool_context: dict[str, Any] | None,
+) -> tuple[str, SkillDisplayConfig] | None:
+    displays = _enabled_skill_displays()
+    if not displays:
+        return None
+    skill_name = str((tool_context or {}).get("skill_name") or "")
+    display = displays.get(skill_name)
+    if display is not None:
+        return skill_name, display
+    tool_name = str((tool_context or {}).get("tool_name") or "")
+    inferred = _infer_skill_name(tool_name, (tool_context or {}).get("args") or {})
+    if inferred and inferred in displays:
+        return inferred, displays[inferred]
+    return None
+
+
+def _normalize_with_display_config(
+    raw: dict[str, Any],
+    fallback_index: int,
+    skill_name: str,
+    display: SkillDisplayConfig,
+) -> dict[str, Any] | None:
+    title = _stringify(_get_by_path(raw, display.title_field), f"结果 {fallback_index}")
+    snippet = _stringify(_get_by_path(raw, display.summary_field), "")
+    content = _stringify(_get_by_path(raw, display.content_field), snippet)
+    source = _stringify(_get_by_path(raw, display.source_field), "") or None
+    author = _stringify(_get_by_path(raw, display.author_field), "") or None
+    item_type = _stringify(_get_by_path(raw, display.type_field), "result") or "result"
+    time = _stringify(_get_by_path(raw, display.time_field), "") or None
+    item_id = _stringify(_get_by_path(raw, display.id_field), "") or None
+    score = _number_or_none(_get_by_path(raw, display.score_field))
+    url = _resolve_display_link(raw, display)
+    if not title and not content and not url:
+        return None
+    return {
+        "id": item_id or url,
+        "skillName": skill_name,
+        "mode": display.mode,
+        "title": title,
+        "type": item_type,
+        "snippet": snippet,
+        "content": content,
+        "source": source,
+        "author": author,
+        "time": time,
+        "url": url,
+        "score": score,
+        "tableColumns": [column.model_dump(by_alias=True) for column in display.table_columns],
+        "raw": raw,
+    }
+
+
+def _resolve_display_link(raw: dict[str, Any], display: SkillDisplayConfig) -> str | None:
+    url = _stringify(_get_by_path(raw, display.url_field), "")
+    item_id = _stringify(_get_by_path(raw, display.id_field), "")
+    if display.link_strategy == "url":
+        return url or None
+    if display.link_strategy == "id":
+        return _render_template(display.link_template, raw) if item_id and display.link_template else None
+    return url or (_render_template(display.link_template, raw) if item_id and display.link_template else None) or None
+
+
+def _render_template(template: str | None, raw: dict[str, Any]) -> str | None:
+    if not template:
+        return None
+
+    def replace(match: re.Match[str]) -> str:
+        value = _get_by_path(raw, match.group(1).strip())
+        return quote(_stringify(value, ""), safe="")
+
+    return re.sub(r"\{([^}]+)\}", replace, template)
+
+
+def _get_by_path(value: Any, path: str | None) -> Any:
+    if not path:
+        return None
+    cursor = value
+    for part in path.split("."):
+        if isinstance(cursor, dict):
+            cursor = cursor.get(part)
+        else:
+            return None
+    return cursor
+
+
+def _stringify(value: Any, fallback: str = "") -> str:
+    if value is None:
+        return fallback
+    if isinstance(value, (str, int, float)):
+        text = str(value).strip()
+        return text if text else fallback
+    if isinstance(value, list):
+        parts = [
+            str(item).strip()
+            for item in value
+            if isinstance(item, (str, int, float)) and str(item).strip()
+        ]
+        return "、".join(parts) if parts else fallback
+    return fallback
+
+
+def _number_or_none(value: Any) -> float | None:
+    if isinstance(value, (int, float)):
+        return float(value)
+    if isinstance(value, str):
+        try:
+            return float(value)
+        except Exception:
+            return None
+    return None
+
+
+def _normalize_name(name: str) -> str:
+    return re.sub(r"[^a-z0-9]+", "", name.lower())
+
+
+def _normalize_reference_item(
+    raw: Any,
+    fallback_index: int,
+    *,
+    payload: Any = None,
+    tool_context: dict[str, Any] | None = None,
+) -> dict[str, Any] | None:
     if not isinstance(raw, dict):
         return None
+    display_match = _resolve_display_match(raw, payload, tool_context)
+    if display_match is not None:
+        skill_name, display = display_match
+        configured = _normalize_with_display_config(raw, fallback_index, skill_name, display)
+        if configured is not None:
+            return configured
     layers = _nested_records(raw)
     title = _pick(layers, ("title", "m_title", "name", "question", "keyword", "query", "text"))
     content = _pick(layers, ("content", "page_content", "summary", "snippet", "description", "content_preview", "abstract", "body"))
     url = _pick(layers, ("url", "link", "href", "source_url", "web_url"))
     source = _pick(layers, ("source", "source1", "site", "provider", "from"))
+    author = _pick(layers, ("author", "authors", "creator", "writer", "byline", "publisher", "owner"))
     time = _pick(layers, ("time", "date", "published_date", "publishedDate", "m_publish", "publish_time", "created_at"))
     item_id = _pick(layers, ("recUuid", "rec_uuid", "uuid", "docId", "doc_id", "id", "recordId", "taskId"))
     score_raw = _pick(layers, ("score", "relevance", "similarity"))
@@ -341,13 +816,14 @@ def _normalize_reference_item(raw: Any, fallback_index: int) -> dict[str, Any] |
         return None
     return {
         "id": item_id or url,
-        "skillName": "backend-reference",
+        "skillName": str((tool_context or {}).get("skill_name") or (tool_context or {}).get("tool_name") or "backend-reference"),
         "mode": "citation",
         "title": title,
         "type": "result",
         "snippet": content,
         "content": content,
         "source": source,
+        "author": author,
         "time": time,
         "url": url,
         "score": score,

offline-backend-20260512/backend/packages/harness/deerflow/runtime/scheduler/__init__.py

@@ -1,4 +1,10 @@
 from .registry import get_scheduled_task_service, set_scheduled_task_service
-from .service import DuplicateTaskNameError, ScheduledTaskService
+from .service import DuplicateTaskNameError, MarkdownRequirementError, ScheduledTaskService
 
-__all__ = ["DuplicateTaskNameError", "ScheduledTaskService", "get_scheduled_task_service", "set_scheduled_task_service"]
+__all__ = [
+    "DuplicateTaskNameError",
+    "MarkdownRequirementError",
+    "ScheduledTaskService",
+    "get_scheduled_task_service",
+    "set_scheduled_task_service",
+]

offline-backend-20260512/backend/packages/harness/deerflow/runtime/scheduler/service.py

@@ -36,6 +36,27 @@ _RUN_ATTEMPT_TIMEOUT_SECONDS = 30 * 60  # 30 minutes
 # Retries allowed after the first attempt before the run is marked "failed".
 _MAX_RUN_RETRIES = 3
 
+# Minimum length (in characters) the generated Markdown file's content must
+# reach when a task has ``require_markdown`` enabled. The model is NOT told about
+# this threshold — it is enforced silently by the service.
+_MARKDOWN_MIN_CHARS = 100
+
+# Instruction appended to the prompt when ``require_markdown`` is on. It only
+# asks the model to produce a ``.md`` file; the length requirement is enforced
+# in code (not surfaced to the model).
+_MARKDOWN_INSTRUCTION = (
+    "\n\n【输出要求】请将本次任务的最终结果生成为一个 Markdown(.md) 格式的文件，"
+    "保存到 /mnt/user-data/outputs 目录，并使用 present_files 展示该文件。"
+)
+
+
+class MarkdownRequirementError(Exception):
+    """Raised when a ``require_markdown`` task did not produce a valid .md file.
+
+    The run is marked failed (no retry) so the user can see the output did not
+    meet the requirement (a Markdown file longer than the minimum length).
+    """
+
 
 class DuplicateTaskNameError(Exception):
     """A user already owns a scheduled task with the requested name.
@@ -166,6 +187,42 @@ def _looks_like_markdown(content: str) -> bool:
     return any(re.search(pattern, content, flags=re.MULTILINE) for pattern in markdown_patterns)
 
 
+def _task_requires_markdown(task: dict[str, Any]) -> bool:
+    """Whether a task has the ``require_markdown`` flag set.
+
+    The flag lives inside ``execution_context`` (no dedicated column) so the
+    feature works without a schema migration on existing databases.
+    """
+    ctx = task.get("execution_context")
+    return bool(isinstance(ctx, dict) and ctx.get("require_markdown"))
+
+
+def _has_valid_markdown_file(result_files: list[dict[str, Any]] | None) -> bool:
+    """True when the run produced a ``.md`` file whose text exceeds the minimum.
+
+    For tasks with ``require_markdown`` enabled the run only succeeds if at least
+    one generated file is Markdown (``.md``/``.markdown``) and its content is
+    longer than :data:`_MARKDOWN_MIN_CHARS` characters. The file is read from its
+    archived path; binary/undecodable files never qualify.
+    """
+    for item in result_files or []:
+        if not isinstance(item, dict):
+            continue
+        name = str(item.get("name") or "")
+        if not name.lower().endswith((".md", ".markdown")):
+            continue
+        path = item.get("path")
+        if not isinstance(path, str) or not path:
+            continue
+        try:
+            text = Path(path).read_text(encoding="utf-8", errors="ignore")
+        except Exception:
+            continue
+        if len(text.strip()) > _MARKDOWN_MIN_CHARS:
+            return True
+    return False
+
+
 def _build_result(
     task: dict[str, Any],
     *,
@@ -451,6 +508,7 @@ class ScheduledTaskService:
         source_agent_name: str | None = None,
         execution_agent_name: str | None = None,
         execution_context: dict[str, Any] | None = None,
+        require_markdown: bool = False,
     ) -> dict[str, Any]:
         tz = _normalize_timezone(timezone)
         cron_expr = _schedule_to_cron(schedule)
@@ -461,6 +519,7 @@ class ScheduledTaskService:
         context = dict(execution_context or {})
         if execution_agent_name:
             context["agent_name"] = execution_agent_name
+        context["require_markdown"] = bool(require_markdown)
         normalized_description = description.strip() if isinstance(description, str) else None
         task = await self.store.create_task(
             {
@@ -519,6 +578,18 @@ class ScheduledTaskService:
             update_data["next_run_at"] = _next_run_at(cron_expr, tz)
         if data.get("execution_context") is not None:
             update_data["execution_context"] = data["execution_context"]
+        # ``require_markdown`` is persisted inside execution_context (no dedicated
+        # column), so merge it into the context being written — preferring a
+        # context supplied in the same request, else the task's existing one.
+        if data.get("require_markdown") is not None:
+            base_ctx = (
+                update_data["execution_context"]
+                if "execution_context" in update_data
+                else (existing or {}).get("execution_context")
+            )
+            ctx = dict(base_ctx or {})
+            ctx["require_markdown"] = bool(data["require_markdown"])
+            update_data["execution_context"] = ctx
         # Re-enabling a task must also schedule it forward from now, otherwise a
         # paused task with a stale/empty next_run_at would never fire again.
         if update_data.get("enabled") is True and "next_run_at" not in update_data:
@@ -831,6 +902,10 @@ class ScheduledTaskService:
                 run_id=run_id,
                 output_snapshot=output_snapshot,
             )
+        if _task_requires_markdown(task) and not _has_valid_markdown_file(result_files):
+            raise MarkdownRequirementError(
+                f"未满足输出要求：需生成 Markdown(.md) 文件，且文件正文不少于 {_MARKDOWN_MIN_CHARS} 字"
+            )
         return _build_result(task, scheduled_for=scheduled_for, content=result_text, files=result_files)
 
     async def _ensure_thread_meta(self, user_id: str, thread_id: str) -> None:
@@ -895,6 +970,8 @@ class ScheduledTaskService:
                 f"计划时间：{scheduled_for.astimezone(ZoneInfo(task.get('timezone') or 'UTC')).isoformat()}\n\n"
                 f"{task['prompt']}"
             )
+            if execution_context.get("require_markdown"):
+                content += _MARKDOWN_INSTRUCTION
             body = RunCreateRequest(
                 assistant_id="lead_agent",
                 input={"messages": [{"type": "human", "content": content}]},

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

@@ -56,7 +56,7 @@ def notebook_search_tool(query: str, top_k: int = 10) -> str:
             headers={"Authorization": f"Bearer {token}"},
             json={
                 "query": query,
-                "source_ids": [source_ids],
+                "source_ids": source_ids,
                 "search_space_id": search_space_id,
                 "top_k": top_k,
             },