[feature] adds web_search and fetch_url tool calls

server/src/llm/streaming.ts

@@ -1,11 +1,12 @@
 import { performance } from "node:perf_hooks";
-import type OpenAI from "openai";
 import { prisma } from "../db.js";
 import { anthropicClient, openaiClient, xaiClient } from "./providers.js";
+import { buildToolLogMessageData, runToolAwareOpenAIChat, type ToolExecutionEvent } from "./chat-tools.js";
 import type { MultiplexRequest, Provider } from "./types.js";
 
 export type StreamEvent =
   | { type: "meta"; chatId: string; callId: string; provider: Provider; model: string }
+  | { type: "tool_call"; event: ToolExecutionEvent }
   | { type: "delta"; text: string }
   | { type: "done"; text: string; usage?: { inputTokens?: number; outputTokens?: number; totalTokens?: number } }
   | { type: "error"; message: string };
@@ -51,28 +52,37 @@ export async function* runMultiplexStream(req: MultiplexRequest): AsyncGenerator
   let text = "";
   let usage: StreamEvent extends any ? any : never;
   let raw: unknown = { streamed: true };
+  let toolMessages: ReturnType<typeof buildToolLogMessageData>[] = [];
 
   try {
     if (req.provider === "openai" || req.provider === "xai") {
       const client = req.provider === "openai" ? openaiClient() : xaiClient();
-
-      const stream = await client.chat.completions.create({
+      const toolEvents: ToolExecutionEvent[] = [];
+      const r = await runToolAwareOpenAIChat({
+        client,
         model: req.model,
-        messages: req.messages.map((m) => ({ role: m.role, content: m.content, name: m.name })) as any,
+        messages: req.messages,
         temperature: req.temperature,
-        max_tokens: req.maxTokens,
-        stream: true,
+        maxTokens: req.maxTokens,
+        onToolEvent: (event) => {
+          toolEvents.push(event);
+        },
+        logContext: {
+          provider: req.provider,
+          model: req.model,
+          chatId,
+        },
       });
-
-      for await (const chunk of stream as any as AsyncIterable<OpenAI.Chat.Completions.ChatCompletionChunk>) {
-        const delta = chunk.choices?.[0]?.delta?.content ?? "";
-        if (delta) {
-          text += delta;
-          yield { type: "delta", text: delta };
-        }
+      raw = r.raw;
+      text = r.text;
+      usage = r.usage;
+      toolMessages = toolEvents.map((event) => buildToolLogMessageData(chatId, event));
+      for (const event of toolEvents) {
+        yield { type: "tool_call", event };
+      }
+      if (text) {
+        yield { type: "delta", text };
       }
-
-      // no guaranteed usage in stream mode across providers; leave empty for now
     } else if (req.provider === "anthropic") {
       const client = anthropicClient();
 
@@ -110,17 +120,29 @@ export async function* runMultiplexStream(req: MultiplexRequest): AsyncGenerator
         }
         // some streams end with message_stop
       }
+      raw = { streamed: true, provider: "anthropic" };
     } else {
       throw new Error(`unknown provider: ${req.provider}`);
     }
 
     const latencyMs = Math.round(performance.now() - t0);
 
-    await prisma.$transaction([
-      prisma.message.create({
+    await prisma.$transaction(async (tx) => {
+      if (toolMessages.length) {
+        await tx.message.createMany({
+          data: toolMessages.map((message) => ({
+            chatId: message.chatId,
+            role: message.role as any,
+            content: message.content,
+            name: message.name,
+            metadata: message.metadata as any,
+          })),
+        });
+      }
+      await tx.message.create({
         data: { chatId, role: "assistant" as any, content: text },
-      }),
-      prisma.llmCall.update({
+      });
+      await tx.llmCall.update({
         where: { id: call.id },
         data: {
           response: raw as any,
@@ -129,8 +151,8 @@ export async function* runMultiplexStream(req: MultiplexRequest): AsyncGenerator
           outputTokens: usage?.outputTokens,
           totalTokens: usage?.totalTokens,
         },
-      }),
-    ]);
+      });
+    });
 
     yield { type: "done", text, usage };
   } catch (e: any) {