fix(ai写作): 技能检索稳定接收数据(复用参考文献提取+截断抢救+上限早停)

- 复用主智能体「参考文献」模式的提取逻辑(deerflow.runtime.references)解析技能JSON，字段并集兼容
- bash输出中间截断时逐条抢救完整记录对象，不再因大JSON解析失败而全丢
- 所有退出路径(超时/递归上限/异常)都从tool输出收割，不依赖模型最终汇总
- 达到max_materials上限即停止驱动模型，修复步骤条爆100+条/素材超额
- 步骤条实时显示每次检索获得多少条数据
- 内网兜底:与技能并行预跑、响应解析兼容两套接口字段、0结果换简单检索词重试、超时放宽到10分钟

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

frontend-web/src/core/auth/task-buttons.ts

@@ -20,7 +20,7 @@ import {
   readCopTaskDetail,
   resolveTaskDeeplinkTarget,
 } from "@/core/auth/task-deeplink";
-import { readUserInfo } from "@/strategy-components/lib/light-app";
+import { readUserInfo, resolveRouteParamValue } from "@/strategy-components/lib/light-app";
 
 /** 「大屏绘制」独立页应用内路由（深链「上一步」默认指向它，跳转追加 ?taskId=）。 */
 export const DASHBOARD_AGENT_ROUTE = "/page/strategy/dashboard-agent";
@@ -106,17 +106,20 @@ export interface TaskButton {
   sortOrder: number;
 }
 
-/** 运行时解析单个登录参数的值（``other`` 取字面值，否则从 ``userInfo`` 取 ``source`` 字段）。 */
+/**
+ * 运行时解析单个登录参数的值——**与「应用管理」完全一致**：直接复用 light-app 的
+ * ``resolveRouteParamValue``（把本按钮的登录参数适配成 ``login`` 类型的路由参数），
+ * 保证取值来源/逻辑跟轻应用一模一样（``other`` 取字面 ``customValue``，否则从
+ * ``userInfo`` 取 ``source`` 字段）。
+ */
 function resolveLoginParamValue(
   p: TaskButtonLoginParam,
   userInfo: Record<string, unknown>,
 ): string {
-  if (p.source === "other") return p.customValue ?? "";
-  if (p.source) {
-    const value = userInfo[p.source];
-    return value == null ? "" : String(value);
-  }
-  return "";
+  return resolveRouteParamValue(
+    { key: p.key, paramType: "login", source: p.source, customValue: p.customValue },
+    userInfo,
+  );
 }
 
 /** 内部切换可选的目标业务（``business`` 类型按钮的下拉项）。 */
@@ -311,13 +314,20 @@ export async function runTaskButton(btn: TaskButton, ctx: RunTaskButtonCtx): Pro
     if (base.startsWith("/") && !base.startsWith("//")) {
       if (btn.openMode === "blank") {
         const full = `${window.location.origin}${window.location.pathname}#${url}`;
+        // 打印实际打开地址，便于排查登录参数/拼接问题。
+        console.info(`[task-button] 「${btn.label}」内部路由(新标签) → ${full}`);
         window.open(full, "_blank", "noopener,noreferrer");
       } else {
+        console.info(`[task-button] 「${btn.label}」内部路由(同窗) → ${url}`);
         ctx.navigate(url);
       }
       return;
     }
     // 外链：self=当前窗口（嵌入态即当前 iframe）；否则 blank=新窗口（默认）。
+    // 打印最终跳转地址（含追加的 id / 登录参数），便于排查问题。
+    console.info(
+      `[task-button] 「${btn.label}」外链(${btn.openMode === "self" ? "当前窗口" : "新窗口"}) → ${url}`,
+    );
     if (btn.openMode === "self") {
       window.location.assign(url);
     } else {

frontend-web/src/roundtable-planning/pages/RoundtablePlanningPage.tsx

@@ -2265,6 +2265,7 @@ export default function RoundtablePlanningPage() {
           <TaskJumpBar
             business={BUSINESS_CODE_TO_TASK_BUSINESS[prefillRef.current.businessCode]}
             taskId={prefillRef.current.taskId}
+            actionId={prefillRef.current.actionId}
             roundtable
           />
         )}

frontend-web/src/strategy-components/components/task-components/TaskJumpBar.tsx

