fix(providers): reject malformed UTF-8 before MCP decoding

src/ida/IdaMcpConnection.ts

@@ -33,6 +33,10 @@ import {
   AnalysisResourceConstraintError,
 } from "../domain/analysisErrorCore.js";
 import { ProviderAdapterError } from "../domain/providerAdapterError.js";
+import {
+  McpHttpUtf8Error,
+  validateMcpHttpResponse,
+} from "../providers/Utf8McpResponse.js";
 
 /** Injectable MCP boundary used by the provider and transport conformance tests. */
 export interface IdaMcpConnection {
@@ -294,9 +298,15 @@ const readBoundedResponseText = async (
 };
 
 const diagnosticFetch =
-  (redactor: IdaDiagnosticRedactor): typeof fetch =>
+  (
+    redactor: IdaDiagnosticRedactor,
+    rejectTransport: (failure: McpHttpUtf8Error) => void,
+  ): typeof fetch =>
   async (input, init) => {
-    const response = await fetch(input, init);
+    const response = validateMcpHttpResponse(
+      await fetch(input, init),
+      rejectTransport,
+    );
     const readText = response.text.bind(response);
     let safeBody: string | undefined;
     if (!response.ok) {
@@ -356,6 +366,7 @@ export const createIdaMcpConnection = (
   )(config);
   let stderr = Buffer.alloc(0);
   let stderrBytes = 0;
+  let httpUtf8Failure: McpHttpUtf8Error | undefined;
   const capturedTransportOutput = () => {
     let start = 0;
     if (stderrBytes > stderr.length)
@@ -407,6 +418,13 @@ export const createIdaMcpConnection = (
         ? /ReadBuffer exceeded maximum size of (\d+) bytes/u.exec(cause.message)
         : null;
     if (cause instanceof IdaDiagnosticResourceError) throw cause;
+    if (cause instanceof McpHttpUtf8Error) throw cause;
+    if (
+      httpUtf8Failure !== undefined &&
+      cause instanceof SdkError &&
+      cause.code === SdkErrorCode.ConnectionClosed
+    )
+      throw httpUtf8Failure;
     let failure: Error;
     if (cause instanceof IdaHttpJsonError)
       failure = new Error(cause.message, { cause });
@@ -459,7 +477,7 @@ export const createIdaMcpConnection = (
       capturedOutput,
     });
   };
-  const transport =
+  const transport: StdioClientTransport | StreamableHTTPClientTransport =
     "command" in config
       ? new StdioClientTransport({
           command: config.command,
@@ -483,7 +501,10 @@ export const createIdaMcpConnection = (
         })
       : new StreamableHTTPClientTransport(new URL(config.url), {
           requestInit: { headers: config.headers, redirect: "error" },
-          fetch: diagnosticFetch(redactor),
+          fetch: diagnosticFetch(redactor, (failure) => {
+            httpUtf8Failure ??= failure;
+            void transport.close().catch(() => undefined);
+          }),
         });
   // Keep a bounded diagnostic tail for genuine failures; never persist ambient variables.
   if (transport instanceof StdioClientTransport)

src/jeb/JebMcpConnection.ts

@@ -180,7 +180,7 @@ export const createStreamableHttpJebMcpConnection: JebMcpConnectionFactory = (
       } catch (cause: unknown) {
         const failure = physicalFailure(
           "catalog",
-          deadline.failure("jeb_mcp_catalog", cause),
+          owner.responseFailure(deadline.failure("jeb_mcp_catalog", cause)),
         );
         owner.acquisitionFailure = { cause: failure };
         owner.signal.abort(failure);
@@ -237,7 +237,7 @@ export const createStreamableHttpJebMcpConnection: JebMcpConnectionFactory = (
       } catch (cause: unknown) {
         throw physicalFailure(
           "catalog",
-          deadline.failure("jeb_mcp_catalog", cause),
+          owner.responseFailure(deadline.failure("jeb_mcp_catalog", cause)),
         );
       } finally {
         deadline.dispose();
@@ -260,7 +260,10 @@ export const createStreamableHttpJebMcpConnection: JebMcpConnectionFactory = (
           { timeout: deadline.remainingMs(), signal: deadline.signal },
         );
       } catch (cause: unknown) {
-        throw physicalFailure("tool", deadline.failure(name, cause));
+        throw physicalFailure(
+          "tool",
+          owner.responseFailure(deadline.failure(name, cause)),
+        );
       } finally {
         deadline.dispose();
       }

src/jeb/JebMcpConnectionOwner.ts

@@ -1,11 +1,17 @@
 import {
   Client,
+  SdkError,
+  SdkErrorCode,
   StreamableHTTPClientTransport,
 } from "@modelcontextprotocol/client";
 import { AnalysisCleanupLease } from "../artifacts/AnalysisCleanupLease.js";
 import { ProviderCleanupError } from "../domain/providerCleanupError.js";
 import { SharedStartupAttempt } from "../process/SharedStartupAttempt.js";
 import type { Result } from "../domain/result.js";
+import {
+  McpHttpUtf8Error,
+  validateMcpHttpResponse,
+} from "../providers/Utf8McpResponse.js";
 
 /** Physical SDK failure captured before any subsequent resource cleanup. */
 export class JebMcpPhysicalFailure extends Error {
@@ -61,10 +67,17 @@ export class JebMcpConnectionOwner {
   #clientClosed = false;
   #transportClosed = false;
   #disposal: Promise<void> | undefined;
+  #httpUtf8Failure: McpHttpUtf8Error | undefined;
 
   constructor(endpoint: URL) {
     this.transport = new StreamableHTTPClientTransport(endpoint, {
       requestInit: { redirect: "error" },
+      fetch: async (input, init) =>
+        validateMcpHttpResponse(await fetch(input, init), (failure) => {
+          this.#httpUtf8Failure ??= failure;
+          this.tools = undefined;
+          void this.transport.close().catch(() => undefined);
+        }),
     });
     this.#clientResource = `${endpoint} (JEB MCP client)`;
     this.#transportResource = `${endpoint} (JEB MCP transport)`;
@@ -78,6 +91,15 @@ export class JebMcpConnectionOwner {
     return this.#disposal !== undefined;
   }
 
+  /** Attribute only the actual transport close caused by malformed response bytes. */
+  responseFailure(cause: unknown): unknown {
+    return this.#httpUtf8Failure !== undefined &&
+      cause instanceof SdkError &&
+      cause.code === SdkErrorCode.ConnectionClosed
+      ? this.#httpUtf8Failure
+      : cause;
+  }
+
   get cleanupResources(): readonly string[] {
     return [
       ...(this.#clientClosed ? [] : [this.#clientResource]),

src/providers/Utf8McpResponse.ts

@@ -0,0 +1,51 @@
+import { AnalysisProtocolError } from "../domain/analysisErrorCore.js";
+
+/** Authored byte-validation failure, distinct from arbitrary upstream protocol diagnostics. */
+export class McpHttpUtf8Error extends AnalysisProtocolError {
+  constructor(cause: unknown) {
+    super("MCP HTTP response body omitted: not valid UTF-8.", { cause });
+  }
+}
+
+/** Validate accepted MCP message bytes before Fetch or the SDK can replace malformed UTF-8. */
+export const validateMcpHttpResponse = (
+  response: Response,
+  rejectTransport: (failure: McpHttpUtf8Error) => void,
+): Response => {
+  // Failed HTTP responses belong to each adapter's existing diagnostic policy.
+  if (!response.ok || response.body === null) return response;
+  const decoder = new TextDecoder("utf-8", { fatal: true, ignoreBOM: true });
+  const validate = (bytes?: Uint8Array): void => {
+    try {
+      decoder.decode(bytes, { stream: bytes !== undefined });
+    } catch (cause: unknown) {
+      const failure = new McpHttpUtf8Error(cause);
+      // SDK 2.3.1 reports SSE read errors without settling pending tools/call.
+      // The lifetime owner closes the actual transport to reject that work.
+      rejectTransport(failure);
+      throw failure;
+    }
+  };
+  const body = response.body.pipeThrough(
+    new TransformStream<Uint8Array, Uint8Array>({
+      transform(bytes, controller) {
+        validate(bytes);
+        controller.enqueue(bytes);
+      },
+      flush() {
+        validate();
+      },
+    }),
+  );
+  const validated = new Response(body, {
+    status: response.status,
+    statusText: response.statusText,
+    headers: response.headers,
+  });
+  Object.defineProperties(validated, {
+    url: { value: response.url },
+    redirected: { value: response.redirected },
+    type: { value: response.type },
+  });
+  return validated;
+};

tests/boundary/providers/mcpHttpUtf8.test.ts

@@ -0,0 +1,217 @@
+import type { ServerResponse } from "node:http";
+import { setTimeout as delay } from "node:timers/promises";
+import { expect, it } from "vitest";
+import { z } from "zod";
+import { createVerifierHttpServer } from "../../../scripts/lib/verifier-http-server.mjs";
+import { createIdaMcpConnection } from "../../../src/ida/IdaMcpConnection.js";
+import { createStreamableHttpJebMcpConnection } from "../../../src/jeb/JebMcpConnection.js";
+import { AnalysisCancelledError } from "../../../src/domain/analysisErrorCore.js";
+
+const requestSchema = z.object({
+  id: z.union([z.string(), z.number()]).optional(),
+  method: z.string(),
+});
+const MARKER = "SOURCE_UTF8_BYTES";
+const payload = {
+  success: true,
+  text: `caf${MARKER}`,
+  source_url: "file:///selected/a%20b.js?query#fragment",
+};
+
+/** Actual HTTP framing and SDK parsing; this server is not an installed IDA/JEB engine. */
+const startFixture = async (format: "json" | "sse") => {
+  let fault: "valid" | "invalid" | "truncated" | "stalled" = "valid";
+  let announceStalled!: () => void;
+  let stalled = Promise.resolve();
+  let release!: () => void;
+  const gate = new Promise<void>((resolve) => {
+    release = resolve;
+  });
+  const stalledResponses: ServerResponse[] = [];
+  const { server, close } = createVerifierHttpServer(
+    async (request, response) => {
+      if (request.method !== "POST") {
+        request.resume();
+        response.writeHead(request.method === "DELETE" ? 200 : 405).end();
+        return;
+      }
+      const chunks: Buffer[] = [];
+      for await (const chunk of request) chunks.push(Buffer.from(chunk));
+      const message = requestSchema.parse(
+        JSON.parse(Buffer.concat(chunks).toString("utf8")),
+      );
+      if (message.id === undefined) {
+        response.writeHead(202).end();
+        return;
+      }
+      const result =
+        message.method === "initialize"
+          ? {
+              protocolVersion: "2025-11-25",
+              capabilities: { tools: {} },
+              serverInfo: { name: "utf8-http-boundary", version: "1" },
+            }
+          : message.method === "tools/list"
+            ? {
+                tools: ["get_client_information"].map((name) => ({
+                  name,
+                  inputSchema: { type: "object" },
+                })),
+              }
+            : {
+                content: [{ type: "text", text: JSON.stringify(payload) }],
+                structuredContent: payload,
+              };
+      const json = JSON.stringify({ jsonrpc: "2.0", id: message.id, result });
+      if (message.method !== "tools/call") {
+        response.writeHead(200, { "Content-Type": "application/json" });
+        response.end(json);
+        return;
+      }
+      const selectedFault = fault;
+      response.writeHead(200, {
+        "Content-Type":
+          format === "json" ? "application/json" : "text/event-stream",
+        "X-Observed-Header": "preserved",
+      });
+      response.flushHeaders();
+      const prefix = format === "sse" ? "event: message\ndata: " : "";
+      const suffix = format === "sse" ? "\n\n" : "";
+      const source = `${prefix}${json}${suffix}`;
+      const start = source.indexOf(MARKER);
+      if (selectedFault === "stalled") {
+        stalledResponses.push(response);
+        response.write(source.slice(0, start));
+        announceStalled();
+        await gate;
+        response.end();
+        return;
+      }
+      const replacement =
+        selectedFault === "invalid" ? Buffer.from([0xe9]) : Buffer.from("é");
+      const parts = source.split(MARKER);
+      const bytes = Buffer.concat(
+        parts.flatMap((part, index) =>
+          index === 0 ? [Buffer.from(part)] : [replacement, Buffer.from(part)],
+        ),
+      );
+      // Leave the first multibyte character incomplete in one actual HTTP chunk.
+      response.write(bytes.subarray(0, start + 1));
+      await delay(10);
+      if (selectedFault === "truncated") response.end();
+      else response.end(bytes.subarray(start + 1));
+    },
+  );
+  await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
+  const address = server.address();
+  if (address === null || typeof address === "string")
+    throw new Error("Missing HTTP listener");
+  return {
+    endpoint: `http://127.0.0.1:${address.port}/mcp`,
+    setFault: (value: typeof fault) => {
+      fault = value;
+      if (value === "stalled")
+        stalled = new Promise<void>((resolve) => {
+          announceStalled = resolve;
+        });
+    },
+    get stalled() {
+      return stalled;
+    },
+    async close() {
+      release();
+      for (const response of stalledResponses) response.destroy();
+      await close();
+    },
+  };
+};
+
+it.each(
+  (["ida", "jeb"] as const).flatMap((provider) =>
+    (["json", "sse"] as const).map((format) => ({ provider, format })),
+  ),
+)(
+  "rejects malformed UTF-8 over real $provider $format HTTP without altering valid source or cancellation",
+  async ({ provider, format }) => {
+    const fixture = await startFixture(format);
+    const createConnection = () => {
+      if (provider === "ida") {
+        const connection = createIdaMcpConnection(
+          {
+            url: fixture.endpoint,
+            headers: {},
+            mode: "attached",
+            protocol: "native",
+            timeoutMs: 1_000,
+          },
+          {},
+        );
+        return {
+          connect: () => connection.connect(),
+          close: () => connection.close(),
+          call: (signal?: AbortSignal) =>
+            connection.call("get_client_information", {}, { signal }),
+        };
+      }
+      const connection = createStreamableHttpJebMcpConnection(
+        new URL(fixture.endpoint),
+        {
+          discoveryTimeoutMs: 1_000,
+          requestTimeoutMs: 1_000,
+        },
+      );
+      return {
+        connect: () => connection.connect(),
+        close: () => connection.close(),
+        call: (signal?: AbortSignal) =>
+          connection.call("get_client_information", {}, signal),
+      };
+    };
+    let connection = createConnection();
+    const call = (signal?: AbortSignal) => connection.call(signal);
+    try {
+      await connection.connect();
+      const expected = { ...payload, text: "café" };
+      expect(await call()).toEqual(expected);
+      for (const fault of ["invalid", "truncated"] as const) {
+        fixture.setFault(fault);
+        await expect(call()).rejects.toThrow("not valid UTF-8");
+        if (provider === "ida") {
+          // IDA's session owner retires failed connections rather than restarting SDK transports.
+          await connection.close();
+          connection = createConnection();
+          await connection.connect();
+        }
+      }
+      fixture.setFault("valid");
+      expect(await call()).toEqual(expected);
+      fixture.setFault("stalled");
+      const sibling = call().catch((failure: unknown) => failure);
+      await fixture.stalled;
+      fixture.setFault("valid");
+      expect(await call()).toEqual(expected);
+      fixture.setFault("invalid");
+      await expect(call()).rejects.toThrow("not valid UTF-8");
+      expect(await sibling).toMatchObject({ _tag: "AnalysisProtocolError" });
+      if (provider === "ida") {
+        await connection.close();
+        connection = createConnection();
+        await connection.connect();
+      }
+      fixture.setFault("stalled");
+      const controller = new AbortController();
+      const reason = new Error("caller stopped the incomplete source reply");
+      const pending = call(controller.signal).catch(
+        (failure: unknown) => failure,
+      );
+      await fixture.stalled;
+      controller.abort(reason);
+      const failure = await pending;
+      expect(failure).toBeInstanceOf(AnalysisCancelledError);
+      expect(failure).toMatchObject({ cause: reason });
+    } finally {
+      await connection.close();
+      await fixture.close();
+    }
+  },
+);