Trim comments and docstrings in the listen driver changes

Comment-only pass over the branch's additions: keep the non-inferable
invariants and motivations (receive-order rationale, sub-resource
admission, the wake-snapshot race, peek/commit semantics) at one to
three lines each, tighten docstrings to Google style with Raises
sections kept, and drop narration the code already states. No code
changes.

Author: maxisbey

src/mcp/client/client.py

@@ -675,28 +675,19 @@ def listen(
     ) -> AbstractAsyncContextManager[Subscription]:
         """Open a `subscriptions/listen` stream of typed change events (2026-07-28 only).
 
-        The keyword arguments mirror the wire `SubscriptionFilter` field for
-        field; `resource_subscriptions` names exact resource URIs to watch.
-        Entering waits for the server's acknowledgment - the subset it agreed
-        to deliver is `sub.honored` - then the handle yields events:
+        Keyword args mirror the wire `SubscriptionFilter`; entering waits for the ack (honored subset: `sub.honored`):
 
             async with client.listen(tools_list_changed=True) as sub:
                 async for event in sub:
                     tools = await client.list_tools()  # refetch on change
 
-        A graceful server close ends the loop; an abrupt drop raises
-        `SubscriptionLost`. Either way the stream is gone and there is no
-        replay: a client that keeps watching re-listens and refetches.
-        Exiting the context ends the subscription. Multiple subscriptions may
-        be open concurrently.
+        A graceful close ends the loop; an abrupt drop raises `SubscriptionLost`. No replay: re-listen and refetch.
 
         Raises:
-            ListenNotSupportedError: The negotiated protocol version predates
-                2026-07-28 (use `subscribe_resource` and `message_handler` there).
-            MCPError: The server rejected the request, or the connection
-                failed before the acknowledgment arrived.
+            ListenNotSupportedError: The negotiated protocol version predates 2026-07-28.
+            MCPError: The server rejected the request or the connection failed first.
             SubscriptionLost: The stream ended before it was acknowledged.
-            TimeoutError: The session's read timeout elapsed before the acknowledgment.
+            TimeoutError: The read timeout elapsed before the acknowledgment.
         """
         return _listen(
             self.session,
@@ -710,14 +701,9 @@ def listen(
     async def _evict_for_listen_event(self, event: ServerEvent) -> None:
         """Finish response-cache eviction before a listen consumer can refetch.
 
-        The eviction wrapper on message_handler runs on a spawned path; the
-        consumer's iterator would otherwise wake first, refetch through a
-        still-warm entry, and - events being deduplicated level triggers -
-        never get a second wake to correct it. The event's notification still
-        tees to that wrapper, so the same eviction fires twice; that is
-        deliberate (eviction is idempotent, and the tee-path one covers
-        non-iterating consumers sharing this cache). Same containment boundary
-        as `_evicting_message_handler`: a cache fault must not block delivery.
+        Without it the iterator wakes first and refetches a still-warm entry, with no
+        corrective wake (events are deduplicated level triggers). The tee path repeats
+        the eviction; deliberate: idempotent, and it covers non-iterating consumers.
         """
         cache = self._response_cache
         assert cache is not None  # installed as the event barrier only when a cache exists

src/mcp/client/session.py

@@ -363,12 +363,7 @@ def __init__(
         self._negotiated_version: str | None = None
         self._stamp: Callable[[dict[str, Any], CallOptions], None] = _preconnect_stamp
         self._task_group: anyio.abc.TaskGroup | None = None
-        # subscriptions/listen demux routes, keyed by the listen request's id
-        # (verbatim-typed: a plain dict already keeps 1 and "1" distinct).
-        # Fed by `_intercept_notification` on the dispatcher's receive path;
-        # membership alone decides ack consumption, so a closed subscription
-        # releases its id completely - raw escape-hatch listens (never
-        # registered here) keep receiving their acks via message_handler.
+        # subscriptions/listen demux routes; membership decides ack consumption (raw listens are never registered)
         self._listen_routes: dict[RequestId, ListenRoute] = {}
         if dispatcher is not None:
             if read_stream is not None or write_stream is not None:
@@ -1257,64 +1252,44 @@ def _unregister_listen_route(self, request_id: RequestId) -> None:
         self._listen_routes.pop(request_id, None)
 
     def _settle_listen_routes_closed(self) -> None:
-        """End every open subscription as lost: the session is gone.
-
-        Without this, a consumer iterating in a sibling task would park forever -
-        the driver task dies by cancellation and can never settle its route.
-        """
+        """Settle all open listen routes as lost on session exit; cancelled driver tasks cannot."""
         closed = MCPError(code=CONNECTION_CLOSED, message="Connection closed")
         for route in self._listen_routes.values():
             route.settle("lost", error=closed)
         self._listen_routes.clear()
 
     def _intercept_notification(self, method: str, params: Mapping[str, Any] | None) -> bool:
-        """Wire-order listen demux, run by the dispatcher on its receive path.
-
-        Route bookkeeping must advance in receive order relative to the listen
-        request's own result: the result resolves synchronously on this same
-        path, so an ack or event handled on the spawned `_on_notify` path could
-        lose the race and be dropped after the stream settled - a graceful
-        close swallowing the events that preceded it. Only the synchronous
-        bookkeeping happens here; the user-facing tee (message_handler,
-        logging) stays on the spawned path.
-
-        Returns True to consume the frame: an ack for a live driver route is
-        driver state, never surfaced. Raw escape-hatch listens have no route
-        registered and keep observing their acks via message_handler - as does
-        a stray ack for an already-closed driver id, whose registration is gone.
-
-        The `as_request_id` guard is not a tripwire: this reads raw wire
-        dicts, where a non-id (even unhashable) `_meta` value is
-        constructible and would fail the dict lookup.
+        """Wire-order listen demux, run synchronously on the dispatcher's receive path.
+
+        Bookkeeping must advance in receive order with the listen result (resolved on
+        this same path); the spawned `_on_notify` path would race it and drop events.
+        Returns True to consume the frame: a live route's ack is driver state, never surfaced.
         """
         if not self._listen_routes:
             return False
         if method == "notifications/cancelled":
             request_id = cancelled_request_id_from_params(params)
             if request_id is not None and (listen_route := self._listen_routes.get(request_id)) is not None:
-                # A server-sent cancel naming one of our listen requests is
-                # that stream's teardown signal (the stream-transport spelling).
+                # a server-sent cancel naming a listen request is that stream's teardown signal
                 listen_route.settle("lost")
             return False  # _on_notify swallows every cancelled either way (v1 parity)
         if params is None:
             return False
         meta = params.get("_meta")
         if not isinstance(meta, Mapping):
             return False
+        # as_request_id is not a tripwire: raw wire _meta can carry a non-id (even unhashable) value
         subscription_id = as_request_id(cast("Mapping[str, Any]", meta).get(SUBSCRIPTION_ID_META_KEY))
         if subscription_id is None or (listen_route := self._listen_routes.get(subscription_id)) is None:
             return False
         if method == "notifications/subscriptions/acknowledged":
             raw_filter = params.get("notifications")
             if raw_filter is None:
-                # The wire shape requires `notifications`; a frame without it is
-                # malformed, not an empty filter - leave it to the spawned
-                # path's validation warning rather than fabricating honored={}.
+                # malformed, not an empty filter: leave it to the spawned path's validation warning
                 return False
             try:
                 honored = types.SubscriptionFilter.model_validate(raw_filter)
             except ValidationError:
-                # Malformed ack: leave it to the spawned path's validation warning.
                 return False
             listen_route.set_acked(honored)
             return True
@@ -1356,9 +1331,7 @@ async def _on_notify(
             logger.warning("Failed to validate notification: %s", method, exc_info=True)
             return
         if isinstance(notification, types.CancelledNotification):
-            # Never surfaced to message_handler (v1 parity): the dispatcher
-            # already applied any in-flight cancellation, and a cancel naming
-            # one of our listen streams was settled by the wire-order intercept.
+            # Never surfaced (v1 parity): the dispatcher already applied it; listen cancels settled by the intercept.
             return
         try:
             if isinstance(notification, types.LoggingMessageNotification):

src/mcp/client/streamable_http.py

@@ -443,10 +443,8 @@ async def _handle_sse_response(
             logger.info("SSE stream disconnected, reconnecting...")
             await self._handle_reconnection(ctx, last_event_id, retry_interval_ms)
         else:
-            # Not resumable: the response can never arrive. Resolve the waiter
-            # instead of leaving the request pending for the session's lifetime
-            # (a listen stream's consumer would otherwise hang instead of
-            # learning the subscription is lost).
+            # Not resumable: resolve the waiter, else a listen stream's consumer
+            # would hang forever instead of learning the subscription is lost.
             await self._resolve_abandoned_request(
                 ctx.read_stream_writer, original_request_id, "SSE stream ended without a response"
             )
@@ -456,9 +454,7 @@ async def _resolve_abandoned_request(
     ) -> None:
         """Resolve a request whose response can never arrive with a synthesized error.
 
-        Best-effort, like the dispatcher's own resolution writes: the read
-        stream closing first means the session is being torn down and nobody
-        is waiting on this request anymore.
+        Best-effort: a closed read stream means the session is tearing down.
         """
         error_data = ErrorData(code=CONNECTION_CLOSED, message=message)
         error_msg = SessionMessage(JSONRPCError(jsonrpc="2.0", id=request_id, error=error_data))
@@ -475,16 +471,13 @@ async def _handle_reconnection(
         attempt: int = 0,
     ) -> None:
         """Reconnect with Last-Event-ID to resume stream after server disconnect."""
-        # Only requests reconnect: every caller reaches here from a request's
-        # response stream (`_handle_sse_response` asserts the same), so the
-        # original id to map responses onto is always present.
+        # Only requests reconnect: every caller arrives from a request's response stream.
         assert isinstance(ctx.session_message.message, JSONRPCRequest)
         original_request_id = ctx.session_message.message.id
 
         if attempt >= MAX_RECONNECTION_ATTEMPTS:
-            # Give up AND resolve: without a synthesized error the waiter is
-            # pending for the session's lifetime, and a request with no read
-            # timeout (a listen stream) would hang its caller forever.
+            # Resolve on give-up: a request with no read timeout (a listen
+            # stream) would otherwise hang its caller forever.
             logger.debug(f"Max reconnection attempts ({MAX_RECONNECTION_ATTEMPTS}) exceeded")
             await self._resolve_abandoned_request(
                 ctx.read_stream_writer,