diff --git a/server/src/__tests__/mcp-http.test.ts b/server/src/__tests__/mcp-http.test.ts index 83ebaca3419..59061802434 100644 --- a/server/src/__tests__/mcp-http.test.ts +++ b/server/src/__tests__/mcp-http.test.ts @@ -5,8 +5,83 @@ import { MCP_PROTOCOL_VERSION, mcpHttpRequestHeaders, parseMcpHttpResponseBody, + readMcpHttpResponse, } from "../services/mcp-http.js"; +describe("bounded MCP responses", () => { + it("retains consumed SSE progress when a partial event fails and releases the reader", async () => { + const partial = 'data: {"jsonrpc":"2.0","method":"notifications/progress"}\n\ndata: {"jsonrpc":'; + const failure = new Error("peer reset"); + let reads = 0; + const body = new ReadableStream({ + pull(controller) { + if (reads++ === 0) controller.enqueue(new TextEncoder().encode(partial)); + else controller.error(failure); + }, + }, { highWaterMark: 0 }); + let progress: unknown; + await expect(readMcpHttpResponse(new Response(body, { headers: { "content-type": "text/event-stream" } }), "request", { + onProgress: (value) => { progress = value; }, + })).rejects.toBe(failure); + expect(progress).toEqual({ bytesReceived: Buffer.byteLength(partial), chunksReceived: 1, eventsReceived: 1, matchedResponse: false, ended: false }); + expect(body.locked).toBe(false); + }); + + for (const contentType of ["application/json", "text/event-stream"]) { + it(`rejects declared ${contentType} overflow before consuming the body`, async () => { + let reads = 0; + let cancelled = false; + const body = new ReadableStream({ + pull(controller) { reads += 1; controller.enqueue(new Uint8Array(1)); }, + cancel() { cancelled = true; }, + }, { highWaterMark: 0 }); + const response = new Response(body, { + headers: { "content-type": contentType, "content-length": "101" }, + }); + await expect(readMcpHttpResponse(response, "request", { maxBytes: 100 })).rejects.toMatchObject({ + name: "McpHttpResponseError", reason: "too_large", maxBytes: 100, responseBytes: 101, + }); + expect(reads).toBe(0); + expect(cancelled).toBe(true); + expect(body.locked).toBe(false); + }); + + it(`cancels ${contentType} at the first streamed byte beyond the limit`, async () => { + let reads = 0; + let cancelled = false; + const body = new ReadableStream({ + pull(controller) { reads += 1; controller.enqueue(new Uint8Array(reads === 1 ? 100 : 1)); }, + cancel() { cancelled = true; }, + }, { highWaterMark: 0 }); + await expect(readMcpHttpResponse(new Response(body, { + headers: { "content-type": contentType, "content-length": "invalid" }, + }), "request", { maxBytes: 100 })).rejects.toMatchObject({ + reason: "too_large", maxBytes: 100, responseBytes: 101, + }); + expect(reads).toBe(2); + expect(cancelled).toBe(true); + expect(body.locked).toBe(false); + }); + + it(`reports UTF-8 byte size for buffered ${contentType} overflow`, async () => { + const response = new Response(null, { headers: { "content-type": contentType } }); + response.text = async () => "\u00e9".repeat(51); + await expect(readMcpHttpResponse(response, "request", { maxBytes: 100 })).rejects.toMatchObject({ + reason: "too_large", maxBytes: 100, responseBytes: 102, + }); + }); + + it(`accepts ${contentType} at exactly the byte limit`, async () => { + const payload = { jsonrpc: "2.0", id: "request", result: {} }; + const json = JSON.stringify(payload); + const body = contentType === "text/event-stream" ? `data: ${json}\n\n` : json; + expect(await readMcpHttpResponse(new Response(body, { + headers: { "content-type": contentType, "content-length": String(Buffer.byteLength(body)) }, + }), "request", { maxBytes: Buffer.byteLength(body) })).toEqual(payload); + }); + } +}); + describe("mcpHttpRequestHeaders", () => { it("advertises both JSON and SSE on every request", () => { expect(mcpHttpRequestHeaders()).toMatchObject({ diff --git a/server/src/__tests__/tool-gateway.test.ts b/server/src/__tests__/tool-gateway.test.ts index 64f09143526..f1bb4d760ac 100644 --- a/server/src/__tests__/tool-gateway.test.ts +++ b/server/src/__tests__/tool-gateway.test.ts @@ -4439,6 +4439,84 @@ rl.on("line", (line) => { } }); + for (const scenario of ["slow success", "default deadline", "explicit deadline", "peer reset", "cyclic peer reset", "primitive peer reset"] as const) { + it(`records remote SSE ${scenario} without confusing a call deadline with a broken transport`, async () => { + const company = await createCompany(db); + const agent = await createAgent(db, company.id); + const { run } = await createIssueAndRun(db, company.id, agent.id); + const { connection } = await createRemoteMcpTool(db, company.id); + await allowAllToolsForAgent(db, company.id, agent.id); + const reading = Promise.withResolvers(); + let bodyController!: ReadableStreamDefaultController; + let requestId!: string; + let signal!: AbortSignal; + const progress = 'data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progress":1}}\n\n'; + const cause = Object.assign(new Error("peer failure Authorization: Bearer fixture-private-value"), { code: "ECONNRESET" }); + const failure = Object.assign(new Error("aborted", { cause }), { code: "ECONNRESET" }); + if (scenario === "cyclic peer reset") Object.assign(cause, { cause: failure }); + const gateway = createTestToolGatewayService(db, { + remoteHttpRequest: async (_url, init) => { + requestId = JSON.parse(String(init.body)).id; + signal = init.signal!; + const body = new ReadableStream({ + start(controller) { + bodyController = controller; + controller.enqueue(new TextEncoder().encode(progress)); + init.signal!.addEventListener("abort", () => controller.error(failure), { once: true }); + }, + pull() { reading.resolve(); }, + }, { highWaterMark: 0 }); + return new Response(body, { headers: { "content-type": "text/event-stream", "x-request-id": "sse-request" } }); + }, + }); + const session = await gateway.createSession({ companyId: company.id, agentId: agent.id, runId: run.id }); + const tool = (await gateway.listToolsForSession(session.token)).find((entry) => entry.providerType === "mcp_remote_http")!; + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + const call = gateway.executeTool({ + sessionToken: session.token, tool: tool.name, parameters: { key: "a", value: "b" }, + ...(scenario === "explicit deadline" ? { timeoutMs: 1500 } : {}), + }).then((value) => ({ value, error: null }), (error: unknown) => ({ value: null, error })); + try { + await reading.promise; + if (scenario.includes("peer reset")) bodyController.error(scenario === "primitive peer reset" ? "peer reset" : failure); + else await vi.advanceTimersByTimeAsync(scenario === "slow success" ? 12_000 : scenario === "explicit deadline" ? 1500 : 30_000); + if (scenario === "slow success") { + expect(signal.aborted).toBe(false); + bodyController.enqueue(new TextEncoder().encode(`data: ${JSON.stringify({ jsonrpc: "2.0", id: requestId, result: { content: [{ type: "text", text: "completed after twelve seconds" }] } })}\n\n`)); + } + const result = await call; + const [event] = await db.select().from(toolCallEvents).where(eq(toolCallEvents.eventType, scenario === "slow success" ? "call_completed" : "call_failed")); + const [stored] = await db.select().from(toolConnections).where(eq(toolConnections.id, connection.id)); + if (scenario === "slow success") { + expect(result.error).toBeNull(); + expect(JSON.stringify(result.value)).toContain("completed after twelve seconds"); + expect(stored.healthStatus).toBe("ok"); + expect(event.metadata).toMatchObject({ execution: { stream: { eventsReceived: 2, matchedResponse: true, ended: false } } }); + } else { + const timeout = scenario === "default deadline" || scenario === "explicit deadline"; + expectGatewayError(result.error, timeout ? 504 : 502, timeout ? "tool_timeout" : "mcp_remote_fetch_failed"); + expect(stored.healthStatus).toBe(timeout ? "degraded" : "error"); + expect(event.metadata).toMatchObject({ execution: { + failureKind: timeout ? "invocation_timeout" : "transport_failure", + deadline: { timeoutMs: scenario === "explicit deadline" ? 1500 : 30_000, aborted: timeout }, + error: scenario === "primitive peer reset" + ? { name: null, message: "peer reset", code: null, cause: null } + : { name: "Error", message: "aborted", code: "ECONNRESET", cause: { name: "Error", code: "ECONNRESET" } }, + response: { httpStatus: 200, contentType: "text/event-stream", upstreamRequestId: "sse-request" }, + stream: { bytesReceived: Buffer.byteLength(progress), chunksReceived: 1, eventsReceived: 1, matchedResponse: false, ended: false }, + } }); + expect(JSON.stringify(event.metadata)).not.toContain("fixture-private-value"); + if (scenario !== "primitive peer reset") expect(JSON.stringify(event.metadata)).toContain("peer failure"); + if (scenario === "cyclic peer reset") expect(JSON.stringify(event.metadata)).toContain("[Circular cause]"); + } + } finally { + vi.useRealTimers(); + bodyController.error(new Error("fixture cleanup")); + await call; + } + }); + } + const remoteFailureCases = [ { name: "HTTP status", diff --git a/server/src/services/mcp-http.ts b/server/src/services/mcp-http.ts index 80db7f9c17e..0a9cf0fbc7e 100644 --- a/server/src/services/mcp-http.ts +++ b/server/src/services/mcp-http.ts @@ -175,13 +175,27 @@ export function parseMcpHttpResponseBody(bodyText: string, contentType: string | throw new SyntaxError("MCP SSE response contained no data events"); } +export type McpHttpResponseProgress = { + bytesReceived: number; + chunksReceived: number; + eventsReceived: number; + matchedResponse: boolean; + ended: boolean; +}; + /** Read until the response for this request arrives, without waiting for an SSE * connection to close. Notifications and responses for other IDs are ignored. */ export async function readMcpHttpResponse( response: Response, requestId: string | number, - options: { maxBytes?: number; onRequest?: (message: Record) => Promise } = {}, + options: { + maxBytes?: number; + onRequest?: (message: Record) => Promise; + onProgress?: (progress: McpHttpResponseProgress) => void; + } = {}, ): Promise { + const progress: McpHttpResponseProgress = { bytesReceived: 0, chunksReceived: 0, eventsReceived: 0, matchedResponse: false, ended: false }; + options.onProgress?.(progress); const maxBytes = options.maxBytes ?? 8 * 1024 * 1024; const isStream = response.headers.get("content-type")?.toLowerCase().includes("text/event-stream"); const reader = response.body?.getReader(); @@ -204,20 +218,27 @@ export async function readMcpHttpResponse( const inspect = async (message: unknown): Promise => { if (!message || typeof message !== "object") return undefined; const record = message as Record; - if (record.id === requestId && ("result" in record || "error" in record)) return record; + if (record.id === requestId && ("result" in record || "error" in record)) { + progress.matchedResponse = true; + return record; + } if ("method" in record && "id" in record) await options.onRequest?.(record); return undefined; }; const event = async (value: string) => { const data = value.split("\n").filter((line) => line.startsWith("data:")).map((line) => line.slice(5).replace(/^ /, "")).join("\n"); if (!data) return undefined; + progress.eventsReceived += 1; return inspect(parse(data)); }; try { while (true) { const { value, done } = await reader.read(); bytes += value?.byteLength ?? 0; - if (bytes > maxBytes) throw new McpHttpResponseError("too_large", "MCP response exceeded the size limit"); + progress.bytesReceived = bytes; + if (value) progress.chunksReceived += 1; + progress.ended = done; + if (bytes > maxBytes) throw new McpHttpResponseError("too_large", "MCP response exceeded the size limit", { maxBytes, responseBytes: bytes }); buffer += decoder.decode(value, { stream: !done }); if (isStream) { // Normalize CRLF after concatenating chunks, including split CR/LF pairs. diff --git a/server/src/services/remote-http-fetch.ts b/server/src/services/remote-http-fetch.ts index b02d8769fa3..57638cb980f 100644 --- a/server/src/services/remote-http-fetch.ts +++ b/server/src/services/remote-http-fetch.ts @@ -33,7 +33,7 @@ const DEFAULT_CONNECT_TIMEOUT_MS = 10_000; * that own a longer budget — `tools/call`, which an operator can raise to 60 s — * pass `responseTimeoutMs` so this default never truncates it. */ -const DEFAULT_RESPONSE_TIMEOUT_MS = 30_000; +export const DEFAULT_REMOTE_HTTP_RESPONSE_TIMEOUT_MS = 30_000; const DNS_RESOLUTION_ERROR_CODES = new Set(["ENODATA", "ENOTFOUND", "EAI_AGAIN"]); /** How a verified socket is opened. Overridable so tests can simulate a rebind. */ @@ -156,7 +156,7 @@ async function pinnedRequest( socket, init, signal, - responseTimeoutMs: options.responseTimeoutMs ?? DEFAULT_RESPONSE_TIMEOUT_MS, + responseTimeoutMs: options.responseTimeoutMs ?? DEFAULT_REMOTE_HTTP_RESPONSE_TIMEOUT_MS, error: options.error, }); } catch (error) { diff --git a/server/src/services/tool-gateway.ts b/server/src/services/tool-gateway.ts index 5c19fe5302d..6d5e3f32e71 100644 --- a/server/src/services/tool-gateway.ts +++ b/server/src/services/tool-gateway.ts @@ -44,6 +44,7 @@ import { extractRemoteMcpPending } from "./remote-mcp-pending.js"; import { logger } from "../middleware/logger.js"; import { getAppDefinitionForApplicationKey, isConnectorFeatureEnabled } from "@paperclipai/shared"; import { instanceSettingsService } from "./instance-settings.js"; +import { redactSensitiveText } from "../redaction.js"; import { spawn } from "node:child_process"; import { createHash, randomBytes, randomUUID } from "node:crypto"; import { isDeepStrictEqual } from "node:util"; @@ -140,6 +141,7 @@ import { readMcpHttpResponse, McpHttpResponseError, McpHttpInitializationError, + type McpHttpResponseProgress, mcpHttpRequestHeaders, parseMcpHttpResponseBody, } from "./mcp-http.js"; @@ -153,6 +155,7 @@ import { assertGmailArchiveToolArgumentsSupported, gmailArchiveToolDescription } import { parseRemoteHttpEndpoint } from "./remote-http-endpoint-guard.js"; import { guardedRemoteHttpFetch, + DEFAULT_REMOTE_HTTP_RESPONSE_TIMEOUT_MS, type GuardedRemoteHttpFetchOptions, } from "./remote-http-fetch.js"; import { @@ -487,8 +490,33 @@ type RemoteHttpExecutionAudit = { bodySizeBytes: number; upstreamRequestId: string | null; }; + failureKind?: "invocation_timeout" | "transport_failure" | "protocol_failure"; + error?: RemoteHttpErrorEvidence; + deadline?: { timeoutMs: number; aborted: boolean }; + stream?: McpHttpResponseProgress; }; +type RemoteHttpErrorEvidence = { + name: string | null; + message: string; + code: string | null; + stack: string | null; + cause: RemoteHttpErrorEvidence | null; +}; + +function remoteHttpErrorEvidence(error: unknown, seen = new Set()): RemoteHttpErrorEvidence { + if (seen.has(error)) return { name: null, message: "[Circular cause]", code: null, stack: null, cause: null }; + const record = asRecord(error); + seen.add(error); + return { + name: typeof record?.name === "string" ? redactSensitiveText(record.name) : null, + message: redactSensitiveText(typeof record?.message === "string" ? record.message : String(error)), + code: typeof record?.code === "string" ? redactSensitiveText(record.code) : null, + stack: typeof record?.stack === "string" ? redactSensitiveText(record.stack) : null, + cause: record?.cause == null ? null : remoteHttpErrorEvidence(record.cause, seen), + }; +} + type LocalStdioRuntimeTemplate = { templateId: string; command: string | null; @@ -6193,6 +6221,7 @@ export function createToolGatewayService( // and connections whose stored grant/catalog predates scope reduction. assertGoogleChatToolArgumentsSupported(connection, entry.toolName, parameters); assertGmailArchiveToolArgumentsSupported(connection, entry.toolName, parameters); + if (useDefaultTimeout) ms = DEFAULT_REMOTE_HTTP_RESPONSE_TIMEOUT_MS; if (useDefaultTimeout && isRailwayConnection(connection) && entry.toolName === `${RAILWAY_TOOL_PREFIX}run-command`) { ms = railwayCommandBudgetMs(parameters); } @@ -6480,6 +6509,7 @@ export function createToolGatewayService( const body = response.ok ? JSON.stringify(await readMcpHttpResponse(response, requestId, { maxBytes: MAX_REMOTE_MCP_RESPONSE_BYTES, + onProgress: (progress) => { execution.stream = progress; }, onRequest: async (message) => { const pending = extractRemoteMcpPending(message); if (pending) { @@ -6607,7 +6637,22 @@ export function createToolGatewayService( ); return { result, headerSummary, execution }; } catch (error) { + execution.error = remoteHttpErrorEvidence(error); + execution.deadline = { timeoutMs: ms, aborted: controller.signal.aborted }; + if (controller.signal.aborted || (error instanceof Error && error.name === "AbortError")) { + if (mcpSession) forgetMcpHttpSession(mcpSession); + execution.failureKind = "invocation_timeout"; + await markCallHealth( + "degraded", + "Remote MCP tool call timed out; transport remains viable.", + ); + throw new ToolGatewayHttpError(504, "Remote MCP tool call timed out", "tool_timeout", { + connectionId: connection.id, catalogEntryId: entry.id, execution, failureKind: "invocation_timeout", + }); + } if (error instanceof McpHttpResponseError) { + execution.failureKind = "protocol_failure"; + if (execution.response && error.responseBytes != null) execution.response.bodySizeBytes = error.responseBytes; const failure = error.reason === "too_large" ? responseTooLargeError() : error.reason === "malformed_response" ? malformedRemoteMcpResponse() : new ToolGatewayHttpError(502, "Remote MCP server returned invalid JSON", "mcp_remote_invalid_json"); @@ -6635,21 +6680,11 @@ export function createToolGatewayService( }, ); } - if (error instanceof Error && error.name === "AbortError") { - if (mcpSession) forgetMcpHttpSession(mcpSession); - throw new ToolGatewayHttpError( - 504, - "Remote MCP tool call timed out", - "tool_timeout", - { - connectionId: connection.id, - catalogEntryId: entry.id, - execution, - }, - ); - } + execution.failureKind = "transport_failure"; if (error instanceof McpHttpInitializationError && (error.status === 401 || error.status === 403)) { await markCallHealth("error", "Remote MCP session authorization failed."); + } else { + await markCallHealth("error", "Remote MCP tool call failed."); } if (mcpSession) forgetMcpHttpSession(mcpSession); throw new ToolGatewayHttpError(