Replace a session's GET SSE stream instead of answering 409

CHANGELOG.md

@@ -1,3 +1,10 @@
+## Unreleased
+
+- A GET SSE stream for a session that already has one replaces it instead of
+  being refused with `409 Conflict`. A client that drops its stream is not
+  noticed until the next write, so its reconnect was refused, failing clients
+  such as Cursor.
+
 ## 0.7.0 (2026-10-04)
 
 - New: the Tasks extension (`io.modelcontextprotocol/tasks`) under MCP

lib/phantom/plug.ex

@@ -479,21 +479,15 @@ defmodule Phantom.Plug do
 
   defp dispatch(%Plug.Conn{method: "GET"} = conn, opts) do
     if opts.pubsub do
-      case maybe_track_session_stream(conn) do
-        %Plug.Conn{halted: true} = conn ->
-          conn
-
-        conn ->
-          session = conn.private.phantom.session
+      conn = maybe_track_session_stream(conn)
 
-          conn
-          |> put_resp_header("mcp-session-id", session.id)
-          |> put_resp_header("cache-control", "no-cache, no-transform")
-          |> put_resp_content_type("text/event-stream")
-          |> put_resp_header("x-accel-buffering", "no")
-          |> send_chunked(200)
-          |> stream_loop(opts)
-      end
+      conn
+      |> put_resp_header("mcp-session-id", conn.private.phantom.session.id)
+      |> put_resp_header("cache-control", "no-cache, no-transform")
+      |> put_resp_content_type("text/event-stream")
+      |> put_resp_header("x-accel-buffering", "no")
+      |> send_chunked(200)
+      |> stream_loop(opts)
     else
       conn
       |> put_status(405)
@@ -1332,36 +1326,22 @@ defmodule Phantom.Plug do
 
   # The GET stream is the session's stream for server-initiated messages.
   # A POST's stream, `initialize` included, closes after its response.
-  defp maybe_track_session_stream(conn) do
-    session_id = conn.private.phantom.session.id
-
-    case {conn.method, Phantom.Tracker.list_session_streams(session_id)} do
-      # Only if no stream exists for the session (on any node)
-      {"GET", []} ->
-        session = %{conn.private.phantom.session | close_after_complete: false}
-
-        Phantom.Tracker.track_session(
-          self(),
-          session.id,
-          %{}
-        )
-
-        Phantom.Tracker.subscribe_session(session.pubsub, session.id)
+  #
+  # A client that drops its GET stream is not noticed until the next write,
+  # so a GET for a session that already has one replaces it (on any node)
+  # rather than being refused while the stale stream lingers.
+  defp maybe_track_session_stream(%Plug.Conn{method: "GET"} = conn) do
+    session = %{conn.private.phantom.session | close_after_complete: false}
 
-        put_in(conn.private.phantom.session, session)
+    session.id
+    |> Phantom.Tracker.list_session_streams()
+    |> Enum.each(fn {_key, %{pid: pid}} -> GenServer.cast(pid, :replaced) end)
 
-      {"GET", _existing} ->
-        conn
-        |> put_status(409)
-        |> json_error(
-          Request.error(%{
-            code: -32000,
-            message: "Only one SSE stream is allowed per session"
-          })
-        )
+    Phantom.Tracker.track_session(self(), session.id, %{})
+    Phantom.Tracker.subscribe_session(session.pubsub, session.id)
 
-      _ ->
-        conn
-    end
+    put_in(conn.private.phantom.session, session)
   end
+
+  defp maybe_track_session_stream(conn), do: conn
 end

lib/phantom/session.ex

@@ -755,6 +755,12 @@ defmodule Phantom.Session do
     {:stop, {:shutdown, :closed}, state}
   end
 
+  # A newer GET stream for the session took over.
+  def handle_cast(:replaced, state) do
+    state = state.stream_fun.(state, nil, "closed", "replaced")
+    {:stop, {:shutdown, :closed}, state}
+  end
+
   @doc false
   def handle_cast({:log_legacy, level, level_name, domain, payload}, state)
       when level <= state.log_level do

test/phantom/plug_test.exs

@@ -664,7 +664,9 @@ defmodule Phantom.PlugTest do
       assert_sse_connected()
     end
 
-    test "second GET on same session returns 409 without raising AlreadySentError" do
+    # A client that drops its GET stream is not noticed until the next write,
+    # so a reconnect must not be refused because of the stale stream.
+    test "a second GET on the same session replaces the first stream" do
       session_id = "019dd3d8-0000-0000-0000-000000000001"
 
       :get
@@ -673,17 +675,22 @@ defmodule Phantom.PlugTest do
       |> call(session_id: session_id)
 
       assert_sse_connected()
+      [{_, %{pid: first}}] = Phantom.Tracker.list_session_streams(session_id)
+      ref = Process.monitor(first)
 
       :get
       |> conn("/mcp")
       |> put_req_header("accept", "text/event-stream")
       |> call(session_id: session_id)
 
-      assert_receive {:conn, conn}
-      assert conn.status == 409
-      error = JSON.decode!(conn.resp_body)
-      assert error["error"]["code"] == -32000
-      assert error["error"]["message"] == "Only one SSE stream is allowed per session"
+      assert_receive {:response, nil, "closed", "replaced"}
+      assert_receive {:DOWN, ^ref, :process, ^first, _}
+      assert_receive {:conn, %{status: 200}}
+      assert_receive {:plug_conn, :sent}
+      refute_receive {:conn, _}
+
+      assert [{_, %{pid: second}}] = Phantom.Tracker.list_session_streams(session_id)
+      assert second != first
     end
   end