Fix Streamable HTTP MCP connection leaks

packages/plugins/mcp/src/sdk/connection.test.ts

@@ -0,0 +1,94 @@
+import { describe, it } from "@effect/vitest";
+import { Effect, Layer, Option, Predicate, Schema } from "effect";
+import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/unstable/http";
+
+import { createMcpConnector } from "./connection";
+
+const JsonRpcId = Schema.Union([Schema.String, Schema.Number, Schema.Null]);
+const JsonRpcRequest = Schema.Struct({
+  id: Schema.optional(JsonRpcId),
+  method: Schema.String,
+});
+type JsonRpcRequest = typeof JsonRpcRequest.Type;
+
+const decodeJsonRpcRequest = Schema.decodeUnknownOption(Schema.fromJsonString(JsonRpcRequest));
+
+const initializeResponse = (request: HttpClientRequest.HttpClientRequest): Response => {
+  const body = Predicate.isTagged(request.body, "Uint8Array")
+    ? new TextDecoder().decode(request.body.body)
+    : "";
+  const rpc = Option.getOrElse(decodeJsonRpcRequest(body), () => ({
+    id: undefined,
+    method: "",
+  }));
+  if (rpc.method !== "initialize") return new Response(null, { status: 202 });
+  return new Response(
+    JSON.stringify({
+      jsonrpc: "2.0",
+      id: rpc.id ?? null,
+      result: {
+        protocolVersion: "2025-06-18",
+        capabilities: {},
+        serverInfo: { name: "connection-fixture", version: "1.0.0" },
+      },
+    }),
+    {
+      headers: {
+        "content-type": "application/json",
+        "mcp-session-id": "connection-fixture-session",
+      },
+    },
+  );
+};
+
+describe("MCP HTTP connection cleanup", () => {
+  it.effect("aborts an open GET SSE response stream when the connection closes", () =>
+    Effect.gen(function* () {
+      let resolveGetStarted!: () => void;
+      const getStarted = new Promise<void>((resolve) => {
+        resolveGetStarted = resolve;
+      });
+      let resolveCancelled!: () => void;
+      const cancelled = new Promise<void>((resolve) => {
+        resolveCancelled = resolve;
+      });
+      const sseBody = new ReadableStream<Uint8Array>({
+        start(controller) {
+          controller.enqueue(new TextEncoder().encode(": priming\n\n"));
+        },
+        cancel() {
+          resolveCancelled();
+        },
+      });
+      const httpClientLayer = Layer.succeed(HttpClient.HttpClient)(
+        HttpClient.make((request) =>
+          Effect.sync(() => {
+            if (request.method === "POST") {
+              return HttpClientResponse.fromWeb(request, initializeResponse(request));
+            }
+            if (request.method === "GET") {
+              resolveGetStarted();
+              return HttpClientResponse.fromWeb(
+                request,
+                new Response(sseBody, {
+                  headers: { "content-type": "text/event-stream" },
+                }),
+              );
+            }
+            return HttpClientResponse.fromWeb(request, new Response(null, { status: 405 }));
+          }),
+        ),
+      );
+      const connection = yield* createMcpConnector({
+        transport: "remote",
+        endpoint: "https://connection-fixture.example/mcp",
+        remoteTransport: "streamable-http",
+        httpClientLayer,
+      });
+
+      yield* Effect.promise(() => getStarted);
+      yield* Effect.promise(() => connection.close());
+      yield* Effect.promise(() => cancelled);
+    }),
+  );
+});

packages/plugins/mcp/src/sdk/connection.ts

@@ -123,6 +123,17 @@ const abortError = (signal: AbortSignal): unknown => {
   return error;
 };
 
+const awaitAbort = (signal: AbortSignal): Effect.Effect<void> =>
+  Effect.callback<void>((resume) => {
+    if (signal.aborted) {
+      resume(Effect.void);
+      return;
+    }
+    const onAbort = () => resume(Effect.void);
+    signal.addEventListener("abort", onAbort, { once: true });
+    return Effect.sync(() => signal.removeEventListener("abort", onAbort));
+  });
+
 const fetchFromHttpClientLayer = (
   httpClientLayer: Layer.Layer<HttpClient.HttpClient>,
 ): FetchLike => {
@@ -142,7 +153,11 @@ const fetchFromHttpClientLayer = (
       const body =
         response.status === 204 || response.status === 205 || response.status === 304
           ? null
-          : Stream.toReadableStream(response.stream);
+          : Stream.toReadableStream(
+              init?.signal
+                ? Stream.interruptWhen(response.stream, awaitAbort(init.signal))
+                : response.stream,
+            );
       return new Response(body, {
         status: response.status,
         headers: responseHeaders,
@@ -156,7 +171,10 @@ const fetchFromHttpClientLayer = (
     // tagged error from the fetch adapter (a true runtime edge: the SDK
     // consumes promise rejections) so it reaches the invoke/connect catch
     // sites verbatim.
-    const promise = Effect.runPromise(effect).then((response) => {
+    const signal = init?.signal;
+    // oxlint-disable-next-line executor/no-promise-reject -- boundary: Fetch-compatible adapter mirrors abort rejection semantics
+    if (signal?.aborted) return Promise.reject(abortError(signal));
+    const promise = Effect.runPromise(effect, { signal: signal ?? undefined }).then((response) => {
       if (response.status === 403) {
         const challenge = response.headers.get("www-authenticate");
         if (
@@ -172,14 +190,10 @@ const fetchFromHttpClientLayer = (
       }
       return response;
     });
-    if (!init?.signal) return promise;
-    // oxlint-disable-next-line executor/no-promise-reject -- boundary: Fetch-compatible adapter mirrors abort rejection semantics
-    if (init.signal.aborted) return Promise.reject(abortError(init.signal));
+    if (!signal) return promise;
     const aborted = new Promise<never>((_, reject) => {
       // oxlint-disable-next-line executor/no-promise-reject -- boundary: Fetch-compatible adapter races the Effect request against AbortSignal
-      init.signal?.addEventListener("abort", () => reject(abortError(init.signal!)), {
-        once: true,
-      });
+      signal.addEventListener("abort", () => reject(abortError(signal)), { once: true });
     });
     return Promise.race([promise, aborted]);
   };