diff --git a/CHANGELOG.md b/CHANGELOG.md index 86c5021..2747d30 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/lib/phantom/plug.ex b/lib/phantom/plug.ex index b2c3f97..75109e8 100644 --- a/lib/phantom/plug.ex +++ b/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 diff --git a/lib/phantom/session.ex b/lib/phantom/session.ex index 315114c..881b1a1 100644 --- a/lib/phantom/session.ex +++ b/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 diff --git a/test/phantom/plug_test.exs b/test/phantom/plug_test.exs index b0f9bb2..6654949 100644 --- a/test/phantom/plug_test.exs +++ b/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