fix(ai写作): 技能步骤从节点协程 emit，custom 事件真正实时刷给前端

前端/SDK 都是实时的（onCustomEvent 逐条即触发，draft_chunk 同链路就实时）。
问题在后端：技能步骤原来在节点内的嵌套 agent.astream 里直接 emit，custom 事件
被攒到整段跑完才一次性吐到 SSE（后端在跑、前端一直 loading）。

改为：嵌套 astream 跑后台任务，只把工具调用塞进 asyncio.Queue；真正的 writer
由 researcher 节点**自身协程**从队列取出来调（与 draft_chunk 同一上下文），
custom 事件即被 LangGraph 实时刷给前端。撤掉上一版无效的 sleep(0)。

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

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

@@ -175,8 +175,7 @@ async def run_skill_via_real_agent(
     )
 
     # 复用 SubagentExecutor 的 agent/state 构建（模型 + 中间件 + 沙箱 + 技能注入），但**不**走
-    # 它阻塞式的 execute()——而是直接在本图的事件循环里 astream，边跑边把每个工具调用实时推到
-    # 步骤条。内网慢时用户能看到「技能正在调 bash/检索…」的进度，而不是干等一段冻住的空白。
+    # 它阻塞式的 execute()。
     thread_id = (config.get("configurable") or {}).get("thread_id")
     try:
         executor = SubagentExecutor(
@@ -201,42 +200,57 @@ async def run_skill_via_real_agent(
     if emit_query:
         emit_query(f"读取技能「{skill_name}」的 SKILL.md，开始执行检索…")
 
+    # ── 实时步骤的关键 ──
+    # 嵌套 agent.astream 跑在**后台任务**里，只把工具调用塞进队列；真正的 emit_query(writer)
+    # 由本协程（researcher 节点自身的协程）从队列里取出来调——和 draft_chunk 同一上下文。
+    # 这样 custom 事件能被 LangGraph **实时**刷给前端 SSE；若在嵌套 astream 里直接 emit，
+    # 事件会被攒到整段跑完才一次性吐出（后端在跑、前端却一直 loading）。
     final_text = ""
     seen_msg_ids: set[str] = set()
+    step_queue: asyncio.Queue = asyncio.Queue()
 
     async def _drive() -> None:
         nonlocal final_text
-        async for chunk in agent.astream(
-            init_state, config=run_config, context=run_context, stream_mode="values"
-        ):
-            msgs = chunk.get("messages") or []
-            emitted = False
-            for m in msgs:
-                mid = getattr(m, "id", None)
-                if not mid or mid in seen_msg_ids:
-                    continue
-                seen_msg_ids.add(mid)
-                # 每个新的工具调用 → 实时推一条步骤（让用户看到技能在执行，而不是冻住）
-                for tc in (getattr(m, "tool_calls", None) or []):
-                    if not emit_query:
-                        break
-                    tname = (tc.get("name") or "").strip() or "工具"
-                    emit_query(f"技能「{skill_name}」调用 {tname}：{_summarize_tool_args(tc.get('args') or {})}")
-                    emitted = True
-                # 记录最后一条 AIMessage 正文（含最终 JSON 素材）
-                if isinstance(m, AIMessage):
-                    text = _content_to_text(m.content)
-                    if text.strip():
-                        final_text = text
-            # 关键：每个 chunk 处理完主动让出事件循环，促使刚 emit 的 custom 事件**实时**
-            # 刷给前端 SSE；否则嵌套 astream 会把它们攒到整段跑完才一次性吐出（前端看着像卡死）。
-            if emitted:
-                await asyncio.sleep(0)
+        try:
+            async for chunk in agent.astream(
+                init_state, config=run_config, context=run_context, stream_mode="values"
+            ):
+                for m in chunk.get("messages") or []:
+                    mid = getattr(m, "id", None)
+                    if not mid or mid in seen_msg_ids:
+                        continue
+                    seen_msg_ids.add(mid)
+                    for tc in (getattr(m, "tool_calls", None) or []):
+                        tname = (tc.get("name") or "").strip() or "工具"
+                        step_queue.put_nowait(
+                            f"技能「{skill_name}」调用 {tname}：{_summarize_tool_args(tc.get('args') or {})}"
+                        )
+                    if isinstance(m, AIMessage):
+                        text = _content_to_text(m.content)
+                        if text.strip():
+                            final_text = text
+        finally:
+            step_queue.put_nowait(None)  # sentinel：跑完/出错都让 pump 收尾
+
+    task = asyncio.create_task(_drive())
+
+    async def _pump() -> None:
+        while True:
+            item = await step_queue.get()
+            if item is None:
+                break
+            if emit_query:
+                emit_query(item)  # 本协程上下文调用 writer → custom 事件实时刷给前端
 
     try:
-        await asyncio.wait_for(_drive(), timeout=max(timeout_s, _REAL_AGENT_TIMEOUT_SECONDS))
+        await asyncio.wait_for(_pump(), timeout=max(timeout_s, _REAL_AGENT_TIMEOUT_SECONDS))
     except asyncio.TimeoutError:
+        task.cancel()
         return [], f"技能检索超时（>{max(timeout_s, _REAL_AGENT_TIMEOUT_SECONDS)}s）"
+
+    # pump 正常结束（收到 sentinel）后等后台任务收尾，拿 final_text / 抛出的异常
+    try:
+        await task
     except Exception as e:  # noqa: BLE001
         logger.warning("skill real-agent[%s]: astream failed: %s", skill_name, e, exc_info=True)
         return [], f"技能检索失败: {e}"