Address review round three: EOF tolerance, docstring, refusal wording

- Treat a read stream whose receive end the transport closed as
  end-of-input in the opening peek and the relay, mirroring the
  tolerance the dispatcher's read loop already had; the read loop moved
  here, so its EOF handling moves with it. Two tests close the stream
  before and after the first request and expect a clean return.
- ListenHandler's docstring no longer claims it needs SSE mode.
- The legacy refusal states the era rule instead of narrating what the
  first request was, so it holds for every opener.

Author: maxisbey

src/mcp/server/runner.py

@@ -677,22 +677,30 @@ async def replay_then_relay() -> None:
         async with replay_send:
             for envelope in lead:
                 await replay_send.send(envelope)
-            async for item in read_stream:
-                await replay_send.send((_sender_context(read_stream), item))
+            try:
+                async for item in read_stream:
+                    await replay_send.send((_sender_context(read_stream), item))
+            except anyio.ClosedResourceError:
+                # Receive end closed under us (stateless SHTTP teardown); same as EOF.
+                logger.debug("read stream closed by transport; treating as EOF")
 
     # This helper takes ownership of `read_stream` from the serving loop, so
     # every exit - including cancellation while awaiting the first request -
     # closes it and the replay channel.
     try:
-        async for item in read_stream:
-            if isinstance(item, SessionMessage) and isinstance(item.message, JSONRPCRequest):
-                opening_request = item.message
-            elif len(lead) >= _PRE_REQUEST_REPLAY_LIMIT:
-                logger.debug("dropped a frame received before the first request: %r", item)
-                continue
-            lead.append((_sender_context(read_stream), item))
-            if opening_request is not None:
-                break
+        try:
+            async for item in read_stream:
+                if isinstance(item, SessionMessage) and isinstance(item.message, JSONRPCRequest):
+                    opening_request = item.message
+                elif len(lead) >= _PRE_REQUEST_REPLAY_LIMIT:
+                    logger.debug("dropped a frame received before the first request: %r", item)
+                    continue
+                lead.append((_sender_context(read_stream), item))
+                if opening_request is not None:
+                    break
+        except anyio.ClosedResourceError:
+            # Receive end closed under us (stateless SHTTP teardown); same as EOF.
+            logger.debug("read stream closed by transport; treating as EOF")
         async with anyio.create_task_group() as tg:
             tg.start_soon(replay_then_relay)
             yield opening_request, replayed
@@ -730,8 +738,8 @@ async def on_request(
         if method != "initialize" and _has_modern_envelope(params):
             raise MCPError(
                 code=INVALID_REQUEST,
-                message="this connection's first request carried no 2026-07-28 envelope, so it "
-                "serves the handshake era; enveloped requests are not accepted on it",
+                message="this connection serves the handshake protocol era; "
+                "requests carrying the 2026-07-28 envelope are not accepted on it",
             )
         return await runner.on_request(dctx, method, params)
 

src/mcp/server/subscriptions.py

@@ -164,8 +164,8 @@ class ListenHandler:
     cancels the handler; the stream just ends, per the spec's abrupt-close
     contract) or `close` ends all streams gracefully.
 
-    Requires a transport that can stream a request's response (streamable
-    HTTP's SSE mode).
+    Served on any transport that can carry the request's response stream:
+    streamable HTTP's SSE mode, or a duplex stream pair such as stdio.
 
     `max_subscriptions` bounds concurrent streams (further listen requests are
     rejected with `INTERNAL_ERROR`, before the ack). `max_buffered_events`

tests/server/test_runner.py

@@ -16,6 +16,7 @@
 
 import anyio
 import anyio.abc
+import anyio.lowlevel
 import pytest
 from mcp_types import (
     CLIENT_CAPABILITIES_META_KEY,
@@ -1544,7 +1545,7 @@ async def test_dual_era_loop_initialize_locks_legacy_and_rejects_modern_traffic(
         assert result["tools"][0]["name"] == "t"
     assert discover_exc.value.error.code == INVALID_REQUEST
     assert envelope_exc.value.error.code == INVALID_REQUEST
-    assert "carried no 2026-07-28 envelope" in envelope_exc.value.error.message
+    assert "serves the handshake protocol era" in envelope_exc.value.error.message
 
 
 @pytest.mark.anyio
@@ -2019,6 +2020,36 @@ async def test_dual_era_loop_leading_notifications_never_decide_the_era(server:
         assert result["tools"][0]["name"] == "t"
 
 
+@pytest.mark.anyio
+async def test_dual_era_loop_treats_a_read_stream_closed_before_the_first_request_as_end_of_input(server: SrvT):
+    """A transport closing the read stream's receive end (the stateless teardown
+    pattern) ends the loop like end-of-input rather than surfacing a
+    closed-resource error - here before the client has sent anything."""
+    c2s_send, c2s_recv = anyio.create_memory_object_stream[SessionMessage | Exception](8)
+    s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](8)
+    with anyio.fail_after(5):
+        async with anyio.create_task_group() as tg, c2s_send, s2c_recv:
+            tg.start_soon(partial(serve_dual_era_loop, server, c2s_recv, s2c_send, lifespan_state=_LIFESPAN))
+            await anyio.lowlevel.checkpoint()
+            c2s_recv.close()  # the transport tears down under the waiting loop
+
+
+@pytest.mark.anyio
+async def test_dual_era_loop_treats_a_read_stream_closed_mid_connection_as_end_of_input(server: SrvT):
+    """The same teardown after the connection is open ends the era loop
+    cleanly: the relay behind the opening request treats the closed receive
+    end as end-of-input."""
+    c2s_send, c2s_recv = anyio.create_memory_object_stream[SessionMessage | Exception](8)
+    s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](8)
+    ping = JSONRPCRequest(jsonrpc="2.0", id=1, method="ping", params=None)
+    with anyio.fail_after(5):
+        async with anyio.create_task_group() as tg, c2s_send, s2c_recv:
+            tg.start_soon(partial(serve_dual_era_loop, server, c2s_recv, s2c_send, lifespan_state=_LIFESPAN))
+            await c2s_send.send(SessionMessage(message=ping))
+            await s2c_recv.receive()  # the connection is open and answered
+            c2s_recv.close()
+
+
 @pytest.mark.anyio
 async def test_dual_era_client_propagates_body_exception_unwrapped(server: SrvT):
     """The harness re-raises body exceptions as-is, not as `ExceptionGroup`."""