Enable real-time agent tool call streaming for all providers (#4279)

* WIP agentic tool call streaming
- OpenAI
- Anthropic
- Azure OpenAI

* WIP rest of providers EXCLUDES Bedrock and GenericOpenAI

* patch untooled complete/streaming to use chatCallback provider from provider class and not assume OpenAI client struct
example: Ollama

* modify ollama to function with its own overrides
normalize completion/stream outputs across providers/untooled

* dev build

* fix message sanization for anthropic agent streaming

* wip fix anthropic agentic streaming sanitization

* patch gemini, webgenui, generic aibitat providers + disable providers unable to test

* refactor anthropic aibitat provider for empty message and tool call formatting

* Add frontend missing prop check
update Azure for streaming support
update Gemini to streamting support on gemini-* models
generic OpenAI disable streaming
verify localAI support
verify NVIDIA Nim support

* DPAIS, remove temp from call, support streaming'

* remove 0 temp to remove possibility of bad temp error/500s/400s

* Patch condition where model is non-streamable and no tools are present or called resulting in the provider `handleFunctionCallChat` being called - which returns a string.

This would then fail in Untooled.complete since response would be a string and not the expected `response.choices?.[0]?.message`

Modified this line to handle both conditions for stream/non-streaming and tool presence or lack thereof

* Allow generic Openai to be streamable since using untooled it should work fine
honor disabled streaming for provider where that concern may apply for regular chats

* rename function and more gemini-specific function to gemini provider

* add comments for readability
.complete on azure should be non-streaming as this is the sync response

* migrate CometAPI, but disable as we cannot test

---------

Co-authored-by: shatfield4 <[email protected]>

Author: timothycarambat

.github/workflows/dev-build.yaml

@@ -6,7 +6,7 @@ concurrency:
 
 on:
   push:
-    branches: ['3999-chromium-flags'] # put your current branch to create a build. Core team only.
+    branches: ['agentic-streaming'] # put your current branch to create a build. Core team only.
     paths-ignore:
       - '**.md'
       - 'cloud-deployments/*'

frontend/src/components/WorkspaceChat/ChatContainer/ChatHistory/HistoricalMessage/index.jsx

@@ -226,6 +226,7 @@ const RenderChatContent = memo(
       );
     let thoughtChain = null;
     let msgToRender = message;
+    if (!message) return null;
 
     // If the message is a perfect thought chain, we can render it directly
     // Complete == open and close tags match perfectly.

frontend/src/components/WorkspaceChat/ChatContainer/ChatHistory/StatusResponse/index.jsx

@@ -4,11 +4,7 @@ import { CaretDown } from "@phosphor-icons/react";
 import AgentAnimation from "@/media/animations/agent-animation.webm";
 import AgentStatic from "@/media/animations/agent-static.png";
 
-export default function StatusResponse({
-  messages = [],
-  isThinking = false,
-  showCheckmark = false,
-}) {
+export default function StatusResponse({ messages = [], isThinking = false }) {
   const [isExpanded, setIsExpanded] = useState(false);
   const currentThought = messages[messages.length - 1];
   const previousThoughts = messages.slice(0, -1);

frontend/src/components/WorkspaceChat/ChatContainer/ChatHistory/index.jsx

@@ -174,10 +174,6 @@ export default function ChatHistory({
           key={`status-group-${index}`}
           messages={item}
           isThinking={!hasSubsequentMessages && lastMessageInfo.isAnimating}
-          showCheckmark={
-            hasSubsequentMessages ||
-            (!lastMessageInfo.isAnimating && !lastMessageInfo.isStatusResponse)
-          }
         />
       );
     },

frontend/src/components/WorkspaceChat/ChatContainer/index.jsx

@@ -234,6 +234,7 @@ export default function ChatContainer({ workspace, knownHistory = [] }) {
         const socket = new WebSocket(
           `${websocketURI()}/api/agent-invocation/${socketId}`
         );
+        socket.supportsAgentStreaming = false;
 
         window.addEventListener(ABORT_STREAM_EVENT, () => {
           window.dispatchEvent(new CustomEvent(AGENT_SESSION_END));
@@ -243,7 +244,7 @@ export default function ChatContainer({ workspace, knownHistory = [] }) {
         socket.addEventListener("message", (event) => {
           setLoadingResponse(true);
           try {
-            handleSocketResponse(event, setChatHistory);
+            handleSocketResponse(socket, event, setChatHistory);
           } catch (e) {
             console.error("Failed to parse data");
             window.dispatchEvent(new CustomEvent(AGENT_SESSION_END));

frontend/src/utils/chat/agent.js

@@ -12,6 +12,8 @@ const handledEvents = [
   "awaitingFeedback",
   "wssFailure",
   "rechartVisualize",
+  // Streaming events
+  "reportStreamEvent",
 ];
 
 export function websocketURI() {
@@ -20,13 +22,13 @@ export function websocketURI() {
   return `${wsProtocol}//${new URL(import.meta.env.VITE_API_BASE).host}`;
 }
 
-export default function handleSocketResponse(event, setChatHistory) {
+export default function handleSocketResponse(socket, event, setChatHistory) {
   const data = safeJsonParse(event.data, null);
   if (data === null) return;
 
   // No message type is defined then this is a generic message
   // that we need to print to the user as a system response
-  if (!data.hasOwnProperty("type")) {
+  if (!data.hasOwnProperty("type") && !socket.supportsAgentStreaming) {
     return setChatHistory((prev) => {
       return [
         ...prev.filter((msg) => !!msg.content),
@@ -46,6 +48,90 @@ export default function handleSocketResponse(event, setChatHistory) {
 
   if (!handledEvents.includes(data.type) || !data.content) return;
 
+  if (data.type === "reportStreamEvent") {
+    // Enable agent streaming for the next message so we can handle streaming or non-streaming responses
+    // If we get this message we know the provider supports agentic streaming
+    socket.supportsAgentStreaming = true;
+
+    return setChatHistory((prev) => {
+      if (data.content.type === "removeStatusResponse")
+        return [...prev.filter((msg) => msg.uuid !== data.content.uuid)];
+
+      const knownMessage = data.content.uuid
+        ? prev.find((msg) => msg.uuid === data.content.uuid)
+        : null;
+      if (!knownMessage) {
+        if (data.content.type === "fullTextResponse") {
+          return [
+            ...prev.filter((msg) => !!msg.content),
+            {
+              uuid: data.content.uuid,
+              type: "textResponse",
+              content: data.content.content,
+              role: "assistant",
+              sources: [],
+              closed: true,
+              error: null,
+              animate: false,
+              pending: false,
+            },
+          ];
+        }
+
+        return [
+          ...prev.filter((msg) => !!msg.content),
+          {
+            uuid: data.content.uuid,
+            type: "statusResponse",
+            content: data.content.content,
+            role: "assistant",
+            sources: [],
+            closed: true,
+            error: null,
+            animate: false,
+            pending: false,
+          },
+        ];
+      } else {
+        const { type, content, uuid } = data.content;
+        // For tool call invocations, we need to update the existing message entirely since it is accumulated
+        // and we dont know if the function will have arguments or not while streaming - so replace the existing message entirely
+        if (type === "toolCallInvocation") {
+          const knownMessage = prev.find((msg) => msg.uuid === uuid);
+          if (!knownMessage)
+            return [...prev, { uuid, type: "toolCallInvocation", content }]; // If the message is not known, add it to the end of the list
+          return [
+            ...prev.filter((msg) => msg.uuid !== uuid),
+            { ...knownMessage, content },
+          ]; // If the message is known, replace it with the new content
+        }
+
+        if (type === "textResponseChunk") {
+          return prev
+            .map((msg) =>
+              msg.uuid === uuid
+                ? {
+                    ...msg,
+                    type: "textResponse",
+                    content: msg.content + content,
+                  }
+                : msg?.content
+                  ? msg
+                  : null
+            )
+            .filter((msg) => !!msg);
+        }
+
+        // Generic text response - will be put in the agent thought bubble
+        return prev.map((msg) =>
+          msg.uuid === data.content.uuid
+            ? { ...msg, content: msg.content + data.content.content }
+            : msg
+        );
+      }
+    });
+  }
+
   if (data.type === "fileDownload") {
     saveAs(data.content.b64Content, data.content.filename ?? "unknown.txt");
     return;