diff --git a/packages/commands/src/commands/mcp/activate-hint.ts b/packages/commands/src/commands/mcp/activate-hint.ts index 5fe4060..0be2e56 100644 --- a/packages/commands/src/commands/mcp/activate-hint.ts +++ b/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); } diff --git a/packages/commands/tests/mcp-activate-hint.test.ts b/packages/commands/tests/mcp-activate-hint.test.ts index 223c7fb..6a1602e 100644 --- a/packages/commands/tests/mcp-activate-hint.test.ts +++ b/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 广场深链", () => { diff --git a/packages/core/src/client/mcp-sse.ts b/packages/core/src/client/mcp-sse.ts index fd62f63..97188b5 100644 --- a/packages/core/src/client/mcp-sse.ts +++ b/packages/core/src/client/mcp-sse.ts @@ -114,8 +114,7 @@ export class McpSseClient { private async openSse(): Promise { 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 }; +} diff --git a/packages/core/src/client/mcp.ts b/packages/core/src/client/mcp.ts index 00228e0..c3ffe9e 100644 --- a/packages/core/src/client/mcp.ts +++ b/packages/core/src/client/mcp.ts @@ -70,12 +70,11 @@ 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); } /** @@ -83,7 +82,7 @@ export function isStreamableHttpUnsupported(error: unknown): boolean { */ 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 = { diff --git a/packages/core/src/client/stream.ts b/packages/core/src/client/stream.ts index bca5e09..05e70ca 100644 --- a/packages/core/src/client/stream.ts +++ b/packages/core/src/client/stream.ts @@ -7,6 +7,71 @@ 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, + maxBuffer: number, +): { event: Partial; 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 = { ...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 { const reader = response.body?.getReader(); if (!reader) return; @@ -14,69 +79,55 @@ export async function* parseSSE(response: Response): AsyncGenerator = {}; 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(); diff --git a/packages/core/tests/mcp.test.ts b/packages/core/tests/mcp.test.ts index d3b4125..55e1034 100644 --- a/packages/core/tests/mcp.test.ts +++ b/packages/core/tests/mcp.test.ts @@ -64,6 +64,12 @@ test("bailianMcp 路径与 isStreamableHttpUnsupported", () => { expect(isStreamableHttpUnsupported(new BailianError("MCP error (405): Method Not Allowed"))).toBe( false, ); + // Nested wrapper phrase in a JSON-RPC message must not trigger fallback. + expect( + isStreamableHttpUnsupported( + new BailianError("MCP error (-32000): MCP request failed: 405 Method Not Allowed"), + ), + ).toBe(false); expect( isUrlOverrideSseFallbackCandidate(new BailianError("MCP request failed: 404 Not Found")), @@ -76,6 +82,11 @@ test("bailianMcp 路径与 isStreamableHttpUnsupported", () => { expect(isUrlOverrideSseFallbackCandidate(new BailianError("MCP error (404): not found"))).toBe( false, ); + expect( + isUrlOverrideSseFallbackCandidate( + new BailianError("MCP error (-32000): MCP request failed: 404 Not Found"), + ), + ).toBe(false); }); test("resolveSameOriginMessageUrl:同源通过、跨域拒绝", () => { @@ -507,3 +518,169 @@ test("McpSseClient:非 2xx 不产生 unhandledRejection", async () => { globalThis.fetch = originalFetch; } }); + +test("McpSseClient:非 2xx 读 body 仍受 --timeout 约束", async () => { + const originalFetch = globalThis.fetch; + + globalThis.fetch = async (_input, init) => { + const signal = init?.signal; + return { + ok: false, + status: 500, + statusText: "Internal Server Error", + async text() { + 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 }, + ); + }); + }, + } as Response; + }; + + 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 reading error response body/i); + expect(Date.now() - started).toBeLessThan(2500); + client.close(); + } finally { + globalThis.fetch = originalFetch; + } +}); + +test("McpSseClient:fetch 失败保留 cause", async () => { + const originalFetch = globalThis.fetch; + const root = Object.assign(new Error("getaddrinfo ENOTFOUND example.test"), { + code: "ENOTFOUND", + }); + const fetchFailed = new TypeError("fetch failed", { cause: root }); + + globalThis.fetch = async () => { + throw fetchFailed; + }; + + try { + const client = new McpSseClient(testDeps(), "https://example.test/sse", "sk-test"); + await expect(client.initialize()).rejects.toMatchObject({ + message: expect.stringMatching(/MCP SSE request failed:\s*fetch failed/i), + exitCode: 6, + cause: fetchFailed, + }); + client.close(); + } finally { + globalThis.fetch = originalFetch; + } +}); + +test("McpSseClient:fetch 失败后同实例可重新 openSse", async () => { + const originalFetch = globalThis.fetch; + let attempt = 0; + let sseController: ReadableStreamDefaultController | undefined; + const encoder = new TextEncoder(); + + globalThis.fetch = async (input, init) => { + const url = requestUrl(input); + const method = init?.method ?? "GET"; + if (method === "GET" || url.endsWith("/sse")) { + attempt += 1; + if (attempt === 1) { + throw new TypeError("fetch failed"); + } + const stream = new ReadableStream({ + start(controller) { + sseController = controller; + controller.enqueue(encoder.encode("event: endpoint\ndata: /message\n\n")); + }, + }); + return new Response(stream, { + status: 200, + headers: { "Content-Type": "text/event-stream" }, + }); + } + const body = typeof init?.body === "string" ? JSON.parse(init.body) : {}; + queueMicrotask(() => { + if (body.id != null && sseController) { + sseController.enqueue(encoder.encode(jsonRpcResult(body.id, {}))); + } + }); + return new Response("{}", { status: 200, headers: { "Content-Type": "application/json" } }); + }; + + try { + const client = new McpSseClient(testDeps(), "https://example.test/sse", "sk-test"); + await expect(client.initialize()).rejects.toThrow(/fetch failed/i); + await client.initialize(); + client.close(); + } finally { + globalThis.fetch = originalFetch; + } +}); + +test("McpSseClient:close 可中止进行中的 POST", async () => { + const originalFetch = globalThis.fetch; + let postAborted = false; + const encoder = new TextEncoder(); + + globalThis.fetch = async (input, init) => { + const url = requestUrl(input); + const method = init?.method ?? "GET"; + if (method === "GET" || url.endsWith("/sse")) { + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode("event: endpoint\ndata: /message\n\n")); + }, + }); + return new Response(stream, { + status: 200, + headers: { "Content-Type": "text/event-stream" }, + }); + } + + const signal = init?.signal; + return new Promise((_resolve, reject) => { + if (!signal) { + reject(new Error("missing signal")); + return; + } + const onAbort = () => { + postAborted = true; + reject(new DOMException("This operation was aborted.", "AbortError")); + }; + if (signal.aborted) { + onAbort(); + return; + } + signal.addEventListener("abort", onAbort, { once: true }); + }); + }; + + try { + const client = new McpSseClient( + testDeps({ timeout: 5 }), + "https://example.test/sse", + "sk-test", + ); + const initPromise = client.initialize(); + await new Promise((resolve) => setTimeout(resolve, 30)); + client.close(); + await expect(initPromise).rejects.toThrow(/session closed|aborted/i); + expect(postAborted).toBe(true); + } finally { + globalThis.fetch = originalFetch; + } +}); diff --git a/packages/core/tests/stream.test.ts b/packages/core/tests/stream.test.ts index 12fe3f8..2bf4479 100644 --- a/packages/core/tests/stream.test.ts +++ b/packages/core/tests/stream.test.ts @@ -52,3 +52,28 @@ test("parseSSE:跨 chunk 保留 id,且多事件连续正确", async () => { { data: '{"n":2}', event: "message" }, ]); }); + +test("parseSSE:CRLF 行尾可解析 endpoint", async () => { + const events = await collectEvents(["event: endpoint\r\ndata: /message\r\n\r\n"]); + expect(events).toEqual([{ data: "/message", event: "endpoint" }]); +}); + +test("parseSSE:纯 CR 行尾可解析 endpoint", async () => { + const events = await collectEvents(["event: endpoint\rdata: /message\r\r"]); + expect(events).toEqual([{ data: "/message", event: "endpoint" }]); +}); + +test("parseSSE:跨 chunk 的 CRLF(\\r|\\n)不丢事件", async () => { + const events = await collectEvents(["event: endpoint\r", "\ndata: /message\r\n\r\n"]); + expect(events).toEqual([{ data: "/message", event: "endpoint" }]); +}); + +test("parseSSE:EOF without blank line keeps event type", async () => { + const events = await collectEvents(["event: endpoint\ndata: /message"]); + expect(events).toEqual([{ data: "/message", event: "endpoint" }]); +}); + +test("parseSSE:EOF data-only flush keeps prior behavior", async () => { + const events = await collectEvents(["data: plain"]); + expect(events).toEqual([{ data: "plain" }]); +});