From 8dd88fcf8dfebcc10dfcf94c7b4b2217ef370fe3 Mon Sep 17 00:00:00 2001 From: OpenAI Codex Date: Tue, 14 Jul 2026 18:35:41 +0100 Subject: [PATCH] fix(everything): return errors for missing SSE sessions --- src/everything/__tests__/sse.test.ts | 66 ++++++++++++++++ src/everything/transports/sse.ts | 114 +++++++++++++++------------ 2 files changed, 129 insertions(+), 51 deletions(-) create mode 100644 src/everything/__tests__/sse.test.ts diff --git a/src/everything/__tests__/sse.test.ts b/src/everything/__tests__/sse.test.ts new file mode 100644 index 0000000000..c8e0eea8b0 --- /dev/null +++ b/src/everything/__tests__/sse.test.ts @@ -0,0 +1,66 @@ +import { afterEach, describe, expect, it } from "vitest"; +import { + createServer as createHttpServer, + request as httpRequest, + type IncomingMessage, + type Server, +} from "node:http"; +import { createSseApp } from "../transports/sse.js"; + +const servers: Server[] = []; + +afterEach(async () => { + await Promise.all( + servers.splice(0).map( + (server) => + new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())); + }) + ) + ); +}); + +async function request( + path: string, + method = "GET" +): Promise<{ status: number; body: string }> { + const server = createHttpServer(createSseApp()); + servers.push(server); + await new Promise((resolve) => server.listen(0, resolve)); + const address = server.address(); + if (!address || typeof address === "string") { + throw new Error("Test server did not expose a TCP address"); + } + + return new Promise((resolve, reject) => { + const req = httpRequest( + { host: "127.0.0.1", port: address.port, path, method }, + (res: IncomingMessage) => { + const chunks: Buffer[] = []; + res.on("data", (chunk: Buffer) => chunks.push(chunk)); + res.on("end", () => + resolve({ + status: res.statusCode ?? 0, + body: Buffer.concat(chunks).toString(), + }) + ); + } + ); + req.on("error", reject); + req.end(); + }); +} + +describe("SSE missing-session handling", () => { + it("returns 404 for GET requests with an unknown session", async () => { + const response = await request("/sse?sessionId=missing"); + expect(response.status).toBe(404); + expect(response.body).toContain("No transport found"); + }); + + it("returns 404 for POST requests with an unknown session", async () => { + const response = await request("/message?sessionId=missing", "POST"); + expect(response.status).toBe(404); + expect(response.body).toContain("No transport found"); + }); +}); diff --git a/src/everything/transports/sse.ts b/src/everything/transports/sse.ts index 2406db7cfa..1d5073eb03 100644 --- a/src/everything/transports/sse.ts +++ b/src/everything/transports/sse.ts @@ -1,43 +1,48 @@ import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js"; -import express from "express"; +import express, { type Express } from "express"; import { createServer } from "../server/index.js"; import cors from "cors"; -console.error("Starting SSE server..."); +type ServerFactory = typeof createServer; -// Express app with permissive CORS for testing with Inspector direct connect mode -const app = express(); -app.use( - cors({ - origin: "*", // use "*" with caution in production - methods: "GET,POST", - preflightContinue: false, - optionsSuccessStatus: 204, - }) -); +export function createSseApp( + transports: Map = new Map(), + serverFactory: ServerFactory = createServer +): Express { + // Express app with permissive CORS for testing with Inspector direct connect mode + const app = express(); + app.use( + cors({ + origin: "*", // use "*" with caution in production + methods: "GET,POST", + preflightContinue: false, + optionsSuccessStatus: 204, + }) + ); -// Map sessionId to transport for each client -const transports: Map = new Map< - string, - SSEServerTransport ->(); + // Handle GET requests for new SSE streams + app.get("/sse", async (req, res) => { + // Session Id should not exist for GET /sse requests + if (req?.query?.sessionId) { + const sessionId = String(req.query.sessionId); + const transport = transports.get(sessionId); + if (!transport) { + res.status(404).json({ + error: `No transport found for sessionId ${sessionId}`, + }); + return; + } -// Handle GET requests for new SSE streams -app.get("/sse", async (req, res) => { - let transport: SSEServerTransport; - const { server, cleanup } = createServer(); + console.error( + "Client Reconnecting? This shouldn't happen; when client has a sessionId, GET /sse should not be called again.", + transport.sessionId + ); + return; + } - // Session Id should not exist for GET /sse requests - if (req?.query?.sessionId) { - const sessionId = req?.query?.sessionId as string; - transport = transports.get(sessionId) as SSEServerTransport; - console.error( - "Client Reconnecting? This shouldn't happen; when client has a sessionId, GET /sse should not be called again.", - transport.sessionId - ); - } else { + const { server, cleanup } = serverFactory(); // Create and store transport for the new session - transport = new SSEServerTransport("/message", res); + const transport = new SSEServerTransport("/message", res); transports.set(transport.sessionId, transport); // Connect server to transport @@ -52,26 +57,33 @@ app.get("/sse", async (req, res) => { transports.delete(sessionId); cleanup(sessionId); }; - } -}); + }); -// Handle POST requests for client messages -app.post("/message", async (req, res) => { - // Session Id should exist for POST /message requests - const sessionId = req?.query?.sessionId as string; + // Handle POST requests for client messages + app.post("/message", async (req, res) => { + // Session Id should exist for POST /message requests + const sessionId = req?.query?.sessionId as string | undefined; - // Get the transport for this session and use it to handle the request - const transport = transports.get(sessionId); - if (transport) { - console.error("Client Message from", sessionId); - await transport.handlePostMessage(req, res); - } else { - console.error(`No transport found for sessionId ${sessionId}`); - } -}); + // Get the transport for this session and use it to handle the request + const transport = sessionId ? transports.get(sessionId) : undefined; + if (transport) { + console.error("Client Message from", sessionId); + await transport.handlePostMessage(req, res); + return; + } -// Start the express server -const PORT = process.env.PORT || 3001; -app.listen(PORT, () => { - console.error(`Server is running on port ${PORT}`); -}); + res.status(404).json({ + error: `No transport found for sessionId ${sessionId ?? ""}`, + }); + }); + + return app; +} + +if (process.env.NODE_ENV !== "test") { + console.error("Starting SSE server..."); + const PORT = process.env.PORT || 3001; + createSseApp().listen(PORT, () => { + console.error(`Server is running on port ${PORT}`); + }); +}