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 { useCallback, useEffect, useRef, useState } from "react"; import { toast } from "sonner"; import type { PromptInputMessage } from "@/components/ai-elements/prompt-input"; import { getAPIClient } from "../api"; import { useI18n } from "../i18n/hooks"; import type { FileInMessage } from "../messages/utils"; import type { LocalSettings } from "../settings"; import { useUpdateSubtask } from "../tasks/context"; import type { UploadedFileInfo } from "../uploads"; import { uploadFiles } from "../uploads"; import type { AgentThread, AgentThreadState } from "./types"; export type ToolEndEvent = { name: string; data: unknown; }; export type ThreadStreamOptions = { threadId?: string | null | undefined; context: LocalSettings["context"]; isMock?: boolean; onStart?: (threadId: string) => void; onFinish?: (state: AgentThreadState) => void; onToolEnd?: (event: ToolEndEvent) => void; }; export function useThreadStream({ threadId, context, isMock, onStart, onFinish, onToolEnd, }: ThreadStreamOptions) { const { t } = useI18n(); const threadIdRef = useRef(threadId ?? null); const startedRef = useRef(false); const listeners = useRef({ onStart, onFinish, onToolEnd, }); // Keep listeners ref updated with latest callbacks useEffect(() => { listeners.current = { onStart, onFinish, onToolEnd }; }, [onStart, onFinish, onToolEnd]); useEffect(() => { if (threadIdRef.current && threadIdRef.current !== threadId) { threadIdRef.current = threadId ?? null; startedRef.current = false; // Reset for new thread } }, [threadId]); const _handleStart = useCallback((id: string) => { if (!startedRef.current) { listeners.current.onStart?.(id); startedRef.current = true; } }, []); const queryClient = useQueryClient(); const updateSubtask = useUpdateSubtask(); const thread = useStream({ client: getAPIClient(isMock), assistantId: "lead_agent", threadId: threadIdRef.current, reconnectOnMount: true, fetchStateHistory: { limit: 1 }, onCreated(meta) { threadIdRef.current = meta.thread_id; _handleStart(meta.thread_id); }, onLangChainEvent(event) { if (event.event === "on_tool_end") { listeners.current.onToolEnd?.({ name: event.name, data: event.data, }); } }, onUpdateEvent(data) { const updates: Array | null> = Object.values( data || {}, ); for (const update of updates) { if (update && "title" in update && update.title) { void queryClient.setQueriesData( { queryKey: ["threads", "search"], exact: false, }, (oldData: Array | undefined) => { return oldData?.map((t) => { if (t.thread_id === threadIdRef.current) { return { ...t, values: { ...t.values, title: update.title, }, }; } return t; }); }, ); } } }, onCustomEvent(event: unknown) { 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 }); } }, onFinish(state) { listeners.current.onFinish?.(state.values); void queryClient.invalidateQueries({ queryKey: ["threads", "search"] }); }, }); // Optimistic messages shown before the server stream responds const [optimisticMessages, setOptimisticMessages] = useState([]); // Track message count before sending so we know when server has responded const prevMsgCountRef = useRef(thread.messages.length); // Clear optimistic when server messages arrive (count increases) useEffect(() => { if ( optimisticMessages.length > 0 && thread.messages.length > prevMsgCountRef.current ) { setOptimisticMessages([]); } }, [thread.messages.length, optimisticMessages.length]); const sendMessage = useCallback( async ( threadId: string, message: PromptInputMessage, extraContext?: Record, ) => { const text = message.text.trim(); // 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, }), ); // Create optimistic human message (shown immediately) const optimisticHumanMsg: Message = { type: "human", id: `opt-human-${Date.now()}`, content: text ? [{ type: "text", text }] : "", additional_kwargs: optimisticFiles.length > 0 ? { files: optimisticFiles } : {}, }; const newOptimistic: Message[] = [optimisticHumanMsg]; if (optimisticFiles.length > 0) { // 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); _handleStart(threadId); let uploadedFileInfo: UploadedFileInfo[] = []; try { // Upload files first if any if (message.files && message.files.length > 0) { try { // Convert FileUIPart to File objects by fetching blob URLs const filePromises = message.files.map(async (fileUIPart) => { if (fileUIPart.url && fileUIPart.filename) { try { // Fetch the blob URL to get the file data const response = await fetch(fileUIPart.url); const blob = await response.blob(); // Create a File object from the blob return new File([blob], fileUIPart.filename, { type: fileUIPart.mediaType || blob.type, }); } catch (error) { console.error( `Failed to fetch file ${fileUIPart.filename}:`, error, ); return null; } } return null; }); 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: { files: uploadedFiles }, }, ...messages.slice(1), ]; } return messages; }); } } catch (error) { console.error("Failed to upload files:", error); const errorMessage = error instanceof Error ? error.message : "Failed to upload files."; toast.error(errorMessage); setOptimisticMessages([]); throw error; } } // 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, }), ); await thread.submit( { messages: [ { type: "human", content: [ { type: "text", text, }, ], additional_kwargs: filesForSubmit.length > 0 ? { files: filesForSubmit } : {}, }, ], }, { threadId: threadId, streamSubgraphs: true, streamResumable: true, streamMode: ["values", "messages-tuple", "custom"], config: { recursion_limit: 1000, }, context: { ...extraContext, ...context, thinking_enabled: context.mode !== "flash", is_plan_mode: context.mode === "pro" || context.mode === "ultra", subagent_enabled: context.mode === "ultra", thread_id: threadId, }, }, ); void queryClient.invalidateQueries({ queryKey: ["threads", "search"] }); } catch (error) { setOptimisticMessages([]); throw error; } }, [thread, _handleStart, t.uploads.uploadingFiles, context, queryClient], ); // Merge thread with optimistic messages for display const mergedThread = optimisticMessages.length > 0 ? ({ ...thread, messages: [...thread.messages, ...optimisticMessages], } as typeof thread) : thread; return [mergedThread, sendMessage] as const; } export function useThreads( params: Parameters[0] = { limit: 50, sortBy: "updated_at", sortOrder: "desc", select: ["thread_id", "updated_at", "values"], }, ) { const apiClient = getAPIClient(); return useQuery({ queryKey: ["threads", "search", params], queryFn: async () => { const response = await apiClient.threads.search(params); return response as AgentThread[]; }, refetchOnWindowFocus: false, }); } export function useDeleteThread() { const queryClient = useQueryClient(); const apiClient = getAPIClient(); return useMutation({ mutationFn: async ({ threadId }: { threadId: string }) => { await apiClient.threads.delete(threadId); }, onSuccess(_, { threadId }) { queryClient.setQueriesData( { queryKey: ["threads", "search"], exact: false, }, (oldData: Array) => { return oldData.filter((t) => t.thread_id !== threadId); }, ); }, }); } 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 }, }); }, onSuccess(_, { threadId, title }) { queryClient.setQueriesData( { queryKey: ["threads", "search"], exact: false, }, (oldData: Array) => { return oldData.map((t) => { if (t.thread_id === threadId) { return { ...t, values: { ...t.values, title, }, }; } return t; }); }, ); }, }); }