🐛 fix: fix mcp server connect issue and refactor web search implement (#9694)

* add

* clean

* refactor

* refactor

* add test

* fix style

* refactor to improve search performance

* refactor types

* refactor types

* refactor types

* fix mcp retry issue

* add more tests

* fix test and types

* fix test

* fix desktop remote streamable http

* add local

* fix tests

* update

Author: arvinxx

packages/const/src/index.ts

@@ -2,6 +2,7 @@ export * from './auth';
 export * from './branding';
 export * from './currency';
 export * from './desktop';
+export * from './discover';
 export * from './guide';
 export * from './layoutTokens';
 export * from './message';

packages/context-engine/src/__tests__/pipeline.test.ts

@@ -243,8 +243,6 @@ describe('ContextEngine', () => {
 
       expect(result.stats.processedCount).toBe(2);
       expect(result.stats.totalDuration).toBeGreaterThanOrEqual(20);
-      expect(result.stats.processorDurations.p1).toBeGreaterThanOrEqual(10);
-      expect(result.stats.processorDurations.p2).toBeGreaterThanOrEqual(20);
     });
 
     it('should stop processing when aborted', async () => {

packages/database/src/models/plugin.ts

@@ -1,9 +1,8 @@
+import { LobeTool } from '@lobechat/types';
 import { and, desc, eq } from 'drizzle-orm';
 
-import { LobeChatDatabase } from '../type';
-import { LobeTool } from '@/types/tool';
-
 import { InstalledPluginItem, NewInstalledPlugin, userInstalledPlugins } from '../schemas';
+import { LobeChatDatabase } from '../type';
 
 export class PluginModel {
   private userId: string;

packages/model-runtime/src/index.ts

@@ -34,6 +34,7 @@ export { LobeZeroOneAI } from './providers/zeroone';
 export { LobeZhipuAI } from './providers/zhipu';
 export * from './types';
 export * from './types/error';
+export { consumeStreamUntilDone } from './utils/consumeStream';
 export { AgentRuntimeError } from './utils/createError';
 export { getModelPropertyWithFallback } from './utils/getFallbackModelProperty';
 export { getModelPricing } from './utils/getModelPricing';

packages/model-runtime/src/types/chat.ts

@@ -1,4 +1,4 @@
-import { ModelSpeed, ModelTokensUsage, ModelUsage } from '@/types/message';
+import { ModelSpeed, ModelTokensUsage, ModelUsage } from '@lobechat/types';
 
 import { MessageToolCall, MessageToolCallChunk } from './toolsCalling';
 
@@ -207,6 +207,9 @@ export interface ChatStreamCallbacks {
   onThinking?: (content: string) => Promise<void> | void;
   onToolsCalling?: (data: {
     chunk: MessageToolCallChunk[];
+    /**
+     * full tools calling array
+     */
     toolsCalling: MessageToolCall[];
   }) => Promise<void> | void;
   onUsage?: (usage: ModelTokensUsage) => Promise<void> | void;

packages/model-runtime/src/utils/consumeStream.test.ts

@@ -0,0 +1,89 @@
+import { describe, expect, it, vi } from 'vitest';
+
+import { consumeStreamUntilDone } from './consumeStream';
+
+describe('consumeStreamUntilDone', () => {
+  it('should consume a stream completely', async () => {
+    const chunks = ['chunk1', 'chunk2', 'chunk3'];
+    const stream = new ReadableStream({
+      start(controller) {
+        for (const chunk of chunks) {
+          controller.enqueue(new TextEncoder().encode(chunk));
+        }
+        controller.close();
+      },
+    });
+
+    const response = new Response(stream);
+    await consumeStreamUntilDone(response);
+
+    // Stream should be consumed (reader should be locked and released)
+    expect(response.body?.locked).toBe(false);
+  });
+
+  it('should handle response without body', async () => {
+    const response = new Response(null);
+    await expect(consumeStreamUntilDone(response)).resolves.toBeUndefined();
+  });
+
+  it('should release reader lock even when stream errors occur', async () => {
+    const stream = new ReadableStream({
+      start(controller) {
+        controller.error(new Error('Stream error'));
+      },
+    });
+
+    const response = new Response(stream);
+
+    await expect(consumeStreamUntilDone(response)).rejects.toThrow('Stream error');
+
+    // Reader lock should still be released
+    expect(response.body?.locked).toBe(false);
+  });
+
+  it('should handle empty stream', async () => {
+    const stream = new ReadableStream({
+      start(controller) {
+        controller.close();
+      },
+    });
+
+    const response = new Response(stream);
+    await expect(consumeStreamUntilDone(response)).resolves.toBeUndefined();
+  });
+
+  it('should consume stream with large number of chunks', async () => {
+    const chunkCount = 100;
+    const stream = new ReadableStream({
+      start(controller) {
+        for (let i = 0; i < chunkCount; i++) {
+          controller.enqueue(new TextEncoder().encode(`chunk${i}`));
+        }
+        controller.close();
+      },
+    });
+
+    const response = new Response(stream);
+    await expect(consumeStreamUntilDone(response)).resolves.toBeUndefined();
+    expect(response.body?.locked).toBe(false);
+  });
+
+  it('should ensure reader.releaseLock is called', async () => {
+    const stream = new ReadableStream({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode('test'));
+        controller.close();
+      },
+    });
+
+    const response = new Response(stream);
+    const reader = response.body!.getReader();
+    const releaseLockSpy = vi.spyOn(reader, 'releaseLock');
+    reader.releaseLock(); // Release the lock we just acquired
+
+    await consumeStreamUntilDone(response);
+
+    // The function should acquire a new reader and release it
+    expect(releaseLockSpy).toHaveBeenCalled();
+  });
+});

packages/model-runtime/src/utils/consumeStream.ts

@@ -0,0 +1,35 @@
+/**
+ * Consumes a Response stream completely to ensure all callbacks are executed
+ * @param response - The Response object with a ReadableStream body
+ * @returns Promise that resolves when the stream is fully consumed
+ *
+ * @example
+ * ```ts
+ * const response = await modelRuntime.chat(payload, {
+ *   callback: {
+ *     onText: async (text) => {
+ *       await saveToDatabase(text);
+ *     }
+ *   }
+ * });
+ *
+ * // Ensure all callbacks complete before proceeding
+ * await consumeStreamUntilDone(response);
+ * ```
+ */
+export async function consumeStreamUntilDone(response: Response): Promise<void> {
+  if (!response.body) {
+    return;
+  }
+
+  const reader = response.body.getReader();
+  try {
+    // eslint-disable-next-line no-constant-condition
+    while (true) {
+      const { done } = await reader.read();
+      if (done) break;
+    }
+  } finally {
+    reader.releaseLock();
+  }
+}

packages/types/src/aiProvider.ts

@@ -6,7 +6,6 @@ export type ResponseAnimation =
   | {
       speed?: number;
       text?: ResponseAnimationStyle;
-      toolsCalling?: ResponseAnimationStyle;
     }
   | ResponseAnimationStyle;
 

packages/types/src/index.ts

@@ -16,6 +16,7 @@ export * from './knowledgeBase';
 export * from './llm';
 export * from './message';
 export * from './meta';
+export * from './plugins';
 export * from './rag';
 export * from './search';
 export * from './serverConfig';

packages/types/src/message/tools.ts

@@ -2,7 +2,7 @@ import { IPluginErrorType } from '@lobehub/chat-plugin-sdk';
 import type { PartialDeep } from 'type-fest';
 import { z } from 'zod';
 
-import { LobeToolRenderType } from '@/types/tool';
+import { LobeToolRenderType } from '../tool';
 
 export interface ChatPluginPayload {
   apiName: string;