fix(mcp): fix SSE header timeout, 405 fallback matching, and parseSSE chunking

hopelynd · hopelynd · commit 4dcec7d075fa · 2026-08-13T18:32:54.000+08:00
diff --git a/packages/core/src/client/mcp-sse.ts b/packages/core/src/client/mcp-sse.ts
@@ -24,7 +24,7 @@ type PendingResolver = {
   reject: (reason: unknown) => void;
 };
 
-/** 用字符串键匹配 JSON-RPC id（兼容 number / string 回传）。 */
+/** Match JSON-RPC ids with string keys (number or string echo from server). */
 function pendingKey(id: number | string): string {
   return String(id);
 }
@@ -41,7 +41,7 @@ export class McpSseClient {
   private resolveEndpoint: (() => void) | undefined;
   private rejectEndpoint: ((reason: unknown) => void) | undefined;
   private closed = false;
-  /** SSE GET 已结束（非主动 close）时置位，后续 RPC 立即失败。 */
+  /** Set when the SSE GET ends without an intentional close(); later RPCs fail fast. */
   private streamEnded = false;
 
   constructor(deps: HttpDeps, sseUrl: string, authToken?: string) {
@@ -114,8 +114,15 @@ export class McpSseClient {
   private async openSse(): Promise<void> {
     if (this.abortController) return;
 
-    // Keep the GET open until close(); timeouts apply only to endpoint wait / per-RPC.
+    // use shared abortController：header wait use timer abort；after getting header, clearTimeout,
+    // the long-lived stream is only ended by close()/session abort (compatible with Node 18, no AbortSignal.any).
     this.abortController = new AbortController();
+    const timeoutMs = this.deps.settings.timeout * 1000;
+    let headerTimedOut = false;
+    const headerTimer = setTimeout(() => {
+      headerTimedOut = true;
+      this.abortController?.abort();
+    }, timeoutMs);
 
     const headers: Record<string, string> = {
       Accept: "text/event-stream",
@@ -130,11 +137,28 @@ export class McpSseClient {
       console.error(`> GET ${this.sseUrl}`);
     }
 
-    const response = await fetch(this.sseUrl, {
-      method: "GET",
-      headers,
-      signal: this.abortController.signal,
-    });
+    let response: Response;
+    try {
+      response = await fetch(this.sseUrl, {
+        method: "GET",
+        headers,
+        signal: this.abortController.signal,
+      });
+    } catch (error) {
+      clearTimeout(headerTimer);
+      if (this.closed) {
+        throw new BailianError("MCP SSE session closed.", ExitCode.GENERAL);
+      }
+      if (headerTimedOut) {
+        throw new BailianError("MCP SSE timed out waiting for response headers.", ExitCode.TIMEOUT);
+      }
+      throw new BailianError(
+        `MCP SSE request failed: ${error instanceof Error ? error.message : String(error)}`,
+        ExitCode.NETWORK,
+      );
+    }
+    // 已收到响应头：取消 header 等待，后续仅由 abortController 结束流。
+    clearTimeout(headerTimer);
 
     if (this.deps.settings.verbose) {
       console.error(`< ${response.status} ${response.statusText}`);
@@ -148,9 +172,8 @@ export class McpSseClient {
       } catch {
         /* ignore */
       }
-      const error = new BailianError(errMsg, ExitCode.GENERAL);
-      this.rejectEndpoint?.(error);
-      throw error;
+      // Throw only — do not rejectEndpoint; this path never awaits endpointReady.
+      throw new BailianError(errMsg, ExitCode.GENERAL);
     }
 
     void this.consumeSse(response).catch((error) => {
@@ -163,13 +186,12 @@ export class McpSseClient {
               ExitCode.GENERAL,
             );
       this.rejectEndpoint?.(reason);
-      // consumeSse 在正常结束路径已 markStreamEnded；此处覆盖解析/读取异常。
+      // consumeSse already markStreamEnded on a clean end; cover parse/read failures here.
       if (!this.streamEnded) {
         this.markStreamEnded(reason);
       }
     });
 
-    const timeoutMs = this.deps.settings.timeout * 1000;
     const endpointTimeout = cancellableTimeoutReject(
       timeoutMs,
       "MCP SSE timed out waiting for endpoint event.",
@@ -185,7 +207,7 @@ export class McpSseClient {
     for await (const event of parseSSE(response)) {
       if (this.closed) break;
 
-      // 规范要求首事件为 event: endpoint；不接受无名事件以免误把 JSON 当 URL。
+      // Spec requires event: endpoint; ignore unnamed events so JSON is not treated as a URL.
       if (event.event === "endpoint") {
         const raw = event.data.trim();
         if (!raw) continue;
@@ -197,7 +219,7 @@ export class McpSseClient {
         continue;
       }
 
-      // 缺省 event 类型在 SSE 中等同 message。
+      // Omitted SSE event type defaults to "message".
       if (event.event === "message" || event.event === undefined) {
         let payload: JsonRpcResponse;
         try {
@@ -225,8 +247,8 @@ export class McpSseClient {
       throw error;
     }
 
-    // 已拿到 endpoint 后流仍结束：标记会话死亡并唤醒 pending；不再 throw，
-    // 避免 void consumeSse().catch 之外再冒出未处理 rejection。
+    // Stream ended after endpoint: mark session dead and wake pending; do not throw,
+    // so void consumeSse().catch does not surface an extra unhandled rejection.
     this.markStreamEnded(new BailianError("MCP SSE stream ended unexpectedly.", ExitCode.GENERAL));
   }
 
@@ -248,7 +270,7 @@ export class McpSseClient {
     const responsePromise = new Promise<JsonRpcResponse>((resolve, reject) => {
       this.pending.set(key, { resolve, reject });
     });
-    // 流可能在 Promise.race 之前结束并 reject pending，先挂上 catch 避免 unhandledRejection。
+    // Stream may end and reject pending before Promise.race; attach catch to avoid unhandledRejection.
     void responsePromise.catch(() => undefined);
     const responseTimeout = cancellableTimeoutReject(
       timeoutMs,
diff --git a/packages/core/src/client/mcp.ts b/packages/core/src/client/mcp.ts
@@ -70,20 +70,20 @@ export function bailianMcpSsePath(serverCode: string): string {
 
 /**
  * True when Streamable HTTP is unsupported and classic SSE fallback should be tried.
- * 以 HTTP 405 为准，不依赖服务端英文文案（避免文案变更导致降级失效）。
- * Bailian 的 404（未开通）不在此列，避免误降级。
+ * Match HTTP wrapper text `MCP request failed: 405` only — not JSON-RPC `MCP error (405)`.
+ * Bailian HTTP 404 (not activated) is intentionally excluded.
  */
 export function isStreamableHttpUnsupported(error: unknown): boolean {
   if (!(error instanceof BailianError)) return false;
-  return /405\b/i.test(error.message);
+  return /MCP request failed:\s*405\b/i.test(error.message);
 }
 
 /**
- * `--url` 覆盖时的 SSE 降级条件（官方 backwards-compat：同 URL 上 405/404 后尝试 GET SSE）。
+ * SSE fallback for `--url` (official backwards-compat: same URL, HTTP 405/404 then GET SSE).
  */
 export function isUrlOverrideSseFallbackCandidate(error: unknown): boolean {
   if (!(error instanceof BailianError)) return false;
-  return /405\b/i.test(error.message) || /404\b/i.test(error.message);
+  return /MCP request failed:\s*(405|404)\b/i.test(error.message);
 }
 
 export type McpConnectedClient = {
@@ -242,7 +242,7 @@ export class McpClient {
   }
 
   /**
-   * 按 Content-Type 读取 JSON-RPC 响应：支持 application/json 与 text/event-stream。
+   * Read a JSON-RPC response by Content-Type: application/json or text/event-stream.
    */
   private async readJsonRpcResponse(
     response: Response,
diff --git a/packages/core/src/client/stream.ts b/packages/core/src/client/stream.ts
@@ -20,6 +20,8 @@ export async function* parseSSE(response: Response): AsyncGenerator<ServerSentEv
   const MAX_SSE_BUFFER = 16 * 1024 * 1024; // 16 MiB
 
   try {
+    let event: Partial<ServerSentEvent> = {};
+
     while (true) {
       const { done, value } = await reader.read();
       if (done) break;
@@ -32,8 +34,6 @@ export async function* parseSSE(response: Response): AsyncGenerator<ServerSentEv
       const lines = buffer.split("\n");
       buffer = lines.pop() || "";
 
-      let event: Partial<ServerSentEvent> = {};
-
       for (const line of lines) {
         if (line === "") {
           if (event.data !== undefined) {
diff --git a/packages/core/tests/mcp.test.ts b/packages/core/tests/mcp.test.ts
@@ -53,14 +53,17 @@ test("bailianMcp 路径与 isStreamableHttpUnsupported", () => {
       ),
     ),
   ).toBe(true);
-  // 裸 405 也应触发降级，不依赖英文文案
   expect(
     isStreamableHttpUnsupported(new BailianError("MCP request failed: 405 Method Not Allowed")),
   ).toBe(true);
   expect(isStreamableHttpUnsupported(new BailianError("MCP request failed: 404 Not Found"))).toBe(
     false,
   );
   expect(isStreamableHttpUnsupported(new Error("405 streamableHttp"))).toBe(false);
+  // JSON-RPC business 405 must not trigger HTTP transport fallback
+  expect(isStreamableHttpUnsupported(new BailianError("MCP error (405): Method Not Allowed"))).toBe(
+    false,
+  );
 
   expect(
     isUrlOverrideSseFallbackCandidate(new BailianError("MCP request failed: 404 Not Found")),
@@ -70,6 +73,9 @@ test("bailianMcp 路径与 isStreamableHttpUnsupported", () => {
       new BailianError("MCP request failed: 405 Method Not Allowed"),
     ),
   ).toBe(true);
+  expect(isUrlOverrideSseFallbackCandidate(new BailianError("MCP error (404): not found"))).toBe(
+    false,
+  );
 });
 
 test("resolveSameOriginMessageUrl：同源通过、跨域拒绝", () => {
@@ -118,7 +124,7 @@ test("connectBailianMcpWithFallback：成功走 Streamable；405 降级 SSE", as
     globalThis.fetch = originalFetch;
   }
 
-  // 裸 405（无 streamableHttp 文案）→ SSE
+  // Bare HTTP 405 (no streamableHttp body text) → SSE
   let sseController: ReadableStreamDefaultController<Uint8Array> | undefined;
   const encoder = new TextEncoder();
   const urls: string[] = [];
@@ -207,7 +213,7 @@ test("connectBailianMcpWithFallback：WebSearch 不降级；urlOverride 同 URL
     globalThis.fetch = originalFetch;
   }
 
-  // urlOverride：POST 405 后应对同一 URL 发 GET SSE
+  // urlOverride: after POST 405, fall back with GET SSE on the same URL
   urls.length = 0;
   let sseController: ReadableStreamDefaultController<Uint8Array> | undefined;
   const encoder = new TextEncoder();
@@ -292,7 +298,7 @@ test("McpSseClient：流结束后立刻失败 pending（不干等到 timeout）"
   globalThis.fetch = async (input, init) => {
     const url = requestUrl(input);
     if ((init?.method ?? "GET") === "GET" || url.endsWith("/sse")) {
-      // 发完 endpoint 后立刻关流
+      // Close the stream immediately after the endpoint event
       const stream = new ReadableStream<Uint8Array>({
         start(controller) {
           controller.enqueue(
@@ -335,7 +341,7 @@ test("McpSseClient：string JSON-RPC id 可匹配；仅认 event:endpoint", asyn
       const stream = new ReadableStream<Uint8Array>({
         start(controller) {
           sseController = controller;
-          // 无名事件不应被当成 endpoint
+          // Untyped events must not be treated as endpoint
           controller.enqueue(
             encoder.encode(`data:${JSON.stringify({ jsonrpc: "2.0", id: 99, result: {} })}\n\n`),
           );
@@ -354,7 +360,7 @@ test("McpSseClient：string JSON-RPC id 可匹配；仅认 event:endpoint", asyn
       const body = typeof init?.body === "string" ? JSON.parse(init.body) : {};
       queueMicrotask(() => {
         if (body.id != null && sseController) {
-          // 以 string id 回传
+          // Echo id as a string
           sseController.enqueue(encoder.encode(jsonRpcResult(String(body.id), {})));
         }
       });
@@ -441,3 +447,63 @@ test("McpSseClient.close 可中止挂起 GET", async () => {
     globalThis.fetch = originalFetch;
   }
 });
+
+test("McpSseClient：等待响应头受 --timeout 约束", async () => {
+  const originalFetch = globalThis.fetch;
+
+  globalThis.fetch = async (_input, init) => {
+    const signal = init?.signal;
+    return new Promise((_resolve, reject) => {
+      if (!signal) {
+        reject(new Error("missing signal"));
+        return;
+      }
+      if (signal.aborted) {
+        reject(new DOMException("This operation was aborted.", "AbortError"));
+        return;
+      }
+      signal.addEventListener(
+        "abort",
+        () => reject(new DOMException("This operation was aborted.", "AbortError")),
+        { once: true },
+      );
+    });
+  };
+
+  try {
+    const client = new McpSseClient(
+      testDeps({ timeout: 1 }),
+      "https://example.test/sse",
+      "sk-test",
+    );
+    const started = Date.now();
+    await expect(client.initialize()).rejects.toThrow(/timed out waiting for response headers/i);
+    expect(Date.now() - started).toBeLessThan(2500);
+    client.close();
+  } finally {
+    globalThis.fetch = originalFetch;
+  }
+});
+
+test("McpSseClient：非 2xx 不产生 unhandledRejection", async () => {
+  const originalFetch = globalThis.fetch;
+  const unhandled: unknown[] = [];
+  const onUnhandled = (reason: unknown) => {
+    unhandled.push(reason);
+  };
+  process.on("unhandledRejection", onUnhandled);
+
+  globalThis.fetch = async () =>
+    new Response("boom", { status: 500, statusText: "Internal Server Error" });
+
+  try {
+    const client = new McpSseClient(testDeps(), "https://example.test/sse", "sk-test");
+    await expect(client.initialize()).rejects.toThrow(/MCP request failed:\s*500/i);
+    await new Promise((resolve) => setTimeout(resolve, 30));
+    expect(unhandled).toEqual([]);
+    client.close();
+  } finally {
+    process.off("unhandledRejection", onUnhandled);
+    globalThis.fetch = originalFetch;
+  }
+});
diff --git a/packages/core/tests/stream.test.ts b/packages/core/tests/stream.test.ts
@@ -0,0 +1,54 @@
+import { expect, test } from "vite-plus/test";
+import { parseSSE } from "../src/client/stream.ts";
+
+async function collectEvents(
+  chunks: string[],
+): Promise<Array<{ data: string; event?: string; id?: string }>> {
+  const encoder = new TextEncoder();
+  const stream = new ReadableStream<Uint8Array>({
+    start(controller) {
+      for (const chunk of chunks) {
+        controller.enqueue(encoder.encode(chunk));
+      }
+      controller.close();
+    },
+  });
+  const response = new Response(stream, {
+    headers: { "Content-Type": "text/event-stream" },
+  });
+  const events: Array<{ data: string; event?: string; id?: string }> = [];
+  for await (const event of parseSSE(response)) {
+    events.push(event);
+  }
+  return events;
+}
+
+test("parseSSE：单 chunk 完整事件保持原行为", async () => {
+  const events = await collectEvents([
+    'event: message\ndata: {"ok":true}\nid: 1\n\ndata: plain\n\n',
+  ]);
+  expect(events).toEqual([{ data: '{"ok":true}', event: "message", id: "1" }, { data: "plain" }]);
+});
+
+test("parseSSE：多行 data 与注释保持原行为", async () => {
+  const events = await collectEvents([": keep-alive\ndata: line1\ndata: line2\n\n"]);
+  expect(events).toEqual([{ data: "line1\nline2" }]);
+});
+
+test("parseSSE：跨 chunk 保留 event 类型", async () => {
+  const events = await collectEvents(["event: endpoint\n", "data: /message?sessionId=abc\n\n"]);
+  expect(events).toEqual([{ data: "/message?sessionId=abc", event: "endpoint" }]);
+});
+
+test("parseSSE：跨 chunk 保留 id，且多事件连续正确", async () => {
+  const events = await collectEvents([
+    "id: a\nevent: message\n",
+    'data: {"n":1}\n\n',
+    "event: message\ndata: ",
+    '{"n":2}\n\n',
+  ]);
+  expect(events).toEqual([
+    { data: '{"n":1}', event: "message", id: "a" },
+    { data: '{"n":2}', event: "message" },
+  ]);
+});
