fix(everything): close streamable HTTP transports on SIGINT

src/everything/__tests__/streamableHttpShutdown.test.ts

@@ -0,0 +1,19 @@
+import { describe, expect, it, vi } from "vitest";
+import { closeActiveTransports } from "../transports/streamableHttpShutdown.js";
+
+describe("closeActiveTransports", () => {
+  it("closes and removes every active transport", async () => {
+    const firstClose = vi.fn<() => Promise<void>>().mockResolvedValue();
+    const secondClose = vi.fn<() => Promise<void>>().mockResolvedValue();
+    const transports = new Map([
+      ["first-session", { close: firstClose }],
+      ["second-session", { close: secondClose }],
+    ]);
+
+    await closeActiveTransports(transports);
+
+    expect(firstClose).toHaveBeenCalledOnce();
+    expect(secondClose).toHaveBeenCalledOnce();
+    expect(transports.size).toBe(0);
+  });
+});

src/everything/transports/streamableHttp.ts

@@ -6,6 +6,7 @@ import express, { Request, Response } from "express";
 import { createServer } from "../server/index.js";
 import { randomUUID } from "node:crypto";
 import cors from "cors";
+import { closeActiveTransports } from "./streamableHttpShutdown.js";
 
 // Simple in-memory event store for SSE resumability
 class InMemoryEventStore implements EventStore {
@@ -225,15 +226,7 @@ process.on("SIGINT", async () => {
   console.log("Shutting down server...");
 
   // Close all active transports to properly clean up resources
-  for (const sessionId in transports) {
-    try {
-      console.log(`Closing transport for session ${sessionId}`);
-      await transports.get(sessionId)!.close();
-      transports.delete(sessionId);
-    } catch (error) {
-      console.log(`Error closing transport for session ${sessionId}:`, error);
-    }
-  }
+  await closeActiveTransports(transports);
 
   console.log("Server shutdown complete");
   process.exit(0);

src/everything/transports/streamableHttpShutdown.ts

@@ -0,0 +1,17 @@
+type ClosableTransport = {
+  close: () => Promise<void>;
+};
+
+export async function closeActiveTransports(
+  transports: Map<string, ClosableTransport>
+): Promise<void> {
+  for (const [sessionId, transport] of transports) {
+    try {
+      console.log(`Closing transport for session ${sessionId}`);
+      await transport.close();
+      transports.delete(sessionId);
+    } catch (error) {
+      console.log(`Error closing transport for session ${sessionId}:`, error);
+    }
+  }
+}