From 00150e59c1cfaf5f1e1f18300a2ee92b5306b186 Mon Sep 17 00:00:00 2001 From: "pavel.guseynov" Date: Sat, 7 Nov 2026 15:56:50 +0500 Subject: [PATCH] fix(server): keep tool questions and classify every remote call failure - A deliberate stop of a remote tool call (elicitation_required, provider_interaction_required, oauth_insufficient_scope) keeps its own reason when the call deadline passes while it is recorded. Before, the deadline check came first and turned a tool question into tool_timeout. - An error that is already a ToolGatewayHttpError now gets a failure kind: tool_timeout is invocation_timeout, a malformed, invalid or oversized response is protocol_failure, and remote_http_* and mcp_remote_fetch_failed are transport_failure. - A non-success response body is read as a stream, so a body that fails partway records the bytes that arrived. The reader also stops at the size limit and cancels the body, as the tools/call reader does. Co-Authored-By: Claude Opus 5.5 --- server/src/__tests__/tool-gateway.test.ts | 70 +++++++++++++++++++++++ server/src/services/tool-gateway.ts | 69 ++++++++++++++++++++-- 2 files changed, 134 insertions(+), 5 deletions(-) diff --git a/server/src/__tests__/tool-gateway.test.ts b/server/src/__tests__/tool-gateway.test.ts index 51562c33b3c..9a5c4a83a1d 100644 --- a/server/src/__tests__/tool-gateway.test.ts +++ b/server/src/__tests__/tool-gateway.test.ts @@ -4651,6 +4651,76 @@ rl.on("line", (line) => { }); } + 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/tool-gateway.ts b/server/src/services/tool-gateway.ts index 089137875fb..084c96610df 100644 --- a/server/src/services/tool-gateway.ts +++ b/server/src/services/tool-gateway.ts @@ -507,6 +507,27 @@ type RemoteHttpErrorEvidence = { 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", +]); + +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); @@ -5942,6 +5963,7 @@ export function createToolGatewayService( async function readBoundedRemoteResponse( response: Response, + onBytes?: (bytes: number) => void, ): Promise { const contentLength = response.headers.get("content-length"); if ( @@ -5950,11 +5972,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 { @@ -6552,7 +6599,9 @@ export function createToolGatewayService( if (request) await requestElicitationForRecordedToolCall({ session, tool, invocationId, request }); }, })) - : await readBoundedRemoteResponse(response); + : 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); @@ -6657,6 +6706,14 @@ export function createToolGatewayService( 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"; @@ -6683,6 +6740,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,