fix(mcp): close the connection unless the incoming body was consumed to its end

Addresses the twelfth automated review finding: bodyUsed only means a body
was disturbed, so a route that read one chunk of an upload left the rest
paused on a keep-alive socket. The close decision now follows the incoming
message's end of stream.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

docs/specs/agent-environments/http-transport/verification.md

@@ -281,3 +281,10 @@ An eleventh finding (Codex): URL construction canonicalized dot segments, so
 differs after URL parsing, so routing always sees the path the client sent
 (test: `/public/../mcp`, `/public/%2e%2e/mcp` and `/./mcp` get `400`; `/card?x=1`
 still routes).
+
+A twelfth finding (Codex): `bodyUsed` means disturbed, not consumed, so a
+composed route that read one chunk of an upload left the rest paused on a
+keep-alive socket. The close decision now uses the incoming message's end of
+stream (test: a route that reads one chunk of a 256 KiB upload is answered
+with `Connection: close` and a closed socket; a route that reads it all keeps
+the connection).

src/server/http/serveStreamableHttp.test.ts

@@ -711,3 +711,54 @@ it("retries a shutdown hook that threw instead of marking it complete", async ()
   await slot.close();
   expect(attempts).toBe(2);
 });
+
+it("closes the connection when a route read only part of the body and keeps it after a full read", async () => {
+  const { served } = await start({
+    routes: [
+      {
+        method: "POST",
+        path: "/partial",
+        handle: async (request) => {
+          const reader = request.body?.getReader();
+          await reader?.read();
+          return new Response("stopped early");
+        },
+      },
+      {
+        method: "POST",
+        path: "/whole",
+        handle: async (request) =>
+          Response.json({ bytes: (await request.text()).length }),
+      },
+    ],
+  });
+  const url = new URL(served.url);
+  const agent = new Agent({ keepAlive: true, maxSockets: 1 });
+  const body = "x".repeat(256 * 1024);
+  try {
+    const partial = await keepAliveRequest(agent, new URL("/partial", url), {
+      method: "POST",
+      headers: {
+        "content-type": "text/plain",
+        "content-length": String(body.length),
+      },
+      body,
+    });
+    expect(partial.status).toBe(200);
+    expect(partial.headers.connection).toBe("close");
+    expect(partial.socketClosed).toBe(true);
+    const whole = await keepAliveRequest(agent, new URL("/whole", url), {
+      method: "POST",
+      headers: {
+        "content-type": "text/plain",
+        "content-length": String(body.length),
+      },
+      body,
+    });
+    expect(whole.status).toBe(200);
+    expect(whole.headers.connection).not.toBe("close");
+    expect(whole.socketClosed).toBe(false);
+  } finally {
+    agent.destroy();
+  }
+});

src/server/http/serveStreamableHttp.ts

@@ -398,15 +398,15 @@ const serveNodeRequest = async (
   response: ServerResponse,
 ): Promise<void> => {
   const { route, origin, logger } = target;
-  // A reply sent before an incoming body was read leaves that body on the socket, which the
-  // bridged stream keeps paused; closing the connection after the reply releases it. The
-  // incoming message decides, so a GET or HEAD that carried a body closes too.
+  // A reply sent before an incoming body was fully consumed leaves the rest on the socket, which
+  // the bridged stream keeps paused; closing the connection after the reply releases it. The
+  // incoming message decides (end of stream, not "body disturbed"), so a partially read upload
+  // and a GET or HEAD that carried a body close too.
   const declaresBody =
     request.headers["content-length"] !== undefined ||
     request.headers["transfer-encoding"] !== undefined;
   let webRequest: Request | undefined;
-  const unreadBody = (): boolean =>
-    declaresBody && !(webRequest?.bodyUsed ?? false);
+  const unreadBody = (): boolean => declaresBody && !request.readableEnded;
   try {
     try {
       webRequest = toWebRequest(request, response, origin());