From 2213ce75702be15b4d1b2d5e309ef8d05db0336f Mon Sep 17 00:00:00 2001 From: Donald Date: Wed, 28 Oct 2026 03:05:10 +0000 Subject: [PATCH] fix: reclaim disconnected gateway MCP sessions --- docs/deployment.md | 13 ++ src/mcp-http.ts | 148 +++++++++++++++++++++++ src/mcp-session-lifecycle.ts | 63 ++++++++++ src/server.ts | 134 ++------------------ test/mcp-http.test.ts | 228 +++++++++++++++++++++++++++++++++++ 5 files changed, 465 insertions(+), 121 deletions(-) create mode 100644 src/mcp-http.ts create mode 100644 src/mcp-session-lifecycle.ts create mode 100644 test/mcp-http.test.ts diff --git a/docs/deployment.md b/docs/deployment.md index dfabfde..c20be5e 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -128,6 +128,19 @@ https://approval.justconn.top/downloads/dingmcp/install.md ## 运维检查 +网关会话上限为每个已认证用户 8 个、全局 1000 个，同时计入尚未完成初始化的连接。 +现有 Bridge 客户端退出或刷新登录时，即使没有发送 MCP DELETE，网关也会在观察到 +Streamable HTTP GET 长连接断开后保留 60 秒重连宽限期，再回收遗留会话；后台每 15 秒 +检查一次，新连接进入时也会检查，因此通常在断开后的 60–75 秒内释放名额。 +宽限期内恢复长连接或继续发送请求会保留原会话；仅使用 POST 的客户端仍使用 2 小时 +闲置期限。正常连接中的长会话不按闲置期限清理，最长存活时间仍为 24 小时。 +实际请求处理结束前不会因为断线、取消或超时回收会话，也不会重试任何审批写入。 +回收状态只保存在内存中，不记录审批参数、正文、Token 或用户身份到日志。 + +如果 8 个会话确实仍在连接，网关仍返回 HTTP 429 和 `Retry-After: 60`，需要关闭不用的 +Agent 连接再重试。旧客户端可能仍将 429 显示为“审批网关暂时无法连接”；此次服务端 +修复无需用户更新客户端，但不会改变客户端内置的错误文案。 + ```bash systemctl --user status dingtalk-approval-mcp.service curl --fail http://127.0.0.1:3010/healthz diff --git a/src/mcp-http.ts b/src/mcp-http.ts new file mode 100644 index 0000000..d4492dc --- /dev/null +++ b/src/mcp-http.ts @@ -0,0 +1,148 @@ +import { randomUUID } from "node:crypto"; +import type { Express, Request, Response, RequestHandler } from "express"; +import type { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; +import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js"; +import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js"; +import { isInitializeRequest } from "@modelcontextprotocol/sdk/types.js"; +import type { AuthenticatedDingTalkUser } from "./xft.js"; +import { McpSessionLifecycle } from "./mcp-session-lifecycle.js"; + +type TransportEntry = + | { kind: "streamable"; transport: StreamableHTTPServerTransport; userId: string; lifecycle: McpSessionLifecycle } + | { kind: "sse"; transport: SSEServerTransport; userId: string; lifecycle: McpSessionLifecycle }; + +const MCP_SESSION_MAX_PER_USER = 8; +const MCP_SESSION_MAX_TOTAL = 1000; + +interface McpTransportOptions { + authenticate?: RequestHandler; + // Production always resolves this from verified OAuth, never request parameters. + requestUser: (req: Request) => AuthenticatedDingTalkUser; + createServer: (user: AuthenticatedDingTalkUser, lifecycle: McpSessionLifecycle) => McpServer; + now?: () => number; +} + +export function installMcpTransports(app: Express, options: McpTransportOptions) { + const now = options.now ?? Date.now; + const transports = new Map(); + let pendingTransportSessions = 0; + const pendingTransportSessionsByUser = new Map(); + const closeTransport = (id: string, entry: TransportEntry) => { + transports.delete(id); + void entry.transport.close().catch(() => undefined); + }; + const cleanupTransports = () => { + for (const [id, entry] of transports) { + if (entry.lifecycle.expired()) { + closeTransport(id, entry); + } + } + }; + const sessionCapacityAvailable = (userId: string) => transports.size + pendingTransportSessions < MCP_SESSION_MAX_TOTAL + && [...transports.values()].filter((entry) => entry.userId === userId).length + + (pendingTransportSessionsByUser.get(userId) ?? 0) < MCP_SESSION_MAX_PER_USER; + const reserveSession = (userId: string) => { + pendingTransportSessions += 1; + pendingTransportSessionsByUser.set(userId, (pendingTransportSessionsByUser.get(userId) ?? 0) + 1); + }; + const releaseSessionReservation = (userId: string) => { + pendingTransportSessions = Math.max(0, pendingTransportSessions - 1); + const remaining = Math.max(0, (pendingTransportSessionsByUser.get(userId) ?? 1) - 1); + if (remaining === 0) pendingTransportSessionsByUser.delete(userId); + else pendingTransportSessionsByUser.set(userId, remaining); + }; + const transportCleanupTimer = setInterval(cleanupTransports, 15_000); + transportCleanupTimer.unref(); + + const mcpHandler = async (req: Request, res: Response) => { + let reservedUserId: string | undefined; + try { + cleanupTransports(); + const user = options.requestUser(req); + const sessionId = req.get("mcp-session-id"); + let entry = sessionId ? transports.get(sessionId) : undefined; + if (entry && entry.kind !== "streamable") { + res.status(400).json({ error: "Invalid MCP session transport" }); + return; + } + if (entry && entry.userId !== user.userId) { + res.status(403).json({ error: "MCP session belongs to another authenticated user" }); + return; + } + if (entry) entry.lifecycle.touch(); + if (!entry && req.method === "POST" && !sessionId && isInitializeRequest(req.body)) { + if (!sessionCapacityAvailable(user.userId)) { + res.set("Retry-After", "60").status(429).json({ error: "MCP session limit reached; close an old Agent session and retry" }); + return; + } + reserveSession(user.userId); + reservedUserId = user.userId; + const transport: StreamableHTTPServerTransport = new StreamableHTTPServerTransport({ + sessionIdGenerator: () => randomUUID(), + onsessioninitialized: (id) => { + transports.set(id, entry!); + // The live entry now owns the slot; do not count its reservation twice. + releaseSessionReservation(user.userId); + reservedUserId = undefined; + }, + }); + transport.onclose = () => { if (transport.sessionId) transports.delete(transport.sessionId); }; + entry = { kind: "streamable", transport, userId: user.userId, lifecycle: new McpSessionLifecycle(now) }; + await options.createServer(user, entry.lifecycle).connect(transport); + } + if (!entry || entry.kind !== "streamable") { + res.status(sessionId ? 404 : 400).json({ error: "Missing or invalid MCP session" }); + return; + } + if (req.method === "GET") entry.lifecycle.trackStream(res); + await entry.transport.handleRequest(req, res, req.body); + } catch (error) { sendServerError(res, error); } + finally { if (reservedUserId) releaseSessionReservation(reservedUserId); } + }; + app.all("/mcp", ...(options.authenticate ? [options.authenticate, mcpHandler] : [mcpHandler])); + + const sseHandler = async (req: Request, res: Response) => { + const user = options.requestUser(req); + cleanupTransports(); + if (!sessionCapacityAvailable(user.userId)) { + res.set("Retry-After", "60").status(429).send("MCP session limit reached"); + return; + } + const transport = new SSEServerTransport("/messages", res); + const lifecycle = new McpSessionLifecycle(now); + lifecycle.trackStream(res); + transports.set(transport.sessionId, { kind: "sse", transport, userId: user.userId, lifecycle }); + res.on("close", () => transports.delete(transport.sessionId)); + await options.createServer(user, lifecycle).connect(transport); + }; + app.get("/sse", ...(options.authenticate ? [options.authenticate, sseHandler] : [sseHandler])); + const messagesHandler = async (req: Request, res: Response) => { + const user = options.requestUser(req); + const entry = transports.get(String(req.query.sessionId ?? "")); + if (!entry || entry.kind !== "sse") { + res.status(404).send("Invalid SSE session"); + return; + } + if (entry.userId !== user.userId) { + res.status(403).send("Invalid SSE session"); + return; + } + entry.lifecycle.touch(); + await entry.transport.handlePostMessage(req, res, req.body); + }; + app.post("/messages", ...(options.authenticate ? [options.authenticate, messagesHandler] : [messagesHandler])); + return { + cleanup: cleanupTransports, + async close() { + clearInterval(transportCleanupTimer); + const entries = [...transports.values()]; + transports.clear(); + await Promise.all(entries.map((entry) => entry.transport.close())); + }, + }; +} + +function sendServerError(res: Response, _error: unknown): void { + console.error("[dingtalk-approval-mcp] MCP transport request failed"); + if (!res.headersSent) res.status(500).json({ error: "Internal server error" }); +} diff --git a/src/mcp-session-lifecycle.ts b/src/mcp-session-lifecycle.ts new file mode 100644 index 0000000..0c9b097 --- /dev/null +++ b/src/mcp-session-lifecycle.ts @@ -0,0 +1,63 @@ +import type { Response } from "express"; +import type { Server } from "@modelcontextprotocol/sdk/server/index.js"; + +const DISCONNECT_GRACE_MS = 60_000; +const IDLE_TTL_MS = 2 * 60 * 60_000; +const MAX_AGE_MS = 24 * 60 * 60_000; + +/** Only transport metadata lives here; never approval arguments or results. */ +export class McpSessionLifecycle { + private readonly createdAt: number; + private lastSeenAt: number; + private disconnectedAt?: number; + private openStreams = 0; + private activeRequests = 0; + + constructor(private readonly now: () => number = Date.now) { + this.createdAt = this.lastSeenAt = now(); + } + + /** + * Install before registering tools, using the SDK's public handler API. + * Track actual handler completion, not HTTP close or JSON-RPC replies: + * cancelled requests may finish upstream without ever sending a reply. + */ + trackRequests(server: Server): void { + const setRequestHandler = server.setRequestHandler.bind(server); + server.setRequestHandler = (schema, handler) => { + setRequestHandler(schema, async (request, extra) => { + this.activeRequests += 1; + try { return await handler(request, extra); } + finally { + this.activeRequests -= 1; + this.touch(); + } + }); + }; + } + + touch(): void { this.lastSeenAt = this.now(); } + + trackStream(res: Response): void { + this.openStreams += 1; + res.once("close", () => { + this.openStreams -= 1; + // A rejected duplicate GET (409) must not start the disconnect timer. + if (res.statusCode === 200) { + this.disconnectedAt = this.now(); + } + }); + } + + expired(): boolean { + if (this.activeRequests > 0) return false; + const now = this.now(); + if (now - this.createdAt >= MAX_AGE_MS) return true; + if (this.openStreams > 0) return false; + if (this.disconnectedAt !== undefined + && now - Math.max(this.disconnectedAt, this.lastSeenAt) >= DISCONNECT_GRACE_MS) return true; + // GET streams are optional in Streamable HTTP. POST-only clients retain + // the existing idle timeout instead of being mistaken for dead clients. + return now - this.lastSeenAt >= IDLE_TTL_MS; + } +} diff --git a/src/server.ts b/src/server.ts index f0beef7..a9a29eb 100644 --- a/src/server.ts +++ b/src/server.ts @@ -1,13 +1,11 @@ -import { randomUUID } from "node:crypto"; +import type { McpSessionLifecycle } from "./mcp-session-lifecycle.js"; +import { installMcpTransports } from "./mcp-http.js"; import express, { type NextFunction, type Request, type Response } from "express"; import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; -import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js"; -import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js"; import { hostHeaderValidation } from "@modelcontextprotocol/sdk/server/middleware/hostHeaderValidation.js"; import { getOAuthProtectedResourceMetadataUrl, mcpAuthRouter } from "@modelcontextprotocol/sdk/server/auth/router.js"; import { requireBearerAuth } from "@modelcontextprotocol/sdk/server/auth/middleware/bearerAuth.js"; import type { AuthInfo } from "@modelcontextprotocol/sdk/server/auth/types.js"; -import { isInitializeRequest } from "@modelcontextprotocol/sdk/types.js"; import { z } from "zod"; import { APPROVAL_BATCH_LIMIT, @@ -28,15 +26,6 @@ import { InvoiceDeduplicationService, InvoiceDedupStore, invoiceDedupDatabasePat import { APPLICATION_VERSION } from "./version.js"; import { SignedReferenceCodec } from "./signed-reference.js"; -type TransportEntry = - | { kind: "streamable"; transport: StreamableHTTPServerTransport; userId: string; createdAt: number; lastSeenAt: number } - | { kind: "sse"; transport: SSEServerTransport; userId: string; createdAt: number; lastSeenAt: number }; - -const MCP_SESSION_IDLE_TTL_MS = 2 * 60 * 60_000; -const MCP_SESSION_MAX_AGE_MS = 24 * 60 * 60_000; -const MCP_SESSION_MAX_PER_USER = 8; -const MCP_SESSION_MAX_TOTAL = 1000; - export const APPROVAL_MCP_INSTRUCTIONS = [ "先调用 approval_list_todos 获取当前待办。薪福通待办不按单据类型过滤；availableActions=[] 且 manualAction.reason=unsupported_xft_task_type 的项目仍要展示给用户，但只能引导到薪福通客户端查看和处理，禁止尝试 MCP 写入。用户只是在查看时，仅回答标题、申请人、金额、时间等业务信息，不主动讨论审批接口、批处理方式或索取写操作授权。用户明确要求处理后，在其授权范围内内部选择整组引用或自动分批；技术分批不得造成重复确认。", "approvalId、selectionId、targetId、单次条数限制、分批方式以及内部引用是否返回均属于 Agent 与 MCP 的内部调用细节，绝不向用户展示、解释或要求用户操作。面向用户只说明目标审批、业务动作、成功结果和需要用户决策的业务失败；不要说引用是否签发、是否失效、是否存在一键全选或如何分组。", @@ -61,11 +50,13 @@ export function createApprovalMcpServer( authenticatedUser?: AuthenticatedDingTalkUser, audit?: (userId: string, action: string, reference: string | undefined, outcome: string) => void, invoiceDedup?: InvoiceDeduplicationService, + lifecycle?: McpSessionLifecycle, ): McpServer { const server = new McpServer( { name: "dingtalk-approval-mcp", version: APPLICATION_VERSION }, { instructions: APPROVAL_MCP_INSTRUCTIONS }, ); + lifecycle?.trackRequests(server.server); const unified = new UnifiedApprovalService(config, service, xft, xftChallenges, officialOaReturns, invoiceDedup); const user = authenticatedUser ?? config.trustedLocalUser; if (!user) throw new Error("MCP 会话缺少已认证用户"); @@ -469,114 +460,15 @@ export async function startHttpServer(config: AppConfig): Promise { ...(authMiddleware ? [authMiddleware, attachmentBody, attachmentUploadHandler] : [attachmentBody, attachmentUploadHandler]), ); - const transports = new Map(); - let pendingTransportSessions = 0; - const pendingTransportSessionsByUser = new Map(); - const closeTransport = (id: string, entry: TransportEntry) => { - transports.delete(id); - void entry.transport.close().catch(() => undefined); - }; - const cleanupTransports = () => { - const now = Date.now(); - for (const [id, entry] of transports) { - if (entry.lastSeenAt < now - MCP_SESSION_IDLE_TTL_MS || entry.createdAt < now - MCP_SESSION_MAX_AGE_MS) { - closeTransport(id, entry); - } - } - }; - const sessionCapacityAvailable = (userId: string) => transports.size + pendingTransportSessions < MCP_SESSION_MAX_TOTAL - && [...transports.values()].filter((entry) => entry.userId === userId).length - + (pendingTransportSessionsByUser.get(userId) ?? 0) < MCP_SESSION_MAX_PER_USER; - const reserveSession = (userId: string) => { - pendingTransportSessions += 1; - pendingTransportSessionsByUser.set(userId, (pendingTransportSessionsByUser.get(userId) ?? 0) + 1); - }; - const releaseSessionReservation = (userId: string) => { - pendingTransportSessions = Math.max(0, pendingTransportSessions - 1); - const remaining = Math.max(0, (pendingTransportSessionsByUser.get(userId) ?? 1) - 1); - if (remaining === 0) pendingTransportSessionsByUser.delete(userId); - else pendingTransportSessionsByUser.set(userId, remaining); - }; - const transportCleanupTimer = setInterval(cleanupTransports, 10 * 60_000); - transportCleanupTimer.unref(); - - const mcpHandler = async (req: Request, res: Response) => { - let reservedUserId: string | undefined; - try { - cleanupTransports(); - const user = requestUser(config, oauth, req.auth); - const sessionId = headerString(req.headers["mcp-session-id"]); - let entry = sessionId ? transports.get(sessionId) : undefined; - if (entry && entry.kind !== "streamable") { - res.status(400).json({ error: "Invalid MCP session transport" }); - return; - } - if (entry && entry.userId !== user.userId) { - res.status(403).json({ error: "MCP session belongs to another authenticated user" }); - return; - } - if (entry) entry.lastSeenAt = Date.now(); - if (!entry && req.method === "POST" && !sessionId && isInitializeRequest(req.body)) { - if (!sessionCapacityAvailable(user.userId)) { - res.status(429).json({ error: "MCP session limit reached; close an old Agent session and retry" }); - return; - } - reserveSession(user.userId); - reservedUserId = user.userId; - const createdAt = Date.now(); - const transport: StreamableHTTPServerTransport = new StreamableHTTPServerTransport({ - sessionIdGenerator: () => randomUUID(), - onsessioninitialized: (id) => { - transports.set(id, { kind: "streamable", transport, userId: user.userId, createdAt, lastSeenAt: Date.now() }); - }, - }); - transport.onclose = () => { if (transport.sessionId) transports.delete(transport.sessionId); }; - entry = { kind: "streamable", transport, userId: user.userId, createdAt, lastSeenAt: createdAt }; - await createApprovalMcpServer( - config, service, xft, undefined, officialOaReturns, user, - oauth ? (userId, action, reference, outcome) => oauth.audit(userId, action, reference, outcome) : undefined, - invoiceDedup, - ).connect(transport); - } - if (!entry || entry.kind !== "streamable") { - res.status(sessionId ? 404 : 400).json({ error: "Missing or invalid MCP session" }); - return; - } - await entry.transport.handleRequest(req, res, req.body); - } catch (error) { sendServerError(res, error); } - finally { if (reservedUserId) releaseSessionReservation(reservedUserId); } - }; - app.all("/mcp", ...(authMiddleware ? [authMiddleware, mcpHandler] : [mcpHandler])); - - const sseHandler = async (req: Request, res: Response) => { - const user = requestUser(config, oauth, req.auth); - cleanupTransports(); - if (!sessionCapacityAvailable(user.userId)) { - res.status(429).send("MCP session limit reached"); - return; - } - const transport = new SSEServerTransport("/messages", res); - const createdAt = Date.now(); - transports.set(transport.sessionId, { kind: "sse", transport, userId: user.userId, createdAt, lastSeenAt: createdAt }); - res.on("close", () => transports.delete(transport.sessionId)); - await createApprovalMcpServer(config, service, xft, undefined, officialOaReturns, user, undefined, invoiceDedup).connect(transport); - }; - app.get("/sse", ...(authMiddleware ? [authMiddleware, sseHandler] : [sseHandler])); - const messagesHandler = async (req: Request, res: Response) => { - const user = requestUser(config, oauth, req.auth); - const entry = transports.get(String(req.query.sessionId ?? "")); - if (!entry || entry.kind !== "sse") { - res.status(404).send("Invalid SSE session"); - return; - } - if (entry.userId !== user.userId) { - res.status(403).send("Invalid SSE session"); - return; - } - entry.lastSeenAt = Date.now(); - await entry.transport.handlePostMessage(req, res, req.body); - }; - app.post("/messages", ...(authMiddleware ? [authMiddleware, messagesHandler] : [messagesHandler])); + installMcpTransports(app, { + authenticate: authMiddleware, + requestUser: (req) => requestUser(config, oauth, req.auth), + createServer: (user, lifecycle) => createApprovalMcpServer( + config, service, xft, undefined, officialOaReturns, user, + oauth ? (userId, action, reference, outcome) => oauth.audit(userId, action, reference, outcome) : undefined, + invoiceDedup, lifecycle, + ), + }); app.get("/healthz", (_req, res) => res.json({ status: "ok", mode: config.appMode })); await new Promise((resolve, reject) => { diff --git a/test/mcp-http.test.ts b/test/mcp-http.test.ts new file mode 100644 index 0000000..038907b --- /dev/null +++ b/test/mcp-http.test.ts @@ -0,0 +1,228 @@ +import assert from "node:assert/strict"; +import { once } from "node:events"; +import { setTimeout as delay } from "node:timers/promises"; +import test, { type TestContext } from "node:test"; +import express from "express"; +import { Client } from "@modelcontextprotocol/sdk/client/index.js"; +import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; +import { installMcpTransports } from "../src/mcp-http.js"; +import { createApprovalMcpServer } from "../src/server.js"; +import type { ApprovalService } from "../src/approval-service.js"; +import type { AppConfig } from "../src/config.js"; + +const config: AppConfig = { + clientId: "mock-client", clientSecret: "mock-secret", apiBaseUrl: "https://api.test", + oapiBaseUrl: "https://oapi.test", publicBaseUrl: "http://localhost:3000", port: 3000, + requestTimeoutMs: 1000, officialLookbackDays: 120, officialScanConcurrency: 2, + officialClientOnlyProcessCodes: [], batchLimit: 20, approvalRefSecret: "mock-session-test-key", +}; + +async function until(condition: () => boolean) { + for (let i = 0; i < 200; i += 1) { + if (condition()) return; + await delay(10); + } + assert.fail("HTTP lifecycle event did not arrive"); +} + +async function harness(t: TestContext, readTodos: () => Promise = async () => {}) { + let time = Date.now(); + let closedStreams = 0; + const app = express(); + app.use(express.json()); + app.use((req, res, next) => { + if (req.method === "GET" && req.path === "/mcp") res.once("close", () => { closedStreams += 1; }); + next(); + }); + const routes = installMcpTransports(app, { + now: () => time, + authenticate: (req, res, next) => { + // Synthetic bearer credentials; no external approval or OAuth writes. + const token = req.get("authorization"); + if (token !== "Bearer mock-a" && token !== "Bearer mock-b") { res.sendStatus(401); return; } + req.auth = { token: "mock", clientId: "mock-client", scopes: ["mcp:tools"], extra: { userId: token === "Bearer mock-a" ? "mock-user-a" : "mock-user-b" } }; + next(); + }, + requestUser: (req) => ({ userId: String(req.auth!.extra!.userId) }), + createServer: (user, lifecycle) => createApprovalMcpServer(config, { + async listMyTodos() { await readTodos(); return { items: [], diagnostics: {} }; }, + } as unknown as ApprovalService, undefined, undefined, undefined, user, undefined, undefined, lifecycle), + }); + const http = app.listen(0, "127.0.0.1"); + await once(http, "listening"); + const address = http.address(); + assert.ok(address && typeof address !== "string"); + const endpoint = new URL(`http://127.0.0.1:${address.port}/mcp`); + const clients: Client[] = []; + t.after(async () => { + await Promise.all(clients.map((client) => client.close())); + await routes.close(); + http.closeAllConnections(); + await new Promise((resolve) => http.close(() => resolve())); + }); + + async function connect(account = "a") { + let streamReady = false; + const client = new Client({ name: "mock-old-bridge", version: "1.0" }); + clients.push(client); + const transport = new StreamableHTTPClientTransport(endpoint, { + requestInit: { headers: { authorization: `Bearer mock-${account}` } }, + fetch: async (input, init) => { + const res = await fetch(input, init); + if (init?.method === "GET" && res.status === 200) streamReady = true; + return res; + }, + }); + await client.connect(transport); + await until(() => streamReady); + return { client, transport }; + } + + async function rpc(sessionId?: string, account = "a", method = "initialize") { + const response = await fetch(endpoint, { + method: "POST", + headers: { authorization: `Bearer mock-${account}`, "content-type": "application/json", + accept: "application/json, text/event-stream", ...(sessionId ? { "mcp-session-id": sessionId } : {}) }, + body: JSON.stringify({ jsonrpc: "2.0", id: 123, method, + ...(method === "initialize" ? { params: { protocolVersion: "2025-03-26", capabilities: {}, clientInfo: { name: "mock", version: "1" } } } : {}) }), + }); + const body = await response.text(); + return { response, body }; + } + + return { connect, rpc, endpoint, cleanup: routes.cleanup, + advance: (ms: number) => { time += ms; }, + async disconnect(client: Client) { + const before = closedStreams; + await client.close(); // Deliberately no terminateSession(), like existing clients. + await until(() => closedStreams > before); + }, + }; +} + +test("old bridge disconnects release eight occupied slots without a client upgrade", async (t) => { + const h = await harness(t); + const sessions = []; + for (let i = 0; i < 8; i += 1) sessions.push(await h.connect()); + const full = await h.rpc(); + assert.equal(full.response.status, 429); + assert.equal(full.response.headers.get("retry-after"), "60"); + assert.match(full.body, /MCP session limit reached/); + for (const { client } of sessions) await h.disconnect(client); + h.advance(59_000); + assert.equal((await h.rpc()).response.status, 429, "retain reconnect grace"); + h.advance(1_001); + const replacement = await h.connect(); // Admission itself sweeps expired sessions. + const result = await replacement.client.callTool({ name: "approval_list_todos", arguments: { source: "dingtalk" } }); + assert.notEqual(result.isError, true); + assert.equal((await h.rpc(sessions[0]!.transport.sessionId, "a", "tools/list")).response.status, 404); +}); + +test("live streams survive idle cleanup and capacity does not evict another user", async (t) => { + const h = await harness(t); + const sessions = []; + for (let i = 0; i < 8; i += 1) sessions.push(await h.connect()); + h.advance(3 * 60 * 60_000); + h.cleanup(); + assert.equal((await h.rpc()).response.status, 429); + const other = await h.connect("b"); + assert.ok((await other.client.listTools()).tools.length > 0); + assert.ok((await sessions[0]!.client.listTools()).tools.length > 0); + assert.equal((await h.rpc(sessions[0]!.transport.sessionId, "b", "tools/list")).response.status, 403); +}); + +test("GET reconnect within grace preserves the original session", async (t) => { + const h = await harness(t); + const original = await h.connect(); + const id = original.transport.sessionId!; + await h.disconnect(original.client); + h.advance(30_000); + const abort = new AbortController(); + const response = await fetch(h.endpoint, { headers: { + authorization: "Bearer mock-a", accept: "text/event-stream", "mcp-session-id": id, + }, signal: abort.signal }); + t.after(() => abort.abort()); + assert.equal(response.status, 200); + h.advance(90_000); + h.cleanup(); + assert.equal((await h.rpc(id, "a", "tools/list")).response.status, 200); + abort.abort(); +}); + +test("clients without GET retain the existing two-hour idle lifetime", async (t) => { + const h = await harness(t); + const initialized = await h.rpc(); + const id = initialized.response.headers.get("mcp-session-id")!; + assert.ok(id); + h.advance(5 * 60_000); + h.cleanup(); + assert.equal((await h.rpc(id, "a", "tools/list")).response.status, 200); + h.advance(2 * 60 * 60_000 + 1); + h.cleanup(); + assert.equal((await h.rpc(id, "a", "tools/list")).response.status, 404); +}); + +test("disconnected in-flight approval reads finish before their session is reclaimed", async (t) => { + let started = false; + let finish!: () => void; + const gate = new Promise((resolve) => { finish = resolve; }); + t.after(() => finish()); + const h = await harness(t, async () => { started = true; await gate; }); + const active = await h.connect(); + const id = active.transport.sessionId!; + const pending = active.client.callTool({ name: "approval_list_todos", arguments: { source: "dingtalk" } }).catch(() => undefined); + await until(() => started); + await h.disconnect(active.client); + await pending; + h.advance(25 * 60 * 60_000); + h.cleanup(); + assert.equal((await h.rpc(id, "a", "tools/list")).response.status, 200, "even max age must not interrupt work"); + finish(); + await delay(20); + h.cleanup(); + assert.equal((await h.rpc(id, "a", "tools/list")).response.status, 404); +}); + +test("concurrent initialization cannot bypass the per-user session cap", async (t) => { + const h = await harness(t); + const results = await Promise.all(Array.from({ length: 12 }, () => h.rpc())); + assert.equal(results.filter(({ response }) => response.status === 200).length, 8); + assert.equal(results.filter(({ response }) => response.status === 429).length, 4); +}); + +test("cancelled requests retain their slot until the handler actually settles", async (t) => { + let started = false; + let finish!: () => void; + const gate = new Promise((resolve) => { finish = resolve; }); + t.after(() => finish()); + const h = await harness(t, async () => { started = true; await gate; }); + const active = await h.connect(); + const id = active.transport.sessionId!; + const abort = new AbortController(); + const pending = active.client.callTool({ name: "approval_list_todos", arguments: { source: "dingtalk" } }, undefined, { signal: abort.signal }).catch(() => undefined); + await until(() => started); + abort.abort(); // SDK sends notifications/cancelled; the server suppresses the reply. + await pending; + await h.disconnect(active.client); + h.advance(60_001); + h.cleanup(); + assert.equal((await h.rpc(id, "a", "tools/list")).response.status, 200); + finish(); + await delay(20); + h.advance(60_001); + h.cleanup(); + assert.equal((await h.rpc(id, "a", "tools/list")).response.status, 404, "cancellation must not leak the slot forever"); +}); + +test("rejected duplicate GET does not evict the valid stream", async (t) => { + const h = await harness(t); + const active = await h.connect(); + const response = await fetch(h.endpoint, { headers: { + authorization: "Bearer mock-a", accept: "text/event-stream", "mcp-session-id": active.transport.sessionId!, + } }); + await response.text(); + assert.equal(response.status, 409); + h.advance(60_001); + h.cleanup(); + assert.ok((await active.client.listTools()).tools.length > 0); +});