diff --git a/server/src/__tests__/mcp-http.test.ts b/server/src/__tests__/mcp-http.test.ts index 83ebaca3419..9e220008645 100644 --- a/server/src/__tests__/mcp-http.test.ts +++ b/server/src/__tests__/mcp-http.test.ts @@ -5,8 +5,29 @@ import { MCP_PROTOCOL_VERSION, mcpHttpRequestHeaders, parseMcpHttpResponseBody, + readMcpHttpResponse, } from "../services/mcp-http.js"; +describe("MCP response progress", () => { + 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); + }); +}); + describe("mcpHttpRequestHeaders", () => { it("advertises both JSON and SSE on every request", () => { expect(mcpHttpRequestHeaders()).toMatchObject({ diff --git a/server/src/__tests__/tool-gateway-service.test.ts b/server/src/__tests__/tool-gateway-service.test.ts index 78c3b8ea9ff..32168c3a800 100644 --- a/server/src/__tests__/tool-gateway-service.test.ts +++ b/server/src/__tests__/tool-gateway-service.test.ts @@ -3330,7 +3330,9 @@ describeEmbeddedPostgres("tool gateway service", () => { }); }); - it("refreshes Vercel tokens after an upstream 401 without replaying the tool call", async () => { + for (const deadlinePassesDuringRefresh of [false, true]) it(deadlinePassesDuringRefresh + ? "keeps the explicit retry reason when the call deadline passes during the Vercel token refresh" + : "refreshes Vercel tokens after an upstream 401 without replaying the tool call", async () => { const { company, agent, run } = await createRunFixture(db); const { connection } = await createRemoteMcpToolFixture(db, company.id); await db.update(toolConnections).set({ @@ -3355,12 +3357,20 @@ describeEmbeddedPostgres("tool gateway service", () => { }).where(eq(connectionGrants.connectionId, connection.id)); await db.insert(toolPolicies).values(readPolicyRow(company, "Allow Vercel-backed reads")); - const getToken = vi.fn(async (_request, options) => ({ - token: options?.forceRefresh ? "fresh-provider-bearer" : "stale-provider-bearer", - tokenId: options?.forceRefresh ? "stk_fresh" : "stk_stale", - expiresAt: Date.now() + 60_000, - connector: { id: "scl_posthog", uid: "posthog-paperclip", type: "api-key" }, - })); + const refreshing = Promise.withResolvers(); + const refreshGate = Promise.withResolvers(); + const getToken = vi.fn(async (_request, options) => { + if (options?.forceRefresh && deadlinePassesDuringRefresh) { + refreshing.resolve(); + await refreshGate.promise; + } + return { + token: options?.forceRefresh ? "fresh-provider-bearer" : "stale-provider-bearer", + tokenId: options?.forceRefresh ? "stk_fresh" : "stk_stale", + expiresAt: Date.now() + 60_000, + connector: { id: "scl_posthog", uid: "posthog-paperclip", type: "api-key" }, + }; + }); const evict = vi.fn(); const vercelConnectClient: VercelConnectClient = { getConnectorMetadata: vi.fn(), @@ -3387,11 +3397,23 @@ describeEmbeddedPostgres("tool gateway service", () => { const tool = (await gateway.listToolsForSession(session.token)) .find((candidate) => candidate.providerType === "mcp_remote_http"); - await expect(gateway.executeTool({ + if (deadlinePassesDuringRefresh) vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + const call = gateway.executeTool({ sessionToken: session.token, tool: tool!.name, parameters: {}, - })).rejects.toMatchObject({ + ...(deadlinePassesDuringRefresh ? { timeoutMs: 1500 } : {}), + }); + if (deadlinePassesDuringRefresh) { + try { + await refreshing.promise; + await vi.advanceTimersByTimeAsync(1500); + } finally { + vi.useRealTimers(); + refreshGate.resolve(); + } + } + await expect(call).rejects.toMatchObject({ status: 409, reasonCode: "oauth_refreshed_retry_required", }); diff --git a/server/src/__tests__/tool-gateway.test.ts b/server/src/__tests__/tool-gateway.test.ts index 61d26d36d1e..03c7369bf93 100644 --- a/server/src/__tests__/tool-gateway.test.ts +++ b/server/src/__tests__/tool-gateway.test.ts @@ -4627,6 +4627,154 @@ 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("ok"); + 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; + } + }); + } + + async function startRemoteCallWithControlledBody(respond: (body: ReadableStream) => Response, waitForRead = true) { + const company = await createCompany(db); + const agent = await createAgent(db, company.id); + const { run } = await createIssueAndRun(db, company.id, agent.id); + await createRemoteMcpTool(db, company.id); + await allowAllToolsForAgent(db, company.id, agent.id); + const reading = Promise.withResolvers(); + let bodyController!: ReadableStreamDefaultController; + let requestId!: string; + const gateway = createTestToolGatewayService(db, { + remoteHttpRequest: async (_url, init) => { + requestId = JSON.parse(String(init.body)).id; + return respond(new ReadableStream({ + start(controller) { bodyController = controller; }, + pull() { reading.resolve(); }, + }, { highWaterMark: 0 })); + }, + }); + 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" }, timeoutMs: 1500 }) + .then((value) => ({ value, error: null }), (error: unknown) => ({ value: null, error })); + if (waitForRead) await reading.promise; + const failedEvent = async () => (await db.select().from(toolCallEvents).where(eq(toolCallEvents.eventType, "call_failed")))[0]; + return { call, failedEvent, body: () => bodyController, requestId: () => requestId }; + } + + it("keeps a tool question when the call deadline passes while the question is recorded", async () => { + const remote = await startRemoteCallWithControlledBody((body) => new Response(body, { headers: { "content-type": "text/event-stream" } })); + try { + await vi.advanceTimersByTimeAsync(1500); + const question = { jsonrpc: "2.0", id: "question-1", method: "elicitation/create", params: { message: "Which folder?", requestedSchema: { type: "object" } } }; + remote.body().enqueue(new TextEncoder().encode(`data: ${JSON.stringify(question)}\n\n`)); + expectGatewayError((await remote.call).error, 409, "elicitation_required"); + } finally { + vi.useRealTimers(); + remote.body().error(new Error("fixture cleanup")); + await remote.call; + } + }); + + it("records a transport failure kind when the remote hostname does not resolve", async () => { + const remote = await startRemoteCallWithControlledBody(() => { + throw new ToolGatewayHttpError(502, "Remote MCP connection hostname did not resolve", "remote_http_dns_failed"); + }, false); + try { + expectGatewayError((await remote.call).error, 502, "remote_http_dns_failed"); + expect((await remote.failedEvent()).metadata).toMatchObject({ execution: { failureKind: "transport_failure" } }); + } finally { + vi.useRealTimers(); + } + }); + + it("records the bytes of an error response body that fails partway", async () => { + const remote = await startRemoteCallWithControlledBody((body) => new Response(body, { status: 503, headers: { "content-type": "text/plain" } })); + try { + const partial = "upstream overloaded, partial"; + remote.body().enqueue(new TextEncoder().encode(partial)); + remote.body().error(Object.assign(new Error("aborted"), { code: "ECONNRESET" })); + expectGatewayError((await remote.call).error, 502, "mcp_remote_fetch_failed"); + expect((await remote.failedEvent()).metadata).toMatchObject({ execution: { + failureKind: "transport_failure", + response: { httpStatus: 503, bodySizeBytes: Buffer.byteLength(partial) }, + } }); + } finally { + vi.useRealTimers(); + } + }); + const remoteFailureCases = [ { name: "HTTP status", diff --git a/server/src/services/mcp-http.ts b/server/src/services/mcp-http.ts index 80db7f9c17e..b43334a5553 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,19 +218,26 @@ 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; + 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"); buffer += decoder.decode(value, { stream: !done }); if (isStream) { 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 d8b2b01fb4e..1c1cad85772 100644 --- a/server/src/services/tool-gateway.ts +++ b/server/src/services/tool-gateway.ts @@ -47,6 +47,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"; @@ -143,6 +144,7 @@ import { readMcpHttpResponse, McpHttpResponseError, McpHttpInitializationError, + type McpHttpResponseProgress, mcpHttpRequestHeaders, parseMcpHttpResponseBody, } from "./mcp-http.js"; @@ -156,6 +158,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 { @@ -491,8 +494,55 @@ 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; +}; + +// Deliberate stops of a remote tool call. They are not failures, and a deadline +// that passes while they are recorded must not turn them into a timeout. +const REMOTE_TOOL_CALL_WORKFLOW_REASON_CODES = new Set([ + "elicitation_required", + "provider_interaction_required", + "oauth_insufficient_scope", + "oauth_refreshed_retry_required", +]); + +const REMOTE_TOOL_CALL_PROTOCOL_REASON_CODES = new Set([ + "mcp_remote_invalid_json", + "remote_mcp_malformed_response", + "mcp_remote_response_too_large", +]); + +function remoteToolCallFailureKind(reasonCode: string): RemoteHttpExecutionAudit["failureKind"] { + if (reasonCode === "tool_timeout") return "invocation_timeout"; + if (REMOTE_TOOL_CALL_PROTOCOL_REASON_CODES.has(reasonCode)) return "protocol_failure"; + if (reasonCode === "mcp_remote_fetch_failed" || reasonCode.startsWith("remote_http_")) return "transport_failure"; + return undefined; +} + +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; @@ -5920,6 +5970,7 @@ export function createToolGatewayService( async function readBoundedRemoteResponse( response: Response, + onBytes?: (bytes: number) => void, ): Promise { const contentLength = response.headers.get("content-length"); if ( @@ -5928,11 +5979,36 @@ export function createToolGatewayService( ) { throw responseTooLargeError(); } - const body = await response.text(); - if (Buffer.byteLength(body, "utf8") > MAX_REMOTE_MCP_RESPONSE_BYTES) { - throw responseTooLargeError(); + const reader = response.body?.getReader(); + // Injected HTTP transports can expose a buffered text response rather than a + // Web ReadableStream. Keep the same size check for that form. + if (!reader) { + const body = await response.text(); + onBytes?.(Buffer.byteLength(body, "utf8")); + if (Buffer.byteLength(body, "utf8") > MAX_REMOTE_MCP_RESPONSE_BYTES) { + throw responseTooLargeError(); + } + return body; + } + // Count each chunk as it arrives, so a body that fails partway still + // reports the bytes that were read. + const decoder = new TextDecoder(); + let body = ""; + let bytes = 0; + try { + while (true) { + const { value, done } = await reader.read(); + if (done) break; + bytes += value.byteLength; + onBytes?.(bytes); + if (bytes > MAX_REMOTE_MCP_RESPONSE_BYTES) throw responseTooLargeError(); + body += decoder.decode(value, { stream: true }); + } + return body + decoder.decode(); + } finally { + await reader.cancel().catch(() => undefined); + reader.releaseLock(); } - return body; } function malformedRemoteMcpResponse(): ToolGatewayHttpError { @@ -6214,6 +6290,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); } @@ -6498,9 +6575,21 @@ export function createToolGatewayService( // automatically: the failed call may have changed app data. if (mcpSession) forgetMcpHttpSession(mcpSession); } + // Record the response headers before the body is read, so a body that + // fails partway still has its status and content type in the evidence. + execution.response = { + httpStatus: response.status, + contentType: response.headers.get("content-type"), + bodySizeBytes: 0, + upstreamRequestId: + response.headers.get("x-request-id") ?? + response.headers.get("x-zapier-request-id") ?? + response.headers.get("traceparent"), + }; 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) { @@ -6517,16 +6606,10 @@ export function createToolGatewayService( if (request) await requestElicitationForRecordedToolCall({ session, tool, invocationId, request }); }, })) - : await readBoundedRemoteResponse(response); - execution.response = { - httpStatus: response.status, - contentType: response.headers.get("content-type"), - bodySizeBytes: Buffer.byteLength(body, "utf8"), - upstreamRequestId: - response.headers.get("x-request-id") ?? - response.headers.get("x-zapier-request-id") ?? - response.headers.get("traceparent"), - }; + : await readBoundedRemoteResponse(response, (bytes) => { + execution.response!.bodySizeBytes = bytes; + }); + execution.response.bodySizeBytes = Buffer.byteLength(body, "utf8"); if (isInsufficientConnectionScope(response, body)) { await markCallHealth("degraded", INSUFFICIENT_CONNECTION_SCOPE_MESSAGE); throw new ToolGatewayHttpError(403, INSUFFICIENT_CONNECTION_SCOPE_MESSAGE, "oauth_insufficient_scope", { @@ -6627,7 +6710,26 @@ export function createToolGatewayService( ); return { result, failureInstructions, headerSummary, execution }; } catch (error) { + execution.error = remoteHttpErrorEvidence(error); + execution.deadline = { timeoutMs: ms, aborted: controller.signal.aborted }; + if (execution.response && execution.stream) execution.response.bodySizeBytes = execution.stream.bytesReceived; + // A deliberate workflow stop, such as a tool question for the user, keeps + // its own reason even when the deadline passes while it is recorded. + if (error instanceof ToolGatewayHttpError && REMOTE_TOOL_CALL_WORKFLOW_REASON_CODES.has(error.reasonCode)) { + throw new ToolGatewayHttpError(error.status, error.message, error.reasonCode, { + ...error.details, + execution: error.details.execution ?? execution, + }); + } + if (controller.signal.aborted || (error instanceof Error && error.name === "AbortError")) { + if (mcpSession) forgetMcpHttpSession(mcpSession); + execution.failureKind = "invocation_timeout"; + 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"; 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"); @@ -6645,6 +6747,8 @@ export function createToolGatewayService( throw new ToolGatewayHttpError(error.status, error.message, error.code, { connectionId: connection.id, catalogEntryId: entry.id, execution }); } if (error instanceof ToolGatewayHttpError) { + const failureKind = remoteToolCallFailureKind(error.reasonCode); + if (failureKind) execution.failureKind = failureKind; throw new ToolGatewayHttpError( error.status, error.message, @@ -6655,19 +6759,7 @@ 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."); }