Serve the 2026-07-28 protocol over stdio: decide the era from the opening request (#3152)

docs/handlers/subscriptions.md

@@ -52,13 +52,6 @@ Two things the stream is *not*:
     handler on the low-level `Server` and acknowledge a smaller filter than the client asked
     for; the acknowledgment is how the client learns what it actually got.
 
-!!! warning "Streamable HTTP only, for now"
-    `subscriptions/listen` needs a transport that can stream a request's response, which today
-    means streamable HTTP. Over stdio a 2026-07-28 connection rejects the method with
-    METHOD_NOT_FOUND, even though `server/discover` advertises the subscription capabilities
-    there. Serving it over stdio is planned; the open-stream semantics for that transport are
-    not built yet.
-
 ## The client end
 
 Here is a client on the other side of that stream, following the board:

docs/whats-new.md

@@ -191,7 +191,7 @@ That file is the pitch in one place: one server, one `Resolve`-backed tool, and
 
 ### Change notifications become one stream
 
-At 2026-07-28 the standalone HTTP GET stream and `resources/subscribe` are replaced by `subscriptions/listen`: the client opens one long-lived stream and names the notification kinds it wants. `MCPServer` serves it out of the box; you publish with `await ctx.notify_resource_updated(uri)` (and `notify_tools_changed()`, and so on), and multi-replica deployments plug in a shared `SubscriptionBus`. On the client (since `2.0.0b2`), `async with client.listen(...)` opens the stream: the filter goes in as keyword arguments, typed change events come back, and `sub.honored` is the subset the server agreed to deliver. One honest caveat: over stdio the server does not serve the stream yet.
+At 2026-07-28 the standalone HTTP GET stream and `resources/subscribe` are replaced by `subscriptions/listen`: the client opens one long-lived stream and names the notification kinds it wants. `MCPServer` serves it out of the box; you publish with `await ctx.notify_resource_updated(uri)` (and `notify_tools_changed()`, and so on), and multi-replica deployments plug in a shared `SubscriptionBus`. On the client (since `2.0.0b2`), `async with client.listen(...)` opens the stream: the filter goes in as keyword arguments, typed change events come back, and `sub.honored` is the subset the server agreed to deliver.
 
 **[Subscriptions](handlers/subscriptions.md)** covers publishing and serving, **[its Clients twin](client/subscriptions.md)** the watching end, and **[Deploy & scale](run/deploy.md)** the bus.
 

src/mcp/server/lowlevel/server.py

@@ -703,9 +703,9 @@ class Server(Generic[LifespanResultT]):
 
         Thin wrapper over `serve_dual_era_loop`: enters the server lifespan,
         then drives the loop, serving the legacy handshake era and the modern
-        per-request-envelope era (the first era-distinctive message to succeed
-        locks the connection). Transports with their own lifespan owner (the
-        streamable-HTTP manager) call `serve_loop` directly instead.
+        per-request-envelope era (the client's first request decides which).
+        Transports with their own lifespan owner (the streamable-HTTP manager)
+        call `serve_loop` directly instead.
         """
         async with self.lifespan(self) as lifespan_context:
             await serve_dual_era_loop(

src/mcp/server/runner.py

@@ -13,11 +13,13 @@ the entry constructs the `Connection`, the driver tears it down.
 
 from __future__ import annotations
 
+import contextvars
 import logging
-from collections.abc import Awaitable, Mapping
+from collections.abc import AsyncIterator, Awaitable, Mapping
+from contextlib import asynccontextmanager
 from dataclasses import KW_ONLY, dataclass, replace
 from functools import cached_property, partial
-from typing import TYPE_CHECKING, Any, Generic, Literal, cast
+from typing import TYPE_CHECKING, Any, Generic, cast
 
 import anyio
 import anyio.abc
@@ -37,6 +39,7 @@ from mcp_types import (
     Implementation,
     InitializeRequestParams,
     InitializeResult,
+    JSONRPCRequest,
     RequestId,
     RequestParams,
     RequestParamsMeta,
@@ -57,6 +60,7 @@ from mcp.server.connection import Connection, NotifyOnlyOutbound
 from mcp.server.context import CallNext, HandlerResult, ServerMiddleware, ServerRequestContext
 from mcp.server.models import InitializationOptions
 from mcp.server.session import ServerSession
+from mcp.shared._context_streams import ContextReceiveStream
 from mcp.shared._stream_protocols import ReadStream, WriteStream
 from mcp.shared.dispatcher import CallOptions, DispatchContext, Dispatcher, OnNotify, OnRequest
 from mcp.shared.exceptions import MCPError, NoBackChannelError
@@ -497,15 +501,12 @@ async def serve_loop(
 def _has_modern_envelope(params: Mapping[str, Any] | None) -> bool:
     """Whether `params._meta` carries the reserved protocol-version key.
 
-    Era evidence is the client's explicit version declaration: the
-    `io.modelcontextprotocol/protocolVersion` key exists only in 2026-07-28+
-    envelopes, and the `io.modelcontextprotocol/` prefix is spec-reserved, so
-    legacy traffic never mints it (bare `_meta` is NOT evidence - legacy
-    requests carry `progressToken` there). Presence of the version key alone
-    is the rule, not the full required pair, so a half-built envelope
-    (version present, capabilities missing) still routes modern and gets the
-    classifier's INVALID_PARAMS naming the missing key instead of the legacy
-    path's generic one - and, like every failed classification, locks no era.
+    The `io.modelcontextprotocol/protocolVersion` key exists only in
+    2026-07-28+ envelopes and its prefix is spec-reserved, so legacy traffic
+    never mints it (a bare `_meta` is not evidence - legacy requests carry
+    `progressToken` there). The version key alone is the signal, not the full
+    required pair, so a half-built envelope still routes modern and gets the
+    classifier's INVALID_PARAMS naming the missing key.
     """
     if not params:
         return False
@@ -605,91 +606,180 @@ async def serve_dual_era_loop(
     init_options: InitializationOptions | None = None,
     raise_exceptions: bool = False,
 ) -> None:
-    """Drive `server` over a duplex stream pair, serving both protocol eras.
+    """Drive `server` over a duplex stream pair, in the era the client opens with.
 
-    The stream-pair counterpart of the modern HTTP entry's era router. Era is
-    a property of the connection, decided by how the client opens it, and
-    mid-stream switching is undefined - so the first era-distinctive message
-    to SUCCEED locks the connection (matching the typescript-sdk):
-
-    - A successful `initialize` locks legacy: the connection behaves exactly
-      like `serve_loop` for its lifetime, and modern envelope traffic is then
-      rejected with INVALID_REQUEST. `initialize` never routes modern - the
-      method is legacy-distinctive by definition - even when a confused
-      client stamps the envelope keys on it.
-    - A request whose `_meta` declares the modern protocol version - or
-      `server/discover`, a modern-only method - is classified
-      (`classify_inbound_request`) and served single-exchange via `serve_one`
-      with a born-ready per-request `Connection`, the same dispatch model as
-      the modern HTTP entry. The first such request to succeed locks the
-      connection modern; a later `initialize` is then rejected with
-      UNSUPPORTED_PROTOCOL_VERSION naming the modern versions.
-
-    Modern connections push notifications over the duplex pipe but refuse
-    server-initiated requests on both channels (the modern protocol forbids
-    them). A request that fails - rejected classification, malformed envelope
-    content, unknown method - never locks either era, so a failed probe
-    leaves the legacy handshake available: released auto-negotiating clients
-    fall back on any error code except -32022, and that code is only emitted
-    for genuine version negotiation or for `initialize` on an
-    already-modern connection.
-
-    The era lock rides the request's own dispatch. For the inline methods
-    (`initialize`, `server/discover`) that completes before the next frame is
-    read, so the canonical probe-then-go flow is race-free; a pinned-modern
-    client that pipelines frames ahead of its first response should expect
-    envelope-less notifications sent in that window to be dropped. The lock
-    settles exactly once: a request from the other era that was already in
-    flight when the lock committed may still complete and its response
-    stands, but the era does not move; and a success the peer cancelled away
-    (it sees "Request cancelled", not the result) does not lock either.
+    The client's first request decides the connection's protocol era, once:
+    a request carrying the 2026-07-28 per-request `_meta` envelope opens a
+    modern connection, and anything else - the `initialize` handshake, which
+    does not exist at 2026 versions even when a client stamps the envelope on
+    it - opens a legacy one. The deciding frame is replayed into the chosen
+    serving loop along with everything the client sent before it. A later
+    claim from the other era is refused: `initialize` on a modern connection
+    gets UNSUPPORTED_PROTOCOL_VERSION naming the served versions, and an
+    enveloped request on a legacy connection gets INVALID_REQUEST.
     """
+    # This loop owns both streams from the moment it is called, so the write
+    # stream is closed even if the client leaves before sending any request.
+    try:
+        async with _replay_from_opening_request(read_stream) as (opening, replayed):
+            opens_modern = (
+                opening is not None and opening.method != "initialize" and _has_modern_envelope(opening.params)
+            )
+            if opens_modern:
+                await _serve_modern_stream(
+                    server, replayed, write_stream, lifespan_state=lifespan_state, raise_exceptions=raise_exceptions
+                )
+            else:
+                await _serve_legacy_stream(
+                    server,
+                    replayed,
+                    write_stream,
+                    lifespan_state=lifespan_state,
+                    session_id=session_id,
+                    init_options=init_options,
+                    raise_exceptions=raise_exceptions,
+                )
+    finally:
+        await write_stream.aclose()
+
+
+_PRE_REQUEST_REPLAY_LIMIT: int = 8
+"""How many frames arriving ahead of the client's first request are kept
+for the chosen era's loop (a bare `notifications/initialized` is the one that
+matters); further ones are dropped and never decide the era."""
+
+
+def _sender_context(stream: ReadStream[Any]) -> contextvars.Context:
+    """The per-message sender context a context-aware stream carries, else the current one."""
+    ctx = getattr(stream, "last_context", None)
+    return ctx if ctx is not None else contextvars.copy_context()
+
+
+@asynccontextmanager
+async def _replay_from_opening_request(
+    read_stream: ReadStream[SessionMessage | Exception],
+) -> AsyncIterator[tuple[JSONRPCRequest | None, ReadStream[SessionMessage | Exception]]]:
+    """Peek at the client's first request without consuming it.
+
+    Yields that request together with a stream that replays it - preceded by
+    up to `_PRE_REQUEST_REPLAY_LIMIT` earlier frames - and relays the rest of
+    `read_stream` behind it, sender contexts included. The request is `None`
+    if the channel closes before one arrives.
+    """
+    lead: list[tuple[contextvars.Context, SessionMessage | Exception]] = []
+    opening_request: JSONRPCRequest | None = None
+    replay_send, replay_receive = anyio.create_memory_object_stream[
+        tuple[contextvars.Context, SessionMessage | Exception]
+    ]()
+    replayed = ContextReceiveStream(replay_receive)
+
+    async def replay_then_relay() -> None:
+        async with replay_send:
+            for envelope in lead:
+                await replay_send.send(envelope)
+            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:
+        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
+            tg.cancel_scope.cancel()
+    finally:
+        await read_stream.aclose()
+        replay_send.close()
+        replay_receive.close()
+
+
+async def _serve_legacy_stream(
+    server: Server[LifespanT],
+    read_stream: ReadStream[SessionMessage | Exception],
+    write_stream: WriteStream[SessionMessage],
+    *,
+    lifespan_state: LifespanT,
+    session_id: str | None,
+    init_options: InitializationOptions | None,
+    raise_exceptions: bool,
+) -> None:
+    """Serve a 2025 handshake connection; enveloped requests are refused."""
     dispatcher: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(
         read_stream,
         write_stream,
         raise_handler_exceptions=raise_exceptions,
-        # `initialize` inline for the same pipelining reason as `serve_loop`;
-        # `server/discover` inline so the modern era lock commits before the
-        # next pipelined message is read.
-        inline_methods=frozenset({"initialize", "server/discover"}),
+        # `initialize` inline for the same pipelining reason as `serve_loop`.
+        inline_methods=frozenset({"initialize"}),
     )
-    loop_connection = Connection.for_loop(dispatcher, session_id=session_id)
-    loop_runner = ServerRunner(server, loop_connection, lifespan_state, init_options=init_options)
-    standalone_outbound = NotifyOnlyOutbound(dispatcher)
-    era: Literal["unlocked", "legacy", "modern"] = "unlocked"
-    modern_version = LATEST_MODERN_VERSION
+    connection = Connection.for_loop(dispatcher, session_id=session_id)
+    runner = ServerRunner(server, connection, lifespan_state, init_options=init_options)
 
-    def era_settles(dctx: DispatchContext[TransportContext]) -> bool:
-        # The one definition of "this request may lock the era": it settled as
-        # a client-visible success on a still-unlocked connection. The lock is
-        # monotone - the first success wins, so a straggling request from the
-        # other era can never overwrite a committed lock. A pending peer
-        # cancel means the dispatcher is about to replace this response with
-        # "Request cancelled": the client never sees the success, no lock.
-        return era == "unlocked" and not dctx.cancel_requested.is_set()
-
-    async def serve_modern(
+    async def on_request(
         dctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None
     ) -> dict[str, Any]:
-        nonlocal era, modern_version
+        if method != "initialize" and _has_modern_envelope(params):
+            raise MCPError(
+                code=INVALID_REQUEST,
+                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)
+
+    try:
+        await dispatcher.run(on_request, runner.on_notify)
+    finally:
+        await aclose_shielded(connection)
+
+
+async def _serve_modern_stream(
+    server: Server[LifespanT],
+    read_stream: ReadStream[SessionMessage | Exception],
+    write_stream: WriteStream[SessionMessage],
+    *,
+    lifespan_state: LifespanT,
+    raise_exceptions: bool,
+) -> None:
+    """Serve a 2026-07-28 connection: every request carries its own envelope."""
+    dispatcher: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(
+        read_stream, write_stream, raise_handler_exceptions=raise_exceptions
+    )
+    outbound = NotifyOnlyOutbound(dispatcher)
+
+    async def on_request(
+        dctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None
+    ) -> dict[str, Any]:
+        if method == "initialize":
+            raise MCPError(
+                code=UNSUPPORTED_PROTOCOL_VERSION,
+                message="connection is serving the 2026-07-28 protocol; the initialize handshake is not accepted",
+                data=_initialize_after_modern_data(params),
+            )
         route = classify_inbound_request({"method": method, "params": params})
         if isinstance(route, InboundLadderRejection):
             raise MCPError(code=route.code, message=route.message, data=route.data)
-        if method == "subscriptions/listen":
-            # The registered listen handler assumes the HTTP entry's stream
-            # semantics; served over a stream pair it would wedge. Reject until
-            # this transport grows its own listen design.
-            raise MCPError(
-                code=METHOD_NOT_FOUND, message="subscriptions/listen is not served over this transport", data=method
-            )
         connection = Connection.from_envelope(
-            route.protocol_version,
-            route.client_info,
-            route.client_capabilities,
-            outbound=standalone_outbound,
+            route.protocol_version, route.client_info, route.client_capabilities, outbound=outbound
         )
         try:
-            result = await serve_one(
+            return await serve_one(
                 server,
                 _NoServerRequestsDispatchContext(dctx),
                 method,
@@ -698,68 +788,25 @@ async def serve_dual_era_loop(
                 lifespan_state=lifespan_state,
             )
         except (MCPError, ValidationError):
-            # The dispatcher's shared ladder maps these to the same wire error
-            # the modern HTTP entry produces.
+            # The dispatcher's shared ladder maps these to the wire error.
             raise
         except Exception as exc:
             if raise_exceptions:
                 raise
             error = modern_error_data(exc)
             raise MCPError(code=error.code, message=error.message, data=error.data) from exc
-        if era_settles(dctx):
-            era, modern_version = "modern", route.protocol_version
-        return result
-
-    async def on_request(
-        dctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None
-    ) -> dict[str, Any]:
-        nonlocal era
-        if era == "legacy":
-            if _has_modern_envelope(params):
-                raise MCPError(
-                    code=INVALID_REQUEST,
-                    message="connection is locked to the legacy handshake era; "
-                    "modern envelope requests are not accepted",
-                )
-            # Bare modern-only methods (e.g. `server/discover`) fall through to
-            # the loop runner's per-version surface validation - the same
-            # METHOD_NOT_FOUND a handshake-only server produced, byte for byte.
-            return await loop_runner.on_request(dctx, method, params)
-        if era == "modern":
-            if method == "initialize":
-                raise MCPError(
-                    code=UNSUPPORTED_PROTOCOL_VERSION,
-                    message="connection already negotiated a modern protocol version",
-                    data=_initialize_after_modern_data(params),
-                )
-            return await serve_modern(dctx, method, params)
-        # Unlocked. `initialize` is legacy-distinctive by definition (the
-        # method does not exist at modern versions), so it takes the handshake
-        # path even when the envelope keys are stamped on it.
-        if method != "initialize" and (method == "server/discover" or _has_modern_envelope(params)):
-            return await serve_modern(dctx, method, params)
-        result = await loop_runner.on_request(dctx, method, params)
-        if method == "initialize" and era_settles(dctx):
-            # Lock only on success: a failed handshake leaves both eras open.
-            era = "legacy"
-        return result
 
     async def on_notify(dctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None) -> None:
-        if era != "modern":
-            return await loop_runner.on_notify(dctx, method, params)
-        # The envelope is request-only, so notifications inherit the
-        # connection's locked version.
-        connection = Connection.from_envelope(modern_version, None, None, outbound=standalone_outbound)
+        # The envelope is request-only, so a notification runs at the latest
+        # served version; the modern protocol has nothing version-specific here.
+        connection = Connection.from_envelope(LATEST_MODERN_VERSION, None, None, outbound=outbound)
         notify_runner = ServerRunner(server, connection, lifespan_state)
         try:
             await notify_runner.on_notify(_NoServerRequestsDispatchContext(dctx), method, params)
         finally:
             await aclose_shielded(connection)
 
-    try:
-        await dispatcher.run(on_request, on_notify)
-    finally:
-        await aclose_shielded(loop_connection)
+    await dispatcher.run(on_request, on_notify)
 
 
 async def serve_one(

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

@@ -7,6 +7,7 @@ behaviour under test. Driver tests (`serve_connection`, `serve_one`,
 `aclose_shielded`) follow at the bottom.
 """
 
+import contextvars
 from collections.abc import AsyncIterator, Mapping
 from contextlib import asynccontextmanager
 from dataclasses import dataclass, field, replace
@@ -15,6 +16,7 @@ from typing import Any, cast
 
 import anyio
 import anyio.abc
+import anyio.lowlevel
 import pytest
 from mcp_types import (
     CLIENT_CAPABILITIES_META_KEY,
@@ -68,6 +70,8 @@ from mcp.server.runner import (
     serve_one,
 )
 from mcp.server.session import ServerSession
+from mcp.server.subscriptions import SUBSCRIPTION_ID_META_KEY, InMemorySubscriptionBus, ListenHandler
+from mcp.shared._context_streams import create_context_streams
 from mcp.shared.dispatcher import CallOptions
 from mcp.shared.exceptions import MCPError, NoBackChannelError
 from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher
@@ -1355,6 +1359,8 @@ async def _append_async(dst: list[int], v: int) -> None:
 
 _LIFESPAN: dict[str, Any] = {}
 
+_SENDER_VAR: contextvars.ContextVar[str] = contextvars.ContextVar("dual_era_sender_var", default="unset")
+
 
 @pytest.mark.anyio
 async def test_serve_one_runs_handler_and_returns_result_dict(server: SrvT):
@@ -1539,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 "locked to the legacy handshake era" in discover_exc.value.error.message
+    assert "serves the handshake protocol era" in envelope_exc.value.error.message
 
 
 @pytest.mark.anyio
@@ -1558,44 +1564,49 @@ async def test_dual_era_loop_bare_discover_after_legacy_lock_is_byte_identical(s
 
 
 @pytest.mark.anyio
-async def test_dual_era_loop_unsupported_modern_version_rejects_without_locking(server: SrvT):
-    """A probe at an unknown modern version gets -32022 with the supported
-    list, and the rejection does not lock the era: the legacy handshake still
-    succeeds afterwards (the released auto clients' retry/fallback contract)."""
+async def test_dual_era_loop_unsupported_modern_version_gets_the_supported_list(server: SrvT):
+    """A probe at an unknown modern version is answered -32022 with the
+    supported list; the connection is a 2026 one regardless, so the client
+    picks a listed version instead of falling back to `initialize` - which the
+    modern connection refuses with -32022 too, exactly as stdio.mdx directs."""
     async with dual_era_client(server) as (client, _):
         with pytest.raises(MCPError) as exc_info:
             await client.send_raw_request("server/discover", _modern_params(version="2099-01-01"))
-        init = await client.send_raw_request("initialize", _initialize_params())
-        assert init["protocolVersion"] == LATEST_HANDSHAKE_VERSION
+        with pytest.raises(MCPError) as init_exc:
+            await client.send_raw_request("initialize", _initialize_params())
     assert exc_info.value.error.code == UNSUPPORTED_PROTOCOL_VERSION
     assert exc_info.value.error.data == {
         "supported": list(MODERN_PROTOCOL_VERSIONS),
         "requested": "2099-01-01",
     }
+    assert init_exc.value.error.code == UNSUPPORTED_PROTOCOL_VERSION
 
 
 @pytest.mark.anyio
-async def test_dual_era_loop_bare_discover_rejects_without_locking(server: SrvT):
-    """A `server/discover` with no envelope triple is INVALID_PARAMS - never
-    -32022, so a released auto client's code-keyed fallback predicate takes the
-    legacy branch - and the connection can still complete the handshake."""
+async def test_dual_era_loop_bare_discover_opens_legacy_and_keeps_the_handshake_available(server: SrvT):
+    """`server/discover` without the envelope is not 2026 vocabulary, so it
+    opens a handshake connection: the answer is METHOD_NOT_FOUND (never -32022,
+    the one code a probing client must not treat as "fall back"), and the
+    client's fallback `initialize` then succeeds on the same connection."""
     async with dual_era_client(server) as (client, _):
         with pytest.raises(MCPError) as exc_info:
             await client.send_raw_request("server/discover", None)
         init = await client.send_raw_request("initialize", _initialize_params())
         assert init["protocolVersion"] == LATEST_HANDSHAKE_VERSION
-    assert exc_info.value.error.code == INVALID_PARAMS
+    assert exc_info.value.error.code == METHOD_NOT_FOUND
     assert exc_info.value.error.code != UNSUPPORTED_PROTOCOL_VERSION
 
 
 @pytest.mark.anyio
-async def test_dual_era_loop_ping_before_any_lock_stays_exempt_and_neutral(server: SrvT):
-    """A pre-handshake `ping` is answered (the init-gate exemption) and does
-    not lock an era: the connection can still go modern."""
+async def test_dual_era_loop_an_envelopeless_first_request_opens_a_legacy_connection(server: SrvT):
+    """A first request without the 2026 envelope - here a pre-handshake `ping`,
+    answered under the init-gate exemption - is handshake-era vocabulary and
+    opens a legacy connection: enveloped requests are refused after it."""
     async with dual_era_client(server) as (client, _):
         assert await client.send_raw_request("ping", None) == {}
-        result = await client.send_raw_request("tools/list", _modern_params())
-        assert result["tools"][0]["name"] == "t"
+        with pytest.raises(MCPError) as exc_info:
+            await client.send_raw_request("tools/list", _modern_params())
+    assert exc_info.value.error.code == INVALID_REQUEST
 
 
 @pytest.mark.anyio
@@ -1610,34 +1621,43 @@ async def test_dual_era_loop_modern_request_without_envelope_rejects(server: Srv
 
 
 @pytest.mark.anyio
-async def test_dual_era_loop_rejects_subscriptions_listen_on_modern(server: SrvT):
-    """`subscriptions/listen` is rejected before dispatch on the stream-pair
-    modern path (the registered handler assumes the HTTP entry's stream
-    semantics) - and like every failed request it does not lock the era, so
-    the legacy handshake stays available."""
-    async with dual_era_client(server) as (client, _):
-        with pytest.raises(MCPError) as exc_info:
-            await client.send_raw_request("subscriptions/listen", _modern_params())
-        init = await client.send_raw_request("initialize", _initialize_params())
-        assert init["protocolVersion"] == LATEST_HANDSHAKE_VERSION
-    assert exc_info.value.error.code == METHOD_NOT_FOUND
-    assert "not served over this transport" in exc_info.value.error.message
+async def test_dual_era_loop_serves_subscriptions_listen_over_the_stream_pair():
+    """`subscriptions/listen` is served over the duplex stream like any other
+    modern request: the acknowledgement notification rides the pipe first,
+    and the server ending the stream yields the request's graceful-close
+    result, stamped with the subscription id."""
+    listen = ListenHandler(InMemorySubscriptionBus())
+    listener = Server(name="listener-server", version="0.0.1", on_subscriptions_listen=listen)
+    async with dual_era_client(listener) as (client, recorder):
+        result: dict[str, Any] = {}
+
+        async def open_listen() -> None:
+            params = _modern_params(notifications={"toolsListChanged": True})
+            result.update(await client.send_raw_request("subscriptions/listen", params))
+
+        async with anyio.create_task_group() as tg:
+            tg.start_soon(open_listen)
+            await recorder.notified.wait()
+            listen.close()
+    assert recorder.notifications[0][0] == "notifications/subscriptions/acknowledged"
+    assert result["resultType"] == "complete"
+    assert SUBSCRIPTION_ID_META_KEY in result["_meta"]
 
 
 @pytest.mark.anyio
-async def test_dual_era_loop_malformed_envelope_content_never_locks(server: SrvT):
-    """The envelope triple with mis-shaped values fails the request but never
-    locks the era: the lock commits only when a modern request SUCCEEDS, so a
-    buggy client's initialize fallback still works (it must never see -32022
-    for a request that failed)."""
+async def test_dual_era_loop_malformed_envelope_content_is_a_modern_era_error(server: SrvT):
+    """An envelope with mis-shaped values still declares the 2026 era: the
+    request is rejected INVALID_PARAMS in modern vocabulary, and the connection
+    is a modern one, so a handshake sent afterwards is refused rather than
+    served on a connection that already speaks the other era."""
     params: dict[str, Any] = {"_meta": {**_modern_envelope(), CLIENT_INFO_META_KEY: 42}}
     async with dual_era_client(server) as (client, _):
         with pytest.raises(MCPError) as exc_info:
             await client.send_raw_request("tools/list", params)
-        init = await client.send_raw_request("initialize", _initialize_params())
-        assert init["protocolVersion"] == LATEST_HANDSHAKE_VERSION
+        with pytest.raises(MCPError) as init_exc:
+            await client.send_raw_request("initialize", _initialize_params())
     assert exc_info.value.error.code == INVALID_PARAMS
-    assert exc_info.value.error.code != UNSUPPORTED_PROTOCOL_VERSION
+    assert init_exc.value.error.code == UNSUPPORTED_PROTOCOL_VERSION
 
 
 @pytest.mark.anyio
@@ -1661,32 +1681,32 @@ async def test_dual_era_loop_pair_only_envelope_serves_modern_and_locks(server:
 
 
 @pytest.mark.anyio
-async def test_dual_era_loop_version_without_capabilities_rejects_naming_the_key_and_never_locks(server: SrvT):
+async def test_dual_era_loop_version_without_capabilities_rejects_naming_the_key(server: SrvT):
     """A `_meta` declaring the protocol version but missing the required
     client-capabilities key routes modern - never the legacy path with its
     generic 'Invalid request parameters' - and is rejected INVALID_PARAMS
-    naming the missing key; like every failed classification it locks no era,
-    so the legacy handshake stays available."""
+    naming the missing key; a correctly enveloped request then serves."""
     params = _modern_params()
     del params["_meta"][CLIENT_CAPABILITIES_META_KEY]
     async with dual_era_client(server) as (client, _):
         with pytest.raises(MCPError) as exc_info:
             await client.send_raw_request("tools/list", params)
-        init = await client.send_raw_request("initialize", _initialize_params())
-        assert init["protocolVersion"] == LATEST_HANDSHAKE_VERSION
+        result = await client.send_raw_request("tools/list", _modern_params())
+        assert result["tools"][0]["name"] == "t"
     assert exc_info.value.error.code == INVALID_PARAMS
     assert CLIENT_CAPABILITIES_META_KEY in exc_info.value.error.message
 
 
 @pytest.mark.anyio
-async def test_dual_era_loop_failed_modern_request_never_locks(server: SrvT):
-    """A well-formed modern request for an unknown method fails without
-    locking; the next modern request locks on its own success."""
+async def test_dual_era_loop_failed_modern_request_leaves_the_connection_modern(server: SrvT):
+    """A well-formed modern request for an unknown method fails on its own;
+    the connection is a 2026 one either way, so the next modern request
+    serves."""
     async with dual_era_client(server) as (client, _):
         with pytest.raises(MCPError) as exc_info:
             await client.send_raw_request("nope/missing", _modern_params())
-        init = await client.send_raw_request("initialize", _initialize_params())
-        assert init["protocolVersion"] == LATEST_HANDSHAKE_VERSION
+        result = await client.send_raw_request("tools/list", _modern_params())
+        assert result["tools"][0]["name"] == "t"
     assert exc_info.value.error.code == METHOD_NOT_FOUND
 
 
@@ -1705,9 +1725,9 @@ async def test_dual_era_loop_initialize_with_envelope_takes_the_handshake_path(s
 
 
 @pytest.mark.anyio
-async def test_dual_era_loop_modern_notification_dispatches_at_locked_version(server: SrvT):
-    """Notifications carry no envelope, so on a modern-locked connection they
-    dispatch with the locked protocol version."""
+async def test_dual_era_loop_modern_notification_dispatches_at_the_served_version(server: SrvT):
+    """Notifications carry no envelope, so on a 2026 connection they dispatch
+    at the modern version the server serves."""
     seen_versions: list[str] = []
     handled = anyio.Event()
 
@@ -1774,11 +1794,11 @@ async def test_dual_era_loop_modern_refuses_server_initiated_requests(server: Sr
 
 
 @pytest.mark.anyio
-async def test_dual_era_loop_late_modern_success_does_not_overwrite_a_committed_legacy_lock():
-    """The era settles exactly once, on the FIRST client-visible success: a
-    modern request that was already in flight when a legacy handshake
-    committed may still complete - its response stands - but the connection
-    stays legacy, so the handshaked client is never stranded."""
+async def test_dual_era_loop_initialize_during_an_in_flight_modern_request_is_refused():
+    """The era is fixed by the opening request, not by what has completed: a
+    handshake arriving while the opening modern request is still parked in
+    its handler is refused with -32022, and the parked request finishes on
+    the modern connection it opened."""
     entered = anyio.Event()
     release = anyio.Event()
 
@@ -1796,60 +1816,12 @@ async def test_dual_era_loop_late_modern_success_does_not_overwrite_a_committed_
 
         async with anyio.create_task_group() as tg:
             tg.start_soon(modern_call)
-            # The modern dispatch is parked in its handler before the
-            # handshake frame is even written, so the initialize commits first.
             await entered.wait()
-            init = await client.send_raw_request("initialize", _initialize_params())
-            assert init["protocolVersion"] == LATEST_HANDSHAKE_VERSION
+            with pytest.raises(MCPError) as exc_info:
+                await client.send_raw_request("initialize", _initialize_params())
+            assert exc_info.value.error.code == UNSUPPORTED_PROTOCOL_VERSION
             release.set()
         assert modern_result["tools"][0]["name"] == "t"
-        # The straggler's success did not move the era: plain legacy requests
-        # still serve (a modern overwrite would demand the envelope triple).
-        result = await client.send_raw_request("tools/list", None)
-        assert result["tools"][0]["name"] == "t"
-
-
-@pytest.mark.anyio
-async def test_dual_era_loop_modern_success_cancelled_away_at_the_response_write_never_locks():
-    """A peer cancel that lands while the handler is finishing means the
-    dispatcher replaces the computed result with "Request cancelled" - the
-    client never sees the success, so the era must not lock and the legacy
-    handshake must stay available."""
-    entered = anyio.Event()
-    release = anyio.Event()
-
-    async def list_tools(ctx: Ctx, params: PaginatedRequestParams | None) -> ListToolsResult:
-        entered.set()
-        # Survive the interrupt-mode scope cancel so the handler completes
-        # with the cancel pending - the cancellation is then delivered at the
-        # dispatcher's response-write checkpoint, after the era commit ran.
-        with anyio.CancelScope(shield=True):
-            await release.wait()
-        return ListToolsResult(tools=[])
-
-    parked = Server(name="parked-server", version="0.0.1", on_list_tools=list_tools)
-    async with dual_era_client(parked) as (client, _):
-        failures: list[MCPError] = []
-
-        async def modern_call() -> None:
-            with pytest.raises(MCPError) as exc_info:
-                await client.send_raw_request("tools/list", _modern_params())
-            failures.append(exc_info.value)
-
-        async with anyio.create_task_group() as tg:
-            tg.start_soon(modern_call)
-            await entered.wait()
-            # First request on a fresh dispatcher pair, so its id is 1.
-            await client.notify("notifications/cancelled", {"requestId": 1})
-            # The read loop handles frames in order: this marker's response
-            # proves the cancel was processed before the handler resumes.
-            with pytest.raises(MCPError):
-                await client.send_raw_request("probe/marker", None)
-            release.set()
-        assert failures[0].error.message == "Request cancelled"
-        # The cancelled-away success never locked the era.
-        init = await client.send_raw_request("initialize", _initialize_params())
-        assert init["protocolVersion"] == LATEST_HANDSHAKE_VERSION
 
 
 @pytest.mark.anyio
@@ -1857,8 +1829,7 @@ async def test_dual_era_loop_maps_unmapped_handler_exceptions_like_the_modern_ht
     """An unmapped handler exception on a modern request surfaces as the
     generic INTERNAL_ERROR - the same boundary as the modern HTTP entry - so
     handler internals never reach the wire. (The dispatcher's code-0
-    catch-all is a handshake-era compat pin and stays legacy-only.) The
-    failed request never locks, so the handshake stays available."""
+    catch-all is a handshake-era compat pin and stays legacy-only.)"""
 
     async def list_tools(ctx: Ctx, params: PaginatedRequestParams | None) -> ListToolsResult:
         raise RuntimeError("handler internals")
@@ -1867,8 +1838,6 @@ async def test_dual_era_loop_maps_unmapped_handler_exceptions_like_the_modern_ht
     async with dual_era_client(exploding) as (client, _):
         with pytest.raises(MCPError) as exc_info:
             await client.send_raw_request("tools/list", _modern_params())
-        init = await client.send_raw_request("initialize", _initialize_params())
-        assert init["protocolVersion"] == LATEST_HANDSHAKE_VERSION
     assert exc_info.value.error.code == INTERNAL_ERROR
     assert exc_info.value.error.message == "Internal server error"
     assert "handler internals" not in str(exc_info.value.error)
@@ -2008,6 +1977,79 @@ async def test_notify_only_outbound_forwards_notifications_and_refuses_requests(
         await outbound.send_raw_request("ping", None)
 
 
+@pytest.mark.anyio
+async def test_dual_era_loop_carries_the_sender_context_through_the_replay():
+    """A context-aware read stream's per-message sender context still reaches
+    handlers over the dual-era loop (the SSE transport delivers requests this
+    way): the opening-request replay forwards each frame's captured context."""
+    seen: list[str] = []
+
+    async def probe(ctx: Ctx, params: RequestParams | None) -> dict[str, Any]:
+        seen.append(_SENDER_VAR.get())
+        return {}
+
+    server = Server(name="ctx-server", version="0.0.1")
+    server.add_request_handler("x/probe", RequestParams, probe)
+    c2s_send, c2s_recv = create_context_streams[SessionMessage | Exception](8)
+    s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](8)
+    frame = JSONRPCRequest(jsonrpc="2.0", id=1, method="x/probe", params=_modern_params())
+    async with anyio.create_task_group() as tg, s2c_recv:
+        tg.start_soon(partial(serve_dual_era_loop, server, c2s_recv, s2c_send, lifespan_state=_LIFESPAN))
+        token = _SENDER_VAR.set("from-the-sender")
+        try:
+            await c2s_send.send(SessionMessage(message=frame))
+        finally:
+            _SENDER_VAR.reset(token)
+        with anyio.fail_after(5):
+            await s2c_recv.receive()
+        tg.cancel_scope.cancel()
+        await c2s_send.aclose()
+    assert seen == ["from-the-sender"]
+
+
+@pytest.mark.anyio
+async def test_dual_era_loop_leading_notifications_never_decide_the_era(server: SrvT):
+    """Frames a peer sends ahead of its first request never decide the era: a
+    stream of leading notifications well past the retained lead is followed by
+    an enveloped request, which still opens a 2026 connection and serves. The
+    lead is bounded, so the flood cannot grow the loop's buffer either."""
+    async with dual_era_client(server) as (client, _):
+        for _ in range(50):
+            await client.notify("notifications/lead", None)
+        result = await client.send_raw_request("tools/list", _modern_params())
+        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`."""