@@ -44,10 +44,19 @@ import { TaskPurposeDialog } from "@/strategy-components/components/task-compone
 export function TaskJumpBar({
   business,
   taskId,
+  actionId,
   roundtable,
 }: {
   business: TaskBusiness;
   taskId: string;
+  /**
+   * xdfx 深链原始「行动id」（URL ?taskId= 传入值，与解析后的任务id 不同）。
+   * 单智能体问答（``TaskFxWorkspace``）由 ``persistTaskContext`` 写进 ``TaskContext``，
+   * 故省略时回退 ``readTaskContext()?.actionId``；多智能体会商（``RoundtablePlanningPage``）
+   * 不写 ``TaskContext``（只在 ``roundtablePrefill`` 里），必须**显式**把 ``prefill.actionId``
+   * 传进来，否则「追加行动id」的按钮会拿不到行动id 而回退任务id。
+   */
+  actionId?: string;
   /** 多智能体会商深链模式：不展示「编辑」与「弹框查看」（purpose）按钮。 */
   roundtable?: boolean;
 }) {
@@ -73,10 +82,13 @@ export function TaskJumpBar({
 
   // 目的树弹框（purpose 类型按钮）由组件层打开。
   const [purposeOpen, setPurposeOpen] = useState(false);
-  // xdfx：深链原始「行动id」（与解析后的任务id 不同才算真行动id），弹框据此换回任务id 再查目的。
-  const ctxActionId = readTaskContext()?.actionId;
+  // xdfx 深链原始「行动id」：会商模式由 prop 显式传入；单智能体回退 TaskContext。
+  const effectiveActionId = actionId ?? readTaskContext()?.actionId;
+  // 与解析后的任务id 不同才算真行动id，弹框据此换回任务id 再查目的。
   const purposeActionId =
-    business === "xdfx" && ctxActionId && ctxActionId !== taskId ? ctxActionId : undefined;
+    business === "xdfx" && effectiveActionId && effectiveActionId !== taskId
+      ? effectiveActionId
+      : undefined;
 
   if (jumpButtons.length === 0) return null;
   // 深链 `hideJump=1`（外部 iframe 套用时）：整条跳转栏隐藏——底部「下一步」等按钮 +
@@ -90,7 +102,7 @@ export function TaskJumpBar({
     }
     void runTaskButton(btn, {
       taskId,
-      actionId: readTaskContext()?.actionId,
+      actionId: effectiveActionId,
       navigate,
       onError: (m) => toast.error(m),
     });

offline-backend-20260512/backend/CLAUDE.md

@@ -274,9 +274,11 @@ Seeding mirrors the roundtable pattern: verified `config.yaml` + `SOUL.md` ship
 
 **Skill-driven retrieval (检索类型).** `material_source` has exactly two values: `general` (通用检索, default) and `notebook` (我的空间); legacy `knowledge_base`/`skill` are normalized to `general` by the researcher node. General retrieval is orchestrated from the **skills configured on the `ai-writing-researcher` agent** (`config.yaml → skills`, editable from the agent management page): the built-in `knowledge-base-search` skill (知识库检索) runs the native keyword fan-out web/intranet search (the old default path), while any other configured skill searches with a `description`-targeted query; skills run in configured order with early-stop at `max_materials`, and an empty/unresolvable skills config falls back to the built-in knowledge search so retrieval always works. The skill itself is seeded from `_ai_writing_seed_assets/skills/knowledge-base-search/SKILL.md` into `skills/public/` (never overwrites), and a pre-existing researcher `config.yaml` lacking the `skills` key is patched in place with the default (explicit `skills: []` is respected). **PAUSE-1 input append flow**: when the user types natural language into the material-confirm input (`re_search`/legacy `enable_skill_search` + `user_query`), the node matches the most relevant configured skill via `SkillPicker(candidates=…)` (skipped when only one skill is configured), searches with the user's words, and **appends** the new materials to the existing package (URL-dedup, original order preserved, no LLM re-rank/truncation, 100-item hard cap). **Skill-failure fault tolerance**: a configured skill that times out / raises / has all its keyword searches fail (vs. legitimately returning 0 results) no longer just silently moves on — the researcher node emits a `skill_search_step` of type `skill_error` (carrying the error `reason`), and if the configured skills collected **no** new materials at all it runs a `fallback` pass over other available skills (`_fallback_candidate_skills`: the built-in `knowledge-base-search` native path first, then other enabled catalog skills not yet tried) instead of erroring out. The frontend `SkillStepsCard` renders `skill_error` (rose) and `fallback` (amber) so the timeline shows "技能报错 → 改用其他技能搜集". Tests: `tests/test_ai_writing_material_source.py`.
 
-**Skill-call timeout** is hardcoded **300s (5 min)** — `researcher.py` `_SKILL_SEARCH_TIMEOUT_SECONDS` (per-skill wrapper) and `_skill_agent.py` `_REAL_AGENT_TIMEOUT_SECONDS` (whole real-agent skill run). The **final intranet ES fallback** (`IntranetSearchProvider`) gets a longer **600s (10 min)** default (`intranet_stub.py`; overridable via `ai_writing.intranet_search_timeout`) — that interface is the slow one, so our own fallback timeout is generous to avoid being blamed for "检不出".
+**Skill-call timeout** is hardcoded **300s (5 min)** — `researcher.py` `_SKILL_SEARCH_TIMEOUT_SECONDS` (per-skill wrapper) and `_skill_agent.py` `_REAL_AGENT_TIMEOUT_SECONDS` (whole real-agent skill run). The intranet ES fallback (`IntranetSearchProvider`) gets a longer **600s (10 min)** default (`intranet_stub.py`; overridable via `ai_writing.intranet_search_timeout`) — that interface is the slow one, so our own fallback timeout is generous to avoid being blamed for "检不出".
 
-**Partial-result harvest + query simplification (检不出/超时的容错).** When `researcher_skill_agent: true` runs a non-native skill via `run_skill_via_real_agent` (the deployment's real path — the skill, e.g. `knowledge-search-v2`, is invoked by the real agent through `bash python3 scripts/search.py "<query>"`), one writing run issues many queries and **some time out** (the skill script prints `{"success":false,"error":"请求超时…","count":0}`) while others succeed (`{"success":true,"count":30,"results":[…]}`). Two fixes so successful data is never dropped: (1) `_drive()` captures every **ToolMessage** stdout and `_harvest_tool_results()` parses the skill's JSON directly (`_try_load_json` tolerates a trailing `Exit Code: 1`; skips `success:false`; field-name union across `knowledge-search-v2`/intranet-ES shapes) — so even if the model's final `{materials:[]}` summary gives up (confused by the timeouts), the successful queries' real results are still collected; the function merges model materials + harvested results (dedup by url/recUuid/title, capped at `max_results`). (2) The model is instructed (in `_MATERIAL_OUTPUT_CONTRACT`) that a 0-result/timeout query must be **retried with a simpler, core query** (drop topic suffixes like 竞选/诉讼/政策, keep the core entity: 「特朗普竞选」→「特朗普」) and that any data already retrieved must always be kept. The same deterministic simplifier `search/base.py::simplify_query` is wired into `IntranetSearchProvider.search`: an empty first attempt retries once with the simplified query. Tests: `tests/test_ai_writing_skill_real_agent.py`, `tests/test_ai_writing_query_simplify.py`.
+**Parallel intranet fallback (技能 0 条 → 直接用并行预跑的内网结果).** When `ai_writing.intranet_search_url` is configured, the researcher node launches the intranet ES search (`IntranetSearchProvider` over the generated keywords) as a **concurrent `asyncio.Task` the moment the skill loop starts** (not sequentially after all skills fail). If the configured + fallback skills collect **nothing**, the node awaits that already-running task and uses its results (`_collect`) — no extra wait. If the skills **did** get materials, the task is cancelled (`_cancel_intranet_task`, suppresses CancelledError) so the intranet call isn't wasted. URL empty → no task, behavior unchanged. Tests: `tests/test_ai_writing_intranet_parallel.py`.
+
+**Partial-result harvest + query simplification (检不出/超时的容错).** When `researcher_skill_agent: true` runs a non-native skill via `run_skill_via_real_agent` (the deployment's real path — the skill, e.g. `knowledge-search-v2`, is invoked by the real agent through `bash python3 scripts/search.py "<query>"`), one writing run issues many queries and **some time out** (the skill script prints `{"success":false,"error":"请求超时…","count":0}`) while others succeed (`{"success":true,"count":30,"results":[…]}`). Two fixes so successful data is never dropped: (1) `_drive()` captures every **ToolMessage** stdout and `_harvest_tool_results()` parses the skill's JSON **by reusing the lead-agent 参考文献 mode's proven extractor** (`deerflow.runtime.references`: `_parse_json_like` + `_first_list` + `_normalize_reference_item`) — the same logic that already extracts skill JSON into citations 100% reliably on ChatPage/AgentChatPage, with its huge field-name union (title/m_title/name, content/page_content/content_preview/snippet/…, recUuid/uuid/docId/…) and per-skill display-config support — plus skips `success:false` outputs — so even if the model's final `{materials:[]}` summary gives up (confused by the timeouts), the successful queries' real results are still collected; the function merges model materials + harvested results (dedup by url/recUuid/title, capped at `max_results`). **The harvest+merge (`_finalize`) runs on EVERY exit path — normal end, overall timeout, AND `await task` exceptions (e.g. `GraphRecursionError`)** — so when the run is cut off before the model writes its summary (observed: 10 queries × ~2 graph steps > the old `recursion_limit`, so the agent errored out with the real 30+ results already in `tool_outputs`), those results are still salvaged instead of returning `[]`. `recursion_limit` was also raised from `_REAL_AGENT_MAX_TURNS` to `_REAL_AGENT_MAX_TURNS*2+4` (it counts graph super-steps ≈ 2 per model→tool round, not model turns) so the model normally has room to finish its summary. **Early-stop on `max_results`:** `_drive` re-harvests after every tool output and, once the deduped harvested count reaches `max_results` (= `per_skill` = `max_materials`), **breaks the nested `astream`** and emits a 「已获取 N 条素材，达到上限，停止本技能检索」 step — without it a weak model keeps issuing new queries forever (observed: 100+ step-bar entries, materials far over the limit). Because the skill now returns exactly `max_materials`, the researcher loop's own `collect_limit` early-stop fires right after, so later configured skills don't run either. (2) The model is instructed (in `_MATERIAL_OUTPUT_CONTRACT`) that a 0-result/timeout query must be **retried with a simpler, core query** (drop topic suffixes like 竞选/诉讼/政策, keep the core entity: 「特朗普竞选」→「特朗普」) and that any data already retrieved must always be kept. The same deterministic simplifier `search/base.py::simplify_query` is wired into `IntranetSearchProvider.search`: an empty first attempt retries once with the simplified query. **(3) Root cause of the instability — bash output middle-truncation.** The skill prints its results as JSON to stdout, and the `bash` tool **middle-truncates** any output over `sandbox.bash_output_max_chars` (default **20000**, inserting a `... [middle truncated …] ...` marker). A `count:30` skill response (~30 records × ~600 chars) exceeds 20000 → the JSON is cut in the middle → **unparseable as a whole** by both the model and `_try_load_json` → all materials lost (while a small `count:9` response parses fine — hence the flapping). Fix: when whole-JSON parse yields no items, `_harvest_tool_results` falls back to `_scan_balanced_json_objects` — a brace-depth/string-aware scanner that recovers **every individually-complete `{…}` record object** from the truncated text (the records on each side of the truncation marker are still valid JSON), so ~16+16 records survive a 20000-char cut, far above `max_materials`. This is why the lead-agent chat pages (`/page/workspace/chats/new`, agent chat) call the same skill with 100% success — they only need the model to summarize prose from whatever it sees, never to parse a structured list, so truncation doesn't hurt them; ai-writing needs the structured records, so it must salvage them itself. Tests: `tests/test_ai_writing_skill_real_agent.py`, `tests/test_ai_writing_query_simplify.py`.
 
 **Agentic skill execution (`ai_writing.researcher_skill_agent`, default false).** By default a configured non-native skill is *not* run agent-style — the node just turns its `description` into a query for the generic search provider. With this flag on, the directed-skill branch instead runs `nodes/_skill_agent.py::run_skill_search_agent`: it injects the skill's **SKILL.md** (read from `skill.skill_file`) alongside the researcher SOUL as the system prompt, binds a `search` tool to the researcher model, and runs a bounded (`_MAX_TURNS=4`) tool-call loop so the model drives the retrieval per the skill's instructions; the `SearchResult`s the tool returns are harvested directly as materials (no fuzzy text extraction). Requires a tool-calling model; if the model can't tool-call (returns no `tool_calls`) or the agent yields nothing, `_execute_skill` falls back to the legacy description→query search so materials are never empty. The built-in `knowledge-base-search` native concurrent path is unchanged. Tests: `tests/test_ai_writing_skill_agent.py`.
 

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

@@ -70,108 +70,152 @@ _MATERIAL_OUTPUT_CONTRACT = (
 )
 
 
-# 技能 stdout 里「结果列表」常见的字段名（不同技能各异，做并集兼容）
-_RESULT_LIST_KEYS = ("results", "data", "items", "materials", "documents", "list", "hits")
-_TITLE_KEYS = ("title", "m_title", "name", "doc_title")
-_CONTENT_KEYS = ("content", "content_preview", "page_content", "text", "summary", "abstract")
-_SOURCE_KEYS = ("source", "source_category", "source1", "source_name", "kb_name")
-_DATE_KEYS = ("publish_time", "published_at", "m_publish", "date", "create_time")
+def _scan_balanced_json_objects(text: str) -> list:
+    """从可能被**中间截断**的文本里扫出所有「括号平衡且完整」的 ``{...}`` 子串（含各嵌套层级）。
 
-
-def _try_load_json(text: str):
-    """尽量从技能 stdout 解析出 JSON：先整体 loads，失败再截取首个平衡的 {…}/[…]。
-
-    技能脚本的输出常带尾巴（如 ``...}\n\nExit Code: 1``），整体 loads 会失败，故再用
-    「首个左括号 → 最后一个对应右括号」截取主体重试。
+    根因：bash 工具对超过 ~20000 字的输出做**中间截断**（插入「... [middle truncated …] ...」），
+    使整段大 JSON（如 count=30 的技能返回）无法解析；但被截断点**两侧的单条结果对象**通常仍是
+    完整合法的 JSON，可逐个抢救出来。带字符串状态机（正确处理引号内的 {}/转义），逐个返回闭合的
+    对象子串，调用方再逐个 json.loads + 字段识别。
     """
     import json as _json
 
-    text = (text or "").strip()
-    if not text:
-        return None
-    try:
-        return _json.loads(text)
-    except Exception:  # noqa: BLE001
-        pass
-    for open_c, close_c in (("{", "}"), ("[", "]")):
-        start = text.find(open_c)
-        end = text.rfind(close_c)
-        if 0 <= start < end:
-            try:
-                return _json.loads(text[start : end + 1])
-            except Exception:  # noqa: BLE001
-                continue
-    return None
+    out: list = []
+    stack: list[int] = []
+    in_str = False
+    esc = False
+    for i, ch in enumerate(text):
+        if in_str:
+            if esc:
+                esc = False
+            elif ch == "\\":
+                esc = True
+            elif ch == '"':
+                in_str = False
+            continue
+        if ch == '"':
+            in_str = True
+        elif ch == "{":
+            stack.append(i)
+        elif ch == "}":
+            if stack:
+                s = stack.pop()
+                frag = text[s : i + 1]
+                try:
+                    parsed = _json.loads(frag)
+                except Exception:  # noqa: BLE001
+                    continue
+                if isinstance(parsed, dict):
+                    out.append(parsed)
+    return out
 
 
-def _extract_result_items(obj) -> list:
-    """从解析出的 JSON 里取「结果数组」：顶层就是数组则直接用，否则按常见字段名找。"""
-    if isinstance(obj, list):
-        return obj
-    if isinstance(obj, dict):
-        for key in _RESULT_LIST_KEYS:
-            v = obj.get(key)
-            if isinstance(v, list):
-                return v
-    return []
+def _describe_tool_output(raw: str, skill_name: str) -> str:
+    """从单次技能输出里提炼一句步骤条文案：**检索词 + 本次命中条数**（用户要求实时显示每次拿到多少条）。
 
+    返回空串 = 这次工具输出不是可识别的检索结果（如 read_file 读 SKILL.md），不发步骤。
+    """
+    import re as _re
 
-def _pick(d: dict, keys: tuple) -> str:
-    for k in keys:
-        v = d.get(k)
-        if isinstance(v, str) and v.strip():
-            return v.strip()
-        if v not in (None, "") and not isinstance(v, (dict, list)):
-            return str(v).strip()
-    return ""
+    from deerflow.runtime import references as _ref
+
+    if not isinstance(raw, str) or not raw.strip():
+        return ""
+    payload = _ref._parse_json_like(raw)
+    query = ""
+    if isinstance(payload, dict):
+        query = str(payload.get("query") or "").strip()
+        # 明确失败（超时/报错）：也给一条「0 条 + 原因」，让用户看到这次没出数据
+        if payload.get("success") is False:
+            err = str(payload.get("error") or "").strip()
+            q = f"「{query}」" if query else ""
+            return f"技能「{skill_name}」检索{q}未获得数据（0 条）{('：' + err) if err else ''}"
+
+    count: int | None = None
+    if isinstance(payload, dict) and isinstance(payload.get("count"), int):
+        count = payload["count"]
+    items = _ref._first_list(payload) if payload is not None else []
+    if count is None and items:
+        count = len(items)
+    if count is None:
+        # 整段没解析出来（最常见：bash 输出被中间截断）→ 从头部正则抢救 query/count，
+        # 再退化为「抢救出的完整记录对象」计数。
+        if not query:
+            m = _re.search(r'"query"\s*:\s*"([^"]{1,80})"', raw)
+            query = m.group(1).strip() if m else ""
+        m2 = _re.search(r'"count"\s*:\s*(\d+)', raw)
+        if m2:
+            count = int(m2.group(1))
+        else:
+            salvaged = _scan_balanced_json_objects(raw)
+            n = sum(1 for o in salvaged if o.get("content") or o.get("content_preview") or o.get("title") or o.get("m_title"))
+            count = n if n > 0 else None
+    if count is None:
+        return ""  # 非检索类输出，不发步骤
+    q = f"「{query}」" if query else ""
+    return f"技能「{skill_name}」检索{q}获得 {count} 条数据"
 
 
 def _harvest_tool_results(tool_outputs: list, skill_name: str, limit: int) -> list:
     """**直接从技能 stdout 收割真实检索结果**（不依赖模型最后的 JSON 汇总）。
 
-    关键容错：一次写作会对同一技能跑多个检索词，部分检索词可能超时（``success:false``），
-    但成功的那些（``success:true``）拿回的真实数据**必须照单全收**，不能因为别的检索词没出
-    数据就整段丢弃。本函数逐条解析技能输出、跳过 ``success:false`` 的、把结果数组映射成
-    SearchResult，按 url/recUuid/title 去重，截断到 limit。字段名做并集兼容（适配 knowledge-
-    search-v2 的 ``{success,query,count,results:[{recUuid,title,url,content_preview,...}]}``
-    及内网 ES 的 ``{m_title,content,recUuid,...}`` 等多种技能输出）。
+    **复用主智能体「参考文献」模式同一套久经考验的提取逻辑**（`deerflow.runtime.references`）：
+    那套逻辑在 ChatPage/AgentChatPage 已 100% 稳定地把技能 JSON 解析成参考文献——它对结果列表
+    字段（results/data/items/documents/...）、单条字段（title/m_title、content/page_content/
+    content_preview/snippet/...、recUuid/uuid/docId/...）都做了超大并集兼容，并尊重每个技能的
+    展示配置。ai 写作直接复用它，确保和参考文献模式「解析能力完全一致」。
+
+    在此基础上再叠加两层本地容错：
+      - 跳过 ``success:false`` 的输出（某检索词超时不影响其它成功的）；
+      - 整段 JSON 解析不出结果时（最常见：bash 输出被**中间截断**），退化为「逐个抢救完整的
+        记录对象」（`_scan_balanced_json_objects`），截断点两侧的单条记录仍能救回。
     """
     from ..search.base import SearchResult
+    from deerflow.runtime import references as _ref
 
+    tool_context = {"skill_name": skill_name, "tool_name": skill_name}
     harvested: list = []
     seen: set[str] = set()
     for raw in tool_outputs:
-        obj = _try_load_json(raw)
-        if obj is None:
+        if not isinstance(raw, str):
             continue
+        payload = _ref._parse_json_like(raw)
         # 明确标记失败的（超时 / 报错）跳过，但不影响其它成功的输出
-        if isinstance(obj, dict) and obj.get("success") is False:
+        if isinstance(payload, dict) and payload.get("success") is False:
             continue
-        for it in _extract_result_items(obj):
+        items = _ref._first_list(payload) if payload is not None else []
+        if not items:
+            # 整段解析不出结果（bash 中间截断把大 JSON 剪断）→ 逐个抢救完整的记录对象
+            items = _scan_balanced_json_objects(raw)
+        for idx, it in enumerate(items):
             if not isinstance(it, dict):
                 continue
-            # 内网 ES 风格：真实字段藏在 metadata 里
-            md = it.get("metadata") if isinstance(it.get("metadata"), dict) else {}
-            merged = {**md, **it} if md else it
-            content = _pick(merged, _CONTENT_KEYS)
-            title = _pick(merged, _TITLE_KEYS)
+            # 用参考文献模式的归一化（含字段并集 + 每技能展示配置），与右侧参考文献完全一致
+            norm = _ref._normalize_reference_item(
+                it, len(harvested) + 1, payload=payload, tool_context=tool_context
+            )
+            if not norm:
+                continue
+            content = str(norm.get("content") or norm.get("snippet") or "").strip()
+            title = str(norm.get("title") or "").strip()
             if not content and not title:
                 continue
-            url = _pick(merged, ("url",))
-            rec = _pick(merged, ("recUuid", "rec_uuid", "id"))
+            raw_rec = norm.get("raw") if isinstance(norm.get("raw"), dict) else {}
+            rec = str(raw_rec.get("recUuid") or raw_rec.get("uuid") or raw_rec.get("docId") or "").strip()
+            url = str(norm.get("url") or "").strip()
             key = url or rec or title
             if key in seen:
                 continue
             seen.add(key)
             try:
-                score = float(merged.get("relevance_score") or 0.0)
+                score = float(norm.get("score"))
             except (TypeError, ValueError):
-                score = 0.0
+                score = 0.75
             harvested.append(SearchResult(
                 title=title or "（无标题）",
                 content=content[:500],
-                source=_pick(merged, _SOURCE_KEYS) or skill_name,
-                published_at=_pick(merged, _DATE_KEYS),
+                source=str(norm.get("source") or skill_name).strip() or skill_name,
+                published_at=str(norm.get("time") or "").strip(),
                 url=url,
                 relevance_score=score or 0.75,
                 source_type="skill",
@@ -304,7 +348,10 @@ async def run_skill_via_real_agent(
         logger.warning("skill real-agent[%s]: init failed: %s", skill_name, e)
         return [], f"运行时初始化失败: {e}"
 
-    run_config: dict = {"recursion_limit": _REAL_AGENT_MAX_TURNS}
+    # recursion_limit 计的是 graph 超级步（每个「模型→工具」回合≈2 步），不能直接等于 max_turns，
+    # 否则模型做了 N 个检索词后还没来得及写最终汇总就被 GraphRecursionError 截断（实测 10 个
+    # 检索词≈20 步 > 16，结果汇总丢失）。给到 ~2×turns + 余量，让模型正常情况下能写完汇总。
+    run_config: dict = {"recursion_limit": _REAL_AGENT_MAX_TURNS * 2 + 4}
     run_context: dict = {}
     if thread_id:
         run_config["configurable"] = {"thread_id": thread_id}
@@ -330,6 +377,7 @@ async def run_skill_via_real_agent(
 
     async def _drive() -> None:
         nonlocal final_text
+        reached_limit = False
         try:
             async for chunk in agent.astream(
                 init_state, config=run_config, context=run_context, stream_mode="values"
@@ -352,6 +400,18 @@ async def run_skill_via_real_agent(
                         out = _content_to_text(m.content)
                         if out.strip():
                             tool_outputs.append(out)
+                            # 实时步骤条：这次检索拿到多少条数据（含失败/截断的诚实计数）
+                            desc = _describe_tool_output(out, skill_name)
+                            if desc:
+                                step_queue.put_nowait(desc)
+                            # **早停**：已收割素材达上限就停止驱动模型——否则弱模型会无限换检索词
+                            # （步骤条爆 100+ 条、素材远超上限）。停掉嵌套 astream 即不再检索。
+                            if max_results and len(_harvest_tool_results(tool_outputs, skill_name, max_results)) >= max_results:
+                                step_queue.put_nowait(f"已获取 {max_results} 条素材，达到上限，停止本技能检索")
+                                reached_limit = True
+                                break
+                if reached_limit:
+                    break
         finally:
             step_queue.put_nowait(None)  # sentinel：跑完/出错都让 pump 收尾
 
@@ -365,10 +425,62 @@ async def run_skill_via_real_agent(
             if emit_query:
                 emit_query(item)  # 本协程上下文调用 writer → custom 事件实时刷给前端
 
+    def _finalize() -> list:
+        """把「模型最终汇总的 materials」与「技能 stdout 直接收割的结果」合并去重。
+
+        **关键**：无论本次运行是正常结束、超时、还是抛异常，只要已经收到过技能的工具输出，
+        就必须把这些真实结果收割下来——绝不能因为模型还没来得及汇总、或整段被切断，就把已经
+        拿到的数据整段丢掉（用户实测：技能明明检到了 30+ 条，却因没汇总而全丢）。
+        """
+        # 1) 模型最终汇总的 materials（它可能已按相关度筛选/去重）
+        model_results: list = []
+        parsed = parse_json(final_text or "", {"materials": []})
+        raw_materials = parsed.get("materials")
+        if isinstance(raw_materials, list):
+            for m in raw_materials:
+                if not isinstance(m, dict):
+                    continue
+                content = str(m.get("content") or "").strip()
+                title = str(m.get("title") or "").strip()
+                if not content and not title:
+                    continue
+                model_results.append(SearchResult(
+                    title=title or "（无标题）",
+                    content=content[:500],
+                    source=str(m.get("source") or skill_name).strip() or skill_name,
+                    published_at=str(m.get("published_at") or "").strip(),
+                    url=str(m.get("url") or "").strip(),
+                    relevance_score=0.8,
+                    source_type="skill",
+                ))
+
+        # 2) **直接从技能 stdout 收割真实检索结果**——哪怕模型最后没汇总、或别的检索词超时，
+        #    成功检索词拿回的数据也照单全收。
+        harvested = _harvest_tool_results(tool_outputs, skill_name, max_results)
+
+        # 3) 合并去重（模型汇总优先、技能原始结果补齐），按 url/recUuid/title 去重，截断到上限。
+        merged: list = []
+        seen_keys: set[str] = set()
+        for r in model_results + harvested:
+            key = (r.url or "").strip() or (r.recUuid or "").strip() or (r.title or "").strip()
+            if key and key in seen_keys:
+                continue
+            if key:
+                seen_keys.add(key)
+            merged.append(r)
+            if len(merged) >= max_results:
+                break
+        return merged
+
     try:
         await asyncio.wait_for(_pump(), timeout=max(timeout_s, _REAL_AGENT_TIMEOUT_SECONDS))
     except asyncio.TimeoutError:
         task.cancel()
+        # 超时也要收割已拿到的技能数据：有就返回（无错误），实在没有才报超时。
+        salvaged = _finalize()
+        if salvaged:
+            logger.info("skill real-agent[%s]: timed out but salvaged %d results", skill_name, len(salvaged))
+            return salvaged, ""
         return [], f"技能检索超时（>{max(timeout_s, _REAL_AGENT_TIMEOUT_SECONDS)}s）"
 
     # pump 正常结束（收到 sentinel）后等后台任务收尾，拿 final_text / 抛出的异常
@@ -376,47 +488,14 @@ async def run_skill_via_real_agent(
         await task
     except Exception as e:  # noqa: BLE001
         logger.warning("skill real-agent[%s]: astream failed: %s", skill_name, e, exc_info=True)
+        # 递归上限 / 中途异常同样要收割已拿到的技能数据，不整段丢。
+        salvaged = _finalize()
+        if salvaged:
+            logger.info("skill real-agent[%s]: errored but salvaged %d results", skill_name, len(salvaged))
+            return salvaged, ""
         return [], f"技能检索失败: {e}"
 
-    # 1) 模型最终汇总的 materials（它可能已按相关度筛选/去重）
-    model_results: list = []
-    parsed = parse_json(final_text or "", {"materials": []})
-    raw_materials = parsed.get("materials")
-    if isinstance(raw_materials, list):
-        for m in raw_materials:
-            if not isinstance(m, dict):
-                continue
-            content = str(m.get("content") or "").strip()
-            title = str(m.get("title") or "").strip()
-            if not content and not title:
-                continue
-            model_results.append(SearchResult(
-                title=title or "（无标题）",
-                content=content[:500],
-                source=str(m.get("source") or skill_name).strip() or skill_name,
-                published_at=str(m.get("published_at") or "").strip(),
-                url=str(m.get("url") or "").strip(),
-                relevance_score=0.8,
-                source_type="skill",
-            ))
-
-    # 2) **直接从技能 stdout 收割真实检索结果**——哪怕模型最后没汇总、或别的检索词超时，
-    #    成功检索词拿回的数据也照单全收（用户诉求：拿到的数据必须接收，不能整段丢）。
-    harvested = _harvest_tool_results(tool_outputs, skill_name, max_results)
-
-    # 3) 合并去重（模型汇总优先、技能原始结果补齐），按 url/recUuid/title 去重，截断到上限。
-    merged: list = []
-    seen_keys: set[str] = set()
-    for r in model_results + harvested:
-        key = (r.url or "").strip() or (r.recUuid or "").strip() or (r.title or "").strip()
-        if key and key in seen_keys:
-            continue
-        if key:
-            seen_keys.add(key)
-        merged.append(r)
-        if len(merged) >= max_results:
-            break
-    return merged, ""
+    return _finalize(), ""
 
 
 def _read_skill_body(skill) -> str:

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

@@ -767,6 +767,60 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
         # 早停上限：通常为 max_materials；用户显式要「再追加 N 条」时放宽到 N，确保凑够
         collect_limit = max(max_materials, requested_append_count or 0)
 
+        # ── 内网检索接口：与技能检索**并行**预跑 ──
+        # 用户诉求：调用技能的同时就把内网检索接口跑起来；技能没拿到结果时，直接用这里
+        # 预跑好的内网结果兜底（省去"技能全失败后再串行等一遍内网"的时间）。需在 config.yaml
+        # 的 ai_writing.intranet_search_url 配地址才启用；技能拿到结果则取消这个任务、不浪费。
+        intranet_url = (ai_writing_config.get("intranet_search_url") or "").strip()
+        intranet_task: "asyncio.Task | None" = None
+        intranet_holder: dict = {}
+
+        async def _run_intranet_search() -> list[SearchResult]:
+            # 全程吞异常：并行内网任务**绝不**能把主流程带挂（实测内网接口可能瞬间失败）。
+            # 失败只记 search_errors + 日志，返回空列表，交由技能结果或空结果文案兜底。
+            try:
+                from ..search.intranet_stub import IntranetSearchProvider
+
+                intranet = IntranetSearchProvider(
+                    base_url=intranet_url,
+                    knowledge_base_name=ai_writing_config.get("intranet_knowledge_base", "samples"),
+                    # 兜底接口超时可在 config.yaml 的 ai_writing.intranet_search_timeout 调大，
+                    # 默认 600s（10 分钟），给内网慢检索留足时间。
+                    timeout=ai_writing_config.get("intranet_search_timeout", 600),
+                )
+                intranet_holder["provider"] = intranet
+                fb_keywords = keywords[:keyword_count] or [intent]
+                gathered = await asyncio.gather(
+                    *[intranet.search(kw, max_results=per_skill) for kw in fb_keywords],
+                    return_exceptions=True,
+                )
+                collected: list[SearchResult] = []
+                for res in gathered:
+                    if isinstance(res, Exception):
+                        logger.warning("[intranet-parallel] one keyword failed: %s", res)
+                        search_errors.append(f"{type(res).__name__}: {res}")
+                        continue
+                    collected.extend(res or [])
+                return collected
+            except asyncio.CancelledError:
+                raise  # 取消要正常传播（技能已拿到结果时主动取消）
+            except Exception as e:  # noqa: BLE001
+                logger.warning("[intranet-parallel] intranet fallback crashed: %s", e, exc_info=True)
+                search_errors.append(f"内网兜底异常：{type(e).__name__}: {e}")
+                return []
+
+        if intranet_url:
+            intranet_task = asyncio.create_task(_run_intranet_search())
+
+        async def _cancel_intranet_task() -> None:
+            """技能已拿到素材：取消并行的内网检索任务，避免无谓占用 / 未取回异常告警。"""
+            if intranet_task is not None and not intranet_task.done():
+                intranet_task.cancel()
+                try:
+                    await intranet_task
+                except BaseException:  # noqa: BLE001 — CancelledError / 任务内异常都吞掉
+                    pass
+
         for sk in ordered:
             added, error_detail = await _execute_skill(sk)
             tried_names.add(sk.name)
@@ -802,45 +856,34 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
                         # 兜底已拿到素材即止，不再无谓尝试后续技能
                         break
 
-        # ── 最终兜底：技能（含兜底技能）都没检索到 → 直连内网知识库接口（ES）兜一把 ──
-        # 用户诉求：技能因网络/配置连不上而 0 条时，至少用内网 ES 接口直检，避免素材全空。
-        # 需在 config.yaml 的 ai_writing 段配 intranet_search_url 才启用；接口契约见
-        # IntranetSearchProvider（POST {query,knowledge_base_name,ts} → [{page_content,metadata}]）。
+        # ── 最终兜底：技能（含兜底技能）都没检索到 → 用并行预跑的内网知识库接口结果兜底 ──
+        # 用户诉求：调用技能时已同时跑了内网检索接口；技能因网络/配置连不上而 0 条时，直接用
+        # 那份并行结果，避免素材全空、也省去串行等待。需在 config.yaml 配 intranet_search_url。
         if not new_results:
-            intranet_url = (ai_writing_config.get("intranet_search_url") or "").strip()
-            if intranet_url:
-                from ..search.intranet_stub import IntranetSearchProvider
-
+            if intranet_task is not None:
                 _emit_skill_step({
                     "type": "fallback",
-                    "message": "技能均未检索到素材，改为直连内网知识库接口兜底检索…",
+                    "message": "技能均未检索到素材，使用并行预跑的内网知识库接口结果兜底…",
                 })
-                intranet = IntranetSearchProvider(
-                    base_url=intranet_url,
-                    knowledge_base_name=ai_writing_config.get("intranet_knowledge_base", "samples"),
-                    # 兜底接口超时可在 config.yaml 的 ai_writing.intranet_search_timeout 调大，
-                    # 默认 600s（10 分钟），给内网慢检索留足时间。
-                    timeout=ai_writing_config.get("intranet_search_timeout", 600),
-                )
-                fb_keywords = keywords[:keyword_count] or [intent]
-                fb_gathered = await asyncio.gather(
-                    *[intranet.search(kw, max_results=per_skill) for kw in fb_keywords],
-                    return_exceptions=True,
-                )
-                fb_added = 0
-                for fb_results in fb_gathered:
-                    if isinstance(fb_results, Exception):
-                        search_errors.append(f"{type(fb_results).__name__}: {fb_results}")
-                        continue
-                    fb_added += _collect(fb_results)
-                search_errors.extend(intranet.errors)
-                # 让下方 notice 拿到内网接口的原始响应体（解析失败时便于定位）
-                provider = intranet
+                try:
+                    fb_results = await intranet_task
+                except BaseException as e:  # noqa: BLE001 — 任务内异常/取消都按"无结果"处理
+                    fb_results = []
+                    search_errors.append(f"{type(e).__name__}: {e}")
+                fb_added = _collect(fb_results)
+                intranet = intranet_holder.get("provider")
+                if intranet is not None:
+                    search_errors.extend(getattr(intranet, "errors", []))
+                    # 让下方 notice 拿到内网接口的原始响应体（解析失败时便于定位）
+                    provider = intranet
                 _emit_skill_step({
                     "type": "skill_result",
                     "message": f"内网知识库接口兜底检索到 {fb_added} 条新素材",
                     "matched_count": fb_added,
                 })
+        else:
+            # 技能已拿到素材：取消并行的内网检索，不再浪费内网接口调用
+            await _cancel_intranet_task()
 
     # 合并去重
     # - 首次检索 / 普通 re_search（无输入，用户只想重出）：旧素材丢弃，覆盖式替换

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

@@ -15,6 +15,17 @@ _DEFAULT_KB = "samples"
 _RAW_SNIPPET_LIMIT = 2000
 
 
+def _pick(d: dict, keys: tuple) -> str:
+    """从结果项里按候选字段名取第一个非空值（两套接口字段名不同，做并集兼容）。"""
+    for k in keys:
+        v = d.get(k)
+        if isinstance(v, str) and v.strip():
+            return v.strip()
+        if v not in (None, "") and not isinstance(v, (dict, list)):
+            return str(v).strip()
+    return ""
+
+
 class IntranetSearchProvider(BaseSearchProvider):
     """内网知识库搜索接口适配器。
 
@@ -92,11 +103,17 @@ class IntranetSearchProvider(BaseSearchProvider):
             self._record_raw(raw_text)
             return []
 
-        # 正常情况下接口直接返回数组；兼容个别情况下包了一层 data/results 的结构
+        # 正常情况下接口直接返回数组；兼容包了一层 data/results 的结构（含工作正常的
+        # knowledge-search-v2：{"success":true,"results":[...]}）。
         if isinstance(data, list):
             items = data
         elif isinstance(data, dict):
-            items = data.get("data") or data.get("results") or []
+            # 接口显式标失败时把整体回传前端便于定位
+            if data.get("success") is False:
+                self.errors.append(f"接口返回 success=false：{str(data.get('error') or '')[:200]}")
+                self._record_raw(raw_text)
+                return []
+            items = data.get("data") or data.get("results") or data.get("documents") or []
         else:
             # 返回结构既不是数组也不是已知 dict 包装：视为异常，把原始响应回传前端
             logger.warning("intranet search unexpected response type for query=%r: %s", query, type(data).__name__)
@@ -106,18 +123,32 @@ class IntranetSearchProvider(BaseSearchProvider):
 
         results: List[SearchResult] = []
         for i, item in enumerate(items[:max_results]):
-            md = (item or {}).get("metadata") or {}
-            content = md.get("content") or (item or {}).get("page_content") or ""
+            if not isinstance(item, dict):
+                continue
+            # 字段名做并集兼容两套接口：
+            #   老 ES：     {page_content, metadata:{m_title, content, recUuid, m_publish, source1}}
+            #   knowledge-search-v2：{title, content_preview, url, recUuid, source_category, publish_time, relevance_score}
+            md = item.get("metadata") if isinstance(item.get("metadata"), dict) else {}
+            merged = {**md, **item} if md else item
+            content = _pick(merged, ("content", "content_preview", "page_content", "text", "summary"))
+            title = _pick(merged, ("title", "m_title", "name", "doc_title"))
+            if not content and not title:
+                continue
+            try:
+                score = float(merged.get("relevance_score"))
+            except (TypeError, ValueError):
+                # 接口未返回相关度分值：按返回顺序赋递减分以保留排序
+                score = max(0.5, 0.95 - i * 0.03)
             results.append(SearchResult(
-                title=str(md.get("m_title") or "").strip() or "（无标题）",
+                title=title or "（无标题）",
                 content=str(content)[:500],
-                source=str(md.get("source1") or "").strip() or "内网知识库",
-                published_at=str(md.get("m_publish") or "").strip(),
-                url="",  # 内网素材的原文地址由前端依据 recUuid 拼接
-                # 接口未返回相关度分值，按 ES 返回顺序赋递减分以保留排序
-                relevance_score=max(0.5, 0.95 - i * 0.03),
+                source=_pick(merged, ("source_category", "source1", "source", "source_name", "kb_name")) or "内网知识库",
+                published_at=_pick(merged, ("publish_time", "m_publish", "published_at", "date")),
+                # 工作接口直接给了 url；老 ES 没有 url（由前端依据 recUuid 拼接），保持空串
+                url=_pick(merged, ("url",)),
+                relevance_score=score,
                 source_type="intranet",
-                recUuid=str(md.get("recUuid") or "").strip(),
+                recUuid=_pick(merged, ("recUuid", "rec_uuid", "id")),
             ))
         return results