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

offline-backend-20260512/backend/CLAUDE.md

@@ -244,6 +244,7 @@ FastAPI application on port 8001 with health check at `GET /health`. Set `GATEWA
 | **Runs** (`/api/runs`) | `POST /stream` - stateless run + SSE; `POST /wait` - stateless run + block; `GET /{rid}/messages` - paginated messages by run_id `{data, has_more}` (cursor: `after_seq`/`before_seq`); `GET /{rid}/feedback` - list feedback by run_id |
 | **Thread Shares** (`/api/thread-shares`) | Per-conversation sharing with two modes (`thread_shares.mode`). **`import`**: `POST /` `{thread_id, mode}` - create (or reuse) a revocable share code; `GET /{code}/preview` + `POST /{code}/import` - copy that conversation into the caller's account (fresh `thread_id` via `app/gateway/thread_copy.py` + new `threads_meta` row + `user-data` copy; dedup on `metadata.shared_origin`). **`view`**: a public, read-only static page — see Public Shares below. Both modes: `GET /` - list my codes; `DELETE /{code}` - revoke. |
 | **Public Shares** (`/api/public/shares`) | **Unauthenticated** (the `/api/public/` prefix is whitelisted in `auth_middleware._PUBLIC_PATH_PREFIXES`). `GET /{code}` - read-only conversation snapshot (title + serialized messages + artifacts) for a `view`-mode share; messages are read straight from the latest checkpoint. `GET /{code}/files/{path}` - serves a file referenced by that conversation (images, Markdown, text, PDF, …) so the shared page's links are openable; path resolution is confined to the owner's thread directory. Active content (`text/html`, `application/xhtml+xml`, `image/svg+xml`) is forced to download (`_public_share_disposition`); everything else is served inline. Missing/revoked/non-`view` codes all return 404. |
+| **Roundtable Drafts** (`/api/roundtable-drafts`) | Per-user persistence for the roundtable-planning page (replaces the old browser `localStorage`). `GET /` - list drafts (lightweight meta); `POST /` - create; `GET/PUT/DELETE /{id}` - full read / partial update / delete. A draft = one full session (Step 1 intent + Step 2 roundtable), step snapshots stored as JSON in `PortableLongText`. Recommendation history bound to a draft: `GET /{id}/recommendations` - list newest-first; `POST /{id}/recommendations` - append one run (objective/status/rationale/picks/candidates). Backed by `deerflow.persistence.roundtable_drafts` (tables `roundtable_drafts`, `roundtable_recommend_history`); store wired in `deps.py` as `app.state.roundtable_draft_store`. |
 | **Scheduled Tasks** (`/api/scheduled-tasks`) | Per-user cron-like agent tasks. `GET/POST /` - list/create owned tasks; `PATCH/DELETE /{id}` - update/delete (owner); `POST /{id}/{pause,resume,trigger}`; `GET /{id}/runs`, `GET /runs/{rid}`, run-file preview/edit/download; `GET /runs/{rid}/messages` - agent messages for the run's underlying agent run (used to inspect a stuck/running run; authorized via `_load_run`); `DELETE /runs/{rid}` - owner-scoped, with an admin fallback so admins can delete any user's run; publish/subscribe under `/{id}/{publish,unpublish,subscribe}` and `GET /published`. **Admin (cross-user):** `GET /admin/all` - every task; `PATCH/DELETE /admin/{id}` - edit/delete any task; `POST /admin/{id}/trigger`; `GET /admin/{id}/runs`. Admin routes require `system_role == "admin"`. |
 
 Proxied through nginx: `/api/langgraph/*` → LangGraph, all other `/api/*` → Gateway.

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

@@ -42,6 +42,7 @@ from app.gateway.routers import (
     public_scheduled_tasks,
     recommend,
     recommended_questions,
+    roundtable_drafts,
     runs,
     scheduled_tasks,
     skills,
@@ -833,6 +834,9 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
 
     app.include_router(user_prompts.router)
 
+    # Roundtable-planning drafts + recommendation history at /api/roundtable-drafts
+    app.include_router(roundtable_drafts.router)
+
     # Suggestions API is mounted at /api/threads/{thread_id}/suggestions
     app.include_router(suggestions.router)
 

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

@@ -152,6 +152,10 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
 
         app.state.user_prompt_store = make_user_prompt_store(sf)
 
+        from deerflow.persistence.roundtable_drafts import make_roundtable_draft_store
+
+        app.state.roundtable_draft_store = make_roundtable_draft_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_drafts.py

@@ -0,0 +1,223 @@
+"""CRUD API for roundtable-planning drafts + recommendation history.
+
+A draft is one full roundtable session (Step 1 intent + Step 2 multi-agent
+roundtable). Bound to the requesting user (not the browser), so it follows the
+account across devices/sessions — replacing the old ``localStorage`` storage.
+
+Recommendation history is bound to a draft: the recommend dialog shows "last
+recommendation for this session" on open, while still letting the user
+re-analyze (which appends a new history row).
+
+Routes (prefix ``/api/roundtable-drafts``):
+    GET    ""                         list current user's drafts (lightweight)
+    POST   ""                         create a draft
+    GET    "/{draft_id}"              full draft (incl. step1 / step2 blobs)
+    PUT    "/{draft_id}"              partial update (title / step / snapshots)
+    DELETE "/{draft_id}"              delete
+    GET    "/{draft_id}/recommendations"   list recommendation history
+    POST   "/{draft_id}/recommendations"   append one recommendation run
+"""
+
+from __future__ import annotations
+
+import logging
+from datetime import datetime
+from typing import Any
+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/roundtable-drafts", tags=["roundtable-drafts"])
+
+
+# ── schemas ────────────────────────────────────────────────────────────────
+
+
+class DraftMetaResponse(BaseModel):
+    id: str
+    title: str = ""
+    furthest_step: int = 1
+    created_at: datetime | str | None = None
+    updated_at: datetime | str | None = None
+
+
+class DraftListResponse(BaseModel):
+    drafts: list[DraftMetaResponse]
+
+
+class DraftResponse(DraftMetaResponse):
+    step1: dict[str, Any] | None = None
+    step2: dict[str, Any] | None = None
+
+
+class DraftCreateRequest(BaseModel):
+    # Optional client-provided id so the browser can keep one stable id across
+    # the create + subsequent updates without an extra round-trip.
+    id: str | None = Field(default=None, max_length=64)
+    title: str = Field(default="", max_length=512)
+    furthest_step: int = Field(default=1, ge=1, le=3)
+    step1: dict[str, Any] | None = None
+    step2: dict[str, Any] | None = None
+
+
+class DraftUpdateRequest(BaseModel):
+    # All optional → callers send only what changed. Auto-save omits ``title``
+    # so it never clobbers a manual rename.
+    title: str | None = Field(default=None, max_length=512)
+    furthest_step: int | None = Field(default=None, ge=1, le=3)
+    step1: dict[str, Any] | None = None
+    step2: dict[str, Any] | None = None
+
+
+class RecommendPickSchema(BaseModel):
+    agent_id: str
+    reason: str = ""
+
+
+class RecommendCandidateSchema(BaseModel):
+    agent_id: str
+    name: str = ""
+    description: str = ""
+
+
+class RecommendHistoryResponse(BaseModel):
+    id: str
+    draft_id: str
+    objective: str = ""
+    status: str = "done"
+    model: str | None = None
+    rationale: str | None = None
+    picks: list[RecommendPickSchema] = []
+    candidates: list[RecommendCandidateSchema] = []
+    created_at: datetime | str | None = None
+
+
+class RecommendHistoryListResponse(BaseModel):
+    history: list[RecommendHistoryResponse]
+
+
+class RecommendHistoryCreateRequest(BaseModel):
+    objective: str = Field(default="", max_length=512)
+    status: str = Field(default="done", max_length=32)
+    model: str | None = None
+    rationale: str | None = None
+    picks: list[RecommendPickSchema] = []
+    candidates: list[RecommendCandidateSchema] = []
+
+
+# ── helpers ──────────────────────────────────────────────────────────────
+
+
+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, "roundtable_draft_store", None)
+    if store is None:
+        raise HTTPException(status_code=503, detail="Roundtable draft store not available")
+    return store
+
+
+# ── draft routes ───────────────────────────────────────────────────────────
+
+
+@router.get("", response_model=DraftListResponse)
+async def list_drafts(request: Request) -> DraftListResponse:
+    store = _get_store(request)
+    user_id = _current_user_id(request)
+    rows = await store.list_drafts(user_id)
+    return DraftListResponse(drafts=[DraftMetaResponse(**r) for r in rows])
+
+
+@router.post("", response_model=DraftResponse, status_code=201)
+async def create_draft(request: Request, body: DraftCreateRequest) -> DraftResponse:
+    store = _get_store(request)
+    user_id = _current_user_id(request)
+    row = await store.create_draft(
+        user_id,
+        {
+            "id": (body.id or uuid4().hex),
+            "title": body.title.strip(),
+            "furthest_step": body.furthest_step,
+            "step1": body.step1,
+            "step2": body.step2,
+        },
+    )
+    return DraftResponse(**row)
+
+
+@router.get("/{draft_id}", response_model=DraftResponse)
+async def get_draft(request: Request, draft_id: str) -> DraftResponse:
+    store = _get_store(request)
+    user_id = _current_user_id(request)
+    row = await store.get_draft(draft_id, user_id)
+    if row is None:
+        raise HTTPException(status_code=404, detail="Draft not found")
+    return DraftResponse(**row)
+
+
+@router.put("/{draft_id}", response_model=DraftResponse)
+async def update_draft(request: Request, draft_id: str, body: DraftUpdateRequest) -> DraftResponse:
+    store = _get_store(request)
+    user_id = _current_user_id(request)
+    # Build kwargs that distinguish "omitted" from "explicitly None". The store
+    # uses a sentinel; here we pass only the fields the client actually sent.
+    fields = body.model_dump(exclude_unset=True)
+    row = await store.update_draft(draft_id, user_id, **fields)
+    if row is None:
+        raise HTTPException(status_code=404, detail="Draft not found")
+    return DraftResponse(**row)
+
+
+@router.delete("/{draft_id}", status_code=204)
+async def delete_draft(request: Request, draft_id: str) -> None:
+    store = _get_store(request)
+    user_id = _current_user_id(request)
+    deleted = await store.delete_draft(draft_id, user_id)
+    if not deleted:
+        raise HTTPException(status_code=404, detail="Draft not found")
+
+
+# ── recommendation history routes ────────────────────────────────────────
+
+
+@router.get("/{draft_id}/recommendations", response_model=RecommendHistoryListResponse)
+async def list_recommendations(request: Request, draft_id: str) -> RecommendHistoryListResponse:
+    store = _get_store(request)
+    user_id = _current_user_id(request)
+    rows = await store.list_recommendations(draft_id, user_id)
+    return RecommendHistoryListResponse(history=[RecommendHistoryResponse(**r) for r in rows])
+
+
+@router.post("/{draft_id}/recommendations", response_model=RecommendHistoryResponse, status_code=201)
+async def add_recommendation(
+    request: Request, draft_id: str, body: RecommendHistoryCreateRequest
+) -> RecommendHistoryResponse:
+    store = _get_store(request)
+    user_id = _current_user_id(request)
+    # Ownership: the draft must exist and belong to the caller.
+    draft = await store.get_draft(draft_id, user_id)
+    if draft is None:
+        raise HTTPException(status_code=404, detail="Draft not found")
+    row = await store.add_recommendation(
+        user_id,
+        draft_id,
+        {
+            "id": uuid4().hex,
+            "objective": body.objective,
+            "status": body.status,
+            "model": body.model,
+            "rationale": body.rationale,
+            "picks": [p.model_dump() for p in body.picks],
+            "candidates": [c.model_dump() for c in body.candidates],
+        },
+    )
+    return RecommendHistoryResponse(**row)

offline-backend-20260512/backend/config.yaml

@@ -14,7 +14,7 @@ models:
     display_name: zai-org/GLM-5-FP8
     use: langchain_openai:ChatOpenAI
     model: zai-org/GLM-5-FP8
-    api_key: Bearer modalresearch_gWu46FDpVLYZWPVEA3b_2fq8Jw2UJiKMgEm_Rxmn1SA
+    api_key: modalresearch_gWu46FDpVLYZWPVEA3b_2fq8Jw2UJiKMgEm_Rxmn1SA
     base_url: https://api.us-west-2.modal.direct/v1
     request_timeout: 600.0
     max_retries: 2
@@ -71,6 +71,16 @@ models:
     temperature: 1.0
     supports_vision: true
     supports_thinking: true
+  - name: glm-5.1
+    display_name: glm-5.1
+    use: langchain_openai:ChatOpenAI
+    model: glm-5.1
+    api_key: 75121e4bed7640d5b71ac95cf8b4310a.RA3GyRBAR3kv7VrR
+    base_url: https://open.bigmodel.cn/api/paas/v4/
+    max_tokens: 4096
+    temperature: 1.0
+    supports_vision: true
+    supports_thinking: true
 
 title:
   enabled: true

offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/_utils.py

@@ -74,6 +74,28 @@ def get_model(config: RunnableConfig, model_name: str | None = None):
     return create_chat_model(name=name)
 
 
+def get_writing_model(config: RunnableConfig, state: dict):
+    """大纲 / 写作 / 审核专用：严格使用用户在表单选定的模型，**不回退默认**。
+
+    用户模型在 intent_parser（首次 invocation，configurable.model_name 可用）阶段
+    捕获进 ``state['model_name']``，经 checkpoint 持久化。这三个节点都在
+    「确认素材 / 大纲」resume 之后才运行，而 resume 是新 invocation、其 config 不再
+    带 model_name —— 所以必须从 state 读，不能从 configurable 读（否则回退到 config
+    列表第一个模型，常是推理模型，既影响质量也会让 reasoning_content 走不出流式）。
+
+    缺失视为配置错误直接抛 ValueError —— 由各节点的 try/except 兜底成 status:error，
+    而不是静默回退默认模型。
+    """
+    name = ((state or {}).get("model_name") or "").strip()
+    if not name:
+        # 兜底再看一眼 configurable（首次同 invocation 内直跑到这里的极端情况）
+        name = (config.get("configurable", {}).get("model_name", "") or "").strip()
+    if not name:
+        raise ValueError("未指定写作模型（model_name 为空）：请在写作表单选择模型后重试")
+    from deerflow.models import create_chat_model
+    return create_chat_model(name=name)
+
+
 # 弱模型常把结构性标点写成中文全角符号，仅在常规解析失败后作为修复手段
 _FULLWIDTH_FIX = {
     "“": '"',   # 左双引号 “
@@ -175,6 +197,21 @@ def parse_json(text: str, fallback: dict) -> dict:
     return fallback
 
 
+def reasoning_delta(chunk) -> str:
+    """提取流式 chunk 的增量「推理内容」。
+
+    推理模型（deepseek-reasoner / GLM 等）会把思考过程放在
+    ``additional_kwargs.reasoning_content`` 增量字段里，而 ``content`` 为空 ——
+    只读 content 的流式循环会把这些 chunk 全部跳过，前端就看不到任何实时输出。
+    各写作节点用本函数把推理增量当作 ``is_thinking=True`` 推给前端的思考框。
+
+    兼容不同 provider 的字段名：``reasoning_content``（主流）/ ``reasoning``。
+    """
+    ak = getattr(chunk, "additional_kwargs", None) or {}
+    val = ak.get("reasoning_content") or ak.get("reasoning") or ""
+    return val if isinstance(val, str) else ""
+
+
 def _content_to_text(content) -> str:
     """将 LLM 响应内容统一为字符串（部分模型返回结构化 content 块列表）。"""
     if isinstance(content, str):

offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/editor.py

@@ -11,7 +11,7 @@ from ..state import AIWritingState, ReviewResult, ReviewIssue
 from ..prompts.editor_prompts import (
     EDITOR_SYSTEM, FACT_REVIEW_USER, LOGIC_REVIEW_USER, LANGUAGE_REVIEW_USER,
 )
-from ._utils import get_ai_writing_config, get_model, parse_json, _content_to_text, ThinkTagParser, safe_get_stream_writer
+from ._utils import get_ai_writing_config, get_writing_model, parse_json, _content_to_text, reasoning_delta, ThinkTagParser, safe_get_stream_writer
 
 logger = logging.getLogger(__name__)
 
@@ -49,6 +49,10 @@ async def _stream_one_review(
         SystemMessage(content=EDITOR_SYSTEM),
         HumanMessage(content=user_prompt),
     ]):
+        # 推理模型的思考走 reasoning_content（content 为空）：作为思考实时推送
+        reasoning = reasoning_delta(chunk)
+        if reasoning:
+            writer({"type": "review_chunk", "category": category, "text": reasoning, "is_thinking": True})
         delta = _content_to_text(chunk.content)
         if not delta:
             continue
@@ -118,7 +122,7 @@ async def editor_node(state: AIWritingState, config: RunnableConfig) -> dict:
     # 模型创建守卫：模型名配错时 create_chat_model 抛 ValueError，
     # 不拦截会硬崩溃整个图；此处转成可恢复的 status:error。
     try:
-        model = get_model(config)
+        model = get_writing_model(config, state)
     except Exception as e:
         logger.error("editor model creation failed: %s", e, exc_info=True)
         return {

offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/intent_parser.py

@@ -14,6 +14,7 @@ from ._utils import (
     _content_to_text,
     get_model,
     parse_json,
+    reasoning_delta,
     safe_get_stream_writer,
 )
 
@@ -215,6 +216,10 @@ async def intent_parser_node(state: AIWritingState, config: RunnableConfig) -> d
         async for chunk in model.astream(
             [SystemMessage(content=system), HumanMessage(content=user)]
         ):
+            # 推理模型的思考走 reasoning_content（content 为空）：作为思考实时推送
+            reasoning = reasoning_delta(chunk)
+            if reasoning:
+                writer({"type": "intent_chunk", "text": reasoning, "is_thinking": True})
             delta = _content_to_text(chunk.content)
             if not delta:
                 continue
@@ -259,6 +264,12 @@ async def intent_parser_node(state: AIWritingState, config: RunnableConfig) -> d
     progress_events.append({"type": "step_done", "agent_name": "系统", "message": "意图解析完成"})
 
     updates: dict = {"status": "researching", "progress_events": progress_events}
+    # 捕获用户表单选定的主模型进 state（仅本节点能可靠拿到 configurable.model_name —
+    # 它运行在首次 invocation；后续大纲/写作/审核在 resume 后跑，config 已不带它）。
+    # 大纲/写作/审核经 get_writing_model 严格读 state['model_name']，不回退默认。
+    selected_model = (configurable.get("model_name") or "").strip()
+    if selected_model:
+        updates["model_name"] = selected_model
     updates["article_type"] = final_article_type
     updates["target_audience"] = final_target_audience
     updates["word_count_target"] = final_word_count

offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/researcher.py

@@ -5,6 +5,7 @@ import asyncio
 import logging
 import os
 import re
+import time
 import uuid
 from typing import List
 
@@ -34,6 +35,11 @@ _MAX_MAX_MATERIALS = 100
 _DEFAULT_KEYWORD_COUNT = 6
 _MAX_KEYWORD_COUNT = 10
 
+# 单个技能的 web 检索超时（秒）。技能检索是顺序循环、最多 5 个技能，
+# DuckDuckGo 等 provider 偶发限速/挂起会把整段"技能检索过程"拖到几十秒看起来卡死；
+# 超时即放弃该技能、emit 0 条结果继续下一个，保证时间线持续推进。
+_SKILL_SEARCH_TIMEOUT_SECONDS = 25
+
 
 def _resolve_keyword_count(raw: object) -> int:
     """把用户传入的检索关键词数量规整到 [1, _MAX_KEYWORD_COUNT]，非法值回退默认 6。"""
@@ -174,11 +180,19 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
     is_user_query_search = action == "re_search" and bool(user_query)
     # 兼容旧版"追加关键词"调用方（如果还有 client 传 extra_keywords 数组而非 user_query）
     is_append_only = action == "re_search" and not user_query and bool(extra_keywords)
-    # PAUSE-1 素材确认卡用户点「启用技能检索」：
+    # 用户在写作表单选择的素材来源：knowledge_base（默认 web/intranet 检索）/ skill（技能检索）。
+    # 首次检索（没有用户干预）时按它决定检索方式；缺省 knowledge_base。
+    material_source = (state.get("material_source") or "knowledge_base").strip()
+    is_first_run = not intervention
+    is_initial_skill_source = is_first_run and material_source == "skill"
+    # 技能检索触发条件（任一即可）：
+    #   - 首次检索且表单选了「技能检索」来源
+    #   - PAUSE-1 素材确认卡用户点「启用技能检索」按钮
+    # 技能检索流程：
     #   - 不再走 web/intranet 检索，改用 SkillSearchProvider
     #   - 顺序调用候选技能 + 早停（第一个技能够就停，不够再加下一个）
     #   - 旧 material_package 完全丢弃（覆盖式替换，让用户对比 web vs skill 结果）
-    is_skill_search = action == "enable_skill_search"
+    is_skill_search = action == "enable_skill_search" or is_initial_skill_source
 
     # 用户指定的检索关键词数量（1~10），决定生成多少个检索词、检索多少个方向
     keyword_count = _resolve_keyword_count(state.get("keyword_count"))
@@ -258,6 +272,10 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
 
     def _emit_skill_step(step: dict) -> None:
         skill_steps.append(step)
+        # 技能检索时机 trace：grep "[skill-timing]" 看每步 emit 的真实时刻，
+        # 与前端 [ai-writing:custom] 日志对齐，钉死是否实时推送。
+        logger.info("[skill-timing] t=%.3f emit skill_search_step type=%s skill=%s",
+                    time.monotonic(), step.get("type"), step.get("skill_name") or "")
         writer({"type": "skill_search_step", "agent_name": "素材收集专家", **step})
 
     if is_skill_search:
@@ -285,15 +303,25 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
 
         for entry in selected:
             sk = entry["skill"]
-            # 用 LLM 把 user_query + skill.description 转成针对性关键词
+            skill_query = f"{user_query} {sk.description or ''}".strip() or intent
+            # invoke_skill 步骤带上实际检索词（截断防过长），让用户在每次长 await 之前
+            # 就看到"这一步在用技能 X 检索什么"，配合前端 spinner 体现进行中。
             _emit_skill_step({
                 "type": "invoke_skill",
-                "message": f"使用技能「{sk.name}」开始检索…",
+                "message": f"使用技能「{sk.name}」检索：{skill_query[:60]}",
                 "skill_name": sk.name,
             })
-            skill_query = f"{user_query} {sk.description or ''}".strip() or intent
             try:
-                res = await web_provider.search(skill_query, max_results=per_skill)
+                # 加超时兜底：单个技能检索最多等 _SKILL_SEARCH_TIMEOUT_SECONDS 秒，
+                # 超时不再阻塞后续技能，避免整段过程看起来"冻结"。
+                res = await asyncio.wait_for(
+                    web_provider.search(skill_query, max_results=per_skill),
+                    timeout=_SKILL_SEARCH_TIMEOUT_SECONDS,
+                )
+            except asyncio.TimeoutError:
+                logger.warning("skill[%s] search timed out after %ss", sk.name, _SKILL_SEARCH_TIMEOUT_SECONDS)
+                search_errors.append(f"skill[{sk.name}]: 检索超时（>{_SKILL_SEARCH_TIMEOUT_SECONDS}s）")
+                res = []
             except Exception as e:
                 logger.warning("skill[%s] search failed: %s", sk.name, e)
                 search_errors.append(f"skill[{sk.name}]: {e}")

offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/writer_draft.py

@@ -12,7 +12,7 @@ from ..prompts.writer_prompts import (
     DRAFT_SECTION_SYSTEM, DRAFT_SECTION_USER, DRAFT_EXPAND_USER,
     WRITER_MODE_LOOSE, WRITER_MODE_STRICT, SECTION_BLOCKED_MARKER,
 )
-from ._utils import get_model, _content_to_text, count_words, ThinkTagParser, safe_get_stream_writer
+from ._utils import get_writing_model, _content_to_text, count_words, reasoning_delta, ThinkTagParser, safe_get_stream_writer
 
 logger = logging.getLogger(__name__)
 
@@ -197,7 +197,7 @@ async def writer_draft_node(state: AIWritingState, config: RunnableConfig) -> di
     # 模型创建守卫：模型名配错时 create_chat_model 抛 ValueError，
     # 不拦截会硬崩溃整个图；此处转成可恢复的 status:error。
     try:
-        model = get_model(config)
+        model = get_writing_model(config, state)
     except Exception as e:
         logger.error("draft model creation failed: %s", e, exc_info=True)
         return {
@@ -250,6 +250,17 @@ async def writer_draft_node(state: AIWritingState, config: RunnableConfig) -> di
                 SystemMessage(content=DRAFT_SECTION_SYSTEM),
                 HumanMessage(content=user_msg),
             ]):
+                # 推理模型的思考走 reasoning_content（content 为空）：作为思考实时推送
+                reasoning = reasoning_delta(chunk)
+                if reasoning:
+                    writer({
+                        "type": "draft_chunk",
+                        "section_index": i,
+                        "section_title": sec_title,
+                        "text": reasoning,
+                        "is_thinking": True,
+                        "is_section_done": False,
+                    })
                 delta = _content_to_text(chunk.content)
                 if not delta:
                     continue

offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/writer_outline.py

@@ -11,7 +11,7 @@ from ..state import AIWritingState, KeyPoint, Outline, OutlineSection
 from ..prompts.writer_prompts import (
     OUTLINE_SYSTEM, OUTLINE_USER, OUTLINE_REVISE_SYSTEM, OUTLINE_REVISE_USER,
 )
-from ._utils import get_model, parse_json, _content_to_text, ThinkTagParser, safe_get_stream_writer
+from ._utils import get_writing_model, parse_json, _content_to_text, reasoning_delta, ThinkTagParser, safe_get_stream_writer
 
 logger = logging.getLogger(__name__)
 
@@ -125,13 +125,17 @@ async def writer_outline_node(state: AIWritingState, config: RunnableConfig) ->
     parser = ThinkTagParser()
 
     try:
-        # get_model 放进 try：模型名配错时 create_chat_model 抛 ValueError，
+        # get_writing_model 放进 try：用户模型缺失 / 模型名配错时抛 ValueError，
         # 留在 try 外会硬崩溃；放进来则统一走下方的 status:error 兜底。
-        model = get_model(config)
+        model = get_writing_model(config, state)
         async for chunk in model.astream([
             SystemMessage(content=system_prompt),
             HumanMessage(content=user_prompt),
         ]):
+            # 推理模型的思考走 reasoning_content（content 为空）：作为思考实时推送
+            reasoning = reasoning_delta(chunk)
+            if reasoning:
+                writer({"type": "outline_chunk", "text": reasoning, "is_thinking": True})
             delta = _content_to_text(chunk.content)
             if not delta:
                 continue

offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/search/skill_search.py

@@ -16,6 +16,7 @@
 from __future__ import annotations
 
 import logging
+import time
 from typing import Callable, List, Optional
 
 from langchain_core.messages import HumanMessage, SystemMessage
@@ -26,6 +27,7 @@ from ..nodes._utils import (
     _content_to_text,
     get_model,
     parse_json,
+    reasoning_delta,
     safe_get_stream_writer,
 )
 
@@ -127,12 +129,26 @@ class SkillPicker:
 
         fallback = {"selected_skills": [s.name for s in skills[: self._max_select]], "reason": ""}
 
+        # list_skills 之后立刻发一个"分析中"提示步骤，避免 LLM 第一个 chunk 到来前
+        # 用户看到空窗期（某些模型 astream 在头部 buffer 时间较长）
+        self._emit_step({
+            "type": "match_skills",
+            "message": "正在让模型从候选技能中筛选最合适的…",
+        })
+
         # 流式 LLM 调用：与 intent_parser / writer_outline 同套机制
         writer = safe_get_stream_writer()
         parser = ThinkTagParser()
         full_text = ""
 
+        _chunk_count = {"n": 0}
+
         def _emit_chunk(text: str, is_thinking: bool) -> None:
+            # 只在首个 chunk 打一条 timing，避免逐 token 刷屏；用来和 skill_search_step
+            # 的 [skill-timing] 对比"选择打字"与"步骤事件"谁先到。
+            if _chunk_count["n"] == 0:
+                logger.info("[skill-timing] t=%.3f first skill_select_chunk", time.monotonic())
+            _chunk_count["n"] += 1
             writer({
                 "type": "skill_select_chunk",
                 "text": text,
@@ -140,6 +156,10 @@ class SkillPicker:
                 "agent_name": "素材收集专家",
             })
 
+        # 立刻推一个空 chunk，让前端 streaming view 立即挂载（占位 + 闪烁光标），
+        # 即使后续 LLM 首个 chunk 到达较慢，UI 也不再静默
+        _emit_chunk("", False)
+
         try:
             model = get_model(self._config, model_name=self._model_name)
             async for chunk in model.astream([
@@ -152,6 +172,10 @@ class SkillPicker:
                     max_select=self._max_select,
                 )),
             ]):
+                # 推理模型的思考走 reasoning_content（content 为空）：作为思考实时推送
+                reasoning = reasoning_delta(chunk)
+                if reasoning:
+                    _emit_chunk(reasoning, True)
                 delta = _content_to_text(chunk.content)
                 if not delta:
                     continue

offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/state.py

@@ -132,9 +132,14 @@ class AIWritingState(TypedDict):
     keyword_count: int
     # 写作模式：strict=严格依据素材 / loose=可结合通用知识（默认 loose）
     writing_mode: str
-    # 笔记来源（可选）：传入后 researcher 阶段会额外从笔记检索素材
-    notebook_search_space_id: Optional[int]
-    notebook_source_ids: Optional[List[int]]
+    # 素材来源：knowledge_base=知识库（默认 web/intranet 检索）/ skill=技能检索。
+    # 用户在写作表单选择，首次检索时由 researcher 节点据此决定检索方式（缺省 knowledge_base）。
+    material_source: str
+    # 用户在写作表单选定的主模型。大纲/写作/审核严格使用它（不回退默认）。
+    # 由 intent_parser 在首次 invocation 从 configurable.model_name 捕获进 state，
+    # 经 checkpoint 持久化 —— interrupt/resume 是新 invocation、其 config 不带原
+    # model_name，存进 state 才能让 resume 后才运行的节点拿到用户模型。
+    model_name: str
 
     # Agent A 输出
     material_package: Optional[MaterialPackage]

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

@@ -22,6 +22,10 @@ from deerflow.persistence.llm_metrics.model import LlmCallMetricRow
 from deerflow.persistence.models.run_event import RunEventRow
 from deerflow.persistence.notifications.model import NotificationRow
 from deerflow.persistence.recommended_questions.model import RecommendedQuestionRow
+from deerflow.persistence.roundtable_drafts.model import (
+    RoundtableDraftRow,
+    RoundtableRecommendHistoryRow,
+)
 from deerflow.persistence.run.model import RunRow
 from deerflow.persistence.scheduled_tasks.model import (
     ScheduledTaskDeliveryProfileRow,
@@ -45,6 +49,8 @@ __all__ = [
     "LlmCallMetricRow",
     "NotificationRow",
     "RecommendedQuestionRow",
+    "RoundtableDraftRow",
+    "RoundtableRecommendHistoryRow",
     "RunEventRow",
     "RunRow",
     "ScheduledTaskDeliveryProfileRow",

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

@@ -0,0 +1,22 @@
+"""Roundtable-planning draft + recommendation-history persistence."""
+
+from __future__ import annotations
+
+from deerflow.persistence.roundtable_drafts.model import (
+    RoundtableDraftRow,
+    RoundtableRecommendHistoryRow,
+)
+from deerflow.persistence.roundtable_drafts.sql import RoundtableDraftRepository
+
+__all__ = [
+    "RoundtableDraftRepository",
+    "RoundtableDraftRow",
+    "RoundtableRecommendHistoryRow",
+    "make_roundtable_draft_store",
+]
+
+
+def make_roundtable_draft_store(session_factory):
+    if session_factory is None:
+        return None
+    return RoundtableDraftRepository(session_factory)

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

@@ -0,0 +1,69 @@
+"""ORM models for roundtable-planning drafts and recommendation history.
+
+A "draft" is one full roundtable-planning session (Step 1 intent + Step 2
+multi-agent roundtable), persisted per user so it follows the account across
+devices/sessions — replacing the old browser-only ``localStorage`` storage.
+
+``step1`` / ``step2`` are stored as JSON serialized into ``PortableLongText``
+(MySQL ``LONGTEXT``) rather than ``PortableJSON`` (MySQL ``TEXT``, 64 KB cap),
+because the Step 2 snapshot can carry a long roundtable transcript plus per-seat
+stub messages that easily exceed 64 KB — mirroring ``ai_writing_sessions``.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+from sqlalchemy import Integer, String
+from sqlalchemy.orm import Mapped, mapped_column
+
+from deerflow.persistence.base import Base
+from deerflow.persistence.types import BeijingDateTime, PortableLongText
+
+
+class RoundtableDraftRow(Base):
+    __tablename__ = "roundtable_drafts"
+
+    id: Mapped[str] = mapped_column(String(64), primary_key=True)
+    user_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    # Auto-derived task summary (objective → first user message → timestamp).
+    title: Mapped[str] = mapped_column(String(512), nullable=False, default="")
+    # Furthest step reached when this snapshot was taken (1 | 2 | 3).
+    furthest_step: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
+    # JSON blobs matching the frontend Step1Snapshot / Step2Snapshot shapes.
+    step1: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)  # JSON object
+    step2: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)  # JSON object
+    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 RoundtableRecommendHistoryRow(Base):
+    """One agent-recommendation run, bound to a draft/session.
+
+    Lets the recommend dialog show "last recommendation for this session" on
+    open, while still letting the user re-analyze (which appends a new row).
+    """
+
+    __tablename__ = "roundtable_recommend_history"
+
+    id: Mapped[str] = mapped_column(String(64), primary_key=True)
+    user_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    draft_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    # Denormalized intent objective at recommend time (list display / context).
+    objective: Mapped[str] = mapped_column(String(512), nullable=False, default="")
+    # How the run ended on the backend: "done" | "fallback" | "asking" | "error".
+    status: Mapped[str] = mapped_column(String(32), nullable=False, default="done")
+    model: Mapped[str | None] = mapped_column(String(128), nullable=True)
+    # Natural-language rationale (may be long → LONGTEXT).
+    rationale: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
+    # JSON: [{"agent_id": str, "reason": str}, ...]
+    picks: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
+    # JSON snapshot of the candidate pool at recommend time:
+    # [{"agent_id": str, "name": str, "description": str}, ...]
+    candidates: Mapped[str | None] = mapped_column(PortableLongText, nullable=True)
+    created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC))

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

@@ -0,0 +1,191 @@
+"""SQLAlchemy-backed roundtable-planning draft + recommendation-history store.
+
+JSON columns (``step1`` / ``step2`` / ``picks`` / ``candidates``) are stored as
+serialized strings in ``PortableLongText`` and round-tripped here with
+``json.dumps`` / ``json.loads`` — same approach as ``ai_writing_sessions`` — so
+large Step 2 transcripts never hit MySQL's 64 KB ``TEXT`` cap.
+
+All write/read paths are scoped by ``user_id`` for ownership isolation.
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import UTC, datetime
+from typing import Any
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from deerflow.persistence.roundtable_drafts.model import (
+    RoundtableDraftRow,
+    RoundtableRecommendHistoryRow,
+)
+
+_UNSET = object()
+
+
+def _loads(value: Any) -> Any:
+    if not isinstance(value, str):
+        return value
+    if not value:
+        return None
+    try:
+        return json.loads(value)
+    except Exception:
+        return None
+
+
+def _dumps(value: Any) -> str | None:
+    if value is None:
+        return None
+    return json.dumps(value, ensure_ascii=False)
+
+
+class RoundtableDraftRepository:
+    def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
+        self._sf = session_factory
+
+    # ── drafts ────────────────────────────────────────────────────────────
+
+    @staticmethod
+    def _draft_to_meta(row: RoundtableDraftRow) -> dict[str, Any]:
+        """Lightweight projection for the history dropdown (no big blobs)."""
+        return {
+            "id": row.id,
+            "title": row.title,
+            "furthest_step": row.furthest_step,
+            "created_at": row.created_at.isoformat() if isinstance(row.created_at, datetime) else row.created_at,
+            "updated_at": row.updated_at.isoformat() if isinstance(row.updated_at, datetime) else row.updated_at,
+        }
+
+    @classmethod
+    def _draft_to_dict(cls, row: RoundtableDraftRow) -> dict[str, Any]:
+        d = cls._draft_to_meta(row)
+        d["step1"] = _loads(row.step1)
+        d["step2"] = _loads(row.step2)
+        return d
+
+    async def list_drafts(self, user_id: str, *, limit: int = 100) -> list[dict[str, Any]]:
+        stmt = (
+            select(RoundtableDraftRow)
+            .where(RoundtableDraftRow.user_id == user_id)
+            .order_by(RoundtableDraftRow.updated_at.desc())
+            .limit(limit)
+        )
+        async with self._sf() as session:
+            result = await session.execute(stmt)
+            return [self._draft_to_meta(row) for row in result.scalars()]
+
+    async def get_draft(self, draft_id: str, user_id: str) -> dict[str, Any] | None:
+        async with self._sf() as session:
+            row = await session.get(RoundtableDraftRow, draft_id)
+            if row is None or row.user_id != user_id:
+                return None
+            return self._draft_to_dict(row)
+
+    async def create_draft(self, user_id: str, data: dict[str, Any]) -> dict[str, Any]:
+        now = datetime.now(UTC)
+        row = RoundtableDraftRow(
+            id=data["id"],
+            user_id=user_id,
+            title=(data.get("title") or "").strip(),
+            furthest_step=int(data.get("furthest_step") or 1),
+            step1=_dumps(data.get("step1")),
+            step2=_dumps(data.get("step2")),
+            created_at=now,
+            updated_at=now,
+        )
+        async with self._sf() as session:
+            session.add(row)
+            await session.commit()
+            await session.refresh(row)
+            return self._draft_to_dict(row)
+
+    async def update_draft(
+        self,
+        draft_id: str,
+        user_id: str,
+        *,
+        title=_UNSET,
+        furthest_step=_UNSET,
+        step1=_UNSET,
+        step2=_UNSET,
+    ) -> dict[str, Any] | None:
+        async with self._sf() as session:
+            row = await session.get(RoundtableDraftRow, draft_id)
+            if row is None or row.user_id != user_id:
+                return None
+            if title is not _UNSET and title is not None:
+                row.title = str(title).strip()
+            if furthest_step is not _UNSET and furthest_step is not None:
+                row.furthest_step = int(furthest_step)
+            if step1 is not _UNSET:
+                row.step1 = _dumps(step1)
+            if step2 is not _UNSET:
+                row.step2 = _dumps(step2)
+            row.updated_at = datetime.now(UTC)
+            await session.commit()
+            await session.refresh(row)
+            return self._draft_to_dict(row)
+
+    async def delete_draft(self, draft_id: str, user_id: str) -> bool:
+        async with self._sf() as session:
+            row = await session.get(RoundtableDraftRow, draft_id)
+            if row is None or row.user_id != user_id:
+                return False
+            await session.delete(row)
+            await session.commit()
+            return True
+
+    # ── recommendation history ────────────────────────────────────────────
+
+    @staticmethod
+    def _history_to_dict(row: RoundtableRecommendHistoryRow) -> dict[str, Any]:
+        return {
+            "id": row.id,
+            "draft_id": row.draft_id,
+            "objective": row.objective,
+            "status": row.status,
+            "model": row.model,
+            "rationale": row.rationale,
+            "picks": _loads(row.picks) or [],
+            "candidates": _loads(row.candidates) or [],
+            "created_at": row.created_at.isoformat() if isinstance(row.created_at, datetime) else row.created_at,
+        }
+
+    async def list_recommendations(
+        self, draft_id: str, user_id: str, *, limit: int = 20
+    ) -> list[dict[str, Any]]:
+        stmt = (
+            select(RoundtableRecommendHistoryRow)
+            .where(
+                RoundtableRecommendHistoryRow.draft_id == draft_id,
+                RoundtableRecommendHistoryRow.user_id == user_id,
+            )
+            .order_by(RoundtableRecommendHistoryRow.created_at.desc())
+            .limit(limit)
+        )
+        async with self._sf() as session:
+            result = await session.execute(stmt)
+            return [self._history_to_dict(row) for row in result.scalars()]
+
+    async def add_recommendation(self, user_id: str, draft_id: str, data: dict[str, Any]) -> dict[str, Any]:
+        now = datetime.now(UTC)
+        row = RoundtableRecommendHistoryRow(
+            id=data["id"],
+            user_id=user_id,
+            draft_id=draft_id,
+            objective=(data.get("objective") or "").strip()[:512],
+            status=(data.get("status") or "done")[:32],
+            model=(data.get("model") or None),
+            rationale=data.get("rationale") or None,
+            picks=_dumps(data.get("picks") or []),
+            candidates=_dumps(data.get("candidates") or []),
+            created_at=now,
+        )
+        async with self._sf() as session:
+            session.add(row)
+            await session.commit()
+            await session.refresh(row)
+            return self._history_to_dict(row)