Address review round two: bounded lead, exact context check, prose

- Frames arriving ahead of the client's first request are kept only up
  to a small bound and never decide the era: past the bound they are
  dropped while the loop keeps waiting for the opening request, so a
  bare notifications/initialized still opens the gate and a flood cannot
  grow the buffer or pick an era.
- Check the captured sender context with `is not None`, since an empty
  contextvars.Context is falsy.
- Word the legacy refusal to the actual rule (first request carried no
  2026 envelope) and update the Server.run docstring and one test to the
  decide-from-the-opening-request model.

Author: maxisbey

src/mcp/server/lowlevel/server.py

@@ -703,9 +703,9 @@ async def run(
 
         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

@@ -643,39 +643,39 @@ async def serve_dual_era_loop(
         await write_stream.aclose()
 
 
-_OPENING_PEEK_LIMIT: int = 32
-"""How many frames may precede the client's first request before the loop
-stops looking for one. Every legitimate client opens with a request, so this
-only bounds a peer that streams other frames without ever sending one."""
+_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."""
-    return getattr(stream, "last_context", None) or contextvars.copy_context()
+    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 anything.
+    """Peek at the client's first request without consuming it.
 
-    Reads frames until the first JSON-RPC request arrives, then yields that
-    request together with a stream that replays every frame read so far and
-    relays the rest of `read_stream` behind it, sender contexts included. The
-    request is `None` if the channel closes - or `_OPENING_PEEK_LIMIT` frames
-    pass - before any request appears.
+    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.
     """
-    peeked: list[tuple[contextvars.Context, SessionMessage | Exception]] = []
-    opening: JSONRPCRequest | None = None
+    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 peeked:
+            for envelope in lead:
                 await replay_send.send(envelope)
             async for item in read_stream:
                 await replay_send.send((_sender_context(read_stream), item))
@@ -685,15 +685,17 @@ async def replay_then_relay() -> None:
     # closes it and the replay channel.
     try:
         async for item in read_stream:
-            peeked.append((_sender_context(read_stream), item))
             if isinstance(item, SessionMessage) and isinstance(item.message, JSONRPCRequest):
-                opening = item.message
-                break
-            if len(peeked) >= _OPENING_PEEK_LIMIT:
+                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
         async with anyio.create_task_group() as tg:
             tg.start_soon(replay_then_relay)
-            yield opening, replayed
+            yield opening_request, replayed
             tg.cancel_scope.cancel()
     finally:
         await read_stream.aclose()
@@ -728,8 +730,8 @@ async def on_request(
         if method != "initialize" and _has_modern_envelope(params):
             raise MCPError(
                 code=INVALID_REQUEST,
-                message="connection was opened with the initialize handshake; "
-                "2026-07-28 envelope requests are not accepted on it",
+                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",
             )
         return await runner.on_request(dctx, method, params)
 

tests/server/test_runner.py

@@ -58,7 +58,6 @@
 from mcp.server.lowlevel.server import NotificationOptions, Server
 from mcp.server.models import InitializationOptions
 from mcp.server.runner import (
-    _OPENING_PEEK_LIMIT,
     ServerRunner,
     _extract_meta,
     _has_modern_envelope,
@@ -1545,7 +1544,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 "opened with the initialize handshake" in envelope_exc.value.error.message
+    assert "carried no 2026-07-28 envelope" in envelope_exc.value.error.message
 
 
 @pytest.mark.anyio
@@ -1725,9 +1724,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()
 
@@ -2008,16 +2007,16 @@ async def probe(ctx: Ctx, params: RequestParams | None) -> dict[str, Any]:
 
 
 @pytest.mark.anyio
-async def test_dual_era_loop_stops_peeking_after_a_flood_of_pre_request_frames(server: SrvT):
-    """A peer that streams notifications and never opens with a request only
-    gets the first `_OPENING_PEEK_LIMIT` frames buffered: past that the loop
-    stops looking for an opening request and serves the rest as an ordinary
-    handshake connection, so the handshake that follows still lands."""
+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(_OPENING_PEEK_LIMIT + 5):
-            await client.notify("notifications/flood", None)
-        init = await client.send_raw_request("initialize", _initialize_params())
-        assert init["protocolVersion"] == LATEST_HANDSHAKE_VERSION
+        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