董宏凯andCursor
b81d200eca
澄清中间件改为 ToolMessage + after_model 停轮，前端提交前取消残留 run；系统/向导 thread_type 从对话列表排除；岗位总结席位加 write_file 上限。 Co-authored-by: Cursor <cursoragent@cursor.com>
1889 lines
65 KiB
TypeScript
1889 lines
65 KiB
TypeScript
import type { AIMessage, Message } from "@langchain/langgraph-sdk";
|
||
import type { ThreadsClient } from "@langchain/langgraph-sdk/client";
|
||
import { useStream } from "@langchain/langgraph-sdk/react";
|
||
import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query";
|
||
import type { QueryClient } from "@tanstack/react-query";
|
||
import { useCallback, useEffect, useRef, useState } from "react";
|
||
import { useNavigate } from "react-router-dom";
|
||
import { toast } from "sonner";
|
||
|
||
import type { PromptInputMessage } from "@/components/ai-elements/prompt-input";
|
||
|
||
import { getAPIClient } from "../api";
|
||
import { apiFetch } from "../api";
|
||
import { getStoredAuth } from "../auth";
|
||
import { rememberAskTime } from "../chat-timestamps";
|
||
import { getBackendBaseURL, getLangGraphBaseURL } from "../config";
|
||
import { useI18n } from "../i18n/hooks";
|
||
import { OptType, writeYLog } from "../log/y-log";
|
||
import type { FileInMessage } from "../messages/utils";
|
||
import type { LocalSettings } from "../settings";
|
||
import { isThreadRagModeLocked, lockThreadRagMode, useLocalSettings } from "../settings";
|
||
import { useUpdateSubtask } from "../tasks/context";
|
||
import type { UploadedFileInfo } from "../uploads";
|
||
import { promptInputFilePartToFile, uploadFiles } from "../uploads";
|
||
|
||
import type { AgentThread, AgentThreadState } from "./types";
|
||
import {
|
||
isAbortLikeError,
|
||
isTransientStreamTransportError,
|
||
} from "./thread-access-error";
|
||
import { isHiddenFromRecentChats, shouldHideSystemThreadsFromList } from "./utils";
|
||
|
||
export type ToolEndEvent = {
|
||
name: string;
|
||
data: unknown;
|
||
toolCallId?: string;
|
||
};
|
||
|
||
export type ToolStartEvent = ToolEndEvent;
|
||
|
||
export type ThreadStreamOptions = {
|
||
threadId?: string | null | undefined;
|
||
context: LocalSettings["context"];
|
||
isMock?: boolean;
|
||
assistantId?: string;
|
||
/**
|
||
* Per-page run behaviour. Most chats keep the SDK defaults; long-running
|
||
* workspace flows can opt into a resumable stream without changing global
|
||
* chat semantics.
|
||
*/
|
||
runOptions?: {
|
||
onDisconnect?: "cancel" | "continue";
|
||
multitaskStrategy?: "reject" | "rollback" | "interrupt" | "enqueue";
|
||
};
|
||
/** 额外写到新建 thread 的 metadata（与 agent_id 合并）。
|
||
* 任务深链用 `{ taskId }` 把对话挂到任务下（1:N），列表按 metadata.taskId 过滤。 */
|
||
threadMetadata?: Record<string, unknown>;
|
||
onStart?: (threadId: string) => void;
|
||
onFinish?: (state: AgentThreadState) => void;
|
||
onToolStart?: (event: ToolStartEvent) => void;
|
||
onToolEnd?: (event: ToolEndEvent) => void;
|
||
};
|
||
|
||
type SendMessageOptions = {
|
||
additionalKwargs?: Record<string, unknown>;
|
||
/**
|
||
* Extra human-context messages submitted before the visible user message.
|
||
* They are persisted with `hide_from_ui` so the model receives rich business
|
||
* context without dumping long prompts into the conversation surface.
|
||
*/
|
||
hiddenContext?:
|
||
| string
|
||
| {
|
||
text: string;
|
||
name?: string;
|
||
additionalKwargs?: Record<string, unknown>;
|
||
}
|
||
| Array<{
|
||
text: string;
|
||
name?: string;
|
||
additionalKwargs?: Record<string, unknown>;
|
||
}>;
|
||
};
|
||
|
||
type ThreadTitleRefreshState = Pick<AgentThreadState, "title"> & {
|
||
title_provisional?: boolean;
|
||
};
|
||
|
||
function normalizeHiddenContextMessages(options?: SendMessageOptions) {
|
||
const value = options?.hiddenContext;
|
||
if (!value) return [];
|
||
const items =
|
||
typeof value === "string"
|
||
? [{ text: value }]
|
||
: Array.isArray(value)
|
||
? value
|
||
: [value];
|
||
return items
|
||
.map((item) => ({
|
||
...item,
|
||
text: item.text.trim(),
|
||
}))
|
||
.filter((item) => item.text.length > 0);
|
||
}
|
||
|
||
function normalizeStoredRunId(runId: string | null): string | null {
|
||
if (!runId) {
|
||
return null;
|
||
}
|
||
|
||
const trimmed = runId.trim();
|
||
if (!trimmed) {
|
||
return null;
|
||
}
|
||
|
||
const queryIndex = trimmed.indexOf("?");
|
||
if (queryIndex >= 0) {
|
||
const params = new URLSearchParams(trimmed.slice(queryIndex + 1));
|
||
const queryRunId = params.get("run_id")?.trim();
|
||
if (queryRunId) {
|
||
return queryRunId;
|
||
}
|
||
}
|
||
|
||
const pathWithoutQueryOrHash = trimmed.split(/[?#]/, 1)[0]?.trim() ?? "";
|
||
if (!pathWithoutQueryOrHash) {
|
||
return null;
|
||
}
|
||
|
||
const runsMarker = "/runs/";
|
||
const runsIndex = pathWithoutQueryOrHash.lastIndexOf(runsMarker);
|
||
if (runsIndex >= 0) {
|
||
const runIdAfterMarker = pathWithoutQueryOrHash
|
||
.slice(runsIndex + runsMarker.length)
|
||
.split("/", 1)[0]
|
||
?.trim();
|
||
if (runIdAfterMarker) {
|
||
return runIdAfterMarker;
|
||
}
|
||
return null;
|
||
}
|
||
|
||
const segments = pathWithoutQueryOrHash
|
||
.split("/")
|
||
.map((segment) => segment.trim())
|
||
.filter(Boolean);
|
||
return segments.at(-1) ?? null;
|
||
}
|
||
|
||
function getRunMetadataStorage(): {
|
||
getItem(key: `lg:stream:${string}`): string | null;
|
||
setItem(key: `lg:stream:${string}`, value: string): void;
|
||
removeItem(key: `lg:stream:${string}`): void;
|
||
} {
|
||
return {
|
||
getItem(key) {
|
||
const normalized = normalizeStoredRunId(
|
||
window.sessionStorage.getItem(key),
|
||
);
|
||
if (normalized) {
|
||
window.sessionStorage.setItem(key, normalized);
|
||
return normalized;
|
||
}
|
||
window.sessionStorage.removeItem(key);
|
||
return null;
|
||
},
|
||
setItem(key, value) {
|
||
const normalized = normalizeStoredRunId(value);
|
||
if (normalized) {
|
||
window.sessionStorage.setItem(key, normalized);
|
||
return;
|
||
}
|
||
window.sessionStorage.removeItem(key);
|
||
},
|
||
removeItem(key) {
|
||
window.sessionStorage.removeItem(key);
|
||
},
|
||
};
|
||
}
|
||
|
||
/**
|
||
* 显式停止某会话「服务端正在跑」的 run。
|
||
*
|
||
* 注意：useStream / `thread.stop()` 只中断本地 SSE 监听，服务端 run 仍会继续跑。
|
||
* 任务流程要求「开启新对话时老对话要停止」，所以这里读出 SDK 存在 sessionStorage
|
||
* （``lg:stream:{threadId}``）里的 run id，调用 ``runs.cancel`` 真正取消后端 run，
|
||
* 再清掉本地 reconnect 记录。已结束 / 不存在的 run 取消失败时静默忽略。
|
||
*/
|
||
export async function stopThreadRun(
|
||
threadId: string | null | undefined,
|
||
): Promise<void> {
|
||
if (!threadId || typeof window === "undefined") return;
|
||
let runId: string | null = null;
|
||
try {
|
||
runId = normalizeStoredRunId(
|
||
window.sessionStorage.getItem(`lg:stream:${threadId}`),
|
||
);
|
||
} catch {
|
||
runId = null;
|
||
}
|
||
if (!runId) return;
|
||
try {
|
||
await getAPIClient().runs.cancel(threadId, runId);
|
||
} catch {
|
||
/* 已结束 / 不存在 / 网络错误等忽略 */
|
||
}
|
||
try {
|
||
window.sessionStorage.removeItem(`lg:stream:${threadId}`);
|
||
} catch {
|
||
/* noop */
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 协助卡提交前：若上一轮还在流式（弱模型出卡后偶发未停干净），先取消服务端 run
|
||
* 并断开本地 SSE，避免 409「已有活跃 run」或用户点了选项却发不出去。
|
||
*/
|
||
export async function stopActiveThreadBeforeClarificationReply(
|
||
threadId: string | null | undefined,
|
||
thread: { isLoading?: boolean; stop: () => void | Promise<void> },
|
||
): Promise<void> {
|
||
let storedRunId: string | null = null;
|
||
if (threadId && typeof window !== "undefined") {
|
||
try {
|
||
storedRunId = normalizeStoredRunId(
|
||
window.sessionStorage.getItem(`lg:stream:${threadId}`),
|
||
);
|
||
} catch {
|
||
storedRunId = null;
|
||
}
|
||
}
|
||
if (!thread.isLoading && !storedRunId) return;
|
||
try {
|
||
await stopThreadRun(threadId);
|
||
} catch {
|
||
/* best-effort */
|
||
}
|
||
try {
|
||
await Promise.resolve(thread.stop());
|
||
} catch {
|
||
/* best-effort */
|
||
}
|
||
// cancel 落地需要一拍，立刻 submit 仍可能 409。
|
||
await new Promise((resolve) => setTimeout(resolve, 120));
|
||
}
|
||
|
||
function getStreamErrorMessage(error: unknown): string {
|
||
if (typeof error === "string" && error.trim()) {
|
||
return error;
|
||
}
|
||
if (error instanceof Error && error.message.trim()) {
|
||
return error.message;
|
||
}
|
||
if (typeof error === "object" && error !== null) {
|
||
const message = Reflect.get(error, "message");
|
||
if (typeof message === "string" && message.trim()) {
|
||
return message;
|
||
}
|
||
const nestedError = Reflect.get(error, "error");
|
||
if (nestedError instanceof Error && nestedError.message.trim()) {
|
||
return nestedError.message;
|
||
}
|
||
if (typeof nestedError === "string" && nestedError.trim()) {
|
||
return nestedError;
|
||
}
|
||
}
|
||
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;
|
||
// A run can legitimately be quiet while a long tool call / slow model write is
|
||
// in progress. Reconnect only after a long stall; replacing the local reader
|
||
// must NOT cancel the backend run (see reconnect below).
|
||
const STREAM_RECONNECT_IDLE_MS = 90_000;
|
||
const STREAM_RECONNECT_MAX_ATTEMPTS = 3;
|
||
|
||
async function fetchThreadTitleRefreshState(
|
||
threadId: string,
|
||
signal?: AbortSignal,
|
||
): Promise<ThreadTitleRefreshState | null> {
|
||
const response = await apiFetch(
|
||
`${getBackendBaseURL()}/api/threads/${encodeURIComponent(threadId)}/state`,
|
||
{ signal },
|
||
);
|
||
if (!response.ok) {
|
||
return null;
|
||
}
|
||
const json = (await response.json()) as {
|
||
values?: { title?: string; title_provisional?: boolean };
|
||
};
|
||
if (!json.values) {
|
||
return null;
|
||
}
|
||
return {
|
||
title: json.values.title ?? "",
|
||
title_provisional: json.values.title_provisional,
|
||
};
|
||
}
|
||
|
||
/**
|
||
* Detect "thread not found" / forbidden errors from the LangGraph SDK.
|
||
*
|
||
* Backend returns 404 both when a thread truly doesn't exist and when the
|
||
* currently authenticated user doesn't own it — the SDK surfaces both as
|
||
* ``HTTP 404: {"detail":"Thread <uuid> not found"}``. The most common cause
|
||
* we see in practice is a stale tab whose URL still points at the previous
|
||
* login session's thread. Raw-toasting the 404 just confuses the user, so
|
||
* we re-frame it around the current login identity.
|
||
*/
|
||
function isThreadNotAccessibleError(message: string): boolean {
|
||
const lowered = message.toLowerCase();
|
||
// Backend now returns "not found" (no row + no checkpoint) and
|
||
// "not accessible" (row owned by another user / stale login session) as
|
||
// distinct details — both mean the current user can't use this thread.
|
||
if (/thread\s+[\w-]{8,}\s+not\s+(found|accessible)/i.test(message)) {
|
||
return true;
|
||
}
|
||
// Fallback: a bare 404/403 from the threads endpoint when the SDK has
|
||
// wrapped the detail differently.
|
||
return (
|
||
(lowered.includes("http 404") || lowered.includes("http 403")) &&
|
||
lowered.includes("thread")
|
||
);
|
||
}
|
||
|
||
/**
|
||
* Detect a "Run <id> not found" 404 from the LangGraph SDK.
|
||
*
|
||
* The SDK persists the in-flight run id under ``lg:stream:{threadId}`` in
|
||
* sessionStorage and, on mount, auto-``joinStream``s it to resume a dropped
|
||
* SSE. After a backend restart the in-memory run registry is gone, so the join
|
||
* 404s — and the SDK only clears that key on reconnect *success*, never on
|
||
* failure. So every switch back into the conversation re-fires the same 404.
|
||
* We detect it here to clear the stale key ourselves and stay silent (the
|
||
* conversation renders from history; there is nothing live to reconnect to).
|
||
*/
|
||
function isRunNotFoundError(message: string): boolean {
|
||
return /run\s+[\w-]{8,}\s+not\s+found/i.test(message);
|
||
}
|
||
|
||
/**
|
||
* 同一会话上一轮还没生成完，又提交了新问题 → 后端返回 409
|
||
* 「already has an active run」。匹配 HTTP 409、英文兜底文案，或后端已中文化
|
||
* 的提示，统一映射成一句友好的中文。
|
||
*/
|
||
function isActiveRunConflictError(message: string): boolean {
|
||
return (
|
||
/\b409\b/.test(message) ||
|
||
/already has an active run/i.test(message) ||
|
||
/还在生成回答/.test(message)
|
||
);
|
||
}
|
||
|
||
function isModelConcurrencyLimitError(message: string): boolean {
|
||
return (
|
||
/\b429\b/.test(message) ||
|
||
/并发已达到上限/.test(message) ||
|
||
/并发调用已达到上限/.test(message) ||
|
||
/使用人数较多/.test(message) ||
|
||
/too many requests/i.test(message)
|
||
);
|
||
}
|
||
|
||
function friendlySendLimitMessage(message: string): string | null {
|
||
if (isActiveRunConflictError(message)) {
|
||
return "当前对话还在生成回答，请等待本轮回复结束，或先点“停止”再发送新的问题。";
|
||
}
|
||
if (isModelConcurrencyLimitError(message)) {
|
||
return /当前账号对该模型/.test(message)
|
||
? "当前账号对该模型的并发调用已达到上限，请等待已有回答结束后再试。"
|
||
: "当前使用人数较多，请稍等一下再试。";
|
||
}
|
||
return null;
|
||
}
|
||
|
||
function buildCurrentUserLabel(): string {
|
||
const auth = getStoredAuth();
|
||
if (!auth) {
|
||
return "未登录";
|
||
}
|
||
const email = auth.email?.trim();
|
||
if (email) {
|
||
return email;
|
||
}
|
||
const userId = auth.user_id?.trim();
|
||
return userId ? userId : "未知用户";
|
||
}
|
||
|
||
/**
|
||
* 从 `on_chat_model_end` 事件的 output 里尽力取出模型输出文本。
|
||
*
|
||
* output 形态可能是 AIMessage（`{ content: string | parts[] }`），也可能是
|
||
* LangChain 的 `{ generations: [[{ text, message }]] }`。全部容错，取不到返回空串。
|
||
*/
|
||
function textFromChatModelOutput(output: unknown): string {
|
||
if (output == null) return "";
|
||
if (typeof output === "string") return output;
|
||
if (typeof output !== "object") return String(output);
|
||
const o = output as Record<string, unknown>;
|
||
if ("content" in o) {
|
||
const c = o.content;
|
||
if (typeof c === "string") return c;
|
||
if (Array.isArray(c)) {
|
||
const parts = c
|
||
.map((p) =>
|
||
p && typeof p === "object" && "text" in p
|
||
? String((p as { text?: unknown }).text ?? "")
|
||
: "",
|
||
)
|
||
.filter(Boolean);
|
||
if (parts.length) return parts.join("");
|
||
}
|
||
}
|
||
if (Array.isArray(o.generations)) {
|
||
const texts: string[] = [];
|
||
const collect = (gen: unknown) => {
|
||
if (Array.isArray(gen)) {
|
||
gen.forEach(collect);
|
||
return;
|
||
}
|
||
if (gen && typeof gen === "object") {
|
||
const gg = gen as Record<string, unknown>;
|
||
if (typeof gg.text === "string" && gg.text) texts.push(gg.text);
|
||
else if (gg.message) {
|
||
const t = textFromChatModelOutput(gg.message);
|
||
if (t) texts.push(t);
|
||
}
|
||
}
|
||
};
|
||
collect(o.generations);
|
||
if (texts.length) return texts.join("");
|
||
}
|
||
return "";
|
||
}
|
||
|
||
/**
|
||
* 从 `on_chat_model_*` 事件的 input 里尽力取出最后一条 user/human 文本，
|
||
* 作为本次大模型调用的「问题」。input 可能是 `{ messages: BaseMessage[] | BaseMessage[][] }`。
|
||
*/
|
||
function lastHumanFromChatModelInput(input: unknown): string {
|
||
try {
|
||
const raw =
|
||
input && typeof input === "object" && "messages" in input
|
||
? (input as { messages?: unknown }).messages
|
||
: input;
|
||
const flat: unknown[] = [];
|
||
const push = (m: unknown) => {
|
||
if (Array.isArray(m)) m.forEach(push);
|
||
else if (m != null) flat.push(m);
|
||
};
|
||
push(raw);
|
||
for (let i = flat.length - 1; i >= 0; i--) {
|
||
const m = flat[i];
|
||
if (!m || typeof m !== "object") continue;
|
||
const obj = m as Record<string, unknown>;
|
||
const type = obj.type ?? obj.role;
|
||
if (type === "human" || type === "user") {
|
||
const kwargs = obj.additional_kwargs;
|
||
if (
|
||
kwargs &&
|
||
typeof kwargs === "object" &&
|
||
(kwargs as Record<string, unknown>).hide_from_ui === true
|
||
) {
|
||
continue;
|
||
}
|
||
const text = textFromChatModelOutput(m);
|
||
if (text) return text;
|
||
}
|
||
}
|
||
} catch {
|
||
/* 取不到忽略 */
|
||
}
|
||
return "";
|
||
}
|
||
|
||
export function useThreadStream({
|
||
threadId,
|
||
context,
|
||
isMock,
|
||
assistantId,
|
||
runOptions,
|
||
threadMetadata,
|
||
onStart,
|
||
onFinish,
|
||
onToolStart,
|
||
onToolEnd,
|
||
}: ThreadStreamOptions) {
|
||
const { t } = useI18n();
|
||
const [localSettings] = useLocalSettings();
|
||
// Track the thread ID that is currently streaming to handle thread changes during streaming
|
||
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<string | null>(threadId ?? null);
|
||
const startedRef = useRef(false);
|
||
const lastStreamActivityAtRef = useRef(0);
|
||
const streamReconciliationInFlightRef = useRef(false);
|
||
const streamReconnectAttemptsRef = useRef(0);
|
||
const streamReconnectFailureToastShownRef = useRef(false);
|
||
|
||
const markStreamActivity = useCallback(() => {
|
||
lastStreamActivityAtRef.current = Date.now();
|
||
streamReconnectAttemptsRef.current = 0;
|
||
streamReconnectFailureToastShownRef.current = false;
|
||
}, []);
|
||
|
||
const listeners = useRef({
|
||
onStart,
|
||
onFinish,
|
||
onToolStart,
|
||
onToolEnd,
|
||
});
|
||
|
||
// Keep listeners ref updated with latest callbacks
|
||
useEffect(() => {
|
||
listeners.current = { onStart, onFinish, onToolStart, onToolEnd };
|
||
}, [onStart, onFinish, onToolStart, onToolEnd]);
|
||
|
||
useEffect(() => {
|
||
const normalizedThreadId = threadId ?? null;
|
||
if (!normalizedThreadId) {
|
||
// Reset when the UI moves back to a brand new unsaved thread.
|
||
startedRef.current = false;
|
||
setOnStreamThreadId(normalizedThreadId);
|
||
} else {
|
||
setOnStreamThreadId(normalizedThreadId);
|
||
}
|
||
threadIdRef.current = normalizedThreadId;
|
||
}, [threadId]);
|
||
|
||
const _handleOnStart = useCallback((id: string) => {
|
||
if (!startedRef.current) {
|
||
listeners.current.onStart?.(id);
|
||
startedRef.current = true;
|
||
}
|
||
}, []);
|
||
|
||
const handleStreamStart = useCallback(
|
||
(_threadId: string) => {
|
||
threadIdRef.current = _threadId;
|
||
markStreamActivity();
|
||
_handleOnStart(_threadId);
|
||
},
|
||
[_handleOnStart, markStreamActivity],
|
||
);
|
||
|
||
const queryClient = useQueryClient();
|
||
const updateSubtask = useUpdateSubtask();
|
||
const navigate = useNavigate();
|
||
const runMetadataStorageRef = useRef<
|
||
ReturnType<typeof getRunMetadataStorage> | undefined
|
||
>(undefined);
|
||
const joinStreamRef = useRef<(runId: string) => Promise<unknown>>(
|
||
async () => undefined,
|
||
);
|
||
|
||
if (
|
||
typeof window !== "undefined" &&
|
||
runMetadataStorageRef.current === undefined
|
||
) {
|
||
runMetadataStorageRef.current = getRunMetadataStorage();
|
||
}
|
||
|
||
const thread = useStream<AgentThreadState>({
|
||
client: getAPIClient(isMock),
|
||
assistantId: assistantId ?? "lead_agent",
|
||
threadId: onStreamThreadId,
|
||
reconnectOnMount: runMetadataStorageRef.current
|
||
? () => runMetadataStorageRef.current!
|
||
: false,
|
||
fetchStateHistory: { limit: 1 },
|
||
onCreated(meta) {
|
||
handleStreamStart(meta.thread_id);
|
||
setOnStreamThreadId(meta.thread_id);
|
||
const metadataAgentId = context.agent_id ?? context.agent_name;
|
||
const extraMetadata =
|
||
threadMetadata && typeof threadMetadata === "object" ? threadMetadata : {};
|
||
const metadata: Record<string, unknown> = {
|
||
...(metadataAgentId ? { agent_id: metadataAgentId } : {}),
|
||
...extraMetadata,
|
||
};
|
||
if (Object.keys(metadata).length > 0 && !isMock) {
|
||
void getAPIClient()
|
||
.threads.update(meta.thread_id, { metadata })
|
||
.catch(() => ({}));
|
||
}
|
||
},
|
||
onLangChainEvent(event) {
|
||
markStreamActivity();
|
||
// 操作日志(Y-Log)：每一次大模型调用结束都各上报一条（一次 LLM 调用 = 一条），
|
||
// 让问答日志尽量「多、全」——一个问答轮次里 lead agent / 子代理的每次模型调用
|
||
// 都会落一条。门控与失败静默都在 writeYLog 内部处理，绝不影响问答。
|
||
if (event.event === "on_chat_model_end") {
|
||
try {
|
||
const data = (event.data ?? {}) as {
|
||
input?: unknown;
|
||
output?: unknown;
|
||
};
|
||
const answer = textFromChatModelOutput(data.output);
|
||
const question = lastHumanFromChatModelInput(data.input);
|
||
writeYLog({
|
||
optType: OptType.other,
|
||
optDetail: "CM助手-全量-智能问答",
|
||
url: `/api/threads/${threadIdRef.current ?? ""}/runs/stream`,
|
||
reqParam: { model: event.name ?? "", question },
|
||
returnParam: { data: answer },
|
||
});
|
||
} catch {
|
||
/* 上报绝不影响问答 */
|
||
}
|
||
}
|
||
},
|
||
// The SDK emits UI tool lifecycle events on the dedicated `tools` stream
|
||
// mode. `events` is a separate LangChain trace channel; using it for UI
|
||
// reactions can miss a tool that the conversation has already rendered.
|
||
onToolEvent:
|
||
onToolStart || onToolEnd
|
||
? (event) => {
|
||
markStreamActivity();
|
||
if (event.event === "on_tool_start") {
|
||
listeners.current.onToolStart?.({
|
||
name: event.name,
|
||
data: event.input,
|
||
toolCallId: event.toolCallId,
|
||
});
|
||
return;
|
||
}
|
||
if (event.event === "on_tool_end") {
|
||
listeners.current.onToolEnd?.({
|
||
name: event.name,
|
||
data: event.output,
|
||
toolCallId: event.toolCallId,
|
||
});
|
||
}
|
||
}
|
||
: undefined,
|
||
onUpdateEvent(data) {
|
||
markStreamActivity();
|
||
const updates: Array<Partial<AgentThreadState> | null> = Object.values(
|
||
data || {},
|
||
);
|
||
for (const update of updates) {
|
||
if (update && "title" in update && update.title) {
|
||
const nextTitle = update.title;
|
||
updateCachedThreadLists(queryClient, (threads) =>
|
||
threads.map((t) =>
|
||
t.thread_id === threadIdRef.current
|
||
? {
|
||
...t,
|
||
values: {
|
||
...t.values,
|
||
title: nextTitle,
|
||
},
|
||
}
|
||
: t,
|
||
),
|
||
);
|
||
}
|
||
}
|
||
},
|
||
onCustomEvent(event: unknown) {
|
||
markStreamActivity();
|
||
if (
|
||
typeof event === "object" &&
|
||
event !== null &&
|
||
"type" in event &&
|
||
event.type === "task_running"
|
||
) {
|
||
const e = event as {
|
||
type: "task_running";
|
||
task_id: string;
|
||
message: AIMessage;
|
||
};
|
||
updateSubtask({ id: e.task_id, latestMessage: e.message });
|
||
return;
|
||
}
|
||
|
||
if (
|
||
typeof event === "object" &&
|
||
event !== null &&
|
||
"type" in event &&
|
||
event.type === "llm_retry" &&
|
||
"message" in event &&
|
||
typeof event.message === "string" &&
|
||
event.message.trim()
|
||
) {
|
||
// The backend is already handling this with bounded retry/backoff.
|
||
// Surfacing every transient model retry as a toast made normal model
|
||
// stalls look like repeated "network fluctuation" failures.
|
||
return;
|
||
}
|
||
|
||
// Context compression about to run a (blocking) summarization LLM call —
|
||
// surface a transient hint so the wait before the answer streams isn't a
|
||
// silent dead air. Emitted by DeerFlowSummarizationMiddleware.
|
||
if (
|
||
typeof event === "object" &&
|
||
event !== null &&
|
||
"type" in event &&
|
||
event.type === "context_compacting" &&
|
||
"message" in event &&
|
||
typeof event.message === "string" &&
|
||
event.message.trim()
|
||
) {
|
||
const e = event as { type: "context_compacting"; message: string };
|
||
toast(e.message);
|
||
}
|
||
},
|
||
onError(error) {
|
||
markStreamActivity();
|
||
const message = getStreamErrorMessage(error);
|
||
// Replacing a stalled local SSE reader aborts the in-flight fetch. That
|
||
// is expected and must not look like a user-facing "network" failure.
|
||
if (isAbortLikeError(error)) {
|
||
return;
|
||
}
|
||
// A dropped SSE while the backend run is still alive: rejoin instead of
|
||
// toasting "Failed to fetch" / 网络波动 and leaving the answer half-done.
|
||
if (isTransientStreamTransportError(message)) {
|
||
const staleThreadId = threadIdRef.current ?? onStreamThreadId ?? null;
|
||
const runId = staleThreadId
|
||
? runMetadataStorageRef.current?.getItem(`lg:stream:${staleThreadId}`)
|
||
: null;
|
||
if (
|
||
runId &&
|
||
streamReconnectAttemptsRef.current < STREAM_RECONNECT_MAX_ATTEMPTS
|
||
) {
|
||
streamReconnectAttemptsRef.current += 1;
|
||
lastStreamActivityAtRef.current = Date.now();
|
||
void joinStreamRef.current(runId).catch(() => {
|
||
// Reconciliation will retry while the run stays non-terminal.
|
||
});
|
||
return;
|
||
}
|
||
}
|
||
setOptimisticMessages([]);
|
||
// A stale auto-reconnect ("Run <id> not found") after a backend restart:
|
||
// drop the dirty ``lg:stream:{threadId}`` key so switching back into this
|
||
// conversation stops re-firing the same 404, and render from history
|
||
// silently — there is no live run to rejoin.
|
||
if (isRunNotFoundError(message)) {
|
||
const staleThreadId = threadIdRef.current ?? onStreamThreadId ?? null;
|
||
if (staleThreadId && runMetadataStorageRef.current) {
|
||
runMetadataStorageRef.current.removeItem(
|
||
`lg:stream:${staleThreadId}`,
|
||
);
|
||
}
|
||
return;
|
||
}
|
||
// A "thread not found" 404 almost always means the URL points at a
|
||
// thread from a previous login session — either this tab outlived a
|
||
// logout/re-login or the thread genuinely belongs to a different user.
|
||
// Re-frame the error around the *current* logged-in identity and bounce
|
||
// the user to a fresh chat instead of leaving them stuck on the stale
|
||
// route looking at a raw HTTP 404.
|
||
if (isThreadNotAccessibleError(message)) {
|
||
const userLabel = buildCurrentUserLabel();
|
||
toast.info(
|
||
`当前登录用户为 ${userLabel}，原对话不存在或无权访问，正在为你刷新页面…`,
|
||
);
|
||
// Prefer staying inside an agent chat context when the user is on an
|
||
// agent thread URL — otherwise fall back to the global new chat.
|
||
const hash = typeof window !== "undefined" ? window.location.hash : "";
|
||
const agentMatch = hash.match(/#\/page\/workspace\/agents\/([^/]+)\//);
|
||
const iframeMatch = hash.match(/#\/page\/workspace\/chats_iframe\//);
|
||
const target = agentMatch
|
||
? `/page/workspace/agents/${agentMatch[1]}/chats/new`
|
||
: iframeMatch
|
||
? "/page/workspace/chats_iframe/new"
|
||
: "/page/workspace/chats/new";
|
||
// First navigate to a safe URL so the post-reload page lands on a
|
||
// valid route. Then full-reload so TanStack Query cache, context
|
||
// providers and any other state captured under the previous user's
|
||
// identity get rebuilt against the current login.
|
||
if (typeof window !== "undefined") {
|
||
window.location.hash = target;
|
||
window.setTimeout(() => {
|
||
window.location.reload();
|
||
}, 900);
|
||
} else {
|
||
navigate(target, { replace: true });
|
||
}
|
||
return;
|
||
}
|
||
// 上一轮还没结束又发了新问题（409）或并发达到上限（429）：
|
||
// 给一句明确的中文提示，而不是把原始 HTTP 错误抛给用户。
|
||
const friendly = friendlySendLimitMessage(message);
|
||
if (friendly) {
|
||
toast.info(friendly);
|
||
return;
|
||
}
|
||
toast.error(message);
|
||
},
|
||
onFinish(state) {
|
||
markStreamActivity();
|
||
listeners.current.onFinish?.(state.values);
|
||
void queryClient.invalidateQueries({ queryKey: ["threads", "search"] });
|
||
},
|
||
});
|
||
|
||
// Optimistic messages shown before the server stream responds
|
||
const [optimisticMessages, setOptimisticMessages] = useState<Message[]>([]);
|
||
const [isUploading, setIsUploading] = useState(false);
|
||
const [threadValueOverrides, setThreadValueOverrides] = useState<
|
||
Partial<AgentThreadState>
|
||
>({});
|
||
const sendInFlightRef = useRef(false);
|
||
// Carries the ask time of an in-flight send until the server-assigned human
|
||
// message id shows up, so we can bind the timestamp to the real id (see the
|
||
// reconciliation effect below). ``known`` is the set of human ids that
|
||
// already existed at send time, used to spot the newly-arrived one.
|
||
const pendingAskTimeRef = useRef<{ ts: string; known: Set<string> } | null>(
|
||
null,
|
||
);
|
||
// Track message count before sending so we know when server has responded
|
||
const prevMsgCountRef = useRef(thread.messages.length);
|
||
|
||
// 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;
|
||
joinStreamRef.current = (runId: string) => thread.joinStream(runId);
|
||
|
||
useEffect(() => {
|
||
const activeThreadId = threadIdRef.current ?? onStreamThreadId ?? null;
|
||
if (!activeThreadId || typeof window === "undefined") {
|
||
return;
|
||
}
|
||
|
||
let disposed = false;
|
||
let refreshTimer: number | null = null;
|
||
|
||
const reconcileRun = 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) {
|
||
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.
|
||
//
|
||
// SDK ``stop()`` is *not* a local-only abort: its onStop hook calls
|
||
// ``runs.cancel`` and deletes ``lg:stream:{threadId}``, which is
|
||
// exactly "答一半就不答了". Hide the run id first so stop() only
|
||
// aborts the browser reader, then join the still-running server run.
|
||
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();
|
||
|
||
const stream = threadStreamRef.current;
|
||
storage?.removeItem(`lg:stream:${activeThreadId}`);
|
||
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;
|
||
try {
|
||
storage?.setItem(`lg:stream:${activeThreadId}`, runId);
|
||
} catch {
|
||
/* noop */
|
||
}
|
||
void stream.joinStream(runId).catch(() => {
|
||
// The status poll will either reconcile a terminal run or
|
||
// schedule the next bounded reconnect attempt.
|
||
});
|
||
});
|
||
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 or already reported a transport
|
||
// error and dropped ``isLoading``.
|
||
const finalState = await client.threads.getState(activeThreadId);
|
||
if (disposed) {
|
||
return;
|
||
}
|
||
|
||
storage?.removeItem(`lg:stream:${activeThreadId}`);
|
||
streamReconnectAttemptsRef.current = 0;
|
||
streamReconnectFailureToastShownRef.current = false;
|
||
if (run.status === "success") {
|
||
setOptimisticMessages([]);
|
||
setThreadValueOverrides(finalState.values as Partial<AgentThreadState>);
|
||
listeners.current.onFinish?.(finalState.values as AgentThreadState);
|
||
void queryClient.invalidateQueries({ queryKey: ["threads", "search"] });
|
||
} else {
|
||
setThreadValueOverrides(finalState.values as Partial<AgentThreadState>);
|
||
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 reconcileRun(),
|
||
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
|
||
// a fresh chat" signal — abort the in-flight SSE and clear optimistic state.
|
||
//
|
||
// We deliberately do NOT reset on every threadId change. During a normal
|
||
// "user just sent the first message on /chats/new" flow, useThreadStream
|
||
// sees threadId transition from undefined (the landing state) to the
|
||
// server-assigned thread id once onCreated fires. Stopping the stream and
|
||
// clearing optimistic messages in that case is exactly what made submitted
|
||
// questions visually disappear — the live run that just started would be
|
||
// aborted by the very effect intended to reset stale state.
|
||
useEffect(() => {
|
||
if (threadId != null) {
|
||
// Forward direction: starting or continuing with a real thread.
|
||
// No reset needed; the upstream useStream will load history for this id.
|
||
setThreadValueOverrides({});
|
||
return;
|
||
}
|
||
const prevThread = threadStreamRef.current;
|
||
if (prevThread?.isLoading) {
|
||
try {
|
||
prevThread.stop();
|
||
} catch {
|
||
// best-effort; nothing useful to do if the SDK fails to abort
|
||
}
|
||
}
|
||
startedRef.current = false;
|
||
sendInFlightRef.current = false;
|
||
prevMsgCountRef.current = 0;
|
||
setOptimisticMessages([]);
|
||
setIsUploading(false);
|
||
setThreadValueOverrides({});
|
||
}, [threadId]);
|
||
|
||
useEffect(() => {
|
||
const activeThreadId = threadIdRef.current ?? onStreamThreadId ?? null;
|
||
const effectiveTitleProvisional =
|
||
(threadValueOverrides.title_provisional ??
|
||
thread.values?.title_provisional) === true;
|
||
|
||
if (!activeThreadId || thread.isLoading || !effectiveTitleProvisional) {
|
||
return;
|
||
}
|
||
|
||
let cancelled = false;
|
||
let timeoutId: ReturnType<typeof setTimeout> | null = null;
|
||
const controller = new AbortController();
|
||
|
||
const poll = async () => {
|
||
while (!cancelled) {
|
||
try {
|
||
const nextState = await fetchThreadTitleRefreshState(
|
||
activeThreadId,
|
||
controller.signal,
|
||
);
|
||
if (cancelled || !nextState) {
|
||
return;
|
||
}
|
||
|
||
setThreadValueOverrides((prev) => ({
|
||
...prev,
|
||
title: nextState.title,
|
||
title_provisional: nextState.title_provisional,
|
||
}));
|
||
|
||
if (nextState.title) {
|
||
updateCachedThreadLists(queryClient, (threads) =>
|
||
threads.map((t) =>
|
||
t.thread_id === activeThreadId
|
||
? {
|
||
...t,
|
||
values: {
|
||
...t.values,
|
||
title: nextState.title,
|
||
title_provisional: nextState.title_provisional,
|
||
},
|
||
}
|
||
: t,
|
||
),
|
||
);
|
||
}
|
||
|
||
if (nextState.title_provisional !== true) {
|
||
void queryClient.invalidateQueries({ queryKey: ["threads", "search"] });
|
||
return;
|
||
}
|
||
} catch {
|
||
return;
|
||
}
|
||
|
||
await new Promise<void>((resolve) => {
|
||
timeoutId = setTimeout(resolve, 2000);
|
||
});
|
||
}
|
||
};
|
||
|
||
void poll();
|
||
|
||
return () => {
|
||
cancelled = true;
|
||
controller.abort();
|
||
if (timeoutId) {
|
||
clearTimeout(timeoutId);
|
||
}
|
||
};
|
||
}, [
|
||
onStreamThreadId,
|
||
queryClient,
|
||
thread.isLoading,
|
||
thread.values?.title_provisional,
|
||
threadValueOverrides.title_provisional,
|
||
]);
|
||
|
||
// Clear optimistic when server messages arrive (count increases)
|
||
useEffect(() => {
|
||
if (
|
||
optimisticMessages.length > 0 &&
|
||
thread.messages.length > prevMsgCountRef.current
|
||
) {
|
||
setOptimisticMessages([]);
|
||
}
|
||
}, [thread.messages.length, optimisticMessages.length]);
|
||
|
||
// Bind a just-sent question's ask time to the server-assigned human message
|
||
// id once it appears. Persists to localStorage so the time shows on every
|
||
// question and survives reloads, independent of the backend echoing it back.
|
||
useEffect(() => {
|
||
const pending = pendingAskTimeRef.current;
|
||
if (!pending) return;
|
||
for (const message of thread.messages) {
|
||
if (
|
||
message.type === "human" &&
|
||
message.id &&
|
||
!message.id.startsWith("opt-") &&
|
||
!pending.known.has(message.id)
|
||
) {
|
||
rememberAskTime(message.id, pending.ts);
|
||
pendingAskTimeRef.current = null;
|
||
break;
|
||
}
|
||
}
|
||
}, [thread.messages]);
|
||
|
||
const sendMessage = useCallback(
|
||
async (
|
||
threadId: string,
|
||
message: PromptInputMessage,
|
||
extraContext?: Record<string, unknown>,
|
||
options?: SendMessageOptions,
|
||
) => {
|
||
if (sendInFlightRef.current) {
|
||
return;
|
||
}
|
||
sendInFlightRef.current = true;
|
||
|
||
const text = message.text.trim();
|
||
// Wall-clock time the question was asked. Stored on the human message so
|
||
// the chat can show "几点提问" and have it survive a reload (it round-trips
|
||
// through the checkpoint inside additional_kwargs, same as `files`).
|
||
const clientTs = new Date().toISOString();
|
||
// Remember which human messages already exist so the reconciliation
|
||
// effect can bind ``clientTs`` to the new one once the server returns it.
|
||
pendingAskTimeRef.current = {
|
||
ts: clientTs,
|
||
known: new Set(
|
||
thread.messages
|
||
.filter((m) => m.type === "human" && m.id)
|
||
.map((m) => m.id as string),
|
||
),
|
||
};
|
||
|
||
// Capture current count before showing optimistic messages
|
||
prevMsgCountRef.current = thread.messages.length;
|
||
|
||
// Build optimistic files list with uploading status
|
||
const optimisticFiles: FileInMessage[] = (message.files ?? []).map(
|
||
(f) => ({
|
||
filename: f.filename ?? "",
|
||
size: 0,
|
||
status: "uploading" as const,
|
||
}),
|
||
);
|
||
|
||
const hideFromUI = options?.additionalKwargs?.hide_from_ui === true;
|
||
const optimisticAdditionalKwargs = {
|
||
...options?.additionalKwargs,
|
||
client_ts: clientTs,
|
||
...(optimisticFiles.length > 0 ? { files: optimisticFiles } : {}),
|
||
};
|
||
|
||
const newOptimistic: Message[] = [];
|
||
if (!hideFromUI) {
|
||
newOptimistic.push({
|
||
type: "human",
|
||
id: `opt-human-${Date.now()}`,
|
||
content: text ? [{ type: "text", text }] : "",
|
||
additional_kwargs: optimisticAdditionalKwargs,
|
||
});
|
||
}
|
||
|
||
if (optimisticFiles.length > 0 && !hideFromUI) {
|
||
// Mock AI message while files are being uploaded
|
||
newOptimistic.push({
|
||
type: "ai",
|
||
id: `opt-ai-${Date.now()}`,
|
||
content: t.uploads.uploadingFiles,
|
||
additional_kwargs: { element: "task" },
|
||
});
|
||
}
|
||
setOptimisticMessages(newOptimistic);
|
||
|
||
// Only fire onStart immediately for an existing persisted thread.
|
||
// Brand-new chats should wait for onCreated(meta.thread_id) so URL sync
|
||
// uses the real server-generated thread id.
|
||
if (threadIdRef.current) {
|
||
_handleOnStart(threadId);
|
||
}
|
||
|
||
let uploadedFileInfo: UploadedFileInfo[] = [];
|
||
|
||
try {
|
||
// Upload files first if any
|
||
if (message.files && message.files.length > 0) {
|
||
setIsUploading(true);
|
||
try {
|
||
const filePromises = message.files.map((fileUIPart) =>
|
||
promptInputFilePartToFile(fileUIPart),
|
||
);
|
||
|
||
const conversionResults = await Promise.all(filePromises);
|
||
const files = conversionResults.filter(
|
||
(file): file is File => file !== null,
|
||
);
|
||
const failedConversions = conversionResults.length - files.length;
|
||
|
||
if (failedConversions > 0) {
|
||
throw new Error(
|
||
`Failed to prepare ${failedConversions} attachment(s) for upload. Please retry.`,
|
||
);
|
||
}
|
||
|
||
if (!threadId) {
|
||
throw new Error("Thread is not ready for file upload.");
|
||
}
|
||
|
||
if (files.length > 0) {
|
||
const uploadResponse = await uploadFiles(threadId, files);
|
||
uploadedFileInfo = uploadResponse.files;
|
||
|
||
// Update optimistic human message with uploaded status + paths
|
||
const uploadedFiles: FileInMessage[] = uploadedFileInfo.map(
|
||
(info) => ({
|
||
filename: info.filename,
|
||
size: info.size,
|
||
path: info.virtual_path,
|
||
status: "uploaded" as const,
|
||
}),
|
||
);
|
||
setOptimisticMessages((messages) => {
|
||
if (messages.length > 1 && messages[0]) {
|
||
const humanMessage: Message = messages[0];
|
||
return [
|
||
{
|
||
...humanMessage,
|
||
additional_kwargs: {
|
||
...humanMessage.additional_kwargs,
|
||
files: uploadedFiles,
|
||
},
|
||
},
|
||
...messages.slice(1),
|
||
];
|
||
}
|
||
return messages;
|
||
});
|
||
}
|
||
} catch (error) {
|
||
const errorMessage =
|
||
error instanceof Error
|
||
? error.message
|
||
: "Failed to upload files.";
|
||
toast.error(errorMessage);
|
||
setOptimisticMessages([]);
|
||
throw error;
|
||
} finally {
|
||
setIsUploading(false);
|
||
}
|
||
}
|
||
|
||
// Build files metadata for submission (included in additional_kwargs)
|
||
const filesForSubmit: FileInMessage[] = uploadedFileInfo.map(
|
||
(info) => ({
|
||
filename: info.filename,
|
||
size: info.size,
|
||
path: info.virtual_path,
|
||
status: "uploaded" as const,
|
||
}),
|
||
);
|
||
|
||
const mergedContext = {
|
||
...context,
|
||
...extraContext,
|
||
};
|
||
const selectedLlmWikiIds = Array.isArray(
|
||
mergedContext.llmwiki_knowledge_base_ids,
|
||
)
|
||
? mergedContext.llmwiki_knowledge_base_ids
|
||
: [];
|
||
const ragModeEnabled =
|
||
localSettings.display.ragModeEnabled ||
|
||
isThreadRagModeLocked(threadId) ||
|
||
selectedLlmWikiIds.length > 0;
|
||
if (ragModeEnabled) {
|
||
lockThreadRagMode(threadId);
|
||
}
|
||
|
||
const contextCompressionEnabled =
|
||
localSettings.display.contextCompressionEnabled ?? false;
|
||
|
||
const hiddenContextMessages = normalizeHiddenContextMessages(options);
|
||
const submittedMessages: Message[] = [
|
||
...hiddenContextMessages.map((item) => ({
|
||
type: "human" as const,
|
||
...(item.name ? { name: item.name } : {}),
|
||
content: [
|
||
{
|
||
type: "text" as const,
|
||
text: item.text,
|
||
},
|
||
],
|
||
additional_kwargs: {
|
||
...item.additionalKwargs,
|
||
hide_from_ui: true,
|
||
},
|
||
})),
|
||
{
|
||
type: "human" as const,
|
||
content: [
|
||
{
|
||
type: "text" as const,
|
||
text,
|
||
},
|
||
],
|
||
additional_kwargs: {
|
||
...options?.additionalKwargs,
|
||
client_ts: clientTs,
|
||
...(filesForSubmit.length > 0
|
||
? { files: filesForSubmit }
|
||
: {}),
|
||
},
|
||
},
|
||
];
|
||
|
||
await thread.submit(
|
||
{
|
||
messages: submittedMessages,
|
||
},
|
||
{
|
||
threadId: threadId,
|
||
streamSubgraphs: true,
|
||
streamResumable: true,
|
||
onDisconnect: runOptions?.onDisconnect,
|
||
multitaskStrategy: runOptions?.multitaskStrategy,
|
||
config: {
|
||
recursion_limit: 1000,
|
||
},
|
||
context: {
|
||
...mergedContext,
|
||
thinking_enabled: mergedContext.mode !== "flash",
|
||
is_plan_mode:
|
||
mergedContext.mode === "pro" || mergedContext.mode === "ultra",
|
||
subagent_enabled: mergedContext.mode === "ultra",
|
||
reasoning_effort:
|
||
mergedContext.reasoning_effort ??
|
||
(mergedContext.mode === "ultra"
|
||
? "high"
|
||
: mergedContext.mode === "pro"
|
||
? "medium"
|
||
: mergedContext.mode === "thinking"
|
||
? "low"
|
||
: undefined),
|
||
memory_injection_enabled:
|
||
mergedContext.memory_injection_enabled ?? true,
|
||
rag_mode_enabled: ragModeEnabled,
|
||
summarization_enabled: contextCompressionEnabled,
|
||
thread_id: threadId,
|
||
},
|
||
},
|
||
);
|
||
void queryClient.invalidateQueries({ queryKey: ["threads", "search"] });
|
||
} catch (error) {
|
||
const message =
|
||
error instanceof Error ? error.message : String(error);
|
||
const friendly = friendlySendLimitMessage(message);
|
||
if (friendly) {
|
||
setOptimisticMessages((messages) => {
|
||
const human = messages.find((item) => item.type === "human");
|
||
const kept = human ? [human] : [];
|
||
return [
|
||
...kept,
|
||
{
|
||
type: "ai",
|
||
id: `opt-ai-limit-${Date.now()}`,
|
||
content: friendly,
|
||
additional_kwargs: { element: "task" },
|
||
} as Message,
|
||
];
|
||
});
|
||
setIsUploading(false);
|
||
return;
|
||
}
|
||
setOptimisticMessages([]);
|
||
setIsUploading(false);
|
||
throw error;
|
||
} finally {
|
||
sendInFlightRef.current = false;
|
||
}
|
||
},
|
||
[thread, _handleOnStart, t.uploads.uploadingFiles, context, localSettings.display.ragModeEnabled, queryClient, runOptions],
|
||
);
|
||
|
||
// Merge thread with optimistic messages for display
|
||
const mergedValues =
|
||
Object.keys(threadValueOverrides).length > 0
|
||
? {
|
||
...thread.values,
|
||
...threadValueOverrides,
|
||
}
|
||
: thread.values;
|
||
|
||
const mergedMessages =
|
||
Array.isArray(mergedValues.messages)
|
||
? (mergedValues.messages as Message[])
|
||
: thread.messages;
|
||
|
||
const mergedThread =
|
||
optimisticMessages.length > 0 || mergedValues !== thread.values
|
||
? ({
|
||
...thread,
|
||
values: mergedValues,
|
||
messages: [...mergedMessages, ...optimisticMessages],
|
||
} as typeof thread)
|
||
: thread;
|
||
|
||
// 第 4 个返回值 `onStreamThreadId` = 底层 useStream 当前实际绑定的线程 id（新建/未保存时为 null）。
|
||
// 调用方据此判断「流是否已切到全新线程」，避免在切换会话的那一帧把消息发进上一个会话
|
||
// （submit 的 threadId 选项仅在 hook 线程为 null/undefined 时才生效）。
|
||
return [mergedThread, sendMessage, isUploading, onStreamThreadId] as const;
|
||
}
|
||
|
||
export function useThreads(
|
||
params: Parameters<ThreadsClient["search"]>[0] = {
|
||
limit: 50,
|
||
sortBy: "updated_at",
|
||
sortOrder: "desc",
|
||
select: ["thread_id", "updated_at", "values", "metadata"],
|
||
},
|
||
) {
|
||
const apiClient = getAPIClient();
|
||
return useQuery<AgentThread[]>({
|
||
queryKey: ["threads", "search", params],
|
||
queryFn: async () => {
|
||
const maxResults = params.limit;
|
||
const initialOffset = params.offset ?? 0;
|
||
const DEFAULT_PAGE_SIZE = 50;
|
||
|
||
// Preserve prior semantics: if a non-positive limit is explicitly provided,
|
||
// delegate to a single search call with the original parameters.
|
||
if (maxResults !== undefined && maxResults <= 0) {
|
||
const response =
|
||
await apiClient.threads.search<AgentThreadState>(params);
|
||
return response as AgentThread[];
|
||
}
|
||
|
||
const pageSize =
|
||
typeof maxResults === "number" && maxResults > 0
|
||
? Math.min(DEFAULT_PAGE_SIZE, maxResults)
|
||
: DEFAULT_PAGE_SIZE;
|
||
|
||
const threads: AgentThread[] = [];
|
||
let offset = initialOffset;
|
||
|
||
while (true) {
|
||
if (typeof maxResults === "number" && threads.length >= maxResults) {
|
||
break;
|
||
}
|
||
|
||
const currentLimit =
|
||
typeof maxResults === "number"
|
||
? Math.min(pageSize, maxResults - threads.length)
|
||
: pageSize;
|
||
|
||
if (typeof maxResults === "number" && currentLimit <= 0) {
|
||
break;
|
||
}
|
||
|
||
const response = (await apiClient.threads.search<AgentThreadState>({
|
||
...params,
|
||
limit: currentLimit,
|
||
offset,
|
||
})) as AgentThread[];
|
||
|
||
threads.push(...response);
|
||
|
||
if (response.length < currentLimit) {
|
||
break;
|
||
}
|
||
|
||
offset += response.length;
|
||
}
|
||
|
||
// Safety net on top of the backend's system-thread exclusion: drop
|
||
// machine-created / wizard threads (scheduled tasks, report-structure
|
||
// setup, writing setup, notebooks, …) in case an older backend still
|
||
// returns them.
|
||
return threads.filter((t) => !isHiddenFromRecentChats(t));
|
||
},
|
||
refetchOnWindowFocus: false,
|
||
});
|
||
}
|
||
|
||
export interface ThreadsPageResult {
|
||
threads: AgentThread[];
|
||
hasMore: boolean;
|
||
}
|
||
|
||
/**
|
||
* Apply ``transform`` to the thread array inside every cached
|
||
* ``["threads", "search"]`` entry. The sidebar queries cache a plain
|
||
* ``AgentThread[]`` while the paged chats list (`useThreadsPage`) caches a
|
||
* ``ThreadsPageResult`` object under the same key prefix, so fuzzy
|
||
* ``setQueriesData`` updaters must handle both shapes.
|
||
*/
|
||
function updateCachedThreadLists(
|
||
queryClient: QueryClient,
|
||
transform: (threads: AgentThread[]) => AgentThread[],
|
||
) {
|
||
queryClient.setQueriesData<AgentThread[] | ThreadsPageResult | undefined>(
|
||
{ queryKey: ["threads", "search"], exact: false },
|
||
(oldData) => {
|
||
if (Array.isArray(oldData)) {
|
||
return transform(oldData);
|
||
}
|
||
if (oldData && Array.isArray(oldData.threads)) {
|
||
return { ...oldData, threads: transform(oldData.threads) };
|
||
}
|
||
return oldData;
|
||
},
|
||
);
|
||
}
|
||
|
||
/**
|
||
* Server-side paginated conversation list for the chats page. Unlike
|
||
* `useThreads` (which loads a bounded recent window for the sidebar), this
|
||
* fetches exactly one page per request so a long history is never pulled in
|
||
* one go. `search` matches thread titles on the backend; one extra row is
|
||
* over-fetched only to detect whether a next page exists. The query key stays
|
||
* under ["threads", "search"] so existing invalidations refresh pages too.
|
||
*/
|
||
export function useThreadsPage(options: {
|
||
page: number;
|
||
pageSize: number;
|
||
search?: string;
|
||
metadata?: Record<string, unknown>;
|
||
}) {
|
||
const { page, pageSize, search, metadata } = options;
|
||
const normalizedSearch = search?.trim() ?? "";
|
||
return useQuery<ThreadsPageResult>({
|
||
queryKey: [
|
||
"threads",
|
||
"search",
|
||
"paged",
|
||
page,
|
||
pageSize,
|
||
normalizedSearch,
|
||
metadata ?? null,
|
||
],
|
||
queryFn: async () => {
|
||
// Direct fetch instead of the LangGraph SDK: the SDK's threads.search
|
||
// drops fields it doesn't know about, and `query` (title search) is a
|
||
// gateway extension.
|
||
const response = await apiFetch(
|
||
`${getLangGraphBaseURL()}/threads/search`,
|
||
{
|
||
method: "POST",
|
||
headers: { "Content-Type": "application/json" },
|
||
body: JSON.stringify({
|
||
limit: pageSize + 1,
|
||
offset: (page - 1) * pageSize,
|
||
metadata: metadata ?? {},
|
||
query: normalizedSearch || undefined,
|
||
}),
|
||
},
|
||
);
|
||
if (!response.ok) {
|
||
throw new Error("Failed to load conversations");
|
||
}
|
||
const rows = (await response.json()) as AgentThread[];
|
||
// 与后端 exclude_system 对齐：普通列表再兜一层。调用方显式按
|
||
// thread_type / system 取数（笔记本等）时不滤，避免把目标会话藏掉。
|
||
const visible = shouldHideSystemThreadsFromList(metadata)
|
||
? rows.filter((thread) => !isHiddenFromRecentChats(thread))
|
||
: rows;
|
||
return {
|
||
threads: visible.slice(0, pageSize),
|
||
hasMore: rows.length > pageSize,
|
||
};
|
||
},
|
||
placeholderData: (previous) => previous,
|
||
refetchOnWindowFocus: false,
|
||
});
|
||
}
|
||
|
||
/**
|
||
* Total number of conversations matching the same filters as
|
||
* `useThreadsPage`, via the lightweight `/threads/count` endpoint — powers
|
||
* the "共 N 条 / X 页" display without fetching any rows. Depends only on the
|
||
* filters (not the page), so paging doesn't refire it.
|
||
*/
|
||
export function useThreadsCount(options: {
|
||
search?: string;
|
||
metadata?: Record<string, unknown>;
|
||
}) {
|
||
const normalizedSearch = options.search?.trim() ?? "";
|
||
return useQuery<number>({
|
||
queryKey: [
|
||
"threads",
|
||
"search",
|
||
"count",
|
||
normalizedSearch,
|
||
options.metadata ?? null,
|
||
],
|
||
queryFn: async () => {
|
||
const response = await apiFetch(
|
||
`${getLangGraphBaseURL()}/threads/count`,
|
||
{
|
||
method: "POST",
|
||
headers: { "Content-Type": "application/json" },
|
||
body: JSON.stringify({
|
||
metadata: options.metadata ?? {},
|
||
query: normalizedSearch || undefined,
|
||
}),
|
||
},
|
||
);
|
||
if (!response.ok) {
|
||
throw new Error("Failed to count conversations");
|
||
}
|
||
const data = (await response.json()) as { total: number };
|
||
return data.total;
|
||
},
|
||
refetchOnWindowFocus: false,
|
||
});
|
||
}
|
||
|
||
export function useDeleteThread() {
|
||
const queryClient = useQueryClient();
|
||
const apiClient = getAPIClient();
|
||
return useMutation({
|
||
mutationFn: async ({ threadId }: { threadId: string }) => {
|
||
await apiClient.threads.delete(threadId);
|
||
|
||
const response = await apiFetch(
|
||
`${getBackendBaseURL()}/api/threads/${encodeURIComponent(threadId)}`,
|
||
{
|
||
method: "DELETE",
|
||
},
|
||
);
|
||
|
||
if (!response.ok) {
|
||
const error = await response
|
||
.json()
|
||
.catch(() => ({ detail: "Failed to delete local thread data." }));
|
||
throw new Error(error.detail ?? "Failed to delete local thread data.");
|
||
}
|
||
},
|
||
onSuccess(_, { threadId }) {
|
||
updateCachedThreadLists(queryClient, (threads) =>
|
||
threads.filter((t) => t.thread_id !== threadId),
|
||
);
|
||
},
|
||
onSettled() {
|
||
void queryClient.invalidateQueries({ queryKey: ["threads", "search"] });
|
||
},
|
||
});
|
||
}
|
||
|
||
/**
|
||
* Delete a conversation turn (a user question together with its answer) from a
|
||
* thread. Hits the Gateway endpoint that removes the question and every
|
||
* following message up to the next question from the latest checkpoint, and
|
||
* returns the ids that were actually removed. The live `useStream` cache is not
|
||
* auto-refreshed by this call — callers hide the returned ids locally (see
|
||
* `MessageList`) so the removal shows immediately; a reload then reads the
|
||
* persisted, already-trimmed state.
|
||
*/
|
||
export function useDeleteMessage() {
|
||
return useMutation({
|
||
mutationFn: async ({
|
||
threadId,
|
||
messageId,
|
||
}: {
|
||
threadId: string;
|
||
messageId: string;
|
||
}): Promise<string[]> => {
|
||
const response = await apiFetch(
|
||
`${getBackendBaseURL()}/api/threads/${encodeURIComponent(
|
||
threadId,
|
||
)}/messages/${encodeURIComponent(messageId)}`,
|
||
{
|
||
method: "DELETE",
|
||
},
|
||
);
|
||
|
||
if (!response.ok) {
|
||
const error = await response
|
||
.json()
|
||
.catch(() => ({ detail: "删除消息失败。" }));
|
||
throw new Error(error.detail ?? "删除消息失败。");
|
||
}
|
||
|
||
const body = (await response.json().catch(() => ({}))) as {
|
||
deleted_ids?: string[];
|
||
};
|
||
// Fall back to the requested id so the caller can always hide something.
|
||
return body.deleted_ids?.length ? body.deleted_ids : [messageId];
|
||
},
|
||
});
|
||
}
|
||
|
||
/**
|
||
* Clear all messages from a thread checkpoint. The caller should immediately
|
||
* hide the returned ids locally because the active stream cache may still hold
|
||
* the old message objects until a reload/refetch reads the trimmed checkpoint.
|
||
*/
|
||
export function useClearThreadMessages() {
|
||
return useMutation({
|
||
mutationFn: async ({
|
||
threadId,
|
||
}: {
|
||
threadId: string;
|
||
}): Promise<string[]> => {
|
||
const response = await apiFetch(
|
||
`${getBackendBaseURL()}/api/threads/${encodeURIComponent(
|
||
threadId,
|
||
)}/messages`,
|
||
{
|
||
method: "DELETE",
|
||
},
|
||
);
|
||
|
||
if (!response.ok) {
|
||
const error = await response
|
||
.json()
|
||
.catch(() => ({ detail: "清空对话失败。" }));
|
||
throw new Error(error.detail ?? "清空对话失败。");
|
||
}
|
||
|
||
const body = (await response.json().catch(() => ({}))) as {
|
||
deleted_ids?: string[];
|
||
};
|
||
return body.deleted_ids ?? [];
|
||
},
|
||
});
|
||
}
|
||
|
||
/**
|
||
* Persist an in-place message edit in the latest thread checkpoint.
|
||
*
|
||
* This is opt-in UI behaviour: most chat pages keep their existing
|
||
* "edit question and resubmit" affordance, while position roundtable can pass
|
||
* this mutation down to save the original message content itself.
|
||
*/
|
||
export function useEditMessage() {
|
||
return useMutation({
|
||
mutationFn: async ({
|
||
threadId,
|
||
messageId,
|
||
content,
|
||
}: {
|
||
threadId: string;
|
||
messageId: string;
|
||
content: string;
|
||
}): Promise<Message> => {
|
||
const response = await apiFetch(
|
||
`${getBackendBaseURL()}/api/threads/${encodeURIComponent(
|
||
threadId,
|
||
)}/messages/${encodeURIComponent(messageId)}`,
|
||
{
|
||
method: "PATCH",
|
||
headers: { "Content-Type": "application/json" },
|
||
body: JSON.stringify({ content }),
|
||
},
|
||
);
|
||
|
||
if (!response.ok) {
|
||
const error = await response
|
||
.json()
|
||
.catch(() => ({ detail: "消息保存失败。" }));
|
||
throw new Error(error.detail ?? "消息保存失败。");
|
||
}
|
||
|
||
const body = (await response.json().catch(() => ({}))) as {
|
||
message?: Message;
|
||
};
|
||
return body.message ?? ({ id: messageId, type: "human", content } as Message);
|
||
},
|
||
});
|
||
}
|
||
|
||
/**
|
||
* Merge metadata into a conversation's record via the Gateway
|
||
* `PATCH /api/threads/{id}` route. Used to persist per-conversation flags such
|
||
* as `is_pinned` (置顶) and `is_favorite` (收藏) — a conversation belongs to a
|
||
* single user, so these live directly on the thread metadata (no join table).
|
||
* Updates the `["threads", "search"]` cache optimistically so the UI reflects
|
||
* the toggle immediately, mirroring `useRenameThread`.
|
||
*/
|
||
export function useUpdateThreadMetadata() {
|
||
const queryClient = useQueryClient();
|
||
return useMutation({
|
||
mutationFn: async ({
|
||
threadId,
|
||
metadata,
|
||
}: {
|
||
threadId: string;
|
||
metadata: Record<string, unknown>;
|
||
}) => {
|
||
const response = await apiFetch(
|
||
`${getBackendBaseURL()}/api/threads/${encodeURIComponent(threadId)}`,
|
||
{
|
||
method: "PATCH",
|
||
headers: { "Content-Type": "application/json" },
|
||
body: JSON.stringify({ metadata }),
|
||
},
|
||
);
|
||
if (!response.ok) {
|
||
const error = await response
|
||
.json()
|
||
.catch(() => ({ detail: "更新对话失败。" }));
|
||
throw new Error(error.detail ?? "更新对话失败。");
|
||
}
|
||
},
|
||
onSuccess(_, { threadId, metadata }) {
|
||
updateCachedThreadLists(queryClient, (threads) =>
|
||
threads.map((t) =>
|
||
t.thread_id === threadId
|
||
? {
|
||
...t,
|
||
metadata: {
|
||
...(t.metadata as Record<string, unknown> | null),
|
||
...metadata,
|
||
},
|
||
}
|
||
: t,
|
||
),
|
||
);
|
||
},
|
||
});
|
||
}
|
||
|
||
/** Toggle a conversation's pinned (置顶) flag. */
|
||
export function useToggleThreadPin() {
|
||
const { mutate, ...rest } = useUpdateThreadMetadata();
|
||
return {
|
||
...rest,
|
||
toggle: (threadId: string, pinned: boolean) =>
|
||
mutate({ threadId, metadata: { is_pinned: pinned } }),
|
||
};
|
||
}
|
||
|
||
/** Toggle a conversation's favorite (收藏) flag. */
|
||
export function useToggleThreadFavorite() {
|
||
const { mutate, ...rest } = useUpdateThreadMetadata();
|
||
return {
|
||
...rest,
|
||
toggle: (threadId: string, favorite: boolean) =>
|
||
mutate({ threadId, metadata: { is_favorite: favorite } }),
|
||
};
|
||
}
|
||
|
||
export function useRenameThread() {
|
||
const queryClient = useQueryClient();
|
||
const apiClient = getAPIClient();
|
||
return useMutation({
|
||
mutationFn: async ({
|
||
threadId,
|
||
title,
|
||
}: {
|
||
threadId: string;
|
||
title: string;
|
||
}) => {
|
||
await apiClient.threads.updateState(threadId, {
|
||
values: { title, title_provisional: false },
|
||
});
|
||
},
|
||
onSuccess(_, { threadId, title }) {
|
||
updateCachedThreadLists(queryClient, (threads) =>
|
||
threads.map((t) =>
|
||
t.thread_id === threadId
|
||
? {
|
||
...t,
|
||
values: {
|
||
...t.values,
|
||
title,
|
||
title_provisional: false,
|
||
},
|
||
}
|
||
: t,
|
||
),
|
||
);
|
||
},
|
||
});
|
||
}
|