代码修改

This commit is contained in:
董宏凯 committed 2026-07-23 16:00:39 +08:00
1 parent 4948a80c56
commit f49def09ea
3 files changed
+140 -8

No files matched your search

+118 -1
View File
@@ -225,6 +225,22 @@ function getStreamErrorMessage(error: unknown): string {
return "Request failed."; 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( async function fetchThreadTitleRefreshState(
threadId: string, threadId: string,
signal?: AbortSignal, signal?: AbortSignal,
@@ -430,6 +446,12 @@ export function useThreadStream({
// and to allow access to the current thread id in onUpdateEvent // and to allow access to the current thread id in onUpdateEvent
const threadIdRef = useRef<string | null>(threadId ?? null); const threadIdRef = useRef<string | null>(threadId ?? null);
const startedRef = useRef(false); const startedRef = useRef(false);
const lastStreamActivityAtRef = useRef(0);
const streamReconciliationInFlightRef = useRef(false);
const markStreamActivity = useCallback(() => {
lastStreamActivityAtRef.current = Date.now();
}, []);
const listeners = useRef({ const listeners = useRef({
onStart, onStart,
@@ -464,9 +486,10 @@ export function useThreadStream({
const handleStreamStart = useCallback( const handleStreamStart = useCallback(
(_threadId: string) => { (_threadId: string) => {
threadIdRef.current = _threadId; threadIdRef.current = _threadId;
markStreamActivity();
_handleOnStart(_threadId); _handleOnStart(_threadId);
}, },
[_handleOnStart], [_handleOnStart, markStreamActivity],
); );
const queryClient = useQueryClient(); const queryClient = useQueryClient();
@@ -508,6 +531,7 @@ export function useThreadStream({
} }
}, },
onLangChainEvent(event) { onLangChainEvent(event) {
markStreamActivity();
if (event.event === "on_tool_end") { if (event.event === "on_tool_end") {
listeners.current.onToolEnd?.({ listeners.current.onToolEnd?.({
name: event.name, name: event.name,
@@ -538,6 +562,7 @@ export function useThreadStream({
} }
}, },
onUpdateEvent(data) { onUpdateEvent(data) {
markStreamActivity();
const updates: Array<Partial<AgentThreadState> | null> = Object.values( const updates: Array<Partial<AgentThreadState> | null> = Object.values(
data || {}, data || {},
); );
@@ -561,6 +586,7 @@ export function useThreadStream({
} }
}, },
onCustomEvent(event: unknown) { onCustomEvent(event: unknown) {
markStreamActivity();
if ( if (
typeof event === "object" && typeof event === "object" &&
event !== null && event !== null &&
@@ -607,6 +633,7 @@ export function useThreadStream({
} }
}, },
onError(error) { onError(error) {
markStreamActivity();
setOptimisticMessages([]); setOptimisticMessages([]);
const message = getStreamErrorMessage(error); const message = getStreamErrorMessage(error);
// A stale auto-reconnect ("Run <id> not found") after a backend restart: // A stale auto-reconnect ("Run <id> not found") after a backend restart:
@@ -670,6 +697,7 @@ export function useThreadStream({
toast.error(message); toast.error(message);
}, },
onFinish(state) { onFinish(state) {
markStreamActivity();
listeners.current.onFinish?.(state.values); listeners.current.onFinish?.(state.values);
void queryClient.invalidateQueries({ queryKey: ["threads", "search"] }); void queryClient.invalidateQueries({ queryKey: ["threads", "search"] });
}, },
@@ -698,6 +726,95 @@ export function useThreadStream({
const threadStreamRef = useRef(thread); const threadStreamRef = useRef(thread);
threadStreamRef.current = 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 // Reset thread-local pending UI state when the caller actually switches AWAY
// from a thread (threadId becomes nullish). This is the "click 新对话 / open // from a thread (threadId becomes nullish). This is the "click 新对话 / open
// a fresh chat" signal — abort the in-flight SSE and clear optimistic state. // a fresh chat" signal — abort the in-flight SSE and clear optimistic state.
@@ -62,6 +62,8 @@ interface PositionNodeChatProps {
path: string; path: string;
name: string; name: string;
mimeType: string | null; mimeType: string | null;
/** Unique write_file invocation used to distinguish rewrites of one path. */
previewKey?: string;
auto?: boolean; auto?: boolean;
}) => void; }) => void;
onThreadSnapshot?: (snapshot: { onThreadSnapshot?: (snapshot: {
@@ -478,7 +480,9 @@ export function PositionNodeChat({
} }
// One write_file call should open at most one preview. The parent further // 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. // 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; if (autoPreviewedArtifactKeyRef.current === artifactKey) return;
autoPreviewedArtifactKeyRef.current = artifactKey; autoPreviewedArtifactKeyRef.current = artifactKey;
onPreviewArtifact?.({ onPreviewArtifact?.({
@@ -486,6 +490,7 @@ export function PositionNodeChat({
path: streamingArtifact.path, path: streamingArtifact.path,
name: streamingArtifact.name, name: streamingArtifact.name,
mimeType: streamingArtifact.mimeType, mimeType: streamingArtifact.mimeType,
previewKey: streamingArtifact.key,
auto: true, auto: true,
}); });
}, [activeThreadId, onPreviewArtifact, streamingArtifact, thread.isLoading]); }, [activeThreadId, onPreviewArtifact, streamingArtifact, thread.isLoading]);
@@ -131,6 +131,8 @@ type PositionArtifactPreview = {
path: string; path: string;
name: string; name: string;
mimeType: string | null; mimeType: string | null;
/** A write_file call id. It differentiates two generations of one path. */
previewKey?: string;
auto?: boolean; auto?: boolean;
}; };
type ClosureStageKind = PositionBuiltinAgentKind; type ClosureStageKind = PositionBuiltinAgentKind;
@@ -167,7 +169,12 @@ function normalizePreviewArtifactPath(path: string): string {
return normalized.startsWith("/") ? normalized : `/${normalized}`; return normalized.startsWith("/") ? normalized : `/${normalized}`;
} }
function sandboxCloseKey(artifact: Pick<PositionArtifactPreview, "threadId" | "path">): string { function sandboxCloseKey(
artifact: Pick<PositionArtifactPreview, "threadId" | "path" | "previewKey">,
): string {
if (artifact.previewKey) {
return `${artifact.threadId}:write:${artifact.previewKey}`;
}
return `${artifact.threadId}:${normalizePreviewArtifactPath(artifact.path)}`; return `${artifact.threadId}:${normalizePreviewArtifactPath(artifact.path)}`;
} }
@@ -1323,11 +1330,13 @@ export default function PositionRoundtablePage({
} }
closedSandboxKeyRef.current = null; closedSandboxKeyRef.current = null;
setSandboxArtifact((current) => setSandboxArtifact((current) => {
current?.threadId === artifact.threadId && current.path === artifact.path const isSamePreview =
? current current?.threadId === artifact.threadId &&
: artifact, current.path === artifact.path &&
); current.previewKey === artifact.previewKey;
return isSamePreview ? current : artifact;
});
setSandboxThreadSnapshot((current) => setSandboxThreadSnapshot((current) =>
current?.threadId === artifact.threadId ? current : null, current?.threadId === artifact.threadId ? current : null,
); );
@@ -1939,6 +1948,7 @@ export default function PositionRoundtablePage({
<aside className="flex w-[min(48vw,720px)] min-w-[440px] shrink-0 flex-col overflow-hidden rounded-[4px] border border-slate-200 bg-white shadow-sm dark:border-[#2b8ffe]/45 dark:bg-[#032650]/80 dark:shadow-none"> <aside className="flex w-[min(48vw,720px)] min-w-[440px] shrink-0 flex-col overflow-hidden rounded-[4px] border border-slate-200 bg-white shadow-sm dark:border-[#2b8ffe]/45 dark:bg-[#032650]/80 dark:shadow-none">
<ThreadContext.Provider value={sandboxThreadContextValue}> <ThreadContext.Provider value={sandboxThreadContextValue}>
<ArtifactFileDetail <ArtifactFileDetail
key={sandboxCloseKey(sandboxArtifact)}
className="min-h-0 flex-1" className="min-h-0 flex-1"
filepath={sandboxArtifact.path} filepath={sandboxArtifact.path}
threadId={sandboxArtifact.threadId} threadId={sandboxArtifact.threadId}