From f49def09ea0c948c745ac4767838d52d8e81083d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=91=A3=E5=AE=8F=E5=87=AF?= Date: Thu, 23 Jul 2026 16:00:39 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BB=A3=E7=A0=81=E4=BF=AE=E6=94=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- frontend-web/src/core/threads/hooks.ts | 175 +++++++++++++++--- .../components/PositionNodeChat.tsx | 7 +- .../pages/PositionRoundtablePage.tsx | 22 ++- 3 files changed, 168 insertions(+), 36 deletions(-) diff --git a/frontend-web/src/core/threads/hooks.ts b/frontend-web/src/core/threads/hooks.ts index 4b61756f..61f60f3f 100644 --- a/frontend-web/src/core/threads/hooks.ts +++ b/frontend-web/src/core/threads/hooks.ts @@ -225,6 +225,22 @@ function getStreamErrorMessage(error: unknown): string { return "Request failed."; } +const TERMINAL_RUN_STATUSES = new Set([ + "success", + "error", + "timeout", + "interrupted", +]); + +// A completed server run normally sends a terminal SSE event. In local +// development, however, a Vite hot update, a proxy disconnect, or a brief +// network interruption can make the browser miss that final event while the +// backend continues (and successfully completes) the run. Do not probe on +// every render: only treat the stream as suspicious after it has been quiet +// 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; + async function fetchThreadTitleRefreshState( threadId: string, signal?: AbortSignal, @@ -428,8 +444,14 @@ export function useThreadStream({ const [onStreamThreadId, setOnStreamThreadId] = useState(() => threadId); // Ref to track current thread ID across async callbacks without causing re-renders, // and to allow access to the current thread id in onUpdateEvent - const threadIdRef = useRef(threadId ?? null); - const startedRef = useRef(false); + const threadIdRef = useRef(threadId ?? null); + const startedRef = useRef(false); + const lastStreamActivityAtRef = useRef(0); + const streamReconciliationInFlightRef = useRef(false); + + const markStreamActivity = useCallback(() => { + lastStreamActivityAtRef.current = Date.now(); + }, []); const listeners = useRef({ onStart, @@ -461,13 +483,14 @@ export function useThreadStream({ } }, []); - const handleStreamStart = useCallback( - (_threadId: string) => { - threadIdRef.current = _threadId; - _handleOnStart(_threadId); - }, - [_handleOnStart], - ); + const handleStreamStart = useCallback( + (_threadId: string) => { + threadIdRef.current = _threadId; + markStreamActivity(); + _handleOnStart(_threadId); + }, + [_handleOnStart, markStreamActivity], + ); const queryClient = useQueryClient(); const updateSubtask = useUpdateSubtask(); @@ -491,8 +514,8 @@ export function useThreadStream({ ? () => runMetadataStorageRef.current! : false, fetchStateHistory: { limit: 1 }, - onCreated(meta) { - handleStreamStart(meta.thread_id); + onCreated(meta) { + handleStreamStart(meta.thread_id); setOnStreamThreadId(meta.thread_id); const metadataAgentId = context.agent_id ?? context.agent_name; const extraMetadata = @@ -506,9 +529,10 @@ export function useThreadStream({ .threads.update(meta.thread_id, { metadata }) .catch(() => ({})); } - }, - onLangChainEvent(event) { - if (event.event === "on_tool_end") { + }, + onLangChainEvent(event) { + markStreamActivity(); + if (event.event === "on_tool_end") { listeners.current.onToolEnd?.({ name: event.name, data: event.data, @@ -536,9 +560,10 @@ export function useThreadStream({ /* 上报绝不影响问答 */ } } - }, - onUpdateEvent(data) { - const updates: Array | null> = Object.values( + }, + onUpdateEvent(data) { + markStreamActivity(); + const updates: Array | null> = Object.values( data || {}, ); for (const update of updates) { @@ -559,9 +584,10 @@ export function useThreadStream({ ); } } - }, - onCustomEvent(event: unknown) { - if ( + }, + onCustomEvent(event: unknown) { + markStreamActivity(); + if ( typeof event === "object" && event !== null && "type" in event && @@ -605,9 +631,10 @@ export function useThreadStream({ const e = event as { type: "context_compacting"; message: string }; toast(e.message); } - }, - onError(error) { - setOptimisticMessages([]); + }, + onError(error) { + markStreamActivity(); + setOptimisticMessages([]); const message = getStreamErrorMessage(error); // A stale auto-reconnect ("Run not found") after a backend restart: // drop the dirty ``lg:stream:{threadId}`` key so switching back into this @@ -669,10 +696,11 @@ export function useThreadStream({ } toast.error(message); }, - onFinish(state) { - listeners.current.onFinish?.(state.values); - void queryClient.invalidateQueries({ queryKey: ["threads", "search"] }); - }, + onFinish(state) { + markStreamActivity(); + listeners.current.onFinish?.(state.values); + void queryClient.invalidateQueries({ queryKey: ["threads", "search"] }); + }, }); // Optimistic messages shown before the server stream responds @@ -695,8 +723,97 @@ export function useThreadStream({ // Keep a ref to the latest `thread` so the threadId-change effect can stop // an in-flight stream without re-running on every render that produces a // new `thread` identity. - const threadStreamRef = useRef(thread); - threadStreamRef.current = thread; + const threadStreamRef = useRef(thread); + 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; + } + + let disposed = false; + let refreshTimer: number | null = null; + + const reconcileCompletedRun = async () => { + if ( + disposed || + streamReconciliationInFlightRef.current || + Date.now() - lastStreamActivityAtRef.current < STREAM_RECONCILIATION_IDLE_MS + ) { + return; + } + + const storage = runMetadataStorageRef.current; + const runId = storage?.getItem(`lg:stream:${activeThreadId}`); + if (!runId) return; + + streamReconciliationInFlightRef.current = true; + try { + const client = getAPIClient(isMock); + const run = await client.runs.get(activeThreadId, runId); + if (disposed || !TERMINAL_RUN_STATUSES.has(run.status)) { + return; + } + + // The server is already terminal, so the existing SSE listener cannot + // 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. + const finalState = await client.threads.getState(activeThreadId); + if (disposed || !threadStreamRef.current.isLoading) { + return; + } + + storage?.removeItem(`lg:stream:${activeThreadId}`); + if (run.status === "success") { + listeners.current.onFinish?.(finalState.values as AgentThreadState); + void queryClient.invalidateQueries({ queryKey: ["threads", "search"] }); + } else { + toast.error("本轮生成未能完成，已同步当前对话内容，可重新发起请求。"); + } + + // Changing the controlled thread id clears only the local SDK stream + // (it does not cancel the already-terminal backend run). Switching it + // back immediately makes useStream fetch the persisted messages, so + // the page exits the false loading state without a manual refresh. + setOnStreamThreadId(null); + refreshTimer = window.setTimeout(() => { + if (!disposed && threadIdRef.current === activeThreadId) { + setOnStreamThreadId(activeThreadId); + } + }, 0); + } catch { + // A transient status lookup failure must not disrupt a real stream; + // the next polling interval will try again if it is still quiet. + } finally { + if (!disposed) { + streamReconciliationInFlightRef.current = false; + } + } + }; + + const interval = window.setInterval( + () => void reconcileCompletedRun(), + STREAM_RECONCILIATION_POLL_MS, + ); + + return () => { + disposed = true; + window.clearInterval(interval); + if (refreshTimer !== null) { + window.clearTimeout(refreshTimer); + } + }; + }, [isMock, onStreamThreadId, queryClient, thread.isLoading]); // Reset thread-local pending UI state when the caller actually switches AWAY // from a thread (threadId becomes nullish). This is the "click 新对话 / open diff --git a/frontend-web/src/position-roundtable/components/PositionNodeChat.tsx b/frontend-web/src/position-roundtable/components/PositionNodeChat.tsx index ff6f2053..7b50184e 100644 --- a/frontend-web/src/position-roundtable/components/PositionNodeChat.tsx +++ b/frontend-web/src/position-roundtable/components/PositionNodeChat.tsx @@ -62,6 +62,8 @@ interface PositionNodeChatProps { path: string; name: string; mimeType: string | null; + /** Unique write_file invocation used to distinguish rewrites of one path. */ + previewKey?: string; auto?: boolean; }) => void; onThreadSnapshot?: (snapshot: { @@ -478,7 +480,9 @@ export function PositionNodeChat({ } // One write_file call should open at most one preview. The parent further // guards manual close state, because it owns the page-wide singleton sandbox. - const artifactKey = `${activeThreadId}:${streamingArtifact.path}`; + // A report is often rewritten to the same output path. The tool-call key + // (not just the path) makes that a new previewable delivery. + const artifactKey = `${activeThreadId}:${streamingArtifact.key}`; if (autoPreviewedArtifactKeyRef.current === artifactKey) return; autoPreviewedArtifactKeyRef.current = artifactKey; onPreviewArtifact?.({ @@ -486,6 +490,7 @@ export function PositionNodeChat({ path: streamingArtifact.path, name: streamingArtifact.name, mimeType: streamingArtifact.mimeType, + previewKey: streamingArtifact.key, auto: true, }); }, [activeThreadId, onPreviewArtifact, streamingArtifact, thread.isLoading]); diff --git a/frontend-web/src/position-roundtable/pages/PositionRoundtablePage.tsx b/frontend-web/src/position-roundtable/pages/PositionRoundtablePage.tsx index 8c0d68d6..8463aae2 100644 --- a/frontend-web/src/position-roundtable/pages/PositionRoundtablePage.tsx +++ b/frontend-web/src/position-roundtable/pages/PositionRoundtablePage.tsx @@ -131,6 +131,8 @@ type PositionArtifactPreview = { path: string; name: string; mimeType: string | null; + /** A write_file call id. It differentiates two generations of one path. */ + previewKey?: string; auto?: boolean; }; type ClosureStageKind = PositionBuiltinAgentKind; @@ -167,7 +169,12 @@ function normalizePreviewArtifactPath(path: string): string { return normalized.startsWith("/") ? normalized : `/${normalized}`; } -function sandboxCloseKey(artifact: Pick): string { +function sandboxCloseKey( + artifact: Pick, +): string { + if (artifact.previewKey) { + return `${artifact.threadId}:write:${artifact.previewKey}`; + } return `${artifact.threadId}:${normalizePreviewArtifactPath(artifact.path)}`; } @@ -1323,11 +1330,13 @@ export default function PositionRoundtablePage({ } closedSandboxKeyRef.current = null; - setSandboxArtifact((current) => - current?.threadId === artifact.threadId && current.path === artifact.path - ? current - : artifact, - ); + setSandboxArtifact((current) => { + const isSamePreview = + current?.threadId === artifact.threadId && + current.path === artifact.path && + current.previewKey === artifact.previewKey; + return isSamePreview ? current : artifact; + }); setSandboxThreadSnapshot((current) => current?.threadId === artifact.threadId ? current : null, ); @@ -1939,6 +1948,7 @@ export default function PositionRoundtablePage({