From 62f8c5701c97bb461bb90520726db5e378d8a3d7 Mon Sep 17 00:00:00 2001
From: OpenAI Codex <codex@openai.com>
Date: Tue, 14 Jul 2026 18:37:22 +0100
Subject: [PATCH] fix(everything): close streamable HTTP transports on SIGINT

---
 .../__tests__/streamableHttpShutdown.test.ts  | 19 +++++++++++++++++++
 src/everything/transports/streamableHttp.ts   | 11 ++---------
 .../transports/streamableHttpShutdown.ts      | 17 +++++++++++++++++
 3 files changed, 38 insertions(+), 9 deletions(-)
 create mode 100644 src/everything/__tests__/streamableHttpShutdown.test.ts
 create mode 100644 src/everything/transports/streamableHttpShutdown.ts

diff --git a/src/everything/__tests__/streamableHttpShutdown.test.ts b/src/everything/__tests__/streamableHttpShutdown.test.ts
new file mode 100644
index 0000000000..a03e8ce777
--- /dev/null
+++ b/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);
+  });
+});
diff --git a/src/everything/transports/streamableHttp.ts b/src/everything/transports/streamableHttp.ts
index 2e79abc554..81b4d184f5 100644
--- a/src/everything/transports/streamableHttp.ts
+++ b/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);
diff --git a/src/everything/transports/streamableHttpShutdown.ts b/src/everything/transports/streamableHttpShutdown.ts
new file mode 100644
index 0000000000..1a99cce39b
--- /dev/null
+++ b/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);
+    }
+  }
+}