代码提交

frontend-web/src/core/threads/hooks.ts

@@ -244,6 +244,11 @@ const TERMINAL_RUN_STATUSES = new Set([
 // for a while, then reconcile it from the authoritative run/state endpoints.
 const STREAM_RECONCILIATION_IDLE_MS = 12_000;
 const STREAM_RECONCILIATION_POLL_MS = 5_000;
+// A run can legitimately be quiet while a long tool call is in progress, so
+// reconnect more conservatively than the terminal-status check above.  The
+// reconnection only replaces the local SSE reader; the backend run keeps going.
+const STREAM_RECONNECT_IDLE_MS = 30_000;
+const STREAM_RECONNECT_MAX_ATTEMPTS = 3;
 
 async function fetchThreadTitleRefreshState(
   threadId: string,
@@ -466,9 +471,15 @@ export function useThreadStream({
   const startedRef = useRef(false);
   const lastStreamActivityAtRef = useRef(0);
   const streamReconciliationInFlightRef = useRef(false);
+  const streamReconnectAttemptsRef = useRef(0);
+  const streamReconnectToastShownRef = useRef(false);
+  const streamReconnectFailureToastShownRef = useRef(false);
 
   const markStreamActivity = useCallback(() => {
     lastStreamActivityAtRef.current = Date.now();
+    streamReconnectAttemptsRef.current = 0;
+    streamReconnectToastShownRef.current = false;
+    streamReconnectFailureToastShownRef.current = false;
   }, []);
 
   const listeners = useRef({
@@ -761,13 +772,6 @@ export function useThreadStream({
   threadStreamRef.current = thread;
 
   useEffect(() => {
-    if (!thread.isLoading) {
-      // A normal terminal event reached the browser.  A later run may use the
-      // same thread, so it must be eligible for recovery again.
-      streamReconciliationInFlightRef.current = false;
-      return;
-    }
-
     const activeThreadId = threadIdRef.current ?? onStreamThreadId ?? null;
     if (!activeThreadId || typeof window === "undefined") {
       return;
@@ -776,7 +780,7 @@ export function useThreadStream({
     let disposed = false;
     let refreshTimer: number | null = null;
 
-    const reconcileCompletedRun = async () => {
+    const reconcileRun = async () => {
       if (
         disposed ||
         streamReconciliationInFlightRef.current ||
@@ -793,7 +797,51 @@ export function useThreadStream({
       try {
         const client = getAPIClient(isMock);
         const run = await client.runs.get(activeThreadId, runId);
-        if (disposed || !TERMINAL_RUN_STATUSES.has(run.status)) {
+        if (disposed) {
+          return;
+        }
+
+        if (!TERMINAL_RUN_STATUSES.has(run.status)) {
+          // ``useStream`` retries one failed connection on its own.  If that
+          // reader quietly stalls, however, its internal queue never gets a
+          // terminal error and the page remains frozen despite the backend
+          // continuing the run.  Abort only the local reader, then join the
+          // same run so no work or tokens are lost server-side.
+          if (Date.now() - lastStreamActivityAtRef.current < STREAM_RECONNECT_IDLE_MS) {
+            return;
+          }
+
+          if (streamReconnectAttemptsRef.current >= STREAM_RECONNECT_MAX_ATTEMPTS) {
+            if (!streamReconnectFailureToastShownRef.current) {
+              streamReconnectFailureToastShownRef.current = true;
+              toast.error("回答仍在后台处理中，但页面连接暂未恢复；请稍后返回此会话查看结果。");
+            }
+            return;
+          }
+
+          streamReconnectAttemptsRef.current += 1;
+          // Avoid enqueueing another reconnect before this new reader has had
+          // enough time to receive its first event.  A genuine event resets
+          // the attempt counter through ``markStreamActivity``.
+          lastStreamActivityAtRef.current = Date.now();
+          if (!streamReconnectToastShownRef.current) {
+            streamReconnectToastShownRef.current = true;
+            toast.info("回答连接出现波动，正在自动恢复…");
+          }
+
+          const stream = threadStreamRef.current;
+          void Promise.resolve(stream.stop())
+            .catch(() => {
+              // Stopping a stale local reader is best-effort; joining is still
+              // useful if the SDK already considers the stream stopped.
+            })
+            .finally(() => {
+              if (disposed || threadIdRef.current !== activeThreadId) return;
+              void stream.joinStream(runId).catch(() => {
+                // The status poll will either reconcile a terminal run or
+                // schedule the next bounded reconnect attempt.
+              });
+            });
           return;
         }
 
@@ -801,13 +849,17 @@ export function useThreadStream({
         // receive any more useful events.  Load the checkpoint directly before
         // resetting the SDK stream; this preserves the normal onFinish side
         // effects (e.g. marking a position delivery complete) even when the
-        // browser missed the terminal event.
+        // browser missed the terminal event or already reported a transport
+        // error and dropped ``isLoading``.
         const finalState = await client.threads.getState(activeThreadId);
-        if (disposed || !threadStreamRef.current.isLoading) {
+        if (disposed) {
           return;
         }
 
         storage?.removeItem(`lg:stream:${activeThreadId}`);
+        streamReconnectAttemptsRef.current = 0;
+        streamReconnectToastShownRef.current = false;
+        streamReconnectFailureToastShownRef.current = false;
         if (run.status === "success") {
           listeners.current.onFinish?.(finalState.values as AgentThreadState);
           void queryClient.invalidateQueries({ queryKey: ["threads", "search"] });
@@ -836,7 +888,7 @@ export function useThreadStream({
     };
 
     const interval = window.setInterval(
-      () => void reconcileCompletedRun(),
+      () => void reconcileRun(),
       STREAM_RECONCILIATION_POLL_MS,
     );
 

offline-backend-20260512/.env

@@ -64,3 +64,6 @@ ROUNDTABLE_JOB_GATEWAY=inprocess
 DEEPSEEK_CHAT_API_KEY=sk-686b7bc000be4b7cbed6dccb3897c411
 GLM5_FP8_API_KEY=modalresearch_gWu46FDpVLYZWPVEA3b_2fq8Jw2UJiKMgEm_Rxmn1SA
 SILICONFLOW_API_KEY=sk-ighylafootabtuthwueirmoycccyqtdgbifyhozduvolkdyt
+
+# WeKnora admin credentials (DeerFlow server-side iframe SSO proxy)
+WEKNORA_ADMIN_PASSWORD=1qaz@WSX

offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/worker.py

@@ -15,10 +15,10 @@ internal checkpoint callbacks that are not exposed in the Python public API.
 
 from __future__ import annotations
 
-import asyncio
-import copy
-import inspect
-import logging
+import asyncio
+import copy
+import inspect
+import logging
 from dataclasses import dataclass, field
 from functools import lru_cache
 from typing import TYPE_CHECKING, Any, Literal, cast
@@ -28,31 +28,31 @@ from langgraph.checkpoint.base import empty_checkpoint
 if TYPE_CHECKING:
     from langchain_core.messages import HumanMessage
 
-from deerflow.agents.middlewares.title_middleware import (
-    finalize_thread_title,
-    log_title_debug,
-)
-from deerflow.config.app_config import AppConfig
-from deerflow.runtime.serialization import serialize
-from deerflow.runtime.stream_bridge import StreamBridge
+from deerflow.agents.middlewares.title_middleware import (
+    finalize_thread_title,
+    log_title_debug,
+)
+from deerflow.config.app_config import AppConfig
+from deerflow.runtime.serialization import serialize
+from deerflow.runtime.stream_bridge import StreamBridge
 
 from .manager import RunManager, RunRecord
 from .schemas import RunStatus
 
-logger = logging.getLogger(__name__)
+logger = logging.getLogger(__name__)
 
 # Valid stream_mode values for LangGraph's graph.astream()
-_VALID_LG_MODES = {"values", "updates", "checkpoints", "tasks", "debug", "messages", "custom"}
-
-
-def _log_background_title_finalize_failure(thread_id: str):
-    def _done(task: asyncio.Task[bool]) -> None:
-        try:
-            task.result()
-        except Exception:
-            logger.debug("Failed to finalize background title for thread %s", thread_id, exc_info=True)
-
-    return _done
+_VALID_LG_MODES = {"values", "updates", "checkpoints", "tasks", "debug", "messages", "custom"}
+
+
+def _log_background_title_finalize_failure(thread_id: str):
+    def _done(task: asyncio.Task[bool]) -> None:
+        try:
+            task.result()
+        except Exception:
+            logger.debug("Failed to finalize background title for thread %s", thread_id, exc_info=True)
+
+    return _done
 
 
 def _build_runtime_context(
@@ -364,7 +364,40 @@ async def run_agent(
             else:
                 await run_manager.set_status(run_id, RunStatus.interrupted)
         else:
-            await run_manager.set_status(run_id, RunStatus.success)
+            # LLMErrorHandlingMiddleware intentionally converts an exhausted
+            # provider failure into a user-safe AIMessage instead of raising.
+            # That keeps a malformed upstream response out of the SSE transport,
+            # but previously made this worker record the run as ``success``.
+            # The UI consequently had no error event to show and looked as if
+            # generation had silently stopped halfway through.  Inspect the
+            # terminal checkpoint before declaring success so the persisted run
+            # state and the user-visible SSE outcome stay consistent.
+            terminal_llm_error = await _terminal_llm_error_reason(
+                checkpointer=checkpointer,
+                thread_id=thread_id,
+            )
+            if terminal_llm_error is not None:
+                error_message = "本轮模型生成未能完成，已保留当前对话内容，请稍后重试。"
+                logger.warning(
+                    "Run %s ended with LLM fallback message (reason=%s)",
+                    run_id,
+                    terminal_llm_error,
+                )
+                await run_manager.set_status(
+                    run_id,
+                    RunStatus.error,
+                    error=error_message,
+                )
+                await bridge.publish(
+                    run_id,
+                    "error",
+                    {
+                        "message": error_message,
+                        "name": "ModelGenerationError",
+                    },
+                )
+            else:
+                await run_manager.set_status(run_id, RunStatus.success)
 
     except asyncio.CancelledError:
         action = record.abort_action
@@ -414,47 +447,47 @@ async def run_agent(
             except Exception:
                 logger.warning("Failed to persist run completion for %s (non-fatal)", run_id, exc_info=True)
 
-        # Sync title from checkpoint to threads_meta.display_name
-        if checkpointer is not None and thread_store is not None:
-            try:
-                ckpt_config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
-                ckpt_tuple = await checkpointer.aget_tuple(ckpt_config)
-                if ckpt_tuple is not None:
-                    ckpt = getattr(ckpt_tuple, "checkpoint", {}) or {}
-                    channel_values = ckpt.get("channel_values", {}) or {}
-                    title = channel_values.get("title")
-                    if title:
-                        await thread_store.update_display_name(thread_id, title)
-                    if (
-                        record.status == RunStatus.success
-                        and channel_values.get("title_provisional") is True
-                    ):
-                        log_title_debug(
-                            "Run %s: scheduling background title generation for thread %s",
-                            run_id,
-                            thread_id,
-                        )
-                        title_task = asyncio.create_task(
-                            finalize_thread_title(
-                                checkpointer=checkpointer,
-                                thread_store=thread_store,
-                                thread_id=thread_id,
-                                source_state=copy.deepcopy(channel_values),
-                                app_config=ctx.app_config,
-                            ),
-                            name=f"title-finalize:{thread_id}",
-                        )
-                        title_task.add_done_callback(_log_background_title_finalize_failure(thread_id))
-                    elif record.status == RunStatus.success:
-                        log_title_debug(
-                            "Run %s: background title generation not scheduled for thread %s because title_provisional=%r title=%r",
-                            run_id,
-                            thread_id,
-                            channel_values.get("title_provisional"),
-                            channel_values.get("title"),
-                        )
-            except Exception:
-                logger.debug("Failed to sync title for thread %s (non-fatal)", thread_id)
+        # Sync title from checkpoint to threads_meta.display_name
+        if checkpointer is not None and thread_store is not None:
+            try:
+                ckpt_config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
+                ckpt_tuple = await checkpointer.aget_tuple(ckpt_config)
+                if ckpt_tuple is not None:
+                    ckpt = getattr(ckpt_tuple, "checkpoint", {}) or {}
+                    channel_values = ckpt.get("channel_values", {}) or {}
+                    title = channel_values.get("title")
+                    if title:
+                        await thread_store.update_display_name(thread_id, title)
+                    if (
+                        record.status == RunStatus.success
+                        and channel_values.get("title_provisional") is True
+                    ):
+                        log_title_debug(
+                            "Run %s: scheduling background title generation for thread %s",
+                            run_id,
+                            thread_id,
+                        )
+                        title_task = asyncio.create_task(
+                            finalize_thread_title(
+                                checkpointer=checkpointer,
+                                thread_store=thread_store,
+                                thread_id=thread_id,
+                                source_state=copy.deepcopy(channel_values),
+                                app_config=ctx.app_config,
+                            ),
+                            name=f"title-finalize:{thread_id}",
+                        )
+                        title_task.add_done_callback(_log_background_title_finalize_failure(thread_id))
+                    elif record.status == RunStatus.success:
+                        log_title_debug(
+                            "Run %s: background title generation not scheduled for thread %s because title_provisional=%r title=%r",
+                            run_id,
+                            thread_id,
+                            channel_values.get("title_provisional"),
+                            channel_values.get("title"),
+                        )
+            except Exception:
+                logger.debug("Failed to sync title for thread %s (non-fatal)", thread_id)
 
         # Update threads_meta status based on run outcome
         if thread_store is not None:
@@ -484,6 +517,65 @@ async def _call_checkpointer_method(checkpointer: Any, async_name: str, sync_nam
     return result
 
 
+async def _terminal_llm_error_reason(
+    *,
+    checkpointer: Any | None,
+    thread_id: str,
+) -> str | None:
+    """Return the terminal LLM fallback reason, if this run ended in one.
+
+    ``LLMErrorHandlingMiddleware`` persists its fallback as an AI message with
+    an internal marker.  A normal agent graph treats that message as a valid
+    result and completes without an exception, so the worker must check the
+    saved conversation rather than relying solely on ``astream`` raising.
+    """
+    if checkpointer is None:
+        return None
+
+    try:
+        checkpoint_config = {
+            "configurable": {
+                "thread_id": thread_id,
+                "checkpoint_ns": "",
+            }
+        }
+        checkpoint_tuple = await _call_checkpointer_method(
+            checkpointer,
+            "aget_tuple",
+            "get_tuple",
+            checkpoint_config,
+        )
+        if checkpoint_tuple is None:
+            return None
+        checkpoint = getattr(checkpoint_tuple, "checkpoint", {}) or {}
+        channel_values = checkpoint.get("channel_values", {})
+        messages = channel_values.get("messages", []) if isinstance(channel_values, dict) else []
+        if not isinstance(messages, list):
+            return None
+
+        # Walk backwards to the latest assistant message. Tool/human messages
+        # may be appended around it by middleware, but an older fallback should
+        # never turn a later successful response into an error.
+        for message in reversed(messages):
+            message_type = message.get("type") if isinstance(message, dict) else getattr(message, "type", None)
+            if message_type not in {"ai", "assistant"}:
+                continue
+            from deerflow.agents.middlewares.llm_error_handling_middleware import (
+                is_llm_error_message,
+            )
+
+            return is_llm_error_message(message)
+    except Exception:
+        # A status check must never turn a completed answer into a failed run
+        # merely because a custom checkpointer cannot expose its latest tuple.
+        logger.debug(
+            "Could not inspect terminal LLM state for thread %s",
+            thread_id,
+            exc_info=True,
+        )
+    return None
+
+
 async def _rollback_to_pre_run_checkpoint(
     *,
     checkpointer: Any,

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

@@ -3,7 +3,13 @@ from typing import Literal
 from langchain.tools import tool
 
 
-@tool("ask_clarification", parse_docstring=True, return_direct=True)
+# ClarificationMiddleware owns the stop behaviour: it returns a Command that
+# writes the ToolMessage and explicitly routes to END.  ``return_direct=True``
+# makes the prebuilt tool node short-circuit before that Command update is
+# checkpointed in some LangGraph versions, leaving an assistant tool call
+# without its matching ToolMessage.  The next user reply then fails provider
+# validation with "tool_calls must be followed by tool messages".
+@tool("ask_clarification", parse_docstring=True, return_direct=False)
 def ask_clarification_tool(
     question: str,
     clarification_type: Literal[