{/* Main Content */}
{/* Loading State */}
- {showLoader && (isLoading || isCreating) && (
+ {shouldShowLoader && (
-
- ;
+ completedStreams: Map;
+ activeSessions: Set;
+ streamCompleteCallbacks: Set;
+}
+
+interface ChatStoreActions {
+ startStream: (
+ sessionId: string,
+ message: string,
+ isUserMessage: boolean,
+ context?: { url: string; content: string },
+ onChunk?: (chunk: StreamChunk) => void,
+ ) => Promise;
+ stopStream: (sessionId: string) => void;
+ subscribeToStream: (
+ sessionId: string,
+ onChunk: (chunk: StreamChunk) => void,
+ skipReplay?: boolean,
+ ) => () => void;
+ getStreamStatus: (sessionId: string) => StreamStatus;
+ getCompletedStream: (sessionId: string) => StreamResult | undefined;
+ clearCompletedStream: (sessionId: string) => void;
+ isStreaming: (sessionId: string) => boolean;
+ registerActiveSession: (sessionId: string) => void;
+ unregisterActiveSession: (sessionId: string) => void;
+ isSessionActive: (sessionId: string) => boolean;
+ onStreamComplete: (callback: StreamCompleteCallback) => () => void;
+}
+
+type ChatStore = ChatStoreState & ChatStoreActions;
+
+function notifyStreamComplete(
+ callbacks: Set,
+ sessionId: string,
+) {
+ for (const callback of callbacks) {
+ try {
+ callback(sessionId);
+ } catch (err) {
+ console.warn("[ChatStore] Stream complete callback error:", err);
+ }
+ }
+}
+
+function cleanupExpiredStreams(
+ completedStreams: Map,
+): Map {
+ const now = Date.now();
+ const cleaned = new Map(completedStreams);
+ for (const [sessionId, result] of cleaned) {
+ if (now - result.completedAt > COMPLETED_STREAM_TTL) {
+ cleaned.delete(sessionId);
+ }
+ }
+ return cleaned;
+}
+
+export const useChatStore = create((set, get) => ({
+ activeStreams: new Map(),
+ completedStreams: new Map(),
+ activeSessions: new Set(),
+ streamCompleteCallbacks: new Set(),
+
+ startStream: async function startStream(
+ sessionId,
+ message,
+ isUserMessage,
+ context,
+ onChunk,
+ ) {
+ const state = get();
+ const newActiveStreams = new Map(state.activeStreams);
+ let newCompletedStreams = new Map(state.completedStreams);
+ const callbacks = state.streamCompleteCallbacks;
+
+ const existingStream = newActiveStreams.get(sessionId);
+ if (existingStream) {
+ existingStream.abortController.abort();
+ const normalizedStatus =
+ existingStream.status === "streaming"
+ ? "completed"
+ : existingStream.status;
+ const result: StreamResult = {
+ sessionId,
+ status: normalizedStatus,
+ chunks: existingStream.chunks,
+ completedAt: Date.now(),
+ error: existingStream.error,
+ };
+ newCompletedStreams.set(sessionId, result);
+ newActiveStreams.delete(sessionId);
+ newCompletedStreams = cleanupExpiredStreams(newCompletedStreams);
+ if (normalizedStatus === "completed" || normalizedStatus === "error") {
+ notifyStreamComplete(callbacks, sessionId);
+ }
+ }
+
+ const abortController = new AbortController();
+ const initialCallbacks = new Set<(chunk: StreamChunk) => void>();
+ if (onChunk) initialCallbacks.add(onChunk);
+
+ const stream: ActiveStream = {
+ sessionId,
+ abortController,
+ status: "streaming",
+ startedAt: Date.now(),
+ chunks: [],
+ onChunkCallbacks: initialCallbacks,
+ };
+
+ newActiveStreams.set(sessionId, stream);
+ set({
+ activeStreams: newActiveStreams,
+ completedStreams: newCompletedStreams,
+ });
+
+ try {
+ await executeStream(stream, message, isUserMessage, context);
+ } finally {
+ if (onChunk) stream.onChunkCallbacks.delete(onChunk);
+ if (stream.status !== "streaming") {
+ const currentState = get();
+ const finalActiveStreams = new Map(currentState.activeStreams);
+ let finalCompletedStreams = new Map(currentState.completedStreams);
+
+ const storedStream = finalActiveStreams.get(sessionId);
+ if (storedStream === stream) {
+ const result: StreamResult = {
+ sessionId,
+ status: stream.status,
+ chunks: stream.chunks,
+ completedAt: Date.now(),
+ error: stream.error,
+ };
+ finalCompletedStreams.set(sessionId, result);
+ finalActiveStreams.delete(sessionId);
+ finalCompletedStreams = cleanupExpiredStreams(finalCompletedStreams);
+ set({
+ activeStreams: finalActiveStreams,
+ completedStreams: finalCompletedStreams,
+ });
+ if (stream.status === "completed" || stream.status === "error") {
+ notifyStreamComplete(
+ currentState.streamCompleteCallbacks,
+ sessionId,
+ );
+ }
+ }
+ }
+ }
+ },
+
+ stopStream: function stopStream(sessionId) {
+ const state = get();
+ const stream = state.activeStreams.get(sessionId);
+ if (!stream) return;
+
+ stream.abortController.abort();
+ stream.status = "completed";
+
+ const newActiveStreams = new Map(state.activeStreams);
+ let newCompletedStreams = new Map(state.completedStreams);
+
+ const result: StreamResult = {
+ sessionId,
+ status: stream.status,
+ chunks: stream.chunks,
+ completedAt: Date.now(),
+ error: stream.error,
+ };
+ newCompletedStreams.set(sessionId, result);
+ newActiveStreams.delete(sessionId);
+ newCompletedStreams = cleanupExpiredStreams(newCompletedStreams);
+
+ set({
+ activeStreams: newActiveStreams,
+ completedStreams: newCompletedStreams,
+ });
+
+ notifyStreamComplete(state.streamCompleteCallbacks, sessionId);
+ },
+
+ subscribeToStream: function subscribeToStream(
+ sessionId,
+ onChunk,
+ skipReplay = false,
+ ) {
+ const state = get();
+ const stream = state.activeStreams.get(sessionId);
+
+ if (stream) {
+ if (!skipReplay) {
+ for (const chunk of stream.chunks) {
+ onChunk(chunk);
+ }
+ }
+
+ stream.onChunkCallbacks.add(onChunk);
+
+ return function unsubscribe() {
+ stream.onChunkCallbacks.delete(onChunk);
+ };
+ }
+
+ return function noop() {};
+ },
+
+ getStreamStatus: function getStreamStatus(sessionId) {
+ const { activeStreams, completedStreams } = get();
+
+ const active = activeStreams.get(sessionId);
+ if (active) return active.status;
+
+ const completed = completedStreams.get(sessionId);
+ if (completed) return completed.status;
+
+ return "idle";
+ },
+
+ getCompletedStream: function getCompletedStream(sessionId) {
+ return get().completedStreams.get(sessionId);
+ },
+
+ clearCompletedStream: function clearCompletedStream(sessionId) {
+ const state = get();
+ if (!state.completedStreams.has(sessionId)) return;
+
+ const newCompletedStreams = new Map(state.completedStreams);
+ newCompletedStreams.delete(sessionId);
+ set({ completedStreams: newCompletedStreams });
+ },
+
+ isStreaming: function isStreaming(sessionId) {
+ const stream = get().activeStreams.get(sessionId);
+ return stream?.status === "streaming";
+ },
+
+ registerActiveSession: function registerActiveSession(sessionId) {
+ const state = get();
+ if (state.activeSessions.has(sessionId)) return;
+
+ const newActiveSessions = new Set(state.activeSessions);
+ newActiveSessions.add(sessionId);
+ set({ activeSessions: newActiveSessions });
+ },
+
+ unregisterActiveSession: function unregisterActiveSession(sessionId) {
+ const state = get();
+ if (!state.activeSessions.has(sessionId)) return;
+
+ const newActiveSessions = new Set(state.activeSessions);
+ newActiveSessions.delete(sessionId);
+ set({ activeSessions: newActiveSessions });
+ },
+
+ isSessionActive: function isSessionActive(sessionId) {
+ return get().activeSessions.has(sessionId);
+ },
+
+ onStreamComplete: function onStreamComplete(callback) {
+ const state = get();
+ const newCallbacks = new Set(state.streamCompleteCallbacks);
+ newCallbacks.add(callback);
+ set({ streamCompleteCallbacks: newCallbacks });
+
+ return function unsubscribe() {
+ const currentState = get();
+ const cleanedCallbacks = new Set(currentState.streamCompleteCallbacks);
+ cleanedCallbacks.delete(callback);
+ set({ streamCompleteCallbacks: cleanedCallbacks });
+ };
+ },
+}));
diff --git a/autogpt_platform/frontend/src/components/contextual/Chat/chat-types.ts b/autogpt_platform/frontend/src/components/contextual/Chat/chat-types.ts
new file mode 100644
index 0000000000..8c8aa7b704
--- /dev/null
+++ b/autogpt_platform/frontend/src/components/contextual/Chat/chat-types.ts
@@ -0,0 +1,94 @@
+import type { ToolArguments, ToolResult } from "@/types/chat";
+
+export type StreamStatus = "idle" | "streaming" | "completed" | "error";
+
+export interface StreamChunk {
+ type:
+ | "text_chunk"
+ | "text_ended"
+ | "tool_call"
+ | "tool_call_start"
+ | "tool_response"
+ | "login_needed"
+ | "need_login"
+ | "credentials_needed"
+ | "error"
+ | "usage"
+ | "stream_end";
+ timestamp?: string;
+ content?: string;
+ message?: string;
+ code?: string;
+ details?: Record;
+ tool_id?: string;
+ tool_name?: string;
+ arguments?: ToolArguments;
+ result?: ToolResult;
+ success?: boolean;
+ idx?: number;
+ session_id?: string;
+ agent_info?: {
+ graph_id: string;
+ name: string;
+ trigger_type: string;
+ };
+ provider?: string;
+ provider_name?: string;
+ credential_type?: string;
+ scopes?: string[];
+ title?: string;
+ [key: string]: unknown;
+}
+
+export type VercelStreamChunk =
+ | { type: "start"; messageId: string }
+ | { type: "finish" }
+ | { type: "text-start"; id: string }
+ | { type: "text-delta"; id: string; delta: string }
+ | { type: "text-end"; id: string }
+ | { type: "tool-input-start"; toolCallId: string; toolName: string }
+ | {
+ type: "tool-input-available";
+ toolCallId: string;
+ toolName: string;
+ input: Record;
+ }
+ | {
+ type: "tool-output-available";
+ toolCallId: string;
+ toolName?: string;
+ output: unknown;
+ success?: boolean;
+ }
+ | {
+ type: "usage";
+ promptTokens: number;
+ completionTokens: number;
+ totalTokens: number;
+ }
+ | {
+ type: "error";
+ errorText: string;
+ code?: string;
+ details?: Record;
+ };
+
+export interface ActiveStream {
+ sessionId: string;
+ abortController: AbortController;
+ status: StreamStatus;
+ startedAt: number;
+ chunks: StreamChunk[];
+ error?: Error;
+ onChunkCallbacks: Set<(chunk: StreamChunk) => void>;
+}
+
+export interface StreamResult {
+ sessionId: string;
+ status: StreamStatus;
+ chunks: StreamChunk[];
+ completedAt: number;
+ error?: Error;
+}
+
+export type StreamCompleteCallback = (sessionId: string) => void;
diff --git a/autogpt_platform/frontend/src/components/contextual/Chat/components/ChatContainer/ChatContainer.tsx b/autogpt_platform/frontend/src/components/contextual/Chat/components/ChatContainer/ChatContainer.tsx
index 17748f8dbc..dec221338a 100644
--- a/autogpt_platform/frontend/src/components/contextual/Chat/components/ChatContainer/ChatContainer.tsx
+++ b/autogpt_platform/frontend/src/components/contextual/Chat/components/ChatContainer/ChatContainer.tsx
@@ -4,6 +4,7 @@ import { Text } from "@/components/atoms/Text/Text";
import { Dialog } from "@/components/molecules/Dialog/Dialog";
import { useBreakpoint } from "@/lib/hooks/useBreakpoint";
import { cn } from "@/lib/utils";
+import { GlobeHemisphereEastIcon } from "@phosphor-icons/react";
import { useEffect } from "react";
import { ChatInput } from "../ChatInput/ChatInput";
import { MessageList } from "../MessageList/MessageList";
@@ -15,6 +16,7 @@ export interface ChatContainerProps {
initialPrompt?: string;
className?: string;
onStreamingChange?: (isStreaming: boolean) => void;
+ onOperationStarted?: () => void;
}
export function ChatContainer({
@@ -23,6 +25,7 @@ export function ChatContainer({
initialPrompt,
className,
onStreamingChange,
+ onOperationStarted,
}: ChatContainerProps) {
const {
messages,
@@ -37,6 +40,7 @@ export function ChatContainer({
sessionId,
initialMessages,
initialPrompt,
+ onOperationStarted,
});
useEffect(() => {
@@ -55,24 +59,37 @@ export function ChatContainer({
)}
>
+ }
controlled={{
isOpen: isRegionBlockedModalOpen,
set: handleRegionModalOpenChange,
}}
onClose={handleRegionModalClose}
+ styling={{ maxWidth: 550, width: "100%", minWidth: "auto" }}
>
-
)}
{/* Error State */}
- {error && !isLoading && (
+ {error && !isLoading && !isSwitchingSession && (
+
- Loading your chats...
+ {isSwitchingSession
+ ? "Switching chat..."
+ : "Loading your chat..."}
+
- This model is not available in your region. Please connect via VPN
- and try again.
+ The Autogpt AI model is not available in your region or your
+ connection is blocking it. Please try again with a different
+ connection.
-
+
diff --git a/autogpt_platform/frontend/src/components/contextual/Chat/components/ChatContainer/createStreamEventDispatcher.ts b/autogpt_platform/frontend/src/components/contextual/Chat/components/ChatContainer/createStreamEventDispatcher.ts
index 791cf046d5..82e9b05e88 100644
--- a/autogpt_platform/frontend/src/components/contextual/Chat/components/ChatContainer/createStreamEventDispatcher.ts
+++ b/autogpt_platform/frontend/src/components/contextual/Chat/components/ChatContainer/createStreamEventDispatcher.ts
@@ -1,5 +1,5 @@
import { toast } from "sonner";
-import { StreamChunk } from "../../useChatStream";
+import type { StreamChunk } from "../../chat-types";
import type { HandlerDependencies } from "./handlers";
import {
handleError,
diff --git a/autogpt_platform/frontend/src/components/contextual/Chat/components/ChatContainer/handlers.ts b/autogpt_platform/frontend/src/components/contextual/Chat/components/ChatContainer/handlers.ts
index 96198a0386..f3cac01f96 100644
--- a/autogpt_platform/frontend/src/components/contextual/Chat/components/ChatContainer/handlers.ts
+++ b/autogpt_platform/frontend/src/components/contextual/Chat/components/ChatContainer/handlers.ts
@@ -22,6 +22,7 @@ export interface HandlerDependencies {
setIsStreamingInitiated: Dispatch>;
setIsRegionBlockedModalOpen: Dispatch>;
sessionId: string;
+ onOperationStarted?: () => void;
}
export function isRegionBlockedError(chunk: StreamChunk): boolean {
@@ -48,6 +49,15 @@ export function handleTextEnded(
const completedText = deps.streamingChunksRef.current.join("");
if (completedText.trim()) {
deps.setMessages((prev) => {
+ // Check if this exact message already exists to prevent duplicates
+ const exists = prev.some(
+ (msg) =>
+ msg.type === "message" &&
+ msg.role === "assistant" &&
+ msg.content === completedText,
+ );
+ if (exists) return prev;
+
const assistantMessage: ChatMessageData = {
type: "message",
role: "assistant",
@@ -154,6 +164,11 @@ export function handleToolResponse(
}
return;
}
+ // Trigger polling when operation_started is received
+ if (responseMessage.type === "operation_started") {
+ deps.onOperationStarted?.();
+ }
+
deps.setMessages((prev) => {
const toolCallIndex = prev.findIndex(
(msg) => msg.type === "tool_call" && msg.toolId === chunk.tool_id,
@@ -203,13 +218,24 @@ export function handleStreamEnd(
]);
}
if (completedContent.trim()) {
- const assistantMessage: ChatMessageData = {
- type: "message",
- role: "assistant",
- content: completedContent,
- timestamp: new Date(),
- };
- deps.setMessages((prev) => [...prev, assistantMessage]);
+ deps.setMessages((prev) => {
+ // Check if this exact message already exists to prevent duplicates
+ const exists = prev.some(
+ (msg) =>
+ msg.type === "message" &&
+ msg.role === "assistant" &&
+ msg.content === completedContent,
+ );
+ if (exists) return prev;
+
+ const assistantMessage: ChatMessageData = {
+ type: "message",
+ role: "assistant",
+ content: completedContent,
+ timestamp: new Date(),
+ };
+ return [...prev, assistantMessage];
+ });
}
deps.setStreamingChunks([]);
deps.streamingChunksRef.current = [];
diff --git a/autogpt_platform/frontend/src/components/contextual/Chat/components/ChatContainer/helpers.ts b/autogpt_platform/frontend/src/components/contextual/Chat/components/ChatContainer/helpers.ts
index 9d51003a93..e744c9bc34 100644
--- a/autogpt_platform/frontend/src/components/contextual/Chat/components/ChatContainer/helpers.ts
+++ b/autogpt_platform/frontend/src/components/contextual/Chat/components/ChatContainer/helpers.ts
@@ -1,7 +1,118 @@
+import type { SessionDetailResponse } from "@/app/api/__generated__/models/sessionDetailResponse";
import { SessionKey, sessionStorage } from "@/services/storage/session-storage";
import type { ToolResult } from "@/types/chat";
import type { ChatMessageData } from "../ChatMessage/useChatMessage";
+export function processInitialMessages(
+ initialMessages: SessionDetailResponse["messages"],
+): ChatMessageData[] {
+ const processedMessages: ChatMessageData[] = [];
+ const toolCallMap = new Map();
+
+ for (const msg of initialMessages) {
+ if (!isValidMessage(msg)) {
+ console.warn("Invalid message structure from backend:", msg);
+ continue;
+ }
+
+ let content = String(msg.content || "");
+ const role = String(msg.role || "assistant").toLowerCase();
+ const toolCalls = msg.tool_calls;
+ const timestamp = msg.timestamp
+ ? new Date(msg.timestamp as string)
+ : undefined;
+
+ if (role === "user") {
+ content = removePageContext(content);
+ if (!content.trim()) continue;
+ processedMessages.push({
+ type: "message",
+ role: "user",
+ content,
+ timestamp,
+ });
+ continue;
+ }
+
+ if (role === "assistant") {
+ content = content
+ .replace(/[\s\S]*?<\/thinking>/gi, "")
+ .replace(/[\s\S]*?<\/internal_reasoning>/gi, "")
+ .trim();
+
+ if (toolCalls && isToolCallArray(toolCalls) && toolCalls.length > 0) {
+ for (const toolCall of toolCalls) {
+ const toolName = toolCall.function.name;
+ const toolId = toolCall.id;
+ toolCallMap.set(toolId, toolName);
+
+ try {
+ const args = JSON.parse(toolCall.function.arguments || "{}");
+ processedMessages.push({
+ type: "tool_call",
+ toolId,
+ toolName,
+ arguments: args,
+ timestamp,
+ });
+ } catch (err) {
+ console.warn("Failed to parse tool call arguments:", err);
+ processedMessages.push({
+ type: "tool_call",
+ toolId,
+ toolName,
+ arguments: {},
+ timestamp,
+ });
+ }
+ }
+ if (content.trim()) {
+ processedMessages.push({
+ type: "message",
+ role: "assistant",
+ content,
+ timestamp,
+ });
+ }
+ } else if (content.trim()) {
+ processedMessages.push({
+ type: "message",
+ role: "assistant",
+ content,
+ timestamp,
+ });
+ }
+ continue;
+ }
+
+ if (role === "tool") {
+ const toolCallId = (msg.tool_call_id as string) || "";
+ const toolName = toolCallMap.get(toolCallId) || "unknown";
+ const toolResponse = parseToolResponse(
+ content,
+ toolCallId,
+ toolName,
+ timestamp,
+ );
+ if (toolResponse) {
+ processedMessages.push(toolResponse);
+ }
+ continue;
+ }
+
+ if (content.trim()) {
+ processedMessages.push({
+ type: "message",
+ role: role as "user" | "assistant" | "system",
+ content,
+ timestamp,
+ });
+ }
+ }
+
+ return processedMessages;
+}
+
export function hasSentInitialPrompt(sessionId: string): boolean {
try {
const sent = JSON.parse(
@@ -193,6 +304,7 @@ export function parseToolResponse(
if (isAgentArray(agentsData)) {
return {
type: "agent_carousel",
+ toolId,
toolName: "agent_carousel",
agents: agentsData,
totalCount: parsedResult.total_count as number | undefined,
@@ -205,6 +317,7 @@ export function parseToolResponse(
if (responseType === "execution_started") {
return {
type: "execution_started",
+ toolId,
toolName: "execution_started",
executionId: (parsedResult.execution_id as string) || "",
agentName: (parsedResult.graph_name as string) || undefined,
@@ -213,6 +326,58 @@ export function parseToolResponse(
timestamp: timestamp || new Date(),
};
}
+ if (responseType === "clarification_needed") {
+ return {
+ type: "clarification_needed",
+ toolName,
+ questions:
+ (parsedResult.questions as Array<{
+ question: string;
+ keyword: string;
+ example?: string;
+ }>) || [],
+ message:
+ (parsedResult.message as string) ||
+ "I need more information to proceed.",
+ sessionId: (parsedResult.session_id as string) || "",
+ timestamp: timestamp || new Date(),
+ };
+ }
+ if (responseType === "operation_started") {
+ return {
+ type: "operation_started",
+ toolName: (parsedResult.tool_name as string) || toolName,
+ toolId,
+ operationId: (parsedResult.operation_id as string) || "",
+ message:
+ (parsedResult.message as string) ||
+ "Operation started. You can close this tab.",
+ timestamp: timestamp || new Date(),
+ };
+ }
+ if (responseType === "operation_pending") {
+ return {
+ type: "operation_pending",
+ toolName: (parsedResult.tool_name as string) || toolName,
+ toolId,
+ operationId: (parsedResult.operation_id as string) || "",
+ message:
+ (parsedResult.message as string) ||
+ "Operation in progress. Please wait...",
+ timestamp: timestamp || new Date(),
+ };
+ }
+ if (responseType === "operation_in_progress") {
+ return {
+ type: "operation_in_progress",
+ toolName: (parsedResult.tool_name as string) || toolName,
+ toolCallId: (parsedResult.tool_call_id as string) || toolId,
+ message:
+ (parsedResult.message as string) ||
+ "Operation already in progress. Please wait...",
+ timestamp: timestamp || new Date(),
+ };
+ }
if (responseType === "need_login") {
return {
type: "login_needed",
diff --git a/autogpt_platform/frontend/src/components/contextual/Chat/components/ChatContainer/useChatContainer.ts b/autogpt_platform/frontend/src/components/contextual/Chat/components/ChatContainer/useChatContainer.ts
index 42dd04670d..46f384d055 100644
--- a/autogpt_platform/frontend/src/components/contextual/Chat/components/ChatContainer/useChatContainer.ts
+++ b/autogpt_platform/frontend/src/components/contextual/Chat/components/ChatContainer/useChatContainer.ts
@@ -1,5 +1,6 @@
import type { SessionDetailResponse } from "@/app/api/__generated__/models/sessionDetailResponse";
-import { useCallback, useEffect, useMemo, useRef, useState } from "react";
+import { useEffect, useMemo, useRef, useState } from "react";
+import { useChatStore } from "../../chat-store";
import { toast } from "sonner";
import { useChatStream } from "../../useChatStream";
import { usePageContext } from "../../usePageContext";
@@ -9,23 +10,44 @@ import {
createUserMessage,
filterAuthMessages,
hasSentInitialPrompt,
- isToolCallArray,
- isValidMessage,
markInitialPromptSent,
- parseToolResponse,
- removePageContext,
+ processInitialMessages,
} from "./helpers";
+// Helper to generate deduplication key for a message
+function getMessageKey(msg: ChatMessageData): string {
+ if (msg.type === "message") {
+ // Don't include timestamp - dedupe by role + content only
+ // This handles the case where local and server timestamps differ
+ // Server messages are authoritative, so duplicates from local state are filtered
+ return `msg:${msg.role}:${msg.content}`;
+ } else if (msg.type === "tool_call") {
+ return `toolcall:${msg.toolId}`;
+ } else if (msg.type === "tool_response") {
+ return `toolresponse:${(msg as any).toolId}`;
+ } else if (
+ msg.type === "operation_started" ||
+ msg.type === "operation_pending" ||
+ msg.type === "operation_in_progress"
+ ) {
+ return `op:${(msg as any).toolId || (msg as any).operationId || (msg as any).toolCallId || ""}:${msg.toolName}`;
+ } else {
+ return `${msg.type}:${JSON.stringify(msg).slice(0, 100)}`;
+ }
+}
+
interface Args {
sessionId: string | null;
initialMessages: SessionDetailResponse["messages"];
initialPrompt?: string;
+ onOperationStarted?: () => void;
}
export function useChatContainer({
sessionId,
initialMessages,
initialPrompt,
+ onOperationStarted,
}: Args) {
const [messages, setMessages] = useState([]);
const [streamingChunks, setStreamingChunks] = useState([]);
@@ -41,11 +63,18 @@ export function useChatContainer({
sendMessage: sendStreamMessage,
stopStreaming,
} = useChatStream();
+ const activeStreams = useChatStore((s) => s.activeStreams);
+ const subscribeToStream = useChatStore((s) => s.subscribeToStream);
const isStreaming = isStreamingInitiated || hasTextChunks;
- useEffect(() => {
- if (sessionId !== previousSessionIdRef.current) {
- stopStreaming(previousSessionIdRef.current ?? undefined, true);
+ useEffect(
+ function handleSessionChange() {
+ if (sessionId === previousSessionIdRef.current) return;
+
+ const prevSession = previousSessionIdRef.current;
+ if (prevSession) {
+ stopStreaming(prevSession);
+ }
previousSessionIdRef.current = sessionId;
setMessages([]);
setStreamingChunks([]);
@@ -53,138 +82,11 @@ export function useChatContainer({
setHasTextChunks(false);
setIsStreamingInitiated(false);
hasResponseRef.current = false;
- }
- }, [sessionId, stopStreaming]);
- const allMessages = useMemo(() => {
- const processedInitialMessages: ChatMessageData[] = [];
- const toolCallMap = new Map();
+ if (!sessionId) return;
- for (const msg of initialMessages) {
- if (!isValidMessage(msg)) {
- console.warn("Invalid message structure from backend:", msg);
- continue;
- }
-
- let content = String(msg.content || "");
- const role = String(msg.role || "assistant").toLowerCase();
- const toolCalls = msg.tool_calls;
- const timestamp = msg.timestamp
- ? new Date(msg.timestamp as string)
- : undefined;
-
- if (role === "user") {
- content = removePageContext(content);
- if (!content.trim()) continue;
- processedInitialMessages.push({
- type: "message",
- role: "user",
- content,
- timestamp,
- });
- continue;
- }
-
- if (role === "assistant") {
- content = content
- .replace(/[\s\S]*?<\/thinking>/gi, "")
- .trim();
-
- if (toolCalls && isToolCallArray(toolCalls) && toolCalls.length > 0) {
- for (const toolCall of toolCalls) {
- const toolName = toolCall.function.name;
- const toolId = toolCall.id;
- toolCallMap.set(toolId, toolName);
-
- try {
- const args = JSON.parse(toolCall.function.arguments || "{}");
- processedInitialMessages.push({
- type: "tool_call",
- toolId,
- toolName,
- arguments: args,
- timestamp,
- });
- } catch (err) {
- console.warn("Failed to parse tool call arguments:", err);
- processedInitialMessages.push({
- type: "tool_call",
- toolId,
- toolName,
- arguments: {},
- timestamp,
- });
- }
- }
- if (content.trim()) {
- processedInitialMessages.push({
- type: "message",
- role: "assistant",
- content,
- timestamp,
- });
- }
- } else if (content.trim()) {
- processedInitialMessages.push({
- type: "message",
- role: "assistant",
- content,
- timestamp,
- });
- }
- continue;
- }
-
- if (role === "tool") {
- const toolCallId = (msg.tool_call_id as string) || "";
- const toolName = toolCallMap.get(toolCallId) || "unknown";
- const toolResponse = parseToolResponse(
- content,
- toolCallId,
- toolName,
- timestamp,
- );
- if (toolResponse) {
- processedInitialMessages.push(toolResponse);
- }
- continue;
- }
-
- if (content.trim()) {
- processedInitialMessages.push({
- type: "message",
- role: role as "user" | "assistant" | "system",
- content,
- timestamp,
- });
- }
- }
-
- return [...processedInitialMessages, ...messages];
- }, [initialMessages, messages]);
-
- const sendMessage = useCallback(
- async function sendMessage(
- content: string,
- isUserMessage: boolean = true,
- context?: { url: string; content: string },
- ) {
- if (!sessionId) {
- console.error("[useChatContainer] Cannot send message: no session ID");
- return;
- }
- setIsRegionBlockedModalOpen(false);
- if (isUserMessage) {
- const userMessage = createUserMessage(content);
- setMessages((prev) => [...filterAuthMessages(prev), userMessage]);
- } else {
- setMessages((prev) => filterAuthMessages(prev));
- }
- setStreamingChunks([]);
- streamingChunksRef.current = [];
- setHasTextChunks(false);
- setIsStreamingInitiated(true);
- hasResponseRef.current = false;
+ const activeStream = activeStreams.get(sessionId);
+ if (!activeStream || activeStream.status !== "streaming") return;
const dispatcher = createStreamEventDispatcher({
setHasTextChunks,
@@ -195,44 +97,170 @@ export function useChatContainer({
setIsRegionBlockedModalOpen,
sessionId,
setIsStreamingInitiated,
+ onOperationStarted,
});
- try {
- await sendStreamMessage(
- sessionId,
- content,
- dispatcher,
- isUserMessage,
- context,
- );
- } catch (err) {
- console.error("[useChatContainer] Failed to send message:", err);
- setIsStreamingInitiated(false);
-
- // Don't show error toast for AbortError (expected during cleanup)
- if (err instanceof Error && err.name === "AbortError") return;
-
- const errorMessage =
- err instanceof Error ? err.message : "Failed to send message";
- toast.error("Failed to send message", {
- description: errorMessage,
- });
- }
+ setIsStreamingInitiated(true);
+ const skipReplay = initialMessages.length > 0;
+ return subscribeToStream(sessionId, dispatcher, skipReplay);
},
- [sessionId, sendStreamMessage],
+ [
+ sessionId,
+ stopStreaming,
+ activeStreams,
+ subscribeToStream,
+ onOperationStarted,
+ ],
);
- const handleStopStreaming = useCallback(() => {
+ // Collect toolIds from completed tool results in initialMessages
+ // Used to filter out operation messages when their results arrive
+ const completedToolIds = useMemo(() => {
+ const processedInitial = processInitialMessages(initialMessages);
+ const ids = new Set();
+ for (const msg of processedInitial) {
+ if (
+ msg.type === "tool_response" ||
+ msg.type === "agent_carousel" ||
+ msg.type === "execution_started"
+ ) {
+ const toolId = (msg as any).toolId;
+ if (toolId) {
+ ids.add(toolId);
+ }
+ }
+ }
+ return ids;
+ }, [initialMessages]);
+
+ // Clean up local operation messages when their completed results arrive from polling
+ // This effect runs when completedToolIds changes (i.e., when polling brings new results)
+ useEffect(
+ function cleanupCompletedOperations() {
+ if (completedToolIds.size === 0) return;
+
+ setMessages((prev) => {
+ const filtered = prev.filter((msg) => {
+ if (
+ msg.type === "operation_started" ||
+ msg.type === "operation_pending" ||
+ msg.type === "operation_in_progress"
+ ) {
+ const toolId = (msg as any).toolId || (msg as any).toolCallId;
+ if (toolId && completedToolIds.has(toolId)) {
+ return false; // Remove - operation completed
+ }
+ }
+ return true;
+ });
+ // Only update state if something was actually filtered
+ return filtered.length === prev.length ? prev : filtered;
+ });
+ },
+ [completedToolIds],
+ );
+
+ // Combine initial messages from backend with local streaming messages,
+ // Server messages maintain correct order; only append truly new local messages
+ const allMessages = useMemo(() => {
+ const processedInitial = processInitialMessages(initialMessages);
+
+ // Build a set of keys from server messages for deduplication
+ const serverKeys = new Set();
+ for (const msg of processedInitial) {
+ serverKeys.add(getMessageKey(msg));
+ }
+
+ // Filter local messages: remove duplicates and completed operation messages
+ const newLocalMessages = messages.filter((msg) => {
+ // Remove operation messages for completed tools
+ if (
+ msg.type === "operation_started" ||
+ msg.type === "operation_pending" ||
+ msg.type === "operation_in_progress"
+ ) {
+ const toolId = (msg as any).toolId || (msg as any).toolCallId;
+ if (toolId && completedToolIds.has(toolId)) {
+ return false;
+ }
+ }
+ // Remove messages that already exist in server data
+ const key = getMessageKey(msg);
+ return !serverKeys.has(key);
+ });
+
+ // Server messages first (correct order), then new local messages
+ return [...processedInitial, ...newLocalMessages];
+ }, [initialMessages, messages, completedToolIds]);
+
+ async function sendMessage(
+ content: string,
+ isUserMessage: boolean = true,
+ context?: { url: string; content: string },
+ ) {
+ if (!sessionId) {
+ console.error("[useChatContainer] Cannot send message: no session ID");
+ return;
+ }
+ setIsRegionBlockedModalOpen(false);
+ if (isUserMessage) {
+ const userMessage = createUserMessage(content);
+ setMessages((prev) => [...filterAuthMessages(prev), userMessage]);
+ } else {
+ setMessages((prev) => filterAuthMessages(prev));
+ }
+ setStreamingChunks([]);
+ streamingChunksRef.current = [];
+ setHasTextChunks(false);
+ setIsStreamingInitiated(true);
+ hasResponseRef.current = false;
+
+ const dispatcher = createStreamEventDispatcher({
+ setHasTextChunks,
+ setStreamingChunks,
+ streamingChunksRef,
+ hasResponseRef,
+ setMessages,
+ setIsRegionBlockedModalOpen,
+ sessionId,
+ setIsStreamingInitiated,
+ onOperationStarted,
+ });
+
+ try {
+ await sendStreamMessage(
+ sessionId,
+ content,
+ dispatcher,
+ isUserMessage,
+ context,
+ );
+ } catch (err) {
+ console.error("[useChatContainer] Failed to send message:", err);
+ setIsStreamingInitiated(false);
+
+ if (err instanceof Error && err.name === "AbortError") return;
+
+ const errorMessage =
+ err instanceof Error ? err.message : "Failed to send message";
+ toast.error("Failed to send message", {
+ description: errorMessage,
+ });
+ }
+ }
+
+ function handleStopStreaming() {
stopStreaming();
setStreamingChunks([]);
streamingChunksRef.current = [];
setHasTextChunks(false);
setIsStreamingInitiated(false);
- }, [stopStreaming]);
+ }
const { capturePageContext } = usePageContext();
+ const sendMessageRef = useRef(sendMessage);
+ sendMessageRef.current = sendMessage;
- // Send initial prompt if provided (for new sessions from homepage)
useEffect(
function handleInitialPrompt() {
if (!initialPrompt || !sessionId) return;
@@ -241,15 +269,9 @@ export function useChatContainer({
markInitialPromptSent(sessionId);
const context = capturePageContext();
- sendMessage(initialPrompt, true, context);
+ sendMessageRef.current(initialPrompt, true, context);
},
- [
- initialPrompt,
- sessionId,
- initialMessages.length,
- sendMessage,
- capturePageContext,
- ],
+ [initialPrompt, sessionId, initialMessages.length, capturePageContext],
);
async function sendMessageWithContext(
diff --git a/autogpt_platform/frontend/src/components/contextual/Chat/components/ChatInput/ChatInput.tsx b/autogpt_platform/frontend/src/components/contextual/Chat/components/ChatInput/ChatInput.tsx
index 8cdecf0bf4..c45e8dc250 100644
--- a/autogpt_platform/frontend/src/components/contextual/Chat/components/ChatInput/ChatInput.tsx
+++ b/autogpt_platform/frontend/src/components/contextual/Chat/components/ChatInput/ChatInput.tsx
@@ -21,7 +21,7 @@ export function ChatInput({
className,
}: Props) {
const inputId = "chat-input";
- const { value, setValue, handleKeyDown, handleSend, hasMultipleLines } =
+ const { value, handleKeyDown, handleSubmit, handleChange, hasMultipleLines } =
useChatInput({
onSend,
disabled: disabled || isStreaming,
@@ -29,15 +29,6 @@ export function ChatInput({
inputId,
});
- function handleSubmit(e: React.FormEvent) {
- e.preventDefault();
- handleSend();
- }
-
- function handleChange(e: React.ChangeEvent) {
- setValue(e.target.value);
- }
-
return (