fix(mcp): initialized sessions before opening sse stream

Moved the optional Streamable HTTP GET listener after the initialized notification so stateful servers do not terminate the session during setup.

Added regression coverage for a server that rejects pre-initialization GET traffic.

Fixes #8514

packages/ai/test/remote-auth-store.test.ts

@@ -1110,7 +1110,7 @@ describe("RemoteAuthCredentialStore + AuthStorage integration", () => {
 
 	test("broker invalidation drops server-side last-good usage reports", async () => {
 		const credential = serverStore!.listAuthCredentials("anthropic")[0];
-		if (!credential || credential.credential.type !== "oauth") throw new Error("expected OAuth credential");
+		if (credential?.credential.type !== "oauth") throw new Error("expected OAuth credential");
 		serverStore!.updateAuthCredential(credential.id, {
 			...credential.credential,
 			expires: Date.now() + 3_600_000,

packages/coding-agent/CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## [Unreleased]
 
+### Fixed
+
+- Fixed Streamable HTTP MCP sessions being invalidated by opening the optional GET SSE stream before sending `notifications/initialized`, which prevented Figma Dev Mode MCP from connecting ([#8514](https://github.com/can1357/oh-my-pi/issues/8514)).
+
 ## [17.3.3] - 2026-08-14
 
 ### Fixed

packages/coding-agent/src/mcp/client.ts

@@ -92,7 +92,7 @@ async function initializeConnection(
 	transport: MCPTransport,
 	options?: {
 		signal?: AbortSignal;
-		/** Called after the initialize response (which sets the session ID) but before notifications/initialized. */
+		/** Called after notifications/initialized succeeds. */
 		onInitialized?: () => void | Promise<void>;
 	},
 ): Promise<MCPInitializeResult> {
@@ -119,14 +119,13 @@ async function initializeConnection(
 	// initialize; transports that don't need it ignore this.
 	transport.setProtocolVersion?.(result.protocolVersion);
 
-	// Hook point: the transport now has the session ID from the initialize response.
-	// For HTTP, this is the moment to open the SSE stream so server-to-client requests
-	// triggered by notifications/initialized (e.g. roots/list) can be delivered.
-	await options?.onInitialized?.();
-
-	// Send initialized notification
+	// Send initialized before opening the optional GET SSE stream. Servers may
+	// reject or terminate sessions that receive session traffic before this
+	// notification; POST response streams already carry messages during setup.
 	await transport.notify("notifications/initialized");
 
+	await options?.onInitialized?.();
+
 	return result;
 }
 
@@ -162,8 +161,8 @@ export async function connectToServer(
 			const initResult = await initializeConnection(transport, {
 				signal: options?.signal,
 				async onInitialized() {
-					// Open the SSE stream before sending initialized, so server-to-client
-					// requests triggered by on_initialized (e.g. roots/list) are delivered.
+					// Open the optional GET SSE stream only after the initialized
+					// notification makes the session ready for further traffic.
 					if ("startSSEListener" in transport! && typeof transport!.startSSEListener === "function") {
 						await (transport as { startSSEListener(): Promise<void> }).startSSEListener();
 					}

packages/coding-agent/test/mcp-http-transport.test.ts

@@ -1,4 +1,5 @@
 import { afterEach, describe, expect, it } from "bun:test";
+import { connectToServer } from "@oh-my-pi/pi-coding-agent/mcp/client";
 import { HttpTransport } from "@oh-my-pi/pi-coding-agent/mcp/transports/http";
 
 const encoder = new TextEncoder();
@@ -49,6 +50,58 @@ async function withPendingGuard<T>(promise: Promise<T>, label: string): Promise<
 	]);
 }
 
+describe("MCP Streamable HTTP initialization", () => {
+	it("sends initialized before opening the optional GET SSE stream", async () => {
+		const requests: string[] = [];
+		let initialized = false;
+		let sessionValid = true;
+		server = Bun.serve({
+			port: 0,
+			async fetch(req) {
+				if (req.method === "GET") {
+					requests.push("GET");
+					if (initialized) return new Response(null, { status: 405 });
+					sessionValid = false;
+					return new Response("session is not initialized", { status: 400 });
+				}
+				if (req.method === "DELETE") return new Response(null, { status: 204 });
+
+				const body = (await req.json()) as { id?: string | number; method: string };
+				requests.push(body.method);
+				if (body.method === "initialize") {
+					const response = {
+						jsonrpc: "2.0",
+						id: body.id,
+						result: {
+							protocolVersion: "2025-11-25",
+							capabilities: {},
+							serverInfo: { name: "session-order", version: "1.0.0" },
+						},
+					};
+					return new Response(`event: message\ndata: ${JSON.stringify(response)}\n\n`, {
+						headers: {
+							"Content-Type": "text/event-stream",
+							"Mcp-Session-Id": "session-order",
+						},
+					});
+				}
+				if (!sessionValid) return new Response("session terminated", { status: 409 });
+				initialized = true;
+				return new Response(null, { status: 202 });
+			},
+		});
+
+		const connection = await connectToServer("session-order", {
+			type: "http",
+			url: `http://127.0.0.1:${server.port}/mcp`,
+			timeout: GUARD_TIMEOUT_MS,
+		});
+
+		expect(requests).toEqual(["initialize", "notifications/initialized", "GET"]);
+		await connection.transport.close();
+	});
+});
+
 describe("MCP Streamable HTTP transport timeouts", () => {
 	it("keeps the request timeout active until a JSON response body is fully read", async () => {
 		server = Bun.serve({