fix(mcp): harden SSE parsing, abort, and fallback matching

Author: hopelynd

packages/commands/src/commands/mcp/activate-hint.ts

@@ -5,7 +5,7 @@ import { mcpMarketplaceDetailPage } from "bailian-cli-runtime";
 export function isMcpNotActivated(error: unknown): boolean {
   if (!(error instanceof BailianError)) return false;
   const message = error.message;
-  if (!/MCP request failed:\s*404\b/i.test(message)) return false;
+  if (!/^MCP request failed:\s*404\b/i.test(message)) return false;
   return /未开通|MCP不存在|MCP_IS_INVALID/i.test(message);
 }
 

packages/commands/tests/mcp-activate-hint.test.ts

@@ -26,6 +26,12 @@ describe("mcp-activate-hint", () => {
       false,
     );
     expect(isMcpNotActivated(new Error("MCP不存在或未开通"))).toBe(false);
+    // Nested wrapper phrase must not match (anchored at start).
+    expect(
+      isMcpNotActivated(
+        new BailianError("MCP error (-32000): MCP request failed: 404 Not Found - 未开通"),
+      ),
+    ).toBe(false);
   });
 
   test("hint 含对应 server 的 MCP 广场深链", () => {

packages/core/src/client/mcp-sse.ts

@@ -114,8 +114,7 @@ export class McpSseClient {
   private async openSse(): Promise<void> {
     if (this.abortController) return;
 
-    // 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).
+    // One abortController for header/error-body wait; clear timer before the long-lived stream.
     this.abortController = new AbortController();
     const timeoutMs = this.deps.settings.timeout * 1000;
     let headerTimedOut = false;
@@ -146,6 +145,8 @@ export class McpSseClient {
       });
     } catch (error) {
       clearTimeout(headerTimer);
+      // Allow a later initialize() to openSse again on this instance.
+      this.abortController = undefined;
       if (this.closed) {
         throw new BailianError("MCP SSE session closed.", ExitCode.GENERAL);
       }
@@ -155,27 +156,43 @@ export class McpSseClient {
       throw new BailianError(
         `MCP SSE request failed: ${error instanceof Error ? error.message : String(error)}`,
         ExitCode.NETWORK,
+        undefined,
+        { cause: error },
       );
     }
-    // 已收到响应头：取消 header 等待，后续仅由 abortController 结束流。
-    clearTimeout(headerTimer);
 
     if (this.deps.settings.verbose) {
       console.error(`< ${response.status} ${response.statusText}`);
     }
 
     if (!response.ok) {
+      // Keep headerTimer until error body is read (or times out).
       let errMsg = `MCP request failed: ${response.status} ${response.statusText}`;
       try {
         const errBody = await response.text();
         if (errBody) errMsg += ` - ${errBody.slice(0, 500)}`;
-      } catch {
-        /* ignore */
+      } catch (error) {
+        clearTimeout(headerTimer);
+        this.abortController = undefined;
+        if (this.closed) {
+          throw new BailianError("MCP SSE session closed.", ExitCode.GENERAL);
+        }
+        if (headerTimedOut) {
+          throw new BailianError(
+            "MCP SSE timed out reading error response body.",
+            ExitCode.TIMEOUT,
+          );
+        }
+        throw new BailianError(errMsg, ExitCode.GENERAL, undefined, { cause: error });
       }
-      // Throw only — do not rejectEndpoint; this path never awaits endpointReady.
+      clearTimeout(headerTimer);
+      this.abortController = undefined;
+      // Do not rejectEndpoint — openSse never awaits endpointReady on this path.
       throw new BailianError(errMsg, ExitCode.GENERAL);
     }
 
+    clearTimeout(headerTimer);
+
     void this.consumeSse(response).catch((error) => {
       if (this.closed) return;
       const reason =
@@ -247,8 +264,7 @@ export class McpSseClient {
       throw error;
     }
 
-    // Stream ended after endpoint: mark session dead and wake pending; do not throw,
-    // so void consumeSse().catch does not surface an extra unhandled rejection.
+    // After endpoint: mark dead and wake pending; don't throw (avoid unhandledRejection).
     this.markStreamEnded(new BailianError("MCP SSE stream ended unexpectedly.", ExitCode.GENERAL));
   }
 
@@ -332,12 +348,24 @@ export class McpSseClient {
     }
 
     const timeoutMs = this.deps.settings.timeout * 1000;
-    const res = await fetch(this.messageUrl, {
-      method: "POST",
-      headers,
-      body: JSON.stringify(body),
-      signal: AbortSignal.timeout(timeoutMs),
-    });
+    // Combine per-RPC timeout with session abort so close() cancels in-flight POSTs.
+    const requestSignal = createLinkedAbortSignal(timeoutMs, this.abortController?.signal);
+    let res: Response;
+    try {
+      res = await fetch(this.messageUrl, {
+        method: "POST",
+        headers,
+        body: JSON.stringify(body),
+        signal: requestSignal.signal,
+      });
+    } catch (error) {
+      if (this.closed) {
+        throw new BailianError("MCP SSE session closed.", ExitCode.GENERAL);
+      }
+      throw error;
+    } finally {
+      requestSignal.cleanup();
+    }
 
     if (this.deps.settings.verbose) {
       console.error(`< ${res.status} ${res.statusText}`);
@@ -406,3 +434,23 @@ function cancellableTimeoutReject(
     },
   };
 }
+
+/** Timeout + optional parent abort without AbortSignal.any (Node 18). */
+function createLinkedAbortSignal(
+  timeoutMs: number,
+  parentSignal?: AbortSignal,
+): { signal: AbortSignal; cleanup: () => void } {
+  const controller = new AbortController();
+  const timeout = setTimeout(() => controller.abort(), timeoutMs);
+  const abortFromParent = () => controller.abort(parentSignal?.reason);
+  const cleanup = () => {
+    clearTimeout(timeout);
+    parentSignal?.removeEventListener("abort", abortFromParent);
+  };
+
+  if (parentSignal?.aborted) abortFromParent();
+  else parentSignal?.addEventListener("abort", abortFromParent, { once: true });
+  controller.signal.addEventListener("abort", cleanup, { once: true });
+
+  return { signal: controller.signal, cleanup };
+}

packages/core/src/client/mcp.ts

@@ -70,20 +70,19 @@ export function bailianMcpSsePath(serverCode: string): string {
 
 /**
  * True when Streamable HTTP is unsupported and classic SSE fallback should be tried.
- * Match HTTP wrapper text `MCP request failed: 405` only — not JSON-RPC `MCP error (405)`.
- * Bailian HTTP 404 (not activated) is intentionally excluded.
+ * Anchored to HTTP wrapper text only (not JSON-RPC / nested copies). Bailian 404 excluded.
  */
 export function isStreamableHttpUnsupported(error: unknown): boolean {
   if (!(error instanceof BailianError)) return false;
-  return /MCP request failed:\s*405\b/i.test(error.message);
+  return /^MCP request failed:\s*405\b/i.test(error.message);
 }
 
 /**
  * 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 /MCP request failed:\s*(405|404)\b/i.test(error.message);
+  return /^MCP request failed:\s*(405|404)\b/i.test(error.message);
 }
 
 export type McpConnectedClient = {

packages/core/src/client/stream.ts

@@ -7,76 +7,127 @@ export interface ServerSentEvent {
   id?: string;
 }
 
+/** Normalize CRLF/CR to LF; hold a trailing `\r` so a split CRLF is not double-broken. */
+function takeNormalizedSseLines(buffer: string): { lines: string[]; rest: string } {
+  let text = buffer;
+  let holdTrailingCr = false;
+  if (text.endsWith("\r")) {
+    holdTrailingCr = true;
+    text = text.slice(0, -1);
+  }
+
+  text = text.replace(/\r\n/g, "\n").replace(/\r/g, "\n");
+  const parts = text.split("\n");
+  const incomplete = parts.pop() ?? "";
+  return {
+    lines: parts,
+    rest: holdTrailingCr ? `${incomplete}\r` : incomplete,
+  };
+}
+
+function applySseLine(
+  line: string,
+  event: Partial<ServerSentEvent>,
+  maxBuffer: number,
+): { event: Partial<ServerSentEvent>; completed?: ServerSentEvent } {
+  if (line === "") {
+    if (event.data === undefined) {
+      return { event: {} };
+    }
+    return {
+      event: {},
+      completed: { data: event.data, event: event.event, id: event.id },
+    };
+  }
+
+  if (line.startsWith(":")) {
+    return { event };
+  }
+
+  const colonIndex = line.indexOf(":");
+  if (colonIndex === -1) {
+    return { event };
+  }
+
+  const field = line.slice(0, colonIndex);
+  const fieldValue = line.slice(colonIndex + 1).trimStart();
+  const nextEvent: Partial<ServerSentEvent> = { ...event };
+
+  switch (field) {
+    case "data":
+      nextEvent.data =
+        nextEvent.data !== undefined ? `${nextEvent.data}\n${fieldValue}` : fieldValue;
+      if (nextEvent.data.length > maxBuffer) {
+        throw new BailianError("SSE event exceeded the maximum buffer size.", ExitCode.GENERAL);
+      }
+      break;
+    case "event":
+      nextEvent.event = fieldValue;
+      break;
+    case "id":
+      nextEvent.id = fieldValue;
+      break;
+  }
+
+  return { event: nextEvent };
+}
+
 export async function* parseSSE(response: Response): AsyncGenerator<ServerSentEvent> {
   const reader = response.body?.getReader();
   if (!reader) return;
 
   const decoder = new TextDecoder();
   let buffer = "";
 
-  // Guard against a hostile or malfunctioning stream that never emits a newline
-  // (or builds a single absurdly large event): bound the in-memory buffer so the
-  // parser cannot be driven to exhaust process memory.
   const MAX_SSE_BUFFER = 16 * 1024 * 1024; // 16 MiB
 
   try {
+    // Keep partial event fields across chunks.
     let event: Partial<ServerSentEvent> = {};
 
     while (true) {
       const { done, value } = await reader.read();
-      if (done) break;
+      if (done) {
+        // EOF: treat any held `\r` as a line ending.
+        if (buffer.length > 0) {
+          const finalText = buffer.replace(/\r\n/g, "\n").replace(/\r/g, "\n");
+          const parts = finalText.split("\n");
+          buffer = parts.pop() ?? "";
+          for (const line of parts) {
+            const applied = applySseLine(line, event, MAX_SSE_BUFFER);
+            event = applied.event;
+            if (applied.completed) {
+              yield applied.completed;
+            }
+          }
+        }
+        break;
+      }
 
       buffer += decoder.decode(value, { stream: true });
       if (buffer.length > MAX_SSE_BUFFER) {
         throw new BailianError("SSE stream exceeded the maximum buffer size.", ExitCode.GENERAL);
       }
 
-      const lines = buffer.split("\n");
-      buffer = lines.pop() || "";
+      const { lines, rest } = takeNormalizedSseLines(buffer);
+      buffer = rest;
 
       for (const line of lines) {
-        if (line === "") {
-          if (event.data !== undefined) {
-            yield { data: event.data, event: event.event, id: event.id };
-          }
-          event = {};
-          continue;
-        }
-
-        if (line.startsWith(":")) continue; // comment
-
-        const colonIndex = line.indexOf(":");
-        if (colonIndex === -1) continue;
-
-        const field = line.slice(0, colonIndex);
-        const value = line.slice(colonIndex + 1).trimStart();
-
-        switch (field) {
-          case "data":
-            event.data = event.data !== undefined ? `${event.data}\n${value}` : value;
-            if (event.data.length > MAX_SSE_BUFFER) {
-              throw new BailianError(
-                "SSE event exceeded the maximum buffer size.",
-                ExitCode.GENERAL,
-              );
-            }
-            break;
-          case "event":
-            event.event = value;
-            break;
-          case "id":
-            event.id = value;
-            break;
+        const applied = applySseLine(line, event, MAX_SSE_BUFFER);
+        event = applied.event;
+        if (applied.completed) {
+          yield applied.completed;
         }
       }
     }
 
-    // Flush remaining
-    if (buffer.trim() && buffer.includes("data:")) {
-      const colonIndex = buffer.indexOf(":");
-      if (colonIndex !== -1) {
-        yield { data: buffer.slice(colonIndex + 1).trimStart() };
-      }
+    // Legacy EOF flush: apply trailing field line and dispatch with event/id intact.
+    if (buffer.length > 0) {
+      const applied = applySseLine(buffer, event, MAX_SSE_BUFFER);
+      event = applied.event;
+    }
+    if (event.data !== undefined) {
+      yield { data: event.data, event: event.event, id: event.id };
     }
   } finally {
     reader.releaseLock();