fix(server): complete remote MCP calls over event streams

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<Uint8Array>({
+      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<Uint8Array>({
+        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<Uint8Array>({
+        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({

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<void>();
+      let bodyController!: ReadableStreamDefaultController<Uint8Array>;
+      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<Uint8Array>({
+            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",

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<string, unknown>) => Promise<void> } = {},
+  options: {
+    maxBytes?: number;
+    onRequest?: (message: Record<string, unknown>) => Promise<void>;
+    onProgress?: (progress: McpHttpResponseProgress) => void;
+  } = {},
 ): Promise<unknown> {
+  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<unknown | undefined> => {
     if (!message || typeof message !== "object") return undefined;
     const record = message as Record<string, unknown>;
-    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.

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) {

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<unknown>()): 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(