diff --git a/frontend-web/src/core/threads/hooks.ts b/frontend-web/src/core/threads/hooks.ts index 5da690c8..9e616af9 100644 --- a/frontend-web/src/core/threads/hooks.ts +++ b/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, ); diff --git a/offline-backend-20260512/.env b/offline-backend-20260512/.env index c973e72f..66777cd7 100644 --- a/offline-backend-20260512/.env +++ b/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 diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/worker.py b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/worker.py index 01e205b1..5e36391b 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/worker.py +++ b/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, diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/tools/builtins/clarification_tool.py b/offline-backend-20260512/backend/packages/harness/deerflow/tools/builtins/clarification_tool.py index 7420260a..652a94ac 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/tools/builtins/clarification_tool.py +++ b/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[